daily-robots/daily/github/search.py

"""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 []

    target = max(1, min(int(limit), 1000))
    per_page = min(100, target)
    repos: list[dict[str, Any]] = []
    seen: set[str] = set()
    page = 1

    try:
        with httpx.Client(
            timeout=20.0,
            verify=certifi.where(),
            headers=github_api_headers(),
        ) as client:
            while len(repos) < target:
                resp = client.get(
                    "https://api.github.com/search/repositories",
                    params={
                        "q": query,
                        "sort": sort,
                        "order": "desc",
                        "per_page": per_page,
                        "page": page,
                    },
                )
                if resp.status_code != 200:
                    logger.warning(
                        "GitHub Search 失败 (%s page=%s): %s",
                        resp.status_code,
                        page,
                        query[:80],
                    )
                    break
                items = resp.json().get("items") or []
                if not items:
                    break
                for item in items:
                    full_name = item.get("full_name") or ""
                    if not full_name or full_name in seen:
                        continue
                    seen.add(full_name)
                    repos.append(_repo_from_api_item(item, source="api-search"))
                    if len(repos) >= target:
                        break
                if len(items) < per_page:
                    break
                page += 1
                # Search API 最多约 1000 条 / 10 页
                if page > 10:
                    break
    except Exception as exc:
        logger.warning("GitHub Search 异常: %s", exc)
        return repos[:target]

    return repos[:target]


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]