#!/usr/bin/env python3 import json, datetime, logging, sys, re from pathlib import Path from typing import Dict, List PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT / "platform" / "backend")) from db_helper import get_topic_by_id try: from app.core.nvidia_client import call_llm HAVE_LLM = True except ImportError: HAVE_LLM = False from trends import get_trend_context from web_search import enrich_topic_research DATA_DIR = PROJECT_ROOT / "automation" / "data" CASES_FILE = DATA_DIR / "sustainability_cases.json" OUTPUT_DIR = DATA_DIR / "research" 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"research_{TODAY}.log"), logging.StreamHandler()]) logger = logging.getLogger(__name__) class Researcher: def __init__(self, topic_id: str): self.topic_id = topic_id self.topic = self._load_topic() self.cases = self._load_cases() self.output_dir = OUTPUT_DIR / TODAY self.output_dir.mkdir(parents=True, exist_ok=True) def _load_topic(self) -> Dict: topic = get_topic_by_id(self.topic_id) if not topic: raise ValueError(f"Topic {self.topic_id} not found") return topic def _load_cases(self) -> List[Dict]: if CASES_FILE.exists(): return json.loads(CASES_FILE.read_text(encoding='utf-8')) return [] def find_relevant_cases(self, top_k: int = 5) -> List[Dict]: field = self.topic.get('field', '').lower() title = self.topic.get('title', '').lower() scored = [] for case in self.cases: case_date = case.get('date', '') if case_date: m = re.search(r'(\d{4})', str(case_date)) if m and int(m.group(1)) < 2025: continue score = 0 if field and field in case.get('field', '').lower(): score += 3 case_title = case.get('title', '').lower() for word in title.split(): if len(word) > 2 and word in case_title: score += 1 if score > 0: scored.append((score, case)) scored.sort(key=lambda x: x[0], reverse=True) return [c for _, c in scored[:top_k]] def _llm_summary(self, cases: List[Dict]) -> str: if not HAVE_LLM: return "" cases_text = json.dumps(cases, ensure_ascii=False, indent=2) # 获取实时搜索数据作为 LLM 参考 search_data = enrich_topic_research(self.topic) search_section = f"\n## 实时搜索结果\n{search_data}\n" if search_data else "" prompt = f"""你是一个行业研究员+内容策略师，擅长从案例中发现真洞察，并能判断什么内容对真实读者最有价值。 基于以下选题和相关案例，写出能支撑文章核心观点、对读者真正有用的研究发现。 ## 选题 标题：{self.topic['title']} 领域：{self.topic.get('field', '')} 核心观点：{self.topic.get('core_concept', '')} 受众痛点：{self.topic.get('audience_pain', '')} 独特视角：{self.topic.get('unique_angle', '')} {search_section} ## 相关案例（{len(cases)}个） {cases_text} ## 输出要求（按顺序）： 1. 核心发现：2-3个真正有价值的洞察（不是每个案例凑一条）。每个洞察需包含： - 这个发现对读者意味着什么（不要只说事实，要说意义） - 可以用什么数据或案例支撑 2. SEO关键词建议：这篇文章应该重点布局哪些搜索词（3-5个，包含1-2个长尾词） 3. 讨论点：哪个观点最有争议或最可能引发讨论？这能帮助文章获得平台推荐 4. 待验证：指出1-2个不确定的方向，作者需进一步核实 风格：说人话，每条洞察2-3句话直击要点。避免「首先其次最后」「综上所述」。""" try: return call_llm(prompt, temperature=0.5, max_tokens=1200, system_prompt="你是一个行业研究员，擅长从案例中发现真洞察。") except Exception as e: logger.warning(f"LLM 研究发现摘要生成失败: {e}") return "" def generate_notes(self) -> str: cases = self.find_relevant_cases() trend_context = get_trend_context(self.topic.get('field')) search_data = enrich_topic_research(self.topic) lines = [ f"# 研究笔记：{self.topic['title']}", f"\n## 选题信息", f"- **ID**: {self.topic['id']}", f"- **领域**: {self.topic.get('field')}", f"- **核心观点**: {self.topic.get('core_concept', '待补充')}", f"- **受众痛点**: {self.topic.get('audience_pain', '待补充')}", f"- **独特视角**: {self.topic.get('unique_angle', '待补充')}", f"\n{trend_context}", f"\n{search_data}" if search_data else "", f"\n## 相关案例（{len(cases)}个）\n" ] for i, case in enumerate(cases, 1): lines.extend([ f"### 案例 {i}: {case.get('title')}", f"- **来源**: {case.get('source', '未知')}", f"- **日期**: {case.get('date', '未知')}", f"- **摘要**: {case.get('summary', case.get('description', '无'))}", f"- **关键数据**: {case.get('key_metrics', '无')}", "" ]) llm_summary = self._llm_summary(cases) if llm_summary: lines.extend([ "## 研究发现摘要（LLM 生成）", llm_summary, "" ]) else: insights = [] for case in cases[:3]: summary = case.get('summary', case.get('description', '')) metrics = case.get('key_metrics', '') if summary: insight = f"- {case.get('title', '相关案例')}：{summary[:100]}" if metrics: insight += f"（{metrics[:80]}）" insights.append(insight) if not insights: insights = ["- 暂未匹配到高度相关的历史案例"] pain_text = self.topic.get('audience_pain', '') topic_insights = [f"- {pain_text[:100]}"] if pain_text else [] lines.extend([ "## 研究发现摘要", "", ]) lines.extend(insights) if topic_insights: lines.extend(topic_insights) lines.extend([ "", "## 待深入研究的问题", ]) if cases: lines.append("- [ ] 验证以上案例在当前选题背景下的适用性") lines.extend([ "- [ ] 收集更多本土一手数据", "- [ ] 确认目标受众的实际反馈", "" ]) lines.append(f"*生成时间：{TODAY}*") return "\n".join(lines) def save(self): notes = self.generate_notes() out_path = self.output_dir / f"{self.topic_id}_research.md" out_path.write_text(notes, encoding='utf-8') logger.info(f"研究笔记已保存: {out_path}") return out_path def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument('--topic-id', required=True, help='选题ID') args = parser.parse_args() r = Researcher(args.topic_id) r.save() print(f"SUCCESS: Research notes created for {args.topic_id}") sys.exit(0) if __name__ == "__main__": main()