feat: GEO data model and API endpoints

- SearchRanking: +5 fields (ai_search_engine, citation_snippet, citation_url, geo_score, content_type)
- New GeoReadinessScore model (6-dimension scoring: schema/faq/howto/citations/word_count/readability/headings)
- GET /api/seo/geo/overview, GET /api/seo/geo/readiness, POST /api/seo/geo/track

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
Yuzhiran Dev
2026-06-16 08:25:18 +08:00
parent 63a6fabc00
commit 773080cf4f
2 changed files with 132 additions and 3 deletions
+82 -1
View File
@@ -5,7 +5,7 @@ from typing import Optional
from datetime import datetime, timedelta, timezone
from ..database import get_db
from ..models import SearchRanking, Article, Topic
from ..models import SearchRanking, Article, Topic, GeoReadinessScore
from .auth import get_current_user, org_filter
router = APIRouter(prefix="/api/seo", tags=["seo"])
@@ -75,3 +75,84 @@ def get_rankings_overview(
"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)
base = db.query(SearchRanking).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)
scores = db.query(GeoReadinessScore).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))