yumao
f533e31adb
daily 不再通过 sys.path 导入 bot/skills_service，skills feed 数据层上移到 shared/。 Co-authored-by: Cursor <cursoragent@cursor.com>
110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
"""skills.sh feed.json 拉取、缓存与榜单数据。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
import certifi
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
FEED_URLS = [
|
|
"https://cdn.jsdelivr.net/gh/NeverSight/skills.sh_feed@main/data/feed.json",
|
|
"https://raw.githubusercontent.com/NeverSight/skills.sh_feed/main/data/feed.json",
|
|
]
|
|
CACHE_TTL_SECONDS = 600
|
|
CACHE_DIR = ROOT / ".cache"
|
|
CACHE_FILE = CACHE_DIR / "skills-feed.json"
|
|
USER_AGENT = "Mozilla/5.0 (compatible; daily-robots/1.0; +https://skills.sh)"
|
|
|
|
_cache: dict[str, Any] = {"data": None, "fetched_at": 0.0}
|
|
|
|
Board = Literal["trending", "hot", "all"]
|
|
|
|
BOARD_FEED_KEYS = {
|
|
"trending": "topTrending",
|
|
"hot": "topHot",
|
|
"all": "topAllTime",
|
|
}
|
|
|
|
|
|
def _fetch_json(url: str) -> dict[str, Any]:
|
|
headers = {"User-Agent": USER_AGENT, "Accept": "application/json"}
|
|
with httpx.Client(
|
|
timeout=httpx.Timeout(20.0, connect=10.0),
|
|
verify=certifi.where(),
|
|
follow_redirects=True,
|
|
) as client:
|
|
resp = client.get(url, headers=headers)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def _load_disk_cache() -> dict[str, Any] | None:
|
|
if not CACHE_FILE.exists():
|
|
return None
|
|
try:
|
|
return json.loads(CACHE_FILE.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
logger.warning("读取 skills feed 本地缓存失败: %s", exc)
|
|
return None
|
|
|
|
|
|
def _save_disk_cache(data: dict[str, Any]) -> None:
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
CACHE_FILE.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def load_feed(force: bool = False) -> dict[str, Any]:
|
|
now = time.time()
|
|
if not force and _cache["data"] and now - _cache["fetched_at"] < CACHE_TTL_SECONDS:
|
|
return _cache["data"]
|
|
|
|
errors: list[str] = []
|
|
for url in FEED_URLS:
|
|
for attempt in range(3):
|
|
try:
|
|
data = _fetch_json(url)
|
|
_cache["data"] = data
|
|
_cache["fetched_at"] = now
|
|
_save_disk_cache(data)
|
|
logger.info("skills 数据已更新: %s", url)
|
|
return data
|
|
except Exception as exc:
|
|
msg = f"{url} (#{attempt + 1}): {exc}"
|
|
errors.append(msg)
|
|
logger.debug("拉取失败 %s", msg)
|
|
time.sleep(0.5 * (attempt + 1))
|
|
|
|
stale = _load_disk_cache()
|
|
if stale:
|
|
logger.warning("网络不可用，回退到本地 skills feed 缓存")
|
|
_cache["data"] = stale
|
|
_cache["fetched_at"] = now
|
|
return stale
|
|
|
|
raise RuntimeError(f"无法获取 skills 数据。最近错误: {errors[-1]}")
|
|
|
|
|
|
def warm_feed_cache() -> None:
|
|
"""启动时预加载，避免首条消息才触发网络请求。"""
|
|
load_feed(force=True)
|
|
|
|
|
|
def format_installs(n: int | float) -> str:
|
|
if n >= 1_000_000:
|
|
return f"{n / 1_000_000:.1f}M"
|
|
if n >= 1_000:
|
|
return f"{n / 1_000:.1f}K"
|
|
return str(int(n))
|
|
|
|
|
|
def board_items(feed: dict[str, Any], board: Board) -> list[dict[str, Any]]:
|
|
return feed.get(BOARD_FEED_KEYS[board], [])
|