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 流水线完整通过。
221 lines
7.7 KiB
Python
221 lines
7.7 KiB
Python
#!/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()
|