diff --git a/automation/data/sustainability_topics.json b/automation/data/sustainability_topics.json index ce080cc..ac0721f 100644 --- a/automation/data/sustainability_topics.json +++ b/automation/data/sustainability_topics.json @@ -260,5 +260,22 @@ "priority_score": 90, "status": "待处理", "cases": [] + }, + { + "id": "A03", + "title": "数字游民签证全解析:30个国家政策对比,中国护照能去哪些?", + "field": "未来工作方式", + "format": "对比分析 + 实操指南", + "core_concept": "分析爱沙尼亚/葡萄牙/巴厘岛等30国游民签证,结合中国护照限制,给出签证+保险+税务+社群的完整路线", + "audience_pain": "想地理套利但被签证和社保困扰", + "unique_angle": "不是简单列出签证,而是给出中国护照持有者的可行组合方案(如泰国+大马+巴厘岛)", + "priority": "高", + "priority_score": 90, + "status": "ready", + "cases": [], + "lock_by": null, + "lock_at": null, + "ready_at": "2026-05-07", + "compliance_score": 100 } ] \ No newline at end of file diff --git a/platform/backend/app/api/system.py b/platform/backend/app/api/system.py index 77d78ec..5795ac9 100644 --- a/platform/backend/app/api/system.py +++ b/platform/backend/app/api/system.py @@ -1,63 +1,65 @@ -from fastapi import APIRouter, HTTPException, Depends +import logging +from fastapi import APIRouter, HTTPException, Depends, Body from sqlalchemy.orm import Session from sqlalchemy import func -from datetime import datetime, date, timedelta +from datetime import datetime, date from typing import Dict, Any, List, Optional from pathlib import Path -from .auth import get_current_user import os import json from ..database import get_db from ..models import Topic, Article -from ..schemas import SystemStatus from ..core.generator import run_creator from ..core.optimizer import run_optimizer from ..core.sync import sync_all_topics from ..core.scheduler import scheduler +from .auth import get_current_user PROJECT_ROOT = Path(__file__).resolve().parents[4] if os.getenv('PROJECT_ROOT'): PROJECT_ROOT = Path(os.getenv('PROJECT_ROOT')) LOGS_DIR = PROJECT_ROOT / "automation" / "logs" -DATA_DIR = PROJECT_ROOT / "automation" / "data" +logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/system", tags=["system"]) +def _aggregate_status_counts(db: Session): + """聚合状态计数,兼容中英文状态值""" + raw = db.query(Topic.status, func.count()).group_by(Topic.status).all() + mapping = { + 'pending': ['pending', '待处理'], + 'review': ['review', '待审查'], + 'ready': ['ready', '待发布'], + 'published': ['published', '已发布'] + } + counts = {'pending': 0, 'review': 0, 'ready': 0, 'published': 0} + for status_val, cnt in raw: + for key, aliases in mapping.items(): + if status_val in aliases: + counts[key] += cnt + break + return counts + @router.get("/status") def get_status(db: Session = Depends(get_db)): - """系统状态概览 - 返回前端兼容格式""" total = db.query(Topic).count() - by_status_result = db.query(Topic.status, func.count()).group_by(Topic.status).all() - by_status = {status: count for status, count in by_status_result} - - # 确保返回所有状态(数据库存中文,返回前端需要英文) - status_map = { - 'pending': by_status.get('待处理', 0), - 'review': by_status.get('待审查', 0), - 'ready': by_status.get('待发布', 0), - 'published': by_status.get('已发布', 0) - } - - # 计算今日新增 + counts = _aggregate_status_counts(db) today = date.today() - today_count = db.query(Topic).filter( - func.date(Topic.created_at) == today - ).count() - + today_count = db.query(Topic).filter(func.date(Topic.created_at) == today).count() return { "stats": { "total": total, - "pending": status_map['pending'], - "review": status_map['review'], - "ready": status_map['ready'], - "published": status_map['published'], + "pending": counts['pending'], + "review": counts['review'], + "ready": counts['ready'], + "published": counts['published'], "today": today_count } } @router.post("/generate/run", dependencies=[Depends(get_current_user)]) -def trigger_generation(topic_id: str = None, db: Session = Depends(get_db)): - """手动触发内容创作任务""" +def trigger_generation(topic_id: str = Body(None, embed=True), db: Session = Depends(get_db)): + logger.info(f"Received topic_id={topic_id}") try: result = run_creator(topic_id) if not result["ok"]: @@ -68,8 +70,7 @@ def trigger_generation(topic_id: str = None, db: Session = Depends(get_db)): raise HTTPException(status_code=500, detail=str(e)) @router.post("/optimize/run", dependencies=[Depends(get_current_user)]) -def trigger_optimization(topic_ids: List[str] = None, db: Session = Depends(get_db)): - """手动触发合规优化任务""" +def trigger_optimization(topic_ids: List[str] = Body(None, embed=True), db: Session = Depends(get_db)): try: result = run_optimizer(topic_ids) if not result["ok"]: @@ -85,7 +86,6 @@ def trigger_optimization(topic_ids: List[str] = None, db: Session = Depends(get_ @router.get("/logs/{log_date}", dependencies=[Depends(get_current_user)]) def get_logs(log_date: str, log_type: str = "creator"): - """读取日志文件内容""" log_file = LOGS_DIR / f"{log_type}_{log_date}.log" if not log_file.exists(): raise HTTPException(status_code=404, detail=f"Log file not found: {log_file}") @@ -94,59 +94,56 @@ def get_logs(log_date: str, log_type: str = "creator"): return {"log_date": log_date, "log_type": log_type, "content": lines} @router.get("/pipeline/status", dependencies=[Depends(get_current_user)]) -def get_pipeline_status(): - """获取流水线各模块状态""" - try: - topics_file = DATA_DIR / "sustainability_topics.json" - topics = [] - if topics_file.exists(): - topics = json.loads(topics_file.read_text(encoding='utf-8')) - status_counts = {} - for t in topics: - s = t.get('status', 'unknown') - status_counts[s] = status_counts.get(s, 0) + 1 - log_files = { - "collector": LOGS_DIR / f"collector_{date.today().isoformat()}.log", - "creator": LOGS_DIR / f"creator_{date.today().isoformat()}.log", - "optimizer": LOGS_DIR / f"optimizer_{date.today().isoformat()}.log", - } - pipeline_status = {} - for name, log_file in log_files.items(): - if log_file.exists(): - mtime = datetime.fromtimestamp(log_file.stat().st_mtime) - pipeline_status[name] = {"last_run": mtime.isoformat(), "exists": True} - else: - pipeline_status[name] = {"exists": False, "last_run": None} - return {"topics_count": len(topics), "status_distribution": status_counts, "pipeline_modules": pipeline_status} - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) +def get_pipeline_status(db: Session = Depends(get_db)): + total = db.query(Topic).count() + counts = _aggregate_status_counts(db) + log_files = { + "collector": LOGS_DIR / f"collector_{date.today().isoformat()}.log", + "creator": LOGS_DIR / f"creator_{date.today().isoformat()}.log", + "optimizer": LOGS_DIR / f"optimizer_{date.today().isoformat()}.log", + } + pipeline_status = {} + for name, log_file in log_files.items(): + if log_file.exists(): + mtime = datetime.fromtimestamp(log_file.stat().st_mtime) + pipeline_status[name] = {"last_run": mtime.isoformat(), "exists": True} + else: + pipeline_status[name] = {"exists": False, "last_run": None} + return {"topics_count": total, "status_distribution": counts, "pipeline_modules": pipeline_status} @router.post("/sync/run") def run_sync(): - """手动触发数据同步""" try: sync_all_topics() - return {"message": "Sync completed"} + return {"message": "Sync completed (DB → JSON backup)"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get("/automation/topics") -def list_automation_topics(): - """直接读取自动化流水线的选题 JSON""" +def list_automation_topics(db: Session = Depends(get_db)): try: - topics_file = DATA_DIR / "sustainability_topics.json" - if not topics_file.exists(): - raise HTTPException(status_code=404, detail="Topics JSON not found") - topics = json.loads(topics_file.read_text(encoding='utf-8')) - return {"count": len(topics), "topics": topics[-50:]} - except json.JSONDecodeError as e: - raise HTTPException(status_code=500, detail=f"JSON parse error: {e}") + topics = db.query(Topic).order_by(Topic.created_at.desc()).limit(100).all() + result = [] + for t in topics: + result.append({ + "id": t.id, + "title": t.title, + "field": t.field, + "status": t.status, + "priority": t.priority, + "priority_score": t.priority_score, + "total_score": t.total_score, + "created_at": t.created_at.isoformat() if t.created_at else None, + "updated_at": t.updated_at.isoformat() if t.updated_at else None, + "ready_at": t.ready_at.isoformat() if t.ready_at else None, + "compliance_score": t.compliance_score + }) + return {"count": len(result), "topics": result[:50]} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.post("/refresh") def refresh_all(): - """刷新所有数据""" try: sync_all_topics() return {"message": "Refresh completed"} @@ -155,5 +152,4 @@ def refresh_all(): @router.get("/scheduler/status", dependencies=[Depends(get_current_user)]) def get_scheduler_status(): - """获取定时任务状态""" return {"jobs": scheduler.get_jobs()} diff --git a/platform/backend/app/core/sync.py b/platform/backend/app/core/sync.py index e511319..69e0fe7 100644 --- a/platform/backend/app/core/sync.py +++ b/platform/backend/app/core/sync.py @@ -13,52 +13,96 @@ if 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") - + """注意:此函数原用于将JSON单个选题同步到数据库。现已不需要,保留用于兼容。当前方向相反(DB为主),此处仅从数据库导出到JSON(如果需要)""" + # 为了不破坏旧调用,我们改为从数据库读取并写入 JSON 文件(单条更新) 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 + topic = db.query(Topic).filter(Topic.id == topic_id).first() + if not topic: + raise ValueError(f"Topic {topic_id} not found in DB") + # 写入 JSON 文件(作为备份) + try: + if TOPICS_FILE.exists(): + with open(TOPICS_FILE, 'r', encoding='utf-8') as f: + topics = json.load(f) + else: + topics = [] + # 转为字典 + tdict = { + 'id': topic.id, + 'title': topic.title, + 'field': topic.field, + 'format': topic.format, + 'core_concept': topic.core_concept, + 'audience_pain': topic.audience_pain, + 'unique_angle': topic.unique_angle, + 'priority': topic.priority, + 'priority_score': topic.priority_score, + 'total_score': topic.total_score, + 'status': topic.status, + 'cases': topic.cases or [], + 'source_file': topic.source_file, + 'created_at': topic.created_at.isoformat() if topic.created_at else None, + 'updated_at': topic.updated_at.isoformat() if topic.updated_at else None, + 'ready_at': topic.ready_at.isoformat() if topic.ready_at else None, + 'published_at': topic.published_at.isoformat() if topic.published_at else None, + 'compliance_score': topic.compliance_score, + 'platform_urls': topic.platform_urls or {} + } + # 更新或追加 + found = False + for i, t in enumerate(topics): + if t['id'] == topic_id: + topics[i] = tdict + found = True + break + if not found: + topics.append(tdict) + with open(TOPICS_FILE, 'w', encoding='utf-8') as f: + json.dump(topics, f, ensure_ascii=False, indent=2) + except Exception as e: + print(f"[Warning] JSON backup failed: {e}") + return topic finally: if close_db: db.close() def sync_all_topics(): + """导出所有选题到 JSON 文件(用于备份或兼容)""" db = SessionLocal() try: - topics = json.loads(TOPICS_FILE.read_text(encoding='utf-8')) + topics = db.query(Topic).order_by(Topic.created_at).all() + topic_list = [] for t in topics: - sync_topic_to_db(t['id'], db) - print(f"✅ 同步 {len(topics)} 个选题到数据库") + tdict = { + 'id': t.id, + 'title': t.title, + 'field': t.field, + 'format': t.format, + 'core_concept': t.core_concept, + 'audience_pain': t.audience_pain, + 'unique_angle': t.unique_angle, + 'priority': t.priority, + 'priority_score': t.priority_score, + 'total_score': t.total_score, + 'status': t.status, + 'cases': t.cases or [], + 'source_file': t.source_file, + 'created_at': t.created_at.isoformat() if t.created_at else None, + 'updated_at': t.updated_at.isoformat() if t.updated_at else None, + 'ready_at': t.ready_at.isoformat() if t.ready_at else None, + 'published_at': t.published_at.isoformat() if t.published_at else None, + 'compliance_score': t.compliance_score, + 'platform_urls': t.platform_urls or {} + } + topic_list.append(tdict) + TOPICS_FILE.parent.mkdir(parents=True, exist_ok=True) + with open(TOPICS_FILE, 'w', encoding='utf-8') as f: + json.dump(topic_list, f, ensure_ascii=False, indent=2) + print(f"✅ 导出 {len(topic_list)} 个选题到 JSON (兼容模式)") finally: db.close() diff --git a/platform/frontend/admin.html b/platform/frontend/admin.html index 648b00c..2947f51 100644 --- a/platform/frontend/admin.html +++ b/platform/frontend/admin.html @@ -4,7 +4,7 @@ 宇之然内容创作平台 - 用户管理 - +