diff --git a/routes/epg.py b/routes/epg.py index c3d63d5..d02cb94 100644 --- a/routes/epg.py +++ b/routes/epg.py @@ -1,74 +1,112 @@ #!/usr/bin/env python3 """ -EPG-related route handlers +EPG (Electronic Program Guide) routes. + +These routes provide program schedule information for channels. + +Endpoints: + GET /api/epg/status – Cache/config health check + GET /api/epg/xmltv-channels – All channel IDs + display names from XMLTV source + GET /api/providers//epg – Bulk XMLTV export (download) + GET /api/providers//epg/grid – Multi-channel grid view (JSON) + GET /api/providers//channels//epg – Single-channel program list + GET /api/providers//programs/ – Deep program detail + GET /api/providers//epg-mapping – Load channel→EPG ID mapping + POST /api/providers//epg-mapping – Persist channel→EPG ID mapping """ -from bottle import request, response -from streaming_providers.base.utils import logger -from streaming_providers.base.epg_operations import EPGOperations +import gzip +import os +import xml.etree.ElementTree as ET +from datetime import datetime, timezone +from bottle import request, response +from streaming_providers.base.epg_operations import EPGOperations +from streaming_providers.base.utils import logger + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _parse_time_param(value: str | None) -> datetime | None: + """ + Parse a query-string time value into a timezone-aware datetime. + + Accepts: + - Unix timestamp (integer string): "1710000000" + - ISO 8601 string: "2024-03-10T00:00:00Z" + + Returns None and logs a warning if the value cannot be parsed. + """ + if not value: + return None + try: + return datetime.fromtimestamp(int(value), tz=timezone.utc) + except (ValueError, TypeError): + pass + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + logger.warning(f"Invalid time format, ignoring: {value!r}") + return None + + +def _open_xml(file_path: str): + """Open a plain or gzip-compressed XMLTV file for text reading.""" + if file_path.endswith(".gz"): + return gzip.open(file_path, "rt", encoding="utf-8") + return open(file_path, "r", encoding="utf-8") + + +# --------------------------------------------------------------------------- +# Route setup +# --------------------------------------------------------------------------- def setup_epg_routes(app, manager, service): - """Setup EPG-related routes""" + """Register all EPG-related routes on *app*.""" - # Instantiate EPGOperations once here so that EPGManager, EPGCache, - # EPGMapping and their VFS helpers are only constructed a single time, + # EPGOperations is instantiated once here so that EPGManager, EPGCache, + # EPGMapping and their VFS helpers are constructed only a single time, # not on every incoming HTTP request. epg_ops = EPGOperations(manager.registry) + # ------------------------------------------------------------------ + # 1. EPG system status + # ------------------------------------------------------------------ + @app.route("/api/epg/status", method="GET") def get_epg_status(): - """Get EPG configuration and cache status""" + """Return EPG configuration and cache status.""" try: + configured = bool(service.epg_url) and service.epg_url != "https://example.com/epg.xml.gz" result = { - "configured": bool(service.epg_url) - and service.epg_url != "https://example.com/epg.xml.gz", - "epg_url": service.epg_url if service.epg_url else "Not configured", + "configured": configured, + "epg_url": service.epg_url or "Not configured", "cache_valid": False, "cache_path": None, "channel_count": 0, "environment_used": False, } - # Check if we used the environment variable - import os - env_url = os.environ.get("ULTIMATE_EPG_URL") if env_url and env_url == service.epg_url: result["environment_used"] = True - if ( - result["configured"] - and hasattr(service, "epg_manager") - and service.epg_manager - ): + if configured and hasattr(service, "epg_manager") and service.epg_manager: try: - cache = service.epg_manager.cache - xml_path = cache.get_cached_file_path() - + xml_path = service.epg_manager.cache.get_cached_file_path() if xml_path: result["cache_valid"] = True result["cache_path"] = xml_path - - # Try to count channels - import gzip - import xml.etree.ElementTree as ET - - def open_xml_file(file_path): - if file_path.endswith(".gz"): - return gzip.open(file_path, "rt", encoding="utf-8") - else: - return open(file_path, "r", encoding="utf-8") - - channel_ids = set() + channel_ids: set[str] = set() try: - with open_xml_file(xml_path) as xml_file: - context = ET.iterparse(xml_file, events=("start",)) - for event, elem in context: + with _open_xml(xml_path) as f: + for event, elem in ET.iterparse(f, events=("start",)): if elem.tag == "channel": - channel_id = elem.get("id") - if channel_id: - channel_ids.add(channel_id) + cid = elem.get("id") + if cid: + channel_ids.add(cid) elem.clear() result["channel_count"] = len(channel_ids) except Exception as parse_err: @@ -85,20 +123,19 @@ def setup_epg_routes(app, manager, service): response.status = 500 return {"error": str(e)} + # ------------------------------------------------------------------ + # 2. XMLTV channel list + # ------------------------------------------------------------------ + @app.route("/api/epg/xmltv-channels", method="GET") def get_epg_xmltv_channels(): - """Get all unique channel IDs from EPG XML file with display names""" + """Return all channel IDs and display names from the XMLTV source file.""" try: - # Check if EPG manager is available if not hasattr(service, "epg_manager") or not service.epg_manager: response.status = 404 return {"error": "EPG module not available"} - # Check if we have a valid EPG URL configured - if ( - not service.epg_url - or service.epg_url == "https://example.com/epg.xml.gz" - ): + if not service.epg_url or service.epg_url == "https://example.com/epg.xml.gz": response.status = 400 return { "error": "EPG URL not configured", @@ -106,162 +143,283 @@ def setup_epg_routes(app, manager, service): "current_url": service.epg_url, } - # Get the cache manager from EPG manager cache = service.epg_manager.cache - - logger.info(f"EPG Channels: Using URL: {service.epg_url}") - - # This will download if not cached, or return cached path + logger.info(f"EPG xmltv-channels: using URL {service.epg_url}") xml_path = cache.get_or_download(service.epg_url) - if not xml_path: + if not xml_path or not os.path.exists(xml_path): response.status = 404 return { "error": f"EPG file not available from {service.epg_url}", - "details": "Failed to download or cache EPG file.", "hint": "Check if the URL is accessible and contains valid XMLTV data.", } - # Parse XML to get channel IDs and display names - import gzip - import os - import xml.etree.ElementTree as ET - - if not os.path.exists(xml_path): - response.status = 404 - return {"error": f"EPG file does not exist at path: {xml_path}"} - - channel_ids = [] - channel_map = {} # Map of id -> display name - - def open_xml_file(file_path): - if file_path.endswith(".gz"): - return gzip.open(file_path, "rt", encoding="utf-8") - else: - return open(file_path, "r", encoding="utf-8") - - logger.info(f"Parsing EPG file: {xml_path}") file_size = os.path.getsize(xml_path) - logger.info(f"EPG file size: {file_size} bytes") + logger.info(f"Parsing EPG file: {xml_path} ({file_size} bytes)") - with open_xml_file(xml_path) as xml_file: - # Use iterparse for memory efficiency - context = ET.iterparse(xml_file, events=("start", "end")) + channel_ids: list[str] = [] + channel_map: dict[str, str] = {} + current_id: str | None = None + current_names: list[str] = [] - current_channel_id = None - current_display_names = [] - - for event, elem in context: + with _open_xml(xml_path) as f: + for event, elem in ET.iterparse(f, events=("start", "end")): if event == "start" and elem.tag == "channel": - current_channel_id = elem.get("id") - current_display_names = [] - + current_id = elem.get("id") + current_names = [] elif event == "end" and elem.tag == "display-name": - if current_channel_id and elem.text: - current_display_names.append(elem.text.strip()) - + if current_id and elem.text: + current_names.append(elem.text.strip()) elif event == "end" and elem.tag == "channel": - if current_channel_id: - channel_ids.append(current_channel_id) - # Use the first display name as the primary name - if current_display_names: - channel_map[current_channel_id] = current_display_names[ - 0 - ] - else: - channel_map[current_channel_id] = current_channel_id - current_channel_id = None - - # Clear element to save memory + if current_id: + channel_ids.append(current_id) + channel_map[current_id] = current_names[0] if current_names else current_id + current_id = None if event == "end": elem.clear() - logger.info(f"Found {len(channel_ids)} channels in EPG") - - # Sort channels for consistent output sorted_channels = sorted(channel_ids) + logger.info(f"Found {len(sorted_channels)} channels in EPG") return { "channels": sorted_channels, - "channel_map": channel_map, # Map of id -> display name + "channel_map": channel_map, "count": len(sorted_channels), "source_url": service.epg_url, "cache_path": xml_path, "cache_size_bytes": file_size, } - except ET.ParseError as parse_err: - logger.error(f"XML parse error in EPG file: {parse_err}") + except ET.ParseError as e: + logger.error(f"XML parse error in EPG file: {e}") response.status = 500 return { - "error": f"Failed to parse EPG XML file: {str(parse_err)}", + "error": f"Failed to parse EPG XML file: {e}", "hint": "The EPG file may be malformed or not valid XMLTV format.", } except Exception as e: logger.error(f"Error getting EPG channels: {e}", exc_info=True) response.status = 500 - return {"error": f"Failed to process EPG file: {str(e)}"} + return {"error": f"Failed to process EPG file: {e}"} + + # ------------------------------------------------------------------ + # 3. Bulk XMLTV export (provider-scoped) + # ------------------------------------------------------------------ + + @app.route("/api/providers//epg", method="GET") + def get_provider_epg_xmltv(provider): + """Download the full XMLTV feed for a provider.""" + try: + response.content_type = "application/xml; charset=utf-8" + response.headers["Content-Disposition"] = ( + f'attachment; filename="{provider}_epg.xml"' + ) + + xmltv_data = manager.get_provider_epg_xmltv( + provider_name=provider, + country=request.query.get("country"), + ) + + if not xmltv_data: + response.status = 404 + return {"error": f'EPG data not available for provider "{provider}"'} + + return xmltv_data + + except ValueError as e: + logger.error(f"XMLTV EPG error for {provider}: {e}") + response.status = 404 + return {"error": str(e)} + except Exception as e: + logger.error(f"XMLTV EPG error for {provider}: {e}") + response.status = 500 + return {"error": f"Internal server error: {e}"} + + # ------------------------------------------------------------------ + # 4. Multi-channel grid view (NEW) + # ------------------------------------------------------------------ + + @app.route("/api/providers//epg/grid", method="GET") + def get_provider_epg_grid(provider): + """ + Return a time-windowed grid of programs across multiple channels. + + Query params: + start_time – window start (Unix timestamp or ISO 8601); defaults to now + end_time – window end (Unix timestamp or ISO 8601); defaults to now+3h + channels – comma-separated channel IDs to include (omit = all) + country – optional country filter forwarded to the provider + """ + try: + start_time = _parse_time_param(request.query.get("start_time")) + end_time = _parse_time_param(request.query.get("end_time")) + country = request.query.get("country") + + raw_channels = request.query.get("channels") + channel_ids = [c.strip() for c in raw_channels.split(",") if c.strip()] if raw_channels else None + + grid_data = epg_ops.get_provider_epg_grid( + provider_name=provider, + start_time=start_time, + end_time=end_time, + channel_ids=channel_ids, + country=country, + ) + + response.content_type = "application/json; charset=utf-8" + return { + "provider": provider, + "start_time": request.query.get("start_time"), + "end_time": request.query.get("end_time"), + "channels_count": len(grid_data), + "grid": grid_data, + } + + except ValueError as e: + logger.error(f"Grid EPG error for {provider}: {e}") + response.status = 404 + response.content_type = "application/json; charset=utf-8" + return {"error": str(e)} + except Exception as e: + logger.error(f"Grid EPG error for {provider}: {e}") + response.status = 500 + response.content_type = "application/json; charset=utf-8" + return {"error": f"Internal server error: {e}"} + + # ------------------------------------------------------------------ + # 5. Single-channel program list + # ------------------------------------------------------------------ + + @app.route("/api/providers//channels//epg", method="GET") + def get_channel_epg(provider, channel_id): + """ + Return programs for a single channel within a time window. + + Query params: + start_time / start – window start (Unix timestamp or ISO 8601) + end_time / end – window end (Unix timestamp or ISO 8601) + limit – max number of programs to return (default 100) + country – optional country filter + """ + try: + # Accept both old (start_time/end_time) and new (start/end) param names. + start_time = _parse_time_param( + request.query.get("start_time") or request.query.get("start") + ) + end_time = _parse_time_param( + request.query.get("end_time") or request.query.get("end") + ) + limit = request.query.get("limit", default=100, type=int) + country = request.query.get("country") + + programs = epg_ops.get_channel_epg( + provider_name=provider, + channel_id=channel_id, + start_time=start_time, + end_time=end_time, + limit=limit, + country=country, + ) + + response.content_type = "application/json; charset=utf-8" + return { + "provider": provider, + "channel_id": channel_id, + "start_time": request.query.get("start_time") or request.query.get("start"), + "end_time": request.query.get("end_time") or request.query.get("end"), + "count": len(programs), + "programs": programs, + } + + except ValueError as e: + logger.error(f"EPG error for {provider}/{channel_id}: {e}") + response.status = 404 + response.content_type = "application/json; charset=utf-8" + return {"error": str(e)} + except Exception as e: + logger.error(f"EPG error for {provider}/{channel_id}: {e}") + response.status = 500 + response.content_type = "application/json; charset=utf-8" + return {"error": f"Internal server error: {e}"} + + # ------------------------------------------------------------------ + # 6. Deep program detail (NEW) + # ------------------------------------------------------------------ + + @app.route("/api/providers//programs/", method="GET") + def get_program_details(provider, program_id): + """ + Return full metadata for a single program. + + Useful for detail views, catchup deep-links, and series grouping. + """ + try: + program_data = epg_ops.get_program_details( + provider_name=provider, + program_id=program_id, + ) + + if not program_data: + response.status = 404 + response.content_type = "application/json; charset=utf-8" + return {"error": f"Program '{program_id}' not found for provider '{provider}'"} + + response.content_type = "application/json; charset=utf-8" + return { + "provider": provider, + "program_id": program_id, + "details": program_data, + } + + except ValueError as e: + logger.error(f"Program detail error for {provider}/{program_id}: {e}") + response.status = 404 + response.content_type = "application/json; charset=utf-8" + return {"error": str(e)} + except Exception as e: + logger.error(f"Program detail error for {provider}/{program_id}: {e}") + response.status = 500 + response.content_type = "application/json; charset=utf-8" + return {"error": f"Internal server error: {e}"} + + # ------------------------------------------------------------------ + # 7. EPG channel mapping (GET + POST) + # ------------------------------------------------------------------ @app.route("/api/providers//epg-mapping", method="GET") def get_epg_mapping(provider): - """Get current EPG mapping for a provider""" + """Return the persisted channel→EPG ID mapping for a provider.""" try: from streaming_providers.base.utils.vfs import VFS except ImportError: - # If VFS is not available, return empty mapping return {"provider": provider, "mapping": {}, "exists": False} try: - mapping_file = f"{provider}_epg_mapping.json" vfs = VFS(addon_subdir="") + mapping_file = f"{provider}_epg_mapping.json" - if vfs.exists(mapping_file): - mapping_data = vfs.read_json(mapping_file) - if mapping_data: - # The file structure is: - # { - # "_provider_name": "...", - # "channel_id": {"epg_id": "...", "name": "..."} - # } + if not vfs.exists(mapping_file): + logger.info(f"No EPG mapping file found for {provider}") + return {"provider": provider, "mapping": {}, "exists": False} - # Extract mapping (skip internal fields starting with _) - internal_fields = [ - "_provider_name", - "_created_at", - "_updated_at", - "_version", - ] - actual_mapping = { - k: v - for k, v in mapping_data.items() - if k not in internal_fields - } + mapping_data = vfs.read_json(mapping_file) or {} - logger.info( - f"Loaded EPG mapping for {provider}: {len(actual_mapping)} channels" - ) - logger.debug(f"Sample mappings: {list(actual_mapping.items())[:3]}") + _internal = {"_provider_name", "_created_at", "_updated_at", "_version"} + actual_mapping = {k: v for k, v in mapping_data.items() if k not in _internal} - return { - "provider": provider, - "mapping": actual_mapping, - "exists": True, - } + logger.info(f"Loaded EPG mapping for {provider}: {len(actual_mapping)} channels") + logger.debug(f"Sample mappings: {list(actual_mapping.items())[:3]}") - # Return empty mapping if file doesn't exist - logger.info(f"No EPG mapping file found for {provider}") - return {"provider": provider, "mapping": {}, "exists": False} + return {"provider": provider, "mapping": actual_mapping, "exists": True} except Exception as e: - logger.error( - f"Error getting EPG mapping for {provider}: {e}", exc_info=True - ) + logger.error(f"Error getting EPG mapping for {provider}: {e}", exc_info=True) response.status = 500 - return {"error": f"Failed to load mapping: {str(e)}"} + return {"error": f"Failed to load mapping: {e}"} @app.route("/api/providers//epg-mapping", method="POST") def save_epg_mapping(provider): - """Save EPG mapping for a provider""" + """Persist a channel→EPG ID mapping for a provider.""" try: from streaming_providers.base.utils.vfs import VFS except ImportError: @@ -269,7 +427,6 @@ def setup_epg_routes(app, manager, service): return {"error": "VFS module not available"} try: - # Get JSON data from request body using Bottle's request object try: mapping_data = request.json.get("mapping", {}) if request.json else {} except Exception as json_err: @@ -277,125 +434,49 @@ def setup_epg_routes(app, manager, service): response.status = 400 return {"error": "Invalid JSON in request body"} - # Get provider label if available + # Resolve provider display label if possible. provider_label = provider try: - provider_instance = manager.get_provider(provider) - if provider_instance: - provider_label = getattr( - provider_instance, "provider_label", provider - ) - except: + inst = manager.get_provider(provider) + if inst: + provider_label = getattr(inst, "provider_label", provider) + except Exception: pass - # Build the file structure: - # { - # "_provider_name": "Provider Label", - # "channel_id": {"epg_id": "...", "name": "..."} - # } - full_mapping = {"_provider_name": provider_label} + full_mapping: dict = {"_provider_name": provider_label} + for channel_id, value in mapping_data.items(): + if isinstance(value, dict): + full_mapping[channel_id] = value + elif isinstance(value, str): + full_mapping[channel_id] = {"epg_id": value, "name": ""} - # Add each mapping entry - for channel_id, mapping_value in mapping_data.items(): - if isinstance(mapping_value, dict): - # Already has structure {"epg_id": "...", "name": "..."} - full_mapping[channel_id] = mapping_value - elif isinstance(mapping_value, str): - # Simple string, convert to object - full_mapping[channel_id] = { - "epg_id": mapping_value, - "name": "", # Name not provided - } - - # Save to file vfs = VFS(addon_subdir="") mapping_file = f"{provider}_epg_mapping.json" - success = vfs.write_json(mapping_file, full_mapping) - if success: - logger.info( - f"Saved EPG mapping for {provider}: {len(mapping_data)} channels" - ) - - # Clear mapping cache if it exists - try: - from streaming_providers.base.epg.epg_mapping import EPGMapping - - mapping_manager = EPGMapping() - mapping_manager.reload_mapping(provider) - logger.info(f"Reloaded EPG mapping cache for {provider}") - except ImportError: - logger.debug("EPGMapping not available for cache reload") - pass - except Exception as reload_err: - logger.warning(f"Could not reload mapping cache: {reload_err}") - - return { - "success": True, - "message": f"Mapping saved for {provider}", - "channels_mapped": len(mapping_data), - } - else: + if not success: response.status = 500 return {"error": "Failed to save mapping file"} + logger.info(f"Saved EPG mapping for {provider}: {len(mapping_data)} channels") + + # Best-effort cache invalidation. + try: + from streaming_providers.base.epg.epg_mapping import EPGMapping + EPGMapping().reload_mapping(provider) + logger.info(f"Reloaded EPG mapping cache for {provider}") + except ImportError: + logger.debug("EPGMapping not available for cache reload") + except Exception as reload_err: + logger.warning(f"Could not reload mapping cache: {reload_err}") + + return { + "success": True, + "message": f"Mapping saved for {provider}", + "channels_mapped": len(mapping_data), + } + except Exception as e: logger.error(f"Error saving EPG mapping for {provider}: {e}", exc_info=True) response.status = 500 - return {"error": f"Failed to save mapping: {str(e)}"} - - @app.route("/api/providers//channels//epg", method="GET") - def get_channel_epg(provider, channel_id): - """Get EPG data for a specific channel.""" - # Query parameters (Unix timestamps) - start = request.query.get("start") - end = request.query.get("end") - - # Optional: limit results - limit = request.query.get("limit", default=100, type=int) - - try: - from datetime import datetime, timezone - - # Convert timestamps to datetime objects - start_dt = None - end_dt = None - - if start: - try: - start_dt = datetime.fromtimestamp(int(start), tz=timezone.utc) - except (ValueError, TypeError) as e: - logger.warning(f"Invalid start timestamp: {start}, error: {e}") - - if end: - try: - end_dt = datetime.fromtimestamp(int(end), tz=timezone.utc) - except (ValueError, TypeError) as e: - logger.warning(f"Invalid end timestamp: {end}, error: {e}") - - # Re-use the shared epg_ops instance - programs = epg_ops.get_channel_epg( - provider_name=provider, - channel_id=channel_id, - start_time=start_dt, # ← Now passing datetime, not int - end_time=end_dt, # ← Now passing datetime, not int - limit=limit - ) - - return { - "provider": provider, - "channel_id": channel_id, - "start_time": start, - "end_time": end, - "count": len(programs), - "programs": programs - } - - except ValueError as e: - response.status = 404 - return {"error": str(e)} - except Exception as e: - logger.error(f"Error getting EPG: {e}", exc_info=True) - response.status = 500 - return {"error": str(e)} \ No newline at end of file + return {"error": f"Failed to save mapping: {e}"} \ No newline at end of file diff --git a/routes/streams/__init__.py b/routes/streams/__init__.py index 0f1c879..95a8cc5 100644 --- a/routes/streams/__init__.py +++ b/routes/streams/__init__.py @@ -49,14 +49,12 @@ def setup_stream_routes(app, manager, service): from .events import setup_event_routes from .vod import setup_vod_routes from .recordings import setup_recording_routes - from .epg import setup_epg_routes # Pass manager and service to all submodules setup_channel_routes(app, manager, service) setup_event_routes(app, manager, service) setup_vod_routes(app, manager, service) setup_recording_routes(app, manager, service) - setup_epg_routes(app, manager, service) # ========================================================================= # ORIGINAL MANIFEST ENDPOINT (shared across content types) diff --git a/routes/streams/epg.py b/routes/streams/epg.py deleted file mode 100644 index c86c4b7..0000000 --- a/routes/streams/epg.py +++ /dev/null @@ -1,95 +0,0 @@ -#!/usr/bin/env python3 -""" -EPG (Electronic Program Guide) routes. - -These routes provide program schedule information for channels. -""" - -from datetime import datetime, timezone -from bottle import request, response -from streaming_providers.base.utils import logger - - -def setup_epg_routes(app, manager, service): - """Setup all EPG-related routes.""" - - # Note: EPG routes don't need the full helpers factory since they don't - # use the stream resolution helpers. They only need manager. - - @app.route("/api/providers//channels//epg") - def get_channel_epg(provider, channel_id): - try: - kwargs = {"country": request.query.get("country")} - - if request.query.get("start_time"): - start_time_str = request.query.get("start_time") - try: - kwargs["start_time"] = datetime.fromtimestamp( - int(start_time_str), tz=timezone.utc - ) - except (ValueError, TypeError): - try: - kwargs["start_time"] = datetime.fromisoformat( - start_time_str.replace("Z", "+00:00") - ) - except ValueError: - logger.warning(f"Invalid start_time format: {start_time_str}") - - if request.query.get("end_time"): - end_time_str = request.query.get("end_time") - try: - kwargs["end_time"] = datetime.fromtimestamp( - int(end_time_str), tz=timezone.utc - ) - except (ValueError, TypeError): - try: - kwargs["end_time"] = datetime.fromisoformat( - end_time_str.replace("Z", "+00:00") - ) - except ValueError: - logger.warning(f"Invalid end_time format: {end_time_str}") - - epg_data = manager.get_channel_epg( - provider_name=provider, channel_id=channel_id, **kwargs - ) - - response.content_type = "application/json; charset=utf-8" - return {"provider": provider, "channel_id": channel_id, "epg": epg_data} - - except ValueError as e: - logger.error(f"EPG error for {provider}/{channel_id}: {e}") - response.status = 404 - response.content_type = "application/json; charset=utf-8" - return {"error": str(e)} - except Exception as e: - logger.error(f"EPG error for {provider}/{channel_id}: {e}") - response.status = 500 - response.content_type = "application/json; charset=utf-8" - return {"error": f"Internal server error: {str(e)}"} - - @app.route("/api/providers//epg") - def get_provider_epg_xmltv(provider): - try: - response.content_type = "application/xml; charset=utf-8" - response.headers["Content-Disposition"] = ( - f'attachment; filename="{provider}_epg.xml"' - ) - - xmltv_data = manager.get_provider_epg_xmltv( - provider_name=provider, country=request.query.get("country") - ) - - if not xmltv_data: - response.status = 404 - return {"error": f'EPG data not available for provider "{provider}"'} - - return xmltv_data - - except ValueError as e: - logger.error(f"XMLTV EPG error for {provider}: {e}") - response.status = 404 - return {"error": str(e)} - except Exception as e: - logger.error(f"XMLTV EPG error for {provider}: {e}") - response.status = 500 - return {"error": f"Internal server error: {str(e)}"} \ No newline at end of file