refactor: admin_settings.py 按职责拆分为 4 个模块
681 行单文件拆为(保留 admin_settings 为 include_router 聚合入口): - admin_datasources 数据源列表/连通性测试/KeyRing/ingest status (5 路由) - admin_config settings CRUD + 运行日志 (4 路由) - admin_llm LLM agents/models/ping (3 路由) - admin_quality stats/data-completeness/data-quality (4 路由) 所有路由仍挂 /api/v1/admin 且带 dependencies=[Depends(require_admin)]。 app.py 注册方式不变(仍 import admin_settings.router)。
This commit is contained in:
shangfangjian
@@ -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:来自内存日志缓冲,非专用采集失败表;进程重启后清零",
|
||||
}
|
||||
Reference in New Issue
Block a user