Yuzhiran Dev
194 lines
7.4 KiB
Python
194 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
选题引擎 v2：多趋势加权匹配 + 趋势缺口检测 + 新选题生成
|
||
"""
|
||
|
||
import sys, json, logging
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
|
||
PROJECT_ROOT = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend"))
|
||
|
||
from trends import load_trends
|
||
from db_helper import export_topics_to_json, update_topic_status, save_topics_to_db
|
||
from app.core.nvidia_client import call_llm
|
||
|
||
DATA_DIR = PROJECT_ROOT / "automation" / "data"
|
||
TRENDS_FILE = DATA_DIR / "trends.json"
|
||
TODAY = __import__('datetime').datetime.now().strftime("%Y-%m-%d")
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
TREND_DOMAIN_MAP = {
|
||
"远程工作": "未来工作方式",
|
||
"AI工具": "AI与效率",
|
||
"可持续生活": "可持续生活系统",
|
||
"知识管理": "个人知识工厂",
|
||
"数字生活": "科技人文交叉",
|
||
"科技人文": "科技人文交叉",
|
||
"个人成长": "个人成长",
|
||
"副业": "个人成长",
|
||
"AI创作": "AI与效率",
|
||
"未来工作": "未来工作方式",
|
||
"效率工具": "AI与效率",
|
||
"家庭教育": "科技人文交叉",
|
||
}
|
||
|
||
def _topic_trend_score(topic: Dict, trend: Dict) -> float:
|
||
field = (topic.get("field") or "").lower()
|
||
title = (topic.get("title") or "").lower()
|
||
core = (topic.get("core_concept") or "").lower()
|
||
trend_domain = TREND_DOMAIN_MAP.get(trend.get("domain", ""), "")
|
||
keywords = [trend.get("topic", "")] + trend.get("hot_keywords", [])
|
||
score = 0.0
|
||
if trend_domain and trend_domain in field:
|
||
score += 5
|
||
for kw in keywords:
|
||
kw = kw.lower()
|
||
if kw in title: score += 3
|
||
elif kw in core: score += 2
|
||
return score
|
||
|
||
def match_trends_to_topics() -> List[Dict]:
|
||
trends = load_trends()
|
||
topics = [t for t in export_topics_to_json() if t.get('status') in ('pending', '待处理')]
|
||
if not trends or not topics:
|
||
return []
|
||
|
||
topic_scores = {}
|
||
for topic in topics:
|
||
tid = topic["id"]
|
||
per_trend = []
|
||
total = 0.0
|
||
for trend in trends:
|
||
s = _topic_trend_score(topic, trend)
|
||
if s > 0:
|
||
per_trend.append({"trend": trend["topic"], "domain": trend.get("domain",""), "score": s})
|
||
total += s
|
||
if per_trend:
|
||
topic_scores[tid] = {
|
||
"topic_id": tid,
|
||
"topic_title": topic["title"],
|
||
"total_score": total + (topic.get("priority_score", 0) or 0) * 0.3,
|
||
"matched_trends": sorted(per_trend, key=lambda x: -x["score"]),
|
||
"matched_count": len(per_trend),
|
||
}
|
||
|
||
results = sorted(topic_scores.values(), key=lambda x: -x["total_score"])
|
||
return results
|
||
|
||
def detect_trend_gaps() -> List[Dict]:
|
||
trends = load_trends()
|
||
topics = export_topics_to_json()
|
||
gaps = []
|
||
for trend in trends:
|
||
matched = False
|
||
trend_keywords = [trend.get("topic", "").lower()] + [k.lower() for k in trend.get("hot_keywords", [])]
|
||
for t in topics:
|
||
title = (t.get("title") or "").lower()
|
||
core = (t.get("core_concept") or "").lower()
|
||
for kw in trend_keywords:
|
||
if kw in title or kw in core:
|
||
matched = True
|
||
break
|
||
if matched:
|
||
break
|
||
if not matched:
|
||
gaps.append(trend)
|
||
return gaps
|
||
|
||
def generate_new_topics(gaps: List[Dict]) -> List[Dict]:
|
||
if not gaps:
|
||
return []
|
||
prompt = f"""你是一个敏锐的内容策略师，擅长将热点转化为有价值、有传播力的选题。以下热点当前未覆盖，请为每个热点生成选题建议。
|
||
|
||
注意：选题要兼具SEO价值（能在搜索中被找到）和社交传播力（能在平台引发讨论）。
|
||
|
||
热点列表：
|
||
{chr(10).join(f'- {g["topic"]}（{g.get("domain","")}）— {g.get("reason","")}。热搜词：{", ".join(g.get("hot_keywords", []))}）' for g in gaps)}
|
||
|
||
每个选题需包含：
|
||
- "title": 标题（20字内，包含核心关键词，有吸引力）
|
||
- "field": 所属领域
|
||
- "core_concept": 核心观点（一句话说清独特价值）
|
||
- "audience_pain": 受众痛点（真实用户的困惑/焦虑/需求）
|
||
- "unique_angle": 独特视角（差异化切入点，含SEO关键词潜力）
|
||
- "target_platform": 最适合发布平台（知乎/小红书/微信/多平台）
|
||
- "estimated_search_volume": 预估搜索热度（高/中/低）
|
||
|
||
只输出 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]
|
||
suggestions = json.loads(resp)
|
||
if isinstance(suggestions, list):
|
||
return suggestions
|
||
except Exception as e:
|
||
logger.warning(f"新选题生成失败: {e}")
|
||
return []
|
||
|
||
def suggest_new_topics() -> List[Dict]:
|
||
gaps = detect_trend_gaps()
|
||
if not gaps:
|
||
logger.info("所有趋势均有匹配选题，无需新增")
|
||
return []
|
||
logger.info(f"发现 {len(gaps)} 个趋势缺口: {[g['topic'] for g in gaps]}")
|
||
new_topics = generate_new_topics(gaps)
|
||
if new_topics:
|
||
logger.info(f"生成 {len(new_topics)} 个新选题建议")
|
||
report_path = DATA_DIR / "topic_suggestions.json"
|
||
data = {"date": TODAY, "gaps": gaps, "suggestions": new_topics}
|
||
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||
report_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
return new_topics
|
||
|
||
def select_best_topic() -> Optional[Dict]:
|
||
matches = match_trends_to_topics()
|
||
if matches:
|
||
best = matches[0]
|
||
logger.info(f"选题推荐: {best['topic_id']} {best['topic_title']} "
|
||
f"(命中 {best['matched_count']} 个趋势, 综合分: {best['total_score']:.1f})")
|
||
topic = __import__('db_helper').get_topic_by_id(best['topic_id'])
|
||
if topic:
|
||
return topic
|
||
|
||
from db_helper import get_next_topic
|
||
fallback = get_next_topic(priority='高') or get_next_topic()
|
||
if fallback:
|
||
logger.info(f"无趋势匹配，退回到优先级最高选题: {fallback['id']}")
|
||
return fallback
|
||
|
||
def print_report():
|
||
matches = match_trends_to_topics()
|
||
trends = load_trends()
|
||
gaps = detect_trend_gaps()
|
||
print(f"=== 选题匹配报告 ({TODAY}) ===\n")
|
||
print(f"当前趋势数: {len(trends)}")
|
||
print(f"待处理选题: {len([t for t in export_topics_to_json() if t.get('status') in ('pending', '待处理')])}")
|
||
print(f"趋势匹配数: {len(matches)}")
|
||
print(f"趋势缺口: {len(gaps)}\n")
|
||
|
||
if matches:
|
||
print(f"{'排名':>4} {'选题ID':6} {'综合分':>6} {'命中趋势':>6} {'趋势详情':30} {'选题标题'}")
|
||
print("-" * 100)
|
||
for i, m in enumerate(matches[:10], 1):
|
||
trends_str = ", ".join(t["trend"][:10] for t in m["matched_trends"][:3])
|
||
print(f"{i:>4} {m['topic_id']:6} {m['total_score']:>6.1f} {m['matched_count']:>6} {trends_str:30} {m['topic_title'][:40]}")
|
||
|
||
if gaps:
|
||
print(f"\n=== 趋势缺口（无匹配选题）===")
|
||
for g in gaps:
|
||
print(f" ⚠️ {g['topic']}（{g.get('domain','')}）— {g.get('reason','')}")
|
||
|
||
if __name__ == "__main__":
|
||
logging.basicConfig(level=logging.INFO)
|
||
print_report()
|
||
print()
|
||
new = suggest_new_topics()
|
||
if new:
|
||
print(f"\n新选题建议已保存到 automation/data/topic_suggestions.json")
|