nearxos
6185b9b85a
Self-hosted, Docker-based agentic troubleshooting platform: FastAPI backend + LangGraph agent, Next.js UI, tiered LLM routing (local Ollama -> Gemini -> DeepSeek -> OpenRouter), MCP server manager, encrypted device credentials, RBAC, audit log, project-memory + Obsidian integrations, and editable troubleshooting decision rules tuned for the GeneseasX vessel stack. Co-authored-by: Cursor <cursoragent@cursor.com>
70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Editable troubleshooting rules API."""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
|
|
from app.core.deps import require_admin, require_readonly
|
|
from app.models.user import User
|
|
from app.schemas import TroubleshootingRulesOut, TroubleshootingRulesUpdate
|
|
from app.services.audit import record_audit
|
|
from app.services.troubleshooting_rules import (
|
|
load_rules,
|
|
match_rule,
|
|
read_rules_raw,
|
|
save_rules_raw,
|
|
)
|
|
from app.database import get_db
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
router = APIRouter(prefix="/api/rules", tags=["rules"])
|
|
|
|
|
|
@router.get("/troubleshooting", response_model=TroubleshootingRulesOut)
|
|
async def get_troubleshooting_rules(_: User = Depends(require_readonly)):
|
|
data = load_rules() # mtime-cached; reloads only when the file changes
|
|
return TroubleshootingRulesOut(
|
|
yaml=read_rules_raw(),
|
|
parsed=data,
|
|
path_hint="rules/troubleshooting.yaml",
|
|
)
|
|
|
|
|
|
@router.put("/troubleshooting", response_model=TroubleshootingRulesOut)
|
|
async def update_troubleshooting_rules(
|
|
body: TroubleshootingRulesUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(require_admin),
|
|
):
|
|
try:
|
|
parsed = save_rules_raw(body.yaml)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status_code=400, detail=f"Invalid rules: {exc}") from exc
|
|
|
|
await record_audit(
|
|
db,
|
|
user,
|
|
"rules.troubleshooting.update",
|
|
"rules",
|
|
0,
|
|
{"bytes": len(body.yaml)},
|
|
)
|
|
return TroubleshootingRulesOut(
|
|
yaml=read_rules_raw(),
|
|
parsed=parsed,
|
|
path_hint="rules/troubleshooting.yaml",
|
|
)
|
|
|
|
|
|
@router.post("/troubleshooting/preview")
|
|
async def preview_rule_match(
|
|
body: dict,
|
|
_: User = Depends(require_readonly),
|
|
):
|
|
"""Preview which rule and devices match a title + issue (for the Rules UI)."""
|
|
title = str(body.get("title") or "")
|
|
issue = str(body.get("issue") or "")
|
|
matched = match_rule(title, issue)
|
|
return {"matched": matched.as_dict() if matched else None}
|