64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.deps import require_staff
|
|
from app.db import get_db
|
|
from app.models.expert import admin_question, admin_questions, pending_count, reply_question
|
|
|
|
router = APIRouter(dependencies=[Depends(require_staff)])
|
|
|
|
|
|
class ReplyWrite(BaseModel):
|
|
bodyHtml: str = ""
|
|
attachmentFileIds: list[int] = Field(default_factory=list)
|
|
|
|
|
|
def _staff_id(principal: dict) -> int:
|
|
try:
|
|
return int(principal["sub"])
|
|
except (TypeError, ValueError, KeyError) as exc:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录") from exc
|
|
|
|
|
|
@router.get("/expert-questions/pending-count")
|
|
def read_pending_count(db: Session = Depends(get_db)) -> dict:
|
|
return {"count": pending_count(db)}
|
|
|
|
|
|
@router.get("/expert-questions")
|
|
def list_questions(
|
|
status_filter: str | None = Query(None, alias="status"),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
return admin_questions(db, status=status_filter)
|
|
|
|
|
|
@router.get("/expert-questions/{question_id}")
|
|
def read_question(question_id: int, db: Session = Depends(get_db)) -> dict:
|
|
body = admin_question(db, question_id)
|
|
if body is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="提问不存在")
|
|
return body
|
|
|
|
|
|
@router.post("/expert-questions/{question_id}/reply")
|
|
def write_reply(
|
|
question_id: int,
|
|
body: ReplyWrite,
|
|
principal: dict = Depends(require_staff),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
try:
|
|
reply_question(
|
|
db,
|
|
question_id=question_id,
|
|
staff_id=_staff_id(principal),
|
|
body_html=body.bodyHtml,
|
|
attachment_file_ids=body.attachmentFileIds,
|
|
)
|
|
except ValueError as exc:
|
|
code = status.HTTP_404_NOT_FOUND if str(exc) == "提问不存在" else status.HTTP_400_BAD_REQUEST
|
|
raise HTTPException(status_code=code, detail=str(exc)) from exc
|
|
return {"ok": True}
|