shangfangjian/Profeto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 数据采集健壮性改进

Browse Source 
- 新增 src/core/retry.py: 指数退避 + 抖动的重试装饰器
- bzzoiro: 改进重试逻辑
  - 429/5xx/网络错误均触发指数退避重试
  - 4xx 直接抛(不重试)
- understat: 新增重试(原无重试)
- injuries: 新增重试 + 文件缓存 7 天过期
- sources.py: 延迟导入修复循环导入
This commit is contained in:
shangfangjian
2026-09-09 21:47:56 +08:00
parent 1166a157ec
commit 494751bbb0
5 changed files with 166 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/retry.py
+82
View File
@@ -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
src/data/bzzoiro.py
+19 -4
View File
@@ -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(
src/data/injuries.py
+30 -8
View File
@@ -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", [])
# 写缓存
src/data/sources.py
+15 -3
View File
@@ -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
src/data/understat.py
+20 -3
View File
@@ -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('...');