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

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import desc, func
from typing import Optional
from datetime import datetime, timedelta, timezone

from ..database import get_db
from ..models import SearchRanking, Article, Topic, GeoReadinessScore
from .auth import get_current_user, org_filter

router = APIRouter(prefix="/api/seo", tags=["seo"])

ArticleModel = Article


def _apply_org_filter(query, current_user, db):
    of = org_filter(current_user, ArticleModel)
    if of is not True:
        query = query.join(ArticleModel, SearchRanking.article_id == ArticleModel.id, isouter=True)
        query = query.filter(of)
    return query


@router.get("/rankings")
def get_rankings(
    article_id: Optional[str] = None,
    keyword: Optional[str] = None,
    days: int = Query(30, ge=1, le=365),
    limit: int = Query(50, ge=1, le=200),
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user),
):
    """获取搜索排名数据"""
    query = db.query(SearchRanking)
    query = _apply_org_filter(query, current_user, db)

    if article_id:
        query = query.filter(SearchRanking.article_id == article_id)
    if keyword:
        query = query.filter(SearchRanking.keyword.ilike(f"%{keyword}%"))

    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    query = query.filter(SearchRanking.checked_at >= cutoff)

    rankings = query.order_by(desc(SearchRanking.checked_at)).limit(limit).all()
    return {"ok": True, "data": [r.to_dict() for r in rankings]}


@router.get("/rankings/overview")
def get_rankings_overview(
    days: int = Query(30, ge=1, le=365),
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user),
):
    """搜索排名概览统计"""
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    base = db.query(SearchRanking)
    base = _apply_org_filter(base, current_user, db)
    base = base.filter(SearchRanking.checked_at >= cutoff)

    total_checks = base.count()
    on_page = base.filter(SearchRanking.position.isnot(None)).count()
    ai_cited = base.filter(SearchRanking.ai_cited == True).count()

    best = base.filter(SearchRanking.position.isnot(None)).order_by(SearchRanking.position).first()

    return {
        "ok": True,
        "data": {
            "total_checks": total_checks,
            "on_page": on_page,
            "ai_cited": ai_cited,
            "best_position": best.position if best else None,
            "best_keyword": best.keyword if best else None,
        }
    }


@router.get("/geo/overview")
def get_geo_overview(
    days: int = Query(30, ge=1, le=365),
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user),
):
    """GEO 概览 — AI 搜索引用 + 就绪度评分"""
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    # Filter geo checks by org via articles
    base = db.query(SearchRanking)
    base = _apply_org_filter(base, current_user, db)
    base = base.filter(SearchRanking.checked_at >= cutoff)
    total_geo = base.filter(SearchRanking.search_engine == "geo").count()
    total_cited = base.filter(
        SearchRanking.search_engine == "geo",
        SearchRanking.ai_cited == True
    ).count()

    engine_breakdown = {}
    engines = base.filter(
        SearchRanking.search_engine == "geo",
        SearchRanking.ai_source.isnot(None)
    ).with_entities(
        SearchRanking.ai_source,
        func.count(SearchRanking.id).label("cnt"),
        func.sum(func.cast(SearchRanking.ai_cited, func.Integer)).label("cited"),
    ).group_by(SearchRanking.ai_source).all()
    for row in engines:
        engine_breakdown[row.ai_source] = {
            "total": row.cnt,
            "cited": row.cited or 0,
        }

    latest_scores = db.query(GeoReadinessScore).order_by(
        desc(GeoReadinessScore.checked_at)
    ).limit(50).all()

    avg_score = db.query(func.avg(GeoReadinessScore.total_score)).filter(
        GeoReadinessScore.checked_at >= cutoff
    ).scalar() or 0

    return {
        "ok": True,
        "data": {
            "total_geo_checks": total_geo,
            "total_cited": total_cited,
            "citation_rate": round(total_cited / total_geo * 100, 1) if total_geo else 0,
            "engine_breakdown": engine_breakdown,
            "avg_geo_readiness": round(float(avg_score), 1),
            "latest_scores": [s.to_dict() for s in latest_scores[:10]],
        }
    }


@router.get("/geo/readiness")
def get_geo_readiness_scores(
    days: int = Query(30, ge=1, le=365),
    limit: int = Query(50, ge=1, le=200),
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user),
):
    """获取 GEO 就绪度评分列表"""
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    # Join via articles to filter by org
    scores_query = db.query(GeoReadinessScore).join(
        ArticleModel, GeoReadinessScore.article_id == ArticleModel.id, isouter=True
    )
    of = org_filter(current_user, ArticleModel)
    if of is not True:
        scores_query = scores_query.filter(of)
    scores = scores_query.filter(
        GeoReadinessScore.checked_at >= cutoff
    ).order_by(desc(GeoReadinessScore.checked_at)).limit(limit).all()
    return {"ok": True, "data": [s.to_dict() for s in scores]}


@router.post("/geo/track")
def trigger_geo_tracking(
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user),
):
    """手动触发 GEO 追踪"""
    try:
        from scripts.geo_tracker import run_all
        result = run_all()
        return {"ok": True, "message": "GEO 追踪已启动", "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))