Codex
216 lines
10 KiB
Python
216 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any, Optional
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
from app.core.config import Settings
|
|
from app.core.timezone import to_beijing_naive
|
|
from app.schemas.agent import SignalItem
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class NewsSearchClient:
|
|
"""通过Grok Web Search采集中英文AI新闻候选。"""
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
|
|
def collect_ai_news(self, run_date: date) -> list[SignalItem]:
|
|
# Grok负责联网搜索新闻候选，入库前仍统一校验来源链接和结构字段。
|
|
items: list[dict[str, Any]] = []
|
|
for query_group in self._query_groups():
|
|
payload = self._search_with_grok(run_date=run_date, query_group=query_group)
|
|
items.extend(payload.get("items", []))
|
|
candidates = [self._article_to_signal(item) for item in items]
|
|
candidates = [candidate for candidate in candidates if candidate is not None]
|
|
unique: dict[str, SignalItem] = {}
|
|
for candidate in candidates:
|
|
unique[candidate.source_url] = candidate
|
|
ranked = sorted(
|
|
unique.values(),
|
|
key=lambda item: (item.importance, item.published_at or datetime.min),
|
|
reverse=True,
|
|
)
|
|
logger.info(
|
|
"news_candidates_collected source=grok_web_search days=%s items=%s",
|
|
self.settings.news_recent_days,
|
|
len(ranked),
|
|
)
|
|
return ranked[: self.settings.news_max_items]
|
|
|
|
def _query_groups(self) -> list[dict[str, Any]]:
|
|
# 多组搜索覆盖中文、英文、融资、模型和Agent工具，避免单次搜索只返回少量相似新闻。
|
|
return [
|
|
{
|
|
"name": "中文AI新闻",
|
|
"language": "zh",
|
|
"queries": ["人工智能 大模型 智能体 AI 新闻", "AI 大模型 创业 融资 产品 发布"],
|
|
"target_count": 15,
|
|
},
|
|
{
|
|
"name": "英文AI新闻",
|
|
"language": "en",
|
|
"queries": ["latest AI news LLM agents OpenAI Anthropic", "AI startup funding model release agent tools"],
|
|
"target_count": 15,
|
|
},
|
|
{
|
|
"name": "模型与Agent发布",
|
|
"language": "mixed",
|
|
"queries": ["大模型 发布 Agent 工具 开源", "LLM model release AI agent framework open source"],
|
|
"target_count": 15,
|
|
},
|
|
{
|
|
"name": "AI商业与融资",
|
|
"language": "mixed",
|
|
"queries": ["AI 融资 创业 公司 产品", "AI funding startup acquisition product launch"],
|
|
"target_count": 15,
|
|
},
|
|
]
|
|
|
|
def _search_with_grok(self, run_date: date, query_group: dict[str, Any]) -> dict[str, Any]:
|
|
# Responses API的web_search工具是当前新闻发现入口，确保候选来自真实网页搜索。
|
|
if not self.settings.llm_api_key:
|
|
raise RuntimeError("SIGNALSCOUT_LLM_API_KEY is required")
|
|
since = run_date - timedelta(days=self.settings.news_recent_days)
|
|
body = json.dumps(
|
|
{
|
|
"model": self.settings.llm_model,
|
|
"input": [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"你是 SignalScout 的新闻搜索器。必须使用 web_search 找到真实网页来源，"
|
|
"只输出 JSON object，不输出 Markdown 或额外解释。"
|
|
),
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": json.dumps(
|
|
{
|
|
"task": "联网搜索最近AI新闻，覆盖中文和英文来源。",
|
|
"date_window": {
|
|
"from": since.isoformat(),
|
|
"to": run_date.isoformat(),
|
|
},
|
|
"search_group": query_group["name"],
|
|
"preferred_language": query_group["language"],
|
|
"target_count": query_group["target_count"],
|
|
"queries": query_group["queries"],
|
|
"output_schema": {
|
|
"items": [
|
|
{
|
|
"title": "新闻标题",
|
|
"summary": "中文摘要，说明为什么值得关注",
|
|
"url": "真实可访问来源链接",
|
|
"source": "媒体或网站名称",
|
|
"published_at": "ISO时间；不知道可为null",
|
|
"language": "zh或en",
|
|
"importance": "1到5整数",
|
|
"entities": ["公司、模型、项目、人名等"],
|
|
}
|
|
]
|
|
},
|
|
"rules": [
|
|
"优先返回中文和英文来源各一部分，不要只返回英文来源。",
|
|
"只使用最近日期窗口内或明显接近日内的新闻。",
|
|
"每条必须有真实 url，不要使用搜索结果页、YouTube集合页或空链接。",
|
|
"如果中文来源足够，至少保留三分之一中文新闻。",
|
|
"返回尽可能接近 target_count 的 items。",
|
|
],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
},
|
|
],
|
|
"tools": [{"type": "web_search"}],
|
|
"max_output_tokens": self.settings.llm_max_tokens,
|
|
},
|
|
ensure_ascii=False,
|
|
).encode("utf-8")
|
|
request = Request(
|
|
f"{self.settings.llm_base_url.rstrip('/')}/responses",
|
|
data=body,
|
|
headers={
|
|
"Authorization": f"Bearer {self.settings.llm_api_key}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urlopen(request, timeout=self.settings.llm_timeout_seconds) as response:
|
|
response_body = response.read().decode("utf-8")
|
|
except HTTPError as exc:
|
|
message = exc.read().decode("utf-8", errors="replace")
|
|
raise RuntimeError(f"Grok news search failed: HTTP {exc.code} {message}") from exc
|
|
except URLError as exc:
|
|
raise RuntimeError(f"Grok news search failed: {exc.reason}") from exc
|
|
return self._parse_response_json(json.loads(response_body))
|
|
|
|
def _parse_response_json(self, response_payload: dict[str, Any]) -> dict[str, Any]:
|
|
# Responses返回多段output，最终message里的output_text才是模型给业务层的JSON。
|
|
text_parts: list[str] = []
|
|
for output in response_payload.get("output", []):
|
|
if not isinstance(output, dict):
|
|
continue
|
|
for content in output.get("content", []):
|
|
if isinstance(content, dict) and content.get("type") == "output_text":
|
|
text = content.get("text")
|
|
if isinstance(text, str):
|
|
text_parts.append(text)
|
|
content_text = "\n".join(text_parts).strip()
|
|
if not content_text:
|
|
raise RuntimeError("Grok news search returned no output_text")
|
|
try:
|
|
payload = json.loads(content_text)
|
|
except json.JSONDecodeError as exc:
|
|
raise RuntimeError(f"Grok news search returned invalid JSON: {content_text[:1200]}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise RuntimeError("Grok news search JSON root is not an object")
|
|
return payload
|
|
|
|
def _article_to_signal(self, item: dict[str, Any]) -> Optional[SignalItem]:
|
|
# 搜索结果只接受带真实网页链接的条目，避免把模型说明文本写入情报库。
|
|
title = item.get("title")
|
|
url = item.get("url")
|
|
if not isinstance(title, str) or not isinstance(url, str) or not url.startswith("http"):
|
|
return None
|
|
source = item.get("source")
|
|
language = item.get("language")
|
|
entities = item.get("entities")
|
|
return SignalItem(
|
|
source_type="news",
|
|
topic="中文 AI 新闻" if language == "zh" else "AI 新闻",
|
|
title=title[:300],
|
|
summary=str(item.get("summary") or "Grok Web Search 搜索到的AI新闻候选。"),
|
|
source_url=url,
|
|
source_name=str(source or "Grok Web Search"),
|
|
published_at=self._parse_datetime(item.get("published_at")),
|
|
importance=self._parse_importance(item.get("importance")),
|
|
entities=[str(entity) for entity in entities] if isinstance(entities, list) else [],
|
|
)
|
|
|
|
def _parse_datetime(self, value: Any) -> Optional[datetime]:
|
|
# 搜索结果可能只有日期或ISO时间，能解析则统一换算为北京时间。
|
|
if not isinstance(value, str) or not value:
|
|
return None
|
|
try:
|
|
if len(value) == 10:
|
|
value = f"{value}T00:00:00"
|
|
return to_beijing_naive(datetime.fromisoformat(value.replace("Z", "+00:00")))
|
|
except ValueError:
|
|
return None
|
|
|
|
def _parse_importance(self, value: Any) -> int:
|
|
# 重要度来自搜索模型，但入库前仍限制在业务契约允许范围内。
|
|
try:
|
|
importance = int(value)
|
|
except (TypeError, ValueError):
|
|
return 3
|
|
return max(1, min(importance, 5))
|