from app.config.extensions import db from app.models.channel import Channel from app.models.program import Program from app.models.schedule import Schedule from app.services.mapper_service import IDMapper, ExistChecker class ScheduleService: """ Service class for managing TV schedules. Methods: create_schedule(data: dict) -> dict: Creates a new TV schedule and returns its data as a dictionary. get_schedule_by_id(schedule_id: str) -> dict: Retrieves a single TV schedule by its ID and returns its data as a dictionary. update_schedule(schedule_id: str, data: dict) -> dict: Updates an existing TV schedule and returns its updated data as a dictionary. delete_schedule(schedule_id: str) -> bool: Deletes a TV schedule by its ID and returns True if successful, False otherwise. get_schedules_by_channel(channel_id: str) -> list[dict]: Retrieves all TV schedules for a specific channel. get_schedules_by_program(program_id: str) -> list[dict]: Retrieves all TV schedules for a specific program. count_schedules() -> int: Count the total number of TV schedules in the database. delete_all_schedules() -> int: Delete all TV schedules in the database. schedule_exists(channel_id: str, program_id: str, start_time: str, end_time: str) -> bool: Check if a TV schedule with specific attributes exists in the database. """ @staticmethod def create_schedule(data: dict) -> dict: """ Create a new TV schedule and save it to the database. Args: data (dict): A dictionary containing TV schedule data. Returns: dict: The created TV schedule data as a dictionary. """ # Create a new Schedule object with the provided data schedule = Schedule(**data) # schedule.program = Program(**dict(data['program'])) # # Save the newly created schedule to the database schedule.save() # Convert the schedule object to a dictionary representation return schedule.to_dict() @staticmethod def get_schedule_by_id(schedule_id: str) -> dict: """ Retrieve a single TV schedule by its ID. Args: schedule_id (str): The ID of the schedule to retrieve. Returns: dict: The schedule data as a dictionary. """ # Map the public ID to the internal ID for the Schedule model schedule_id = IDMapper.public_id_mapper(schedule_id, Schedule) # Check if the mapping was successful and if an internal ID exists if not schedule_id: return None # Query the database to retrieve the schedule by its internal ID schedule = Schedule.objects(id=schedule_id).first() # Convert the schedule object to a dictionary representation return schedule @staticmethod def update_schedule(schedule_id: str, data: dict) -> dict: """ Update an existing TV schedule by its ID. Args: schedule_id (str): The ID of the schedule to update. data (dict): A dictionary containing the updated schedule data. Returns: dict: The updated schedule data as a dictionary. """ # Retrieve the existing schedule data by its ID using the get_schedule_by_id function schedule = ScheduleService.get_schedule_by_id(schedule_id) # Check if the schedule exists if schedule: # Update the schedule fields with the provided data schedule.modify(**data) # Save the updated schedule data schedule.update() # Convert the updated schedule object to a dictionary representation return schedule.to_dict() else: # Return None or raise an exception if the schedule with the specified ID doesn't exist. return None @staticmethod def delete_schedule(schedule_id: str) -> bool: """ Delete a TV schedule by its ID. Args: schedule_id (str): The ID of the schedule to delete. Returns: bool: True if the schedule was successfully deleted, False otherwise. """ # Retrieve the existing schedule data by its ID using the get_schedule_by_id function schedule = ScheduleService.get_schedule_by_id(schedule_id) # Check if the schedule exists if schedule: # Delete the schedule from the database schedule.delete() # Return True to indicate a successful deletion return True else: # Return False if the schedule with the specified ID doesn't exist. return False @staticmethod def get_all_schedules() -> list: """ Retrieve all TV Schedules from the database. Returns: List: A list of all TV schedules data as dictionaries. Example: all_schedules_data = ScheduleService.get_all_schedules() """ # Query the database to retrieve all TV schedule objects schedules = Schedule.objects() # Convert each TV schedule object to a dictionary representation and create a list of them schedule_data_list = [schedule.to_dict() for schedule in schedules] # Return the list of TV schedule data as dictionaries return schedule_data_list @staticmethod def get_schedules_by_channel(channel_id: str) -> list: """ Retrieve all TV schedules for a specific channel. Args: channel_id (str): The ID of the channel to retrieve schedules for. Returns: List: A list of TV schedule data as dictionaries for the specified channel. """ # Map the public channel ID to the internal ID for the Channel model channel_id = IDMapper.public_id_mapper(channel_id, Channel) # Check if the mapping was successful and if an internal channel ID exists if not channel_id: return [] # Query the database to retrieve schedules for the specified channel schedules = Schedule.objects(channel=channel_id) # Convert the matching schedule objects to a list of dictionaries return [schedule.to_dict() for schedule in schedules] @staticmethod def get_schedules_by_program(program_id: str) -> list: """ Retrieve all TV schedules for a specific program. Args: program_id (str): The ID of the program to retrieve schedules for. Returns: List: A list of TV schedule data as dictionaries for the specified program. """ # Map the public program ID to the internal ID for the Program model program_id = IDMapper.public_id_mapper(program_id, Program) # Check if the mapping was successful and if an internal program ID exists if not program_id: return [] # Query the database to retrieve schedules for the specified program schedules = Schedule.objects(program=program_id) # Convert the matching schedule objects to a list of dictionaries return [schedule.to_dict() for schedule in schedules] @staticmethod def count_schedules() -> int: """ Count the total number of TV schedules in the database. Returns: int: The total count of TV schedules. """ # Use the .count() method to count the number of schedules in the database return Schedule.objects().count() @staticmethod def delete_all_schedules() -> int: """ Delete all TV schedules in the database. Returns: int: The number of TV schedules deleted. """ # Query the database to retrieve all TV schedules schedules = Schedule.objects() # Initialize a count for deleted schedules deleted_count = 0 # Iterate through the list of TV schedules and delete each one for schedule in schedules: # Delete the schedule from the database schedule.delete() deleted_count += 1 # Return the total count of deleted TV schedules return deleted_count # @staticmethod # def schedule_exists(channel_id: str, program_id: str, start_time: str, end_time: str) -> bool: # """ # Check if a TV schedule with specific attributes exists in the database. # Args: # channel_id (str): The ID of the channel for the schedule. # program_id (str): The ID of the program for the schedule. # start_time (str): The start time of the schedule. # end_time (str): The end time of the schedule. # Returns: # bool: True if a schedule with the specified attributes exists, False otherwise. # """ # # Map the public channel ID to the internal ID for the Channel model # channel_id = IDMapper.public_id_mapper(channel_id, Channel) # # Map the public program ID to the internal ID for the Program model # program_id = IDMapper.public_id_mapper(program_id, Program) # # Check if the mappings were successful and if internal IDs exist # if not channel_id or not program_id: # return False # # Query the database to check if a schedule with the specified attributes exists # return ExistChecker.exist(channel=channel_id, program=program_id, start_time=start_time, end_time=end_time, Schedule)