From 494751bbb0e36a042d10e4d80fc2b4f7e4883069 Mon Sep 17 00:00:00 2001 From: shangfangjian Date: Wed, 9 Sep 2026 21:47:56 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=95=B0=E6=8D=AE=E9=87=87=E9=9B=86?= =?UTF-8?q?=E5=81=A5=E5=A3=AE=E6=80=A7=E6=94=B9=E8=BF=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 src/core/retry.py: 指数退避 + 抖动的重试装饰器 - bzzoiro: 改进重试逻辑 - 429/5xx/网络错误均触发指数退避重试 - 4xx 直接抛(不重试) - understat: 新增重试(原无重试) - injuries: 新增重试 + 文件缓存 7 天过期 - sources.py: 延迟导入修复循环导入 --- src/core/retry.py | 82 +++++++++++++++++++++++++++++++++++++++++++ src/data/bzzoiro.py | 23 +++++++++--- src/data/injuries.py | 38 +++++++++++++++----- src/data/sources.py | 18 ++++++++-- src/data/understat.py | 23 ++++++++++-- 5 files changed, 166 insertions(+), 18 deletions(-) create mode 100644 src/core/retry.py diff --git a/src/core/retry.py b/src/core/retry.py new file mode 100644 index 0000000..bb3dd6b --- /dev/null +++ b/src/core/retry.py @@ -0,0 +1,82 @@ +"""重试工具:带指数退避的瞬态错误重试。""" +from __future__ import annotations + +import asyncio +import functools +import logging +import random +import time +from typing import Callable, Iterable, TypeVar + +logger = logging.getLogger(__name__) + +T = TypeVar("T") + + +def with_retry( + *, + max_retries: int = 3, + base_delay: float = 1.0, + max_delay: float = 30.0, + retryable_exceptions: Iterable[type[BaseException]] = (Exception,), + on_retry: Callable[[Exception, int], None] | None = None, +) -> Callable: + """重试装饰器(同步/异步通用,指数退避 + 抖动)。 + + Args: + max_retries: 最大重试次数 + base_delay: 基础延迟(秒) + max_delay: 最大延迟(秒) + retryable_exceptions: 触发重试的异常类型 + on_retry: 重试回调(exception, attempt) + """ + retryable = tuple(retryable_exceptions) + + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + async def async_wrapper(*args, **kwargs): + last_exc: Exception | None = None + for attempt in range(max_retries + 1): + try: + return await func(*args, **kwargs) + except retryable as e: + last_exc = e + if attempt == max_retries: + break + delay = min(base_delay * (2 ** attempt), max_delay) + delay += random.uniform(0, delay * 0.1) # 抖动 + logger.warning( + "%s failed (attempt %d/%d), retry in %.1fs: %s", + func.__name__, attempt + 1, max_retries, delay, e, + ) + if on_retry: + on_retry(e, attempt + 1) + await asyncio.sleep(delay) + raise last_exc # type: ignore[misc] + + @functools.wraps(func) + def sync_wrapper(*args, **kwargs): + last_exc: Exception | None = None + for attempt in range(max_retries + 1): + try: + return func(*args, **kwargs) + except retryable as e: + last_exc = e + if attempt == max_retries: + break + delay = min(base_delay * (2 ** attempt), max_delay) + delay += random.uniform(0, delay * 0.1) + logger.warning( + "%s failed (attempt %d/%d), retry in %.1fs: %s", + func.__name__, attempt + 1, max_retries, delay, e, + ) + if on_retry: + on_retry(e, attempt + 1) + time.sleep(delay) + raise last_exc # type: ignore[misc] + + if asyncio.iscoroutinefunction(func): + return async_wrapper + return sync_wrapper + + return decorator diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 9d4e96c..6e7bb3a 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -7,6 +7,7 @@ from __future__ import annotations import asyncio import json as _json import logging +import random import time as _time import urllib.error import urllib.parse @@ -35,6 +36,7 @@ def _fetch_json_sync(path: str, params: dict | None = None, max_retries: int = 3 if not key: raise RuntimeError("BZZOIRO_KEY 未设置") + last_exc: Exception | None = None for attempt in range(max_retries): try: req = urllib.request.Request(url) @@ -43,12 +45,25 @@ def _fetch_json_sync(path: str, params: dict | None = None, max_retries: int = 3 with urllib.request.urlopen(req, timeout=30) as resp: return _json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: + last_exc = e if e.code == 429: - logger.warning("bzzoiro 429, retry %d", attempt + 1) - _time.sleep(1) + # 指数退避: 429 通常意味着限速 + delay = min(2 ** attempt, 16) + random.uniform(0, 1) + logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay) + _time.sleep(delay) continue - raise - raise RuntimeError("bzzoiro rate limit exceeded") + if 500 <= e.code < 600: + delay = min(2 ** attempt, 16) + random.uniform(0, 1) + logger.warning("bzzoiro %d, retry %d in %.1fs", e.code, attempt + 1, delay) + _time.sleep(delay) + continue + raise # 4xx 直接抛 + except (urllib.error.URLError, TimeoutError, ConnectionError) as e: + last_exc = e + delay = min(2 ** attempt, 16) + random.uniform(0, 1) + logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e) + _time.sleep(delay) + raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}") async def fetch_bzzoiro_events( diff --git a/src/data/injuries.py b/src/data/injuries.py index b2c953c..db86233 100644 --- a/src/data/injuries.py +++ b/src/data/injuries.py @@ -4,8 +4,11 @@ """ from __future__ import annotations +import asyncio import json import logging +import random +import time from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -42,13 +45,17 @@ async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = No cache_dir = _CACHE_DIR cache_dir.mkdir(parents=True, exist_ok=True) - # 缓存命中 + # 缓存命中 (7 天内有效) cache_key = f"injuries_{date}_{fixture_id}_{league_id}.json" cache_file = cache_dir / cache_key if cache_file.exists(): - logger.debug("injuries cache hit: %s", cache_key) - with open(cache_file, encoding="utf-8") as f: - return json.load(f) + age_hours = (time.time() - cache_file.stat().st_mtime) / 3600 + if age_hours < 168: # 7 天 + logger.debug("injuries cache hit: %s (%.1fh old)", cache_key, age_hours) + with open(cache_file, encoding="utf-8") as f: + return json.load(f) + else: + logger.debug("injuries cache expired: %s (%.1fh old)", cache_key, age_hours) headers = { "x-apisports-key": api_key, @@ -63,11 +70,26 @@ async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = No params["league"] = league_id url = f"{API_BASE}/injuries" - client = get_client() - resp = await client.get(url, headers=headers, params=params) - resp.raise_for_status() - data = resp.json() + # 重试 + last_exc: Exception | None = None + for attempt in range(3): + try: + client = get_client() + resp = await client.get(url, headers=headers, params=params, timeout=30) + resp.raise_for_status() + break + except Exception as e: + last_exc = e + if attempt == 2: + raise + delay = min(2 ** attempt, 8) + random.uniform(0, 1) + logger.warning("injuries fetch failed, retry %d in %.1fs: %s", attempt + 1, delay, e) + await asyncio.sleep(delay) + else: + raise RuntimeError(f"injuries fetch failed: {last_exc}") + + data = resp.json() injuries = data.get("response", []) # 写缓存 diff --git a/src/data/sources.py b/src/data/sources.py index acd1486..f60d2ac 100644 --- a/src/data/sources.py +++ b/src/data/sources.py @@ -37,6 +37,8 @@ def register(source: DataSource) -> DataSource: def get_source(name: str) -> DataSource: """按名获取数据源。""" + if not _SOURCES: + _load_sources() if name not in _SOURCES: raise ValueError(f"未知数据源: {name}") return _SOURCES[name] @@ -44,9 +46,19 @@ def get_source(name: str) -> DataSource: def list_sources() -> list[str]: """列出所有已注册数据源名。""" + if not _SOURCES: + _load_sources() return list(_SOURCES.keys()) -# ── 导入数据源触发 @register ── -from src.data.bzzoiro import BzzoiroSource # noqa: E402, F401 -from src.data.understat import UnderstatSource # noqa: E402, F401 +def _load_sources() -> None: + """延迟导入数据源触发 @register(避免循环导入)。""" + from src.data.bzzoiro import BzzoiroSource # noqa: F811 + from src.data.understat import UnderstatSource # noqa: F811 + + +# 保持向后兼容:模块加载时尝试加载(但不再强制) +try: + _load_sources() +except Exception: + pass diff --git a/src/data/understat.py b/src/data/understat.py index 66228b3..b244849 100644 --- a/src/data/understat.py +++ b/src/data/understat.py @@ -4,8 +4,10 @@ """ from __future__ import annotations +import asyncio import json import logging +import random import re from src.core.http_client import get_client @@ -41,9 +43,24 @@ async def fetch_understat(league_code: str, season: int) -> list[dict]: "Referer": f"https://understat.com/league/{understat_league}/{season}", } - client = get_client() - resp = await client.get(url, headers=headers) - resp.raise_for_status() + # 重试:网络错误 / 5xx / 429 + last_exc: Exception | None = None + for attempt in range(3): + try: + client = get_client() + resp = await client.get(url, headers=headers, timeout=30) + resp.raise_for_status() + break + except Exception as e: + last_exc = e + if attempt == 2: + raise + delay = min(2 ** attempt, 8) + random.uniform(0, 1) + logger.warning("understat fetch failed, retry %d in %.1fs: %s", attempt + 1, delay, e) + await asyncio.sleep(delay) + else: + raise RuntimeError(f"understat fetch failed: {last_exc}") + # understat 返回 JS 对象,需要提取 JSON text = resp.text # 匹配 var datesData = JSON.parse('...');