lt
31d6306e3b
=== 后端核心 === - db_helper: 统一数据库访问抽象层 - system.py API: * 参数绑定修复: 使用 Body(embed=True) 接收 JSON * 添加请求日志记录 - sync.py: 仅导出 DB→JSON（备份） === 合规与流水线 === - compliance_checker: 标签检测优化（仅检查容器，避免正文误判） - 所有脚本（creator/collector/writer/outline/research等）统一使用数据库 === 前端改版 === - topics.html: * 创作/优化 API 路径修正 * 预览弹窗重设计：多平台并行加载、富文本显示、单复制按钮 * 状态中文映射（getStatusLabel） * 认证检查 - 所有 HTML 静态资源路径修复（移除 /static 前缀） === 数据一致性 === - 数据库状态统一为英文（pending/review/ready/published） - 前端显示中文化映射 已测试 A03 流水线完整通过。
125 lines
4.6 KiB
Python
125 lines
4.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))
|
|
|
|
# 导入数据库辅助模块
|
|
from db_helper import get_topic_by_id
|
|
|
|
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:
|
|
# 日期过滤：仅保留 2025 年及以后（支持 YYYY-MM-DD 或 YYYY 格式）
|
|
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 generate_notes(self) -> str:
|
|
"""生成研究笔记 Markdown"""
|
|
cases = self.find_relevant_cases()
|
|
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## 相关案例（{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', '无')}",
|
|
""
|
|
])
|
|
lines.extend([
|
|
"## 研究发现摘要",
|
|
"- 待补充：从案例中提炼的趋势和洞察",
|
|
"- 待补充：数据支撑",
|
|
"",
|
|
"## 待深入研究的问题",
|
|
"- [ ] 需要更多本土数据",
|
|
"- [ ] 需要验证某些结论的适用性",
|
|
"",
|
|
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()
|