daily-robots/daily/github/trending.py

"""GitHub Trending：页面爬取或 Search API。"""

from __future__ import annotations

import logging
import re
from datetime import datetime, timedelta, timezone
from html import unescape
from typing import Any, Literal
from urllib.parse import urlencode

import certifi
import httpx

from daily.config import env
from daily.github.auth import (
    fetch_repo_api,
    format_star_count,
    github_html_headers,
    github_token,
)
from daily.github.search import search_github_repos

logger = logging.getLogger(__name__)

TrendingSince = Literal["daily", "weekly", "monthly"]
TrendingMode = Literal["scrape", "api"]

DEFAULT_LIMIT = 5
DEFAULT_SINCE: TrendingSince = "daily"
DEFAULT_MODE: TrendingMode = "scrape"
TRENDING_URL = "https://github.com/trending"

_ARTICLE_RE = re.compile(r'<article class="Box-row">.*?</article>', re.S)
_REPO_HREF_RE = re.compile(r'h2[^>]*>\s*<a[^>]+href="([^"]+)"')
_DESC_RE = re.compile(r'<p class="col-9[^"]*"[^>]*>([^<]*)</p>')
_STARS_TODAY_RE = re.compile(r"([\d,]+)\s+stars?\s+today", re.I)
_LANG_RE = re.compile(r'itemprop="programmingLanguage"[^>]*>([^<]+)<')
_TOTAL_STARS_RE = re.compile(
    r'href="/[^/]+/[^/]+/stargazers"[^>]*>\s*<svg[^>]*octicon-star[^>]*>.*?</svg>\s*([\d.,kKmM]+)',
    re.S,
)


def _strip_html(text: str) -> str:
    return unescape(re.sub(r"\s+", " ", text or "")).strip()


def trending_mode() -> TrendingMode:
    raw = (env("GITHUB_TRENDING_MODE") or DEFAULT_MODE).strip().lower()
    if raw in {"api", "token", "search"}:
        return "api"
    return "scrape"


def trending_data_source_note() -> str:
    if trending_mode() == "api":
        return "> 数据来源：GitHub Search API（`GITHUB_TRENDING_MODE=api`，需 `GITHUB_TOKEN`）"
    return "> 数据来源：[github.com/trending](https://github.com/trending?since=daily)（页面爬取，失败时 API 降级）"


def _parse_article(article_html: str) -> dict[str, Any] | None:
    href_match = _REPO_HREF_RE.search(article_html)
    if not href_match:
        return None
    href = href_match.group(1).strip("/")
    if href.count("/") != 1:
        return None
    owner, name = href.split("/", 1)
    repo = f"{owner}/{name}"
    desc_match = _DESC_RE.search(article_html)
    stars_today_match = _STARS_TODAY_RE.search(article_html)
    lang_match = _LANG_RE.search(article_html)
    total_stars_match = _TOTAL_STARS_RE.search(article_html)
    stars_today_raw = stars_today_match.group(1).replace(",", "") if stars_today_match else ""
    stars_today = int(stars_today_raw) if stars_today_raw.isdigit() else None
    return {
        "repo": repo,
        "url": f"https://github.com/{repo}",
        "description": _strip_html(desc_match.group(1)) if desc_match else "",
        "language": _strip_html(lang_match.group(1)) if lang_match else "",
        "stars_today": stars_today,
        "stars_today_fmt": stars_today_match.group(1) if stars_today_match else "",
        "total_stars_fmt": _strip_html(total_stars_match.group(1)) if total_stars_match else "",
        "source": "scrape",
    }


def _build_trending_url(*, since: TrendingSince = DEFAULT_SINCE, language: str = "") -> str:
    if language:
        return f"{TRENDING_URL}/{language}?{urlencode({'since': since})}"
    return f"{TRENDING_URL}?{urlencode({'since': since})}"


def _since_push_date(since: TrendingSince) -> str:
    now = datetime.now(timezone.utc)
    if since == "weekly":
        delta = timedelta(days=7)
    elif since == "monthly":
        delta = timedelta(days=30)
    else:
        delta = timedelta(days=1)
    return (now - delta).strftime("%Y-%m-%d")


def _enrich_repo_from_api(item: dict[str, Any]) -> dict[str, Any]:
    if not github_token() and env("GITHUB_API_ENRICH", "1") != "1":
        return item
    meta = fetch_repo_api(item["repo"])
    if not meta:
        return item
    enriched = dict(item)
    if not enriched.get("description"):
        enriched["description"] = meta["description"]
    if not enriched.get("language"):
        enriched["language"] = meta["language"]
    if not enriched.get("total_stars_fmt") and meta["stars"]:
        enriched["total_stars_fmt"] = format_star_count(meta["stars"])
    enriched["source"] = enriched.get("source", "scrape") + "+api"
    return enriched


def _fetch_trending_html(url: str) -> str | None:
    try:
        with httpx.Client(
            timeout=20.0,
            verify=certifi.where(),
            follow_redirects=True,
            headers=github_html_headers(),
        ) as client:
            resp = client.get(url)
            if resp.status_code in {403, 429} and not github_token():
                logger.warning("GitHub Trending %s（匿名可能被限），可配置 GITHUB_TOKEN", resp.status_code)
            resp.raise_for_status()
            return resp.text
    except Exception as exc:
        logger.warning("GitHub Trending 页面抓取失败: %s", exc)
        return None


def _parse_trending_html(html: str, limit: int) -> list[dict[str, Any]]:
    repos: list[dict[str, Any]] = []
    for article_html in _ARTICLE_RE.findall(html):
        item = _parse_article(article_html)
        if item:
            repos.append(_enrich_repo_from_api(item))
        if len(repos) >= limit:
            break
    return repos


def _fetch_trending_via_api(limit: int, since: TrendingSince, language: str) -> list[dict[str, Any]]:
    pushed_after = _since_push_date(since)
    parts = [f"pushed:>{pushed_after}", "stars:>50", "fork:false"]
    if language:
        parts.append(f"language:{language}")
    query = " ".join(parts)
    repos = search_github_repos(query, limit, require_token=True)
    for item in repos:
        item["source"] = "api-search"
    return repos


def fetch_github_trending(
    limit: int = DEFAULT_LIMIT,
    *,
    since: TrendingSince | None = None,
    language: str = "",
) -> list[dict[str, Any]]:
    since = since or env("GITHUB_TRENDING_SINCE", DEFAULT_SINCE)  # type: ignore[assignment]
    if since not in ("daily", "weekly", "monthly"):
        since = DEFAULT_SINCE
    lang = (language or env("GITHUB_TRENDING_LANGUAGE") or "").strip()
    mode = trending_mode()

    if mode == "api":
        repos = _fetch_trending_via_api(limit, since, lang)
        if not repos and not github_token():
            logger.warning("GITHUB_TRENDING_MODE=api 需要配置 GITHUB_TOKEN")
        elif repos:
            logger.info("GitHub Trending 使用 API 模式，共 %d 条", len(repos))
        return repos[:limit]

    url = _build_trending_url(since=since, language=lang)
    html = _fetch_trending_html(url)
    repos: list[dict[str, Any]] = []
    if html:
        repos = _parse_trending_html(html, limit)
        if not repos:
            logger.warning("GitHub Trending 页面解析为空: %s", url)

    if len(repos) < limit:
        before = len(repos)
        fallback = _fetch_trending_via_api(limit, since, lang)
        seen = {r["repo"] for r in repos}
        for item in fallback:
            if item["repo"] in seen:
                continue
            repos.append(item)
            seen.add(item["repo"])
            if len(repos) >= limit:
                break
        if len(repos) > before:
            logger.info("已用 GitHub API 补充 %d 条 Trending 数据", len(repos) - before)

    return repos[:limit]