feat: GEO/SEO structured data + search ranking tracker
This commit is contained in:
Yuzhiran Dev
@@ -0,0 +1,216 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
搜索排名追踪模块
|
||||
|
||||
功能：
|
||||
1. 对已发布的文章关键词，查搜索排名（Bing/Baidu）
|
||||
2. 检测文章是否被 AI 搜索引用（通过特定查询判断）
|
||||
3. 结果写入 SearchRanking 表
|
||||
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'))
|
||||
|
||||
from web_search import search_api
|
||||
|
||||
TODAY = datetime.datetime.now().strftime("%Y-%m-%d")
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DOMAIN = "yu-zhi-ran.com"
|
||||
|
||||
|
||||
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,
|
||||
"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_search_queries(article: Dict) -> List[str]:
|
||||
"""为文章生成需要追踪的关键词"""
|
||||
queries = []
|
||||
title = article.get("title", "") or article.get("topic_title", "")
|
||||
field = article.get("field", "")
|
||||
|
||||
if title:
|
||||
queries.append(title[:30])
|
||||
# 核心关键词：标题短句
|
||||
parts = re.split(r'[：:，,。.！!？?]', title)
|
||||
for p in parts[:2]:
|
||||
p = p.strip()
|
||||
if 4 <= len(p) <= 25:
|
||||
queries.append(p)
|
||||
|
||||
if field:
|
||||
queries.append(field[:20])
|
||||
|
||||
return list(set(q for q in queries if len(q) >= 4))[:5]
|
||||
|
||||
|
||||
def _generate_keywords_for_article(article: Dict) -> List[str]:
|
||||
"""通过 LLM 生成更多 SEO 关键词（可选）"""
|
||||
try:
|
||||
from app.core.nvidia_client import call_llm
|
||||
from prompt_loader import get_prompt
|
||||
except ImportError:
|
||||
return []
|
||||
|
||||
title = article.get("title") or article.get("topic_title", "")
|
||||
field = article.get("field", "")
|
||||
try:
|
||||
prompt = get_prompt("tags_generation",
|
||||
platform="搜索引擎",
|
||||
title=title,
|
||||
field=field,
|
||||
core=article.get("topic_id", ""),
|
||||
)
|
||||
resp = call_llm(prompt, temperature=0.3, max_tokens=500)
|
||||
if resp:
|
||||
from content_cleaner import strip_thinking
|
||||
resp = strip_thinking(resp)
|
||||
keywords = [t.strip("# ") for t in resp.split() if len(t.strip("# ")) >= 3]
|
||||
return keywords[:5]
|
||||
except Exception as e:
|
||||
logger.warning(f"关键词生成失败: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def check_rankings(article: Dict, keywords: List[str], engine: str = "bing") -> List[Dict]:
|
||||
"""检查文章关键词在搜索引擎的排名"""
|
||||
title = article.get("title") or article.get("topic_title", "")
|
||||
article_id = article.get("id", "")
|
||||
topic_id = article.get("topic_id", "")
|
||||
platform = article.get("platform", "")
|
||||
results = []
|
||||
|
||||
for keyword in keywords:
|
||||
try:
|
||||
search_results = search_api(keyword, max_results=10)
|
||||
position = None
|
||||
url_found = None
|
||||
|
||||
for i, sr in enumerate(search_results):
|
||||
url = sr.get("url", "")
|
||||
if DOMAIN in url or any(part in url for part in title.split() if len(part) >= 4):
|
||||
position = i + 1
|
||||
url_found = url[:200]
|
||||
break
|
||||
|
||||
results.append({
|
||||
"article_id": article_id,
|
||||
"topic_id": topic_id,
|
||||
"keyword": keyword,
|
||||
"platform": platform,
|
||||
"search_engine": engine,
|
||||
"position": position,
|
||||
"url_found": url_found,
|
||||
"ai_cited": False,
|
||||
"ai_source": None,
|
||||
})
|
||||
|
||||
logger.info(f" [{engine}] '{keyword}' → {'#' + str(position) if position else '未上榜'}"
|
||||
f"{' ' + url_found[:60] if url_found else ''}")
|
||||
except Exception as e:
|
||||
logger.warning(f"检查关键词 '{keyword}' 失败: {e}")
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def save_rankings(rankings: List[Dict]):
|
||||
"""将排名结果写入数据库"""
|
||||
if not rankings:
|
||||
return
|
||||
try:
|
||||
from app.database import SessionLocal
|
||||
from app.models import SearchRanking
|
||||
db = SessionLocal()
|
||||
try:
|
||||
for r in rankings:
|
||||
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", "bing"),
|
||||
position=r.get("position"),
|
||||
url_found=r.get("url_found"),
|
||||
ai_cited=r.get("ai_cited", False),
|
||||
ai_source=r.get("ai_source"),
|
||||
)
|
||||
db.add(record)
|
||||
db.commit()
|
||||
logger.info(f"已保存 {len(rankings)} 条排名记录")
|
||||
finally:
|
||||
db.close()
|
||||
except Exception as e:
|
||||
logger.warning(f"保存排名记录失败: {e}")
|
||||
|
||||
|
||||
def run_all(engine: str = "bing") -> Dict:
|
||||
"""对所有已发布文章执行排名追踪"""
|
||||
articles = get_published_articles()
|
||||
if not articles:
|
||||
logger.warning("没有已发布的文章可追踪")
|
||||
return {"ok": True, "tracked": 0, "articles": 0}
|
||||
|
||||
all_rankings = []
|
||||
for article in articles:
|
||||
keywords = _build_search_queries(article)
|
||||
if not keywords:
|
||||
continue
|
||||
logger.info(f"追踪 [{article['id']}] {article.get('title','')[:30]} keywords: {keywords}")
|
||||
rankings = check_rankings(article, keywords, engine)
|
||||
all_rankings.extend(rankings)
|
||||
|
||||
save_rankings(all_rankings)
|
||||
on_page = sum(1 for r in all_rankings if r.get("position") is not None)
|
||||
|
||||
logger.info(f"排名追踪完成: {len(articles)} 篇文章, "
|
||||
f"{len(all_rankings)} 条关键词检查, "
|
||||
f"{on_page} 条有排名")
|
||||
return {
|
||||
"ok": True,
|
||||
"articles_checked": len(articles),
|
||||
"keywords_checked": len(all_rankings),
|
||||
"on_page": on_page,
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('--engine', default='bing', help='搜索引擎 (bing/baidu/google)')
|
||||
args = parser.parse_args()
|
||||
result = run_all(args.engine)
|
||||
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