From 220ece45e4de31c09dd6b5a59a6cee29c2ff3a20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=88=9A=28noham=29=C2=B2?= <100566912+NohamR@users.noreply.github.com> Date: Wed, 19 Nov 2025 21:27:15 +0100 Subject: [PATCH] Add EPG program selection to stream workflow Introduces EPG data fetching and program selection in main.py, allowing users to choose a program from the EPG guide if the start date is within the last 7 days. Adds get_epg_data_at and select_program_from_epg utilities to input.py, and updates stream_selection to include channel info in the returned selections. --- main.py | 20 +++++++ utils/input.py | 147 ++++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 166 insertions(+), 1 deletion(-) diff --git a/main.py b/main.py index a2d74db..4c9176c 100644 --- a/main.py +++ b/main.py @@ -1,14 +1,34 @@ """Main module for Oqee channel selection and stream management.""" +from datetime import datetime, timedelta from utils.input import ( stream_selection, get_date_input, + get_epg_data_at, + select_program_from_epg ) if __name__ == "__main__": try: selections = stream_selection() + freebox_id = selections.get("channel", {}).get("freebox_id") + channel_id = selections.get("channel", {}).get("id") + start_date, end_date = get_date_input() + if start_date > datetime.now() - timedelta(days=7): + epg_data = get_epg_data_at(start_date) + + programs = epg_data["entries"][str(channel_id)] + program_selection = select_program_from_epg( + programs, + start_date, + end_date + ) + if program_selection: + start_date = program_selection['start_date'] + end_date = program_selection['end_date'] + title = program_selection['title'] + except KeyboardInterrupt: print("\n\nProgramme interrompu par l'utilisateur. Au revoir !") diff --git a/utils/input.py b/utils/input.py index a873f82..f95d77f 100644 --- a/utils/input.py +++ b/utils/input.py @@ -15,6 +15,7 @@ from utils.stream import ( SERVICE_PLAN_API_URL = "https://api.oqee.net/api/v6/service_plan" EPG_API_URL = "https://api.oqee.net/api/v1/epg/all/{unix}" + class DatetimeValidator(Validator): """ Validateur personnalisé pour les chaînes datetime au format "YYYY-MM-DD HH:MM:SS". @@ -28,6 +29,7 @@ class DatetimeValidator(Validator): cursor_position=len(document.text), ) from exc + class DurationValidator(Validator): """ Validateur personnalisé pour les chaînes de durée au format "HH:MM:SS". @@ -48,6 +50,8 @@ class DurationValidator(Validator): message="Format invalide. Utilisez HH:MM:SS avec des nombres valides.", cursor_position=len(document.text), ) from exc + + def get_date_input(): """Prompt user for start and end date/time or duration. @@ -113,7 +117,9 @@ def get_date_input(): elif end_date_result.get("datetime"): try: - end_date = datetime.datetime.strptime(end_date_result["datetime"], "%Y-%m-%d %H:%M:%S") + end_date = datetime.datetime.strptime( + end_date_result["datetime"], "%Y-%m-%d %H:%M:%S" + ) print(f"\nDate/heure de fin : {end_date}") except (ValueError, TypeError): print("Impossible d'analyser la chaîne de date/heure fournie.") @@ -320,7 +326,146 @@ def stream_selection(): break if final_selections: + final_selections['channel'] = selected_channel return final_selections print("\nAucun flux n'a été sélectionné.") return None + + +def get_epg_data_at(dt: datetime.datetime): + """ + Fetch EPG data from the Oqee API for the nearest aligned hour of a given datetime. + + Args: + dt (datetime.datetime): datetime输入 (with hour, minute, etc.) + + Returns: + dict | None: EPG data or None on error + """ + + # Round to nearest hour + if dt.minute >= 30: + dt_aligned = (dt + datetime.timedelta(hours=1)).replace(minute=0, second=0, microsecond=0) + else: + dt_aligned = dt.replace(minute=0, second=0, microsecond=0) + + unix_time = int(dt_aligned.timestamp()) + print(f"Fetching EPG for aligned time: {dt_aligned} (unix={unix_time})") + + try: + response = requests.get(EPG_API_URL.format(unix=unix_time), timeout=10) + response.raise_for_status() + data = response.json() + + return data.get("result") + + except requests.exceptions.RequestException as e: + print(f"Une erreur réseau est survenue : {e}") + return None + except ValueError: + print("Erreur lors de l'analyse de la réponse JSON.") + return None + + +def select_program_from_epg(programs, original_start_date, original_end_date): + """ + Prompt user to select a program from EPG data or keep original selection. + + Args: + programs (list): List of program dictionaries from EPG data + original_start_date (datetime.datetime): User's original start date selection + original_end_date (datetime.datetime): User's original end date selection + + Returns: + dict: Dictionary containing: + - 'start_date': datetime object for start + - 'end_date': datetime object for end + - 'title': str or None (program title if selected) + - 'program': dict or None (full program data if selected) + """ + if not programs: + print("Aucun programme disponible dans le guide EPG.") + return { + 'start_date': original_start_date, + 'end_date': original_end_date, + 'title': None, + 'program': None + } + + # Create choices list with program information + program_choices = [] + for program in programs: + # Extract the live data from the program + live_data = program.get("live", program) + title = live_data.get('title', 'Sans titre') + start_time = datetime.datetime.fromtimestamp(live_data.get('start', 0)) + end_time = datetime.datetime.fromtimestamp(live_data.get('end', 0)) + duration_min = (end_time - start_time).total_seconds() / 60 + + choice_name = ( + f"{start_time.strftime('%H:%M')} - {end_time.strftime('%H:%M')} | " + f"{title} ({int(duration_min)} min)" + ) + program_choices.append({ + "name": choice_name, + "value": program # Store the full program object + }) + + # Add option to keep original selection + program_choices.insert(0, { + "name": ( + f"Garder la sélection manuelle originale " + f"({original_start_date.strftime('%Y-%m-%d %H:%M:%S')} - " + f"{original_end_date.strftime('%Y-%m-%d %H:%M:%S')})" + ), + "value": None + }) + + questions = [ + { + "type": "list", + "message": "Sélectionnez un programme ou gardez votre sélection manuelle :", + "choices": program_choices, + "long_instruction": "Utilisez les flèches pour naviguer, Entrée pour sélectionner.", + } + ] + + try: + result = prompt(questions) + if not result: + return None + + selected_program = result[0] + + # If user chose to keep original selection + if selected_program is None: + print("\n✅ Sélection manuelle conservée") + return { + 'start_date': original_start_date, + 'end_date': original_end_date, + 'title': None, + 'program': None + } + + # Extract live data and convert program timestamps to datetime objects + live_data = selected_program.get('live', selected_program) + program_start = datetime.datetime.fromtimestamp(live_data.get('start', 0)) + program_end = datetime.datetime.fromtimestamp(live_data.get('end', 0)) + program_title = live_data.get('title', 'Sans titre') + + print("\n✅ Programme sélectionné :") + print(f" - Titre : {program_title}") + print(f" - Début : {program_start.strftime('%Y-%m-%d %H:%M:%S')}") + print(f" - Fin : {program_end.strftime('%Y-%m-%d %H:%M:%S')}") + + return { + 'start_date': program_start, + 'end_date': program_end, + 'title': program_title, + 'program': selected_program + } + + except KeyboardInterrupt: + print("\nOpération annulée par l'utilisateur.") + return None