yuzhiran
9fba863ce7
initial_data: field descriptions humanized, collector search queries include long-tail keywords, platform formats updated collector.py: DEFAULT_CHINA_PAINS expanded with specific pain questions (e.g. '35岁学AI来得及吗') topic_selector.py: domain mapping 11->18 entries (Prompt/AI编程/AI写作/数据隐私/AI教育/一人企业) Ultraworked with Sisyphus Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
218 lines
8.0 KiB
Python
218 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
选题引擎：多趋势加权匹配 + 趋势缺口检测 + 新选题生成
|
||
结合采集到的热点趋势和当前选题库，推荐最优选题并检测趋势缺口
|
||
"""
|
||
|
||
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
|
||
from prompt_loader import get_prompt
|
||
|
||
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__)
|
||
|
||
DEFAULT_TREND_DOMAIN_MAP = {
|
||
"AI工具": "AI与效率",
|
||
"AI创作": "AI与效率",
|
||
"AI职场": "AI与效率",
|
||
"效率工具": "AI与效率",
|
||
"Prompt": "AI与效率",
|
||
"提示词": "AI与效率",
|
||
"AI编程": "AI与效率",
|
||
"AI写作": "AI与效率",
|
||
"未来工作": "未来工作方式",
|
||
"远程工作": "未来工作方式",
|
||
"AI就业": "未来工作方式",
|
||
"一人企业": "未来工作方式",
|
||
"副业": "未来工作方式",
|
||
"科技人文": "科技人文",
|
||
"数字生活": "科技人文",
|
||
"AI伦理": "科技人文",
|
||
"AI情感": "科技人文",
|
||
"数据隐私": "科技人文",
|
||
"AI教育": "科技人文",
|
||
}
|
||
|
||
_cached_trend_domain_map = None
|
||
|
||
def _load_trend_domain_map():
|
||
global _cached_trend_domain_map
|
||
if _cached_trend_domain_map is not None:
|
||
return _cached_trend_domain_map
|
||
try:
|
||
from app.core.prompt_loader import _get_session
|
||
from app.models import TrendFieldMapping
|
||
session = _get_session()
|
||
try:
|
||
rows = session.query(TrendFieldMapping).filter(TrendFieldMapping.is_active == True).order_by(TrendFieldMapping.sort_order).all()
|
||
if rows:
|
||
_cached_trend_domain_map = {r.trend_keyword: r.field_name for r in rows}
|
||
logger.info(f"从DB加载 {len(_cached_trend_domain_map)} 条 trend_domain_map")
|
||
return _cached_trend_domain_map
|
||
finally:
|
||
session.close()
|
||
except Exception as e:
|
||
logger.warning(f"从DB加载 trend_domain_map 失败: {e}")
|
||
_cached_trend_domain_map = DEFAULT_TREND_DOMAIN_MAP
|
||
return _cached_trend_domain_map
|
||
|
||
def get_trend_domain_map():
|
||
return _load_trend_domain_map()
|
||
|
||
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_map = get_trend_domain_map()
|
||
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 []
|
||
gaps_text = chr(10).join(
|
||
f'- {g["topic"]}（{g.get("domain","")}）— {g.get("reason","")}。热搜词：{", ".join(g.get("hot_keywords", []))}）'
|
||
for g in gaps
|
||
)
|
||
prompt = get_prompt("topic_selector_gaps", gaps=gaps_text)
|
||
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")
|