Yuzhiran Dev
217 lines
7.1 KiB
Python
217 lines
7.1 KiB
Python
#!/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()
|