402 lines
12 KiB
Python
402 lines
12 KiB
Python
"""
|
|
满意度调查 API 接口
|
|
"""
|
|
from typing import Optional, List
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from pydantic import BaseModel, Field, ConfigDict
|
|
from datetime import datetime
|
|
import json
|
|
|
|
from app.core.database import get_db
|
|
from app.models.models import SurveyStatus, SurveyType, QuestionType
|
|
from app.services.survey_service import SurveyService
|
|
|
|
|
|
router = APIRouter(prefix="/surveys", tags=["满意度调查"])
|
|
|
|
|
|
# ==================== Pydantic Schemas ====================
|
|
|
|
class QuestionCreate(BaseModel):
|
|
"""题目创建"""
|
|
question_text: str = Field(..., description="题目内容")
|
|
question_type: str = Field(..., description="题目类型")
|
|
options: Optional[List[dict]] = Field(None, description="选项列表")
|
|
score_max: int = Field(5, description="最高分值")
|
|
is_required: bool = Field(True, description="是否必答")
|
|
|
|
|
|
class QuestionResponse(BaseModel):
|
|
"""题目响应"""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
question_text: str
|
|
question_type: str
|
|
options: Optional[str] = None
|
|
score_max: int
|
|
is_required: bool
|
|
sort_order: int
|
|
|
|
|
|
class SurveyCreate(BaseModel):
|
|
"""问卷创建"""
|
|
survey_name: str = Field(..., description="问卷名称")
|
|
survey_code: str = Field(..., description="问卷编码")
|
|
survey_type: str = Field(..., description="问卷类型")
|
|
description: Optional[str] = Field(None, description="描述")
|
|
target_departments: Optional[List[int]] = Field(None, description="目标科室")
|
|
is_anonymous: bool = Field(True, description="是否匿名")
|
|
questions: Optional[List[QuestionCreate]] = Field(None, description="题目列表")
|
|
|
|
|
|
class SurveyUpdate(BaseModel):
|
|
"""问卷更新"""
|
|
survey_name: Optional[str] = None
|
|
description: Optional[str] = None
|
|
target_departments: Optional[List[int]] = None
|
|
is_anonymous: Optional[bool] = None
|
|
|
|
|
|
class SurveyResponse(BaseModel):
|
|
"""问卷响应"""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
survey_name: str
|
|
survey_code: str
|
|
survey_type: str
|
|
description: Optional[str] = None
|
|
target_departments: Optional[str] = None
|
|
total_questions: int
|
|
status: str
|
|
start_date: Optional[datetime] = None
|
|
end_date: Optional[datetime] = None
|
|
is_anonymous: bool
|
|
created_at: datetime
|
|
|
|
|
|
class SurveyDetailResponse(SurveyResponse):
|
|
"""问卷详情响应"""
|
|
questions: Optional[List[QuestionResponse]] = None
|
|
|
|
|
|
class AnswerSubmit(BaseModel):
|
|
"""回答提交"""
|
|
question_id: int
|
|
answer_value: str
|
|
|
|
|
|
class ResponseSubmit(BaseModel):
|
|
"""问卷回答提交"""
|
|
department_id: Optional[int] = None
|
|
respondent_type: str = "patient"
|
|
respondent_id: Optional[int] = None
|
|
respondent_phone: Optional[str] = None
|
|
answers: List[AnswerSubmit]
|
|
|
|
|
|
class ResponseResult(BaseModel):
|
|
"""回答结果"""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
survey_id: int
|
|
department_id: Optional[int] = None
|
|
total_score: float
|
|
max_score: float
|
|
satisfaction_rate: float
|
|
submitted_at: datetime
|
|
|
|
|
|
# ==================== API Endpoints ====================
|
|
|
|
@router.get("", summary="获取问卷列表")
|
|
async def get_survey_list(
|
|
survey_type: Optional[str] = Query(None, description="问卷类型"),
|
|
status: Optional[str] = Query(None, description="状态"),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""获取问卷列表"""
|
|
type_filter = SurveyType(survey_type) if survey_type else None
|
|
status_filter = SurveyStatus(status) if status else None
|
|
|
|
surveys, total = await SurveyService.get_survey_list(
|
|
db, type_filter, status_filter, page, page_size
|
|
)
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": [
|
|
SurveyResponse.model_validate(s) for s in surveys
|
|
],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size
|
|
}
|
|
|
|
|
|
@router.get("/{survey_id}", summary="获取问卷详情")
|
|
async def get_survey_detail(
|
|
survey_id: int,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""获取问卷详情"""
|
|
survey = await SurveyService.get_survey_by_id(db, survey_id)
|
|
if not survey:
|
|
raise HTTPException(status_code=404, detail="问卷不存在")
|
|
|
|
detail = SurveyDetailResponse.model_validate(survey)
|
|
if survey.questions:
|
|
detail.questions = [QuestionResponse.model_validate(q) for q in survey.questions]
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": detail
|
|
}
|
|
|
|
|
|
@router.post("", summary="创建问卷")
|
|
async def create_survey(
|
|
survey_data: SurveyCreate,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""创建问卷"""
|
|
try:
|
|
survey_type = SurveyType(survey_data.survey_type)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="无效的问卷类型")
|
|
|
|
survey = await SurveyService.create_survey(
|
|
db,
|
|
survey_name=survey_data.survey_name,
|
|
survey_code=survey_data.survey_code,
|
|
survey_type=survey_type,
|
|
description=survey_data.description,
|
|
target_departments=survey_data.target_departments,
|
|
is_anonymous=survey_data.is_anonymous
|
|
)
|
|
|
|
# 添加题目
|
|
if survey_data.questions:
|
|
for question in survey_data.questions:
|
|
try:
|
|
question_type = QuestionType(question.question_type)
|
|
except ValueError:
|
|
continue
|
|
|
|
await SurveyService.add_question(
|
|
db,
|
|
survey_id=survey.id,
|
|
question_text=question.question_text,
|
|
question_type=question_type,
|
|
options=question.options,
|
|
score_max=question.score_max,
|
|
is_required=question.is_required
|
|
)
|
|
|
|
await db.commit()
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "创建成功",
|
|
"data": SurveyResponse.model_validate(survey)
|
|
}
|
|
|
|
|
|
@router.put("/{survey_id}", summary="更新问卷")
|
|
async def update_survey(
|
|
survey_id: int,
|
|
survey_data: SurveyUpdate,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""更新问卷"""
|
|
try:
|
|
update_dict = survey_data.model_dump(exclude_unset=True)
|
|
survey = await SurveyService.update_survey(db, survey_id, **update_dict)
|
|
if not survey:
|
|
raise HTTPException(status_code=404, detail="问卷不存在")
|
|
|
|
await db.commit()
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "更新成功",
|
|
"data": SurveyResponse.model_validate(survey)
|
|
}
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.post("/{survey_id}/publish", summary="发布问卷")
|
|
async def publish_survey(
|
|
survey_id: int,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""发布问卷"""
|
|
try:
|
|
survey = await SurveyService.publish_survey(db, survey_id)
|
|
if not survey:
|
|
raise HTTPException(status_code=404, detail="问卷不存在")
|
|
|
|
await db.commit()
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "发布成功",
|
|
"data": SurveyResponse.model_validate(survey)
|
|
}
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.post("/{survey_id}/close", summary="结束问卷")
|
|
async def close_survey(
|
|
survey_id: int,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""结束问卷"""
|
|
try:
|
|
survey = await SurveyService.close_survey(db, survey_id)
|
|
if not survey:
|
|
raise HTTPException(status_code=404, detail="问卷不存在")
|
|
|
|
await db.commit()
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "问卷已结束",
|
|
"data": SurveyResponse.model_validate(survey)
|
|
}
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.delete("/{survey_id}", summary="删除问卷")
|
|
async def delete_survey(
|
|
survey_id: int,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""删除问卷"""
|
|
try:
|
|
success = await SurveyService.delete_survey(db, survey_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="问卷不存在")
|
|
|
|
await db.commit()
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "删除成功"
|
|
}
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.post("/{survey_id}/submit", summary="提交问卷回答")
|
|
async def submit_response(
|
|
survey_id: int,
|
|
response_data: ResponseSubmit,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""提交问卷回答"""
|
|
try:
|
|
answers = [a.model_dump() for a in response_data.answers]
|
|
|
|
response = await SurveyService.submit_response(
|
|
db,
|
|
survey_id=survey_id,
|
|
department_id=response_data.department_id,
|
|
answers=answers,
|
|
respondent_type=response_data.respondent_type,
|
|
respondent_id=response_data.respondent_id,
|
|
respondent_phone=response_data.respondent_phone
|
|
)
|
|
|
|
await db.commit()
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "提交成功",
|
|
"data": ResponseResult.model_validate(response)
|
|
}
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.get("/{survey_id}/stats", summary="获取问卷统计")
|
|
async def get_survey_stats(
|
|
survey_id: int,
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""获取问卷各题目统计"""
|
|
survey = await SurveyService.get_survey_by_id(db, survey_id)
|
|
if not survey:
|
|
raise HTTPException(status_code=404, detail="问卷不存在")
|
|
|
|
stats = await SurveyService.get_question_stats(db, survey_id)
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": {
|
|
"survey_id": survey_id,
|
|
"survey_name": survey.survey_name,
|
|
"total_responses": await _get_response_count(db, survey_id),
|
|
"question_stats": stats
|
|
}
|
|
}
|
|
|
|
|
|
@router.get("/stats/department", summary="获取科室满意度统计")
|
|
async def get_department_satisfaction(
|
|
survey_id: Optional[int] = Query(None, description="问卷ID"),
|
|
department_id: Optional[int] = Query(None, description="科室ID"),
|
|
period_year: Optional[int] = Query(None, description="年度"),
|
|
period_month: Optional[int] = Query(None, description="月份"),
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""获取科室满意度统计"""
|
|
stats = await SurveyService.get_department_satisfaction(
|
|
db, survey_id, department_id, period_year, period_month
|
|
)
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": stats
|
|
}
|
|
|
|
|
|
@router.get("/stats/trend/{department_id}", summary="获取满意度趋势")
|
|
async def get_satisfaction_trend(
|
|
department_id: int,
|
|
months: int = Query(6, ge=1, le=12, description="查询月数"),
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""获取满意度趋势"""
|
|
trend = await SurveyService.get_satisfaction_trend(db, department_id, months)
|
|
|
|
return {
|
|
"code": 200,
|
|
"message": "success",
|
|
"data": trend
|
|
}
|
|
|
|
|
|
async def _get_response_count(db: AsyncSession, survey_id: int) -> int:
|
|
"""获取问卷回答数"""
|
|
from sqlalchemy import select, func
|
|
from app.models.models import SurveyResponse
|
|
|
|
result = await db.execute(
|
|
select(func.count(SurveyResponse.id))
|
|
.where(SurveyResponse.survey_id == survey_id)
|
|
)
|
|
return result.scalar() or 0
|