Codex
116 lines
5.1 KiB
Python
116 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from datetime import date, datetime, time, timedelta
|
|
from typing import Any, Optional
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
from app.core.config import Settings
|
|
from app.core.timezone import BEIJING_TZ, to_beijing_naive
|
|
from app.schemas.agent import SignalItem
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class NewsSearchClient:
|
|
"""通过公开技术新闻检索接口采集动态AI新闻候选。"""
|
|
|
|
endpoint = "https://hn.algolia.com/api/v1/search_by_date"
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
|
|
def collect_ai_news(self, run_date: date) -> list[SignalItem]:
|
|
# HN Algolia按时间倒序返回技术新闻候选，模型后续只从这些证据里挑选日报条目。
|
|
since = run_date - timedelta(days=self.settings.news_recent_days)
|
|
stories: dict[str, dict[str, Any]] = {}
|
|
for query in self._queries():
|
|
payload = self._search(query=query, since=since, until=run_date)
|
|
for story in payload.get("hits", []):
|
|
url = self._story_url(story)
|
|
object_id = story.get("objectID")
|
|
key = str(object_id or url)
|
|
if isinstance(url, str) and url.startswith("http"):
|
|
stories[key] = story
|
|
|
|
candidates = [self._story_to_signal(story) for story in stories.values()]
|
|
candidates = [candidate for candidate in candidates if candidate is not None]
|
|
candidates.sort(key=lambda item: item.published_at or datetime.min, reverse=True)
|
|
logger.info("news_candidates_collected days=%s items=%s", self.settings.news_recent_days, len(candidates))
|
|
return candidates[: self.settings.news_max_items]
|
|
|
|
def _queries(self) -> list[str]:
|
|
# 查询词覆盖模型、Agent、开源AI和融资产品动态，避免单一关键词漏掉日报候选。
|
|
return [
|
|
"AI OR LLM OR agent",
|
|
"large language model",
|
|
"OpenAI Anthropic",
|
|
"AI startup funding",
|
|
]
|
|
|
|
def _search(self, query: str, since: date, until: date) -> dict[str, Any]:
|
|
# search_by_date接口稳定返回最新故事，numericFilters把候选限制在本次日报窗口内。
|
|
start_at = int(datetime.combine(since, time.min, tzinfo=BEIJING_TZ).timestamp())
|
|
end_at = int(datetime.combine(until, time.max, tzinfo=BEIJING_TZ).timestamp())
|
|
params = {
|
|
"query": query,
|
|
"tags": "story",
|
|
"hitsPerPage": str(self.settings.news_search_max_records),
|
|
"numericFilters": f"created_at_i>={start_at},created_at_i<={end_at}",
|
|
}
|
|
url = f"{self.endpoint}?{urlencode(params)}"
|
|
request = Request(url, headers={"User-Agent": "SignalScout"})
|
|
try:
|
|
with urlopen(request, timeout=20) as response:
|
|
body = response.read().decode("utf-8")
|
|
except HTTPError as exc:
|
|
message = exc.read().decode("utf-8", errors="replace")
|
|
raise RuntimeError(f"News search failed: HTTP {exc.code} {message}") from exc
|
|
except URLError as exc:
|
|
raise RuntimeError(f"News search failed: {exc.reason}") from exc
|
|
return json.loads(body)
|
|
|
|
def _story_to_signal(self, story: dict[str, Any]) -> Optional[SignalItem]:
|
|
# HN故事先保留标题、链接和互动指标，重要性与摘要由模型最终确定。
|
|
title = story.get("title")
|
|
url = self._story_url(story)
|
|
if not isinstance(title, str) or not isinstance(url, str):
|
|
return None
|
|
author = str(story.get("author") or "unknown")
|
|
points = int(story.get("points") or 0)
|
|
comments = int(story.get("num_comments") or 0)
|
|
published_at = self._parse_hn_date(story.get("created_at"))
|
|
return SignalItem(
|
|
source_type="news",
|
|
topic="AI 新闻候选",
|
|
title=title[:300],
|
|
summary=f"Hacker News 最新技术讨论，作者 {author}，约 {points} points、{comments} 条评论。",
|
|
source_url=url,
|
|
source_name="Hacker News",
|
|
published_at=published_at,
|
|
importance=3,
|
|
entities=["Hacker News", author],
|
|
)
|
|
|
|
def _story_url(self, story: dict[str, Any]) -> Optional[str]:
|
|
# 外链优先；没有外链的Ask/Show HN故事使用HN讨论页作为证据链接。
|
|
url = story.get("url") or story.get("story_url")
|
|
if isinstance(url, str) and url.startswith("http"):
|
|
return url
|
|
object_id = story.get("objectID")
|
|
if object_id:
|
|
return f"https://news.ycombinator.com/item?id={object_id}"
|
|
return None
|
|
|
|
def _parse_hn_date(self, value: Any) -> Optional[datetime]:
|
|
# HN Algolia返回UTC ISO时间，入库前先换算为北京时间。
|
|
if not isinstance(value, str):
|
|
return None
|
|
try:
|
|
return to_beijing_naive(datetime.fromisoformat(value.replace("Z", "+00:00")))
|
|
except ValueError:
|
|
return None
|