282 lines
8.9 KiB
Python
282 lines
8.9 KiB
Python
"""规则配置 API (MVP-3 / D33 配套后端)
|
||
|
||
提供 YAML 规则文件的增删改查接口，供前端 ``RuleConfiguration.vue`` 调用。
|
||
|
||
接口列表:
|
||
|
||
- ``GET /rules`` 列出所有规则 (按文件分组)
|
||
- ``GET /rules/{name}`` 获取单条规则详情
|
||
- ``POST /rules`` 新增规则
|
||
- ``PUT /rules/{name}`` 更新规则
|
||
- ``DELETE /rules/{name}`` 删除规则
|
||
- ``POST /rules/reload`` 热重载规则目录
|
||
- ``GET /rules/_stats`` 规则引擎运行时状态
|
||
|
||
规则文件落盘策略:
|
||
|
||
- 默认所有自定义规则写入 ``config/rules/custom.yaml``
|
||
- 不修改 MVP-1 内置的事件类型规则文件 (fire.yaml / smoking.yaml 等) 的"内置规则"，
|
||
但可以通过本接口在其上追加新规则或编辑已有规则
|
||
- 安全保护: 规则名作为唯一键，禁止同名覆盖；删除时仅允许删除 "custom.yaml" 内的规则
|
||
|
||
设计要点:
|
||
|
||
- 规则编辑后会刷新 ``DetectionService.rule_engine`` 实例，确保实时生效
|
||
- 写入 YAML 时使用 ``yaml.safe_dump``，保留 schema 兼容性
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import yaml
|
||
from fastapi import APIRouter, HTTPException
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
|
||
from core.settings import get_settings
|
||
from models.event_schemas import EventType, SeverityLevel
|
||
from services.event.rule_engine import AlertRule, AlertRuleEngine
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/rules", tags=["规则配置"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 全局引擎注入
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
_rule_engine_provider = None
|
||
|
||
|
||
def init_rules_api(rule_engine_provider) -> None:
|
||
"""注入规则引擎获取函数。
|
||
|
||
使用 callable 而非实例，便于多个模块 (DetectionService 等) 共享同一引擎，
|
||
并在调用时拿到最新引用。
|
||
"""
|
||
|
||
global _rule_engine_provider
|
||
_rule_engine_provider = rule_engine_provider
|
||
|
||
|
||
def _get_engine() -> AlertRuleEngine:
|
||
if _rule_engine_provider is None:
|
||
raise HTTPException(status_code=503, detail="规则引擎未初始化")
|
||
engine = _rule_engine_provider()
|
||
if engine is None:
|
||
raise HTTPException(status_code=503, detail="规则引擎未就绪")
|
||
return engine
|
||
|
||
|
||
def _custom_rules_path() -> Path:
|
||
settings = get_settings()
|
||
rules_dir = settings.paths.server_dir / settings.event_engine.rules_dir
|
||
rules_dir.mkdir(parents=True, exist_ok=True)
|
||
return rules_dir / "custom.yaml"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 请求 / 响应模型
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class RulePayload(BaseModel):
|
||
"""规则的可序列化表示。"""
|
||
|
||
model_config = ConfigDict(extra="ignore")
|
||
|
||
name: str = Field(..., min_length=1, max_length=64)
|
||
event_type: EventType
|
||
enabled: bool = True
|
||
min_confidence: float = Field(default=0.0, ge=0.0, le=1.0)
|
||
severity: Optional[SeverityLevel] = None
|
||
allowed_sources: Optional[List[str]] = None
|
||
required_labels: Optional[List[str]] = None
|
||
min_bbox_area: int = Field(default=0, ge=0)
|
||
description: str = ""
|
||
|
||
@classmethod
|
||
def from_rule(cls, rule: AlertRule) -> "RulePayload":
|
||
return cls(
|
||
name=rule.name,
|
||
event_type=rule.event_type,
|
||
enabled=rule.enabled,
|
||
min_confidence=rule.min_confidence,
|
||
severity=rule.severity,
|
||
allowed_sources=rule.allowed_sources,
|
||
required_labels=rule.required_labels,
|
||
min_bbox_area=rule.min_bbox_area,
|
||
description=rule.description,
|
||
)
|
||
|
||
def to_yaml_dict(self) -> Dict[str, Any]:
|
||
data: Dict[str, Any] = {
|
||
"name": self.name,
|
||
"event_type": self.event_type.value,
|
||
"enabled": self.enabled,
|
||
"min_confidence": self.min_confidence,
|
||
"min_bbox_area": self.min_bbox_area,
|
||
"description": self.description,
|
||
}
|
||
if self.severity is not None:
|
||
data["severity"] = self.severity.value
|
||
if self.allowed_sources:
|
||
data["allowed_sources"] = list(self.allowed_sources)
|
||
if self.required_labels:
|
||
data["required_labels"] = list(self.required_labels)
|
||
return data
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# YAML 持久化
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _load_custom_rules() -> List[Dict[str, Any]]:
|
||
path = _custom_rules_path()
|
||
if not path.exists():
|
||
return []
|
||
with path.open("r", encoding="utf-8") as fp:
|
||
data = yaml.safe_load(fp) or {}
|
||
if isinstance(data, dict) and "rules" in data:
|
||
return list(data["rules"])
|
||
if isinstance(data, list):
|
||
return list(data)
|
||
return []
|
||
|
||
|
||
def _save_custom_rules(rules: List[Dict[str, Any]]) -> None:
|
||
path = _custom_rules_path()
|
||
payload = {
|
||
"# generated by /api/rules MVP-3 D33": None,
|
||
"rules": rules,
|
||
}
|
||
payload.pop("# generated by /api/rules MVP-3 D33") # 仅作注释占位
|
||
with path.open("w", encoding="utf-8") as fp:
|
||
fp.write("# 由前端规则配置页面写入，请勿手工同时编辑\n")
|
||
yaml.safe_dump(
|
||
{"rules": rules},
|
||
fp,
|
||
allow_unicode=True,
|
||
sort_keys=False,
|
||
indent=2,
|
||
)
|
||
|
||
|
||
def _reload_engine() -> AlertRuleEngine:
|
||
"""重新从规则目录加载，并替换全局引擎的规则集合。"""
|
||
|
||
settings = get_settings()
|
||
rules_dir = settings.paths.server_dir / settings.event_engine.rules_dir
|
||
new_engine = AlertRuleEngine.from_directory(rules_dir)
|
||
|
||
engine = _get_engine()
|
||
engine.rules = new_engine.rules
|
||
engine._stats = new_engine._stats # noqa: SLF001
|
||
logger.info("规则引擎已热重载: 共 %d 条", len(engine.rules))
|
||
return engine
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@router.get("")
|
||
async def list_rules() -> Dict[str, Any]:
|
||
"""列出所有规则 (含来源文件)。"""
|
||
|
||
engine = _get_engine()
|
||
return {
|
||
"rules": [RulePayload.from_rule(r).model_dump(mode="json") for r in engine.rules],
|
||
"stats": engine.stats,
|
||
}
|
||
|
||
|
||
@router.get("/_stats")
|
||
async def rule_stats() -> Dict[str, Any]:
|
||
"""规则引擎运行状态。"""
|
||
|
||
engine = _get_engine()
|
||
return engine.stats
|
||
|
||
|
||
@router.get("/{name}")
|
||
async def get_rule(name: str) -> RulePayload:
|
||
"""获取单条规则。"""
|
||
|
||
engine = _get_engine()
|
||
for rule in engine.rules:
|
||
if rule.name == name:
|
||
return RulePayload.from_rule(rule)
|
||
raise HTTPException(status_code=404, detail=f"规则 {name} 不存在")
|
||
|
||
|
||
@router.post("")
|
||
async def create_rule(payload: RulePayload) -> Dict[str, Any]:
|
||
"""新增规则 (写入 custom.yaml)。"""
|
||
|
||
engine = _get_engine()
|
||
if any(r.name == payload.name for r in engine.rules):
|
||
raise HTTPException(status_code=409, detail=f"规则 {payload.name} 已存在")
|
||
|
||
rules = _load_custom_rules()
|
||
rules.append(payload.to_yaml_dict())
|
||
_save_custom_rules(rules)
|
||
|
||
_reload_engine()
|
||
return {"success": True, "rule": payload.model_dump(mode="json")}
|
||
|
||
|
||
@router.put("/{name}")
|
||
async def update_rule(name: str, payload: RulePayload) -> Dict[str, Any]:
|
||
"""更新规则 (仅支持 custom.yaml 中的规则)。"""
|
||
|
||
if payload.name != name:
|
||
raise HTTPException(status_code=400, detail="路径与请求体中的规则名不一致")
|
||
|
||
rules = _load_custom_rules()
|
||
for idx, item in enumerate(rules):
|
||
if item.get("name") == name:
|
||
rules[idx] = payload.to_yaml_dict()
|
||
_save_custom_rules(rules)
|
||
_reload_engine()
|
||
return {"success": True, "rule": payload.model_dump(mode="json")}
|
||
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail=f"规则 {name} 不在 custom.yaml 中，无法编辑 (内置规则只读)",
|
||
)
|
||
|
||
|
||
@router.delete("/{name}")
|
||
async def delete_rule(name: str) -> Dict[str, Any]:
|
||
"""删除规则 (仅 custom.yaml)。"""
|
||
|
||
rules = _load_custom_rules()
|
||
new_rules = [r for r in rules if r.get("name") != name]
|
||
if len(new_rules) == len(rules):
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail=f"规则 {name} 不在 custom.yaml 中，无法删除 (内置规则只读)",
|
||
)
|
||
|
||
_save_custom_rules(new_rules)
|
||
_reload_engine()
|
||
return {"success": True, "deleted": name}
|
||
|
||
|
||
@router.post("/reload")
|
||
async def reload_rules() -> Dict[str, Any]:
|
||
"""热重载规则目录 (供运维直接编辑 YAML 后调用)。"""
|
||
|
||
engine = _reload_engine()
|
||
return {"success": True, "stats": engine.stats}
|
||
|
||
|
||
__all__ = ["router", "init_rules_api"]
|