93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
"""Administrator-managed exact-match fixed question/answer API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from app.gateway.deps import get_current_user, get_optional_user_from_request
|
|
from deerflow.persistence.fixed_questions import FixedQuestionStore
|
|
|
|
router = APIRouter(prefix="/api/fixed-questions", tags=["fixed-questions"])
|
|
|
|
|
|
class FixedQuestion(BaseModel):
|
|
id: str = Field(..., min_length=1, max_length=64)
|
|
question: str = Field(..., min_length=1, max_length=10_000)
|
|
answer: str = Field(..., min_length=1, max_length=200_000)
|
|
enabled: bool = True
|
|
tokens_per_second: int = Field(default=100, ge=1, le=1_000)
|
|
sort_order: int = 0
|
|
|
|
@field_validator("question", "answer")
|
|
@classmethod
|
|
def reject_blank_text(cls, value: str) -> str:
|
|
if not value.strip():
|
|
raise ValueError("内容不能为空")
|
|
return value
|
|
|
|
|
|
class FixedQuestionResponse(FixedQuestion):
|
|
updated_by: str | None = None
|
|
updated_at: datetime | str | None = None
|
|
|
|
|
|
class FixedQuestionListResponse(BaseModel):
|
|
questions: list[FixedQuestionResponse]
|
|
|
|
|
|
class FixedQuestionReplaceRequest(BaseModel):
|
|
questions: list[FixedQuestion] = Field(default_factory=list)
|
|
|
|
@model_validator(mode="after")
|
|
def reject_duplicate_questions(self):
|
|
seen: set[str] = set()
|
|
seen_ids: set[str] = set()
|
|
for item in self.questions:
|
|
if item.id in seen_ids:
|
|
raise ValueError(f"配置 ID「{item.id}」重复")
|
|
seen_ids.add(item.id)
|
|
if item.question in seen:
|
|
raise ValueError(f"问题「{item.question}」重复")
|
|
seen.add(item.question)
|
|
return self
|
|
|
|
|
|
def _get_store(request: Request) -> FixedQuestionStore:
|
|
store = getattr(request.app.state, "fixed_question_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="Fixed question store not available")
|
|
return store
|
|
|
|
|
|
async def _require_admin(request: Request) -> None:
|
|
"""Require an administrator, while preserving auth-disabled deployments."""
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
return
|
|
if getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(status_code=403, detail="问题配置仅限管理员")
|
|
|
|
|
|
@router.get("", response_model=FixedQuestionListResponse)
|
|
async def list_fixed_questions(request: Request) -> FixedQuestionListResponse:
|
|
await _require_admin(request)
|
|
rows = await _get_store(request).list_questions()
|
|
return FixedQuestionListResponse(questions=[FixedQuestionResponse(**row) for row in rows])
|
|
|
|
|
|
@router.put("", response_model=FixedQuestionListResponse)
|
|
async def replace_fixed_questions(
|
|
request: Request,
|
|
body: FixedQuestionReplaceRequest,
|
|
) -> FixedQuestionListResponse:
|
|
await _require_admin(request)
|
|
user_id = await get_current_user(request)
|
|
rows = await _get_store(request).replace_all(
|
|
[question.model_dump() for question in body.questions],
|
|
updated_by=user_id,
|
|
)
|
|
return FixedQuestionListResponse(questions=[FixedQuestionResponse(**row) for row in rows])
|