Profeto/src/api/routes/ingest.py

"""采集路由(bzzoiro 单一数据源)。

任务类型:
  events    — 比赛日程/比分(/events/)
  standings — 联赛积分榜(/leagues/{id}/standings/)
  stats     — 已完赛比赛详细统计回填(/events/{id}/stats/)
  all       — 依次执行以上三项
"""
from __future__ import annotations

import asyncio
import logging
import uuid
from datetime import datetime, timezone

from fastapi import APIRouter, Depends, HTTPException

from src.api.deps import require_admin
from src.api.schemas import IngestBzzoiroRequest, IngestBzzoiroResponse
from src.data.config import BZZOIRO_LEAGUE_IDS
from src.data.bzzoiro_standings import ingest_bzzoiro_standings
from src.data.bzzoiro_stats import ingest_bzzoiro_event_stats
from src.data.sources import get_source
from src.db.unit_of_work import get_uow

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/v1", tags=["ingest"])

# 后台采集任务注册表:持强引用防止被 GC
_background_tasks: set[asyncio.Task] = set()

VALID_TASKS = {"events", "standings", "stats", "all"}


def _accumulate_ingest_result(merged: dict, code: str, r: dict) -> None:
    """P1-A: 累加单联赛采集结果。联赛级计数读 r["leagues"][code],顶层读 total_*。"""
    merged["total_inserted"] += r.get("total_inserted", 0)
    merged["total_updated"] += r.get("total_updated", 0)
    merged["errors"].extend(r.get("errors", []))
    # 联赛级计数必须来自 leagues[code],而非顶层 r.get("inserted")
    league_r = r.get("leagues", {}).get(code, {})
    acc = merged["leagues"].setdefault(code, {"inserted": 0, "updated": 0, "errors": []})
    acc["inserted"] += league_r.get("inserted", 0)
    acc["updated"] += league_r.get("updated", 0)
    acc["errors"].extend(r.get("errors", []))


def _spawn(coro) -> None:
    """启动后台采集任务;异常已在任务内记录到系统日志。"""
    task = asyncio.create_task(coro)
    _background_tasks.add(task)
    task.add_done_callback(_background_tasks.discard)


@router.post("/ingest/bzzoiro", response_model=IngestBzzoiroResponse, dependencies=[Depends(require_admin)])
async def ingest_bzzoiro_route(req: IngestBzzoiroRequest):
    """触发 bzzoiro 采集(events / standings / stats / all)。

    启动后台任务前写入 ingest_jobs(pending),响应返回 job_id 供前端轮询。
    兼容原 message 字段(仍返回)。
    """
    if req.task not in VALID_TASKS:
        raise HTTPException(status_code=422, detail=f"未知任务类型: {req.task}(可选: {', '.join(sorted(VALID_TASKS))})")
    leagues = req.leagues or list(BZZOIRO_LEAGUE_IDS.keys())
    task_label = {"events": "比赛数据", "standings": "积分榜", "stats": "统计回填", "all": "全量(比赛+积分榜+统计)"}[req.task]

    job_id = await _create_ingest_job(req.task, leagues, req)
    _spawn(_run_bzzoiro(job_id, req.task, leagues, req))

    return IngestBzzoiroResponse(
        ok=True,
        job_id=job_id,
        message=f"采集任务已启动(后台执行,任务: {task_label}),请到「数据采集」页跟踪进度",
    )


async def _create_ingest_job(task: str, leagues: list[str], req: IngestBzzoiroRequest) -> str:
    """写入一条 ingest_jobs(pending),返回 job_id。"""
    from src.db.models import IngestJob

    job_id = str(uuid.uuid4())
    params = {
        "leagues": leagues,
        "date_from": req.date_from,
        "date_to": req.date_to,
        "status": req.status,
        "task": task,
        "limit": req.limit,
        "season": req.season,
    }
    async with get_uow() as session:
        job = IngestJob(id=job_id, task=task, params=params, status="pending")
        session.add(job)
    logger.info("ingest_jobs 创建: job=%s task=%s leagues=%s", job_id, task, leagues)
    return job_id


async def _run_bzzoiro(job_id: str, task: str, leagues: list[str], req: IngestBzzoiroRequest) -> None:
    """后台执行 bzzoiro 采集:上游限速时单次可能耗时数分钟,必须脱离请求生命周期。

    状态流转: pending → running → (success|failed)。
    """
    from src.db.models import IngestJob

    await _update_job(job_id, status="running", started_at=datetime.now(timezone.utc))
    result: dict = {}
    try:
        if task in ("events", "all"):
            statuses = [req.status] if req.status else ["finished", "scheduled"]
            source = get_source("bzzoiro")
            merged: dict = {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
            # D2 修复: 按联赛分批提交,避免超长事务
            for code in leagues:
                for st in statuses:
                    async with get_uow() as session:
                        r = await source.ingest(
                            session, leagues=[code],
                            date_from=req.date_from, date_to=req.date_to, status=st,
                        )
                        _accumulate_ingest_result(merged, code, r)
            logger.info(
                "bzzoiro 比赛采集完成: 新增 %d, 更新 %d, 联赛 %d 个, 状态 %s",
                merged["total_inserted"], merged["total_updated"], len(merged["leagues"]), statuses,
            )
            if merged["errors"]:
                logger.warning("bzzoiro 比赛采集错误 %d 条: %s", len(merged["errors"]), merged["errors"][:3])
            result["events"] = merged

        if task in ("standings", "all"):
            async with get_uow() as session:
                r = await ingest_bzzoiro_standings(session, leagues=leagues, season=req.season)
            if r["errors"]:
                logger.warning("bzzoiro 积分榜采集部分失败: %s", r["errors"][:3])
            else:
                logger.info("bzzoiro 积分榜采集完成: upsert %d 条", r["total_upserted"])
            result["standings"] = r

        if task in ("stats", "all"):
            async with get_uow() as session:
                r = await ingest_bzzoiro_event_stats(
                    session, leagues=leagues, limit=req.limit, only_missing=True
                )
            if r["errors"]:
                logger.warning("bzzoiro 统计回填错误 %d 条: %s", len(r["errors"]), r["errors"][:3])
            result["stats"] = r

        await _update_job(job_id, status="success", result=result, finished_at=datetime.now(timezone.utc))
        logger.info("ingest_jobs 完成: job=%s task=%s", job_id, task)
    except Exception as e:
        logger.exception("bzzoiro 采集任务失败(task=%s)", task)
        await _update_job(
            job_id, status="failed", error=str(e), finished_at=datetime.now(timezone.utc),
        )


async def _update_job(job_id: str, **fields) -> None:
    """更新 ingest_jobs 单行;失败仅记日志,绝不抛异常(避免干扰采集主流程)。"""
    from src.db.models import IngestJob

    try:
        async with get_uow() as session:
            job = await session.get(IngestJob, job_id)
            if job is None:
                logger.warning("ingest_jobs 更新失败: job=%s 不存在", job_id)
                return
            for k, v in fields.items():
                setattr(job, k, v)
    except Exception:
        logger.warning("ingest_jobs 更新异常: job=%s fields=%s", job_id, list(fields.keys()))