#!/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与效率", "家庭教育": "科技人文交叉", } _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")