from fastapi import APIRouter, Depends, HTTPException, Query, Request from sqlalchemy.orm import Session, joinedload from sqlalchemy import func from typing import List, Optional, Dict, Any from datetime import datetime, date from pathlib import Path from ..database import get_db from ..models import Topic, TopicField, TopicConfigField, Article, ContentMetrics from ..schemas import ( TopicCreate, TopicUpdate, TopicResponse, TopicScoreRequest, TopicAnalyzeRequest, TopicAnalyzeResponse, ) from ..core.nvidia_client import call_llm from .auth import get_current_user, org_filter router = APIRouter(prefix="/api/topics", tags=["topics"], dependencies=[Depends(get_current_user)]) PROJECT_ROOT = Path(__file__).parent.parent.parent def _check_org(topic: Topic, current_user, db: Session): """Verify topic belongs to user's org (unless admin).""" if current_user.role != "admin" and topic.org_id != current_user.org_id: raise HTTPException(status_code=404, detail="Topic not found") return topic @router.get("", response_model=List[TopicResponse]) def list_topics( field_id: Optional[int] = None, status: Optional[str] = None, today: Optional[bool] = None, tag: Optional[str] = None, search: Optional[str] = None, sort_by: str = Query("priority_score", enum=["priority_score", "created_at", "title", "updated_at"]), order: str = Query("desc", enum=["asc", "desc"]), limit: int = Query(50, le=200), offset: int = Query(0, ge=0), db: Session = Depends(get_db), current_user=Depends(get_current_user) ): db.expire_all() query = db.query(Topic).options(joinedload(Topic.field)) of = org_filter(current_user, Topic) if of is not True: query = query.filter(of) if field_id: query = query.filter(Topic.field_id == field_id) if status: query = query.filter(Topic.status == status) if today: query = query.filter(func.date(Topic.created_at) == date.today()) if tag: query = query.filter(Topic.tags.contains([tag])) if search: query = query.filter(Topic.title.contains(search)) sort_col = getattr(Topic, sort_by, Topic.priority_score) if order == "desc": query = query.order_by(sort_col.desc()) else: query = query.order_by(sort_col.asc()) return query.offset(offset).limit(limit).all() @router.get("/stats") def topic_stats( db: Session = Depends(get_db), current_user=Depends(get_current_user) ): base_q = db.query(Topic) of = org_filter(current_user, Topic) if of is not True: base_q = base_q.filter(of) total = base_q.count() raw = base_q.with_entities(Topic.status, func.count()).group_by(Topic.status).all() by_status = {s: c for s, c in raw} today = date.today() today_count = base_q.filter(func.date(Topic.created_at) == today).count() published = base_q.filter(Topic.status.in_(["published", "已发布"])).count() metrics_count = db.query(ContentMetrics).count() return { "total": total, "by_status": by_status, "published": published, "today_created": today_count, "metrics_count": metrics_count } @router.post("/analyze", response_model=TopicAnalyzeResponse) def analyze_topic( data: TopicAnalyzeRequest, db: Session = Depends(get_db), current_user=Depends(get_current_user) ): """AI 分析用户输入的选题思路,返回结构化的选题信息""" import json, re from sqlalchemy import text as sa_text refs = "\n".join(f"- {link}" for link in data.reference_links) if data.reference_links else "无" prompt_text = f"""你是一个专业的内容策略师。用户提供了一个选题思路,请分析并提炼为结构化的选题信息。 用户输入的原始内容: {data.raw_input} 参考链接(如有): {refs} 请输出 JSON(只输出 JSON,不要其他文字): {{ "title": "优化后的选题标题(20字内,含核心关键词,有吸引力)", "format": "内容形式(趋势洞察/实操指南/对比分析/案例解读/观点讨论)", "core_concept": "核心观点(一句话说清独特价值,20字内)", "audience_pain": "受众痛点(目标读者的真实困惑或需求,20字内)", "unique_angle": "差异化切入点(与常见文章不同的视角,20字内)", "tags": ["标签1", "标签2", "标签3", "标签4", "标签5"] }}""" try: resp = call_llm(prompt_text, temperature=0.6, max_tokens=1500) except Exception as e: raise HTTPException(status_code=503, detail=f"AI 分析失败: {str(e)}") # Parse JSON from response resp = resp.strip() # Remove markdown code fences if present resp = re.sub(r'^```(?:json)?\s*', '', resp) resp = re.sub(r'\s*```$', '', resp) parsed = json.loads(resp) field_name = None if data.field_id: field = db.query(TopicField).filter(TopicField.id == data.field_id).first() if field: field_name = field.name return TopicAnalyzeResponse( title=parsed.get("title", ""), format=parsed.get("format", ""), core_concept=parsed.get("core_concept", ""), audience_pain=parsed.get("audience_pain", ""), unique_angle=parsed.get("unique_angle", ""), field_name=field_name, tags=parsed.get("tags", []), priority="中", ) @router.post("", response_model=TopicResponse) def create_topic( data: TopicCreate, db: Session = Depends(get_db), current_user=Depends(get_current_user) ): topic_id = data.id if not topic_id: max_topic = db.query(Topic).order_by(Topic.id.desc()).first() if max_topic and max_topic.id.startswith("T"): try: num = int(max_topic.id[1:]) + 1 topic_id = f"T{num:03d}" except (ValueError, TypeError): topic_id = f"T{datetime.now().strftime('%m%d%H%M')}" else: topic_id = f"T{datetime.now().strftime('%m%d%H%M')}" field_name = "未分类" if data.field_id: field = db.query(TopicField).filter(TopicField.id == data.field_id).first() if field: field_name = field.name topic = Topic( id=topic_id, field_id=data.field_id, field_name=field_name, org_id=current_user.org_id or "default", title=data.title, format=data.format, core_concept=data.core_concept, audience_pain=data.audience_pain, unique_angle=data.unique_angle, priority=data.priority, status="pending", tags=data.tags or [], custom_data=data.custom_data or {}, scoring_data=data.scoring_data or {}, ) db.add(topic) db.commit() db.refresh(topic) return topic @router.get("/{topic_id}", response_model=TopicResponse) def get_topic(topic_id: str, db: Session = Depends(get_db), current_user=Depends(get_current_user)): topic = db.query(Topic).options(joinedload(Topic.field)).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) return topic @router.put("/{topic_id}", response_model=TopicResponse) def update_topic( topic_id: str, data: TopicUpdate, db: Session = Depends(get_db), current_user=Depends(get_current_user) ): topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) if data.field_id is not None: topic.field_id = data.field_id if data.field_id: field = db.query(TopicField).filter(TopicField.id == data.field_id).first() topic.field_name = field.name if field else '未分类' for k, v in data.model_dump(exclude_unset=True, exclude={"field_id"}).items(): if k == "tags" or k == "custom_data" or k == "scoring_data": if v is not None: setattr(topic, k, v) elif v is not None: setattr(topic, k, v) topic.updated_at = datetime.now() db.commit() db.refresh(topic) return topic @router.delete("/{topic_id}") def delete_topic( topic_id: str, db: Session = Depends(get_db), current_user=Depends(get_current_user) ): topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) from ..models import ContentTask, Article, PublishRecord, ContentMetrics, ContentCalendar try: db.query(ContentCalendar).filter(ContentCalendar.topic_id == topic_id).delete() db.query(ContentTask).filter(ContentTask.topic_id == topic_id).delete() db.query(Article).filter(Article.topic_id == topic_id).delete() db.query(PublishRecord).filter(PublishRecord.topic_id == topic_id).delete() db.query(ContentMetrics).filter(ContentMetrics.topic_id == topic_id).delete() db.delete(topic) db.commit() return {"ok": True} except Exception as e: db.rollback() raise HTTPException(status_code=400, detail=f"删除失败:{str(e)}") @router.post("/{topic_id}/score") def score_topic( topic_id: str, data: TopicScoreRequest, db: Session = Depends(get_db), current_user=Depends(get_current_user) ): topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) topic.scoring_data = data.scoring_data if data.scoring_data and topic.field_id: configs = db.query(TopicConfigField).filter( TopicConfigField.field_id == topic.field_id ).all() total_weight = 0 weighted_sum = 0 for cfg in configs: val = data.scoring_data.get(cfg.key) if val is not None and cfg.field_type == "number": if cfg.min_value is not None: val = max(val, cfg.min_value) if cfg.max_value is not None: val = min(val, cfg.max_value) normalized = (val - cfg.min_value) / (cfg.max_value - cfg.min_value) if cfg.max_value != cfg.min_value else 0.5 weighted_sum += normalized * cfg.weight total_weight += cfg.weight if total_weight > 0: topic.total_score = round(weighted_sum / total_weight * 100, 1) topic.priority_score = int(topic.total_score) topic.updated_at = datetime.now() db.commit() db.refresh(topic) return {"priority_score": topic.priority_score, "total_score": topic.total_score} @router.get("/{topic_id}/articles", response_model=List[Dict[str, Any]]) def get_topic_articles(topic_id: str, db: Session = Depends(get_db), current_user=Depends(get_current_user)): topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) articles = db.query(Article).filter(Article.topic_id == topic_id).all() return [a.to_dict() if hasattr(a, 'to_dict') else { "id": a.id, "topic_id": a.topic_id, "platform": a.platform, "status": a.status, "file_path": a.file_path } for a in articles] @router.get("/{topic_id}/metrics", response_model=List[Dict[str, Any]]) def get_topic_metrics(topic_id: str, db: Session = Depends(get_db), current_user=Depends(get_current_user)): topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) metrics = db.query(ContentMetrics).filter(ContentMetrics.topic_id == topic_id).all() return [m.to_dict() for m in metrics] @router.post("/{topic_id}/lock") def lock_topic(topic_id: str, db: Session = Depends(get_db), current_user=Depends(get_current_user)): topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) topic.lock_by = current_user.username topic.lock_at = datetime.now() db.commit() return {"ok": True} @router.post("/{topic_id}/unlock") def unlock_topic(topic_id: str, db: Session = Depends(get_db), current_user=Depends(get_current_user)): topic = db.query(Topic).filter(Topic.id == topic_id).first() if not topic: raise HTTPException(status_code=404, detail="Topic not found") _check_org(topic, current_user, db) topic.lock_by = None topic.lock_at = None db.commit() return {"ok": True} @router.post("/batch-update-status") def batch_update_status( topic_ids: List[str], status: str, db: Session = Depends(get_db), current_user=Depends(get_current_user) ): q = db.query(Topic).filter(Topic.id.in_(topic_ids)) of = org_filter(current_user, Topic) if of is not True: q = q.filter(of) now = datetime.now() update_dict = {Topic.status: status, Topic.updated_at: now} if status in ('review', '待审查'): update_dict[Topic.generated_at] = now update_dict[Topic.reviewed_at] = now elif status in ('ready', '待发布'): update_dict[Topic.ready_at] = now.date() update_dict[Topic.reviewed_at] = now elif status in ('published', '已发布'): update_dict[Topic.published_at] = now.date() updated = q.update(update_dict, synchronize_session=False) db.commit() return {"ok": True, "updated": updated} @router.get("/field-distribution") def field_distribution( db: Session = Depends(get_db), current_user=Depends(get_current_user) ): q = db.query(Topic.field_name, func.count(Topic.id).label("count")) of = org_filter(current_user, Topic) if of is not True: q = q.filter(of) rows = q.group_by(Topic.field_name).all() return [{"field": r.field_name or "未分类", "count": r.count} for r in rows]