220 lines
7.4 KiB
Python
220 lines
7.4 KiB
Python
"""관리자 전용 상담 프로토콜 등록·활성화·퇴역 API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Annotated, Literal
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from ..db import acquire
|
|
from ..deps import Principal, Role, require_role
|
|
from ..services import protocol_registry
|
|
|
|
router = APIRouter(prefix="/admin/protocols", tags=["admin-protocols"])
|
|
AdminPrincipal = Annotated[Principal, Depends(require_role(Role.ADMIN))]
|
|
|
|
|
|
class AdminProtocolCreate(BaseModel):
|
|
title: str = Field(..., min_length=1, max_length=240)
|
|
source: str = Field(..., min_length=1, max_length=1000)
|
|
version: int = Field(default=1, ge=1, le=1_000_000)
|
|
license: Literal["A", "B", "C", "D"]
|
|
external_llm_ok: bool = False
|
|
content: str = Field(..., min_length=1, max_length=500_000)
|
|
|
|
@field_validator("title", "source", "content")
|
|
@classmethod
|
|
def reject_blank_text(cls, value: str) -> str:
|
|
if not value.strip():
|
|
raise ValueError("빈 값은 등록할 수 없습니다.")
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def enforce_license_boundary(self) -> "AdminProtocolCreate":
|
|
if self.license in {"C", "D"} and self.external_llm_ok:
|
|
raise ValueError("라이선스 C/D는 외부 LLM 사용을 허용할 수 없습니다.")
|
|
return self
|
|
|
|
|
|
class AdminProtocolResponse(BaseModel):
|
|
protocol_id: str
|
|
source_id: str
|
|
title: str
|
|
source: str
|
|
version: int
|
|
license: Literal["A", "B", "C", "D"]
|
|
external_llm_ok: bool
|
|
content: str
|
|
content_hash: str
|
|
status: Literal["draft", "active", "retired"]
|
|
registered_by: str
|
|
registered_at: datetime
|
|
activated_at: datetime | None = None
|
|
retired_at: datetime | None = None
|
|
|
|
|
|
class AdminProtocolListResponse(BaseModel):
|
|
protocols: list[AdminProtocolResponse]
|
|
total: int
|
|
|
|
|
|
class AdminProtocolActivationResponse(BaseModel):
|
|
protocol: AdminProtocolResponse
|
|
chunks_indexed: int
|
|
skipped_unchanged: bool
|
|
embedded: bool
|
|
degraded: bool
|
|
|
|
|
|
def _response(record: protocol_registry.ProtocolRecord) -> AdminProtocolResponse:
|
|
return AdminProtocolResponse(
|
|
protocol_id=record.protocol_id,
|
|
source_id=record.source_id,
|
|
title=record.title,
|
|
source=record.source,
|
|
version=record.version,
|
|
license=record.license,
|
|
external_llm_ok=record.external_llm_ok,
|
|
content=record.content,
|
|
content_hash=record.content_hash,
|
|
status=record.status,
|
|
registered_by=record.registered_by,
|
|
registered_at=record.registered_at,
|
|
activated_at=record.activated_at,
|
|
retired_at=record.retired_at,
|
|
)
|
|
|
|
|
|
def _raise_http(error: protocol_registry.ProtocolRegistryError) -> None:
|
|
if isinstance(error, protocol_registry.ProtocolNotFound):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, detail=str(error)) from error
|
|
if isinstance(error, protocol_registry.ProtocolTransitionConflict):
|
|
raise HTTPException(status.HTTP_409_CONFLICT, detail=str(error)) from error
|
|
if isinstance(error, protocol_registry.ProtocolPolicyViolation):
|
|
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(error)) from error
|
|
raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(error)) from error
|
|
|
|
|
|
@router.get("", response_model=AdminProtocolListResponse)
|
|
async def list_admin_protocols(
|
|
principal: AdminPrincipal,
|
|
status_filter: Annotated[
|
|
Literal["draft", "active", "retired"] | None,
|
|
Query(alias="status"),
|
|
] = None,
|
|
search: Annotated[str | None, Query(max_length=240)] = None,
|
|
) -> AdminProtocolListResponse:
|
|
try:
|
|
async with acquire(role="admin", user_id=principal.user_id) as conn:
|
|
records = await protocol_registry.list_protocols(
|
|
conn,
|
|
status_filter=status_filter,
|
|
search=search,
|
|
)
|
|
except protocol_registry.ProtocolRegistryError as error:
|
|
_raise_http(error)
|
|
except RuntimeError as error:
|
|
raise HTTPException(
|
|
status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"프로토콜 저장소를 사용할 수 없습니다: {error}",
|
|
) from error
|
|
return AdminProtocolListResponse(
|
|
protocols=[_response(record) for record in records],
|
|
total=len(records),
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=AdminProtocolResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def create_admin_protocol(
|
|
body: AdminProtocolCreate,
|
|
principal: AdminPrincipal,
|
|
) -> AdminProtocolResponse:
|
|
try:
|
|
async with acquire(role="admin", user_id=principal.user_id) as conn:
|
|
record = await protocol_registry.create_protocol(
|
|
conn,
|
|
title=body.title,
|
|
source=body.source,
|
|
version=body.version,
|
|
license_class=body.license,
|
|
external_llm_ok=body.external_llm_ok,
|
|
content=body.content,
|
|
registered_by=principal.user_id,
|
|
)
|
|
except protocol_registry.ProtocolRegistryError as error:
|
|
_raise_http(error)
|
|
except RuntimeError as error:
|
|
raise HTTPException(
|
|
status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"프로토콜 저장소를 사용할 수 없습니다: {error}",
|
|
) from error
|
|
return _response(record)
|
|
|
|
|
|
@router.post(
|
|
"/{protocol_id}/activate",
|
|
response_model=AdminProtocolActivationResponse,
|
|
)
|
|
async def activate_admin_protocol(
|
|
protocol_id: UUID,
|
|
principal: AdminPrincipal,
|
|
) -> AdminProtocolActivationResponse:
|
|
try:
|
|
async with acquire(role="admin", user_id=principal.user_id) as conn:
|
|
async with conn.transaction():
|
|
record, indexed = await protocol_registry.activate_protocol(
|
|
conn,
|
|
protocol_id=str(protocol_id),
|
|
)
|
|
except protocol_registry.ProtocolRegistryError as error:
|
|
_raise_http(error)
|
|
except RuntimeError as error:
|
|
raise HTTPException(
|
|
status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"프로토콜 저장소를 사용할 수 없습니다: {error}",
|
|
) from error
|
|
return AdminProtocolActivationResponse(
|
|
protocol=_response(record),
|
|
chunks_indexed=indexed.chunks_indexed,
|
|
skipped_unchanged=indexed.skipped_unchanged,
|
|
embedded=indexed.embedded,
|
|
degraded=indexed.degraded,
|
|
)
|
|
|
|
|
|
@router.post("/{protocol_id}/retire", response_model=AdminProtocolResponse)
|
|
async def retire_admin_protocol(
|
|
protocol_id: UUID,
|
|
principal: AdminPrincipal,
|
|
) -> AdminProtocolResponse:
|
|
try:
|
|
async with acquire(role="admin", user_id=principal.user_id) as conn:
|
|
async with conn.transaction():
|
|
record = await protocol_registry.retire_protocol(
|
|
conn,
|
|
protocol_id=str(protocol_id),
|
|
)
|
|
except protocol_registry.ProtocolRegistryError as error:
|
|
_raise_http(error)
|
|
except RuntimeError as error:
|
|
raise HTTPException(
|
|
status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=f"프로토콜 저장소를 사용할 수 없습니다: {error}",
|
|
) from error
|
|
return _response(record)
|
|
|
|
|
|
__all__ = [
|
|
"AdminProtocolActivationResponse",
|
|
"AdminProtocolCreate",
|
|
"AdminProtocolListResponse",
|
|
"AdminProtocolResponse",
|
|
"router",
|
|
]
|