Yuzhiran Dev
233e23016c
- 文章 HTML 存储从文件系统迁移至 articles 表，删除 releases 目录 - 合规审查从 DB 读取 HTML，审查结果写回 DB，通过后自动推进至待发布 - 新增 todayCount 筛选按钮，与系统概览统计数据一致 - 全屏预览修复：提升 z-index 超过侧边栏，添加退出全屏/关闭按钮 - 统一 '优化' → '审查' 命名，消除前后端术语不一致 - 调度器创作完成后自动触发审查（生成 → 审查 → 待发布） - 清理旧备份/调试文件、过期大纲和研究笔记
104 lines
3.6 KiB
Python
104 lines
3.6 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. 每个话题包含：领域、话题名称、热度原因（1句话）、相关热搜词（3个）
|
||
3. 优先选择在知乎/微博/小红书上有讨论度的话题
|
||
4. 输出 JSON 数组，格式：
|
||
[{{"domain": "领域", "topic": "话题名", "reason": "热度原因", "hot_keywords": ["词1","词2","词3"], "platform": "知乎/微博/小红书"}}]
|
||
|
||
只输出 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()
|