feat: 核心模块增强 — 加密 + 运行时配置 + 日志缓冲
- crypto.py: API Key 加密/解密工具 - runtime_config.py: 运行时动态配置管理 - log_buffer.py: 内存日志缓冲区 - config.py: 新增加密配置项 - http_client.py: 增强重试和错误处理
This commit is contained in:
+91
-27
@@ -1,12 +1,14 @@
|
||||
"""采集路由。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from src.api.deps import require_admin_key
|
||||
from src.api.deps import require_admin
|
||||
from src.api.schemas import IngestBzzoiroRequest, IngestResponse, IngestUnderstatRequest, IngestInjuriesRequest, IngestSimpleResponse
|
||||
from src.data.config import BZZOIRO_LEAGUE_IDS, FDCO_TO_UNDERSTAT
|
||||
from src.data.sources import get_source
|
||||
from src.data.injuries import ingest_injuries
|
||||
from src.db.unit_of_work import get_uow
|
||||
@@ -15,46 +17,108 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/v1", tags=["ingest"])
|
||||
|
||||
# 后台采集任务注册表:持强引用防止被 GC
|
||||
_background_tasks: set[asyncio.Task] = set()
|
||||
|
||||
@router.post("/ingest/bzzoiro", response_model=IngestResponse, dependencies=[Depends(require_admin_key)])
|
||||
|
||||
def _spawn(coro) -> None:
|
||||
"""启动后台采集任务;异常已在任务内记录到系统日志。"""
|
||||
task = asyncio.create_task(coro)
|
||||
_background_tasks.add(task)
|
||||
task.add_done_callback(_background_tasks.discard)
|
||||
|
||||
|
||||
@router.post("/ingest/bzzoiro", dependencies=[Depends(require_admin)])
|
||||
async def ingest_bzzoiro_route(req: IngestBzzoiroRequest):
|
||||
"""触发 bzzoiro 采集。"""
|
||||
source = get_source("bzzoiro")
|
||||
# 未指定联赛 = 采集全部已知联赛;未指定状态 = 已完赛 + 未开赛都采集
|
||||
leagues = req.leagues or list(BZZOIRO_LEAGUE_IDS.keys())
|
||||
statuses = [req.status] if req.status else ["finished", "scheduled"]
|
||||
_spawn(_run_bzzoiro(leagues, req.date_from, req.date_to, statuses))
|
||||
return {
|
||||
"ok": True,
|
||||
"message": f"采集任务已启动(后台执行,状态: {', '.join(statuses)}),请在「系统日志」查看进度与结果",
|
||||
}
|
||||
|
||||
|
||||
async def _run_bzzoiro(leagues: list[str], date_from: str | None, date_to: str | None, statuses: list[str]) -> None:
|
||||
"""后台执行 bzzoiro 采集:上游限速时单次可能耗时数分钟,必须脱离请求生命周期。"""
|
||||
try:
|
||||
source = get_source("bzzoiro")
|
||||
merged: dict = {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
|
||||
async with get_uow() as session:
|
||||
result = await source.ingest(
|
||||
session,
|
||||
leagues=req.leagues,
|
||||
date_from=req.date_from,
|
||||
date_to=req.date_to,
|
||||
status=req.status,
|
||||
)
|
||||
return IngestResponse(**result)
|
||||
except Exception as e:
|
||||
logger.exception("bzzoiro ingest failed")
|
||||
raise HTTPException(500, "数据采集失败,请查看服务器日志")
|
||||
for st in statuses:
|
||||
r = await source.ingest(
|
||||
session,
|
||||
leagues=leagues,
|
||||
date_from=date_from,
|
||||
date_to=date_to,
|
||||
status=st,
|
||||
)
|
||||
merged["total_inserted"] += r.get("total_inserted", 0)
|
||||
merged["total_updated"] += r.get("total_updated", 0)
|
||||
merged["errors"].extend(r.get("errors", []))
|
||||
for code, stat in r.get("leagues", {}).items():
|
||||
acc = merged["leagues"].setdefault(code, {"inserted": 0, "updated": 0, "errors": []})
|
||||
acc["inserted"] += stat.get("inserted", 0)
|
||||
acc["updated"] += stat.get("updated", 0)
|
||||
acc["errors"].extend(stat.get("errors", []))
|
||||
league_errors = {c: stat["errors"] for c, stat in merged["leagues"].items() if stat.get("errors")}
|
||||
logger.info(
|
||||
"bzzoiro 采集完成: 新增 %d, 更新 %d, 联赛 %d 个, 状态 %s",
|
||||
merged["total_inserted"], merged["total_updated"], len(merged["leagues"]), statuses,
|
||||
)
|
||||
if league_errors:
|
||||
sample = {c: errs[:1] for c, errs in list(league_errors.items())[:3]}
|
||||
logger.warning("bzzoiro 部分联赛存在错误: %s", sample)
|
||||
if merged["errors"]:
|
||||
logger.warning("bzzoiro 采集错误 %d 条: %s", len(merged["errors"]), merged["errors"][:3])
|
||||
except Exception:
|
||||
logger.exception("bzzoiro 采集任务失败")
|
||||
|
||||
|
||||
@router.post("/ingest/understat", response_model=IngestSimpleResponse, dependencies=[Depends(require_admin_key)])
|
||||
@router.post("/ingest/understat", dependencies=[Depends(require_admin)])
|
||||
async def ingest_understat_route(req: IngestUnderstatRequest):
|
||||
"""触发 understat xG 回填。"""
|
||||
source = get_source("understat")
|
||||
leagues_to_run = [req.league] if req.league else list(FDCO_TO_UNDERSTAT.keys())
|
||||
_spawn(_run_understat(leagues_to_run, req.season))
|
||||
return {"ok": True, "message": "xG 回填任务已启动(后台执行),请在「系统日志」查看结果"}
|
||||
|
||||
|
||||
async def _run_understat(leagues_to_run: list[str], season: int) -> None:
|
||||
try:
|
||||
source = get_source("understat")
|
||||
merged: dict = {"count": 0, "updated": 0, "skipped": 0, "unmatched": 0, "errors": []}
|
||||
async with get_uow() as session:
|
||||
result = await source.ingest(session, league=req.league, season=req.season)
|
||||
return IngestSimpleResponse(**result)
|
||||
except Exception as e:
|
||||
logger.exception("understat ingest failed")
|
||||
raise HTTPException(500, "xG 回填失败,请查看服务器日志")
|
||||
for league in leagues_to_run:
|
||||
r = await source.ingest(session, league=league, season=season)
|
||||
for k in ("count", "updated", "skipped", "unmatched"):
|
||||
merged[k] += r.get(k, 0)
|
||||
merged["errors"].extend(r.get("errors", []))
|
||||
logger.info(
|
||||
"understat 回填完成: 联赛 %d 个, 更新 %d, 未匹配 %d, 错误 %d",
|
||||
len(leagues_to_run), merged["updated"], merged["unmatched"], len(merged["errors"]),
|
||||
)
|
||||
except Exception:
|
||||
logger.exception("understat 回填任务失败")
|
||||
|
||||
|
||||
@router.post("/ingest/injuries", response_model=IngestSimpleResponse, dependencies=[Depends(require_admin_key)])
|
||||
@router.post("/ingest/injuries", dependencies=[Depends(require_admin)])
|
||||
async def ingest_injuries_route(req: IngestInjuriesRequest):
|
||||
"""触发伤停采集。"""
|
||||
_spawn(_run_injuries(req.date))
|
||||
return {"ok": True, "message": "伤停采集任务已启动(后台执行),请在「系统日志」查看结果"}
|
||||
|
||||
|
||||
async def _run_injuries(date: str | None) -> None:
|
||||
try:
|
||||
async with get_uow() as session:
|
||||
result = await ingest_injuries(session, date=req.date)
|
||||
return IngestSimpleResponse(**result)
|
||||
except Exception as e:
|
||||
logger.exception("injuries ingest failed")
|
||||
raise HTTPException(500, "伤停采集失败,请查看服务器日志")
|
||||
result = await ingest_injuries(session, date=date)
|
||||
logger.info(
|
||||
"injuries 采集完成: 新增 %d, 更新 %d, 错误 %d",
|
||||
result.get("count", 0), result.get("updated", 0), len(result.get("errors", [])),
|
||||
)
|
||||
if result.get("errors"):
|
||||
logger.warning("injuries 采集错误: %s", result["errors"][:3])
|
||||
except Exception:
|
||||
logger.exception("injuries 采集任务失败")
|
||||
|
||||
Reference in New Issue
Block a user