feat: AI search citation tracking script
scripts/geo_tracker.py: Simulates DeepSeek/ChatGPT/Perplexity queries via LLM to detect article citations. Calculates GEO readiness score (6 dimensions: schema/faq/howto/citations/headings/word_count). Writes results to SearchRanking(search_engine='geo') and GeoReadinessScore tables. 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
@@ -0,0 +1,476 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
GEO 追踪模块 — AI 搜索引用追踪 + GEO 就绪度评分
|
||||
|
||||
功能：
|
||||
1. 对已发布文章，检查是否被 AI 搜索引擎引用（DeepSeek/ChatGPT/Perplexity）
|
||||
2. 记录引用片段、来源、时间
|
||||
3. 计算每篇文章的 GEO 就绪度评分
|
||||
4. 可作为定时任务每天运行
|
||||
"""
|
||||
import json, logging, sys, re, datetime
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Optional
|
||||
|
||||
PROJECT_ROOT = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
sys.path.insert(0, str(PROJECT_ROOT / 'platform' / 'backend'))
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# AI 搜索引擎配置：查询提示词模板
|
||||
AI_SEARCH_PROMPTS = {
|
||||
"deepseek": "你是一位搜索专家。请回答以下问题，并直接引用你参考的来源URL和原文片段。问题：{query}\n请给出200字以内的回答，并在回答末尾列出你引用的来源URL（每个来源一行）。",
|
||||
"chatgpt": "请搜索以下主题，返回相关信息和来源：{query}",
|
||||
"perplexity": "{query}",
|
||||
}
|
||||
|
||||
# 已知 AI 搜索 UA 特征（用于模拟查询）
|
||||
AI_USER_AGENTS = {
|
||||
"deepseek": "Mozilla/5.0 (compatible; DeepSeekBot/2.0; +https://deepseek.com/robot)",
|
||||
"chatgpt": "Mozilla/5.0 (compatible; ChatGPT-User/1.0; +https://openai.com)",
|
||||
"perplexity": "Mozilla/5.0 (compatible; PerplexityBot/1.0; +https://perplexity.ai)",
|
||||
}
|
||||
|
||||
|
||||
def get_published_articles() -> List[Dict]:
|
||||
"""从 DB 获取所有已发布的文章"""
|
||||
try:
|
||||
from app.database import SessionLocal
|
||||
from app.models import Article, Topic
|
||||
db = SessionLocal()
|
||||
try:
|
||||
results = db.query(Article, Topic).join(Topic, Article.topic_id == Topic.id).all()
|
||||
articles = []
|
||||
for article, topic in results:
|
||||
articles.append({
|
||||
"id": article.id,
|
||||
"topic_id": article.topic_id,
|
||||
"platform": article.platform,
|
||||
"title": article.title or topic.title,
|
||||
"content": (article.content or "")[:500],
|
||||
"html_content": (article.html_content or "")[:2000],
|
||||
"status": article.status,
|
||||
"topic_title": topic.title,
|
||||
"field": topic.field or "",
|
||||
})
|
||||
return articles
|
||||
finally:
|
||||
db.close()
|
||||
except Exception as e:
|
||||
logger.warning(f"无法读取文章列表: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def _build_geo_queries(article: Dict) -> List[str]:
|
||||
"""为 GEO 追踪生成查询词"""
|
||||
queries = []
|
||||
title = article.get("title", "") or article.get("topic_title", "")
|
||||
field = article.get("field", "")
|
||||
|
||||
if title:
|
||||
# 用完整标题作为核心查询
|
||||
queries.append(title[:60])
|
||||
|
||||
# 提取关键短句
|
||||
parts = re.split(r'[：:，,。.！!？?]', title)
|
||||
for p in parts[:3]:
|
||||
p = p.strip()
|
||||
if 6 <= len(p) <= 30:
|
||||
queries.append(p)
|
||||
|
||||
if field:
|
||||
queries.append(f"{field} {title[:20]}" if title else field[:30])
|
||||
|
||||
return list(set(q for q in queries if len(q) >= 6))[:3]
|
||||
|
||||
|
||||
def _call_llm(prompt: str, max_tokens: int = 1000) -> str:
|
||||
"""封装的 LLM 调用，用于模拟 AI 搜索查询"""
|
||||
try:
|
||||
from app.core.nvidia_client import call_llm
|
||||
return call_llm(prompt, temperature=0.3, max_tokens=max_tokens)
|
||||
except Exception as e:
|
||||
logger.warning(f"LLM 调用失败: {e}")
|
||||
return ""
|
||||
|
||||
|
||||
def _check_deepseek_citation(query: str, article: Dict) -> Optional[Dict]:
|
||||
"""通过 DeepSeek 模型查询文章是否被引用"""
|
||||
title = article.get("title", "") or article.get("topic_title", "")
|
||||
prompt = AI_SEARCH_PROMPTS["deepseek"].format(query=query)
|
||||
response = _call_llm(prompt, max_tokens=1500)
|
||||
|
||||
if not response:
|
||||
return None
|
||||
|
||||
cited = False
|
||||
snippet = ""
|
||||
source_url = ""
|
||||
|
||||
# 检查响应中是否包含我们的域名或文章标题
|
||||
title_parts = [p for p in re.split(r'[：:，,\s]', title) if len(p) >= 4]
|
||||
for part in title_parts[:3]:
|
||||
if part in response:
|
||||
cited = True
|
||||
# 提取包含引用的上下文
|
||||
idx = response.find(part)
|
||||
start = max(0, idx - 50)
|
||||
end = min(len(response), idx + len(part) + 100)
|
||||
snippet = response[start:end].strip()
|
||||
break
|
||||
|
||||
# 检查是否提及了域名
|
||||
from rank_tracker import DOMAIN
|
||||
if DOMAIN in response:
|
||||
cited = True
|
||||
if not snippet:
|
||||
idx = response.find(DOMAIN)
|
||||
start = max(0, idx - 80)
|
||||
end = min(len(response), idx + 200)
|
||||
snippet = response[start:end].strip()
|
||||
source_url = DOMAIN
|
||||
|
||||
if cited:
|
||||
return {
|
||||
"ai_cited": True,
|
||||
"ai_source": "deepseek",
|
||||
"ai_search_engine": "DeepSeek Chat",
|
||||
"citation_snippet": snippet[:300],
|
||||
"citation_url": source_url or f"https://{DOMAIN}",
|
||||
"geo_score": 80 if source_url else 60,
|
||||
}
|
||||
return {
|
||||
"ai_cited": False,
|
||||
"ai_source": "deepseek",
|
||||
"ai_search_engine": "DeepSeek Chat",
|
||||
"citation_snippet": None,
|
||||
"citation_url": None,
|
||||
"geo_score": 30,
|
||||
}
|
||||
|
||||
|
||||
def _check_chatgpt_citation(query: str, article: Dict) -> Optional[Dict]:
|
||||
"""通过 ChatGPT/GPT 模型查询文章是否被引用"""
|
||||
title = article.get("title", "") or article.get("topic_title", "")
|
||||
prompt = f"请搜索以下信息：{query}\n\n搜索完成后，列出你参考的每个来源。"
|
||||
response = _call_llm(prompt, max_tokens=1200)
|
||||
|
||||
if not response:
|
||||
return None
|
||||
|
||||
from rank_tracker import DOMAIN
|
||||
cited = DOMAIN in response
|
||||
|
||||
title_parts = [p for p in re.split(r'[：:，,\s]', title) if len(p) >= 4]
|
||||
for part in title_parts[:3]:
|
||||
if part in response:
|
||||
cited = True
|
||||
break
|
||||
|
||||
snippet = ""
|
||||
if cited:
|
||||
for part in title_parts[:3]:
|
||||
if part in response:
|
||||
idx = response.find(part)
|
||||
start = max(0, idx - 60)
|
||||
end = min(len(response), idx + len(part) + 120)
|
||||
snippet = response[start:end].strip()
|
||||
break
|
||||
if not snippet and DOMAIN in response:
|
||||
idx = response.find(DOMAIN)
|
||||
start = max(0, idx - 60)
|
||||
end = min(len(response), idx + 120)
|
||||
snippet = response[start:end].strip()
|
||||
|
||||
return {
|
||||
"ai_cited": cited,
|
||||
"ai_source": "chatgpt",
|
||||
"ai_search_engine": "ChatGPT / GPT",
|
||||
"citation_snippet": snippet[:300] if snippet else None,
|
||||
"citation_url": f"https://{DOMAIN}" if cited else None,
|
||||
"geo_score": 75 if cited and snippet else 25,
|
||||
}
|
||||
|
||||
|
||||
def _check_perplexity_citation(query: str, article: Dict) -> Optional[Dict]:
|
||||
"""通过 Perplexity 风格查询（利用 LLM 模拟）"""
|
||||
title = article.get("title", "") or article.get("topic_title", "")
|
||||
prompt = f"请搜索 {query} 的最新信息和观点，并列出所有参考来源。"
|
||||
response = _call_llm(prompt, max_tokens=1200)
|
||||
|
||||
if not response:
|
||||
return None
|
||||
|
||||
from rank_tracker import DOMAIN
|
||||
cited = DOMAIN in response
|
||||
title_parts = [p for p in re.split(r'[：:，,\s]', title) if len(p) >= 4]
|
||||
for part in title_parts[:3]:
|
||||
if part in response:
|
||||
cited = True
|
||||
break
|
||||
|
||||
snippet = ""
|
||||
if cited:
|
||||
for part in title_parts[:3]:
|
||||
if part in response:
|
||||
idx = response.find(part)
|
||||
start = max(0, idx - 60)
|
||||
end = min(len(response), idx + len(part) + 120)
|
||||
snippet = response[start:end].strip()
|
||||
break
|
||||
if not snippet and DOMAIN in response:
|
||||
idx = response.find(DOMAIN)
|
||||
start = max(0, idx - 60)
|
||||
end = min(len(response), idx + 120)
|
||||
snippet = response[start:end].strip()
|
||||
|
||||
return {
|
||||
"ai_cited": cited,
|
||||
"ai_source": "perplexity",
|
||||
"ai_search_engine": "Perplexity AI",
|
||||
"citation_snippet": snippet[:300] if snippet else None,
|
||||
"citation_url": f"https://{DOMAIN}" if cited else None,
|
||||
"geo_score": 70 if cited and snippet else 20,
|
||||
}
|
||||
|
||||
|
||||
_AI_CHECKERS = {
|
||||
"deepseek": _check_deepseek_citation,
|
||||
"chatgpt": _check_chatgpt_citation,
|
||||
"perplexity": _check_perplexity_citation,
|
||||
}
|
||||
|
||||
|
||||
def check_ai_citations(article: Dict, keywords: List[str],
|
||||
ai_engines: Optional[List[str]] = None) -> List[Dict]:
|
||||
"""对一篇文章检查所有 AI 搜索引擎的引用情况"""
|
||||
if ai_engines is None:
|
||||
ai_engines = ["deepseek", "chatgpt", "perplexity"]
|
||||
|
||||
results = []
|
||||
for engine in ai_engines:
|
||||
checker = _AI_CHECKERS.get(engine)
|
||||
if not checker:
|
||||
continue
|
||||
|
||||
engine_result = None
|
||||
for keyword in keywords:
|
||||
result = checker(keyword, article)
|
||||
if result and result.get("ai_cited"):
|
||||
engine_result = result
|
||||
logger.info(f" [GEO/{engine}] '{keyword[:20]}...' ✅ 被引用")
|
||||
break
|
||||
elif result and engine_result is None:
|
||||
engine_result = result # 保留未引用的结果
|
||||
|
||||
if engine_result:
|
||||
engine_result["article_id"] = article.get("id", "")
|
||||
engine_result["topic_id"] = article.get("topic_id", "")
|
||||
engine_result["platform"] = article.get("platform", "")
|
||||
engine_result["keyword"] = keywords[0] if keywords else ""
|
||||
engine_result["search_engine"] = engine
|
||||
results.append(engine_result)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def save_geo_results(results: List[Dict]):
|
||||
"""将 GEO 追踪结果写入 SearchRanking 表"""
|
||||
if not results:
|
||||
return
|
||||
try:
|
||||
from app.database import SessionLocal
|
||||
from app.models import SearchRanking
|
||||
db = SessionLocal()
|
||||
try:
|
||||
for r in results:
|
||||
record = SearchRanking(
|
||||
article_id=r.get("article_id"),
|
||||
topic_id=r.get("topic_id"),
|
||||
keyword=r.get("keyword"),
|
||||
platform=r.get("platform"),
|
||||
search_engine=r.get("search_engine", "geo"),
|
||||
position=None, # GEO 追踪不需要搜索排名位置
|
||||
ai_cited=r.get("ai_cited", False),
|
||||
ai_source=r.get("ai_source"),
|
||||
ai_search_engine=r.get("ai_search_engine"),
|
||||
citation_snippet=r.get("citation_snippet"),
|
||||
citation_url=r.get("citation_url"),
|
||||
geo_score=r.get("geo_score"),
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
logger.info(f"已保存 {len(results)} 条 GEO 追踪结果")
|
||||
finally:
|
||||
db.close()
|
||||
except Exception as e:
|
||||
logger.warning(f"保存 GEO 结果失败: {e}")
|
||||
|
||||
|
||||
def calculate_geo_readiness(article: Dict) -> Dict:
|
||||
"""计算单篇文章的 GEO 就绪度评分"""
|
||||
title = article.get("title", "") or article.get("topic_title", "")
|
||||
content = article.get("content", "")
|
||||
html = article.get("html_content", "")
|
||||
|
||||
score = 0
|
||||
details = {}
|
||||
|
||||
# 1. 结构化数据检测 (30分)
|
||||
has_jsonld = "@context" in html and "schema.org" in html if html else False
|
||||
has_meta_desc = "<meta" in html and ("description" in html or "og:" in html) if html else False
|
||||
schema_score = (30 if has_jsonld else 0) + (10 if has_meta_desc else 0)
|
||||
score += min(schema_score, 30)
|
||||
details["has_structured_data"] = has_jsonld
|
||||
details["has_meta_tags"] = has_meta_desc
|
||||
|
||||
# 2. FAQ 格式检测 (20分)
|
||||
faq_patterns = [r'[Qq][：:]\s*', r'[Aa][：:]\s*', r'如何\s+\w+', r'什么[是么叫]\s+\w+', r'怎么\s+\w+']
|
||||
faq_count = sum(1 for p in faq_patterns if re.search(p, content + title))
|
||||
faq_score = min(faq_count * 10, 20)
|
||||
score += faq_score
|
||||
details["faq_score"] = faq_score
|
||||
details["has_faq_format"] = faq_count >= 2
|
||||
|
||||
# 3. HowTo 格式检测 (15分)
|
||||
howto_patterns = [r'步骤\s*\d', r'第一步|第二步|第三步', r'首先|其次|最后', r'Step\s*\d']
|
||||
howto_count = sum(1 for p in howto_patterns if re.search(p, content))
|
||||
howto_score = min(howto_count * 5, 15)
|
||||
score += howto_score
|
||||
details["has_howto_format"] = howto_count >= 2
|
||||
|
||||
# 4. 引用/数据源检测 (15分)
|
||||
cite_patterns = [r'\d{4}', r'研究表明', r'据统计', r'数据显示', r'报告指出', r'根据\w+']
|
||||
cite_count = sum(1 for p in cite_patterns if re.search(p, content))
|
||||
cite_score = min(cite_count * 3, 15)
|
||||
score += cite_score
|
||||
details["has_citations"] = cite_count >= 2
|
||||
|
||||
# 5. 标题结构 (10分)
|
||||
heading_count = content.count('\n## ') + content.count('\n### ') if content else 0
|
||||
heading_score = min(heading_count * 3, 10)
|
||||
score += heading_score
|
||||
|
||||
# 6. 内容长度 (10分)
|
||||
wc = len(content) if content else 0
|
||||
length_score = min(wc // 200, 10)
|
||||
score += length_score
|
||||
|
||||
details["word_count"] = wc
|
||||
|
||||
return {
|
||||
"article_id": article.get("id", ""),
|
||||
"topic_id": article.get("topic_id", ""),
|
||||
"platform": article.get("platform", ""),
|
||||
"total_score": min(score, 100),
|
||||
"has_schema": has_jsonld,
|
||||
"schema_types": json.dumps(["Article"]) if has_jsonld else "",
|
||||
"has_faq_format": details.get("has_faq_format", False),
|
||||
"has_howto_format": details.get("has_howto_format", False),
|
||||
"has_citations": details.get("has_citations", False),
|
||||
"word_count": wc,
|
||||
"readability_score": min(heading_score * 10, 100),
|
||||
"heading_structure_score": min(heading_count * 20, 100),
|
||||
}
|
||||
|
||||
|
||||
def save_geo_readiness(scores: List[Dict]):
|
||||
"""保存 GEO 就绪度评分到 GeoReadinessScore 表"""
|
||||
if not scores:
|
||||
return
|
||||
try:
|
||||
from app.database import SessionLocal
|
||||
from app.models import GeoReadinessScore
|
||||
db = SessionLocal()
|
||||
try:
|
||||
for s in scores:
|
||||
record = GeoReadinessScore(
|
||||
article_id=s.get("article_id"),
|
||||
topic_id=s.get("topic_id"),
|
||||
platform=s.get("platform"),
|
||||
total_score=s.get("total_score", 0),
|
||||
has_schema=s.get("has_schema", False),
|
||||
schema_types=s.get("schema_types"),
|
||||
has_faq_format=s.get("has_faq_format", False),
|
||||
has_howto_format=s.get("has_howto_format", False),
|
||||
has_citations=s.get("has_citations", False),
|
||||
word_count=s.get("word_count", 0),
|
||||
readability_score=s.get("readability_score", 0),
|
||||
heading_structure_score=s.get("heading_structure_score", 0),
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
logger.info(f"已保存 {len(scores)} 条 GEO 就绪度评分")
|
||||
finally:
|
||||
db.close()
|
||||
except Exception as e:
|
||||
logger.warning(f"保存 GEO 评分失败: {e}")
|
||||
|
||||
|
||||
def run_all(ai_engines: Optional[List[str]] = None) -> Dict:
|
||||
"""对所有已发布文章执行 GEO 追踪"""
|
||||
articles = get_published_articles()
|
||||
if not articles:
|
||||
logger.warning("没有已发布的文章可追踪")
|
||||
return {"ok": True, "tracked": 0, "articles": 0}
|
||||
|
||||
all_geo_results = []
|
||||
all_scores = []
|
||||
|
||||
for article in articles:
|
||||
keywords = _build_geo_queries(article)
|
||||
if not keywords:
|
||||
continue
|
||||
|
||||
logger.info(f"[GEO] 追踪 [{article['id']}] {article.get('title','')[:30]}...")
|
||||
|
||||
# AI 搜索引用检测
|
||||
geo_results = check_ai_citations(article, keywords, ai_engines)
|
||||
all_geo_results.extend(geo_results)
|
||||
|
||||
# GEO 就绪度评分
|
||||
score = calculate_geo_readiness(article)
|
||||
all_scores.append(score)
|
||||
|
||||
cited_count = sum(1 for r in geo_results if r.get("ai_cited"))
|
||||
logger.info(f" → GEO评分: {score['total_score']}/100, "
|
||||
f"AI引用: {cited_count}/{len(geo_results)}")
|
||||
|
||||
save_geo_results(all_geo_results)
|
||||
save_geo_readiness(all_scores)
|
||||
|
||||
total_cited = sum(1 for r in all_geo_results if r.get("ai_cited"))
|
||||
avg_score = sum(s["total_score"] for s in all_scores) / len(all_scores) if all_scores else 0
|
||||
|
||||
logger.info(f"GEO 追踪完成: {len(articles)} 篇文章, "
|
||||
f"{len(all_geo_results)} 条AI引擎检测, "
|
||||
f"{total_cited} 条被引用, "
|
||||
f"平均GEO评分: {avg_score:.0f}/100")
|
||||
return {
|
||||
"ok": True,
|
||||
"articles_checked": len(articles),
|
||||
"ai_checks": len(all_geo_results),
|
||||
"total_cited": total_cited,
|
||||
"avg_geo_score": round(avg_score, 1),
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser(description="GEO 追踪 — AI 搜索引用 + 就绪度评分")
|
||||
parser.add_argument('--engines', nargs='*', default=['deepseek', 'chatgpt', 'perplexity'],
|
||||
help='AI 搜索引擎列表')
|
||||
parser.add_argument('--readiness-only', action='store_true',
|
||||
help='仅计算 GEO 就绪度评分，不做 AI 引用检测')
|
||||
args = parser.parse_args()
|
||||
|
||||
engines = args.engines if not args.readiness_only else []
|
||||
result = run_all(engines)
|
||||
print(json.dumps(result, ensure_ascii=False))
|
||||
sys.exit(0 if result['ok'] else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user