Codex
117 lines
5.3 KiB
Python
117 lines
5.3 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.parse import urlencode
|
||
from urllib.request import Request, urlopen
|
||
|
||
from app.core.config import Settings
|
||
from app.schemas.agent import SignalItem
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class GitHubHotProjectClient:
|
||
"""Collects recently active, high-signal GitHub AI repositories."""
|
||
|
||
endpoint = "https://api.github.com/search/repositories"
|
||
|
||
def __init__(self, settings: Settings) -> None:
|
||
self.settings = settings
|
||
|
||
def collect_hot_projects(self, run_date: date) -> list[SignalItem]:
|
||
# GitHub官方搜索提供开源项目热度候选，最终是否入选交给模型判断。
|
||
since = run_date - timedelta(days=self.settings.github_recent_days)
|
||
repositories: dict[str, dict[str, Any]] = {}
|
||
for query in self._queries(since=since):
|
||
payload = self._search(query)
|
||
for repository in payload.get("items", []):
|
||
html_url = repository.get("html_url")
|
||
if isinstance(html_url, str):
|
||
repositories[html_url] = repository
|
||
|
||
ranked = sorted(repositories.values(), key=lambda item: self._heat_score(item, run_date), reverse=True)
|
||
projects = [self._to_signal_item(repository) for repository in ranked[: self.settings.github_max_projects]]
|
||
logger.info(
|
||
"github_hot_projects_collected days=%s min_stars=%s projects=%s",
|
||
self.settings.github_recent_days,
|
||
self.settings.github_min_stars,
|
||
len(projects),
|
||
)
|
||
return projects
|
||
|
||
def _queries(self, since: date) -> list[str]:
|
||
# 查询集聚焦新近创建的AI项目，避免长期头部仓库挤占每日发现。
|
||
since_text = since.isoformat()
|
||
min_stars = self.settings.github_min_stars
|
||
return [
|
||
f"topic:llm created:>={since_text} stars:>={min_stars}",
|
||
f"topic:ai-agent created:>={since_text} stars:>={min_stars}",
|
||
f"topic:rag created:>={since_text} stars:>={min_stars}",
|
||
f"llm in:name,description created:>={since_text} stars:>={min_stars}",
|
||
f"llm agent in:name,description created:>={since_text} stars:>={min_stars}",
|
||
]
|
||
|
||
def _search(self, query: str) -> dict[str, Any]:
|
||
# GitHub REST搜索直接返回排序后的仓库元数据，便于后续模型研判。
|
||
url = f"{self.endpoint}?{urlencode({'q': query, 'sort': 'stars', 'order': 'desc', 'per_page': '10'})}"
|
||
request = Request(
|
||
url,
|
||
headers={
|
||
"Accept": "application/vnd.github+json",
|
||
"User-Agent": "SignalScout",
|
||
"X-GitHub-Api-Version": "2022-11-28",
|
||
},
|
||
)
|
||
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"GitHub project search failed: HTTP {exc.code} {message}") from exc
|
||
except URLError as exc:
|
||
raise RuntimeError(f"GitHub project search failed: {exc.reason}") from exc
|
||
return json.loads(body)
|
||
|
||
def _to_signal_item(self, repository: dict[str, Any]) -> SignalItem:
|
||
# 仓库元数据先标准化为信号候选，后续由模型统一筛选和改写摘要。
|
||
full_name = str(repository.get("full_name") or repository.get("name") or "Unknown repository")
|
||
description = str(repository.get("description") or "暂无项目描述")
|
||
stars = int(repository.get("stargazers_count") or 0)
|
||
forks = int(repository.get("forks_count") or 0)
|
||
language = repository.get("language") or "未标注语言"
|
||
created_at = self._parse_datetime(repository.get("created_at"))
|
||
pushed_at = self._parse_datetime(repository.get("pushed_at"))
|
||
freshness = created_at.date().isoformat() if created_at else "最近"
|
||
summary = (
|
||
f"{description} 该项目创建于 {freshness}，当前约 {stars} stars、{forks} forks，"
|
||
f"主要语言为 {language}。"
|
||
)
|
||
return SignalItem(
|
||
topic="GitHub 热门项目",
|
||
title=full_name,
|
||
summary=summary,
|
||
source_url=str(repository["html_url"]),
|
||
source_name="GitHub",
|
||
published_at=created_at or pushed_at,
|
||
importance=3,
|
||
entities=[full_name, "GitHub", str(language)],
|
||
)
|
||
|
||
def _heat_score(self, repository: dict[str, Any], run_date: date) -> int:
|
||
# Stars表示关注度，forks表示采用迹象，新项目获得额外新鲜度权重。
|
||
stars = int(repository.get("stargazers_count") or 0)
|
||
forks = int(repository.get("forks_count") or 0)
|
||
created_at = self._parse_datetime(repository.get("created_at"))
|
||
recency_bonus = 100 if created_at and created_at.date() >= run_date - timedelta(days=7) else 0
|
||
return stars * 3 + forks + recency_bonus
|
||
|
||
def _parse_datetime(self, value: Any) -> Optional[datetime]:
|
||
# GitHub时间戳使用UTC ISO格式和Z后缀。
|
||
if not isinstance(value, str):
|
||
return None
|
||
return datetime.fromisoformat(value.replace("Z", "+00:00")).replace(tzinfo=None)
|