wehub-resource-sync
91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""Admin API — 管理操作（迁移、诊断、配置）。"""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import text
|
|
|
|
from ..db import get_db
|
|
from ..services.event_bus import get_event_bus
|
|
|
|
log = logging.getLogger("edict.api.admin")
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/health/deep")
|
|
async def deep_health(db: AsyncSession = Depends(get_db)):
|
|
"""深度健康检查：Postgres + Redis 连通性。"""
|
|
checks = {"postgres": False, "redis": False}
|
|
|
|
# Postgres
|
|
try:
|
|
result = await db.execute(text("SELECT 1"))
|
|
checks["postgres"] = result.scalar() == 1
|
|
except Exception as e:
|
|
checks["postgres_error"] = str(e)
|
|
|
|
# Redis
|
|
try:
|
|
bus = await get_event_bus()
|
|
pong = await bus.redis.ping()
|
|
checks["redis"] = pong is True
|
|
except Exception as e:
|
|
checks["redis_error"] = str(e)
|
|
|
|
status = "ok" if all(checks.get(k) for k in ["postgres", "redis"]) else "degraded"
|
|
return {"status": status, "checks": checks}
|
|
|
|
|
|
@router.get("/pending-events")
|
|
async def pending_events(
|
|
topic: str = "task.dispatch",
|
|
group: str = "dispatcher",
|
|
count: int = 20,
|
|
):
|
|
"""查看未 ACK 的 pending 事件（诊断工具）。"""
|
|
bus = await get_event_bus()
|
|
pending = await bus.get_pending(topic, group, count)
|
|
return {
|
|
"topic": topic,
|
|
"group": group,
|
|
"pending": [
|
|
{
|
|
"entry_id": str(p.get("message_id", "")),
|
|
"consumer": str(p.get("consumer", "")),
|
|
"idle_ms": p.get("time_since_delivered", 0),
|
|
"delivery_count": p.get("times_delivered", 0),
|
|
}
|
|
for p in pending
|
|
] if pending else [],
|
|
}
|
|
|
|
|
|
@router.post("/migrate/check")
|
|
async def migration_check():
|
|
"""检查旧数据文件是否存在。"""
|
|
data_dir = Path(__file__).parents[4] / "data"
|
|
files = {
|
|
"tasks_source": (data_dir / "tasks_source.json").exists(),
|
|
"live_status": (data_dir / "live_status.json").exists(),
|
|
"agent_config": (data_dir / "agent_config.json").exists(),
|
|
"officials_stats": (data_dir / "officials_stats.json").exists(),
|
|
}
|
|
return {"data_dir": str(data_dir), "files": files}
|
|
|
|
|
|
@router.get("/config")
|
|
async def get_config():
|
|
"""获取当前运行配置（脱敏）。"""
|
|
from ..config import get_settings
|
|
settings = get_settings()
|
|
return {
|
|
"port": settings.port,
|
|
"debug": settings.debug,
|
|
"database": settings.database_url.split("@")[-1] if "@" in settings.database_url else "***",
|
|
"redis": settings.redis_url.split("@")[-1] if "@" in settings.redis_url else settings.redis_url,
|
|
"scheduler_scan_interval": settings.scheduler_scan_interval_seconds,
|
|
}
|