255 lines
9.5 KiB
Python
255 lines
9.5 KiB
Python
# ===========================================
|
|
# 班级操行分管理系统 - 学期管理路由
|
|
#
|
|
# 开发者: Canglan
|
|
# 联系方式: admin@sea-studio.top
|
|
# 版权归属: Sea Network Technology Studio
|
|
# 许可证: MIT License
|
|
#
|
|
# 版权所有 © Sea Network Technology Studio
|
|
# ===========================================
|
|
|
|
from fastapi import APIRouter, Request, Query
|
|
from typing import Optional
|
|
|
|
from middleware.permission import (
|
|
get_current_user,
|
|
PermissionChecker
|
|
)
|
|
from services.semester_service import SemesterService
|
|
from services.log_service import LogService
|
|
from schemas.semester import CreateSemesterRequest, UpdateSemesterRequest
|
|
from utils.response import success_response, error_response
|
|
from utils.logger import get_logger
|
|
|
|
router = APIRouter()
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
@router.get("/list")
|
|
async def list_semesters(request: Request):
|
|
"""获取学期列表（仅班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可查看学期列表", code=403)
|
|
result = await SemesterService.list_semesters()
|
|
if result["success"]:
|
|
return success_response(data=result["semesters"])
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.get("/active")
|
|
async def get_active_semester(request: Request):
|
|
"""获取当前活跃学期（含当前周数）"""
|
|
user = await get_current_user(request)
|
|
result = await SemesterService.get_active_semester()
|
|
if result["success"]:
|
|
semester = result.get("semester")
|
|
if semester and semester.get('start_date'):
|
|
from datetime import date, datetime
|
|
try:
|
|
start = semester['start_date']
|
|
if isinstance(start, str):
|
|
start_date = datetime.strptime(start, '%Y-%m-%d').date()
|
|
else:
|
|
start_date = start
|
|
today = date.today()
|
|
delta = (today - start_date).days
|
|
if delta >= 0:
|
|
semester['current_week'] = delta // 7 + 1
|
|
else:
|
|
semester['current_week'] = 0
|
|
except Exception:
|
|
semester['current_week'] = None
|
|
return success_response(data=semester)
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.post("/create")
|
|
async def create_semester(request: Request, req: CreateSemesterRequest):
|
|
"""创建学期（班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可创建学期", code=403)
|
|
|
|
result = await SemesterService.create_semester(
|
|
semester_name=req.semester_name,
|
|
start_date=req.start_date,
|
|
end_date=req.end_date,
|
|
operator_id=user["user_id"]
|
|
)
|
|
if result["success"]:
|
|
await LogService.write_operation_log(
|
|
operator_id=user["user_id"], operator_name=user["real_name"],
|
|
operator_role="班主任", operation_type="create_semester",
|
|
target_type="semester", target_id=result.get("semester_id"),
|
|
details=f"创建学期: {req.semester_name}",
|
|
ip=request.client.host
|
|
)
|
|
return success_response(data=result, message="学期创建成功")
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.put("/activate/{semester_id}")
|
|
async def activate_semester(request: Request, semester_id: int):
|
|
"""设为当前学期（班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可设置当前学期", code=403)
|
|
|
|
result = await SemesterService.activate_semester(
|
|
semester_id=semester_id,
|
|
operator_id=user["user_id"]
|
|
)
|
|
if result["success"]:
|
|
await LogService.write_operation_log(
|
|
operator_id=user["user_id"], operator_name=user["real_name"],
|
|
operator_role="班主任", operation_type="activate_semester",
|
|
target_type="semester", target_id=semester_id,
|
|
details=f"激活学期ID: {semester_id}",
|
|
ip=request.client.host
|
|
)
|
|
return success_response(message=result["message"])
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.put("/update/{semester_id}")
|
|
async def update_semester(request: Request, semester_id: int, req: UpdateSemesterRequest):
|
|
"""编辑学期（班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可编辑学期", code=403)
|
|
|
|
result = await SemesterService.update_semester(
|
|
semester_id=semester_id,
|
|
semester_name=req.semester_name,
|
|
start_date=req.start_date,
|
|
end_date=req.end_date,
|
|
operator_id=user["user_id"]
|
|
)
|
|
if result["success"]:
|
|
await LogService.write_operation_log(
|
|
operator_id=user["user_id"], operator_name=user["real_name"],
|
|
operator_role="班主任", operation_type="update_semester",
|
|
target_type="semester", target_id=semester_id,
|
|
details=f"编辑学期ID: {semester_id}",
|
|
ip=request.client.host
|
|
)
|
|
return success_response(message=result["message"])
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.delete("/delete/{semester_id}")
|
|
async def delete_semester(request: Request, semester_id: int):
|
|
"""删除学期（班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可删除学期", code=403)
|
|
|
|
result = await SemesterService.delete_semester(
|
|
semester_id=semester_id,
|
|
operator_id=user["user_id"]
|
|
)
|
|
if result["success"]:
|
|
await LogService.write_operation_log(
|
|
operator_id=user["user_id"], operator_name=user["real_name"],
|
|
operator_role="班主任", operation_type="delete_semester",
|
|
target_type="semester", target_id=semester_id,
|
|
details=f"删除学期ID: {semester_id}",
|
|
ip=request.client.host
|
|
)
|
|
return success_response(message=result["message"])
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.post("/{semester_id}/associate")
|
|
async def associate_records(request: Request, semester_id: int):
|
|
"""关联记录到学期（班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可关联数据", code=403)
|
|
|
|
result = await SemesterService.associate_records(
|
|
semester_id=semester_id,
|
|
operator_id=user["user_id"]
|
|
)
|
|
if result["success"]:
|
|
await LogService.write_operation_log(
|
|
operator_id=user["user_id"], operator_name=user["real_name"],
|
|
operator_role="班主任", operation_type="associate_records",
|
|
target_type="semester", target_id=semester_id,
|
|
details=f"关联数据到学期ID: {semester_id}, 结果: {result.get('data', {})}",
|
|
ip=request.client.host
|
|
)
|
|
return success_response(data=result.get("data"), message=result["message"])
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.post("/archive/{semester_id}")
|
|
async def archive_semester(
|
|
request: Request,
|
|
semester_id: int,
|
|
reset_scores: bool = Query(False)
|
|
):
|
|
"""归档学期（班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可归档学期", code=403)
|
|
|
|
result = await SemesterService.archive_semester(
|
|
semester_id=semester_id,
|
|
operator_id=user["user_id"],
|
|
reset_scores=reset_scores
|
|
)
|
|
if result["success"]:
|
|
log_detail = f"归档学期ID: {semester_id}"
|
|
if reset_scores:
|
|
log_detail += " 并重置学生操行分"
|
|
await LogService.write_operation_log(
|
|
operator_id=user["user_id"], operator_name=user["real_name"],
|
|
operator_role="班主任", operation_type="archive_semester",
|
|
target_type="semester", target_id=semester_id,
|
|
details=log_detail,
|
|
ip=request.client.host
|
|
)
|
|
return success_response(message=result["message"])
|
|
else:
|
|
return error_response(message=result["message"])
|
|
|
|
|
|
@router.get("/archive/{semester_id}/records")
|
|
async def get_archive_records(
|
|
request: Request,
|
|
semester_id: int,
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200)
|
|
):
|
|
"""查看归档数据（仅班主任）"""
|
|
user = await get_current_user(request)
|
|
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
|
|
if not is_teacher:
|
|
return error_response(message="仅班主任可查看归档数据", code=403)
|
|
result = await SemesterService.get_archive_records(
|
|
semester_id=semester_id,
|
|
page=page,
|
|
page_size=page_size
|
|
)
|
|
if result["success"]:
|
|
return success_response(data=result["data"])
|
|
else:
|
|
return error_response(message=result["message"])
|