SharedClassManager/backend/services/conduct_service.py

# ===========================================
# 班级操行分管理系统 - 后端服务
# 
# 开发者: Canglan
# 联系方式: admin@sea-studio.top
# 版权归属: Sea Network Technology Studio
# 许可证: MIT License
# 
# 版权所有 © Sea Network Technology Studio
# ===========================================

from typing import Dict, Any, List, Optional
from datetime import datetime

from models.student import StudentModel
from models.conduct import ConductModel
from models.user import UserModel
from middleware.permission import PermissionChecker
from config import settings
from utils.logger import get_logger

logger = get_logger(__name__)


class ConductService:
    """操行分服务"""
    
    @staticmethod
    async def add_points(
        student_ids: List[int],
        points_change: int,
        reason: str,
        recorder_id: int,
        recorder_name: str
    ) -> Dict[str, Any]:
        """
        批量加减分
        """
        # 验证分值
        if points_change == 0:
            return {"success": False, "message": "分值不能为0"}
        
        # 获取操作人角色
        role = await PermissionChecker.get_user_role(recorder_id)
        
        # 权限验证
        if role == "班主任":
            # 班主任无限制
            pass
        elif role == "班长":
            # 班长限制 ±5分
            if points_change > settings.MONITOR_MAX_ADD or points_change < settings.MONITOR_MAX_SUBTRACT:
                return {"success": False, "message": f"班长单次只能加减{settings.MONITOR_MAX_ADD}分以内"}
        elif role == "劳动委员":
            # 劳动委员固定 ±1分
            if points_change not in [settings.LABOR_POINTS_ADD, settings.LABOR_POINTS_SUBTRACT]:
                return {"success": False, "message": "劳动委员只能进行±1分操作"}
        elif role in ["科代表", "考勤委员"]:
            # 科代表和考勤委员只能扣分
            if points_change > 0:
                return {"success": False, "message": "该角色只能进行扣分操作"}
        else:
            return {"success": False, "message": "无权进行此操作"}
        
        # 批量处理
        success_count = 0
        fail_count = 0
        details = []
        
        for student_id in student_ids:
            try:
                # 检查学生是否存在
                student = await StudentModel.get_by_id(student_id)
                if not student:
                    details.append({"student_id": student_id, "error": "学生不存在"})
                    fail_count += 1
                    continue
                
                # 创建记录
                record_id = await ConductModel.create_record(
                    student_id=student_id,
                    points_change=points_change,
                    reason=reason,
                    recorder_id=recorder_id,
                    recorder_name=recorder_name
                )
                
                details.append({"student_id": student_id, "success": True, "record_id": record_id})
                success_count += 1
                
                logger.info(f"用户[{recorder_id}] 对学生[{student_id}] 进行 {points_change} 分操作")
                
            except Exception as e:
                details.append({"student_id": student_id, "error": str(e)})
                fail_count += 1
        
        return {
            "success": True,
            "success_count": success_count,
            "fail_count": fail_count,
            "details": details
        }
    
    @staticmethod
    async def revoke_record(record_id: int, revoker_id: int) -> Dict[str, Any]:
        """撤销扣分记录"""
        # 检查权限
        can_revoke = await PermissionChecker.check_can_revoke(revoker_id, record_id)
        if not can_revoke:
            return {"success": False, "message": "无权撤销此记录"}
        
        # 撤销记录
        result = await ConductModel.revoke_record(record_id, revoker_id)
        
        if result:
            logger.info(f"用户[{revoker_id}] 撤销了记录[{record_id}]")
            return {"success": True, "message": "撤销成功"}
        else:
            return {"success": False, "message": "撤销失败"}
    
    @staticmethod
    async def get_history(
        user_id: int,
        student_id: Optional[int] = None,
        page: int = 1,
        page_size: int = 20,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None
    ) -> Dict[str, Any]:
        """获取历史记录"""
        role = await PermissionChecker.get_user_role(user_id)
        offset = (page - 1) * page_size
        
        # 班主任/班长可查看全班
        if role in ["班主任", "班长"]:
            records = await ConductModel.get_all_records(
                limit=page_size,
                offset=offset,
                start_date=start_date,
                end_date=end_date
            )
            
            # 获取总数
            from utils.database import execute_one
            count_sql = """
                SELECT COUNT(*) as total FROM conduct_records cr
                JOIN students s ON cr.student_id = s.student_id
                WHERE cr.is_revoked = 0
            """
            total_result = await execute_one(count_sql)
            total = total_result["total"] if total_result else 0
            
        elif student_id:
            # 管理员可查看指定学生
            
            records = await ConductModel.get_student_records(
                student_id=student_id,
                limit=page_size,
                offset=offset
            )
            total = len(await ConductModel.get_student_records(student_id, limit=10000))
        else:
            # 查看自己提交的记录
            records = await ConductModel.get_records_by_recorder(
                recorder_id=user_id,
                limit=page_size,
                offset=offset
            )
            total = len(await ConductModel.get_records_by_recorder(user_id, limit=10000))
        
        return {
            "records": records,
            "page": page,
            "page_size": page_size,
            "total": total,
            "total_pages": (total + page_size - 1) // page_size
        }