#!/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_name 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 and len(field) >= 4: 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 _call_llm(prompt: str, max_tokens: int = 1000) -> str: try: from app.core.nvidia_client import call_llm return call_llm(prompt, temperature=0.3, max_tokens=max_tokens) or "" except Exception as e: logger.warning(f"LLM 调用失败: {e}") return "" def _check_ai_citation(article: Dict, keyword: str) -> Dict: title = article.get("title") or article.get("topic_title", "") prompt = f"""你是一个 AI 搜索结果评估器。请判断以下文章是否可能被 AI 搜索引擎（如 Bing AI、百度 AI 搜索、Perplexity）引用。 文章标题：{title} 搜索关键词：{keyword} 请用以下格式返回： 被引用：是/否 引擎：Bing AI, 百度 AI 搜索, Perplexity（逗号分隔，若被引用） 证据：简短的屏幕抓取或推理依据 只返回以上三行，不要多余内容。""" response = _call_llm(prompt, max_tokens=500) cited = False ai_sources = [] if response: cited = "是" in response.split("\n")[0] if "\n" in response else "是" in response if cited: for line in response.split("\n"): if line.startswith("引擎"): sources = [s.strip() for s in line.replace("引擎：", "").replace("引擎:", "").split(",") if s.strip()] ai_sources = sources return { "ai_cited": cited, "ai_source": ",".join(ai_sources) if ai_sources else None, "citation_snippet": response[:300] if cited else None, } def check_rankings(article: Dict, keywords: List[str], engine: str = "bing") -> List[Dict]: """检查文章关键词在搜索引擎的排名，并检测 AI 搜索引用""" 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 ai_info = _check_ai_citation(article, keyword) if ai_info.get("ai_cited"): logger.info(f" [AI] '{keyword}' → 被 AI 搜索引用 ({ai_info.get('ai_source', '未知')})") 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": ai_info.get("ai_cited", False), "ai_source": ai_info.get("ai_source"), }) 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()