feat: 数据源统一与前端预览修复
=== 后端核心 === - 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 流水线完整通过。
This commit is contained in:
lt
+106
-64
@@ -1,7 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
将 content/ideas/ 目录下的 Markdown 选题文件转换为 JSON 格式
|
||||
供 content creator 脚本使用
|
||||
将 content/ideas/ 目录下的 Markdown 选题文件转换为并导入数据库
|
||||
"""
|
||||
|
||||
import os
|
||||
@@ -12,13 +11,22 @@ 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"
|
||||
OUTPUT_FILE = DATA_DIR / "sustainability_topics.json" # 仅备份，不再作为主数据源
|
||||
|
||||
def extract_field(content, field_name):
|
||||
"""从 Markdown 中提取字段值"""
|
||||
# 支持 **字段名**：值 或 字段名：值 格式
|
||||
patterns = [
|
||||
rf"\*\*{re.escape(field_name)}\*\*\s*[:：]\s*(.+?)(?:\n|$)",
|
||||
rf"{re.escape(field_name)}\s*[:：]\s*(.+?)(?:\n|$)",
|
||||
@@ -29,28 +37,9 @@ def extract_field(content, field_name):
|
||||
return match.group(1).strip()
|
||||
return None
|
||||
|
||||
def extract_list(content, start_keyword):
|
||||
"""提取列表数据（如数据/案例）"""
|
||||
lines = content.split('\n')
|
||||
result = []
|
||||
capturing = False
|
||||
for line in lines:
|
||||
if start_keyword in line:
|
||||
capturing = True
|
||||
continue
|
||||
if capturing:
|
||||
if line.strip().startswith(('**', '#', '-', '*', '1.', '2.')):
|
||||
if re.match(r'^(#|\*\*|-|\*|\d+\.)\s', line):
|
||||
result.append(line.strip())
|
||||
elif line.strip() == '' or line.startswith('##'):
|
||||
break
|
||||
return result
|
||||
|
||||
def parse_evaluation_matrix(content):
|
||||
"""解析选题评估矩阵表格"""
|
||||
scores = {}
|
||||
lines = content.split('\n')
|
||||
in_table = False
|
||||
for line in lines:
|
||||
if '|' in line and '---' not in line and '维度' not in line:
|
||||
parts = [p.strip() for p in line.split('|')]
|
||||
@@ -69,78 +58,122 @@ def parse_evaluation_matrix(content):
|
||||
return scores
|
||||
|
||||
def md_to_topic(md_path):
|
||||
"""将单个 Markdown 文件转换为 topic 字典"""
|
||||
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, '领域')
|
||||
format_type = extract_field(content, '形式')
|
||||
word_count = extract_field(content, '预估字数')
|
||||
core_concept = extract_field(content, '核心观点')
|
||||
audience_pain = extract_field(content, '受众痛点')
|
||||
unique_angle = extract_field(content, '独特角度')
|
||||
data_cases = extract_list(content, '数据/案例')
|
||||
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, '优先级')
|
||||
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)
|
||||
|
||||
# 生成 topic ID
|
||||
topic_id = md_path.stem.split('-')[0] # 如 "001-上海阳台种菜一年.md" -> "001"
|
||||
# 生成 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}'
|
||||
|
||||
# 构建 topic 对象
|
||||
topic = {
|
||||
return {
|
||||
"id": topic_id,
|
||||
"title": title,
|
||||
"field": field or "未知",
|
||||
"format": format_type or "未指定",
|
||||
"word_count": word_count,
|
||||
"field": field,
|
||||
"format": format_type,
|
||||
"core_concept": core_concept,
|
||||
"audience_pain": audience_pain,
|
||||
"unique_angle": unique_angle,
|
||||
"data_cases": data_cases,
|
||||
"estimated_days": estimated_days,
|
||||
"priority": priority_str,
|
||||
"priority_score": priority_score if priority_score > 0 else (total_score if total_score > 0 else 5),
|
||||
"publish_date": publish_date,
|
||||
"status": status,
|
||||
"evaluation": evaluation,
|
||||
"priority_score": priority_score,
|
||||
"total_score": total_score,
|
||||
"cases": [], # 关联的案例ID列表，待填充
|
||||
"status": status,
|
||||
"cases": [],
|
||||
"source_file": md_path.name,
|
||||
"created_at": datetime.now().isoformat()
|
||||
"created_at": datetime.now().isoformat(),
|
||||
"updated_at": datetime.now().isoformat(),
|
||||
"ready_at": publish_date,
|
||||
"published_at": None,
|
||||
"compliance_score": 100,
|
||||
"platform_urls": {}
|
||||
}
|
||||
|
||||
return topic
|
||||
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():
|
||||
"""主函数：导入所有 Markdown 选题文件"""
|
||||
if not IDEAS_DIR.exists():
|
||||
print(f"错误：选题目录不存在 {IDEAS_DIR}")
|
||||
return
|
||||
|
||||
# 只导入主选题文件（格式：NNN-标题.md），排除 research/compliance 等辅助文件
|
||||
md_files = []
|
||||
for f in IDEAS_DIR.glob("*.md"):
|
||||
if f.name == "README.md":
|
||||
continue
|
||||
# 排除 research 和 compliance 文件
|
||||
if f.name.endswith('-research.md') or f.name.endswith('-compliance.md'):
|
||||
continue
|
||||
# 匹配 001-xxx.md 格式
|
||||
if re.match(r'^\d{3}-.+\.md$', f.name):
|
||||
md_files.append(f)
|
||||
|
||||
@@ -156,23 +189,32 @@ def main():
|
||||
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)
|
||||
|
||||
# 写入 JSON
|
||||
with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
|
||||
json.dump(topics, f, ensure_ascii=False, indent=2)
|
||||
print(f"\n✅ 已备份选题到 {OUTPUT_FILE}")
|
||||
|
||||
print(f"\n✅ 已导入 {len(topics)} 个选题到 {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'] != '已发布']
|
||||
print(f"📊 可用选题数: {len(ready_topics)}")
|
||||
avg_score = sum(t['total_score'] for t in ready_topics) / len(ready_topics) if ready_topics else 0
|
||||
print(f"🎯 平均评分: {avg_score:.1f}")
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user