canglan/ClassManager
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.2测试

Browse Source
This commit is contained in:
canglan
2026-04-10 14:18:07 +08:00
parent 6102774585
commit 9d89e62b63
19 changed files with 461 additions and 995 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
backend/.env.example
View File

@@ -1,164 +1,111 @@
# ===========================================
# 班级操行分管理系统 - 后端服务
#
# 开发者: Canglan
# 联系方式: admin@sea-studio.top
# 版权归属: Sea Network Technology Studio
# 许可证: MIT License
#
# 版权所有 © Sea Network Technology Studio
# ===========================================
# ===========================================
# FastAPI 应用配置
# ===========================================
# 应用名称 - 显示在API文档和日志中
# 应用名称
APP_NAME=班级操行分管理系统
# 运行环境 - production(生产) / development(开发) / testing(测试)
# 生产环境会自动启用HTTPS重定向
# 运行环境 - production / development / testing
APP_ENV=production
# 调试模式 - true开启详细错误信息，生产环境必须为false
DEBUG=False
# 应用密钥 - 用于会话加密，必须32位以上随机字符串
# 生成方法: openssl rand -hex 32
# 应用密钥 - 必须32位以上随机字符串
SECRET_KEY=your-super-secret-key-min-32-characters-long
# API版本号 - 用于路由前缀，如 /api/v1
# API版本号
API_VERSION=v1
# ===========================================
# MySQL 数据库配置
# ===========================================
# 数据库主机地址 - 本地用127.0.0.1，远程用实际IP
DB_HOST=127.0.0.1
# 数据库端口 - MySQL默认3306
DB_PORT=3306
# 数据库用户名 - 建议创建专用账户，不要用root
DB_USER=class_admin
# 数据库密码 - 使用强密码，包含大小写数字特殊字符
DB_PASSWORD=your-strong-db-password
# 数据库名称 - 固定为 classmanagerdb
DB_NAME=classmanagerdb
# 连接池大小 - 同时保持的数据库连接数，根据并发量调整
DB_POOL_SIZE=10
# 最大溢出连接 - 连接池满后最多额外创建的连接数
DB_MAX_OVERFLOW=20
# ===========================================
# Redis 缓存配置
# ===========================================
# Redis主机地址
REDIS_HOST=127.0.0.1
# Redis端口 - 默认6379
REDIS_PORT=6379
# Redis密码 - 建议设置，防止未授权访问
REDIS_PASSWORD=your-redis-password
# Redis数据库编号 - 0-15，建议使用独立数据库避免冲突
REDIS_DB=0
# 最大连接数 - 根据并发量调整，建议50-200
REDIS_MAX_CONNECTIONS=50
# ===========================================
# JWT 认证配置
# ===========================================
# JWT密钥 - 用于签名Token，必须32位以上随机字符串
# 生成方法: openssl rand -hex 32
JWT_SECRET_KEY=your-jwt-secret-key-min-32-chars
# JWT签名算法 - HS256对称加密
JWT_ALGORITHM=HS256
# Token过期时间（分钟）- 30分钟无操作需重新登录
JWT_EXPIRE_MINUTES=30
# ===========================================
# 密码加密配置
# ===========================================
# 密码盐值 - 固定字符串，用于SHA1+MD5双重加密
# 生成后不可更改，否则所有密码失效
# 生成方法: openssl rand -hex 16
PASSWORD_SALT=your-fixed-salt-string-for-password-hash
# ===========================================
# 调试入口配置
# ===========================================
# 调试入口路径 - 随机字符串，用于添加第一批管理员
# 建议使用32位随机字符串，只有开发人员知道此路径
# 添加完管理员后建议注释掉debug路由
# 生成方法: openssl rand -hex 16 | sed 's/\(..\)/\1/g' | cut -c1-32
DEBUG_PATH=/a7k9x2m4q8w1e3r5t6y7u8i9o0p1z2x3
# ===========================================
# 扣分规则配置
# ===========================================
# 注意：这些规则仅用于建议，实际操作时可选择是否应用扣分
# 作业未提交扣分 - 科代表标记未提交时建议扣分数值
DEDUCTION_HOMEWORK_NOT_SUBMIT=2
# 作业迟交扣分 - 迟交作业建议扣分数值
DEDUCTION_HOMEWORK_LATE=1
# 缺勤扣分 - 缺勤建议扣分数值
DEDUCTION_ATTENDANCE_ABSENT=5
# 迟到扣分 - 迟到建议扣分数值
DEDUCTION_ATTENDANCE_LATE=2
# 请假扣分 - 请假建议扣分数值
DEDUCTION_ATTENDANCE_LEAVE=1
# ===========================================
# 劳动委员固定分值配置
# ===========================================
# 劳动委员加分值 - 固定为1分
LABOR_POINTS_ADD=1
# 劳动委员扣分值 - 固定为-1分
LABOR_POINTS_SUBTRACT=-1
# ===========================================
# 班长加减分限制配置
# ===========================================
# 班长最大单次加分值
MONITOR_MAX_ADD=5
# 班长最大单次扣分值（负数）
MONITOR_MAX_SUBTRACT=-5
# ===========================================
# 日志配置
# ===========================================
# 日志级别 - DEBUG/INFO/WARNING/ERROR
# 生产环境建议INFO，开发环境可用DEBUG
LOG_LEVEL=INFO
# 单日志文件最大大小（字节）- 100MB = 104857600
LOG_MAX_BYTES=104857600
# 日志备份数量 - 保留30个历史日志文件
LOG_BACKUP_COUNT=30
# 日志保留天数 - 操作日志保留1年，访问日志保留90天
LOG_RETENTION_DAYS=365
# ===========================================
# CORS 跨域配置
# ===========================================
# 允许的跨域域名 - 多个域名用英文逗号分隔，不要有空格
# 生产环境必须指定具体域名，不能用 *
CORS_ORIGINS=https://your-frontend-domain.com,http://localhost:8080
# 允许的跨域域名 - 多个域名用英文逗号分隔
# 示例: https://example.com,https://api.example.com
CORS_ORIGINS=https://your-frontend-domain.com,https://your-api-domain.com
# ===========================================
# 上传文件配置
# ===========================================
# 最大上传文件大小（字节）- 5MB = 5242880
MAX_UPLOAD_SIZE=5242880
# 允许的文件扩展名 - 多个用英文逗号分隔
ALLOWED_EXTENSIONS=json
# ===========================================
# 学生初始配置
# ===========================================
# 学生初始操行分 - 新生导入时的默认分数，默认60分
STUDENT_INITIAL_POINTS=60

41
backend/config.py
View File

@@ -1,5 +1,5 @@
# ===========================================
# 班级操行分管理系统 - 后端服务
# 班级操行分管理系统 - 配置管理
#
# 开发者: Canglan
# 联系方式: admin@sea-studio.top
@@ -13,21 +13,15 @@ import os
from dotenv import load_dotenv
from typing import List
# 加载环境变量
load_dotenv()
class Settings:
"""应用配置类"""
# ========== 应用配置 ==========
APP_NAME: str = os.getenv("APP_NAME", "班级操行分管理系统")
APP_ENV: str = os.getenv("APP_ENV", "production")
DEBUG: bool = os.getenv("DEBUG", "False").lower() == "true"
SECRET_KEY: str = os.getenv("SECRET_KEY", "")
API_VERSION: str = os.getenv("API_VERSION", "v1")
# ========== 数据库配置 ==========
DB_HOST: str = os.getenv("DB_HOST", "127.0.0.1")
DB_PORT: int = int(os.getenv("DB_PORT", "3306"))
DB_USER: str = os.getenv("DB_USER", "root")
@@ -36,7 +30,6 @@ class Settings:
DB_POOL_SIZE: int = int(os.getenv("DB_POOL_SIZE", "10"))
DB_MAX_OVERFLOW: int = int(os.getenv("DB_MAX_OVERFLOW", "20"))
# ========== Redis配置 ==========
REDIS_HOST: str = os.getenv("REDIS_HOST", "127.0.0.1")
REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))
REDIS_PASSWORD: str = os.getenv("REDIS_PASSWORD", "")
@@ -45,63 +38,47 @@ class Settings:
@property
def REDIS_URL(self) -> str:
"""获取Redis连接URL"""
if self.REDIS_PASSWORD:
return f"redis://:{self.REDIS_PASSWORD}@{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"
return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"
# ========== JWT配置 ==========
JWT_SECRET_KEY: str = os.getenv("JWT_SECRET_KEY", "")
JWT_ALGORITHM: str = os.getenv("JWT_ALGORITHM", "HS256")
JWT_EXPIRE_MINUTES: int = int(os.getenv("JWT_EXPIRE_MINUTES", "30"))
# ========== 密码加密 ==========
PASSWORD_SALT: str = os.getenv("PASSWORD_SALT", "")
# ========== 调试入口 ==========
DEBUG_PATH: str = os.getenv("DEBUG_PATH", "/debug_add_admin")
# ========== 扣分规则 ==========
DEDUCTION_HOMEWORK_NOT_SUBMIT: int = int(os.getenv("DEDUCTION_HOMEWORK_NOT_SUBMIT", "2"))
DEDUCTION_HOMEWORK_LATE: int = int(os.getenv("DEDUCTION_HOMEWORK_LATE", "1"))
DEDUCTION_ATTENDANCE_ABSENT: int = int(os.getenv("DEDUCTION_ATTENDANCE_ABSENT", "5"))
DEDUCTION_ATTENDANCE_LATE: int = int(os.getenv("DEDUCTION_ATTENDANCE_LATE", "2"))
DEDUCTION_ATTENDANCE_LEAVE: int = int(os.getenv("DEDUCTION_ATTENDANCE_LEAVE", "1"))
# ========== 劳动委员固定分值 ==========
LABOR_POINTS_ADD: int = int(os.getenv("LABOR_POINTS_ADD", "1"))
LABOR_POINTS_SUBTRACT: int = int(os.getenv("LABOR_POINTS_SUBTRACT", "-1"))
# ========== 班长加减分限制 ==========
MONITOR_MAX_ADD: int = int(os.getenv("MONITOR_MAX_ADD", "5"))
MONITOR_MAX_SUBTRACT: int = int(os.getenv("MONITOR_MAX_SUBTRACT", "-5"))
# ========== 日志配置 ==========
LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
LOG_MAX_BYTES: int = int(os.getenv("LOG_MAX_BYTES", "104857600"))
LOG_BACKUP_COUNT: int = int(os.getenv("LOG_BACKUP_COUNT", "30"))
LOG_RETENTION_DAYS: int = int(os.getenv("LOG_RETENTION_DAYS", "365"))
# ========== CORS配置 ==========
CORS_ORIGINS: List[str] = os.getenv("CORS_ORIGINS", "http://localhost:8080").split(",")
@property
def CORS_ORIGINS(self) -> List[str]:
origins = os.getenv("CORS_ORIGINS", "")
return [origin.strip() for origin in origins.split(",") if origin.strip()]
# ========== 上传配置 ==========
MAX_UPLOAD_SIZE: int = int(os.getenv("MAX_UPLOAD_SIZE", "5242880"))
ALLOWED_EXTENSIONS: set = set(os.getenv("ALLOWED_EXTENSIONS", "json").split(","))
# ========== 学生初始配置 ==========
STUDENT_INITIAL_POINTS: int = int(os.getenv("STUDENT_INITIAL_POINTS", "60"))
def validate(self) -> None:
"""验证必要配置是否存在"""
required_configs = [
("SECRET_KEY", self.SECRET_KEY),
("JWT_SECRET_KEY", self.JWT_SECRET_KEY),
("PASSWORD_SALT", self.PASSWORD_SALT),
]
for name, value in required_configs:
if not value:
required = ["SECRET_KEY", "JWT_SECRET_KEY", "PASSWORD_SALT"]
for name in required:
if not getattr(self, name):
raise ValueError(f"配置 {name} 不能为空")
settings = Settings()

15
backend/main.py
View File

@@ -1,11 +1,12 @@
# ===========================================
# 班级操行分管理系统 - FastAPI 主入口
# ===========================================
# 班级操行分管理系统 - 主入口
#
# 开发者: Canglan
# 联系方式: admin@sea-studio.top
# 版权归属: Sea Network Technology Studio
# 许可证: MIT License
# 版权所有: Copyright (c) 2024 Sea Network Technology Studio
#
# 版权所有 © Sea Network Technology Studio
# ===========================================
from fastapi import FastAPI, Request
@@ -17,6 +18,7 @@ from config import settings
from utils.logger import setup_logger, log_access
from utils.database import init_db_pool, close_db_pool
from utils.redis_client import init_redis_pool, close_redis_pool
from middleware.auth_middleware import AuthMiddleware
from routes import auth, student, parent, admin, subject, debug
@@ -57,13 +59,14 @@ async def access_log_middleware(request: Request, call_next):
return response
# CORS中间件
# CORS中间件 - 从环境变量读取允许的域名
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
allow_headers=["*"],
expose_headers=["*"],
)
@@ -78,13 +81,11 @@ app.include_router(debug.router, tags=["调试"])
@app.get("/")
async def root():
"""根路径健康检查"""
return {"status": "ok", "message": f"{settings.APP_NAME} API 运行中"}
@app.get("/health")
async def health_check():
"""健康检查接口"""
return {"status": "healthy"}

125
backend/middleware/permission.py
View File

@@ -9,19 +9,13 @@
# 版权所有 © Sea Network Technology Studio
# ===========================================
from fastapi import Request, HTTPException
from fastapi import Request
from typing import List, Optional, Callable, Dict, Any
from functools import wraps
from utils.response import forbidden_response
from utils.database import execute_one
from utils.logger import get_logger
logger = get_logger(__name__)
async def get_current_user(request: Request) -> Dict[str, Any]:
"""获取当前登录用户信息"""
return {
"user_id": getattr(request.state, 'user_id', None),
"username": getattr(request.state, 'username', None),
@@ -30,184 +24,99 @@ async def get_current_user(request: Request) -> Dict[str, Any]:
"role": getattr(request.state, 'role', None)
}
async def get_current_user_id(request: Request) -> int:
"""获取当前用户ID"""
return getattr(request.state, 'user_id', None)
class PermissionChecker:
"""权限检查器"""
@staticmethod
async def get_user_role(user_id: int) -> Optional[str]:
"""获取用户的管理员角色"""
sql = """
SELECT role_type FROM admin_roles
WHERE user_id = %s
LIMIT 1
"""
sql = "SELECT role_type FROM admin_roles WHERE user_id = %s LIMIT 1"
result = await execute_one(sql, (user_id,))
return result["role_type"] if result else None
@staticmethod
async def get_user_class_id(user_id: int) -> Optional[int]:
"""获取用户管理的班级ID"""
sql = """
SELECT class_id FROM admin_roles
WHERE user_id = %s
LIMIT 1
"""
result = await execute_one(sql, (user_id,))
return result["class_id"] if result else None
@staticmethod
async def get_user_subject_ids(user_id: int) -> List[int]:
"""获取科代表管理的科目ID列表"""
sql = """
SELECT subject_id FROM admin_roles
WHERE user_id = %s AND role_type = '科代表'
"""
results = await execute_one(sql, (user_id,))
if results:
return [r["subject_id"] for r in results] if isinstance(results, list) else [results["subject_id"]]
return []
@staticmethod
async def check_is_teacher(user_id: int) -> bool:
"""检查是否为班主任"""
role = await PermissionChecker.get_user_role(user_id)
return role == "班主任"
@staticmethod
async def check_is_monitor(user_id: int) -> bool:
"""检查是否为班长"""
role = await PermissionChecker.get_user_role(user_id)
return role == "班长"
@staticmethod
async def check_is_subject_rep(user_id: int, subject_id: int = None) -> bool:
"""检查是否为科代表"""
async def check_is_study_commissioner(user_id: int) -> bool:
role = await PermissionChecker.get_user_role(user_id)
if role != "科代表":
return False
if subject_id:
subject_ids = await PermissionChecker.get_user_subject_ids(user_id)
return subject_id in subject_ids
return True
return role == "学习委员"
@staticmethod
async def check_is_attendance_rep(user_id: int) -> bool:
"""检查是否为考勤委员"""
role = await PermissionChecker.get_user_role(user_id)
return role == "考勤委员"
@staticmethod
async def check_is_labor_rep(user_id: int) -> bool:
"""检查是否为劳动委员"""
role = await PermissionChecker.get_user_role(user_id)
return role == "劳动委员"
@staticmethod
async def check_can_manage_subjects(user_id: int) -> bool:
role = await PermissionChecker.get_user_role(user_id)
return role in ["班主任", "学习委员"]
@staticmethod
async def check_can_revoke(user_id: int, record_id: int) -> bool:
"""
检查是否可以撤销扣分记录
班主任：可以撤销任何记录
班长：可以撤销任何记录
其他：只能撤销自己的记录
"""
# 获取记录信息
sql = "SELECT recorder_id FROM conduct_records WHERE record_id = %s"
record = await execute_one(sql, (record_id,))
if not record:
return False
role = await PermissionChecker.get_user_role(user_id)
# 班主任或班长可以撤销任何记录
if role in ["班主任", "班长"]:
return True
# 其他人只能撤销自己的记录
return record["recorder_id"] == user_id
@staticmethod
async def check_can_manage_student(user_id: int, student_id: int) -> bool:
"""检查是否可以管理该学生（同班级）"""
# 获取学生班级
sql = "SELECT class_id FROM students WHERE student_id = %s"
student = await execute_one(sql, (student_id,))
if not student:
return False
# 获取管理员管理的班级
admin_class_id = await PermissionChecker.get_user_class_id(user_id)
return admin_class_id == student["class_id"]
def require_auth(func: Callable):
"""需要认证的装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
if not request or not hasattr(request, 'state') or not hasattr(request.state, 'user_id'):
if not request or not hasattr(request.state, 'user_id'):
return forbidden_response("请先登录")
return await func(*args, **kwargs)
return wrapper
def require_role(roles: List[str]):
"""需要特定角色的装饰器"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
if not request or not hasattr(request, 'state'):
if not request or not hasattr(request.state, 'user_id'):
return forbidden_response("请先登录")
user_id = request.state.user_id
user_role = await PermissionChecker.get_user_role(user_id)
if user_role not in roles:
return forbidden_response(f"需要{','.join(roles)}权限")
return await func(*args, **kwargs)
return wrapper
return decorator
def require_teacher(func: Callable):
"""需要班主任权限的装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
if not request or not hasattr(request, 'state'):
if not request or not hasattr(request.state, 'user_id'):
return forbidden_response("请先登录")
user_id = request.state.user_id
is_teacher = await PermissionChecker.check_is_teacher(user_id)
if not is_teacher:
if not await PermissionChecker.check_is_teacher(request.state.user_id):
return forbidden_response("需要班主任权限")
return await func(*args, **kwargs)
return wrapper
def require_monitor(func: Callable):
"""需要班长权限的装饰器"""
def require_study_commissioner(func: Callable):
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
if not request or not hasattr(request, 'state'):
if not request or not hasattr(request.state, 'user_id'):
return forbidden_response("请先登录")
user_id = request.state.user_id
is_monitor = await PermissionChecker.check_is_monitor(user_id)
if not is_monitor:
return forbidden_response("需要班长权限")
if not await PermissionChecker.check_is_study_commissioner(request.state.user_id):
return forbidden_response("需要学习委员权限")
return await func(*args, **kwargs)
return wrapper

35
backend/models/class_model.py
View File

@@ -1,35 +0,0 @@
# ===========================================
# 班级操行分管理系统 - 后端服务
#
# 开发者: Canglan
# 联系方式: admin@sea-studio.top
# 版权归属: Sea Network Technology Studio
# 许可证: MIT License
#
# 版权所有 © Sea Network Technology Studio
# ===========================================
from typing import Optional, Dict, Any, List
from utils.database import execute_one, execute_query, execute_insert, execute_update
class ClassModel:
"""班级数据模型"""
@staticmethod
async def get_by_id(class_id: int) -> Optional[Dict[str, Any]]:
sql = "SELECT * FROM classes WHERE class_id = %s"
return await execute_one(sql, (class_id,))
@staticmethod
async def get_all() -> List[Dict[str, Any]]:
sql = "SELECT * FROM classes ORDER BY class_id"
return await execute_query(sql)
@staticmethod
async def create(class_name: str, grade: str = None, academic_year: str = None) -> int:
sql = """
INSERT INTO classes (class_name, grade, academic_year)
VALUES (%s, %s, %s)
"""
return await execute_insert(sql, (class_name, grade, academic_year))

23
backend/routes/admin.py
View File

@@ -214,6 +214,9 @@ async def get_assignments(request: Request):
获取作业列表
"""
user = await get_current_user(request)
role = await PermissionChecker.get_user_role(user["user_id"])
if role not in ["班主任", "学习委员"]:
return error_response(message="无权限", code=403)
result = await HomeworkService.get_assignments(user["user_id"])
@@ -268,10 +271,10 @@ async def create_assignment(
@router.put("/homework/submission")
async def update_submission_status(request: Request, req: UpdateHomeworkStatusRequest):
"""
更新作业提交状态（科代表）
"""
user = await get_current_user(request)
role = await PermissionChecker.get_user_role(user["user_id"])
if role not in ["班主任", "学习委员"]:
return error_response(message="无权进行此操作", code=403)
result = await HomeworkService.update_submission_status(
submission_id=req.submission_id,
@@ -340,9 +343,19 @@ async def add_admin(request: Request, req: AddAdminRequest):
"""
user = await get_current_user(request)
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
if not is_teacher:
if not await PermissionChecker.check_is_teacher(user["user_id"]):
return error_response(message="仅班主任可添加管理员", code=403)
# 验证角色类型是否合法
if req.role_type not in ["班长", "学习委员", "考勤委员", "劳动委员"]:
return error_response(message="无效的角色类型", code=400)
result = await AdminService.add_admin(
username=req.username,
real_name=req.real_name,
password=req.password,
role_type=req.role_type,
operator_id=user["user_id"]
)
result = await AdminService.add_admin(
username=req.username,

16
backend/routes/debug.py
View File

@@ -1,5 +1,5 @@
# ===========================================
# 班级操行分管理系统 - 后端服务
# 班级操行分管理系统 - 调试入口
#
# 开发者: Canglan
# 联系方式: admin@sea-studio.top
@@ -9,7 +9,7 @@
# 版权所有 © Sea Network Technology Studio
# ===========================================
from fastapi import APIRouter, Request, HTTPException
from fastapi import APIRouter, Request
from pydantic import BaseModel
from typing import Optional
@@ -23,12 +23,10 @@ logger = get_logger(__name__)
class AddAdminDebugRequest(BaseModel):
"""添加管理员请求"""
username: str
password: str
real_name: str
role_type: str # 班主任/班长/科代表/考勤委员/劳动委员
class_id: int
role_type: str # 班主任/班长/学习委员/考勤委员/劳动委员
subject_id: Optional[int] = None
@@ -38,20 +36,22 @@ async def debug_add_admin(request: Request, req: AddAdminDebugRequest):
调试入口 - 添加第一批管理员
注意：此接口仅用于首次部署，使用后建议注释掉此路由
"""
# 检查是否已存在管理员
from models.user import UserModel
existing = await UserModel.get_by_username(req.username)
if existing:
return error_response(message="用户名已存在")
# 验证角色类型
valid_roles = ["班主任", "班长", "学习委员", "考勤委员", "劳动委员"]
if req.role_type not in valid_roles:
return error_response(message=f"无效的角色类型，可选: {', '.join(valid_roles)}")
# 创建管理员账号
result = await AdminService.add_admin(
username=req.username,
real_name=req.real_name,
password=req.password,
role_type=req.role_type,
class_id=req.class_id,
subject_id=req.subject_id,
operator_id=0 # 系统添加
)

84
backend/routes/subject.py
View File

@@ -9,97 +9,41 @@
# 版权所有 © Sea Network Technology Studio
# ===========================================
from fastapi import APIRouter, Request, Query
from fastapi import APIRouter, Request
from typing import Optional
from middleware.permission import get_current_user, PermissionChecker
from services.subject_service import SubjectService
from schemas.subject import CreateSubjectRequest, UpdateSubjectRequest
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 get_subjects(
request: Request,
is_active: Optional[bool] = None
):
"""
获取科目列表
"""
async def get_subjects(request: Request, is_active: Optional[bool] = None):
user = await get_current_user(request)
result = await SubjectService.get_subjects(is_active=is_active)
return success_response(data=result)
@router.post("/create")
async def create_subject(request: Request, req: CreateSubjectRequest):
"""
创建科目（班主任）
"""
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 SubjectService.create_subject(
subject_name=req.subject_name,
subject_code=req.subject_code,
sort_order=req.sort_order
)
if result["success"]:
return success_response(data=result, message="科目创建成功")
else:
return error_response(message=result["message"])
if not await PermissionChecker.check_can_manage_subjects(user["user_id"]):
return error_response(message="无权限", code=403)
result = await SubjectService.create_subject(req.subject_name, req.subject_code, req.sort_order)
return success_response(data=result, message="科目创建成功") if result["success"] else error_response(message=result["message"])
@router.put("/update/{subject_id}")
async def update_subject(
request: Request,
subject_id: int,
req: UpdateSubjectRequest
):
"""
更新科目（班主任）
"""
async def update_subject(request: Request, subject_id: int, req: UpdateSubjectRequest):
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 SubjectService.update_subject(
subject_id=subject_id,
**req.dict(exclude_none=True)
)
if result["success"]:
return success_response(message="科目更新成功")
else:
return error_response(message=result["message"])
if not await PermissionChecker.check_can_manage_subjects(user["user_id"]):
return error_response(message="无权限", code=403)
result = await SubjectService.update_subject(subject_id, **req.dict(exclude_none=True))
return success_response(message="科目更新成功") if result["success"] else error_response(message=result["message"])
@router.delete("/delete/{subject_id}")
async def delete_subject(request: Request, subject_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)
if not await PermissionChecker.check_can_manage_subjects(user["user_id"]):
return error_response(message="无权限", code=403)
result = await SubjectService.delete_subject(subject_id)
if result["success"]:
return success_response(message="科目已禁用")
else:
return error_response(message=result["message"])
return success_response(message="科目已禁用") if result["success"] else error_response(message=result["message"])