diff --git a/platform/backend/app/api/search_rankings.py b/platform/backend/app/api/search_rankings.py index 478b8b7..56d0a69 100644 --- a/platform/backend/app/api/search_rankings.py +++ b/platform/backend/app/api/search_rankings.py @@ -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)) diff --git a/platform/backend/app/models.py b/platform/backend/app/models.py index ae0c567..8283fc7 100644 --- a/platform/backend/app/models.py +++ b/platform/backend/app/models.py @@ -860,7 +860,7 @@ class CollectorSource(Base): class SearchRanking(Base): - """搜索排名追踪""" + """搜索排名追踪 + AI 搜索引用追踪""" __tablename__ = "search_rankings" id = Column(Integer, primary_key=True, index=True, autoincrement=True) @@ -872,7 +872,12 @@ class SearchRanking(Base): position = Column(Integer, nullable=True) # 搜索排名位置（null = 未上榜） url_found = Column(String, nullable=True) # 被找到的具体 URL ai_cited = Column(Boolean, default=False) # 是否被 AI 搜索引用 - ai_source = Column(String, nullable=True) # AI 搜索来源名称 + ai_source = Column(String, nullable=True) # AI 搜索来源（chatgpt/perplexity/deepseek） + ai_search_engine = Column(String, nullable=True) # AI 搜索具体引擎模型名 + citation_snippet = Column(String, nullable=True) # 被引用的文本片段 + citation_url = Column(String, nullable=True) # 引用来源链接 + geo_score = Column(Integer, nullable=True) # GEO 就绪度评分 0-100 + content_type = Column(String, nullable=True) # 内容类型: article/listicle/howto/faq/review checked_at = Column(DateTime(timezone=True), server_default=func.now()) def to_dict(self): @@ -887,5 +892,48 @@ class SearchRanking(Base): "url_found": self.url_found, "ai_cited": self.ai_cited, "ai_source": self.ai_source, + "ai_search_engine": self.ai_search_engine, + "citation_snippet": self.citation_snippet, + "citation_url": self.citation_url, + "geo_score": self.geo_score, + "content_type": self.content_type, + "checked_at": self.checked_at.isoformat() if self.checked_at else None, + } + + +class GeoReadinessScore(Base): + """GEO 就绪度评分 — 按文章评估被 AI 搜索引用的概率""" + __tablename__ = "geo_readiness_scores" + + id = Column(Integer, primary_key=True, index=True, autoincrement=True) + article_id = Column(String, ForeignKey("articles.id"), nullable=True, index=True) + topic_id = Column(String, ForeignKey("topics.id"), nullable=True) + platform = Column(String, nullable=True) + total_score = Column(Integer, default=0) # 总分 0-100 + has_schema = Column(Boolean, default=False) # 是否有结构化数据 + schema_types = Column(String, nullable=True) # 含有的 schema 类型列表 + has_faq_format = Column(Boolean, default=False) # 是否含 FAQ 格式 + has_howto_format = Column(Boolean, default=False) # 是否含 HowTo 格式 + has_citations = Column(Boolean, default=False) # 是否引用数据源 + word_count = Column(Integer, default=0) + readability_score = Column(Integer, default=0) # 可读性评分 0-100 + heading_structure_score = Column(Integer, default=0) # 标题结构评分 0-100 + checked_at = Column(DateTime(timezone=True), server_default=func.now()) + + def to_dict(self): + return { + "id": self.id, + "article_id": self.article_id, + "topic_id": self.topic_id, + "platform": self.platform, + "total_score": self.total_score, + "has_schema": self.has_schema, + "schema_types": self.schema_types, + "has_faq_format": self.has_faq_format, + "has_howto_format": self.has_howto_format, + "has_citations": self.has_citations, + "word_count": self.word_count, + "readability_score": self.readability_score, + "heading_structure_score": self.heading_structure_score, "checked_at": self.checked_at.isoformat() if self.checked_at else None, } \ No newline at end of file