Commit 10e59f3c authored by uuo00_n's avatar uuo00_n

feat: 完善API文档和错误响应

refactor: 统一使用Path/Body参数装饰器
docs: 为所有API端点添加详细文档和错误响应示例
style: 格式化代码并修复文件结尾换行
parent f014fc61
Pipeline #60 canceled with stages
from fastapi import APIRouter, Depends, HTTPException, status, File, UploadFile
from fastapi import APIRouter, Depends, HTTPException, status, File, UploadFile, Query, Path, Body
from typing import List, Optional, Dict
from datetime import datetime
import json
import csv
import io
from app.api.deps import get_current_admin_user, require_edition_for_mode
from pydantic import BaseModel
from app.schemas.sensitive_word import (
SensitiveWordCreate, SensitiveWordResponse, SensitiveRecordResponse,
SensitiveWordBulkImport, CategoryCreate, CategoryResponse, CategoriesResponse
......@@ -17,7 +18,23 @@ from app.services.sensitive_word import (
from app.models.sensitive_word import SENSITIVE_WORD_CATEGORIES, SENSITIVE_WORD_SUBCATEGORIES
router = APIRouter(dependencies=[Depends(require_edition_for_mode())])
@router.post("/sensitive-words", response_model=dict, status_code=status.HTTP_201_CREATED)
class CreatedId(BaseModel):
id: str
class ImportedCount(BaseModel):
imported_count: int
@router.post(
"/sensitive-words",
response_model=CreatedId,
status_code=status.HTTP_201_CREATED,
summary="添加敏感词",
description="新增敏感词,需管理员权限。",
responses={
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足,需要管理员权限"}}}},
},
)
async def create_sensitive_word(
word_data: SensitiveWordCreate,
_: dict = Depends(get_current_admin_user)
......@@ -31,7 +48,17 @@ async def create_sensitive_word(
)
return {"id": word_id}
@router.post("/sensitive-words/bulk", response_model=dict, status_code=status.HTTP_201_CREATED)
@router.post(
"/sensitive-words/bulk",
response_model=ImportedCount,
status_code=status.HTTP_201_CREATED,
summary="批量添加敏感词",
description="批量新增敏感词,需管理员权限。",
responses={
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足,需要管理员权限"}}}},
},
)
async def bulk_create_sensitive_words(
words_data: SensitiveWordBulkImport,
_: dict = Depends(get_current_admin_user)
......@@ -40,7 +67,18 @@ async def bulk_create_sensitive_words(
count = await bulk_import_sensitive_words(words_data.words)
return {"imported_count": count}
@router.post("/sensitive-words/import", status_code=status.HTTP_201_CREATED)
@router.post(
"/sensitive-words/import",
status_code=status.HTTP_201_CREATED,
response_model=ImportedCount,
summary="从文件导入敏感词",
description="支持CSV或JSON导入敏感词,需管理员权限。",
responses={
400: {"description": "文件格式错误", "content": {"application/json": {"example": {"detail": "仅支持CSV和JSON格式文件"}}}},
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足,需要管理员权限"}}}},
},
)
async def import_sensitive_words_from_file(
file: UploadFile = File(...),
_: dict = Depends(get_current_admin_user)
......@@ -85,9 +123,19 @@ async def import_sensitive_words_from_file(
count = await bulk_import_sensitive_words(words)
return {"imported_count": count}
@router.delete("/sensitive-words/{word_id}", status_code=status.HTTP_204_NO_CONTENT)
@router.delete(
"/sensitive-words/{word_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="删除敏感词",
description="根据ID删除敏感词,需管理员权限。",
responses={
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足,需要管理员权限"}}}},
404: {"description": "敏感词不存在", "content": {"application/json": {"example": {"detail": "敏感词不存在"}}}},
},
)
async def remove_sensitive_word(
word_id: str,
word_id: str = Path(..., description="敏感词ID"),
_: dict = Depends(get_current_admin_user)
):
"""删除敏感词(仅管理员)"""
......@@ -99,26 +147,36 @@ async def remove_sensitive_word(
)
return None
@router.get("/sensitive-words", response_model=List[SensitiveWordResponse])
@router.get(
"/sensitive-words",
response_model=List[SensitiveWordResponse],
summary="查询敏感词",
description="按条件查询敏感词列表,需管理员权限。",
)
async def list_sensitive_words(
category: Optional[str] = None,
subcategory: Optional[str] = None,
min_severity: Optional[int] = None,
max_severity: Optional[int] = None,
category: Optional[str] = Query(None, description="主分类"),
subcategory: Optional[str] = Query(None, description="子分类"),
min_severity: Optional[int] = Query(None, description="最小严重程度"),
max_severity: Optional[int] = Query(None, description="最大严重程度"),
_: dict = Depends(get_current_admin_user)
):
return await get_all_sensitive_words(category, subcategory, min_severity, max_severity)
@router.get("/sensitive-records", response_model=List[SensitiveRecordResponse])
@router.get(
"/sensitive-records",
response_model=List[SensitiveRecordResponse],
summary="查询敏感词记录",
description="按条件筛选敏感词命中记录,需管理员权限。",
)
async def list_sensitive_records(
user_id: Optional[str] = None,
conversation_id: Optional[str] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
category: Optional[str] = None,
subcategory: Optional[str] = None,
min_severity: Optional[int] = None,
max_severity: Optional[int] = None,
user_id: Optional[str] = Query(None, description="用户ID"),
conversation_id: Optional[str] = Query(None, description="对话ID"),
start_date: Optional[datetime] = Query(None, description="开始日期"),
end_date: Optional[datetime] = Query(None, description="结束日期"),
category: Optional[str] = Query(None, description="主分类"),
subcategory: Optional[str] = Query(None, description="子分类"),
min_severity: Optional[int] = Query(None, description="最小严重程度"),
max_severity: Optional[int] = Query(None, description="最大严重程度"),
_: dict = Depends(get_current_admin_user)
):
"""获取敏感词记录(仅管理员)
......@@ -138,21 +196,40 @@ async def list_sensitive_records(
category, subcategory, min_severity, max_severity
)
@router.get("/categories", response_model=CategoriesResponse)
@router.get(
"/categories",
response_model=CategoriesResponse,
summary="获取分类配置",
description="获取当前所有敏感词分类配置,需管理员权限。",
)
async def list_categories(
_: dict = Depends(get_current_admin_user)
):
"""获取所有敏感词分类(仅管理员)"""
return {"categories": await get_categories()}
@router.get("/categories/default", response_model=CategoriesResponse)
@router.get(
"/categories/default",
response_model=CategoriesResponse,
summary="获取默认分类",
description="获取默认的敏感词主分类及子分类配置,需管理员权限。",
)
async def get_default_categories(
_: dict = Depends(get_current_admin_user)
):
"""获取默认敏感词分类(仅管理员)"""
return {"categories": {cat: SENSITIVE_WORD_SUBCATEGORIES.get(cat, []) for cat in SENSITIVE_WORD_CATEGORIES}}
@router.post("/categories", response_model=CategoryResponse, status_code=status.HTTP_201_CREATED)
@router.post(
"/categories",
response_model=CategoryResponse,
status_code=status.HTTP_201_CREATED,
summary="新增分类",
description="新增敏感词主分类及子分类配置,需管理员权限。",
responses={
400: {"description": "分类已存在", "content": {"application/json": {"example": {"detail": "分类已存在"}}}},
},
)
async def create_category(
category_data: CategoryCreate,
_: dict = Depends(get_current_admin_user)
......@@ -166,10 +243,18 @@ async def create_category(
)
return {"name": category_data.name, "subcategories": category_data.subcategories}
@router.put("/categories/{category_name}", response_model=CategoryResponse)
@router.put(
"/categories/{category_name}",
response_model=CategoryResponse,
summary="更新分类子项",
description="更新指定主分类的子分类列表,需管理员权限。",
responses={
404: {"description": "分类不存在", "content": {"application/json": {"example": {"detail": "分类不存在"}}}},
},
)
async def update_category_subcategories(
category_name: str,
subcategories: List[str],
category_name: str = Path(..., description="主分类名称"),
subcategories: List[str] = Body(..., description="子分类列表"),
_: dict = Depends(get_current_admin_user)
):
"""更新敏感词分类的子分类(仅管理员)"""
......@@ -181,9 +266,17 @@ async def update_category_subcategories(
)
return {"name": category_name, "subcategories": subcategories}
@router.delete("/categories/{category_name}", status_code=status.HTTP_204_NO_CONTENT)
@router.delete(
"/categories/{category_name}",
status_code=status.HTTP_204_NO_CONTENT,
summary="删除分类",
description="删除指定主分类(默认分类不可删除),需管理员权限。",
responses={
404: {"description": "分类不存在或不可删除", "content": {"application/json": {"example": {"detail": "分类不存在或无法删除默认分类"}}}},
},
)
async def remove_category(
category_name: str,
category_name: str = Path(..., description="主分类名称"),
_: dict = Depends(get_current_admin_user)
):
"""删除敏感词分类(仅管理员)"""
......@@ -193,4 +286,4 @@ async def remove_category(
status_code=status.HTTP_404_NOT_FOUND,
detail="分类不存在或无法删除默认分类"
)
return None
\ No newline at end of file
return None
......@@ -12,8 +12,12 @@ router = APIRouter()
@router.post(
"/register",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
summary="用户注册",
description="创建基础账户(默认教育版、角色等级为1)。",
responses={
400: {"description": "用户名或邮箱已存在", "content": {"application/json": {"example": {"detail": "用户名已存在"}}}},
},
)
async def register(user_data: UserCreate):
"""注册新用户"""
......@@ -63,6 +67,9 @@ async def register(user_data: UserCreate):
response_model=Token,
summary="用户登录",
description="使用表单登录(username/password),返回令牌与用户/绑定信息。",
responses={
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "用户名或密码错误"}}}},
},
)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
"""用户登录"""
......@@ -102,4 +109,4 @@ async def login(form_data: OAuth2PasswordRequestForm = Depends()):
"person_type": b.get("type"),
"bound_primary": True,
} if b else {})
}
\ No newline at end of file
}
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Path
from pydantic import BaseModel
from typing import Optional
from app.api.deps import require_edition_for_mode, get_current_active_user, require_role
......@@ -26,6 +26,10 @@ class BindingOut(BaseModel):
summary="创建主绑定",
description="为当前账号创建人物主绑定(student/teacher),同类型主绑定唯一。",
response_model=ActionResult,
responses={
400: {"description": "主绑定已存在或参数错误", "content": {"application/json": {"example": {"detail": "主绑定已存在或参数错误"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def bind(payload: BindingPayload, current_user: dict = Depends(require_role(2))) -> ActionResult:
ok = await create_binding(str(current_user["_id"]), payload.person_id, payload.type, payload.primary)
......@@ -38,8 +42,12 @@ async def bind(payload: BindingPayload, current_user: dict = Depends(require_rol
summary="删除绑定",
description="删除当前账号与指定人物的绑定。",
response_model=ActionResult,
responses={
404: {"description": "未找到绑定", "content": {"application/json": {"example": {"detail": "未找到绑定"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def unbind(person_id: str, current_user: dict = Depends(require_role(2))) -> ActionResult:
async def unbind(person_id: str = Path(..., description="人物ID"), current_user: dict = Depends(require_role(2))) -> ActionResult:
ok = await delete_binding(str(current_user["_id"]), person_id)
if not ok:
raise HTTPException(status_code=404, detail="未找到绑定")
......@@ -50,6 +58,10 @@ async def unbind(person_id: str, current_user: dict = Depends(require_role(2)))
summary="查询当前账号的主绑定",
description="返回当前账号的主绑定信息(account_id/person_id/type/primary)。",
response_model=BindingOut,
responses={
404: {"description": "未绑定人物", "content": {"application/json": {"example": {"detail": "未绑定人物"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def me(current_user: dict = Depends(get_current_active_user)) -> BindingOut:
b = await get_binding_by_account(str(current_user["_id"]))
......@@ -58,4 +70,4 @@ async def me(current_user: dict = Depends(get_current_active_user)) -> BindingOu
b["_id"] = str(b["_id"])
b["account_id"] = str(b["account_id"])
b["person_id"] = str(b["person_id"])
return b
\ No newline at end of file
return b
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Path, Body
from pydantic import BaseModel
from typing import List, Optional
from bson import ObjectId
......@@ -24,8 +24,13 @@ class ClassOut(BaseModel):
summary="设置班主任人物",
description="为班级设置 head_teacher_person_id(指向教师人物)。",
response_model=ActionResult,
responses={
404: {"description": "班级不存在", "content": {"application/json": {"example": {"detail": "班级不存在"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足"}}}},
},
)
async def set_head_teacher(class_id: str, payload: HeadTeacherPayload, current_user: dict = Depends(require_role(3))) -> ActionResult:
async def set_head_teacher(class_id: str = Path(..., description="班级ID"), payload: HeadTeacherPayload = Body(..., description="班主任人物设置"), current_user: dict = Depends(require_role(3))) -> ActionResult:
res = await db.db.classes.update_one({"class_id": class_id}, {"$set": {"head_teacher_person_id": ObjectId(payload.head_teacher_person_id)}})
if res.matched_count == 0:
raise HTTPException(status_code=404, detail="班级不存在")
......@@ -45,4 +50,4 @@ async def list_classes(current_user: dict = Depends(require_role(2))) -> List[Cl
if d.get("head_teacher_person_id"):
d["head_teacher_person_id"] = str(d["head_teacher_person_id"])
res.append(d)
return res
\ No newline at end of file
return res
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, status, Path, Body
from typing import List
from app.api.deps import get_current_active_user, require_edition_for_mode
from app.schemas.conversation import (
......@@ -20,31 +20,47 @@ from app.services.conversation import (
# 在路由层挂载版别运行模式依赖,限制仅允许当前模式的用户访问
router = APIRouter(dependencies=[Depends(require_edition_for_mode())])
@router.post("/", status_code=status.HTTP_201_CREATED, response_model=CreatedId)
@router.post(
"/",
status_code=status.HTTP_201_CREATED,
response_model=CreatedId,
summary="创建对话",
description="为当前用户新建会话,返回新建的会话ID。",
responses={
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def create_new_conversation(current_user: dict = Depends(get_current_active_user)):
"""创建新对话(返回新建 ID)
用途:为当前用户新建会话,并返回新会话的 ID。
依赖:鉴权用户、版别运行模式。
"""
conversation_id = await create_conversation(str(current_user["_id"]))
return {"id": conversation_id}
@router.get("/", response_model=List[ConversationResponse])
@router.get(
"/",
response_model=List[ConversationResponse],
summary="获取对话列表",
description="返回当前用户的对话列表(包含最近一条消息等精简信息)。",
responses={
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def list_conversations(current_user: dict = Depends(get_current_active_user)):
"""获取用户的所有对话(列表优化)
用途:返回当前用户的对话列表,统一字段并降低负载(仅最近一条消息)。
"""
conversations = await get_user_conversations(str(current_user["_id"]))
return conversations
@router.get("/{conversation_id}", response_model=ConversationDocOut)
@router.get(
"/{conversation_id}",
response_model=ConversationDocOut,
summary="获取对话详情",
description="返回指定对话的完整消息列表。",
responses={
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
404: {"description": "对话不存在", "content": {"application/json": {"example": {"detail": "对话不存在"}}}},
},
)
async def get_single_conversation(
conversation_id: str,
conversation_id: str = Path(..., description="对话ID"),
current_user: dict = Depends(get_current_active_user)
):
"""获取单个对话详情
用途:返回指定对话的完整消息列表,统一 ID 与标题字段。
"""
conversation = await get_conversation(conversation_id, str(current_user["_id"]))
if not conversation:
raise HTTPException(
......@@ -53,29 +69,40 @@ async def get_single_conversation(
)
return conversation
@router.delete("/{conversation_id}", response_model=DeleteResult)
@router.delete(
"/{conversation_id}",
response_model=DeleteResult,
summary="删除对话",
description="删除当前用户归属的指定对话。",
responses={
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
404: {"description": "对话不存在或无权限", "content": {"application/json": {"example": {"detail": "对话不存在或无权限"}}}},
},
)
async def remove_conversation(
conversation_id: str,
conversation_id: str = Path(..., description="对话ID"),
current_user: dict = Depends(get_current_active_user)
):
"""删除对话
用途:仅允许删除当前用户归属的对话,并清理关联敏感记录。
返回:删除结果(deleted: bool)
"""
ok = await delete_conversation(conversation_id, str(current_user["_id"]))
if not ok:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="对话不存在或无权限")
return {"deleted": True, "message": "删除成功"}
@router.post("/{conversation_id}/messages", response_model=MessageSendResult)
@router.post(
"/{conversation_id}/messages",
response_model=MessageSendResult,
summary="发送消息并获取回复",
description="向指定对话发送消息,返回助手回复或敏感词拒绝信息。",
responses={
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
404: {"description": "对话不存在", "content": {"application/json": {"example": {"detail": "对话不存在"}}}},
},
)
async def send_message(
conversation_id: str,
message: MessageCreate,
conversation_id: str = Path(..., description="对话ID"),
message: MessageCreate = Body(..., description="消息内容"),
current_user: dict = Depends(get_current_active_user)
):
"""发送消息并获取回复
用途:在指定对话中发送用户消息;若命中敏感词返回拒绝回复并记录;否则返回模型生成的助手回复。
"""
# 检查对话是否存在
conversation = await get_conversation(conversation_id, str(current_user["_id"]))
if not conversation:
......@@ -87,4 +114,4 @@ async def send_message(
# 添加消息并获取回复
result = await add_message(conversation_id, str(current_user["_id"]), message.content)
return result
\ No newline at end of file
return result
......@@ -89,7 +89,11 @@ class CampusOverview(BaseModel):
"/student/today",
summary="学生端:今日个人课表、出勤与操行",
description="需角色等级≥1且主绑定为学生;返回当天课表(共享节次按班级定位教室)、今日出勤记录与操行评语。"
, response_model=StudentTodaySummary
, response_model=StudentTodaySummary,
responses={
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足或未绑定学生", "content": {"application/json": {"example": {"detail": "实体绑定不存在或类型不匹配"}}}},
}
)
async def student_today(current_user: dict = Depends(require_role(1)), _b: dict = Depends(require_binding("student"))) -> StudentTodaySummary:
return await student_today_summary(current_user)
......@@ -98,7 +102,11 @@ async def student_today(current_user: dict = Depends(require_role(1)), _b: dict
"/homeroom/current",
summary="班主任端:当前节次课程与地点、出勤率、请假、指示",
description="需角色等级≥2且主绑定为教师;按 head_teacher_person_id 定位所辖班,统计当前节次的课程与地点、节次出勤率、今日请假与部门指示。"
, response_model=HomeroomCurrentSummary
, response_model=HomeroomCurrentSummary,
responses={
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足或未绑定教师", "content": {"application/json": {"example": {"detail": "实体绑定不存在或类型不匹配"}}}},
}
)
async def homeroom_current(current_user: dict = Depends(require_role(2)), _b: dict = Depends(require_binding("teacher"))) -> HomeroomCurrentSummary:
return await homeroom_current_summary(current_user)
......@@ -107,7 +115,11 @@ async def homeroom_current(current_user: dict = Depends(require_role(2)), _b: di
"/department/overview",
summary="中层端:教师节次出勤率、学生出勤、异常班级、指示",
description="需角色等级≥3;按今日工作日统计每位教师的节次出勤率、全校学生出勤聚合、异常班级(缺勤+请假占比>0.3)与近期部门/校园指示。"
, response_model=DepartmentOverview
, response_model=DepartmentOverview,
responses={
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足"}}}},
}
)
async def department(current_user: dict = Depends(require_role(3))) -> DepartmentOverview:
return await department_overview(current_user)
......@@ -116,7 +128,11 @@ async def department(current_user: dict = Depends(require_role(3))) -> Departmen
"/campus/overview",
summary="校级端:校园整体总览",
description="需角色等级≥4;返回学生总数、今日出勤、请假与指示数量等宏观数据。"
, response_model=CampusOverview
, response_model=CampusOverview,
responses={
401: {"description": "未认证", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足"}}}},
}
)
async def campus(current_user: dict = Depends(require_role(4))) -> CampusOverview:
return await campus_overview(current_user)
\ No newline at end of file
return await campus_overview(current_user)
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Body
from pydantic import BaseModel, Field
from typing import List, Dict
from bson import ObjectId
......@@ -30,8 +30,13 @@ class PersonsListResponse(BaseModel):
summary="批量导入人物档案",
description="导入学生/教师/职员的基础人物信息(person_id/name/type)。",
response_model=BulkInsertResult,
responses={
400: {"description": "空数据", "content": {"application/json": {"example": {"detail": "空数据"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足"}}}},
},
)
async def bulk_create(persons: List[PersonCreate], current_user: dict = Depends(require_role(3))) -> BulkInsertResult:
async def bulk_create(persons: List[PersonCreate] = Body(..., description="人物档案列表"), current_user: dict = Depends(require_role(3))) -> BulkInsertResult:
docs = [{"person_id": p.person_id, "name": p.name, "type": p.type} for p in persons]
if not docs:
raise HTTPException(status_code=400, detail="空数据")
......@@ -54,4 +59,4 @@ async def list_persons(current_user: dict = Depends(require_role(3))) -> Persons
t = d.get("type")
if t:
counts[t] = counts.get(t, 0) + 1
return {"items": res, "counts_by_type": counts}
\ No newline at end of file
return {"items": res, "counts_by_type": counts}
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Body
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
from bson import ObjectId
......@@ -27,8 +27,13 @@ class ScheduleOut(BaseModel):
summary="为节次设置任课教师人物",
description="将指定 lesson_id 的节次设置为 teacher_person_id(教师人物)。",
response_model=ActionResult,
responses={
404: {"description": "节次不存在", "content": {"application/json": {"example": {"detail": "节次不存在"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足"}}}},
},
)
async def assign_teacher(payload: AssignTeacherPayload, current_user: dict = Depends(require_role(3))) -> ActionResult:
async def assign_teacher(payload: AssignTeacherPayload = Body(..., description="节次与教师人物"), current_user: dict = Depends(require_role(3))) -> ActionResult:
res = await db.db.schedules.update_one({"lesson_id": payload.lesson_id}, {"$set": {"teacher_person_id": ObjectId(payload.teacher_person_id)}})
if res.matched_count == 0:
raise HTTPException(status_code=404, detail="节次不存在")
......@@ -54,4 +59,4 @@ async def list_schedules(current_user: dict = Depends(require_role(2))) -> List[
"teacher_person_id": d.get("teacher_person_id"),
"classes": d.get("classes", []),
})
return res
\ No newline at end of file
return res
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Path, Body
from pydantic import BaseModel
from typing import Optional, Dict, Any
from app.api.deps import require_edition_for_mode, require_role
......@@ -18,24 +18,51 @@ class StudentOut(BaseModel):
name: Optional[str] = None
class_id: Optional[str] = None
@router.post("/{student_id}/bind", response_model=ActionResult)
async def bind(student_id: str, payload: BindPayload, current_user: dict = Depends(require_role(2))) -> ActionResult:
@router.post(
"/{student_id}/bind",
response_model=ActionResult,
summary="绑定学生",
description="为指定学生建立主绑定(需角色等级≥2)。",
responses={
400: {"description": "绑定失败或学生不存在", "content": {"application/json": {"example": {"detail": "用户已绑定其他学生或学生不存在"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def bind(student_id: str = Path(..., description="学生ID"), payload: BindPayload = Body(..., description="绑定参数"), current_user: dict = Depends(require_role(2))) -> ActionResult:
ok = await bind_user_to_student(payload.user_id, student_id)
if not ok:
raise HTTPException(status_code=400, detail="用户已绑定其他学生或学生不存在")
return {"success": True}
@router.delete("/{student_id}/bind", response_model=ActionResult)
async def unbind(student_id: str, current_user: dict = Depends(require_role(2))) -> ActionResult:
@router.delete(
"/{student_id}/bind",
response_model=ActionResult,
summary="解除学生绑定",
description="解除当前用户与指定学生的主绑定(需角色等级≥2)。",
responses={
404: {"description": "未找到或未绑定", "content": {"application/json": {"example": {"detail": "学生不存在或未绑定"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def unbind(student_id: str = Path(..., description="学生ID"), current_user: dict = Depends(require_role(2))) -> ActionResult:
ok = await unbind_user_from_student(student_id)
if not ok:
raise HTTPException(status_code=404, detail="学生不存在或未绑定")
return {"success": True}
@router.get("/me", response_model=StudentOut)
@router.get(
"/me",
response_model=StudentOut,
summary="查询当前绑定学生",
description="返回当前用户绑定的学生信息(需角色等级≥1)。",
responses={
404: {"description": "未绑定学生", "content": {"application/json": {"example": {"detail": "当前用户未绑定学生"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
},
)
async def me(current_user: dict = Depends(require_role(1))) -> StudentOut:
s = await get_student_by_user(str(current_user["_id"]))
if not s:
raise HTTPException(status_code=404, detail="当前用户未绑定学生")
s["_id"] = str(s["_id"]) # 简化返回
return s
\ No newline at end of file
return s
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, Body
from pydantic import BaseModel
from typing import List, Optional
from bson import ObjectId
......@@ -30,8 +30,13 @@ class TeacherOut(BaseModel):
summary="批量导入教师实体",
description="导入教师实体(person_id/teacher_id/department/roles),可选绑定 account_id。",
response_model=BulkInsertResult,
responses={
400: {"description": "空数据", "content": {"application/json": {"example": {"detail": "空数据"}}}},
401: {"description": "认证失败", "content": {"application/json": {"example": {"detail": "无效的认证凭据"}}}},
403: {"description": "权限不足", "content": {"application/json": {"example": {"detail": "权限不足"}}}},
},
)
async def bulk_create(teachers: List[TeacherCreate], current_user: dict = Depends(require_role(3))) -> BulkInsertResult:
async def bulk_create(teachers: List[TeacherCreate] = Body(..., description="教师实体列表"), current_user: dict = Depends(require_role(3))) -> BulkInsertResult:
docs = []
for t in teachers:
doc = {
......@@ -63,4 +68,4 @@ async def list_teachers(current_user: dict = Depends(require_role(3))) -> List[T
if d.get("account_id"):
d["account_id"] = str(d["account_id"])
res.append(d)
return res
\ No newline at end of file
return res
......@@ -8,6 +8,7 @@ from app.utils.sensitive_word_filter import sensitive_word_filter
app = FastAPI(
title=settings.APP_NAME,
version="1.0.0",
description=(
"LLM 过滤系统后端接口\n\n"
"角色等级:1 学生、2 班主任、3 中层、4 校级、5 管理员。\n"
......@@ -28,7 +29,16 @@ app = FastAPI(
],
openapi_url=f"{settings.API_V1_STR}/openapi.json",
docs_url=None, # 禁用默认的/docs
redoc_url=None # 禁用默认的/redoc
redoc_url=None, # 禁用默认的/redoc
contact={
"name": "LLM Filter Team",
"url": settings.APP_BASE_URL,
"email": "huangjunbo1107@outlook.com",
},
license_info={
"name": "MIT",
"url": "https://opensource.org/licenses/MIT",
}
)
origins_cfg = settings.CORS_ALLOWED_ORIGINS
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment