diff --git a/src/api/routes/admin_config.py b/src/api/routes/admin_config.py new file mode 100644 index 0000000..117ddaa --- /dev/null +++ b/src/api/routes/admin_config.py @@ -0,0 +1,88 @@ +"""后台管理:配置项 CRUD(settings)与运行日志查询。 + +所有接口需管理员鉴权(require_admin)。路由前缀 /api/v1/admin。 +配置项白名单见 src/core/runtime_config.py SETTING_DEFS,之外的 key 一律拒绝。 +""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel + +from src.api.deps import require_admin +from src.core.log_buffer import get_entries +from src.core.runtime_config import ( + SETTING_DEFS, + clear_runtime_value, + get_setting_origin, + mask_value, + set_runtime_value, +) + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)]) + + +class SettingUpdateIn(BaseModel): + value: str + + +@router.get("/settings") +async def list_settings(): + """全部可配置项(脱敏),供后台各配置页渲染。""" + out = [] + for key, defn in SETTING_DEFS.items(): + origin, value = await get_setting_origin(key) + out.append( + { + "key": key, + "label": defn.label, + "description": defn.description, + "sensitive": defn.sensitive, + "configured": origin != "none", + "masked": mask_value(value, defn.sensitive), + "origin": origin, + } + ) + return out + + +@router.get("/logs") +async def read_logs( + level: str | None = Query(None, description="最低级别: DEBUG/INFO/WARNING/ERROR"), + keyword: str | None = Query(None, description="消息或 logger 关键字"), + limit: int = Query(200, ge=1, le=1000), +): + """查询应用运行日志(内存环形缓冲,最新在前;进程重启后清零)。""" + entries = get_entries(level, keyword, limit) + return {"entries": entries, "count": len(entries)} + + +@router.put("/settings/{key}") +async def update_setting(key: str, body: SettingUpdateIn): + """更新配置项(写入 app_settings 覆盖 .env)。传空值请改用 DELETE。""" + if key not in SETTING_DEFS: + raise HTTPException(404, f"不支持的配置项: {key}") + value = body.value.strip() + if not value: + raise HTTPException(400, "值不能为空;如需回落 .env 请调用清除接口") + await set_runtime_value(key, value) + defn = SETTING_DEFS[key] + return {"key": key, "masked": mask_value(value, defn.sensitive), "origin": "db"} + + +@router.delete("/settings/{key}") +async def clear_setting(key: str): + """清除 DB 覆盖值,回落 .env 默认。""" + if key not in SETTING_DEFS: + raise HTTPException(404, f"不支持的配置项: {key}") + await clear_runtime_value(key) + origin, value = await get_setting_origin(key) + defn = SETTING_DEFS[key] + return { + "key": key, + "masked": mask_value(value, defn.sensitive), + "origin": origin, + } diff --git a/src/api/routes/admin_datasources.py b/src/api/routes/admin_datasources.py new file mode 100644 index 0000000..462fe42 --- /dev/null +++ b/src/api/routes/admin_datasources.py @@ -0,0 +1,228 @@ +"""后台管理:数据源列表/连通性测试、KeyRing 状态、采集健康概览。 + +所有接口需管理员鉴权(require_admin)。路由前缀 /api/v1/admin。 +""" +from __future__ import annotations + +import logging +import time +from datetime import date, datetime + +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import func, select + +from src.api.deps import require_admin +from src.core.config import settings +from src.core.http_client import get_client +from src.core.runtime_config import ( + SETTING_DEFS, + get_runtime_value, + get_setting_origin, + mask_value, +) +from src.db.base import AsyncSession, get_db_read +from src.db.models import Match, MatchStats, Standing +from src.data.key_ring import get_key_ring + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)]) + +# ── 数据源元数据(bzzoiro 单一数据源) ──────────────────────────── + +_SOURCES: list[dict] = [ + { + "name": "bzzoiro", + "label": "Bzzoiro", + "description": "唯一数据源:赛程比分 + 积分榜 + 比赛详细统计(xG/射门/控球等)", + "setting_keys": ["BZZOIRO_KEY", "BZZOIRO_BASE"], + }, +] + + +async def _last_ingestion(db: AsyncSession, source: str) -> datetime | None: + """源最近一次采集时间(取自数据血缘字段,无记录返回 None)。""" + return ( + await db.execute( + select(func.max(MatchStats.retrieved_at)).where(MatchStats.source == source) + ) + ).scalar() + + +@router.get("/datasources") +async def list_datasources(db: AsyncSession = Depends(get_db_read)): + """数据源列表:各配置项的脱敏值、来源(db/env/none)与最近采集时间。""" + result = [] + for src in _SOURCES: + settings_out = [] + for key in src["setting_keys"]: + origin, value = await get_setting_origin(key) + defn = SETTING_DEFS[key] + settings_out.append( + { + "key": key, + "label": defn.label, + "description": defn.description, + "sensitive": defn.sensitive, + "configured": origin != "none", + "masked": mask_value(value, defn.sensitive), + "origin": origin, + } + ) + key_configured = all(s["configured"] for s in settings_out) if settings_out else True + last = await _last_ingestion(db, src["name"]) + result.append( + { + "name": src["name"], + "label": src["label"], + "description": src["description"], + "key_configured": key_configured, + "last_ingestion": last.isoformat() if last else None, + "settings": settings_out, + } + ) + return result + + +# ── 连通性测试 ────────────────────────────────────────────────── + +_TEST_TIMEOUT = 15 + + +async def _probe(url: str, headers: dict | None = None, params: dict | None = None) -> dict: + """单次 HTTP 探测,返回 (ok, status, latency_ms, detail)。不重试。""" + client = get_client() + start = time.monotonic() + try: + resp = await client.get(url, headers=headers, params=params, timeout=_TEST_TIMEOUT) + except Exception as e: + return { + "ok": False, + "status": None, + "latency_ms": int((time.monotonic() - start) * 1000), + "detail": f"无法连接: {e}", + } + latency = int((time.monotonic() - start) * 1000) + status = resp.status_code + if status == 200: + detail = "连接成功" + elif status in (401, 403): + detail = "服务可达,但密钥无效或无权限" + else: + detail = f"服务返回 HTTP {status}" + return {"ok": status == 200, "status": status, "latency_ms": latency, "detail": detail} + + +@router.post("/datasources/{name}/test") +async def test_datasource(name: str): + """轻量连通性测试:真实请求上游一次,不触发任何入库。""" + src = next((s for s in _SOURCES if s["name"] == name), None) + if src is None: + raise HTTPException(404, f"未知数据源: {name}") + + if name == "bzzoiro": + key = await get_runtime_value("BZZOIRO_KEY") + if not key: + return {"ok": False, "status": None, "latency_ms": 0, "detail": "BZZOIRO_KEY 未配置"} + base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") + today = date.today().isoformat() + return await _probe( + f"{base}/events/", + headers={"Authorization": f"Token {key}", "Accept": "application/json"}, + params={"date_from": today, "date_to": today}, + ) + + raise HTTPException(404, f"未知数据源: {name}") + + +# ── 数据源健康/最近采集状态(只读,不触发采集) ────────────────────── + + +@router.get("/ingest/status") +async def ingest_status(db: AsyncSession = Depends(get_db_read)): + """数据源采集健康概览(bzzoiro 单源;只读,不触发任何采集)。""" + bzzoiro_key = await get_runtime_value("BZZOIRO_KEY") + bzzoiro_base = await get_runtime_value("BZZOIRO_BASE") + + # 比赛覆盖 + match_row = ( + await db.execute( + select( + func.count().label("cnt"), + func.max(Match.match_date).label("latest_match_date"), + func.max(Match.created_at).label("latest_row_at"), + ).where(Match.match_status == "finished") + ) + ).one() + # 统计覆盖(精确 retrieved_at) + stats_row = ( + await db.execute( + select( + func.count().label("cnt"), + func.max(MatchStats.retrieved_at).label("latest_retrieved"), + ).where(MatchStats.source == "bzzoiro") + ) + ).one() + # 积分榜覆盖 + standings_row = ( + await db.execute(select(func.count()).select_from(Standing)) + ).scalar() + + bzzoiro = { + "name": "bzzoiro", + "label": "Bzzoiro", + "key_configured": bool(bzzoiro_key), + "base_url": (bzzoiro_base.rstrip("/") if bzzoiro_base else None) or settings.BZZOIRO_BASE, + "reachable": None, # 不主动探测 + "last_success_at": (stats_row.latest_retrieved or match_row.latest_row_at), + "last_success_at_iso": ( + stats_row.latest_retrieved or match_row.latest_row_at + ).isoformat() if (stats_row.latest_retrieved or match_row.latest_row_at) else None, + "latest_match_date": match_row.latest_match_date.isoformat() if match_row.latest_match_date else None, + "recent_count": match_row.cnt or 0, + "stats_count": stats_row.cnt or 0, + "standings_count": standings_row or 0, + "note": "last_success_at 取 match_stats.retrieved_at(统计回填)与 matches.created_at(比赛行)的较大者", + "last_failure": _last_failure_log("bzzoiro"), + } + + return {"sources": [bzzoiro]} + + +@router.get("/keyring/status") +async def keyring_status(): + """KeyRing 运行状态:当前使用的 key、冷却状态、轮转信息(供管理后台展示)。""" + base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") + raw_keys = await get_runtime_value("BZZOIRO_KEY") + ring = get_key_ring(base, raw_keys) + st = ring.stats() + st["base_url"] = base + st["cooldown_seconds"] = ring._cooldown + st["has_multiple"] = ring.has_multiple + st["active_key"] = ring.active_key + return st + + +@router.post("/keyring/cooldown/reset") +async def keyring_reset_cooldown(): + """手动重置所有 key 的冷却状态(用于紧急恢复)。""" + base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") + raw_keys = await get_runtime_value("BZZOIRO_KEY") + ring = get_key_ring(base, raw_keys) + ring._blocked_until.clear() + return {"ok": True, "message": "已重置所有 key 冷却状态", "stats": ring.stats()} + + +def _last_failure_log(source: str) -> dict | None: + """从系统日志缓冲中查找某数据源的最近一次错误(仅作参考,非专用失败表)。""" + from src.core.log_buffer import get_entries + entries = get_entries(min_level="ERROR", keyword=source, limit=5) + if not entries: + return None + e = entries[0] + return { + "at": datetime.fromtimestamp(e["ts"]).isoformat(), + "logger": e["logger"], + "detail": e["message"][:200], + "note": "approx:来自内存日志缓冲,非专用采集失败表;进程重启后清零", + } diff --git a/src/api/routes/admin_llm.py b/src/api/routes/admin_llm.py new file mode 100644 index 0000000..b432f80 --- /dev/null +++ b/src/api/routes/admin_llm.py @@ -0,0 +1,126 @@ +"""后台管理:LLM 专家/终裁配置、可用模型探测、连通性测试。 + +所有接口需管理员鉴权(require_admin)。路由前缀 /api/v1/admin。 +""" +from __future__ import annotations + +import logging +import time + +import httpx +from fastapi import APIRouter, Depends + +from src.api.deps import require_admin +from src.core.config import settings +from src.core.http_client import get_client +from src.core.runtime_config import ( + AGENT_META, + SETTING_DEFS, + get_runtime_value, + get_setting_origin, + mask_value, +) + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)]) + + +@router.get("/llm/agents") +async def list_llm_agents(): + """各专家/终裁的独立 LLM 配置状态(含当前生效模型的解析结果)。""" + out = [] + for agent in AGENT_META: + aid = agent["id"].upper() + pfx = f"AGENT_{aid}_" + fields = {} + for suffix in ("MODEL", "BASE_URL", "API_KEY"): + origin, value = await get_setting_origin(f"{pfx}{suffix}") + defn = SETTING_DEFS[f"{pfx}{suffix}"] + fields[suffix.lower()] = { + "configured": origin != "none", + "masked": mask_value(value, defn.sensitive), + "origin": origin, + } + # 生效模型 = 覆盖 → 层级默认(专家/终裁 env) → 全局 LLM_MODEL + tier_default = ( + settings.LLM_AGGREGATOR_MODEL if agent["id"] == "aggregator" else settings.LLM_SPECIALIST_MODEL + ) + effective_model = ( + fields["model"]["masked"] + if fields["model"]["configured"] + else (tier_default or await get_runtime_value("LLM_MODEL")) + ) + out.append( + { + "id": agent["id"], + "label": agent["label"], + "fields": fields, + "effective_model": effective_model, + } + ) + return out + + +@router.get("/llm/models") +async def list_llm_models(): + """探测当前 LLM 服务可用的模型列表(OpenAI 兼容 GET /models)。 + + 只读探测,不产生费用;配置缺失或服务不可达时返回 ok=false 与原因。 + """ + base_url = (await get_runtime_value("LLM_BASE_URL")).rstrip("/") + api_key = await get_runtime_value("LLM_API_KEY") + if not base_url or not api_key: + return {"ok": False, "models": [], "detail": "LLM_BASE_URL 或 LLM_API_KEY 未配置"} + + client = get_client() + start = time.monotonic() + try: + resp = await client.get( + f"{base_url}/models", + headers={"Authorization": f"Bearer {api_key}"}, + timeout=httpx.Timeout(connect=10.0, read=20.0, write=10.0, pool=10.0), + ) + except Exception as e: + return { + "ok": False, + "models": [], + "latency_ms": int((time.monotonic() - start) * 1000), + "detail": f"无法连接 LLM 服务: {e}", + } + + latency = int((time.monotonic() - start) * 1000) + if resp.status_code in (401, 403): + return {"ok": False, "models": [], "latency_ms": latency, "detail": "密钥无效或无权限(HTTP 401/403)"} + if resp.status_code != 200: + return {"ok": False, "models": [], "latency_ms": latency, "detail": f"服务返回 HTTP {resp.status_code}"} + + try: + data = resp.json() + except Exception: + return {"ok": False, "models": [], "latency_ms": latency, "detail": "响应不是合法 JSON"} + + models: list[str] = [] + items = data.get("data") if isinstance(data, dict) else None + if isinstance(items, list): + models = sorted( + str(m.get("id")) for m in items if isinstance(m, dict) and m.get("id") + ) + if not models: + return {"ok": False, "models": [], "latency_ms": latency, "detail": "服务未返回模型列表"} + return {"ok": True, "models": models, "latency_ms": latency, "detail": f"共 {len(models)} 个可用模型"} + + +@router.post("/llm/ping") +async def llm_ping(): + """LLM 连通性测试(不依赖比赛)。只发一次 chat 请求验证配置。""" + from src.llm.provider import get_default_provider + p = await get_default_provider() + resp = await p.chat( + system="你是测试助手。", + user="ping", + max_tokens=10, + ) + if resp.error: + return {"ok": False, "message": resp.error} + return {"ok": True, "message": "LLM 连接正常", "model": p.model} diff --git a/src/api/routes/admin_settings.py b/src/api/routes/admin_settings.py index 2f61aef..62ac05e 100644 --- a/src/api/routes/admin_settings.py +++ b/src/api/routes/admin_settings.py @@ -1,668 +1,25 @@ -"""后台管理路由:数据源配置的查看、修改与连通性测试。 +"""后台管理路由聚合入口:按职责拆分为四个子模块,统一挂载。 -所有接口需管理员鉴权(require_admin)。配置项白名单见 -src/core/runtime_config.py SETTING_DEFS,之外的 key 一律拒绝。 +所有路由仍挂在 /api/v1/admin,且均带 dependencies=[Depends(require_admin)] +(鉴权由各子路由器声明,行为与拆分前完全一致)。 + +子模块: + - admin_datasources 数据源列表/连通性测试、KeyRing、采集健康概览 + - admin_config settings CRUD、运行日志 + - admin_llm LLM agents/models/ping + - admin_quality stats、data-completeness、data-quality """ from __future__ import annotations -import logging -import time -from datetime import date, datetime, timedelta, timezone - -from fastapi import APIRouter, Depends, HTTPException, Query -from pydantic import BaseModel -from sqlalchemy import func, select -import httpx - -from src.api.deps import require_admin -from src.core.config import settings -from src.core.http_client import get_client -from src.core.log_buffer import get_entries -from src.core.runtime_config import ( - AGENT_META, - SETTING_DEFS, - clear_runtime_value, - get_runtime_value, - get_setting_origin, - mask_value, - set_runtime_value, -) -from src.db.base import AsyncSession, get_db_read -from src.db.models import League, Match, MatchStats, Standing -from src.data.key_ring import get_key_ring, parse_keys - -logger = logging.getLogger(__name__) - -router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)]) - -# ── 数据源元数据(bzzoiro 单一数据源) ──────────────────────────── - -_SOURCES: list[dict] = [ - { - "name": "bzzoiro", - "label": "Bzzoiro", - "description": "唯一数据源:赛程比分 + 积分榜 + 比赛详细统计(xG/射门/控球等)", - "setting_keys": ["BZZOIRO_KEY", "BZZOIRO_BASE"], - }, -] - - -class SettingUpdateIn(BaseModel): - value: str - - -async def _last_ingestion(db: AsyncSession, source: str) -> datetime | None: - """源最近一次采集时间(取自数据血缘字段,无记录返回 None)。""" - return ( - await db.execute( - select(func.max(MatchStats.retrieved_at)).where(MatchStats.source == source) - ) - ).scalar() - - -@router.get("/datasources") -async def list_datasources(db: AsyncSession = Depends(get_db_read)): - """数据源列表:各配置项的脱敏值、来源(db/env/none)与最近采集时间。""" - result = [] - for src in _SOURCES: - settings_out = [] - for key in src["setting_keys"]: - origin, value = await get_setting_origin(key) - defn = SETTING_DEFS[key] - settings_out.append( - { - "key": key, - "label": defn.label, - "description": defn.description, - "sensitive": defn.sensitive, - "configured": origin != "none", - "masked": mask_value(value, defn.sensitive), - "origin": origin, - } - ) - key_configured = all(s["configured"] for s in settings_out) if settings_out else True - last = await _last_ingestion(db, src["name"]) - result.append( - { - "name": src["name"], - "label": src["label"], - "description": src["description"], - "key_configured": key_configured, - "last_ingestion": last.isoformat() if last else None, - "settings": settings_out, - } - ) - return result - - -@router.get("/settings") -async def list_settings(): - """全部可配置项(脱敏),供后台各配置页渲染。""" - out = [] - for key, defn in SETTING_DEFS.items(): - origin, value = await get_setting_origin(key) - out.append( - { - "key": key, - "label": defn.label, - "description": defn.description, - "sensitive": defn.sensitive, - "configured": origin != "none", - "masked": mask_value(value, defn.sensitive), - "origin": origin, - } - ) - return out - - -# ── LLM 可用模型检测 ──────────────────────────────────────────── - - -@router.get("/logs") -async def read_logs( - level: str | None = Query(None, description="最低级别: DEBUG/INFO/WARNING/ERROR"), - keyword: str | None = Query(None, description="消息或 logger 关键字"), - limit: int = Query(200, ge=1, le=1000), -): - """查询应用运行日志(内存环形缓冲,最新在前;进程重启后清零)。""" - entries = get_entries(level, keyword, limit) - return {"entries": entries, "count": len(entries)} - - -@router.get("/llm/agents") -async def list_llm_agents(): - """各专家/终裁的独立 LLM 配置状态(含当前生效模型的解析结果)。""" - out = [] - for agent in AGENT_META: - aid = agent["id"].upper() - pfx = f"AGENT_{aid}_" - fields = {} - for suffix in ("MODEL", "BASE_URL", "API_KEY"): - origin, value = await get_setting_origin(f"{pfx}{suffix}") - defn = SETTING_DEFS[f"{pfx}{suffix}"] - fields[suffix.lower()] = { - "configured": origin != "none", - "masked": mask_value(value, defn.sensitive), - "origin": origin, - } - # 生效模型 = 覆盖 → 层级默认(专家/终裁 env) → 全局 LLM_MODEL - tier_default = ( - settings.LLM_AGGREGATOR_MODEL if agent["id"] == "aggregator" else settings.LLM_SPECIALIST_MODEL - ) - effective_model = ( - fields["model"]["masked"] - if fields["model"]["configured"] - else (tier_default or await get_runtime_value("LLM_MODEL")) - ) - out.append( - { - "id": agent["id"], - "label": agent["label"], - "fields": fields, - "effective_model": effective_model, - } - ) - return out - - -@router.get("/llm/models") -async def list_llm_models(): - """探测当前 LLM 服务可用的模型列表(OpenAI 兼容 GET /models)。 - - 只读探测,不产生费用;配置缺失或服务不可达时返回 ok=false 与原因。 - """ - base_url = (await get_runtime_value("LLM_BASE_URL")).rstrip("/") - api_key = await get_runtime_value("LLM_API_KEY") - if not base_url or not api_key: - return {"ok": False, "models": [], "detail": "LLM_BASE_URL 或 LLM_API_KEY 未配置"} - - client = get_client() - start = time.monotonic() - try: - resp = await client.get( - f"{base_url}/models", - headers={"Authorization": f"Bearer {api_key}"}, - timeout=httpx.Timeout(connect=10.0, read=20.0, write=10.0, pool=10.0), - ) - except Exception as e: - return { - "ok": False, - "models": [], - "latency_ms": int((time.monotonic() - start) * 1000), - "detail": f"无法连接 LLM 服务: {e}", - } - - latency = int((time.monotonic() - start) * 1000) - if resp.status_code in (401, 403): - return {"ok": False, "models": [], "latency_ms": latency, "detail": "密钥无效或无权限(HTTP 401/403)"} - if resp.status_code != 200: - return {"ok": False, "models": [], "latency_ms": latency, "detail": f"服务返回 HTTP {resp.status_code}"} - - try: - data = resp.json() - except Exception: - return {"ok": False, "models": [], "latency_ms": latency, "detail": "响应不是合法 JSON"} - - models: list[str] = [] - items = data.get("data") if isinstance(data, dict) else None - if isinstance(items, list): - models = sorted( - str(m.get("id")) for m in items if isinstance(m, dict) and m.get("id") - ) - if not models: - return {"ok": False, "models": [], "latency_ms": latency, "detail": "服务未返回模型列表"} - return {"ok": True, "models": models, "latency_ms": latency, "detail": f"共 {len(models)} 个可用模型"} - - -@router.put("/settings/{key}") -async def update_setting(key: str, body: SettingUpdateIn): - """更新配置项(写入 app_settings 覆盖 .env)。传空值请改用 DELETE。""" - if key not in SETTING_DEFS: - raise HTTPException(404, f"不支持的配置项: {key}") - value = body.value.strip() - if not value: - raise HTTPException(400, "值不能为空;如需回落 .env 请调用清除接口") - await set_runtime_value(key, value) - defn = SETTING_DEFS[key] - return {"key": key, "masked": mask_value(value, defn.sensitive), "origin": "db"} - - -@router.delete("/settings/{key}") -async def clear_setting(key: str): - """清除 DB 覆盖值,回落 .env 默认。""" - if key not in SETTING_DEFS: - raise HTTPException(404, f"不支持的配置项: {key}") - await clear_runtime_value(key) - origin, value = await get_setting_origin(key) - defn = SETTING_DEFS[key] - return { - "key": key, - "masked": mask_value(value, defn.sensitive), - "origin": origin, - } - - -# ── 连通性测试 ────────────────────────────────────────────────── - -_TEST_TIMEOUT = 15 - - -async def _probe(url: str, headers: dict | None = None, params: dict | None = None) -> dict: - """单次 HTTP 探测,返回 (ok, status, latency_ms, detail)。不重试。""" - client = get_client() - start = time.monotonic() - try: - resp = await client.get(url, headers=headers, params=params, timeout=_TEST_TIMEOUT) - except Exception as e: - return { - "ok": False, - "status": None, - "latency_ms": int((time.monotonic() - start) * 1000), - "detail": f"无法连接: {e}", - } - latency = int((time.monotonic() - start) * 1000) - status = resp.status_code - if status == 200: - detail = "连接成功" - elif status in (401, 403): - detail = "服务可达,但密钥无效或无权限" - else: - detail = f"服务返回 HTTP {status}" - return {"ok": status == 200, "status": status, "latency_ms": latency, "detail": detail} - - -@router.post("/datasources/{name}/test") -async def test_datasource(name: str): - """轻量连通性测试:真实请求上游一次,不触发任何入库。""" - src = next((s for s in _SOURCES if s["name"] == name), None) - if src is None: - raise HTTPException(404, f"未知数据源: {name}") - - if name == "bzzoiro": - key = await get_runtime_value("BZZOIRO_KEY") - if not key: - return {"ok": False, "status": None, "latency_ms": 0, "detail": "BZZOIRO_KEY 未配置"} - base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") - today = date.today().isoformat() - return await _probe( - f"{base}/events/", - headers={"Authorization": f"Token {key}", "Accept": "application/json"}, - params={"date_from": today, "date_to": today}, - ) - - raise HTTPException(404, f"未知数据源: {name}") - - -@router.post("/llm/ping") -async def llm_ping(): - """LLM 连通性测试(不依赖比赛)。只发一次 chat 请求验证配置。""" - from src.llm.provider import get_default_provider - p = await get_default_provider() - resp = await p.chat( - system="你是测试助手。", - user="ping", - max_tokens=10, - ) - if resp.error: - return {"ok": False, "message": resp.error} - return {"ok": True, "message": "LLM 连接正常", "model": p.model} - - -# ── 数据源健康/最近采集状态(只读,不触发采集) ────────────────────── - - -@router.get("/ingest/status") -async def ingest_status(db: AsyncSession = Depends(get_db_read)): - """数据源采集健康概览(bzzoiro 单源;只读,不触发任何采集)。""" - bzzoiro_key = await get_runtime_value("BZZOIRO_KEY") - bzzoiro_base = await get_runtime_value("BZZOIRO_BASE") - - # 比赛覆盖 - match_row = ( - await db.execute( - select( - func.count().label("cnt"), - func.max(Match.match_date).label("latest_match_date"), - func.max(Match.created_at).label("latest_row_at"), - ).where(Match.match_status == "finished") - ) - ).one() - # 统计覆盖(精确 retrieved_at) - stats_row = ( - await db.execute( - select( - func.count().label("cnt"), - func.max(MatchStats.retrieved_at).label("latest_retrieved"), - ).where(MatchStats.source == "bzzoiro") - ) - ).one() - # 积分榜覆盖 - standings_row = ( - await db.execute(select(func.count()).select_from(Standing)) - ).scalar() - - bzzoiro = { - "name": "bzzoiro", - "label": "Bzzoiro", - "key_configured": bool(bzzoiro_key), - "base_url": (bzzoiro_base.rstrip("/") if bzzoiro_base else None) or settings.BZZOIRO_BASE, - "reachable": None, # 不主动探测 - "last_success_at": (stats_row.latest_retrieved or match_row.latest_row_at), - "last_success_at_iso": ( - stats_row.latest_retrieved or match_row.latest_row_at - ).isoformat() if (stats_row.latest_retrieved or match_row.latest_row_at) else None, - "latest_match_date": match_row.latest_match_date.isoformat() if match_row.latest_match_date else None, - "recent_count": match_row.cnt or 0, - "stats_count": stats_row.cnt or 0, - "standings_count": standings_row or 0, - "note": "last_success_at 取 match_stats.retrieved_at(统计回填)与 matches.created_at(比赛行)的较大者", - "last_failure": _last_failure_log("bzzoiro"), - } - - return {"sources": [bzzoiro]} - - -@router.get("/keyring/status") -async def keyring_status(): - """KeyRing 运行状态:当前使用的 key、冷却状态、轮转信息(供管理后台展示)。""" - base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") - raw_keys = await get_runtime_value("BZZOIRO_KEY") - ring = get_key_ring(base, raw_keys) - st = ring.stats() - st["base_url"] = base - st["cooldown_seconds"] = ring._cooldown - st["has_multiple"] = ring.has_multiple - st["active_key"] = ring.active_key - return st - - -@router.post("/keyring/cooldown/reset") -async def keyring_reset_cooldown(): - """手动重置所有 key 的冷却状态(用于紧急恢复)。""" - base = (await get_runtime_value("BZZOIRO_BASE")).rstrip("/") - raw_keys = await get_runtime_value("BZZOIRO_KEY") - ring = get_key_ring(base, raw_keys) - ring._blocked_until.clear() - return {"ok": True, "message": "已重置所有 key 冷却状态", "stats": ring.stats()} - - -def _last_failure_log(source: str) -> dict | None: - """从系统日志缓冲中查找某数据源的最近一次错误(仅作参考,非专用失败表)。""" - entries = get_entries(min_level="ERROR", keyword=source, limit=5) - if not entries: - return None - e = entries[0] - return { - "at": datetime.fromtimestamp(e["ts"]).isoformat(), - "logger": e["logger"], - "detail": e["message"][:200], - "note": "approx:来自内存日志缓冲,非专用采集失败表;进程重启后清零", - } - - -@router.get("/stats") -async def admin_stats(db: AsyncSession = Depends(get_db_read)): - """管理区统计(只读):预测次数 + 比赛覆盖。轻量聚合,无 LLM 调用。""" - from sqlalchemy import func, text - from src.db.models import Prediction, Match, MatchStats, Standing - day_ago = datetime.now(timezone.utc) - timedelta(days=1) - week_ago = datetime.now(timezone.utc) - timedelta(days=7) - r = ( - await db.execute( - select( - func.count().label("total"), - func.count().filter(Prediction.created_at >= day_ago).label("last_24h"), - func.count().filter(Prediction.created_at >= week_ago).label("last_7d"), - ) - ) - ).one() - # F3 修复: 补充真实比赛计数(非 limit=100 近似) - match_cnt = (await db.execute(select(func.count()).select_from(Match))).scalar() or 0 - finished_cnt = (await db.execute(select(func.count()).where(Match.match_status == "finished"))).scalar() or 0 - stats_cnt = (await db.execute(select(func.count()).select_from(MatchStats))).scalar() or 0 - standings_cnt = (await db.execute(select(func.count()).select_from(Standing))).scalar() or 0 - return { - "predictions": {"total": r.total, "last_24h": r.last_24h, "last_7d": r.last_7d}, - "matches": {"total": match_cnt, "finished": finished_cnt}, - "stats": {"total": stats_cnt}, - "standings": {"total": standings_cnt}, - } - - -# ── 数据完整性分析(可视化数据源) ──────────────────────────────── - - -@router.get("/data-completeness") -async def data_completeness(db: AsyncSession = Depends(get_db_read)): - """按联赛统计数据完整性:比赛覆盖、字段覆盖、积分榜覆盖。 - - 前端「数据完整性」页据此渲染,回答三个问题: - 1. 数据是否齐全(各联赛比赛/统计/积分榜量级) - 2. 字段是否齐全(每张统计表各字段非空率) - 3. 覆盖是否新鲜(最近一场/最近一次采集) - """ - from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_NAMES, LEAGUE_COUNTRIES - - out_leagues: list[dict] = [] - for code, bzz_id in BZZOIRO_LEAGUE_IDS.items(): - # 比赛覆盖 - m = ( - await db.execute( - select( - func.count().label("total"), - func.count().filter(Match.match_status == "finished").label("finished"), - func.count().filter(Match.match_status == "scheduled").label("scheduled"), - func.count().filter(Match.source_event_id.is_not(None)).label("with_source_id"), - func.max(Match.match_date).label("latest_match"), - func.min(Match.match_date).label("earliest_match"), - ) - .select_from(Match) - .join(League, League.id == Match.league_id) - .where(League.code == code) - ) - ).one() - # 统计字段覆盖(联表 matches) - s = ( - await db.execute( - select( - func.count().label("rows"), - func.count(MatchStats.home_xg).label("xg"), - func.count(MatchStats.home_shots).label("shots"), - func.count(MatchStats.home_possession).label("possession"), - func.count(MatchStats.home_corners).label("corners"), - func.count(MatchStats.home_fouls).label("fouls"), - func.count(MatchStats.home_big_chances).label("big_chances"), - func.count(MatchStats.home_yellow_cards).label("cards"), - ) - .select_from(MatchStats) - .join(Match, Match.id == MatchStats.match_id) - .join(League, League.id == Match.league_id) - .where(League.code == code) - ) - ).one() - # 积分榜覆盖 - st = ( - await db.execute( - select( - func.count().label("rows"), - func.max(Standing.retrieved_at).label("latest_retrieved"), - ) - .select_from(Standing) - .join(League, League.id == Standing.league_id) - .where(League.code == code) - ) - ).one() - - stats_rows = s.rows or 0 - pct = lambda n: round(n / stats_rows * 100, 1) if stats_rows else 0.0 # noqa: E731 - out_leagues.append( - { - "code": code, - "name": LEAGUE_NAMES.get(code, code), - "country": LEAGUE_COUNTRIES.get(code), - "matches": { - "total": m.total or 0, - "finished": m.finished or 0, - "scheduled": m.scheduled or 0, - "with_source_id": m.with_source_id or 0, - "earliest_match": m.earliest_match.isoformat() if m.earliest_match else None, - "latest_match": m.latest_match.isoformat() if m.latest_match else None, - }, - "stats": { - "rows": stats_rows, - "fields": { - "xg": {"count": s.xg or 0, "pct": pct(s.xg or 0)}, - "shots": {"count": s.shots or 0, "pct": pct(s.shots or 0)}, - "possession": {"count": s.possession or 0, "pct": pct(s.possession or 0)}, - "corners": {"count": s.corners or 0, "pct": pct(s.corners or 0)}, - "fouls": {"count": s.fouls or 0, "pct": pct(s.fouls or 0)}, - "big_chances": {"count": s.big_chances or 0, "pct": pct(s.big_chances or 0)}, - "cards": {"count": s.cards or 0, "pct": pct(s.cards or 0)}, - }, - }, - "standings": { - "rows": st.rows or 0, - "latest_retrieved": st.latest_retrieved.isoformat() if st.latest_retrieved else None, - }, - } - ) - - # 整体健康信号 - total_finished = sum(l["matches"]["finished"] for l in out_leagues) - total_stats = sum(l["stats"]["rows"] for l in out_leagues) - stats_coverage = round(total_stats / total_finished * 100, 1) if total_finished else 0.0 - issues: list[str] = [] - for l in out_leagues: - if l["matches"]["finished"] == 0: - issues.append(f"{l['name']}: 无已完赛比赛,请先运行「比赛数据」采集") - elif l["stats"]["rows"] == 0: - issues.append(f"{l['name']}: 已完赛 {l['matches']['finished']} 场但无统计回填,请运行「统计回填」采集") - elif stats_coverage < 80: - issues.append(f"{l['name']}: 统计覆盖率仅 {stats_coverage}%,建议增量回填") - if l["standings"]["rows"] == 0: - issues.append(f"{l['name']}: 无积分榜数据,请运行「积分榜」采集") - if not issues: - issues.append("各联赛数据完整度良好") - - return { - "generated_at": datetime.now(timezone.utc).isoformat(), - "leagues": out_leagues, - "totals": { - "finished_matches": total_finished, - "stats_rows": total_stats, - "stats_coverage_pct": stats_coverage, - }, - "issues": issues, - } - - -# ── 数据质量检查 API ──────────────────────────────────────────── - - -@router.get("/data-quality") -async def data_quality_checks(db: AsyncSession = Depends(get_db_read)): - """数据质量检查结果(只读)。""" - from src.db.models import IngestFailure, DataQualityCheck - from sqlalchemy import func - - # 最近的失败记录 - failures = ( - await db.execute( - select(IngestFailure) - .where(IngestFailure.status.in_(["pending", "retrying"])) - .order_by(IngestFailure.created_at.desc()) - .limit(20) - ) - ).scalars().all() - - # 最近的质量检查 - checks = ( - await db.execute( - select(DataQualityCheck) - .order_by(DataQualityCheck.checked_at.desc()) - .limit(20) - ) - ).scalars().all() - - return { - "failures": [ - { - "id": f.id, - "source": f.source_system, - "entity_type": f.entity_type, - "source_record_id": f.source_record_id, - "error_type": f.error_type, - "error_detail": f.error_detail, - "retry_count": f.retry_count, - "status": f.status, - "created_at": f.created_at.isoformat() if f.created_at else None, - } - for f in failures - ], - "checks": [ - { - "id": c.id, - "check_name": c.check_name, - "entity_type": c.entity_type, - "passed": c.passed, - "severity": c.severity, - "detail": c.detail, - "checked_at": c.checked_at.isoformat() if c.checked_at else None, - } - for c in checks - ], - } - - -@router.post("/data-quality/run") -async def run_data_quality_check(db: AsyncSession = Depends(get_db_read)): - """手动触发一次数据质量检查。""" - from src.db.models import DataQualityCheck, Match, MatchStats, Standing, League - from sqlalchemy import func - - checks = [] - - # 检查1: 已完赛但无统计的比赛 - finished_no_stats = ( - await db.execute( - select(func.count()) - .select_from(Match) - .outerjoin(MatchStats, Match.id == MatchStats.match_id) - .where(Match.match_status == "finished") - .where(MatchStats.id.is_(None)) - ) - ).scalar() or 0 - - checks.append(DataQualityCheck( - check_name="finished_without_stats", - entity_type="match", - actual_value=float(finished_no_stats), - passed=finished_no_stats == 0, - severity="warning" if finished_no_stats > 0 else "info", - detail={"message": f"{finished_no_stats} 场已完赛比赛缺少统计数据"}, - )) - - # 检查2: 积分榜缺失的联赛 - leagues_without_standings = ( - await db.execute( - select(func.count()) - .select_from(League) - .outerjoin(Standing, League.id == Standing.league_id) - .where(Standing.id.is_(None)) - ) - ).scalar() or 0 - - checks.append(DataQualityCheck( - check_name="league_without_standings", - entity_type="league", - actual_value=float(leagues_without_standings), - passed=leagues_without_standings == 0, - severity="warning" if leagues_without_standings > 0 else "info", - detail={"message": f"{leagues_without_standings} 个联赛缺少积分榜"}, - )) - - for c in checks: - db.add(c) - await db.commit() - - return {"ok": True, "checks": [{"name": c.check_name, "passed": c.passed} for c in checks]} +from fastapi import APIRouter + +from src.api.routes.admin_config import router as admin_config_router +from src.api.routes.admin_datasources import router as admin_datasources_router +from src.api.routes.admin_llm import router as admin_llm_router +from src.api.routes.admin_quality import router as admin_quality_router + +router = APIRouter() +router.include_router(admin_datasources_router) +router.include_router(admin_config_router) +router.include_router(admin_llm_router) +router.include_router(admin_quality_router)