diff --git a/lib/streaming_providers/providers/hrti/epg_manager.py b/lib/streaming_providers/providers/hrti/epg_manager.py new file mode 100644 index 0000000..61e2b26 --- /dev/null +++ b/lib/streaming_providers/providers/hrti/epg_manager.py @@ -0,0 +1,384 @@ +# streaming_providers/providers/hrti/epg_manager.py +""" +HRTi EPG Manager - Handles Electronic Program Guide operations. +""" + +import json +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Any, Tuple + +from ...base.models.epg_models import EPGEntry +from ...base.utils.logger import logger + + +class HRTiEPGManager: + """Manages HRTi EPG operations.""" + + # Default EPG window: 2 days past, 7 days future + DEFAULT_PAST_DAYS = 2 + DEFAULT_FUTURE_DAYS = 7 + + def __init__(self, provider): + self.provider = provider + self._config = provider.hrti_config + self._http_manager = provider.http_manager + self._authenticator = provider.authenticator + self._channels_cache = None + + def _get_headers(self, referer_path: str = "/tv") -> Dict[str, str]: + """Build authenticated headers for EPG API calls.""" + headers = { + "deviceid": self._authenticator.get_device_id(), + "devicetypeid": self._config.device_reference_id, + "ipaddress": self._authenticator.get_ip_address(), + "operatorreferenceid": self._config.operator_reference_id, + "origin": self._config.base_website, + "referer": f"{self._config.base_website}{referer_path}", + "User-Agent": self._config.user_agent, + "Accept": "application/json", + "Content-Type": "application/json", + } + token = ( + self._authenticator._current_token.access_token + if self._authenticator._current_token + else "" + ) + if token: + headers["authorization"] = f"Client {token}" + return headers + + def get_epg_for_channel( + self, + channel_id: str, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + ) -> List[EPGEntry]: + """ + Get EPG data for a single channel. + + Args: + channel_id: Channel reference ID (e.g., "40013") + start_time: Start time for EPG query (default: now - DEFAULT_PAST_DAYS days) + end_time: End time for EPG query (default: now + DEFAULT_FUTURE_DAYS days) + + Returns: + List of EPGEntry objects + """ + # Set default time range if not provided + now = datetime.now() + if start_time is None: + start_time = now - timedelta(days=self.DEFAULT_PAST_DAYS) + if end_time is None: + end_time = now + timedelta(days=self.DEFAULT_FUTURE_DAYS) + + # Convert to milliseconds timestamp for API + start_ts = int(start_time.timestamp() * 1000) + end_ts = int(end_time.timestamp() * 1000) + + try: + headers = self._get_headers() + payload = { + "ChannelReferenceIds": [channel_id], + "StartTime": f"/Date({start_ts})/", + "EndTime": f"/Date({end_ts})/", + } + + logger.debug(f"Fetching EPG for channel {channel_id}: {start_time} to {end_time}") + + response = self._http_manager.post( + self._config.api_endpoints["programme"], + operation="api", + headers=headers, + data=json.dumps(payload), + ) + response.raise_for_status() + + data = response.json() + if data.get("ErrorCode", 0) != 0: + logger.warning(f"EPG API error: {data.get('ErrorDescription')}") + return [] + + result = data.get("Result", []) + if not result: + return [] + + # Find the channel in the result + channel_data = None + for item in result: + if item.get("ReferenceID") == channel_id: + channel_data = item + break + + if not channel_data: + logger.debug(f"No EPG data found for channel {channel_id}") + return [] + + epg_list = channel_data.get("EpgList", []) + + # Convert to EPGEntry objects + entries = [] + for epg_item in epg_list: + entry = self._parse_epg_item(epg_item, channel_id) + if entry: + entries.append(entry) + + logger.debug(f"Retrieved {len(entries)} EPG entries for channel {channel_id}") + return entries + + except Exception as e: + logger.error(f"Error fetching EPG for channel {channel_id}: {e}") + return [] + + def get_epg_for_all_channels( + self, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + ) -> Dict[str, List[EPGEntry]]: + """ + Get EPG data for all channels in a single API call. + + Note: HRTi API supports fetching EPG for all channels at once, + which is more efficient than per-channel calls. + + Returns: + Dictionary mapping channel_id -> list of EPGEntry objects + """ + # Set default time range + now = datetime.now() + if start_time is None: + start_time = now - timedelta(days=self.DEFAULT_PAST_DAYS) + if end_time is None: + end_time = now + timedelta(days=self.DEFAULT_FUTURE_DAYS) + + start_ts = int(start_time.timestamp() * 1000) + end_ts = int(end_time.timestamp() * 1000) + + # Get all channel IDs from provider + channels = self._get_all_channels() + if not channels: + return {} + + channel_ids = [ch.content_id for ch in channels] + + try: + headers = self._get_headers() + payload = { + "ChannelReferenceIds": channel_ids, + "StartTime": f"/Date({start_ts})/", + "EndTime": f"/Date({end_ts})/", + } + + logger.debug(f"Fetching EPG for {len(channel_ids)} channels") + + response = self._http_manager.post( + self._config.api_endpoints["programme"], + operation="api", + headers=headers, + data=json.dumps(payload), + ) + response.raise_for_status() + + data = response.json() + if data.get("ErrorCode", 0) != 0: + logger.warning(f"EPG API error: {data.get('ErrorDescription')}") + return {} + + result = data.get("Result", []) + + # Parse results per channel + epg_by_channel = {} + for channel_data in result: + channel_id = channel_data.get("ReferenceID") + if not channel_id: + continue + + epg_list = channel_data.get("EpgList", []) + entries = [] + for epg_item in epg_list: + entry = self._parse_epg_item(epg_item, channel_id) + if entry: + entries.append(entry) + + if entries: + epg_by_channel[channel_id] = entries + + logger.debug(f"Retrieved EPG for {len(epg_by_channel)} channels") + return epg_by_channel + + except Exception as e: + logger.error(f"Error fetching EPG for all channels: {e}") + return {} + + def get_epg_details(self, reference_id: str) -> Optional[Dict[str, Any]]: + """ + Get detailed information for a specific EPG entry. + + This is useful for catch-up functionality as it provides the + FileName (streaming URL) for the recorded program. + + Args: + reference_id: EPG entry reference ID + + Returns: + Dictionary with EPG details including FileName + """ + try: + headers = self._get_headers("/tv") + payload = { + "ChannelReferenceId": "", # Not required, but API expects it + "ReferenceId": reference_id, + } + + response = self._http_manager.post( + self._config.api_endpoints["epg_details"], + operation="api", + headers=headers, + data=json.dumps(payload), + ) + response.raise_for_status() + + data = response.json() + if data.get("ErrorCode", 0) != 0: + logger.warning(f"EPG details API error: {data.get('ErrorDescription')}") + return None + + return data.get("Result") + + except Exception as e: + logger.error(f"Error fetching EPG details for {reference_id}: {e}") + return None + + def _parse_epg_item(self, item: Dict[str, Any], channel_id: str) -> Optional[EPGEntry]: + """ + Parse a single EPG item from API response into EPGEntry. + + Expected item format from GetProgramme: + { + "AdultContent": false, + "CategoryReferenceId": "VIJESTI", + "ContentRating": "", + "ImagePath": "https://...", + "ReferenceID": "2059396656", + "TimeEnd": "/Date(1777725000000)/", + "TimeEndUnixEpoch": 1777725000, + "TimeStart": "/Date(1777723200000)/", + "TimeStartUnixEpoch": 1777723200, + "Title": "Vijesti" + } + """ + try: + # Extract timestamps (prefer Unix epoch, fallback to parsing /Date() format) + start = item.get("TimeStartUnixEpoch") + if start is None: + start_str = item.get("TimeStart", "") + start = self._parse_date_string(start_str) + + end = item.get("TimeEndUnixEpoch") + if end is None: + end_str = item.get("TimeEnd", "") + end = self._parse_date_string(end_str) + + if not start or not end: + logger.warning(f"Missing timestamps for EPG item: {item.get('ReferenceID')}") + return None + + title = item.get("Title", "").strip() + if not title: + return None + + # Generate broadcast ID (32-bit unique identifier) + broadcast_id = EPGEntry.encode_broadcast_id( + self.provider.provider_name, + channel_id, + start + ) + + # Get genre type from CategoryReferenceId + genre_type = self._map_category_to_genre(item.get("CategoryReferenceId", "")) + + return EPGEntry( + broadcast_id=broadcast_id, + title=title, + start=start, + end=end, + description=item.get("DescriptionShort") or item.get("DescriptionLong"), + icon=item.get("ImagePath"), + genre=genre_type, + genre_description=item.get("CategoryReferenceId"), + parental_rating=self._parse_parental_rating(item.get("ContentRating")), + flags=EPGFlags.IS_SERIES if item.get("IsSeries", False) else EPGFlags.UNDEFINED, + ) + + except Exception as e: + logger.error(f"Error parsing EPG item: {e}") + return None + + @staticmethod + def _parse_date_string(date_str: str) -> Optional[int]: + """Parse /Date(timestamp)/ format to Unix timestamp.""" + if not date_str: + return None + + import re + match = re.search(r'/Date\((\d+)\)/', date_str) + if match: + # Convert milliseconds to seconds + return int(match.group(1)) // 1000 + + return None + + @staticmethod + def _map_category_to_genre(category: str) -> int: + """ + Map HRTi category to EPG genre type. + Based on ETSI EN 300 468 DVB-SI standard. + """ + category_map = { + "VIJESTI": EPGGenre.NEWSCURRENTAFFAIRS, # News/Current Affairs + "SPORT": EPGGenre.SPORTS, # Sports + "GLAZBA": EPGGenre.MUSICBALLETDANCE, # Music/Ballet/Dance + "IGRANI PROGRAM": EPGGenre.MOVIEDRAMA, # Movie/Drama + "UMJETNOST I KULTURA": EPGGenre.ARTSCULTURE, # Arts/Culture + "ZABAVA": EPGGenre.SHOW, # Show/Game Show + "RELIGIJA": EPGGenre.SOCIALPOLITICALECONOMICS, # Religion + "OBRAZOVANJE": EPGGenre.EDUCATIONALSCIENCE, # Educational + "ZNANOST": EPGGenre.EDUCATIONALSCIENCE, # Science + "DRUGI INFORMATIVNI SADRŽAJI": EPGGenre.NEWSCURRENTAFFAIRS, + "PROMOCIJA": EPGGenre.SHOW, + "KOMERCIJALNI PROGRAMI": EPGGenre.SHOW, + } + + return category_map.get(category, EPGGenre.UNDEFINED) + + @staticmethod + def _parse_parental_rating(rating_str: str) -> Optional[int]: + """Parse parental rating string to integer.""" + if not rating_str: + return None + + # Try to extract number from strings like "12", "FSK 12", "12+" + import re + match = re.search(r'(\d+)', rating_str) + if match: + return int(match.group(1)) + + return None + + def _get_all_channels(self): + """Get all channels from provider (with caching).""" + if self._channels_cache is None: + self._channels_cache = self.provider.get_channels() + return self._channels_cache + + def get_epg_window(self) -> Tuple[int, int]: + """ + Return the EPG window as (past_days, future_days). + + Returns: + Tuple of (past_days, future_days) + """ + return self.DEFAULT_PAST_DAYS, self.DEFAULT_FUTURE_DAYS + + +# Import EPG models at the bottom to avoid circular imports +from ...base.models.epg_models import EPGEntry, EPGGenre, EPGFlags \ No newline at end of file diff --git a/lib/streaming_providers/providers/hrti/provider.py b/lib/streaming_providers/providers/hrti/provider.py index 85060af..142d4aa 100644 --- a/lib/streaming_providers/providers/hrti/provider.py +++ b/lib/streaming_providers/providers/hrti/provider.py @@ -3,7 +3,7 @@ import json import datetime import traceback from urllib.parse import urlparse -from typing import ClassVar, Dict, List, Optional, Union +from typing import ClassVar, Dict, List, Optional, Union, Tuple import requests @@ -15,6 +15,7 @@ from ..lib_drmtoday import create_drmtoday_widevine_config from .auth import HRTiAuthenticator from .constants import HRTiConfig, HRTiDefaults from .vod_manager import HRTiVodManager +from .epg_manager import HRTiEPGManager # Navigation-only prefixes — these content_ids are category/series containers. # They are never directly playable and must never be sent to get_manifest/get_drm. @@ -77,6 +78,9 @@ class HRTiProvider(StreamingProvider): # Initialize VOD manager self._vod_manager = HRTiVodManager(self) + # Initialize EPG manager + self._epg_manager = HRTiEPGManager(self) + try: self.authenticator.get_bearer_token() logger.debug("HRTi authentication successful during initialization") @@ -527,50 +531,19 @@ class HRTiProvider(StreamingProvider): # ============================================================================ def get_epg(self, channel_id: str, **kwargs) -> List[Dict]: - """Get EPG data for a channel.""" - try: - headers = self._get_hrti_authenticated_headers() - start_time = self.authenticator.get_time_offset(-4) - end_time = self.authenticator.get_time_offset(4) + """ + Get EPG data for a channel. - payload = { - "ChannelReferenceIds": [channel_id], - "StartTime": f"/Date({start_time})/", - "EndTime": f"/Date({end_time})/", - } + Returns list of dictionaries for backward compatibility. + """ + entries = self._epg_manager.get_epg_for_channel(channel_id) + # Convert EPGEntry objects to dict for backward compatibility + return [entry.to_dict() for entry in entries] - response = self.http_manager.post( - self.hrti_config.api_endpoints["programme"], - operation="api", - headers=headers, - data=json.dumps(payload), - ) - response.raise_for_status() - - epg_data = response.json() - if "Result" in epg_data: - return [ - { - "title": entry.get("Title", ""), - "description": entry.get("Description", ""), - "start": entry.get("StartTime", ""), - "end": entry.get("EndTime", ""), - "genre": entry.get("Genre", ""), - } - for entry in epg_data["Result"] - ] - return [] - - except Exception as e: - logger.error(f"Error getting EPG data for channel {channel_id}: {e}") - return [] - - def get_license_url(self, channel: StreamingChannel, **kwargs) -> Optional[str]: - """Return the Widevine license server URL for *channel*.""" - drm_configs = self.get_drm(channel.channel_id, **kwargs) - if drm_configs: - return drm_configs[0].license.server_url - return None + @property + def epg_window(self) -> Tuple[int, int]: + """Return EPG window as (past_days, future_days).""" + return self._epg_manager.get_epg_window() @property def catchup_window(self) -> int: