lt
2bd3c83cf5
核心功能: - 新增 JWT 认证系统，支持管理员登录/登出 - 前后端合并为单一 FastAPI 应用 (端口 8001) - 系统概览页：6 个统计卡片，点击跳转筛选 - 选题管理页：批量操作 (刷新/创作/优化)，时间列展示 - 系统日志页：整合日志查看功能 - 用户管理页：管理员可创建/删除用户 - 移动端适配：响应式布局，底部导航栏 - 标题居中显示 技术改进: - 添加 generated_at 字段支持创作时间记录 - 状态更新时自动同步 updated_at - 所有 API 路由添加 JWT 认证保护 - 前端 authFetch 封装自动附加 Token - 升级 FastAPI 0.136, Pydantic 2.13 等依赖 修复: - 修复 API 500 错误 (数据库列缺失) - 修复 formatRelativeTime 未定义错误 - 修复登录 Token 存储和自动附加逻辑
67 lines
2.7 KiB
Python
67 lines
2.7 KiB
Python
import json
|
|
from datetime import datetime, date
|
|
from pathlib import Path
|
|
from sqlalchemy.orm import Session
|
|
from ..database import SessionLocal
|
|
from ..models import Topic
|
|
import os
|
|
|
|
# 计算项目根目录（从本文件位置上升4层）
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
if os.getenv('PROJECT_ROOT'):
|
|
PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT'))
|
|
TOPICS_FILE = PROJECT_ROOT / "automation" / "data" / "sustainability_topics.json"
|
|
|
|
def sync_topic_to_db(topic_id: str, db: Session = None) -> Topic:
|
|
topics = json.loads(TOPICS_FILE.read_text(encoding='utf-8'))
|
|
topic_data = next((t for t in topics if t['id'] == topic_id), None)
|
|
if not topic_data:
|
|
raise ValueError(f"Topic {topic_id} not found in file")
|
|
|
|
close_db = False
|
|
if db is None:
|
|
db = SessionLocal()
|
|
close_db = True
|
|
try:
|
|
db_topic = db.query(Topic).filter(Topic.id == topic_id).first()
|
|
if db_topic is None:
|
|
db_topic = Topic(
|
|
id=topic_data['id'],
|
|
title=topic_data['title'],
|
|
field=topic_data['field'],
|
|
format=topic_data.get('format'),
|
|
core_concept=topic_data.get('core_concept'),
|
|
audience_pain=topic_data.get('audience_pain'),
|
|
unique_angle=topic_data.get('unique_angle'),
|
|
priority=topic_data.get('priority'),
|
|
priority_score=topic_data.get('priority_score', 0),
|
|
total_score=topic_data.get('total_score')
|
|
)
|
|
db.add(db_topic)
|
|
db_topic.status = topic_data.get('status', db_topic.status)
|
|
db_topic.ready_at = datetime.strptime(topic_data['ready_at'], '%Y-%m-%d').date() if topic_data.get('ready_at') else None
|
|
db_topic.published_at = datetime.strptime(topic_data['published_at'], '%Y-%m-%d').date() if topic_data.get('published_at') else None
|
|
db_topic.compliance_score = topic_data.get('compliance_score', db_topic.compliance_score)
|
|
db_topic.platform_urls = topic_data.get('platform_urls', {})
|
|
db_topic.generated_at = datetime.now() if db_topic.generated_at is None and topic_data.get("status") in ["ready", "published"] else db_topic.generated_at
|
|
db_topic.updated_at = datetime.now()
|
|
db.commit()
|
|
db.refresh(db_topic)
|
|
return db_topic
|
|
finally:
|
|
if close_db:
|
|
db.close()
|
|
|
|
def sync_all_topics():
|
|
db = SessionLocal()
|
|
try:
|
|
topics = json.loads(TOPICS_FILE.read_text(encoding='utf-8'))
|
|
for t in topics:
|
|
sync_topic_to_db(t['id'], db)
|
|
print(f"✅ 同步 {len(topics)} 个选题到数据库")
|
|
finally:
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
sync_all_topics()
|