yuzhiran
d11d7f4980
db_helper.py: save_article now calculates and persists word_count generator.py: run_creator_blocking sets word_count for HTML-imported articles writer.py: fix title regex stripping content-leading numbers (35岁后→岁后) trends.py: fix Baidu hot_score str/int type comparison crash database.py: add missing content_tasks.org_id ALTER TABLE migration schemas.py + topics.py: topic list API returns article_count + articles[] previews topics.html: table view and card view show article badges with word counts, clickable to open preview Ultraworked with Sisyphus Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
341 lines
13 KiB
Python
341 lines
13 KiB
Python
#!/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__)
|
||
|
||
DEFAULT_DOMAINS = ["远程工作", "AI工具", "可持续生活", "知识管理", "数字生活", "科技人文"]
|
||
_cached_domains = None
|
||
|
||
def _load_domains():
|
||
global _cached_domains
|
||
if _cached_domains is not None:
|
||
return _cached_domains
|
||
try:
|
||
from app.core.prompt_loader import _get_session
|
||
from app.models import SystemConfig
|
||
session = _get_session()
|
||
try:
|
||
sc = session.query(SystemConfig).filter(SystemConfig.key == "trend_domains").first()
|
||
if sc and sc.value:
|
||
parsed = json.loads(sc.value)
|
||
if isinstance(parsed, list) and parsed:
|
||
_cached_domains = parsed
|
||
logger.info(f"从DB加载 {len(_cached_domains)} 个trend_domains")
|
||
return _cached_domains
|
||
finally:
|
||
session.close()
|
||
except Exception as e:
|
||
logger.warning(f"从DB加载 trend_domains 失败: {e}")
|
||
_cached_domains = DEFAULT_DOMAINS
|
||
return _cached_domains
|
||
|
||
def get_domains():
|
||
return _load_domains()
|
||
|
||
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"科技|人文|教育|心理|哲学|社会学", "科技人文"),
|
||
]
|
||
|
||
_cached_keyword_domain_map = None
|
||
|
||
def _load_keyword_domain_map():
|
||
global _cached_keyword_domain_map
|
||
if _cached_keyword_domain_map is not None:
|
||
return _cached_keyword_domain_map
|
||
|
||
try:
|
||
from app.core.prompt_loader import _get_session
|
||
from app.models import KeywordDomainMap
|
||
session = _get_session()
|
||
try:
|
||
rows = session.query(KeywordDomainMap).filter(KeywordDomainMap.is_active == True).order_by(KeywordDomainMap.sort_order).all()
|
||
if rows:
|
||
_cached_keyword_domain_map = [(r.pattern, r.domain) for r in rows]
|
||
logger.info(f"从DB加载 {len(rows)} 条 keyword_domain_map 规则")
|
||
return _cached_keyword_domain_map
|
||
finally:
|
||
session.close()
|
||
except Exception as e:
|
||
logger.warning(f"从DB加载 keyword_domain_map 失败: {e}")
|
||
|
||
_cached_keyword_domain_map = _KEYWORD_DOMAIN_MAP
|
||
return _cached_keyword_domain_map
|
||
|
||
|
||
def _guess_domain(topic: str, reason: str = "") -> str:
|
||
text = (topic + " " + reason).lower()
|
||
for pattern, domain in _load_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_raw = item.get("hotScore", 0) or item.get("heat", 0)
|
||
hot_score = int(hot_score_raw) if hot_score_raw else 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]:
|
||
try:
|
||
from prompt_loader import get_prompt, get_prompt_params
|
||
prompt = get_prompt("topics_trends", date=datetime.datetime.now().strftime("%Y年%m月%d"),
|
||
domains=", ".join(get_domains()))
|
||
params = get_prompt_params("topics_trends")
|
||
resp = call_llm(prompt, temperature=params.get("temperature", 0.4), max_tokens=params.get("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):
|
||
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 Exception:
|
||
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()
|