""" 课程服务层 """ from typing import Optional, List, Dict, Any from datetime import datetime from sqlalchemy import select, or_, and_, func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.core.logger import get_logger from app.core.exceptions import NotFoundError, BadRequestError, ConflictError from app.models.course import ( Course, CourseStatus, CourseMaterial, KnowledgePoint, GrowthPath, ) from app.models.course_exam_settings import CourseExamSettings from app.models.position_member import PositionMember from app.models.position_course import PositionCourse from app.schemas.course import ( CourseCreate, CourseUpdate, CourseList, CourseInDB, CourseMaterialCreate, KnowledgePointCreate, KnowledgePointUpdate, KnowledgePointInDB, GrowthPathCreate, ) from app.schemas.base import PaginationParams, PaginatedResponse from app.services.base_service import BaseService logger = get_logger(__name__) class CourseService(BaseService[Course]): """ 课程服务类 """ def __init__(self): super().__init__(Course) async def get_course_list( self, db: AsyncSession, *, page_params: PaginationParams, filters: CourseList, user_id: Optional[int] = None, ) -> PaginatedResponse[CourseInDB]: """ 获取课程列表(支持筛选) Args: db: 数据库会话 page_params: 分页参数 filters: 筛选条件 user_id: 用户ID(用于记录访问日志) Returns: 分页的课程列表 """ # 构建筛选条件 filter_conditions = [] # 状态筛选(默认只显示已发布的课程) if filters.status is not None: filter_conditions.append(Course.status == filters.status) else: # 如果没有指定状态,默认只返回已发布的课程 filter_conditions.append(Course.status == CourseStatus.PUBLISHED) # 分类筛选 if filters.category is not None: filter_conditions.append(Course.category == filters.category) # 是否推荐筛选 if filters.is_featured is not None: filter_conditions.append(Course.is_featured == filters.is_featured) # 关键词搜索 if filters.keyword: keyword = f"%{filters.keyword}%" filter_conditions.append( or_(Course.name.like(keyword), Course.description.like(keyword)) ) # 记录查询日志 logger.info( "查询课程列表", user_id=user_id, filters=filters.model_dump(exclude_none=True), page=page_params.page, size=page_params.page_size, ) # 执行分页查询 query = select(Course).where(Course.is_deleted == False) # 添加筛选条件 if filter_conditions: query = query.where(and_(*filter_conditions)) # 添加排序:优先按sort_order升序,其次按创建时间降序(新课程优先) query = query.order_by(Course.sort_order.asc(), Course.created_at.desc()) # 获取总数 count_query = ( select(func.count()).select_from(Course).where(Course.is_deleted == False) ) if filter_conditions: count_query = count_query.where(and_(*filter_conditions)) total_result = await db.execute(count_query) total = total_result.scalar() or 0 # 分页 query = query.offset(page_params.offset).limit(page_params.limit) query = query.options(selectinload(Course.materials)) # 执行查询 result = await db.execute(query) courses = result.scalars().all() # 获取用户所属的岗位ID列表 user_position_ids = [] if user_id: position_result = await db.execute( select(PositionMember.position_id).where( PositionMember.user_id == user_id, PositionMember.is_deleted == False ) ) user_position_ids = [row[0] for row in position_result.fetchall()] # 批量查询课程的岗位分配信息 course_ids = [c.id for c in courses] course_type_map = {} if course_ids and user_position_ids: position_course_result = await db.execute( select(PositionCourse.course_id, PositionCourse.course_type).where( PositionCourse.course_id.in_(course_ids), PositionCourse.position_id.in_(user_position_ids), PositionCourse.is_deleted == False ) ) # 构建课程类型映射:如果有多个岗位,优先取required for course_id, course_type in position_course_result.fetchall(): if course_id not in course_type_map: course_type_map[course_id] = course_type elif course_type == 'required': course_type_map[course_id] = 'required' # 转换为 Pydantic 模型,并附加课程类型 course_list = [] for course in courses: course_data = CourseInDB.model_validate(course) # 设置课程类型:如果用户有岗位分配则使用分配类型,否则为None course_data.course_type = course_type_map.get(course.id) course_list.append(course_data) # 计算总页数 pages = (total + page_params.page_size - 1) // page_params.page_size return PaginatedResponse( items=course_list, total=total, page=page_params.page, page_size=page_params.page_size, pages=pages, ) async def create_course( self, db: AsyncSession, *, course_in: CourseCreate, created_by: int ) -> Course: """ 创建课程 Args: db: 数据库会话 course_in: 课程创建数据 created_by: 创建人ID Returns: 创建的课程 """ # 检查名称是否重复 existing = await db.execute( select(Course).where( and_(Course.name == course_in.name, Course.is_deleted == False) ) ) existing_course = existing.scalar_one_or_none() if existing_course: raise ConflictError( f"课程名称 '{course_in.name}' 已存在", detail={"existing_id": existing_course.id, "existing_name": existing_course.name} ) # 创建课程 course_data = course_in.model_dump() course = await self.create(db, obj_in=course_data, created_by=created_by) # 自动创建默认考试设置 default_exam_settings = CourseExamSettings( course_id=course.id, created_by=created_by, updated_by=created_by # 其他字段使用模型定义的默认值: # single_choice_count=4, multiple_choice_count=2, true_false_count=1, # fill_blank_count=2, essay_count=1, duration_minutes=10, 等 ) db.add(default_exam_settings) await db.commit() await db.refresh(course) logger.info( "创建课程", course_id=course.id, course_name=course.name, created_by=created_by ) logger.info( "自动创建默认考试设置", course_id=course.id, exam_settings_id=default_exam_settings.id ) return course async def update_course( self, db: AsyncSession, *, course_id: int, course_in: CourseUpdate, updated_by: int, ) -> Course: """ 更新课程 Args: db: 数据库会话 course_id: 课程ID course_in: 课程更新数据 updated_by: 更新人ID Returns: 更新后的课程 """ # 获取课程 course = await self.get_by_id(db, course_id) if not course: raise NotFoundError(f"课程ID {course_id} 不存在") # 检查名称是否重复(如果修改了名称) if course_in.name and course_in.name != course.name: existing = await db.execute( select(Course).where( and_( Course.name == course_in.name, Course.id != course_id, Course.is_deleted == False, ) ) ) existing_course = existing.scalar_one_or_none() if existing_course: raise ConflictError( f"课程名称 '{course_in.name}' 已存在", detail={"existing_id": existing_course.id, "existing_name": existing_course.name} ) # 记录状态变更 old_status = course.status # 更新课程 update_data = course_in.model_dump(exclude_unset=True) # 如果状态变为已发布,记录发布时间 if ( update_data.get("status") == CourseStatus.PUBLISHED and old_status != CourseStatus.PUBLISHED ): update_data["published_at"] = datetime.now() update_data["publisher_id"] = updated_by course = await self.update( db, db_obj=course, obj_in=update_data, updated_by=updated_by ) logger.info( "更新课程", course_id=course.id, course_name=course.name, old_status=old_status, new_status=course.status, updated_by=updated_by, ) return course async def delete_course( self, db: AsyncSession, *, course_id: int, deleted_by: int ) -> bool: """ 删除课程(软删除 + 删除相关文件) Args: db: 数据库会话 course_id: 课程ID deleted_by: 删除人ID Returns: 是否删除成功 """ import shutil from pathlib import Path from app.core.config import settings course = await self.get_by_id(db, course_id) if not course: raise NotFoundError(f"课程ID {course_id} 不存在") # 放开删除限制:任意状态均可软删除,由业务方自行控制 # 执行软删除(标记 is_deleted,记录删除时间),由审计日志记录操作者 success = await self.soft_delete(db, id=course_id) if success: # 删除课程文件夹及其所有内容 course_folder = Path(settings.UPLOAD_PATH) / "courses" / str(course_id) if course_folder.exists() and course_folder.is_dir(): try: shutil.rmtree(course_folder) logger.info( "删除课程文件夹成功", course_id=course_id, folder_path=str(course_folder), ) except Exception as e: # 文件夹删除失败不影响业务流程,仅记录日志 logger.error( "删除课程文件夹失败", course_id=course_id, folder_path=str(course_folder), error=str(e), ) logger.warning( "删除课程", course_id=course_id, course_name=course.name, deleted_by=deleted_by, folder_deleted=course_folder.exists(), ) return success async def add_course_material( self, db: AsyncSession, *, course_id: int, material_in: CourseMaterialCreate, created_by: int, ) -> CourseMaterial: """ 添加课程资料 Args: db: 数据库会话 course_id: 课程ID material_in: 资料创建数据 created_by: 创建人ID Returns: 创建的课程资料 """ # 检查课程是否存在 course = await self.get_by_id(db, course_id) if not course: raise NotFoundError(f"课程ID {course_id} 不存在") # 创建资料 material_data = material_in.model_dump() material_data.update({ "course_id": course_id, "created_by": created_by, "updated_by": created_by }) material = CourseMaterial(**material_data) db.add(material) await db.commit() await db.refresh(material) logger.info( "添加课程资料", course_id=course_id, material_id=material.id, material_name=material.name, file_type=material.file_type, file_size=material.file_size, created_by=created_by, ) return material async def get_course_materials( self, db: AsyncSession, *, course_id: int, ) -> List[CourseMaterial]: """ 获取课程资料列表 Args: db: 数据库会话 course_id: 课程ID Returns: 课程资料列表 """ # 确认课程存在 course = await self.get_by_id(db, course_id) if not course: raise NotFoundError(f"课程ID {course_id} 不存在") stmt = ( select(CourseMaterial) .where( CourseMaterial.course_id == course_id, CourseMaterial.is_deleted == False, ) .order_by(CourseMaterial.sort_order.asc(), CourseMaterial.id.asc()) ) result = await db.execute(stmt) materials = result.scalars().all() logger.info( "查询课程资料列表", course_id=course_id, count=len(materials) ) return materials async def delete_course_material( self, db: AsyncSession, *, course_id: int, material_id: int, deleted_by: int, ) -> bool: """ 删除课程资料(软删除 + 删除物理文件) Args: db: 数据库会话 course_id: 课程ID material_id: 资料ID deleted_by: 删除人ID Returns: 是否删除成功 """ import os from pathlib import Path from app.core.config import settings # 先确认课程存在 course = await self.get_by_id(db, course_id) if not course: raise NotFoundError(f"课程ID {course_id} 不存在") # 查找资料并校验归属 material_stmt = select(CourseMaterial).where( CourseMaterial.id == material_id, CourseMaterial.course_id == course_id, CourseMaterial.is_deleted == False, ) result = await db.execute(material_stmt) material = result.scalar_one_or_none() if not material: raise NotFoundError(f"课程资料ID {material_id} 不存在或已删除") # 获取文件路径信息用于删除物理文件 file_url = material.file_url # 软删除数据库记录 material.is_deleted = True material.deleted_at = datetime.now() if hasattr(material, "deleted_by"): # 兼容存在该字段的表 setattr(material, "deleted_by", deleted_by) db.add(material) await db.commit() # 删除物理文件 if file_url and file_url.startswith("/static/uploads/"): try: # 从URL中提取相对路径 relative_path = file_url.replace("/static/uploads/", "") file_path = Path(settings.UPLOAD_PATH) / relative_path # 检查文件是否存在并删除 if file_path.exists() and file_path.is_file(): os.remove(file_path) logger.info( "删除物理文件成功", file_path=str(file_path), material_id=material_id, ) except Exception as e: # 物理文件删除失败不影响业务流程,仅记录日志 logger.error( "删除物理文件失败", file_url=file_url, material_id=material_id, error=str(e), ) logger.warning( "删除课程资料", course_id=course_id, material_id=material_id, deleted_by=deleted_by, file_deleted=file_url is not None, ) return True async def get_material_knowledge_points( self, db: AsyncSession, material_id: int ) -> List[KnowledgePointInDB]: """获取资料关联的知识点列表""" # 获取资料信息 result = await db.execute( select(CourseMaterial).where( CourseMaterial.id == material_id, CourseMaterial.is_deleted == False ) ) material = result.scalar_one_or_none() if not material: raise NotFoundError(f"资料ID {material_id} 不存在") # 直接查询关联到该资料的知识点 query = select(KnowledgePoint).where( KnowledgePoint.material_id == material_id, KnowledgePoint.is_deleted == False ).order_by(KnowledgePoint.created_at.desc()) result = await db.execute(query) knowledge_points = result.scalars().all() from app.schemas.course import KnowledgePointInDB return [KnowledgePointInDB.model_validate(kp) for kp in knowledge_points] async def add_material_knowledge_points( self, db: AsyncSession, material_id: int, knowledge_point_ids: List[int] ) -> List[KnowledgePointInDB]: """ 为资料添加知识点关联 注意:自2025-09-27起,知识点直接通过material_id关联到资料, material_knowledge_points中间表已废弃。此方法将更新知识点的material_id字段。 """ # 验证资料是否存在 result = await db.execute( select(CourseMaterial).where( CourseMaterial.id == material_id, CourseMaterial.is_deleted == False ) ) material = result.scalar_one_or_none() if not material: raise NotFoundError(f"资料ID {material_id} 不存在") # 验证知识点是否存在且属于同一课程 result = await db.execute( select(KnowledgePoint).where( KnowledgePoint.id.in_(knowledge_point_ids), KnowledgePoint.course_id == material.course_id, KnowledgePoint.is_deleted == False ) ) valid_knowledge_points = result.scalars().all() if len(valid_knowledge_points) != len(knowledge_point_ids): raise BadRequestError("部分知识点不存在或不属于同一课程") # 更新知识点的material_id字段 added_knowledge_points = [] for kp in valid_knowledge_points: # 更新知识点的资料关联 kp.material_id = material_id added_knowledge_points.append(kp) await db.commit() # 刷新对象以获取更新后的数据 for kp in added_knowledge_points: await db.refresh(kp) from app.schemas.course import KnowledgePointInDB return [KnowledgePointInDB.model_validate(kp) for kp in added_knowledge_points] async def remove_material_knowledge_point( self, db: AsyncSession, material_id: int, knowledge_point_id: int ) -> bool: """ 移除资料的知识点关联(软删除知识点) 注意:自2025-09-27起,知识点直接通过material_id关联到资料, material_knowledge_points中间表已废弃。此方法将软删除知识点。 """ # 查找知识点并验证归属 result = await db.execute( select(KnowledgePoint).where( KnowledgePoint.id == knowledge_point_id, KnowledgePoint.material_id == material_id, KnowledgePoint.is_deleted == False ) ) knowledge_point = result.scalar_one_or_none() if not knowledge_point: raise NotFoundError(f"知识点ID {knowledge_point_id} 不存在或不属于该资料") # 软删除知识点 knowledge_point.is_deleted = True knowledge_point.deleted_at = datetime.now() await db.commit() logger.info( "移除资料知识点关联", material_id=material_id, knowledge_point_id=knowledge_point_id, ) return True class KnowledgePointService(BaseService[KnowledgePoint]): """ 知识点服务类 """ def __init__(self): super().__init__(KnowledgePoint) async def get_knowledge_points_by_course( self, db: AsyncSession, *, course_id: int, material_id: Optional[int] = None ) -> List[KnowledgePoint]: """ 获取课程的知识点列表 Args: db: 数据库会话 course_id: 课程ID material_id: 资料ID(可选,用于筛选特定资料的知识点) Returns: 知识点列表 """ query = select(KnowledgePoint).where( and_( KnowledgePoint.course_id == course_id, KnowledgePoint.is_deleted == False, ) ) if material_id is not None: query = query.where(KnowledgePoint.material_id == material_id) query = query.order_by(KnowledgePoint.created_at.desc()) result = await db.execute(query) return result.scalars().all() async def create_knowledge_point( self, db: AsyncSession, *, course_id: int, point_in: KnowledgePointCreate, created_by: int, ) -> KnowledgePoint: """ 创建知识点 Args: db: 数据库会话 course_id: 课程ID point_in: 知识点创建数据 created_by: 创建人ID Returns: 创建的知识点 """ # 检查课程是否存在 course_service = CourseService() course = await course_service.get_by_id(db, course_id) if not course: raise NotFoundError(f"课程ID {course_id} 不存在") # 创建知识点 point_data = point_in.model_dump() point_data.update({"course_id": course_id}) knowledge_point = await self.create( db, obj_in=point_data, created_by=created_by ) logger.info( "创建知识点", course_id=course_id, knowledge_point_id=knowledge_point.id, knowledge_point_name=knowledge_point.name, created_by=created_by, ) return knowledge_point async def update_knowledge_point( self, db: AsyncSession, *, point_id: int, point_in: KnowledgePointUpdate, updated_by: int, ) -> KnowledgePoint: """ 更新知识点 Args: db: 数据库会话 point_id: 知识点ID point_in: 知识点更新数据 updated_by: 更新人ID Returns: 更新后的知识点 """ knowledge_point = await self.get_by_id(db, point_id) if not knowledge_point: raise NotFoundError(f"知识点ID {point_id} 不存在") # 验证关联资料是否存在 if hasattr(point_in, 'material_id') and point_in.material_id: result = await db.execute( select(CourseMaterial).where( CourseMaterial.id == point_in.material_id, CourseMaterial.is_deleted == False ) ) material = result.scalar_one_or_none() if not material: raise NotFoundError(f"资料ID {point_in.material_id} 不存在") # 更新知识点 update_data = point_in.model_dump(exclude_unset=True) knowledge_point = await self.update( db, db_obj=knowledge_point, obj_in=update_data, updated_by=updated_by ) logger.info( "更新知识点", knowledge_point_id=knowledge_point.id, knowledge_point_name=knowledge_point.name, updated_by=updated_by, ) return knowledge_point class GrowthPathService(BaseService[GrowthPath]): """ 成长路径服务类 """ def __init__(self): super().__init__(GrowthPath) async def create_growth_path( self, db: AsyncSession, *, path_in: GrowthPathCreate, created_by: int ) -> GrowthPath: """ 创建成长路径 Args: db: 数据库会话 path_in: 成长路径创建数据 created_by: 创建人ID Returns: 创建的成长路径 """ # 检查名称是否重复 existing = await db.execute( select(GrowthPath).where( and_(GrowthPath.name == path_in.name, GrowthPath.is_deleted == False) ) ) existing_path = existing.scalar_one_or_none() if existing_path: raise ConflictError( f"成长路径名称 '{path_in.name}' 已存在", detail={"existing_id": existing_path.id, "existing_name": existing_path.name, "type": "growth_path"} ) # 验证课程是否存在 if path_in.courses: course_ids = [c.course_id for c in path_in.courses] course_service = CourseService() for course_id in course_ids: course = await course_service.get_by_id(db, course_id) if not course: raise NotFoundError(f"课程ID {course_id} 不存在") # 创建成长路径 path_data = path_in.model_dump() # 转换课程列表为JSON格式 if path_data.get("courses"): path_data["courses"] = [c.model_dump() for c in path_in.courses] growth_path = await self.create(db, obj_in=path_data, created_by=created_by) logger.info( "创建成长路径", growth_path_id=growth_path.id, growth_path_name=growth_path.name, course_count=len(path_in.courses) if path_in.courses else 0, created_by=created_by, ) return growth_path # 创建服务实例 course_service = CourseService() knowledge_point_service = KnowledgePointService() growth_path_service = GrowthPathService()