Files
eth_hedge_sim/control/backend/app/api/nodes.py
T

260 lines
8.1 KiB
Python

from __future__ import annotations
import secrets
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from ..auth import require_control_user
from ..config import get_control_settings
from ..crypto import seal
from ..db import get_control_db
from ..proxy import call_node, probe_health
router = APIRouter(prefix="/api/nodes", tags=["nodes"])
def _public_node(row: dict[str, Any]) -> dict[str, Any]:
return {
"id": row["id"],
"name": row["name"],
"base_url": row["base_url"],
"token_configured": bool((row.get("token_sealed") or "").strip()),
"created_at_ms": row["created_at_ms"],
"updated_at_ms": row["updated_at_ms"],
}
def _http_detail(data: Any) -> str:
if isinstance(data, dict):
d = data.get("detail", data)
return d if isinstance(d, str) else str(d)
return str(data)
def _raise_node_error(code: int, data: Any) -> None:
"""策略机错误不得用 401 回传,否则中控前端会误清登录态。"""
detail = _http_detail(data)
if code in (401, 403):
raise HTTPException(
status_code=502,
detail=f"策略机鉴权失败({detail})。请在中控重新生成 Token,并到策略机「系统设置→登录账户」保存同一 Token。",
)
raise HTTPException(
status_code=code if 400 <= code < 600 else 502,
detail=detail,
)
class NodeCreate(BaseModel):
name: str = Field(min_length=1, max_length=64)
base_url: str = Field(min_length=8, max_length=256)
class NodeUpdate(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=64)
base_url: str | None = Field(default=None, min_length=8, max_length=256)
@router.get("/")
async def list_nodes(_user: Annotated[str, Depends(require_control_user)]) -> dict:
db = get_control_db()
nodes = [_public_node(n) for n in db.list_nodes()]
return {"nodes": nodes}
@router.post("/")
async def create_node(
body: NodeCreate,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
db = get_control_db()
try:
row = db.create_node(body.name, body.base_url)
except Exception as e:
raise HTTPException(status_code=400, detail=f"创建失败: {e}") from e
return _public_node(row)
@router.get("/status/all")
async def status_all(_user: Annotated[str, Depends(require_control_user)]) -> dict:
db = get_control_db()
items = []
for node in db.list_nodes():
probe = await probe_health(node)
item = {**_public_node(node), **probe}
if probe.get("online") and node.get("token_sealed"):
code, data = await call_node(node, "GET", "/api/fleet/status")
if code == 200 and isinstance(data, dict):
item["fleet"] = data
item["fleet_ok"] = True
else:
item["fleet_ok"] = False
item["fleet_error"] = _http_detail(data) if data else f"HTTP {code}"
elif not node.get("token_sealed"):
item["fleet_ok"] = False
item["fleet_error"] = "未生成 Token"
items.append(item)
return {"nodes": items}
@router.post("/update-batch")
async def update_batch(
body: dict,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
ids = body.get("ids") or []
if not isinstance(ids, list) or not ids:
raise HTTPException(status_code=400, detail="ids 不能为空")
db = get_control_db()
results = []
for nid in ids:
node = db.get_node(int(nid))
if not node:
results.append({"id": nid, "ok": False, "detail": "不存在"})
continue
code, data = await call_node(node, "POST", "/api/fleet/update")
results.append(
{
"id": nid,
"ok": code < 400,
"status": code,
"result": data,
}
)
return {"results": results}
@router.patch("/{node_id}")
async def update_node(
node_id: int,
body: NodeUpdate,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
db = get_control_db()
row = db.update_node(node_id, name=body.name, base_url=body.base_url)
if not row:
raise HTTPException(status_code=404, detail="节点不存在")
return _public_node(row)
@router.delete("/{node_id}")
async def delete_node(
node_id: int,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
db = get_control_db()
if not db.delete_node(node_id):
raise HTTPException(status_code=404, detail="节点不存在")
return {"ok": True}
@router.post("/{node_id}/generate-token")
async def generate_token(
node_id: int,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
"""生成新 Token,加密存中控;明文仅返回一次,需粘贴到策略机设置。"""
db = get_control_db()
node = db.get_node(node_id)
if not node:
raise HTTPException(status_code=404, detail="节点不存在")
plain = secrets.token_urlsafe(32)
sealed = seal(plain, get_control_settings().control_auth_secret)
db.update_node(node_id, token_sealed=sealed)
return {
"ok": True,
"token": plain,
"msg": "请立即复制并到策略机「系统设置 → 登录账户 → 中控 API Token」保存",
}
@router.get("/{node_id}/status")
async def node_status(
node_id: int,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
db = get_control_db()
node = db.get_node(node_id)
if not node:
raise HTTPException(status_code=404, detail="节点不存在")
probe = await probe_health(node)
out = {**_public_node(node), **probe}
if probe.get("online") and node.get("token_sealed"):
code, data = await call_node(node, "GET", "/api/fleet/status")
if code == 200 and isinstance(data, dict):
out["fleet"] = data
return out
@router.post("/{node_id}/start")
async def node_start(
node_id: int,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
db = get_control_db()
node = db.get_node(node_id)
if not node:
raise HTTPException(status_code=404, detail="节点不存在")
code, data = await call_node(node, "POST", "/api/fleet/start")
if code >= 400:
_raise_node_error(code, data)
return {"ok": True, "result": data}
@router.post("/{node_id}/pause")
async def node_pause(
node_id: int,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
db = get_control_db()
node = db.get_node(node_id)
if not node:
raise HTTPException(status_code=404, detail="节点不存在")
code, data = await call_node(node, "POST", "/api/fleet/pause")
if code >= 400:
_raise_node_error(code, data)
return {"ok": True, "result": data}
@router.post("/{node_id}/update")
async def node_update(
node_id: int,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
db = get_control_db()
node = db.get_node(node_id)
if not node:
raise HTTPException(status_code=404, detail="节点不存在")
code, data = await call_node(node, "POST", "/api/fleet/update")
if code >= 400:
_raise_node_error(code, data)
return {"ok": True, "result": data}
@router.post("/{node_id}/login-url")
async def node_login_url(
node_id: int,
_user: Annotated[str, Depends(require_control_user)],
) -> dict:
"""用 Fleet Token 向策略机签发一次性登录票,返回可打开的 URL。"""
db = get_control_db()
node = db.get_node(node_id)
if not node:
raise HTTPException(status_code=404, detail="节点不存在")
code, data = await call_node(node, "POST", "/api/fleet/issue-login")
if code >= 400:
_raise_node_error(code, data)
path = ""
if isinstance(data, dict):
path = str(data.get("login_path") or "")
if not path:
raise HTTPException(status_code=502, detail="策略机未返回 login_path")
base = str(node["base_url"]).rstrip("/")
return {
"ok": True,
"url": f"{base}{path}",
"expires_in": data.get("expires_in") if isinstance(data, dict) else None,
}