lt
df5a9db3ad
- 完成系统管理页面与导航整合 - 在 users.html/topics.html/logs.html 侧边栏与移动导航添加系统管理入口 - 创建 admin.html 管理页面（案例/任务日志/LLM配置/系统配置） - 新增核心模块：collector.py（数据采集器）、scheduler.py（任务调度器） - 新增脚本：collector_db_integration.py（采集器数据库整合） - 更新项目文档并验证路由注册
78 lines
3.5 KiB
Python
78 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
collector 数据库集成模块
|
|
将采集到的选题写入数据库
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PROJECT_ROOT = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from app.database import SessionLocal
|
|
from app.models import Topic
|
|
from datetime import datetime
|
|
|
|
def save_topics_to_db(topics_data: list):
|
|
"""将选题列表保存到数据库（插入或更新）"""
|
|
db = SessionLocal()
|
|
try:
|
|
for t in topics_data:
|
|
# 检查是否存在
|
|
existing = db.query(Topic).filter(Topic.id == t['id']).first()
|
|
if existing:
|
|
# 更新字段
|
|
existing.title = t['title']
|
|
existing.field = t.get('field', existing.field)
|
|
existing.format = t.get('format', existing.format)
|
|
existing.core_concept = t.get('core_concept', existing.core_concept)
|
|
existing.audience_pain = t.get('audience_pain', existing.audience_pain)
|
|
existing.unique_angle = t.get('unique_angle', existing.unique_angle)
|
|
existing.priority = t.get('priority', existing.priority)
|
|
existing.priority_score = t.get('priority_score', existing.priority_score)
|
|
existing.total_score = t.get('total_score', existing.total_score)
|
|
existing.status = t.get('status', existing.status)
|
|
existing.cases = t.get('cases', existing.cases)
|
|
existing.source_file = t.get('source_file', existing.source_file)
|
|
existing.ready_at = datetime.strptime(t['ready_at'], '%Y-%m-%d').date() if t.get('ready_at') else existing.ready_at
|
|
existing.published_at = datetime.strptime(t['published_at'], '%Y-%m-%d').date() if t.get('published_at') else existing.published_at
|
|
existing.compliance_score = t.get('compliance_score', existing.compliance_score)
|
|
existing.platform_urls = t.get('platform_urls', existing.platform_urls)
|
|
existing.updated_at = datetime.now()
|
|
else:
|
|
# 新增
|
|
topic = Topic(
|
|
id=t['id'],
|
|
title=t['title'],
|
|
field=t.get('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', 100),
|
|
platform_urls=t.get('platform_urls', {}),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now()
|
|
)
|
|
db.add(topic)
|
|
db.commit()
|
|
print(f"✅ 保存/更新了 {len(topics_data)} 个选题到数据库")
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise e
|
|
finally:
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
# 测试示例
|
|
print("collector_db_integration module - for import only")
|