lt
56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
import json
|
||
import os
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from .database import SessionLocal, init_db
|
||
from .models import Topic
|
||
|
||
# 计算项目根目录（backend/app/initial_data.py -> 上升3层到 yu-zhi-ran）
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
||
if os.getenv('PROJECT_ROOT'):
|
||
PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT'))
|
||
TOPICS_FILE = PROJECT_ROOT / "automation" / "data" / "sustainability_topics.json"
|
||
|
||
def import_topics_from_json():
|
||
db = SessionLocal()
|
||
try:
|
||
if db.query(Topic).count() > 0:
|
||
print("数据库已有数据，跳过导入")
|
||
return
|
||
if not __import__('os').path.exists(TOPICS_FILE):
|
||
print(f"选题文件不存在: {TOPICS_FILE}")
|
||
return
|
||
topics = json.loads(open(TOPICS_FILE, encoding='utf-8').read())
|
||
for t in topics:
|
||
topic = Topic(
|
||
id=t['id'],
|
||
title=t['title'],
|
||
field=t['field'],
|
||
format=t.get('format'),
|
||
core_concept=t.get('core_concept'),
|
||
audience_pain=t.get('audience_pain'),
|
||
unique_angle=t.get('unique_angle'),
|
||
priority=t.get('priority'),
|
||
priority_score=t.get('priority_score', 0),
|
||
total_score=t.get('total_score'),
|
||
status=t.get('status', 'pending'),
|
||
cases=t.get('cases', []),
|
||
source_file=t.get('source_file'),
|
||
ready_at=datetime.strptime(t['ready_at'], '%Y-%m-%d').date() if t.get('ready_at') else None,
|
||
published_at=datetime.strptime(t['published_at'], '%Y-%m-%d').date() if t.get('published_at') else None,
|
||
compliance_score=t.get('compliance_score'),
|
||
platform_urls=t.get('platform_urls', {})
|
||
)
|
||
db.add(topic)
|
||
db.commit()
|
||
print(f"✅ 导入 {len(topics)} 个选题到数据库")
|
||
except Exception as e:
|
||
print(f"导入失败: {e}")
|
||
db.rollback()
|
||
finally:
|
||
db.close()
|
||
|
||
if __name__ == "__main__":
|
||
init_db()
|
||
import_topics_from_json()
|