From 97de54c883a0e5ca7dbd240bd02efdb563a0825e Mon Sep 17 00:00:00 2001 From: claude_dev Date: Fri, 25 Sep 2026 22:54:07 +0800 Subject: [PATCH] =?UTF-8?q?feat(api):=20/alerts=20=E5=88=97=E8=A1=A8+summa?= =?UTF-8?q?ry+ack=20=E8=B7=AF=E7=94=B1(=E8=AF=BB=20data/alerts.db,router?= =?UTF-8?q?=20=E7=BA=A7=E9=89=B4=E6=9D=83,spec=20=C2=A720.15.4)=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- sanguo_api/app.py | 3 ++ sanguo_api/routes_alerts.py | 61 +++++++++++++++++++++++++++++++ tests/api/test_routes_alerts.py | 65 +++++++++++++++++++++++++++++++++ 3 files changed, 129 insertions(+) create mode 100644 sanguo_api/routes_alerts.py create mode 100644 tests/api/test_routes_alerts.py diff --git a/sanguo_api/app.py b/sanguo_api/app.py index c379ee04..1fe7632c 100644 --- a/sanguo_api/app.py +++ b/sanguo_api/app.py @@ -9,6 +9,7 @@ from .routes_portfolio import router as portfolio_router from .routes_strategy import router as strategy_router from .routes_factor import router as factor_eval_router, set_eval_db_path as set_factor_eval_db_path from .routes_pipeline import router as pipeline_router, set_pipeline_eval_db_path +from .routes_alerts import router as alerts_router, set_alerts_db_path from .auth import set_jwt_config from .ws import manager from sanguo_orchestrator.runner import Orchestrator, resolve_eval_db @@ -45,10 +46,12 @@ def create_app(db_path: str, file_dir=None, auth_config=None, max_workers: int = app.include_router(strategy_router, prefix="/api/v1") app.include_router(factor_eval_router, prefix="/api/v1") app.include_router(pipeline_router, prefix="/api/v1") + app.include_router(alerts_router, prefix="/api/v1") set_paper_db_path(db_path) set_live_db_path(db_path) set_factor_eval_db_path(resolve_eval_db(db_path)) set_pipeline_eval_db_path(resolve_eval_db(db_path)) + set_alerts_db_path(None) # None=env SANGUO_ALERTS_DB > 各机缺省 data/alerts.db @app.on_event("startup") def _register_live_step(): diff --git a/sanguo_api/routes_alerts.py b/sanguo_api/routes_alerts.py new file mode 100644 index 00000000..e5c2d5d6 --- /dev/null +++ b/sanguo_api/routes_alerts.py @@ -0,0 +1,61 @@ +"""运维告警 API(spec §20.15.4): 只读列表+summary+人工 ack; 写入方=监控脚本(单写者). + +alert 库路径: env SANGUO_ALERTS_DB > 各机缺省 data/alerts.db(相对 cwd)—— +与监控脚本 wrapper(Set-Location C:\\sanguo_vnpy_v2)解析到同一文件。 +""" +from fastapi import APIRouter, Depends, HTTPException + +from sanguo_api.routes import verify_token +from sanguo_data import alerts_store as store + +router = APIRouter(dependencies=[Depends(verify_token)]) + +_db = {"path": None} + + +def set_alerts_db_path(p): + """create_app 注入(同 routes_paper.set_db_path 模式); None=缺省解析.""" + _db["path"] = p + + +def _conn(): + return store.connect(_db["path"] or store.default_alerts_db_path()) + + +@router.get("/alerts") +def list_alerts(source: str = None, severity: str = None, status: str = None, + host: str = None, limit: int = 500, offset: int = 0) -> dict: + limit = max(1, min(limit, 2000)) + offset = max(0, offset) + conn = _conn() + try: + items, total = store.list_alerts(conn, source=source, severity=severity, + status=status, host=host, + limit=limit, offset=offset) + finally: + conn.close() + return {"items": items, "total": total} + + +@router.get("/alerts/summary") +def alerts_summary() -> dict: + conn = _conn() + try: + return store.summary(conn) + finally: + conn.close() + + +@router.post("/alerts/{alert_id}/ack") +def ack_alert(alert_id: str) -> dict: + conn = _conn() + try: + try: + status = store.ack_alert(conn, alert_id, acked_by="user") + except KeyError: + raise HTTPException(404, f"alert 不存在: {alert_id}") + finally: + conn.close() + if status is None: + raise HTTPException(400, "已 resolved 的告警无需 ack") + return {"alert_id": alert_id, "status": status} diff --git a/tests/api/test_routes_alerts.py b/tests/api/test_routes_alerts.py new file mode 100644 index 00000000..db886a30 --- /dev/null +++ b/tests/api/test_routes_alerts.py @@ -0,0 +1,65 @@ +"""routes_alerts 薄层测试(本地跑, CI 不含 tests/api): 列表筛选/summary/ack 三态.""" +import pytest +from fastapi.testclient import TestClient + +from sanguo_api.main import build_app +from sanguo_api.auth import create_token, set_jwt_config +from sanguo_data import alerts_store as store + + +def _cfg(tmp_path) -> str: + cfg = tmp_path / "bt.yaml" + cfg.write_text( + "backtest:\n max_workers: 1\n db_path: %s\n file_dir: %s\n" + "api:\n host: 0.0.0.0\n port: 8000\n" + "auth:\n username: admin\n password_hash: x\n jwt_secret: s\n token_expire_minutes: 60\n" + "pool:\n max_workers: 1\n" % (tmp_path / "r.db", tmp_path / "f") + ) + return str(cfg) + + +@pytest.fixture() +def client(tmp_path, monkeypatch): + monkeypatch.setenv("SANGUO_ALERTS_DB", str(tmp_path / "alerts.db")) + with store.connect(str(tmp_path / "alerts.db")) as conn: + store.upsert_alert(conn, { + "alert_id": "data-t-1-20260925", "check_key": "data-t-1", + "source": "data", "host": "vps", "severity": "red", + "title": "t1", "detail": "", "evidence": {"x": 1}}) + set_jwt_config(secret="s", expire_minutes=60) + app = build_app(_cfg(tmp_path)) + c = TestClient(app) + c.headers.update({"Authorization": f"Bearer {create_token('admin')}"}) + return c + + +def test_list_and_summary(client): + r = client.get("/api/v1/alerts") + assert r.status_code == 200 and r.json()["total"] == 1 + assert r.json()["items"][0]["evidence"] == {"x": 1} + r = client.get("/api/v1/alerts/summary") + assert r.json() == {"open_red": 1, + "by_source": {"data": 1, "strategy": 0, + "factor": 0, "infra": 0}} + + +def test_list_filter_param(client): + assert client.get("/api/v1/alerts", + params={"severity": "yellow"}).json()["total"] == 0 + + +def test_ack_flow(client): + r = client.post("/api/v1/alerts/data-t-1-20260925/ack") + assert r.status_code == 200 and r.json()["status"] == "acked" + # 已 ack(非 open) 再 ack → 400 语义(resolved/no-op 同通道) + r2 = client.post("/api/v1/alerts/data-t-1-20260925/ack") + assert r2.status_code == 400 + r3 = client.post("/api/v1/alerts/nope/ack") + assert r3.status_code == 404 + + +def test_unauthorized_rejected(tmp_path, monkeypatch): + monkeypatch.setenv("SANGUO_ALERTS_DB", str(tmp_path / "alerts.db")) + set_jwt_config(secret="s", expire_minutes=60) + c = TestClient(build_app(_cfg(tmp_path))) + assert c.get("/api/v1/alerts").status_code == 401