diff --git a/scripts/rank_tracker.py b/scripts/rank_tracker.py index cdd8c19..34fd3e3 100644 --- a/scripts/rank_tracker.py +++ b/scripts/rank_tracker.py @@ -42,7 +42,7 @@ def get_published_articles() -> List[Dict]: "title": article.title or topic.title, "status": article.status, "topic_title": topic.title, - "field": topic.field or "", + "field": topic.field_name or "", }) return articles finally: @@ -67,7 +67,7 @@ def _build_search_queries(article: Dict) -> List[str]: if 4 <= len(p) <= 25: queries.append(p) - if field: + if field and len(field) >= 4: queries.append(field[:20]) return list(set(q for q in queries if len(q) >= 4))[:5] @@ -101,8 +101,47 @@ def _generate_keywords_for_article(article: Dict) -> List[str]: 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", "") @@ -122,6 +161,10 @@ def check_rankings(article: Dict, keywords: List[str], engine: str = "bing") -> 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, @@ -130,8 +173,8 @@ def check_rankings(article: Dict, keywords: List[str], engine: str = "bing") -> "search_engine": engine, "position": position, "url_found": url_found, - "ai_cited": False, - "ai_source": None, + "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 '未上榜'}"