#!/usr/bin/env python3 # streaming_providers/base/epg/epg_parser.py """ XMLTV EPG Parser for Kodi PVR Backend Parses XMLTV format and converts to Kodi PVR EPG format """ import xml.etree.ElementTree as ET import gzip import hashlib from datetime import datetime from typing import List, Dict, Optional, Tuple from ..utils.logger import logger class EPGParser: """ Parses XMLTV EPG data and converts to Kodi PVR format. Uses streaming parse for performance with large EPG files. """ @staticmethod def parse_xmltv_time(time_str: str) -> Optional[int]: """ Parse XMLTV time format to Unix timestamp. Format: yyyyMMddHHmmss +HHMM Example: 20240101200000 +0100 Args: time_str: XMLTV formatted time string Returns: Unix timestamp or None if parsing fails """ if not time_str: return None try: # Split timestamp and timezone parts = time_str.strip().split() if not parts: return None dt_str = parts[0] # Parse basic timestamp dt = datetime.strptime(dt_str, '%Y%m%d%H%M%S') # Handle timezone offset if present if len(parts) > 1: tz_str = parts[1] if tz_str.startswith(('+', '-')): sign = 1 if tz_str[0] == '+' else -1 hours = int(tz_str[1:3]) minutes = int(tz_str[3:5]) if len(tz_str) > 3 else 0 # Adjust for timezone (subtract offset to get UTC) from datetime import timedelta dt = dt - timedelta(hours=sign * hours, minutes=sign * minutes) return int(dt.timestamp()) except (ValueError, IndexError) as e: logger.warning(f"Failed to parse XMLTV time '{time_str}': {e}") return None @staticmethod def generate_broadcast_id(channel_id: str, start_time: int) -> int: """ Generate deterministic broadcast ID from channel and start time. Args: channel_id: EPG channel ID start_time: Unix timestamp of programme start Returns: Unique integer broadcast ID """ # Create hash from channel_id and start_time hash_input = f"{channel_id}_{start_time}".encode('utf-8') hash_digest = hashlib.sha256(hash_input).hexdigest() # Convert first 8 hex chars to int (32-bit) return int(hash_digest[:8], 16) @staticmethod def parse_episode_num(episode_elem: ET.Element) -> Tuple[Optional[int], Optional[int], Optional[int]]: """ Parse episode numbering from XMLTV episode-num element. Supports two systems: - xmltv_ns: "season.episode.part/total" (0-indexed, e.g., "4.11.0/1" = S05E12) - onscreen: "S05E12" format Args: episode_elem: episode-num XML element Returns: Tuple of (series_number, episode_number, episode_part_number) """ system = episode_elem.get('system', '') text = episode_elem.text or '' if system == 'xmltv_ns': # Format: "season.episode.part/total_parts" # Example: "4.11.0/1" means season 5, episode 12, part 1 of 1 try: parts = text.split('.') if len(parts) >= 2: series_num = int(parts[0].split('/')[0]) + 1 if parts[0] else None # 0-indexed, so +1 episode_num = int(parts[1].split('/')[0]) + 1 if parts[1] else None part_num = None if len(parts) >= 3 and parts[2]: part_str = parts[2].split('/')[0] part_num = int(part_str) + 1 if part_str else None return series_num, episode_num, part_num except (ValueError, IndexError) as e: logger.debug(f"Failed to parse xmltv_ns episode number '{text}': {e}") elif system == 'onscreen': # Format: "S05E12" or "5x12" try: import re # Try S##E## format match = re.match(r'S(\d+)E(\d+)', text, re.IGNORECASE) if match: return int(match.group(1)), int(match.group(2)), None # Try #x## format match = re.match(r'(\d+)x(\d+)', text) if match: return int(match.group(1)), int(match.group(2)), None except (ValueError, AttributeError) as e: logger.debug(f"Failed to parse onscreen episode number '{text}': {e}") return None, None, None @staticmethod def parse_star_rating(rating_elem: ET.Element) -> Optional[int]: """ Parse star rating from XMLTV format to 0-10 scale. Args: rating_elem: star-rating XML element Returns: Rating as integer 0-10, or None """ value_elem = rating_elem.find('value') if value_elem is not None and value_elem.text: try: # Format can be "8/10" or "4/5" or just "8" text = value_elem.text.strip() if '/' in text: parts = text.split('/') numerator = float(parts[0]) denominator = float(parts[1]) # Normalize to 0-10 scale rating = int((numerator / denominator) * 10) return min(10, max(0, rating)) else: # Assume already 0-10 scale rating = int(float(text)) return min(10, max(0, rating)) except (ValueError, IndexError) as e: logger.debug(f"Failed to parse star rating '{value_elem.text}': {e}") return None @staticmethod def parse_programme(programme_elem: ET.Element, epg_channel_id: str) -> Optional[Dict]: """ Parse a single XMLTV programme element to Kodi EPG format. Args: programme_elem: XML programme element epg_channel_id: EPG channel ID for validation Returns: Dictionary with EPG data, or None if required fields missing """ # Verify channel matches channel = programme_elem.get('channel', '') if channel != epg_channel_id: return None # Parse required fields start_str = programme_elem.get('start', '') stop_str = programme_elem.get('stop', '') start_time = EPGParser.parse_xmltv_time(start_str) end_time = EPGParser.parse_xmltv_time(stop_str) if not start_time or not end_time: logger.debug(f"Programme missing valid start/end time: start={start_str}, stop={stop_str}") return None # Get title title_elem = programme_elem.find('title') title = title_elem.text if title_elem is not None and title_elem.text else "Unknown" # Generate broadcast ID broadcast_id = EPGParser.generate_broadcast_id(epg_channel_id, start_time) # Build base EPG entry with required fields epg_entry = { 'broadcast_id': broadcast_id, 'title': title.strip(), 'start_time': start_time, 'end_time': end_time } # Parse optional fields # Sub-title (episode name) subtitle_elem = programme_elem.find('sub-title') if subtitle_elem is not None and subtitle_elem.text: epg_entry['episode_name'] = subtitle_elem.text.strip() # Description (plot) desc_elem = programme_elem.find('desc') if desc_elem is not None and desc_elem.text: plot = desc_elem.text.strip() epg_entry['plot'] = plot # Use first sentence or first 100 chars as plot_outline outline = plot.split('.')[0] if '.' in plot else plot[:100] epg_entry['plot_outline'] = outline.strip() # Credits credits_elem = programme_elem.find('credits') if credits_elem is not None: # Director director_elem = credits_elem.find('director') if director_elem is not None and director_elem.text: epg_entry['director'] = director_elem.text.strip() # Actors (join with comma) actors = credits_elem.findall('actor') if actors: cast_list = [actor.text.strip() for actor in actors if actor.text] if cast_list: epg_entry['cast'] = ', '.join(cast_list) # Writer writer_elem = credits_elem.find('writer') if writer_elem is not None and writer_elem.text: epg_entry['writer'] = writer_elem.text.strip() # Year/Date date_elem = programme_elem.find('date') if date_elem is not None and date_elem.text: try: year = int(date_elem.text.strip()[:4]) # Take first 4 chars epg_entry['year'] = year except ValueError: pass # Category (genre) category_elem = programme_elem.find('category') if category_elem is not None and category_elem.text: epg_entry['genre_description'] = category_elem.text.strip() # Icon icon_elem = programme_elem.find('icon') if icon_elem is not None: icon_src = icon_elem.get('src', '') if icon_src: epg_entry['icon_path'] = icon_src # Episode numbering (try all episode-num elements) episode_nums = programme_elem.findall('episode-num') series_num = episode_num = part_num = None # Try xmltv_ns first for ep_elem in episode_nums: if ep_elem.get('system') == 'xmltv_ns': series_num, episode_num, part_num = EPGParser.parse_episode_num(ep_elem) break # Fall back to onscreen if xmltv_ns didn't work if series_num is None: for ep_elem in episode_nums: if ep_elem.get('system') == 'onscreen': series_num, episode_num, part_num = EPGParser.parse_episode_num(ep_elem) break if series_num is not None: epg_entry['series_number'] = series_num if episode_num is not None: epg_entry['episode_number'] = episode_num if part_num is not None: epg_entry['episode_part_number'] = part_num # Star rating star_rating_elem = programme_elem.find('star-rating') if star_rating_elem is not None: rating = EPGParser.parse_star_rating(star_rating_elem) if rating is not None: epg_entry['star_rating'] = rating # Parental rating rating_elems = programme_elem.findall('rating') for rating_elem in rating_elems: value_elem = rating_elem.find('value') if value_elem is not None and value_elem.text: try: # Try to extract numeric rating import re match = re.search(r'\d+', value_elem.text) if match: epg_entry['parental_rating'] = int(match.group()) break except (ValueError, AttributeError): pass # Previously shown (first aired) prev_shown_elem = programme_elem.find('previously-shown') if prev_shown_elem is not None: first_aired_str = prev_shown_elem.get('start', '') if first_aired_str: first_aired = EPGParser.parse_xmltv_time(first_aired_str) if first_aired: epg_entry['first_aired'] = first_aired return epg_entry @staticmethod def open_xml_file(file_path: str): """ Open XML file, handling both plain and gzipped files. Args: file_path: Path to XML file (can be .xml or .xml.gz) Returns: File object suitable for ET.iterparse """ if file_path.endswith('.gz'): logger.debug(f"Opening gzipped EPG file: {file_path}") return gzip.open(file_path, 'rt', encoding='utf-8') else: logger.debug(f"Opening plain EPG file: {file_path}") return open(file_path, 'r', encoding='utf-8') @staticmethod def parse_epg_for_channel( xml_path: str, epg_channel_id: str, start_time: Optional[int] = None, end_time: Optional[int] = None ) -> List[Dict]: """ Parse EPG data for a specific channel within a time range. Uses streaming parse for performance. Args: xml_path: Path to XMLTV file (can be .xml or .xml.gz) epg_channel_id: EPG channel ID to filter for start_time: Start of time range (Unix timestamp), None for no lower bound end_time: End of time range (Unix timestamp), None for no upper bound Returns: List of EPG entries as dictionaries """ logger.info(f"Parsing EPG for channel '{epg_channel_id}' from {xml_path}") programmes = [] try: with EPGParser.open_xml_file(xml_path) as xml_file: # Use iterparse for memory-efficient streaming context = ET.iterparse(xml_file, events=('end',)) for event, elem in context: if elem.tag == 'programme': # Check if this programme matches our channel if elem.get('channel') == epg_channel_id: # Parse the programme epg_entry = EPGParser.parse_programme(elem, epg_channel_id) if epg_entry: # Filter by time range prog_start = epg_entry['start_time'] prog_end = epg_entry['end_time'] # Include if programme overlaps with requested range if start_time is not None and prog_end < start_time: # Programme ends before requested range elem.clear() continue if end_time is not None and prog_start > end_time: # Programme starts after requested range # Since XMLTV is chronological, we can stop here elem.clear() break programmes.append(epg_entry) # Clear element to free memory elem.clear() logger.info(f"Parsed {len(programmes)} programmes for channel '{epg_channel_id}'") return programmes except ET.ParseError as e: logger.error(f"XML parse error in EPG file: {e}") return [] except Exception as e: logger.error(f"Error parsing EPG file: {e}") return []