207 lines
7.1 KiB
Python
207 lines
7.1 KiB
Python
"""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]
|