Yuzhiran Dev
fef435cc78
- 审查：移除 manual_review，改为迭代LLM修复(最多3次)，合规分回写Topic - 调度：scheduler 新增话题采集定时任务 scheduled_collect (01:30) - 提示词：全链路8文件≈24个提示词升级，增强SEO/平台推荐/真人感
115 lines
4.4 KiB
Python
115 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
热点趋势感知模块
|
||
- LLM 生成当前领域热点话题
|
||
- 可扩展接入外部热搜 API
|
||
"""
|
||
|
||
import json, datetime, logging, sys
|
||
from pathlib import Path
|
||
from typing import List, Dict
|
||
|
||
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工具", "可持续生活", "知识管理", "数字生活", "科技人文"]
|
||
|
||
def fetch_llm_trends() -> List[Dict]:
|
||
prompt = f"""你是中文互联网趋势分析师，擅长发现真正有价值、能出爆款的热点话题。请列出今天（2026年5月）中文互联网上最值得创作的10个话题。
|
||
|
||
要求：
|
||
1. 覆盖领域：{', '.join(DOMAINS)}
|
||
2. 每个话题必须从「真实用户」角度出发——不是学术热门，而是普通人正在搜、在讨论的
|
||
3. 判断依据：知乎有高赞讨论？小红书有爆款笔记趋势？微信有刷屏文章？
|
||
4. SEO价值：这个话题是否有持续搜索量，还是纯短期流量？
|
||
5. 每个话题需包含：
|
||
- "domain": 领域
|
||
- "topic": 话题名称（老百姓能听懂的说法）
|
||
- "reason": 为什么现在讨论这个（1句话，说人话，有具体事件/数据支撑）
|
||
- "hot_keywords": 用户真实搜索时会用的词（3-5个，包含1-2个长尾词）
|
||
- "platform": 最适合分发此话题的平台（知乎/小红书/微信/多平台）
|
||
- "seo_angle": 从什么角度切入能获得搜索流量（1句话）
|
||
- "engagement": 预估互动潜力（高/中/低）
|
||
|
||
输出 JSON 数组：
|
||
[{{"domain": "领域", "topic": "话题名", "reason": "热度原因", "hot_keywords": ["词1","词2","词3"], "platform": "知乎/小红书/微信/多平台", "seo_angle": "SEO切入点", "engagement": "高/中/低"}}]
|
||
|
||
只输出 JSON，不要其他文字。"""
|
||
try:
|
||
resp = call_llm(prompt, temperature=0.4, max_tokens=2000)
|
||
resp = resp.strip()
|
||
if resp.startswith("```"):
|
||
resp = resp.split("\n", 1)[1].rsplit("\n", 1)[0]
|
||
trends = json.loads(resp)
|
||
if isinstance(trends, list):
|
||
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", []))
|
||
lines.append(f"- **{t['topic']}**（{t.get('platform','')}）：{t.get('reason','')}")
|
||
if keywords:
|
||
lines.append(f" 热搜词：{keywords}")
|
||
return "\n".join(lines)
|
||
|
||
def main():
|
||
logger.info("开始获取热点趋势...")
|
||
trends = fetch_llm_trends()
|
||
if trends:
|
||
save_trends(trends)
|
||
for t in trends:
|
||
print(f" [{t.get('domain','?')}] {t['topic']} — {t.get('platform','')}")
|
||
else:
|
||
print("未获取到趋势数据")
|
||
print(f"完成，共 {len(trends)} 条")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|