20.8.2 抖音/豆包工具（巨量算数、抖音热榜、字节爬虫日志）

# 伪代码示例：通过模拟请求获取巨量算数趋势数据
import requests
import json

def get_juliang_trend(keyword, date_range='7d'):
    """
    获取指定关键词的巨量算数趋势数据
    注意：此接口为模拟，实际需处理反爬与认证
    """
    url = "https://trendinsight.oceanengine.com/api/v1/trend/keyword"
    headers = {
        "User-Agent": "Mozilla/5.0...",
        "Cookie": "your_cookie_here"
    }
    params = {
        "keyword": keyword,
        "date_range": date_range
    }
    response = requests.get(url, headers=headers, params=params)
    return response.json()

# 示例：监控核心产品关键词趋势
trend_data = get_juliang_trend("智能家居")
print(trend_data['data']['trend_list'])

# 伪代码示例：监控抖音热榜变化
import requests
import time
from datetime import datetime

class DouyinHotListMonitor:
    def __init__(self):
        self.base_url = "https://www.douyin.com/aweme/v1/web/hot/search/list/"
        self.headers = {
            "User-Agent": "Mozilla/5.0...",
            "Cookie": "your_cookie"
        }
        self.previous_list = []
    
    def fetch_hot_list(self):
        """获取当前热榜"""
        params = {
            "detail_list": 1,
            "source": 0,
            "main_billboard_count": 10
        }
        response = requests.get(self.base_url, headers=self.headers, params=params)
        data = response.json()
        
        hot_list = []
        for item in data.get('data', {}).get('word_list', []):
            hot_list.append({
                'word': item.get('word'),
                'hot_value': item.get('hot_value'),
                'rank': item.get('rank'),
                'timestamp': datetime.now().isoformat()
            })
        return hot_list
    
    def detect_new_hot_words(self):
        """检测新增热点"""
        current_list = self.fetch_hot_list()
        current_words = {item['word'] for item in current_list}
        previous_words = {item['word'] for item in self.previous_list}
        
        new_words = current_words - previous_words
        if new_words:
            print(f"[{datetime.now()}] 检测到新热点: {new_words}")
            # 触发内容生成流程
            self.trigger_content_generation(new_words)
        
        self.previous_list = current_list
        return current_list
    
    def trigger_content_generation(self, new_words):
        """根据新热点触发内容生成（示例）"""
        for word in new_words:
            # 1. 检查是否与产品相关
            # 2. 如果相关，自动生成视频脚本/图文
            # 3. 推送到内容管理系统
            print(f"正在为热点 '{word}' 生成内容...")

# 启动监控（每15分钟执行一次）
monitor = DouyinHotListMonitor()
while True:
    monitor.detect_new_hot_words()
    time.sleep(900)

import re
from collections import Counter
from datetime import datetime

def analyze_bytespider_log(log_file_path):
    """分析Bytespider爬虫日志"""
    bytespider_pattern = re.compile(r'.*Bytespider.*')
    url_pattern = re.compile(r'GET\s+(\S+)\s+HTTP')
    status_pattern = re.compile(r'HTTP/\d\.\d"\s+(\d{3})')
    time_pattern = re.compile(r'\[(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2})')
    
    url_counter = Counter()
    status_counter = Counter()
    hourly_counter = Counter()
    
    with open(log_file_path, 'r') as f:
        for line in f:
            if bytespider_pattern.match(line):
                # 提取URL
                url_match = url_pattern.search(line)
                if url_match:
                    url_counter[url_match.group(1)] += 1
                
                # 提取状态码
                status_match = status_pattern.search(line)
                if status_match:
                    status_counter[status_match.group(1)] += 1
                
                # 提取时间（按小时）
                time_match = time_pattern.search(line)
                if time_match:
                    hour = time_match.group(1)[:13]  # 取到小时
                    hourly_counter[hour] += 1
    
    print("=== Bytespider 抓取统计 ===")
    print(f"总请求数: {sum(url_counter.values())}")
    print(f"\nTop 10 URL:")
    for url, count in url_counter.most_common(10):
        print(f"  {count:5d} {url}")
    print(f"\n状态码分布:")
    for status, count in status_counter.most_common():
        print(f"  {status}: {count}")
    print(f"\n每小时抓取分布:")
    for hour, count in hourly_counter.most_common(10):
        print(f"  {hour}: {count}")

# 使用示例
analyze_bytespider_log("/var/log/nginx/access.log")

# 使用 prometheus_client 暴露Bytespider指标
from prometheus_client import start_http_server, Counter, Gauge
import time

# 定义指标
bytespider_requests = Counter('bytespider_requests_total', 'Total Bytespider requests')
bytespider_errors = Counter('bytespider_errors_total', 'Bytespider requests with errors')
bytespider_last_seen = Gauge('bytespider_last_seen_timestamp', 'Last time Bytespider was seen')

def log_parser_loop():
    """持续解析日志并更新指标"""
    with open('/var/log/nginx/access.log', 'r') as f:
        f.seek(0, 2)  # 移动到文件末尾
        while True:
            line = f.readline()
            if not line:
                time.sleep(1)
                continue
            if 'Bytespider' in line:
                bytespider_requests.inc()
                bytespider_last_seen.set(time.time())
                # 检查是否返回错误
                if ' 404 ' in line or ' 500 ' in line:
                    bytespider_errors.inc()

if __name__ == '__main__':
    start_http_server(8000)  # 暴露指标在 :8000/metrics
    log_parser_loop()