wren
188 lines
11 KiB
Python
188 lines
11 KiB
Python
"""规则配置控制器。"""
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import Depends
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from fastapi_common.fastapi_common_security.security import verify_access_token
|
|
from fastapi_common.fastapi_common_web.controller import BaseController
|
|
from fastapi_common.fastapi_common_web.domain.responses import Result
|
|
|
|
from fastapi_modules.fastapi_leaudit.domian.Dto.ruleBindingDto import (
|
|
RuleBindingCreateDTO,
|
|
RuleBindingUpdateDTO,
|
|
)
|
|
from fastapi_modules.fastapi_leaudit.domian.Dto.rulePublishDto import RulePublishDTO
|
|
from fastapi_modules.fastapi_leaudit.domian.Dto.ruleValidateDto import RuleValidateDTO
|
|
from fastapi_modules.fastapi_leaudit.domian.Dto.ruleVersionCreateDto import RuleVersionCreateDTO
|
|
from fastapi_modules.fastapi_leaudit.domian.vo.ruleVo import (
|
|
RuleBindingVO,
|
|
RuleContentVO,
|
|
RuleSetVO,
|
|
RuleValidationVO,
|
|
RuleVersionVO,
|
|
)
|
|
from fastapi_modules.fastapi_leaudit.services import IRuleService
|
|
from fastapi_modules.fastapi_leaudit.services.impl.permissionServiceImpl import PermissionServiceImpl
|
|
from fastapi_modules.fastapi_leaudit.services.impl.ruleServiceImpl import GetRuleServiceSingleton
|
|
from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissionService
|
|
|
|
|
|
class RuleController(BaseController):
|
|
"""规则配置控制器。"""
|
|
|
|
def __init__(self):
|
|
super().__init__(prefix="/rule-sets", tags=["规则配置"])
|
|
self.RuleService: IRuleService = GetRuleServiceSingleton()
|
|
self.PermissionService: IPermissionService = PermissionServiceImpl()
|
|
self._PERMISSIONS = {
|
|
"list": "rules:list:read",
|
|
"version_list": "rules:version_list:read",
|
|
"content": "rules:content:read",
|
|
"validate": "rules:validate:execute",
|
|
"create": "rules:version_create:write",
|
|
"publish": "rules:publish:write",
|
|
"rollback": "rules:rollback:write",
|
|
"binding_read": "rules:binding_list:read",
|
|
"binding_create": "rules:binding_create:write",
|
|
"binding_update": "rules:binding_update:write",
|
|
"binding_delete": "rules:binding_delete:delete",
|
|
}
|
|
|
|
@self.router.get("", response_model=Result[list[RuleSetVO]])
|
|
async def ListRuleSets(payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""列出规则集。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["list"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有规则集查看权限", "data": None})
|
|
Data = await self.RuleService.ListSets(CurrentUserId=int(payload["user_id"]))
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.get("/{RuleType}/versions", response_model=Result[list[RuleVersionVO]])
|
|
async def GetVersions(RuleType: str, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""列出规则集的所有版本。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["version_list"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有规则版本查看权限", "data": None})
|
|
Data = await self.RuleService.GetVersions(RuleType=RuleType, CurrentUserId=int(payload["user_id"]))
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.get("/versions/{VersionId}/content", response_model=Result[RuleContentVO])
|
|
async def GetVersionContent(VersionId: int, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""获取规则版本正文。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["content"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有规则正文查看权限", "data": None})
|
|
Data = await self.RuleService.GetContent(VersionId=VersionId, CurrentUserId=int(payload["user_id"]))
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.post("/{RuleType}/validate", response_model=Result[RuleValidationVO])
|
|
async def ValidateRuleYaml(RuleType: str, body: RuleValidateDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""校验规则 YAML。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["validate"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有规则校验权限", "data": None})
|
|
Data = await self.RuleService.Validate(
|
|
RuleType=RuleType,
|
|
YamlText=body.yamlText,
|
|
)
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.post("/{RuleType}/versions", response_model=Result[RuleVersionVO])
|
|
async def CreateRuleVersion(RuleType: str, body: RuleVersionCreateDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""创建规则版本。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["create"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有创建规则版本权限", "data": None})
|
|
Data = await self.RuleService.CreateVersion(
|
|
RuleType=RuleType,
|
|
YamlText=body.yamlText,
|
|
ChangeNote=body.changeNote,
|
|
EditorUserId=int(payload["user_id"]),
|
|
CurrentUserId=int(payload["user_id"]),
|
|
)
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.post("/{RuleType}/publish", response_model=Result[RuleVersionVO])
|
|
async def PublishRuleVersion(RuleType: str, body: RulePublishDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""发布规则版本。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["publish"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有发布规则权限", "data": None})
|
|
Data = await self.RuleService.Publish(
|
|
RuleType=RuleType,
|
|
VersionId=body.versionId,
|
|
OperatorUserId=int(payload["user_id"]),
|
|
CurrentUserId=int(payload["user_id"]),
|
|
)
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.post("/{RuleType}/rollback", response_model=Result[RuleVersionVO])
|
|
async def RollbackRuleVersion(RuleType: str, body: RulePublishDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""回滚到指定规则版本。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["rollback"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有回滚规则权限", "data": None})
|
|
Data = await self.RuleService.Rollback(
|
|
RuleType=RuleType,
|
|
VersionId=body.versionId,
|
|
OperatorUserId=int(payload["user_id"]),
|
|
CurrentUserId=int(payload["user_id"]),
|
|
)
|
|
return Result.success(data=Data)
|
|
|
|
# ── 规则类型绑定 ──────────────────────────────────────────
|
|
|
|
@self.router.get("/bindings", response_model=Result[list[RuleBindingVO]])
|
|
async def ListBindings(
|
|
ruleType: str | None = None,
|
|
region: str | None = None,
|
|
payload: dict[str, Any] = Depends(verify_access_token),
|
|
):
|
|
"""列出规则类型绑定。当前主要按规则类型过滤，region 仅兼容保留。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["binding_read"], self._PERMISSIONS["list"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有规则绑定查看权限", "data": None})
|
|
Data = await self.RuleService.ListBindings(
|
|
RuleType=ruleType,
|
|
Region=region,
|
|
CurrentUserId=int(payload["user_id"]),
|
|
)
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.post("/{RuleType}/bindings", response_model=Result[RuleBindingVO])
|
|
async def CreateBinding(RuleType: str, body: RuleBindingCreateDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""创建规则类型绑定。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["binding_create"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有创建规则绑定权限", "data": None})
|
|
Data = await self.RuleService.CreateBinding(
|
|
DocTypeId=body.docTypeId,
|
|
RuleSetId=body.ruleSetId,
|
|
Region=body.region,
|
|
BindingMode=body.bindingMode,
|
|
Priority=body.priority,
|
|
DocTypeCode=body.docTypeCode,
|
|
Note=body.note,
|
|
CurrentUserId=int(payload["user_id"]),
|
|
)
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.put("/bindings/{BindingId}", response_model=Result[RuleBindingVO])
|
|
async def UpdateBinding(BindingId: int, body: RuleBindingUpdateDTO, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""更新规则类型绑定。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["binding_update"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有更新规则绑定权限", "data": None})
|
|
Data = await self.RuleService.UpdateBinding(
|
|
BindingId=BindingId,
|
|
IsActive=body.isActive,
|
|
Priority=body.priority,
|
|
BindingMode=body.bindingMode,
|
|
Note=body.note,
|
|
CurrentUserId=int(payload["user_id"]),
|
|
)
|
|
return Result.success(data=Data)
|
|
|
|
@self.router.delete("/bindings/{BindingId}", response_model=Result[None])
|
|
async def DeleteBinding(BindingId: int, payload: dict[str, Any] = Depends(verify_access_token)):
|
|
"""删除规则类型绑定。"""
|
|
if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["binding_delete"]]):
|
|
return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有删除规则绑定权限", "data": None})
|
|
await self.RuleService.DeleteBinding(BindingId=BindingId, CurrentUserId=int(payload["user_id"]))
|
|
return Result.success()
|
|
|
|
async def _check_permission(self, user_id: int, permission_keys: list[str]) -> bool:
|
|
return await self.PermissionService.HasAnyPermission(UserId=user_id, PermissionKeys=permission_keys)
|