from typing import Dict, List, Optional from datetime import datetime import uuid from src.models import DatabaseContextManager from src.models.models import ModuleVideo, CourseModule, CourseWeek, CourseLevel from src.utils import custom_response class VideoManager: """Manager class for video operations""" def __init__(self): self.table = ModuleVideo def create_video(self, payload: Dict) -> Dict: """ Create a new video Args: payload: Video data including module_id, title, description, etc. Returns: Response with created video or error """ try: # Validate required fields required_fields = ['module_id', 'video_title'] for field in required_fields: if field not in payload or not payload[field]: return custom_response( success=False, data=f"Missing required field: {field}", status_code=400 ) with DatabaseContextManager() as ctx: # Validate module exists module = ctx.session.query(CourseModule).filter( CourseModule.id == payload['module_id'] ).first() if not module: return custom_response( success=False, data="Module not found", status_code=404 ) # Create the video video = ModuleVideo( id=str(uuid.uuid4()), module_id=payload['module_id'], video_title=payload['video_title'], video_description=payload.get('video_description'), video_url=payload.get('video_url'), video_file_path=payload.get('video_file_path'), video_duration_seconds=payload.get('video_duration_seconds'), thumbnail_url=payload.get('thumbnail_url'), video_quality=payload.get('video_quality', 'HD'), is_required=payload.get('is_required', True), is_active=payload.get('is_active', True), created_at=datetime.utcnow() ) ctx.session.add(video) ctx.session.commit() return custom_response( success=True, data={ "id": video.id, "module_id": video.module_id, "video_title": video.video_title, "video_description": video.video_description, "video_url": video.video_url, "video_file_path": video.video_file_path, "video_duration_seconds": video.video_duration_seconds, "thumbnail_url": video.thumbnail_url, "video_quality": video.video_quality, "is_required": video.is_required, "is_active": video.is_active, "created_at": video.created_at.isoformat() if video.created_at else None }, status_code=201 ) except Exception as e: return custom_response( success=False, data=f"Error creating video: {str(e)}", status_code=500 ) def get_videos(self, requester_id: str, course_id: Optional[str] = None, module_id: Optional[str] = None, week_id: Optional[str] = None) -> Dict: """ Get videos with optional filtering Args: requester_id: ID of the user making the request course_id: Optional course ID filter module_id: Optional module ID filter week_id: Optional week ID filter Returns: Response with videos list """ try: with DatabaseContextManager() as ctx: query = ctx.session.query(ModuleVideo).join(CourseModule) # Apply filters if module_id: query = query.filter(ModuleVideo.module_id == module_id) elif week_id: # Get modules for the week query = query.join(CourseWeek, CourseModule.course_week_id == CourseWeek.id).filter( CourseWeek.id == week_id ) elif course_id: # Get modules for the course through levels and weeks query = query.join(CourseWeek, CourseModule.course_week_id == CourseWeek.id).join( CourseLevel, CourseWeek.course_level_id == CourseLevel.id ).filter(CourseLevel.course_id == course_id) videos = query.all() videos_data = [] for video in videos: videos_data.append({ "id": video.id, "module_id": video.module_id, "video_title": video.video_title, "video_description": video.video_description, "video_url": video.video_url, "video_file_path": video.video_file_path, "video_duration_seconds": video.video_duration_seconds, "thumbnail_url": video.thumbnail_url, "video_quality": video.video_quality, "is_required": video.is_required, "is_active": video.is_active, "created_at": video.created_at.isoformat() if video.created_at else None, "updated_at": video.updated_at.isoformat() if video.updated_at else None }) return custom_response( success=True, data=videos_data, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error fetching videos: {str(e)}", status_code=500 ) def get_video_by_id(self, video_id: str, requester_id: str) -> Dict: """ Get a specific video by ID Args: video_id: ID of the video requester_id: ID of the user making the request Returns: Response with video data or error """ try: with DatabaseContextManager() as ctx: video = ctx.session.query(ModuleVideo).filter( ModuleVideo.id == video_id ).first() if not video: return custom_response( success=False, data="Video not found", status_code=404 ) video_data = { "id": video.id, "module_id": video.module_id, "video_title": video.video_title, "video_description": video.video_description, "video_url": video.video_url, "video_file_path": video.video_file_path, "video_duration_seconds": video.video_duration_seconds, "thumbnail_url": video.thumbnail_url, "video_quality": video.video_quality, "is_required": video.is_required, "is_active": video.is_active, "created_at": video.created_at.isoformat() if video.created_at else None, "updated_at": video.updated_at.isoformat() if video.updated_at else None } return custom_response( success=True, data=video_data, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error fetching video: {str(e)}", status_code=500 ) def update_video(self, video_id: str, payload: Dict) -> Dict: """ Update video details Args: video_id: ID of the video to update payload: Updated video data Returns: Response with updated video or error """ try: with DatabaseContextManager() as ctx: video = ctx.session.query(ModuleVideo).filter( ModuleVideo.id == video_id ).first() if not video: return custom_response( success=False, data="Video not found", status_code=404 ) # Update fields if 'video_title' in payload: video.video_title = payload['video_title'] if 'video_description' in payload: video.video_description = payload['video_description'] if 'video_url' in payload: video.video_url = payload['video_url'] if 'video_file_path' in payload: video.video_file_path = payload['video_file_path'] if 'video_duration_seconds' in payload: video.video_duration_seconds = payload['video_duration_seconds'] if 'thumbnail_url' in payload: video.thumbnail_url = payload['thumbnail_url'] if 'video_quality' in payload: video.video_quality = payload['video_quality'] if 'is_required' in payload: video.is_required = payload['is_required'] if 'is_active' in payload: video.is_active = payload['is_active'] video.updated_at = datetime.utcnow() ctx.session.commit() return custom_response( success=True, data={ "id": video.id, "module_id": video.module_id, "video_title": video.video_title, "video_description": video.video_description, "video_url": video.video_url, "video_file_path": video.video_file_path, "video_duration_seconds": video.video_duration_seconds, "thumbnail_url": video.thumbnail_url, "video_quality": video.video_quality, "is_required": video.is_required, "is_active": video.is_active, "created_at": video.created_at.isoformat() if video.created_at else None, "updated_at": video.updated_at.isoformat() if video.updated_at else None }, status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error updating video: {str(e)}", status_code=500 ) def delete_video(self, video_id: str, requester_id: str) -> Dict: """ Delete a video Args: video_id: ID of the video to delete requester_id: ID of the user making the request Returns: Response with success or error """ try: with DatabaseContextManager() as ctx: video = ctx.session.query(ModuleVideo).filter( ModuleVideo.id == video_id ).first() if not video: return custom_response( success=False, data="Video not found", status_code=404 ) ctx.session.delete(video) ctx.session.commit() return custom_response( success=True, data="Video deleted successfully", status_code=200 ) except Exception as e: return custom_response( success=False, data=f"Error deleting video: {str(e)}", status_code=500 ) def get_videos_by_course(self, course_id: str, requester_id: str) -> Dict: """ Get all videos for a specific course Args: course_id: ID of the course requester_id: ID of the user making the request Returns: Response with videos list """ return self.get_videos(requester_id, course_id=course_id)