121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
"""从 skills.sh 官网抓取 Trending / Hot 完整榜单（突破 feed.json 50 条限制）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from typing import Any, Literal
|
|
|
|
import certifi
|
|
import httpx
|
|
|
|
from daily.config import env
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
Board = Literal["trending", "hot"]
|
|
SKILLS_SITE = "https://www.skills.sh"
|
|
USER_AGENT = "Mozilla/5.0 (compatible; skills-hot-daily/1.0; +https://skills.sh)"
|
|
|
|
_SKILL_RE = re.compile(
|
|
r'\{"source":"(?P<source>[^"]+)","skillId":"(?P<skill_id>[^"]+)",'
|
|
r'"name":"(?P<name>[^"]+)","installs":(?P<installs>\d+)'
|
|
)
|
|
_RSC_CHUNK_RE = re.compile(r"self\.__next_f\.push\(\[1,\"(.*?)\"\]\)", re.DOTALL)
|
|
|
|
|
|
def board_source() -> str:
|
|
return (env("SKILLS_BOARD_SOURCE") or "website").strip().lower()
|
|
|
|
|
|
def _fetch_html(path: str) -> str:
|
|
url = f"{SKILLS_SITE}{path}"
|
|
headers = {"User-Agent": USER_AGENT, "Accept": "text/html"}
|
|
with httpx.Client(timeout=30.0, verify=certifi.where(), follow_redirects=True) as client:
|
|
resp = client.get(url, headers=headers)
|
|
resp.raise_for_status()
|
|
return resp.text
|
|
|
|
|
|
def _rsc_blob(html: str) -> str:
|
|
chunks = _RSC_CHUNK_RE.findall(html)
|
|
blob = "\n".join(chunks)
|
|
return blob.encode("utf-8").decode("unicode_escape", errors="ignore")
|
|
|
|
|
|
def _parse_initial_skills(blob: str, *, limit: int) -> list[dict[str, Any]]:
|
|
items: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
for match in _SKILL_RE.finditer(blob):
|
|
source = match.group("source")
|
|
skill_id = match.group("skill_id")
|
|
uid = f"{source}/{skill_id}"
|
|
if uid in seen:
|
|
continue
|
|
seen.add(uid)
|
|
items.append(
|
|
{
|
|
"id": uid,
|
|
"title": skill_id,
|
|
"source": source,
|
|
"installs": int(match.group("installs")),
|
|
"link": f"{SKILLS_SITE}/{source}/{skill_id}",
|
|
"description": "",
|
|
}
|
|
)
|
|
if len(items) >= limit:
|
|
break
|
|
return items
|
|
|
|
|
|
def fetch_board(board: Board, *, limit: int) -> list[dict[str, Any]]:
|
|
path = "/trending" if board == "trending" else "/hot"
|
|
try:
|
|
html = _fetch_html(path)
|
|
items = _parse_initial_skills(_rsc_blob(html), limit=limit)
|
|
if items:
|
|
logger.info("skills.sh %s: %d items (limit=%d)", board, len(items), limit)
|
|
return items
|
|
except Exception as exc:
|
|
logger.warning("skills.sh %s fetch failed, fallback to feed.json: %s", board, exc)
|
|
return []
|
|
|
|
|
|
def enrich_from_feed(items: list[dict[str, Any]], feed: dict[str, Any]) -> None:
|
|
desc_by_id: dict[str, str] = {}
|
|
for key in ("topTrending", "topHot", "topAllTime"):
|
|
for row in feed.get(key, []):
|
|
uid = str(row.get("id") or f"{row.get('source')}/{row.get('title')}")
|
|
desc = (row.get("description") or "").strip()
|
|
if desc:
|
|
desc_by_id[uid] = desc
|
|
for item in items:
|
|
if not item.get("description"):
|
|
item["description"] = desc_by_id.get(item["id"], "")
|
|
|
|
|
|
def load_boards(
|
|
feed: dict[str, Any],
|
|
*,
|
|
trending_limit: int,
|
|
hot_limit: int,
|
|
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
|
"""优先 skills.sh 官网；失败时回退 feed.json。"""
|
|
if board_source() == "feed":
|
|
return (
|
|
list(feed.get("topTrending", [])[:trending_limit]),
|
|
list(feed.get("topHot", [])[:hot_limit]),
|
|
)
|
|
|
|
trending = fetch_board("trending", limit=trending_limit)
|
|
hot = fetch_board("hot", limit=hot_limit)
|
|
if not trending:
|
|
trending = list(feed.get("topTrending", [])[:trending_limit])
|
|
else:
|
|
enrich_from_feed(trending, feed)
|
|
if not hot:
|
|
hot = list(feed.get("topHot", [])[:hot_limit])
|
|
else:
|
|
enrich_from_feed(hot, feed)
|
|
return trending, hot
|