nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-16 06:02:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

EPG: reject out of window request

Browse Source
This commit is contained in:
Nirvana
2026-06-30 12:03:17 +02:00
parent 63ad335535
commit f44f0d6dfe
2 changed files with 150 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/base/epg_operations.py
+125 -37
View File
@@ -5,12 +5,26 @@ EPG-related operations.
"""
from typing import Dict, List, Optional
from datetime import datetime, timedelta, timezone
from .epg import EPGManager
from .utils.logger import logger
from .models.epg_models import EPGEntry, EPGProgramDetails
class EPGWindowError(Exception):
"""Raised when EPG request is outside the valid time window."""
def __init__(self, message, requested_start=None, requested_end=None,
min_allowed=None, max_allowed=None, provider_window_days=None):
self.message = message
self.requested_start = requested_start
self.requested_end = requested_end
self.min_allowed = min_allowed
self.max_allowed = max_allowed
self.provider_window_days = provider_window_days
super().__init__(message)
class EPGOperations:
"""Handles all EPG-related operations.
@@ -36,32 +50,112 @@ class EPGOperations:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
return provider
def _validate_epg_window(self, provider, start_time, end_time):
"""
Validate EPG time window against provider's supported window.
Raises:
EPGWindowError: If the requested window is outside the valid range.
Returns:
Tuple of (start_time, end_time) validated and clamped to provider's window.
"""
if not start_time and not end_time:
# If no times provided, use provider's default (usually today)
return start_time, end_time
# Get provider's EPG window (days back, days forward)
epg_window = getattr(provider, 'epg_window', (7, 7))
days_back, days_forward = epg_window
now_utc = datetime.now(tz=timezone.utc)
min_allowed = now_utc - timedelta(days=days_back)
max_allowed = now_utc + timedelta(days=days_forward)
# Ensure times are timezone-aware
if start_time and start_time.tzinfo is None:
start_time = start_time.replace(tzinfo=timezone.utc)
if end_time and end_time.tzinfo is None:
end_time = end_time.replace(tzinfo=timezone.utc)
# Check if the requested window is completely outside the valid range
if start_time and end_time:
# If the entire window is outside the valid range
if end_time < min_allowed or start_time > max_allowed:
raise EPGWindowError(
message="Requested EPG time window is outside the valid range",
requested_start=start_time.isoformat(),
requested_end=end_time.isoformat(),
min_allowed=min_allowed.isoformat(),
max_allowed=max_allowed.isoformat(),
provider_window_days=days_back
)
# Clamp individual times if they're partially outside
if start_time:
if start_time < min_allowed:
logger.warning(
f"EPG start_time {start_time} is before the minimum allowed "
f"({min_allowed}). Clamping to {min_allowed}"
)
start_time = min_allowed
elif start_time > max_allowed:
logger.warning(
f"EPG start_time {start_time} is after the maximum allowed "
f"({max_allowed}). Clamping to {max_allowed}"
)
start_time = max_allowed
if end_time:
if end_time < min_allowed:
logger.warning(
f"EPG end_time {end_time} is before the minimum allowed "
f"({min_allowed}). Clamping to {min_allowed}"
)
end_time = min_allowed
elif end_time > max_allowed:
logger.warning(
f"EPG end_time {end_time} is after the maximum allowed "
f"({max_allowed}). Clamping to {max_allowed}"
)
end_time = max_allowed
# Ensure start_time <= end_time
if start_time and end_time and start_time > end_time:
logger.warning(
f"EPG start_time {start_time} is after end_time {end_time}. Swapping."
)
start_time, end_time = end_time, start_time
return start_time, end_time
# ------------------------------------------------------------------
# Channel EPG
# ------------------------------------------------------------------
def get_channel_epg(
self,
provider_name: str,
channel_id: str,
start_time=None,
end_time=None,
limit: int = 100,
country: Optional[str] = None,
self,
provider_name: str,
channel_id: str,
start_time=None,
end_time=None,
limit: int = 100,
country: Optional[str] = None,
) -> List[EPGEntry]:
"""Get EPG data for a specific channel.
Args:
provider_name: Registered provider identifier.
channel_id: Provider-scoped channel identifier.
start_time: Window start as a timezone-aware datetime (optional).
end_time: Window end as a timezone-aware datetime (optional).
limit: Maximum number of programs to return (default 100).
country: Optional country filter forwarded to the provider.
"""
"""Get EPG data for a specific channel."""
provider = self._get_provider(provider_name)
if provider.implements_epg:
# Native providers hit a real upstream API per request — clamp/reject
# out-of-window requests so we don't overwhelm it. Generic/XMLTV-backed
# providers (epg_window == (0, 0)) read from a locally cached feed and
# don't need this throttling.
try:
start_time, end_time = self._validate_epg_window(provider, start_time, end_time)
except EPGWindowError as e:
logger.warning(f"EPG window validation failed for {provider_name}: {e.message}")
raise
logger.debug(f"Using native EPG for '{provider_name}'")
epg_data = provider.get_epg(
channel_id,
@@ -81,7 +175,6 @@ class EPGOperations:
country=country,
)
# Guard against a provider returning None instead of an empty list.
if epg_data is None:
logger.warning(
f"get_epg() returned None for channel '{channel_id}' "
@@ -97,28 +190,23 @@ class EPGOperations:
# ------------------------------------------------------------------
def get_provider_epg_grid(
self,
provider_name: str,
start_time=None,
end_time=None,
channel_ids: Optional[List[str]] = None,
country: Optional[str] = None,
self,
provider_name: str,
start_time=None,
end_time=None,
channel_ids: Optional[List[str]] = None,
country: Optional[str] = None,
) -> Dict[str, List[EPGEntry]]:
"""Get a time-windowed EPG grid across multiple channels.
Returns a dict keyed by channel_id, each value being a list of
program dicts within the requested window.
Args:
provider_name: Registered provider identifier.
start_time: Window start as a timezone-aware datetime (optional).
end_time: Window end as a timezone-aware datetime (optional).
channel_ids: Subset of channels to include; None means all.
country: Optional country filter forwarded to the provider.
"""
"""Get a time-windowed EPG grid across multiple channels."""
provider = self._get_provider(provider_name)
if provider.implements_epg:
try:
start_time, end_time = self._validate_epg_window(provider, start_time, end_time)
except EPGWindowError as e:
logger.warning(f"EPG window validation failed for {provider_name}: {e.message}")
raise
logger.debug(f"Using native EPG grid for '{provider_name}'")
return provider.get_epg_grid(
start_time=start_time,
@@ -127,7 +215,7 @@ class EPGOperations:
country=country,
) or {}
# Fallback: fan out to the generic EPG manager per channel.
# Fallback: generic EPG manager (XMLTV-backed, no native window throttling needed)
logger.debug(f"Using generic EPG grid fallback for '{provider_name}'")
channels = channel_ids or self.epg_manager.get_channel_ids(provider_name)
grid: Dict[str, List[Dict]] = {}
routes/epg.py
+25 -1
View File
@@ -21,7 +21,7 @@ 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.epg_operations import EPGOperations, EPGWindowError
from streaming_providers.base.utils import logger
@@ -282,6 +282,18 @@ def setup_epg_routes(app, manager, service):
"grid": grid_data,
}
except EPGWindowError as e:
logger.warning(f"EPG window rejected for {provider}: {e.message}")
response.status = 416 # Range Not Satisfiable
response.content_type = "application/json; charset=utf-8"
return {
"error": e.message,
"requested_start": e.requested_start,
"requested_end": e.requested_end,
"min_allowed": e.min_allowed,
"max_allowed": e.max_allowed,
"provider_window_days": e.provider_window_days,
}
except ValueError as e:
logger.error(f"Grid EPG error for {provider}: {e}")
response.status = 404
@@ -342,6 +354,18 @@ def setup_epg_routes(app, manager, service):
"programs": programs,
}
except EPGWindowError as e:
logger.warning(f"EPG window rejected for {provider}: {e.message}")
response.status = 416 # Range Not Satisfiable
response.content_type = "application/json; charset=utf-8"
return {
"error": e.message,
"requested_start": e.requested_start,
"requested_end": e.requested_end,
"min_allowed": e.min_allowed,
"max_allowed": e.max_allowed,
"provider_window_days": e.provider_window_days,
}
except ValueError as e:
logger.error(f"EPG error for {provider}/{channel_id}: {e}")
response.status = 404