#!/usr/bin/env python3 """ 热点趋势感知模块 - 实时热搜API（百度/微博/知乎） - LLM 生成作为 fallback """ import json, datetime, logging, sys, re from pathlib import Path from typing import List, Dict import requests PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend")) from app.core.nvidia_client import call_llm DATA_DIR = PROJECT_ROOT / "automation" / "data" TRENDS_FILE = DATA_DIR / "trends.json" LOGS_DIR = PROJECT_ROOT / "automation" / "logs" TODAY = datetime.datetime.now().strftime("%Y-%m-%d") logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler(LOGS_DIR / f"trends_{TODAY}.log"), logging.StreamHandler()] ) logger = logging.getLogger(__name__) DOMAINS = ["远程工作", "AI工具", "可持续生活", "知识管理", "数字生活", "科技人文"] UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" _KEYWORD_DOMAIN_MAP = [ (r"AI|人工智能|大模型|GPT|机器学习|深度学习|聊天机器人|LLM", "AI工具"), (r"远程|居家办公|自由职业|数字游民|远程协作", "远程工作"), (r"可持续|环保|低碳|绿色|碳中和|循环|零浪费|垃圾分类|节能", "可持续生活"), (r"知识管理|笔记|Obsidian|Notion|第二大脑|读书|阅读", "知识管理"), (r"数字生活|数码|手机|电脑|智能|APP|应用|软件", "数字生活"), (r"科技|人文|教育|心理|哲学|社会学", "科技人文"), ] def _guess_domain(topic: str, reason: str = "") -> str: text = (topic + " " + reason).lower() for pattern, domain in _KEYWORD_DOMAIN_MAP: if re.search(pattern, text, re.IGNORECASE): return domain return "科技人文" def fetch_weibo_hot() -> List[Dict]: """微博热搜""" try: resp = requests.get("https://weibo.com/ajax/side/hotSearch", headers={"User-Agent": UA}, timeout=10) if resp.status_code != 200: logger.warning(f"微博热搜返回 {resp.status_code}") return [] data = resp.json() realtime = (data.get("data", {}) or {}).get("realtime", []) results = [] for item in realtime[:30]: word = item.get("word", "").strip() if not word: continue num = item.get("raw_hot", 0) or item.get("num", 0) results.append({ "domain": _guess_domain(word), "topic": word, "reason": f"微博热搜 热度{num}", "hot_keywords": [word], "platform": "微博", "seo_angle": word, "engagement": "高" if num > 500000 else "中", "source": "weibo", }) logger.info(f"微博热搜获取 {len(results)} 条") return results except Exception as e: logger.warning(f"微博热搜失败: {e}") return [] def fetch_zhihu_hot() -> List[Dict]: """知乎热榜""" try: resp = requests.get( "https://www.zhihu.com/api/v3/feed/topstory/hot-lists/total", headers={"User-Agent": UA}, timeout=10, ) if resp.status_code != 200: logger.warning(f"知乎热榜返回 {resp.status_code}") return [] data = resp.json() items = data.get("data", []) results = [] for item in items[:20]: target = item.get("target", {}) title = target.get("title", "").strip() if not title: continue metrics = target.get("metrics_area", {}) if isinstance(metrics, str): metrics = {} detail_text = target.get("excerpt", "") or target.get("answer", "") results.append({ "domain": _guess_domain(title, detail_text), "topic": title, "reason": f"知乎热榜", "hot_keywords": [title] + re.findall(r'[^\s，。！？、,\.]{2,6}', title)[:3], "platform": "知乎", "seo_angle": title, "engagement": "高", "source": "zhihu", }) logger.info(f"知乎热榜获取 {len(results)} 条") return results except Exception as e: logger.warning(f"知乎热榜失败: {e}") return [] def fetch_baidu_hot() -> List[Dict]: """百度热搜""" try: resp = requests.get( "https://top.baidu.com/api/board?tab=realtime", headers={ "User-Agent": UA, "Referer": "https://top.baidu.com/", }, timeout=10, ) if resp.status_code == 200: data = resp.json() cards = data.get("data", {}).get("cards", []) results = [] for card in cards: items = card.get("content", []) for item in items[:30]: word = item.get("query", "").strip() or item.get("word", "").strip() if not word: continue hot_score = item.get("hotScore", 0) or item.get("heat", 0) desc = item.get("desc", "") results.append({ "domain": _guess_domain(word, desc), "topic": word, "reason": f"百度热搜 热度{hot_score}", "hot_keywords": [word], "platform": "百度", "seo_angle": word, "engagement": "高" if hot_score > 500000 else "中", "source": "baidu", }) if results: logger.info(f"百度热搜获取 {len(results)} 条") return results resp2 = requests.get("https://top.baidu.com/board?tab=realtime", headers={"User-Agent": UA}, timeout=10) if resp2.status_code == 200: html = resp2.text words = re.findall(r'"word":"([^"]+)"', html) hot_scores = re.findall(r'"hotScore":(\d+)', html) results = [] for i, word in enumerate(words[:30]): score = int(hot_scores[i]) if i < len(hot_scores) else 0 results.append({ "domain": _guess_domain(word), "topic": word, "reason": f"百度热搜 热度{score}", "hot_keywords": [word], "platform": "百度", "seo_angle": word, "engagement": "高" if score > 500000 else "中", "source": "baidu", }) if results: logger.info(f"百度热搜（HTML解析）获取 {len(results)} 条") return results logger.warning("百度热搜获取失败") return [] except Exception as e: logger.warning(f"百度热搜失败: {e}") return [] def fetch_llm_trends() -> List[Dict]: prompt = f"""你是中文互联网趋势分析师。请列出今天（2026年5月）中文互联网上最值得创作的10个话题。 要求： 1. 覆盖领域：{', '.join(DOMAINS)} 2. 从真实用户角度出发 3. 每个话题需包含： - "domain": 领域 - "topic": 话题名称 - "reason": 为什么现在讨论这个（1句话，有具体事件/数据支撑） - "hot_keywords": 3-5个搜索词（含1-2个长尾词） - "platform": 最适合分发的平台（知乎/小红书/微信/多平台） - "seo_angle": 从什么角度切入能获得搜索流量（1句话） - "engagement": 高/中/低 输出 JSON 数组： [{{"domain": "...", "topic": "...", "reason": "...", "hot_keywords": ["..."], "platform": "...", "seo_angle": "...", "engagement": "..."}}] 只输出 JSON，不要其他文字。""" try: resp = call_llm(prompt, temperature=0.4) resp = resp.strip() if resp.startswith("```"): resp = resp.split("\n", 1)[1].rsplit("\n", 1)[0] trends = json.loads(resp) if isinstance(trends, list): for t in trends: t["source"] = "llm" return trends except Exception as e: logger.warning(f"LLM 趋势获取失败: {e}") return [] def save_trends(trends: List[Dict]): data = { "date": TODAY, "updated_at": datetime.datetime.now().isoformat(), "trends": trends } DATA_DIR.mkdir(parents=True, exist_ok=True) TRENDS_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8') logger.info(f"趋势数据已保存: {len(trends)} 条") def load_trends() -> List[Dict]: if TRENDS_FILE.exists(): try: data = json.loads(TRENDS_FILE.read_text(encoding='utf-8')) if data.get("date") == TODAY: return data.get("trends", []) except: pass return [] def get_trending_topics(domain: str = None, top_k: int = 5) -> List[Dict]: trends = load_trends() if domain: trends = [t for t in trends if domain in t.get("domain", "") or t.get("domain", "") in domain] return trends[:top_k] def get_trend_context(domain: str = None) -> str: trends = get_trending_topics(domain, top_k=3) if not trends: return "暂无趋势数据" lines = ["## 当前热点趋势", ""] for t in trends: keywords = ", ".join(t.get("hot_keywords", [])) source_tag = {"weibo": "🔥", "zhihu": "📖", "baidu": "🔍", "llm": "🤖"}.get(t.get("source", ""), "") lines.append(f"- {source_tag} **{t['topic']}**（{t.get('platform','')}）：{t.get('reason','')}") if keywords: lines.append(f" 热搜词：{keywords}") return "\n".join(lines) def main(): logger.info("开始获取热点趋势...") all_trends = [] seen = set() for fetcher_name, fetcher in [("百度热搜", fetch_baidu_hot), ("微博热搜", fetch_weibo_hot), ("知乎热榜", fetch_zhihu_hot)]: try: items = fetcher() for item in items: topic = item.get("topic", "").strip() if topic and topic not in seen: seen.add(topic) all_trends.append(item) logger.info(f"{fetcher_name}: {len(items)} 条") except Exception as e: logger.warning(f"{fetcher_name} 失败: {e}") logger.info(f"实时热搜合计 {len(all_trends)} 条，尝试LLM补充...") llm_trends = fetch_llm_trends() for item in llm_trends: topic = item.get("topic", "").strip() if topic and topic not in seen: seen.add(topic) all_trends.append(item) if all_trends: save_trends(all_trends) for t in all_trends[:15]: s = {"weibo": "微", "zhihu": "知", "baidu": "百", "llm": "AI"}.get(t.get("source", ""), "?") print(f" [{s}][{t.get('domain','?')}] {t['topic']}") print(f"完成，共 {len(all_trends)} 条（实时 {sum(1 for t in all_trends if t.get('source') != 'llm')} 条，LLM {sum(1 for t in all_trends if t.get('source') == 'llm')} 条）") else: print("未获取到趋势数据") if __name__ == "__main__": main()