yu-zhi-ran/platform/backend/app/api/topics.py

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,
    series: 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 series:
        query = query.filter(Topic.series == series)
    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]