Files
yu-zhi-ran/scripts/import_topics.py
T

179 lines
6.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
将 content/ideas/ 目录下的 Markdown 选题文件转换为 JSON 格式
供 content creator 脚本使用
"""
import os
import sys
import json
import re
from pathlib import Path
from datetime import datetime
PROJECT_ROOT = Path(__file__).parent.parent
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):
"""从 Markdown 中提取字段值"""
# 支持 **字段名**:值 或 字段名:值 格式
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 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('|')]
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):
"""将单个 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, '数据/案例')
estimated_days = extract_field(content, '预估完成时间')
priority_str = extract_field(content, '优先级')
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"
# 构建 topic 对象
topic = {
"id": topic_id,
"title": title,
"field": field or "未知",
"format": format_type or "未指定",
"word_count": word_count,
"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,
"total_score": total_score,
"cases": [], # 关联的案例ID列表,待填充
"source_file": md_path.name,
"created_at": datetime.now().isoformat()
}
return topic
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)
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" 总分: {topic['total_score']}")
print(f" 状态: {topic['status']}")
# 确保输出目录存在
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✅ 已导入 {len(topics)} 个选题到 {OUTPUT_FILE}")
# 统计
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 __name__ == "__main__":
main()