Yuzhiran Dev
9c37c9a574
Phase 4: org_id 注入 JWT/API 过滤/组织管理 CRUD/前端组织列 测试: tests/test_phase_upgrades.py 97项全覆盖 CSS: theme-modern.css 共享 mobile-card-list/status-dot/search-bar 等模式 修复: initial_data.py LLM配置 NOT NULL 约束, TopicResponse 含 org_id
193 lines
7.6 KiB
Python
193 lines
7.6 KiB
Python
#!/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个真正有价值的洞察。每条需包含这个发现对读者意味着什么，以及支撑数据。**所有数据必须是2025-2026年最新数据，禁用过时数据**
|
||
2. SEO关键词建议：重点布局哪些搜索词（3-5个，含1-2个长尾词）
|
||
3. 讨论点：哪个观点最有争议或最可能引发讨论？
|
||
4. 待验证：指出1-2个不确定方向
|
||
|
||
风格：说人话，直击要点。避免「首先其次最后」「综上所述」。直接输出内容，不要输出思考过程。"""
|
||
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()
|