58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""Operation Catalog API。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
|
|
from ..core.response import success_response
|
|
from ..dependencies.auth import get_current_user
|
|
from ..models import User
|
|
from ..services.auth import has_any_role
|
|
from ..services.operation_catalog_service import operation_catalog_service
|
|
|
|
router = APIRouter(prefix="/ai-steward/operation-catalog", tags=["AI Steward"])
|
|
|
|
|
|
def _require_manager(user: User) -> None:
|
|
if not has_any_role(str(user.role or ""), ["BOSS", "ADMIN"]):
|
|
raise HTTPException(status_code=403, detail="仅管理员可查看 Operation Catalog 质量报告")
|
|
|
|
|
|
@router.get("")
|
|
async def list_operation_catalog(
|
|
domain: Optional[str] = Query(default=None),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
_ = current_user
|
|
rows = operation_catalog_service.list_operations(domain=domain)
|
|
return success_response(
|
|
{
|
|
"items": rows,
|
|
"count": len(rows),
|
|
"domain": domain,
|
|
}
|
|
)
|
|
|
|
|
|
@router.get("/lint")
|
|
async def lint_operation_catalog(
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
_require_manager(current_user)
|
|
report = operation_catalog_service.lint_catalog()
|
|
return success_response(report)
|
|
|
|
|
|
@router.get("/{operation_code}")
|
|
async def get_operation_catalog_item(
|
|
operation_code: str,
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
_ = current_user
|
|
row = operation_catalog_service.get_operation(operation_code=operation_code)
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="operation_code not found")
|
|
return success_response(row)
|