122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
"""GitHub Search API：新兴项目、Topic 热点。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
import certifi
|
||
import httpx
|
||
|
||
from daily.config import env, env_int
|
||
from daily.github.auth import format_star_count, github_api_headers, github_token
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _repo_from_api_item(item: dict[str, Any], *, source: str) -> dict[str, Any]:
|
||
full_name = item.get("full_name") or ""
|
||
stars = item.get("stargazers_count", 0)
|
||
created = (item.get("created_at") or "")[:10]
|
||
return {
|
||
"repo": full_name,
|
||
"url": item.get("html_url") or f"https://github.com/{full_name}",
|
||
"description": item.get("description") or "",
|
||
"language": item.get("language") or "",
|
||
"stars_today": None,
|
||
"stars_today_fmt": "",
|
||
"total_stars_fmt": format_star_count(stars),
|
||
"created_at": created,
|
||
"source": source,
|
||
}
|
||
|
||
|
||
def search_github_repos(
|
||
query: str,
|
||
limit: int,
|
||
*,
|
||
sort: str = "stars",
|
||
require_token: bool = True,
|
||
) -> list[dict[str, Any]]:
|
||
if require_token and not github_token():
|
||
logger.warning("GitHub Search 需要 GITHUB_TOKEN: %s", query[:80])
|
||
return []
|
||
|
||
try:
|
||
with httpx.Client(
|
||
timeout=20.0,
|
||
verify=certifi.where(),
|
||
headers=github_api_headers(),
|
||
) as client:
|
||
resp = client.get(
|
||
"https://api.github.com/search/repositories",
|
||
params={
|
||
"q": query,
|
||
"sort": sort,
|
||
"order": "desc",
|
||
"per_page": min(max(limit, 1), 30),
|
||
},
|
||
)
|
||
if resp.status_code != 200:
|
||
logger.warning("GitHub Search 失败 (%s): %s", resp.status_code, query[:80])
|
||
return []
|
||
items = resp.json().get("items") or []
|
||
except Exception as exc:
|
||
logger.warning("GitHub Search 异常: %s", exc)
|
||
return []
|
||
|
||
repos: list[dict[str, Any]] = []
|
||
for item in items:
|
||
full_name = item.get("full_name") or ""
|
||
if not full_name:
|
||
continue
|
||
repos.append(_repo_from_api_item(item, source="api-search"))
|
||
if len(repos) >= limit:
|
||
break
|
||
return repos
|
||
|
||
|
||
def _date_days_ago(days: int) -> str:
|
||
return (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")
|
||
|
||
|
||
def fetch_emerging_repos(
|
||
limit: int = 3,
|
||
*,
|
||
days: int | None = None,
|
||
min_stars: int | None = None,
|
||
exclude: set[str] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
days = days if days is not None else env_int("GITHUB_EMERGING_DAYS", 14)
|
||
min_stars = min_stars if min_stars is not None else env_int("GITHUB_EMERGING_MIN_STARS", 200)
|
||
created_after = _date_days_ago(days)
|
||
query = f"created:>{created_after} stars:>{min_stars} fork:false"
|
||
repos = search_github_repos(query, limit + len(exclude or set()))
|
||
if exclude:
|
||
repos = [r for r in repos if r["repo"] not in exclude]
|
||
for item in repos:
|
||
item["source"] = "api-emerging"
|
||
return repos[:limit]
|
||
|
||
|
||
def fetch_topic_hot_repos(
|
||
limit: int = 3,
|
||
*,
|
||
topic: str | None = None,
|
||
pushed_days: int | None = None,
|
||
min_stars: int | None = None,
|
||
exclude: set[str] | None = None,
|
||
) -> tuple[str, list[dict[str, Any]]]:
|
||
topic = (topic or env("GITHUB_TOPIC") or "llm").strip()
|
||
pushed_days = pushed_days if pushed_days is not None else env_int("GITHUB_TOPIC_PUSHED_DAYS", 7)
|
||
min_stars = min_stars if min_stars is not None else env_int("GITHUB_TOPIC_MIN_STARS", 50)
|
||
pushed_after = _date_days_ago(pushed_days)
|
||
query = f"topic:{topic} pushed:>{pushed_after} stars:>{min_stars} fork:false"
|
||
repos = search_github_repos(query, limit + len(exclude or set()))
|
||
if exclude:
|
||
repos = [r for r in repos if r["repo"] not in exclude]
|
||
for item in repos:
|
||
item["source"] = "api-topic"
|
||
return topic, repos[:limit]
|