#!/usr/bin/env python3 """ 将 content/ideas/ 目录下的 Markdown 选题文件转换为并导入数据库 """ import os import sys import json import re from pathlib import Path from datetime import datetime PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT)) # 数据库导入 try: from app.database import SessionLocal from app.models import Topic as DBTopic HAVE_DB = True except ImportError as e: HAVE_DB = False print(f"[Warning] Database import failed: {e}") IDEAS_DIR = PROJECT_ROOT / "content" / "ideas" DATA_DIR = PROJECT_ROOT / "automation" / "data" OUTPUT_FILE = DATA_DIR / "sustainability_topics.json" # 仅备份，不再作为主数据源 def extract_field(content, field_name): patterns = [ rf"\*\*{re.escape(field_name)}\*\*\s*[:：]\s*(.+?)(?:\n|$)", rf"{re.escape(field_name)}\s*[:：]\s*(.+?)(?:\n|$)", ] for pattern in patterns: match = re.search(pattern, content, re.MULTILINE) if match: return match.group(1).strip() return None def parse_evaluation_matrix(content): scores = {} lines = content.split('\n') for line in lines: if '|' in line and '---' not in line and '维度' not in line: parts = [p.strip() for p in line.split('|')] if len(parts) >= 3: dimension = parts[1] score_str = parts[2] try: score = int(score_str) scores[dimension] = score except: pass if '**总分**' in line: total_match = re.search(r'\*\*总分\*\*\s*\|\s*\*\*(\d+)\*\*', line) if total_match: scores['总分'] = int(total_match.group(1)) return scores def md_to_topic(md_path): with open(md_path, 'r', encoding='utf-8') as f: content = f.read() title_match = re.search(r'^#\s+(.+)$', content, re.MULTILINE) title = title_match.group(1).strip() if title_match else md_path.stem field = extract_field(content, '领域') or '可持续生活系统' format_type = extract_field(content, '形式') or '趋势洞察 + 实操指南' core_concept = extract_field(content, '核心观点') or '' audience_pain = extract_field(content, '受众痛点') or '' unique_angle = extract_field(content, '独特角度') or '' estimated_days = extract_field(content, '预估完成时间') priority_str = extract_field(content, '优先级') or '中' publish_date = extract_field(content, '预计发布时间') status = extract_field(content, '状态') or '待处理' priority_map = {'高': 10, '中': 7, '低': 4} priority_score = priority_map.get(priority_str, 5) evaluation = parse_evaluation_matrix(content) total_score = evaluation.get('总分', 0) # 生成 ID：从文件名提取前缀数字，如果没有则使用标题哈希 stem = md_path.stem # e.g., "001-上海阳台种菜一年" m = re.match(r'^(\d{3})', stem) if m: num = m.group(1) topic_id = f'M{num}' # M 系列表示手动导入 else: import hashlib short = hashlib.md5(title.encode()).hexdigest()[:6].upper() topic_id = f'M{short}' return { "id": topic_id, "title": title, "field": field, "format": format_type, "core_concept": core_concept, "audience_pain": audience_pain, "unique_angle": unique_angle, "priority": priority_str, "priority_score": priority_score, "total_score": total_score, "status": status, "cases": [], "source_file": md_path.name, "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), "ready_at": publish_date, "published_at": None, "compliance_score": 100, "platform_urls": {} } def save_to_db(topic_dict): if not HAVE_DB: print("数据库不可用，跳过入库") return False db = SessionLocal() try: existing = db.query(DBTopic).filter(DBTopic.id == topic_dict['id']).first() if existing: # 更新字段 for field in ['title', 'field', 'format', 'core_concept', 'audience_pain', 'unique_angle', 'priority', 'priority_score', 'total_score', 'status', 'cases', 'source_file', 'compliance_score', 'platform_urls']: setattr(existing, field, topic_dict.get(field, getattr(existing, field))) if topic_dict.get('ready_at'): try: existing.ready_at = datetime.strptime(topic_dict['ready_at'], '%Y-%m-%d').date() except: pass existing.updated_at = datetime.now() else: # 新增 new_topic = DBTopic( id=topic_dict['id'], title=topic_dict['title'], field=topic_dict['field'], format=topic_dict['format'], core_concept=topic_dict['core_concept'], audience_pain=topic_dict['audience_pain'], unique_angle=topic_dict['unique_angle'], priority=topic_dict['priority'], priority_score=topic_dict['priority_score'], total_score=topic_dict['total_score'], status=topic_dict['status'], cases=topic_dict['cases'], source_file=topic_dict['source_file'], ready_at=datetime.strptime(topic_dict['ready_at'], '%Y-%m-%d').date() if topic_dict.get('ready_at') else None, published_at=None, compliance_score=topic_dict['compliance_score'], platform_urls=topic_dict['platform_urls'], created_at=datetime.now(), updated_at=datetime.now() ) db.add(new_topic) db.commit() return True except Exception as e: db.rollback() print(f"数据库保存失败: {e}") return False finally: db.close() def main(): if not IDEAS_DIR.exists(): print(f"错误：选题目录不存在 {IDEAS_DIR}") return md_files = [] for f in IDEAS_DIR.glob("*.md"): if f.name == "README.md": continue if f.name.endswith('-research.md') or f.name.endswith('-compliance.md'): continue if re.match(r'^\d{3}-.+\.md$', f.name): md_files.append(f) if not md_files: print("未找到选题文件") return print(f"找到 {len(md_files)} 个选题文件，开始导入...") topics = [] for md_file in sorted(md_files): print(f" 处理: {md_file.name}") topic = md_to_topic(md_file) topics.append(topic) print(f" 标题: {topic['title']}") print(f" ID: {topic['id']}") print(f" 总分: {topic['total_score']}") print(f" 状态: {topic['status']}") # 保存 JSON 备份 DATA_DIR.mkdir(parents=True, exist_ok=True) with open(OUTPUT_FILE, 'w', encoding='utf-8') as f: json.dump(topics, f, ensure_ascii=False, indent=2) print(f"\n✅ 已备份选题到 {OUTPUT_FILE}") # 导入数据库 if HAVE_DB: success_count = 0 for t in topics: if save_to_db(t): success_count += 1 print(f"✅ 已导入 {success_count}/{len(topics)} 个选题到数据库") else: print("⚠️ 数据库不可用，仅生成了 JSON 备份") # 统计 ready_topics = [t for t in topics if t['status'] != '已发布'] if ready_topics: avg_score = sum(t['total_score'] for t in ready_topics) / len(ready_topics) print(f"📊 可用选题数: {len(ready_topics)}") print(f"🎯 平均评分: {avg_score:.1f}") if __name__ == "__main__": main()