Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-25 11:04:07 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1465 from Dispatcharr/sd-refactor

Browse source 
Enhance Schedules Direct integration with debugging, caching, and refactoring
This commit is contained in:
SergeantPanda 2026-07-21 14:19:23 -05:00 • committed by GitHub
commit 6b2ed0754a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 4112 additions and 2349 deletions
Download patch file Download diff file Expand all files Collapse all files

16
CHANGELOG.md
View file

@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
### Added
- **Schedules Direct Extra Debugging option.** EPG source settings include an **Extra Schedules Direct Debugging** toggle that adds a `RouteTo: debug` header so Schedules Direct support can steer traffic to their debug server. The tooltip states it should only be enabled when SD support asks. If SD returns code 2055 (unexpected debug connection), the toggle is turned off automatically.
### Changed
- **EPG source form places Auto-Apply EPG Logos in the shared middle column for XMLTV and Schedules Direct.** The toggle previously lived in the SD-only right panel when editing a Schedules Direct source; it now uses the same middle-column control as XMLTV so SD-specific options (logo style, posters, debug) stay in the right panel.
- **Schedules Direct poster proxy shares auth tokens across uWSGI workers via Redis.** Tokens are stored in Django's cache until near `tokenExpires` (24h fallback), so concurrent `/poster/` requests on different workers reuse one SD session instead of each process hitting `/token` separately. Redis failures fall back to re-authentication.
### Fixed
- **Schedules Direct poster proxy no longer ignores daily image download limits or missing-image errors.** SD returns codes 5002/5003 as HTTP 200 with a JSON body; the proxy previously only inspected HTTP 400 and kept requesting images after the limit, which can get accounts blocked. It now detects 5002/5003 (subscriber and trial), stops further image fetches for that source until the next midnight UTC (persisted on the EPG source so all workers honor it), and on explicit IMAGE_NOT_FOUND (code 5000) clears that program's `sd_icon` so the same dead URI is not retried. Transient bare HTTP 404s do not blacklist the URI. Successful SD posters use a dedicated nginx cache under `/data/cache/sd_posters` (14 days); channel/VOD logos remain at `/data/cache/logos` (24 hours). On startup, an existing `/data/logo_cache` directory is moved to `/data/cache/logos` once so warm logo entries are kept. API and XMLTV icon URLs include a `?v=` hash of the stored SD image URI so a new artwork link after refresh uses a new cache key without waiting for TTL expiry.
- **Schedules Direct auth and lineup change handling better match SD guidelines.** Token codes 3000/3001 (offline/busy) stop as idle with clear "do not retry" messaging instead of looking like credential failures; 4010 (too many unique IPs) and related account codes get explicit messages. Lineup add/delete respects a known daily change lockout before calling SD, handles 4100 on deletes, and the UI blocks remove when no changes remain (adds and deletes both count toward the 6/day limit).
- **Schedules Direct stops retrying `/token` on auth failures that will not clear by themselves.** Codes 4002/4003 (bad credentials), 4001/4005/4007/4008 (account state), 4009/4010 (login/IP limits), and 4004 (15-minute account lock) persist a cooldown on the EPG source. Soft codes 3000/3001 use a 1-hour idle cooldown. Lockouts clear when username/password change or the cooldown expires. Token auth is centralized in `sd_obtain_token` for refresh, lineup/form, and poster paths. Redis-cached SD tokens are bound to a credential fingerprint (and cleared when the EPG source username/password change) so switching accounts cannot reuse another session.
- **Schedules Direct code split out of general EPG modules.** Refresh pipeline and Celery tasks live in `apps/epg/sd_tasks.py`; lineup/poster API mixins in `apps/epg/sd_api.py`; shared protocol helpers remain in `sd_utils.py`. Progress WebSocket updates (`send_epg_update`) live in `apps/epg/utils.py` so XMLTV and SD tasks can import cleanly. `tasks.py` / `api_views.py` re-export or inherit so existing imports and Celery task names are unchanged.
## [0.28.1] - 2026-07-20
### Fixed

453
apps/epg/api_views.py
View file

@ -1,4 +1,3 @@
import hashlib
import logging
import os
import re
@ -8,6 +7,10 @@ from rest_framework.pagination import PageNumberPagination
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework.views import APIView
from apps.epg.sd_api import (
SchedulesDirectPosterMixin,
SchedulesDirectSourceMixin,
)
from rest_framework.decorators import action
from drf_spectacular.utils import extend_schema, OpenApiParameter, inline_serializer
from drf_spectacular.types import OpenApiTypes
@ -40,7 +43,7 @@ logger = logging.getLogger(__name__)
# ─────────────────────────────
# 1) EPG Source API (CRUD)
# ─────────────────────────────
class EPGSourceViewSet(viewsets.ModelViewSet):
class EPGSourceViewSet(SchedulesDirectSourceMixin, viewsets.ModelViewSet):
"""
API endpoint that allows EPG sources to be viewed or edited.
"""
@ -125,368 +128,20 @@ class EPGSourceViewSet(viewsets.ModelViewSet):
return super().partial_update(request, *args, **kwargs)
def _sd_authenticate(self, source):
"""
Authenticate with Schedules Direct using stored credentials.
Returns (token, None) on success or (None, Response) on failure.
"""
import hashlib
import requests as http_requests
from apps.epg.tasks import SD_BASE_URL
from core.utils import dispatcharr_http_headers
username = (source.username or '').strip()
password = (source.password or '').strip()
if not username or not password:
return None, Response(
{"error": "Username and password are required."},
status=status.HTTP_400_BAD_REQUEST
)
sha1_password = hashlib.sha1(password.encode('utf-8')).hexdigest()
try:
auth_response = http_requests.post(
f"{SD_BASE_URL}/token",
json={'username': username, 'password': sha1_password},
headers=dispatcharr_http_headers(),
timeout=15,
)
auth_response.raise_for_status()
token = auth_response.json().get('token')
if not token:
return None, Response(
{"error": "Authentication failed. Check your credentials."},
status=status.HTTP_401_UNAUTHORIZED
)
return token, None
except http_requests.exceptions.RequestException as e:
return None, Response(
{"error": f"Authentication failed: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
def _get_sd_reset_at(self, source):
"""Retrieve stored reset timestamp from EPGSource model field."""
reset_at_str = (source.custom_properties or {}).get('sd_changes_reset_at')
return reset_at_str
def _get_sd_changes_remaining(self, source):
"""
Retrieve stored changesRemaining from EPGSource model field.
If a reset timestamp exists and has passed (midnight UTC), clears the
lockout automatically so the user can make adds again.
"""
from django.utils import timezone
cp = source.custom_properties or {}
changes_remaining = cp.get('sd_changes_remaining')
reset_at_str = cp.get('sd_changes_reset_at')
from django.utils.dateparse import parse_datetime
reset_at = parse_datetime(reset_at_str) if reset_at_str else None
# If we have a reset timestamp and it has passed, clear the lockout
if changes_remaining == 0 and reset_at:
if timezone.now() >= reset_at:
cp = source.custom_properties or {}
cp.pop('sd_changes_remaining', None)
cp.pop('sd_changes_reset_at', None)
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
return None
return changes_remaining
def _save_sd_changes_remaining(self, source, changes_remaining):
"""Persist changesRemaining to EPGSource model field."""
cp = source.custom_properties or {}
cp['sd_changes_remaining'] = changes_remaining
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
def _save_sd_lockout(self, source):
"""
Persist a hard lockout to EPGSource custom_properties when SD returns
4100 MAX_LINEUP_CHANGES_REACHED. SD lineup change counters reset at
00:00Z (midnight UTC) per SD's documented behavior — error 4100 states
"lineup changes for today" and all SD rate counters reset at midnight UTC.
Lockout clears automatically when the next midnight UTC passes.
"""
from django.utils import timezone
from datetime import datetime, timedelta, timezone as dt_timezone
now = timezone.now()
# Calculate next midnight UTC — SD resets at 00:00Z not on a rolling window
tomorrow = (now + timedelta(days=1)).replace(
hour=0, minute=0, second=0, microsecond=0,
tzinfo=dt_timezone.utc
)
reset_at = tomorrow
cp = source.custom_properties or {}
cp['sd_changes_remaining'] = 0
cp['sd_changes_reset_at'] = reset_at.isoformat()
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
logger.warning(
f"SD source {source.id}: daily add limit reached (4100). "
f"Lockout set until {reset_at.isoformat()}."
)
def _fetch_sd_countries(self):
"""Fetch the SD country list (token not required; User-Agent is)."""
import requests as http_requests
from apps.epg.tasks import SD_BASE_URL
from core.utils import dispatcharr_http_headers
try:
resp = http_requests.get(
f"{SD_BASE_URL}/available/countries",
headers=dispatcharr_http_headers(content_type=None),
timeout=15,
)
resp.raise_for_status()
return resp.json()
except http_requests.exceptions.RequestException as e:
logger.warning(f"Failed to fetch SD countries: {e}")
return None
@action(detail=True, methods=["get", "post", "delete"], url_path="sd-lineups")
def sd_lineups(self, request, pk=None):
"""
GET — list lineups currently on the SD account
POST — add a lineup (body: {"lineup": "USA-NJ29486-X"})
DELETE — remove a lineup (body: {"lineup": "USA-NJ29486-X"})
"""
import requests as http_requests
from apps.epg.tasks import SD_BASE_URL
from core.utils import dispatcharr_http_headers
source = self.get_object()
if source.source_type != 'schedules_direct':
return Response(
{"error": "This action is only available for Schedules Direct sources."},
status=status.HTTP_400_BAD_REQUEST
)
token, error = self._sd_authenticate(source)
if error:
return error
headers = dispatcharr_http_headers(token=token)
if request.method == "GET":
countries = self._fetch_sd_countries()
try:
resp = http_requests.get(
f"{SD_BASE_URL}/lineups",
headers=headers,
timeout=15,
)
if resp.status_code == 400:
sd_data = resp.json()
sd_code = sd_data.get('code')
if sd_code == 4102:
return Response({
"lineups": [],
"max_lineups": 4,
"changes_remaining": self._get_sd_changes_remaining(source),
"changes_reset_at": self._get_sd_reset_at(source),
"notice": "No lineups are currently configured on this Schedules Direct account. Use the search below to add one.",
"countries": countries,
})
resp.raise_for_status()
data = resp.json()
lineups = [l for l in data.get('lineups', []) if not l.get('isDeleted', False)]
return Response({
"lineups": lineups,
"max_lineups": 4,
"changes_remaining": self._get_sd_changes_remaining(source),
"changes_reset_at": self._get_sd_reset_at(source),
"countries": countries,
})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to fetch lineups: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
elif request.method == "POST":
lineup_id = request.data.get('lineup')
if not lineup_id:
return Response({"error": "lineup field is required."}, status=status.HTTP_400_BAD_REQUEST)
try:
resp = http_requests.put(
f"{SD_BASE_URL}/lineups/{lineup_id}",
headers=headers,
timeout=15,
)
sd_data = resp.json()
sd_code = sd_data.get('code')
if resp.status_code == 400 or resp.status_code == 403:
if sd_code == 4100:
self._save_sd_lockout(source)
return Response({
"error": "daily_limit_reached",
"message": "You have reached your daily Schedules Direct lineup addition limit. SD allows 6 adds per 24-hour period. Resets at midnight UTC.",
"changes_remaining": 0,
"docs_url": "https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201#tasks-your-client-must-perform",
}, status=status.HTTP_200_OK)
if sd_code == 4101:
return Response({
"error": "max_lineups_reached",
"message": "Your Schedules Direct account has reached the maximum of 4 lineups. Remove one before adding another.",
"changes_remaining": self._get_sd_changes_remaining(source),
}, status=status.HTTP_200_OK)
if sd_code == 2100:
return Response({
"error": "duplicate_lineup",
"message": "This lineup is already on your Schedules Direct account.",
"changes_remaining": self._get_sd_changes_remaining(source),
}, status=status.HTTP_200_OK)
return Response({
"error": sd_data.get('message', 'Failed to add lineup.'),
"changes_remaining": self._get_sd_changes_remaining(source),
}, status=status.HTTP_200_OK)
resp.raise_for_status()
# Persist changesRemaining to custom_properties
changes_remaining = sd_data.get('changesRemaining')
if changes_remaining is not None:
self._save_sd_changes_remaining(source, changes_remaining)
logger.info(
f"SD lineup added for source {source.id}: {lineup_id}. "
f"changesRemaining: {changes_remaining}"
)
# Re-fetch stations so the new lineup's stations are available for matching
from apps.epg.tasks import fetch_schedules_direct_stations
fetch_schedules_direct_stations.delay(source.id)
return Response({
**sd_data,
"changes_remaining": changes_remaining,
})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to add lineup: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
elif request.method == "DELETE":
lineup_id = request.data.get('lineup')
if not lineup_id:
return Response({"error": "lineup field is required."}, status=status.HTTP_400_BAD_REQUEST)
try:
resp = http_requests.delete(
f"{SD_BASE_URL}/lineups/{lineup_id}",
headers=headers,
timeout=15,
)
if resp.status_code == 400:
sd_data = resp.json()
sd_code = sd_data.get('code')
if sd_code == 2103:
return Response({
"response": "OK",
"code": 0,
"message": "Lineup not found on account — already removed.",
"changes_remaining": self._get_sd_changes_remaining(source),
})
resp.raise_for_status()
sd_data = resp.json()
# SD returns changesRemaining on deletes — persist it
changes_remaining = sd_data.get('changesRemaining')
if changes_remaining is not None:
self._save_sd_changes_remaining(source, changes_remaining)
logger.info(f"SD lineup deleted for source {source.id}: {lineup_id}")
return Response({
**sd_data,
"changes_remaining": self._get_sd_changes_remaining(source),
})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to remove lineup: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
@action(detail=True, methods=["post"], url_path="sd-lineups/search")
def sd_lineups_search(self, request, pk=None):
"""
Search available headends/lineups by country and postal code.
Body: {"country": "USA", "postalcode": "07030"}
Returns a flat list of lineups across all matching headends.
"""
import requests as http_requests
from apps.epg.tasks import SD_BASE_URL
from core.utils import dispatcharr_http_headers
source = self.get_object()
if source.source_type != 'schedules_direct':
return Response(
{"error": "This action is only available for Schedules Direct sources."},
status=status.HTTP_400_BAD_REQUEST
)
country = request.data.get('country', '').strip()
postalcode = request.data.get('postalcode', '').strip()
if not country or not postalcode:
return Response(
{"error": "country and postalcode are required."},
status=status.HTTP_400_BAD_REQUEST
)
token, error = self._sd_authenticate(source)
if error:
return error
headers = dispatcharr_http_headers(token=token)
try:
resp = http_requests.get(
f"{SD_BASE_URL}/headends",
params={'country': country, 'postalcode': postalcode},
headers=headers,
timeout=15,
)
resp.raise_for_status()
headends = resp.json()
lineups = []
for headend in headends:
for lineup in headend.get('lineups', []):
lineups.append({
'lineup': lineup.get('lineup'),
'name': lineup.get('name'),
'transport': headend.get('transport'),
'location': headend.get('location'),
'headend': headend.get('headend'),
})
return Response({"lineups": lineups})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to search headends: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
# ─────────────────────────────
# 2) Program API (CRUD)
# ─────────────────────────────
class ProgramSearchPagination(PageNumberPagination):
page_size = 50
page_size_query_param = 'page_size'
max_page_size = 500
class ProgramViewSet(viewsets.ModelViewSet):
class ProgramViewSet(SchedulesDirectPosterMixin, viewsets.ModelViewSet):
"""Handles CRUD operations for EPG programs"""
queryset = ProgramData.objects.select_related("epg").all()
serializer_class = ProgramDataSerializer
# Per-source in-memory caches (token and error state)
_sd_poster_token_cache: dict = {}
_sd_poster_error_cache: dict = {}
# Short process-local cooldown for transient poster errors (auth/network).
# Image download limits are persisted on the EPG source (shared across workers).
def get_permissions(self):
if self.action == 'poster':
@ -506,98 +161,6 @@ class ProgramViewSet(viewsets.ModelViewSet):
serializer = self.get_serializer(instance)
return Response(serializer.data)
@action(detail=True, methods=['get'], url_path='poster', permission_classes=[AllowAny])
def poster(self, request, pk=None):
"""Proxy endpoint for SD program poster images. Nginx caches the response."""
import requests as http_requests
from apps.epg.tasks import SD_BASE_URL
from core.utils import dispatcharr_http_headers
program = self.get_object()
poster_sd_url = (program.custom_properties or {}).get('sd_icon')
if not poster_sd_url:
return Response(status=status.HTTP_404_NOT_FOUND)
source = program.epg.epg_source if program.epg else None
if not source or source.source_type != 'schedules_direct':
return Response(status=status.HTTP_404_NOT_FOUND)
error_cache = ProgramViewSet._sd_poster_error_cache.get(source.id)
if error_cache and time.time() < error_cache['until']:
return Response(
{'error': f"SD temporarily unavailable: {error_cache['reason']}"},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
cached = ProgramViewSet._sd_poster_token_cache.get(source.id)
token = cached['token'] if cached and time.time() < cached['expires'] else None
if not token:
sha1_password = hashlib.sha1(source.password.encode('utf-8')).hexdigest()
try:
auth_resp = http_requests.post(
f"{SD_BASE_URL}/token",
json={'username': source.username, 'password': sha1_password},
headers=dispatcharr_http_headers(),
timeout=10,
)
auth_data = auth_resp.json()
token = auth_data.get('token')
if not token:
ProgramViewSet._sd_poster_error_cache[source.id] = {
'until': time.time() + 3600,
'reason': auth_data.get('message', 'Authentication failed'),
}
return Response(status=status.HTTP_502_BAD_GATEWAY)
token_expires = auth_data.get('tokenExpires', time.time() + 86400)
ProgramViewSet._sd_poster_token_cache[source.id] = {
'token': token,
'expires': token_expires,
}
except http_requests.exceptions.RequestException:
ProgramViewSet._sd_poster_error_cache[source.id] = {
'until': time.time() + 300,
'reason': 'Network error reaching Schedules Direct',
}
return Response(status=status.HTTP_502_BAD_GATEWAY)
try:
img_resp = http_requests.get(
poster_sd_url,
headers=dispatcharr_http_headers(token=token, content_type=None),
timeout=15,
allow_redirects=True,
)
if img_resp.status_code in (401, 403):
ProgramViewSet._sd_poster_token_cache.pop(source.id, None)
ProgramViewSet._sd_poster_error_cache[source.id] = {
'until': time.time() + 3600,
'reason': f'SD returned {img_resp.status_code}',
}
return Response(status=status.HTTP_502_BAD_GATEWAY)
if img_resp.status_code == 400:
try:
err_code = img_resp.json().get('code')
except Exception:
err_code = None
if err_code == 5002:
ProgramViewSet._sd_poster_error_cache[source.id] = {
'until': time.time() + 3600,
'reason': 'Daily image download limit reached (SD error 5002)',
}
return Response(status=status.HTTP_429_TOO_MANY_REQUESTS)
if img_resp.status_code != 200:
return Response(status=status.HTTP_502_BAD_GATEWAY)
from django.http import HttpResponse
content_type = img_resp.headers.get('Content-Type', 'image/jpeg')
response = HttpResponse(img_resp.content, content_type=content_type)
response['Cache-Control'] = 'public, max-age=86400'
return response
except http_requests.exceptions.RequestException:
return Response(status=status.HTTP_502_BAD_GATEWAY)
def list(self, request, *args, **kwargs):
logger.debug("Listing all EPG programs.")
return super().list(request, *args, **kwargs)

570
apps/epg/sd_api.py Normal file
View file

@ -0,0 +1,570 @@
"""
Schedules Direct HTTP API helpers and ViewSet mixins.
Keeps lineup management and poster proxy out of the general EPG view modules.
Protocol/auth helpers live in ``sd_utils``; the refresh pipeline in ``sd_tasks``.
"""
from __future__ import annotations
import logging
import time
import requests as http_requests
from django.http import HttpResponse
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from apps.epg.sd_utils import (
SD_AUTH_CREDENTIAL_LOCKOUT_CODES,
SD_BASE_URL,
SD_CODE_IMAGE_NOT_FOUND,
SD_IMAGE_LIMIT_CODES,
sd_auth_lockout_active,
sd_clear_cached_token,
sd_get_cached_token,
sd_handle_2055,
sd_headers_for_source,
sd_image_limit_active,
sd_mark_icon_missing,
sd_next_midnight_utc,
sd_obtain_token,
sd_parse_response_payload,
sd_save_image_limit_lockout,
)
from core.utils import dispatcharr_http_headers
logger = logging.getLogger(__name__)
class SchedulesDirectSourceMixin:
"""Lineup management actions and helpers for Schedules Direct EPG sources."""
def _sd_authenticate(self, source):
"""
Authenticate with Schedules Direct using stored credentials.
Returns (token, None) on success or (None, Response) on failure.
"""
result = sd_obtain_token(source, timeout=15)
if result.ok:
return result.token, None
if result.debug_rejected:
return None, Response(
{"error": result.message},
status=status.HTTP_400_BAD_REQUEST,
)
if result.message == 'Username and password are required.':
http_status = status.HTTP_400_BAD_REQUEST
elif result.code in SD_AUTH_CREDENTIAL_LOCKOUT_CODES:
http_status = status.HTTP_401_UNAUTHORIZED
elif result.lockout or result.soft or result.code:
http_status = status.HTTP_503_SERVICE_UNAVAILABLE
else:
http_status = status.HTTP_502_BAD_GATEWAY
return None, Response({"error": result.message}, status=http_status)
def _get_sd_reset_at(self, source):
"""Retrieve stored reset timestamp from EPGSource model field."""
reset_at_str = (source.custom_properties or {}).get('sd_changes_reset_at')
return reset_at_str
def _get_sd_changes_remaining(self, source):
"""
Retrieve stored changesRemaining from EPGSource model field.
If a reset timestamp exists and has passed (midnight UTC), clears the
lockout automatically so the user can make adds again.
"""
cp = source.custom_properties or {}
changes_remaining = cp.get('sd_changes_remaining')
reset_at_str = cp.get('sd_changes_reset_at')
reset_at = parse_datetime(reset_at_str) if reset_at_str else None
# If we have a reset timestamp and it has passed, clear the lockout
if changes_remaining == 0 and reset_at:
if timezone.now() >= reset_at:
cp = source.custom_properties or {}
cp.pop('sd_changes_remaining', None)
cp.pop('sd_changes_reset_at', None)
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
return None
return changes_remaining
def _save_sd_changes_remaining(self, source, changes_remaining):
"""
Persist changesRemaining on the EPG source.
When remaining hits 0, also store sd_changes_reset_at (next midnight UTC)
so the lockout can clear automatically. A positive remaining clears any
prior reset timestamp.
"""
cp = dict(source.custom_properties or {})
cp['sd_changes_remaining'] = changes_remaining
if changes_remaining == 0:
cp['sd_changes_reset_at'] = sd_next_midnight_utc().isoformat()
else:
cp.pop('sd_changes_reset_at', None)
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
def _save_sd_lockout(self, source):
"""
Persist a hard lockout when SD returns 4100 MAX_LINEUP_CHANGES_REACHED.
SD lineup change counters reset at 00:00Z (midnight UTC). Lockout clears
automatically when that reset time passes.
"""
self._save_sd_changes_remaining(source, 0)
reset_at = (source.custom_properties or {}).get('sd_changes_reset_at')
logger.warning(
f"SD source {source.id}: daily lineup change limit reached (4100). "
f"Lockout set until {reset_at}."
)
def _fetch_sd_countries(self):
"""Fetch the SD country list (token not required; User-Agent is)."""
try:
resp = http_requests.get(
f"{SD_BASE_URL}/available/countries",
headers=dispatcharr_http_headers(content_type=None),
timeout=15,
)
resp.raise_for_status()
return resp.json()
except http_requests.exceptions.RequestException as e:
logger.warning(f"Failed to fetch SD countries: {e}")
return None
@action(detail=True, methods=["get", "post", "delete"], url_path="sd-lineups")
def sd_lineups(self, request, pk=None):
"""
GET — list lineups currently on the SD account
POST — add a lineup (body: {"lineup": "USA-NJ29486-X"})
DELETE — remove a lineup (body: {"lineup": "USA-NJ29486-X"})
"""
source = self.get_object()
if source.source_type != 'schedules_direct':
return Response(
{"error": "This action is only available for Schedules Direct sources."},
status=status.HTTP_400_BAD_REQUEST
)
token, error = self._sd_authenticate(source)
if error:
return error
headers = sd_headers_for_source(source, token=token)
if request.method == "GET":
countries = self._fetch_sd_countries()
try:
resp = http_requests.get(
f"{SD_BASE_URL}/lineups",
headers=headers,
timeout=15,
)
if resp.status_code == 400:
sd_data = resp.json()
sd_code = sd_data.get('code')
if sd_code == 4102:
return Response({
"lineups": [],
"max_lineups": 4,
"changes_remaining": self._get_sd_changes_remaining(source),
"changes_reset_at": self._get_sd_reset_at(source),
"notice": "No lineups are currently configured on this Schedules Direct account. Use the search below to add one.",
"countries": countries,
})
resp.raise_for_status()
data = resp.json()
lineups = [l for l in data.get('lineups', []) if not l.get('isDeleted', False)]
return Response({
"lineups": lineups,
"max_lineups": 4,
"changes_remaining": self._get_sd_changes_remaining(source),
"changes_reset_at": self._get_sd_reset_at(source),
"countries": countries,
})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to fetch lineups: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
elif request.method == "POST":
lineup_id = request.data.get('lineup')
if not lineup_id:
return Response({"error": "lineup field is required."}, status=status.HTTP_400_BAD_REQUEST)
# Honor a known daily change lockout without calling SD again.
if self._get_sd_changes_remaining(source) == 0:
return Response({
"error": "daily_limit_reached",
"message": (
"You have reached your daily Schedules Direct lineup "
"change limit (6 add/delete operations per 24 hours). "
"Resets at midnight UTC."
),
"changes_remaining": 0,
"changes_reset_at": self._get_sd_reset_at(source),
"docs_url": "https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201#tasks-your-client-must-perform",
}, status=status.HTTP_200_OK)
try:
resp = http_requests.put(
f"{SD_BASE_URL}/lineups/{lineup_id}",
headers=headers,
timeout=15,
)
sd_data = resp.json()
sd_code = sd_data.get('code')
if resp.status_code == 400 or resp.status_code == 403:
if sd_code == 4100:
self._save_sd_lockout(source)
return Response({
"error": "daily_limit_reached",
"message": (
"You have reached your daily Schedules Direct lineup "
"change limit (6 add/delete operations per 24 hours). "
"Resets at midnight UTC."
),
"changes_remaining": 0,
"docs_url": "https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201#tasks-your-client-must-perform",
}, status=status.HTTP_200_OK)
if sd_code == 4101:
return Response({
"error": "max_lineups_reached",
"message": "Your Schedules Direct account has reached the maximum of 4 lineups. Remove one before adding another.",
"changes_remaining": self._get_sd_changes_remaining(source),
}, status=status.HTTP_200_OK)
if sd_code == 2100:
return Response({
"error": "duplicate_lineup",
"message": "This lineup is already on your Schedules Direct account.",
"changes_remaining": self._get_sd_changes_remaining(source),
}, status=status.HTTP_200_OK)
return Response({
"error": sd_data.get('message', 'Failed to add lineup.'),
"changes_remaining": self._get_sd_changes_remaining(source),
}, status=status.HTTP_200_OK)
resp.raise_for_status()
# Persist changesRemaining to custom_properties
changes_remaining = sd_data.get('changesRemaining')
if changes_remaining is not None:
self._save_sd_changes_remaining(source, changes_remaining)
logger.info(
f"SD lineup added for source {source.id}: {lineup_id}. "
f"changesRemaining: {changes_remaining}"
)
# Re-fetch stations so the new lineup's stations are available for matching
from apps.epg.tasks import fetch_schedules_direct_stations
fetch_schedules_direct_stations.delay(source.id)
return Response({
**sd_data,
"changes_remaining": changes_remaining,
})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to add lineup: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
elif request.method == "DELETE":
lineup_id = request.data.get('lineup')
if not lineup_id:
return Response({"error": "lineup field is required."}, status=status.HTTP_400_BAD_REQUEST)
if self._get_sd_changes_remaining(source) == 0:
return Response({
"error": "daily_limit_reached",
"message": (
"You have reached your daily Schedules Direct lineup "
"change limit (6 add/delete operations per 24 hours). "
"Resets at midnight UTC."
),
"changes_remaining": 0,
"changes_reset_at": self._get_sd_reset_at(source),
"docs_url": "https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201#tasks-your-client-must-perform",
}, status=status.HTTP_200_OK)
try:
resp = http_requests.delete(
f"{SD_BASE_URL}/lineups/{lineup_id}",
headers=headers,
timeout=15,
)
if resp.status_code == 400:
sd_data = resp.json()
sd_code = sd_data.get('code')
if sd_code == 2103:
return Response({
"response": "OK",
"code": 0,
"message": "Lineup not found on account — already removed.",
"changes_remaining": self._get_sd_changes_remaining(source),
})
if sd_code == 4100:
self._save_sd_lockout(source)
return Response({
"error": "daily_limit_reached",
"message": (
"You have reached your daily Schedules Direct lineup "
"change limit (6 add/delete operations per 24 hours). "
"Resets at midnight UTC."
),
"changes_remaining": 0,
"docs_url": "https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201#tasks-your-client-must-perform",
}, status=status.HTTP_200_OK)
resp.raise_for_status()
sd_data = resp.json()
# SD returns changesRemaining on deletes — persist it
changes_remaining = sd_data.get('changesRemaining')
if changes_remaining is not None:
self._save_sd_changes_remaining(source, changes_remaining)
logger.info(f"SD lineup deleted for source {source.id}: {lineup_id}")
return Response({
**sd_data,
"changes_remaining": self._get_sd_changes_remaining(source),
})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to remove lineup: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
@action(detail=True, methods=["post"], url_path="sd-lineups/search")
def sd_lineups_search(self, request, pk=None):
"""
Search available headends/lineups by country and postal code.
Body: {"country": "USA", "postalcode": "07030"}
Returns a flat list of lineups across all matching headends.
"""
source = self.get_object()
if source.source_type != 'schedules_direct':
return Response(
{"error": "This action is only available for Schedules Direct sources."},
status=status.HTTP_400_BAD_REQUEST
)
country = request.data.get('country', '').strip()
postalcode = request.data.get('postalcode', '').strip()
if not country or not postalcode:
return Response(
{"error": "country and postalcode are required."},
status=status.HTTP_400_BAD_REQUEST
)
token, error = self._sd_authenticate(source)
if error:
return error
headers = sd_headers_for_source(source, token=token)
try:
resp = http_requests.get(
f"{SD_BASE_URL}/headends",
params={'country': country, 'postalcode': postalcode},
headers=headers,
timeout=15,
)
try:
headends = resp.json()
except ValueError:
headends = None
if isinstance(headends, dict) and sd_handle_2055(source, headends):
return Response(
{
"error": (
"Schedules Direct rejected the debug routing header "
"(code 2055). Extra Schedules Direct Debugging has "
"been turned off."
),
},
status=status.HTTP_400_BAD_REQUEST,
)
resp.raise_for_status()
if not isinstance(headends, list):
headends = []
lineups = []
for headend in headends:
for lineup in headend.get('lineups', []):
lineups.append({
'lineup': lineup.get('lineup'),
'name': lineup.get('name'),
'transport': headend.get('transport'),
'location': headend.get('location'),
'headend': headend.get('headend'),
})
return Response({"lineups": lineups})
except http_requests.exceptions.RequestException as e:
return Response(
{"error": f"Failed to search headends: {str(e)}"},
status=status.HTTP_502_BAD_GATEWAY
)
# ─────────────────────────────
# 2) Program API (CRUD)
# ─────────────────────────────
class SchedulesDirectPosterMixin:
"""Program poster proxy for Schedules Direct artwork URIs."""
_sd_poster_error_cache: dict = {}
@action(detail=True, methods=['get'], url_path='poster', permission_classes=[AllowAny])
def poster(self, request, pk=None):
"""Proxy endpoint for SD program poster images. Nginx caches the response."""
program = self.get_object()
cp = program.custom_properties or {}
if cp.get('sd_icon_missing'):
return Response(status=status.HTTP_404_NOT_FOUND)
poster_sd_url = cp.get('sd_icon')
if not poster_sd_url:
return Response(status=status.HTTP_404_NOT_FOUND)
source = program.epg.epg_source if program.epg else None
if not source or source.source_type != 'schedules_direct':
return Response(status=status.HTTP_404_NOT_FOUND)
# Persisted lockout (shared across workers) until next midnight UTC.
limit_active, limit_reason = sd_image_limit_active(source)
if limit_active:
self._sd_poster_error_cache[source.id] = {
'until': time.time() + 300,
'reason': limit_reason,
}
return Response(
{'error': f"SD temporarily unavailable: {limit_reason}"},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
auth_lockout_active, auth_lockout_reason, _lockout_code = (
sd_auth_lockout_active(source)
)
if auth_lockout_active:
# Rely on the persisted lockout only. Do not seed the process-local
# poster error cache here: that cache outlives credential changes and
# cooldown expiry and would keep returning 503 afterward.
return Response(
{'error': auth_lockout_reason},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
error_cache = self._sd_poster_error_cache.get(source.id)
if error_cache and time.time() < error_cache['until']:
return Response(
{'error': f"SD temporarily unavailable: {error_cache['reason']}"},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
token = sd_get_cached_token(
source.id, username=source.username, password=source.password
)
if not token:
auth = sd_obtain_token(source, timeout=10)
if auth.debug_rejected:
return Response(
{'error': auth.message},
status=status.HTTP_400_BAD_REQUEST,
)
if not auth.ok:
# Lockout codes are persisted by sd_obtain_token. Other failures
# use a short process-local cache so workers do not hammer /token.
if not auth.lockout:
self._sd_poster_error_cache[source.id] = {
'until': time.time() + (3600 if auth.code else 300),
'reason': auth.message or 'Authentication failed',
}
return Response(
{'error': auth.message},
status=status.HTTP_502_BAD_GATEWAY,
)
token = auth.token
self._sd_poster_error_cache.pop(source.id, None)
try:
img_resp = http_requests.get(
poster_sd_url,
headers=sd_headers_for_source(source, token=token, content_type=None),
timeout=15,
allow_redirects=True,
)
err_code, err_data = sd_parse_response_payload(img_resp)
if err_data is not None and sd_handle_2055(source, err_data):
return Response(
{
'error': (
'Schedules Direct rejected the debug routing header '
'(code 2055). Extra Schedules Direct Debugging has '
'been turned off.'
),
},
status=status.HTTP_400_BAD_REQUEST,
)
# SD documents 5002/5003 as HTTP 200 + JSON. Also accept 4xx.
if err_code in SD_IMAGE_LIMIT_CODES:
sd_save_image_limit_lockout(source, err_code)
self._sd_poster_error_cache[source.id] = {
'until': time.time() + 300,
'reason': (
f'Daily image download limit reached (SD error {err_code})'
),
}
return Response(status=status.HTTP_429_TOO_MANY_REQUESTS)
# Only blacklist on explicit IMAGE_NOT_FOUND (5000). Bare HTTP 404 can
# be a transient CDN/S3 miss for ephemeral URIs and must not permanently
# clear sd_icon (SD docs: code 5000 + HTTP 404).
if err_code == SD_CODE_IMAGE_NOT_FOUND:
sd_mark_icon_missing(program)
return Response(status=status.HTTP_404_NOT_FOUND)
if img_resp.status_code == 404:
return Response(status=status.HTTP_404_NOT_FOUND)
if img_resp.status_code in (401, 403):
sd_clear_cached_token(source.id)
self._sd_poster_error_cache[source.id] = {
'until': time.time() + 3600,
'reason': f'SD returned {img_resp.status_code}',
}
return Response(status=status.HTTP_502_BAD_GATEWAY)
# JSON error body (even on HTTP 200) is never a valid image.
if err_data is not None and err_code not in (None, 0):
logger.warning(
"SD poster proxy: unexpected JSON error code %s for program %s",
err_code,
program.id,
)
return Response(status=status.HTTP_502_BAD_GATEWAY)
if img_resp.status_code != 200:
return Response(status=status.HTTP_502_BAD_GATEWAY)
content_type = img_resp.headers.get('Content-Type', 'image/jpeg')
if 'json' in (content_type or '').lower():
return Response(status=status.HTTP_502_BAD_GATEWAY)
response = HttpResponse(img_resp.content, content_type=content_type)
response['Cache-Control'] = 'public, max-age=86400'
return response
except http_requests.exceptions.RequestException:
return Response(status=status.HTTP_502_BAD_GATEWAY)

1816
apps/epg/sd_tasks.py Normal file
View file

File diff suppressed because it is too large Load diff

694
apps/epg/sd_utils.py Normal file
View file

@ -0,0 +1,694 @@
"""
Schedules Direct API helpers: headers, error codes, and rate-limit lockouts.
See https://github.com/SchedulesDirect/JSON-Service/wiki/API-20141201
"""
from __future__ import annotations
import hashlib
import json
import logging
import time
from dataclasses import dataclass
from datetime import timedelta, timezone as dt_timezone
import requests
from django.core.cache import cache
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from core.utils import dispatcharr_http_headers
logger = logging.getLogger(__name__)
SD_BASE_URL = 'https://json.schedulesdirect.org/20141201'
# SD JSON error codes we must honor to avoid account blocks.
SD_CODE_INVALID_DEBUG = 2055
SD_CODE_SERVICE_OFFLINE = 3000
SD_CODE_SERVICE_BUSY = 3001
SD_CODE_ACCOUNT_EXPIRED = 4001
SD_CODE_INVALID_PASSWORD_HASH = 4002
SD_CODE_INVALID_USER_OR_PASSWORD = 4003
SD_CODE_ACCOUNT_LOCKED = 4004
SD_CODE_JSON_DISABLED = 4005
SD_CODE_APP_NOT_AUTHORIZED = 4007
SD_CODE_ACCOUNT_INACTIVE = 4008
SD_CODE_TOO_MANY_LOGINS = 4009
SD_CODE_TOO_MANY_UNIQUE_IPS = 4010
SD_CODE_IMAGE_NOT_FOUND = 5000
SD_CODE_MAX_IMAGE_DOWNLOADS = 5002
SD_CODE_MAX_IMAGE_DOWNLOADS_TRIAL = 5003
SD_IMAGE_LIMIT_CODES = frozenset({
SD_CODE_MAX_IMAGE_DOWNLOADS,
SD_CODE_MAX_IMAGE_DOWNLOADS_TRIAL,
})
# Soft /token failures: stop and retry later (idle), not a hard account error.
SD_AUTH_SOFT_CODES = frozenset({
SD_CODE_SERVICE_OFFLINE,
SD_CODE_SERVICE_BUSY,
})
# Wrong username/password (or hash): clear early when credentials change.
SD_AUTH_CREDENTIAL_LOCKOUT_CODES = frozenset({
SD_CODE_INVALID_PASSWORD_HASH,
SD_CODE_INVALID_USER_OR_PASSWORD,
})
# All /token codes we must not hammer. Includes soft codes (shorter cooldown).
SD_AUTH_LOCKOUT_CODES = frozenset({
SD_CODE_SERVICE_OFFLINE,
SD_CODE_SERVICE_BUSY,
SD_CODE_ACCOUNT_EXPIRED,
SD_CODE_INVALID_PASSWORD_HASH,
SD_CODE_INVALID_USER_OR_PASSWORD,
SD_CODE_ACCOUNT_LOCKED,
SD_CODE_JSON_DISABLED,
SD_CODE_APP_NOT_AUTHORIZED,
SD_CODE_ACCOUNT_INACTIVE,
SD_CODE_TOO_MANY_LOGINS,
SD_CODE_TOO_MANY_UNIQUE_IPS,
})
SD_AUTH_LOCKOUT_SECONDS = 24 * 3600
SD_AUTH_LOCKOUT_SECONDS_SOFT = 3600
SD_AUTH_LOCKOUT_SECONDS_ACCOUNT_LOCK = 15 * 60
def sd_auth_lockout_seconds_for_code(code):
"""Cooldown length for a /token error code."""
if code == SD_CODE_ACCOUNT_LOCKED:
return SD_AUTH_LOCKOUT_SECONDS_ACCOUNT_LOCK
if code in SD_AUTH_SOFT_CODES:
return SD_AUTH_LOCKOUT_SECONDS_SOFT
return SD_AUTH_LOCKOUT_SECONDS
def sd_auth_lockout_retry_message():
"""Suffix explaining how to clear an auth lockout."""
return (
"Not retrying Schedules Direct authentication until the username or "
"password is updated, or the cooldown expires."
)
# Shared across uWSGI workers via Django's Redis cache.
_SD_TOKEN_CACHE_PREFIX = 'sd:token:'
# Expire a bit early so we re-auth before SD rejects an almost-expired token.
_SD_TOKEN_CACHE_SKEW_SECONDS = 60
_SD_TOKEN_DEFAULT_TTL_SECONDS = 86400
def sd_token_cache_key(source_id):
return f'{_SD_TOKEN_CACHE_PREFIX}{source_id}'
def sd_credential_fingerprint(username, password):
"""
Stable fingerprint of SD credentials for lockout and token-cache matching.
Used so a persisted auth lockout or cached token clears automatically when
the user changes username or password.
"""
raw = f'{(username or "").strip()}\0{(password or "").strip()}'
return hashlib.sha256(raw.encode('utf-8')).hexdigest()
def sd_get_cached_token(source_id, username=None, password=None):
"""
Return a cached SD token string for this source, or None.
Requires ``username`` / ``password`` so a cached session is only reused when
credentials still match (avoids using account A's token after switching to B).
On Redis failure, returns None so the caller re-authenticates.
"""
if source_id is None:
return None
current_fp = sd_credential_fingerprint(username, password)
try:
payload = cache.get(sd_token_cache_key(source_id))
except Exception as exc:
logger.warning(
"SD token cache get failed for source %s (%s); re-authenticating",
source_id,
type(exc).__name__,
)
return None
if not isinstance(payload, dict):
return None
token = payload.get('token')
expires = payload.get('expires')
if not token or not isinstance(expires, (int, float)):
return None
if payload.get('fp') != current_fp:
sd_clear_cached_token(source_id)
return None
if time.time() >= float(expires) - _SD_TOKEN_CACHE_SKEW_SECONDS:
return None
return token
def sd_set_cached_token(source_id, token, expires=None, username=None, password=None):
"""
Cache an SD token for this source until near tokenExpires.
``expires`` is a UNIX epoch seconds value from SD's tokenExpires field.
Stores a credential fingerprint so callers must still match to reuse it.
"""
if source_id is None or not token:
return False
now = time.time()
if expires is None:
expires = now + _SD_TOKEN_DEFAULT_TTL_SECONDS
try:
expires = float(expires)
except (TypeError, ValueError):
expires = now + _SD_TOKEN_DEFAULT_TTL_SECONDS
ttl = int(expires - now - _SD_TOKEN_CACHE_SKEW_SECONDS)
if ttl < 1:
return False
try:
cache.set(
sd_token_cache_key(source_id),
{
'token': token,
'expires': expires,
'fp': sd_credential_fingerprint(username, password),
},
timeout=ttl,
)
return True
except Exception as exc:
logger.warning(
"SD token cache set failed for source %s (%s)",
source_id,
type(exc).__name__,
)
return False
def sd_clear_cached_token(source_id):
"""Drop a cached SD token (e.g. after 401/403 from an image request)."""
if source_id is None:
return False
try:
cache.delete(sd_token_cache_key(source_id))
return True
except Exception as exc:
logger.warning(
"SD token cache delete failed for source %s (%s)",
source_id,
type(exc).__name__,
)
return False
def sd_auth_failure_message(code, sd_message=None):
"""Human-readable message for a Schedules Direct /token error code."""
if code == SD_CODE_SERVICE_OFFLINE:
return (
"Schedules Direct is offline for maintenance. "
"Do not retry for at least 1 hour."
)
if code == SD_CODE_SERVICE_BUSY:
return "Schedules Direct is busy. Stop requesting and retry later."
if code == SD_CODE_ACCOUNT_EXPIRED:
return (
"Schedules Direct: account has expired. Please renew your "
"subscription at schedulesdirect.org."
)
if code == SD_CODE_INVALID_PASSWORD_HASH:
return (
"Schedules Direct: invalid password hash. Check that credentials "
"are stored correctly."
)
if code == SD_CODE_INVALID_USER_OR_PASSWORD:
return "Schedules Direct: invalid username or password."
if code == SD_CODE_ACCOUNT_LOCKED:
return (
"Schedules Direct: account locked due to too many failed login "
"attempts. Try again in 15 minutes."
)
if code == SD_CODE_JSON_DISABLED:
return (
"Schedules Direct: JSON API access is disabled for this account. "
"Contact Schedules Direct support."
)
if code == SD_CODE_APP_NOT_AUTHORIZED:
return (
"Schedules Direct: this application is not authorized. Please "
"contact the Dispatcharr maintainers."
)
if code == SD_CODE_ACCOUNT_INACTIVE:
return (
"Schedules Direct: account is inactive. Please log in to "
"schedulesdirect.org to reactivate."
)
if code == SD_CODE_TOO_MANY_LOGINS:
return (
"Schedules Direct: too many login attempts in 24 hours. Token is "
"valid for 24 hours. Check for misconfiguration."
)
if code == SD_CODE_TOO_MANY_UNIQUE_IPS:
return (
"Schedules Direct: too many unique IP addresses in 24 hours. "
"Avoid VPNs or multiple locations, or contact SD support to raise "
"the limit."
)
if sd_message:
return f"Schedules Direct authentication failed (code {code}): {sd_message}"
return f"Schedules Direct authentication failed (code {code})."
def sd_token_response_code(auth_data):
"""Return integer SD ``code`` from a /token JSON body, or 0."""
if not isinstance(auth_data, dict):
return 0
code = auth_data.get('code', 0)
return code if isinstance(code, int) else 0
def sd_auth_lockout_active(source, username=None, password=None):
"""
Return (active, reason, code) for a persisted auth lockout.
Cleared when username/password changed, or sd_auth_lockout_until has passed.
"""
if source is None:
return False, None, None
cp = source.custom_properties or {}
if not cp.get('sd_auth_lockout'):
return False, None, None
until_str = cp.get('sd_auth_lockout_until')
until = parse_datetime(until_str) if until_str else None
if until is None or timezone.now() >= until:
sd_clear_auth_lockout(source)
return False, None, None
if username is None:
username = source.username
if password is None:
password = source.password
stored_fp = cp.get('sd_auth_lockout_credential_fp')
current_fp = sd_credential_fingerprint(username, password)
if not stored_fp or stored_fp != current_fp:
sd_clear_auth_lockout(source)
return False, None, None
code = cp.get('sd_auth_lockout_code')
reason = cp.get('sd_auth_lockout_reason') or (
'Schedules Direct authentication is temporarily blocked.'
)
return True, reason, code if isinstance(code, int) else None
def sd_save_auth_lockout(source, code, username=None, password=None, reason=None):
"""
Persist an auth lockout so we stop calling /token for a cooldown period.
Duration depends on the code (15m for 4004, 1h for offline/busy, else 24h).
Cleared early if username/password change.
"""
if source is None:
return
if code not in SD_AUTH_LOCKOUT_CODES:
return
if username is None:
username = source.username
if password is None:
password = source.password
if reason is None:
reason = sd_auth_failure_message(code)
seconds = sd_auth_lockout_seconds_for_code(code)
until = timezone.now() + timedelta(seconds=seconds)
cp = dict(source.custom_properties or {})
cp['sd_auth_lockout'] = True
cp['sd_auth_lockout_code'] = code
cp['sd_auth_lockout_reason'] = reason
cp['sd_auth_lockout_until'] = until.isoformat()
cp['sd_auth_lockout_credential_fp'] = sd_credential_fingerprint(
username, password
)
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
sd_clear_cached_token(source.id)
logger.warning(
"SD source %s: auth lockout (code %s). Not calling /token until "
"username or password changes, or after %s.",
source.id,
code,
until.isoformat(),
)
def sd_clear_auth_lockout(source):
"""Clear a persisted auth lockout (success, credentials changed, or expired)."""
if source is None:
return
cp = dict(source.custom_properties or {})
changed = False
for key in (
'sd_auth_lockout',
'sd_auth_lockout_code',
'sd_auth_lockout_reason',
'sd_auth_lockout_until',
'sd_auth_lockout_credential_fp',
):
if key in cp:
cp.pop(key, None)
changed = True
if changed:
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
@dataclass(frozen=True)
class SDTokenAuthResult:
"""Outcome of POST /token (or a short-circuit from a persisted lockout)."""
ok: bool
token: str | None = None
token_expires: float | None = None
code: int | None = None
message: str = ''
soft: bool = False
debug_rejected: bool = False
lockout: bool = False
_DEBUG_REJECTED_MESSAGE = (
"Schedules Direct rejected the debug routing header (code 2055). "
"Extra Schedules Direct Debugging has been turned off. Retry without it "
"unless Schedules Direct support asked you to enable it."
)
def sd_obtain_token(source, username=None, password=None, *, timeout=30):
"""
Authenticate with Schedules Direct and return a session token.
Shared by refresh, lineup/form auth, and the poster proxy. Honors persisted
lockouts, reads JSON ``code`` before HTTP status, and persists cooldowns for
codes that must not be retried until the user or cooldown clears them.
"""
if source is None:
return SDTokenAuthResult(
ok=False,
message='Schedules Direct source is required.',
)
if username is None:
username = (source.username or '').strip()
else:
username = (username or '').strip()
if password is None:
password = (source.password or '').strip()
else:
password = (password or '').strip()
if not username or not password:
return SDTokenAuthResult(
ok=False,
message='Username and password are required.',
)
active, reason, lockout_code = sd_auth_lockout_active(
source, username, password
)
if active:
msg = f"{reason} {sd_auth_lockout_retry_message()}"
return SDTokenAuthResult(
ok=False,
code=lockout_code,
message=msg,
soft=lockout_code in SD_AUTH_SOFT_CODES if lockout_code else False,
lockout=True,
)
sha1_password = hashlib.sha1(password.encode('utf-8')).hexdigest()
try:
response = requests.post(
f"{SD_BASE_URL}/token",
json={'username': username, 'password': sha1_password},
headers=sd_headers_for_source(source),
timeout=timeout,
)
except requests.exceptions.RequestException as exc:
return SDTokenAuthResult(
ok=False,
message=f'Network error authenticating with Schedules Direct: {exc}',
)
try:
auth_data = response.json()
except ValueError:
auth_data = {}
if sd_handle_2055(source, auth_data):
return SDTokenAuthResult(
ok=False,
code=SD_CODE_INVALID_DEBUG,
message=_DEBUG_REJECTED_MESSAGE,
debug_rejected=True,
)
# Honor JSON ``code`` before raise_for_status. SD error semantics are in
# the body; HTTP status alone is not reliable for auth failures.
auth_code = sd_token_response_code(auth_data)
if auth_code != 0:
msg = sd_auth_failure_message(
auth_code, auth_data.get('message', 'Unknown error')
)
soft = auth_code in SD_AUTH_SOFT_CODES
locked = False
if auth_code in SD_AUTH_LOCKOUT_CODES:
sd_save_auth_lockout(
source, auth_code, username, password, reason=msg
)
msg = f"{msg} {sd_auth_lockout_retry_message()}"
locked = True
return SDTokenAuthResult(
ok=False,
code=auth_code,
message=msg,
soft=soft,
lockout=locked,
)
try:
response.raise_for_status()
except requests.exceptions.RequestException as exc:
return SDTokenAuthResult(
ok=False,
message=f'Network error authenticating with Schedules Direct: {exc}',
)
token = auth_data.get('token') if isinstance(auth_data, dict) else None
if not token:
return SDTokenAuthResult(
ok=False,
message='Schedules Direct returned no token.',
)
sd_clear_auth_lockout(source)
expires = None
if isinstance(auth_data, dict):
expires = auth_data.get('tokenExpires')
if not isinstance(expires, (int, float)):
expires = time.time() + _SD_TOKEN_DEFAULT_TTL_SECONDS
expires = float(expires)
sd_set_cached_token(
source.id, token, expires, username=username, password=password
)
return SDTokenAuthResult(
ok=True,
token=token,
token_expires=expires,
code=0,
)
def sd_next_midnight_utc():
"""Return the next Schedules Direct counter reset (00:00Z)."""
now = timezone.now()
return (now + timedelta(days=1)).replace(
hour=0, minute=0, second=0, microsecond=0,
tzinfo=dt_timezone.utc,
)
def sd_headers_for_source(source, *, token=None, content_type='application/json'):
"""
Build outbound headers for Schedules Direct requests for this source.
When Extra Schedules Direct Debugging is enabled, adds RouteTo: debug so the
SD load balancer can steer traffic to their debug server (support-coordinated).
"""
cp = source.custom_properties or {} if source is not None else {}
route_to = 'debug' if cp.get('sd_extra_debugging') else None
return dispatcharr_http_headers(
token=token,
content_type=content_type,
route_to=route_to,
)
def sd_disable_extra_debugging(source):
"""Clear the user-facing debug toggle after SD returns code 2055."""
if source is None:
return False
cp = dict(source.custom_properties or {})
if not cp.get('sd_extra_debugging'):
return False
cp['sd_extra_debugging'] = False
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
logger.warning(
"SD source %s: received code 2055 (unexpected debug connection). "
"Disabled Extra Schedules Direct Debugging.",
source.id,
)
return True
def sd_handle_2055(source, data):
"""
If data is an SD JSON error with code 2055, disable debug routing.
Returns True when 2055 was handled.
"""
if not isinstance(data, dict):
return False
if data.get('code') != SD_CODE_INVALID_DEBUG:
return False
sd_disable_extra_debugging(source)
return True
def sd_parse_response_payload(response):
"""Return (code, data_dict_or_None) for an SD HTTP response body."""
if response is None:
return None, None
content_type = (response.headers.get('Content-Type') or '').lower()
body = response.content or b''
looks_json = (
'json' in content_type
or body.lstrip()[:1] in (b'{', b'[')
)
if not looks_json:
return None, None
try:
data = response.json()
except (ValueError, json.JSONDecodeError):
return None, None
if not isinstance(data, dict):
return None, None
code = data.get('code')
return (code if isinstance(code, int) else None), data
def sd_image_limit_active(source):
"""
Return (active: bool, reason: str|None) for a persisted image download lockout.
Clears the lockout automatically once the next midnight UTC has passed.
"""
if source is None:
return False, None
cp = source.custom_properties or {}
if not cp.get('sd_image_limit_hit'):
return False, None
reset_at_str = cp.get('sd_image_limit_reset_at')
reset_at = parse_datetime(reset_at_str) if reset_at_str else None
if reset_at and timezone.now() >= reset_at:
sd_clear_image_limit_lockout(source)
return False, None
reason = cp.get('sd_image_limit_reason') or (
'Daily image download limit reached'
)
return True, reason
def sd_save_image_limit_lockout(source, code):
"""
Persist a source-wide image download lockout until next 00:00Z.
Used for SD codes 5002 (subscriber) and 5003 (trial).
"""
if source is None:
return
reset_at = sd_next_midnight_utc()
reason = (
f'Daily image download limit reached (SD error {code})'
if code
else 'Daily image download limit reached'
)
cp = dict(source.custom_properties or {})
cp['sd_image_limit_hit'] = True
cp['sd_image_limit_reset_at'] = reset_at.isoformat()
cp['sd_image_limit_reason'] = reason
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
logger.warning(
"SD source %s: image download limit (code %s). Lockout until %s.",
source.id,
code,
reset_at.isoformat(),
)
def sd_clear_image_limit_lockout(source):
"""Clear a persisted image download lockout after midnight UTC."""
if source is None:
return
cp = dict(source.custom_properties or {})
changed = False
for key in (
'sd_image_limit_hit',
'sd_image_limit_reset_at',
'sd_image_limit_reason',
):
if key in cp:
cp.pop(key, None)
changed = True
if changed:
source.custom_properties = cp
source.save(update_fields=['custom_properties'])
def sd_mark_icon_missing(program):
"""
Clear a bad image URI so we never re-request it (SD code 5000).
Continues requesting the same missing URI can accumulate toward code 5004
and get the account blocked.
"""
if program is None:
return
cp = dict(program.custom_properties or {})
if 'sd_icon' not in cp and cp.get('sd_icon_missing'):
return
cp.pop('sd_icon', None)
cp['sd_icon_missing'] = True
program.custom_properties = cp
program.save(update_fields=['custom_properties'])
logger.info(
"SD program %s: IMAGE_NOT_FOUND (5000). Cleared sd_icon to avoid retries.",
program.id,
)

18
apps/epg/serializers.py
View file

@ -1,3 +1,8 @@
from apps.epg.sd_utils import (
sd_clear_cached_token,
sd_credential_fingerprint,
)
from apps.epg.utils import sd_poster_proxy_path
from core.utils import validate_flexible_url, build_absolute_uri_with_port
from rest_framework import serializers
from .models import EPGSource, EPGData, ProgramData
@ -67,11 +72,16 @@ class EPGSourceSerializer(serializers.ModelSerializer):
ct = instance.refresh_task.crontab
cron_expr = f'{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}'
instance._cron_expression = cron_expr
prior_fp = sd_credential_fingerprint(instance.username, instance.password)
for attr, value in validated_data.items():
if attr == 'password' and not value:
continue
setattr(instance, attr, value)
instance.save()
# Drop any Redis SD session tied to the previous username/password so
# poster traffic cannot keep using another account's token.
if prior_fp != sd_credential_fingerprint(instance.username, instance.password):
sd_clear_cached_token(instance.id)
return instance
def create(self, validated_data):
@ -169,9 +179,11 @@ class ProgramDetailSerializer(ProgramDataSerializer):
data['icon'] = cp.get('icon')
data['images'] = cp.get('images') or []
# SD poster: expose as absolute proxy URL so frontend/img tags never need SD auth
if cp.get('sd_icon'):
poster_path = f"/api/epg/programs/{obj.id}/poster/"
# SD poster: expose as absolute proxy URL so frontend/img tags never need SD auth.
# ``?v=`` tracks the sd_icon URI so nginx/browser caches bust when artwork changes.
sd_icon = cp.get('sd_icon')
if sd_icon:
poster_path = sd_poster_proxy_path(obj.id, sd_icon)
request = self.context.get('request')
if request:
data['poster_url'] = build_absolute_uri_with_port(request, poster_path)

1838
apps/epg/tasks.py
View file

File diff suppressed because it is too large Load diff

834
apps/epg/tests/test_schedules_direct.py
View file

@ -14,6 +14,7 @@ Covers:
"""
import hashlib
import time
from datetime import date, datetime, timedelta, timezone as dt_timezone
from unittest.mock import MagicMock, patch
@ -128,8 +129,8 @@ class FetchSchedulesDirectCredentialTests(TestCase):
class FetchSchedulesDirectAuthTests(TestCase):
"""fetch_schedules_direct must SHA1-hash the password before sending."""
@patch('apps.epg.tasks.requests.post')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.sd_tasks.requests.post')
@patch('apps.epg.sd_tasks.requests.get')
def test_password_sha1_hashed_in_token_request(self, mock_get, mock_post):
"""The token POST body must contain the SHA1 hash of the plaintext password."""
plaintext = 'mysecretpassword'
@ -153,7 +154,7 @@ class FetchSchedulesDirectAuthTests(TestCase):
password=plaintext,
)
with patch('apps.epg.tasks.send_epg_update'):
with patch('apps.epg.sd_tasks.send_epg_update'):
fetch_schedules_direct(source)
# Verify the POST was called and the body contained the hash
@ -163,14 +164,251 @@ class FetchSchedulesDirectAuthTests(TestCase):
self.assertEqual(posted_json.get('password'), expected_hash)
self.assertEqual(posted_json.get('username'), 'sduser')
@patch('apps.epg.tasks.requests.post')
def test_auth_failure_sets_error_status(self, mock_post):
"""A non-zero SD response code must set STATUS_ERROR on the source."""
class SDLineupChangesRemainingTests(TestCase):
"""Persisting changesRemaining=0 must include a midnight-UTC reset timestamp."""
def test_save_zero_remaining_sets_reset_at(self):
from apps.epg.api_views import EPGSourceViewSet
from apps.epg.sd_utils import sd_next_midnight_utc
source = EPGSource.objects.create(
name='SD Changes Remaining',
source_type='schedules_direct',
username='u',
password='p',
)
view = EPGSourceViewSet()
view._save_sd_changes_remaining(source, 0)
source.refresh_from_db()
self.assertEqual(source.custom_properties.get('sd_changes_remaining'), 0)
self.assertEqual(
source.custom_properties.get('sd_changes_reset_at'),
sd_next_midnight_utc().isoformat(),
)
def test_save_positive_remaining_clears_reset_at(self):
from apps.epg.api_views import EPGSourceViewSet
source = EPGSource.objects.create(
name='SD Changes Unlock',
source_type='schedules_direct',
username='u',
password='p',
custom_properties={
'sd_changes_remaining': 0,
'sd_changes_reset_at': '2099-01-01T00:00:00+00:00',
},
)
view = EPGSourceViewSet()
view._save_sd_changes_remaining(source, 3)
source.refresh_from_db()
self.assertEqual(source.custom_properties.get('sd_changes_remaining'), 3)
self.assertNotIn('sd_changes_reset_at', source.custom_properties or {})
def test_lockout_uses_shared_save_path(self):
from apps.epg.api_views import EPGSourceViewSet
from apps.epg.sd_utils import sd_next_midnight_utc
source = EPGSource.objects.create(
name='SD Lockout',
source_type='schedules_direct',
username='u',
password='p',
)
view = EPGSourceViewSet()
view._save_sd_lockout(source)
source.refresh_from_db()
self.assertEqual(source.custom_properties.get('sd_changes_remaining'), 0)
self.assertEqual(
source.custom_properties.get('sd_changes_reset_at'),
sd_next_midnight_utc().isoformat(),
)
class SDAuthCredentialLockoutTests(TestCase):
"""4002/4003 lockouts must stop /token until credentials change or 24h."""
def test_lockout_active_until_password_changes(self):
from apps.epg.sd_utils import (
sd_auth_lockout_active,
sd_save_auth_lockout,
)
source = EPGSource.objects.create(
name='SD Cred Lockout Utils',
source_type='schedules_direct',
username='u',
password='old',
)
sd_save_auth_lockout(source, 4003, 'u', 'old')
active, reason, code = sd_auth_lockout_active(source, 'u', 'old')
self.assertTrue(active)
self.assertEqual(code, 4003)
self.assertIn('invalid username or password', reason.lower())
active, reason, code = sd_auth_lockout_active(source, 'u', 'new')
self.assertFalse(active)
source.refresh_from_db()
self.assertFalse((source.custom_properties or {}).get('sd_auth_lockout'))
def test_lockout_clears_when_username_changes(self):
"""4003 is username or password; either field change must allow retry."""
from apps.epg.sd_utils import (
sd_auth_lockout_active,
sd_save_auth_lockout,
)
source = EPGSource.objects.create(
name='SD Cred Lockout Username',
source_type='schedules_direct',
username='olduser',
password='samepass',
)
sd_save_auth_lockout(source, 4003, 'olduser', 'samepass')
self.assertTrue(sd_auth_lockout_active(source, 'olduser', 'samepass')[0])
active, _, _ = sd_auth_lockout_active(source, 'newuser', 'samepass')
self.assertFalse(active)
source.refresh_from_db()
self.assertFalse((source.custom_properties or {}).get('sd_auth_lockout'))
def test_lockout_expires_after_24_hours(self):
"""Auto-clear after 24h so a transient SD-side failure can retry."""
from apps.epg.sd_utils import (
sd_auth_lockout_active,
sd_save_auth_lockout,
)
source = EPGSource.objects.create(
name='SD Cred Lockout Expiry',
source_type='schedules_direct',
username='u',
password='p',
)
frozen_now = timezone.now()
with patch('apps.epg.sd_utils.timezone.now', return_value=frozen_now):
sd_save_auth_lockout(source, 4003, 'u', 'p')
self.assertTrue(sd_auth_lockout_active(source, 'u', 'p')[0])
source.refresh_from_db()
self.assertTrue(
(source.custom_properties or {}).get('sd_auth_lockout_until')
)
later = frozen_now + timedelta(hours=24, seconds=1)
with patch('apps.epg.sd_utils.timezone.now', return_value=later):
active, _, _ = sd_auth_lockout_active(source, 'u', 'p')
self.assertFalse(active)
source.refresh_from_db()
self.assertFalse((source.custom_properties or {}).get('sd_auth_lockout'))
def test_account_expired_4001_locks_for_24_hours(self):
from django.utils.dateparse import parse_datetime
from apps.epg.sd_utils import (
SD_AUTH_LOCKOUT_SECONDS,
sd_auth_lockout_seconds_for_code,
sd_save_auth_lockout,
)
self.assertEqual(sd_auth_lockout_seconds_for_code(4001), SD_AUTH_LOCKOUT_SECONDS)
self.assertEqual(sd_auth_lockout_seconds_for_code(4004), 15 * 60)
self.assertEqual(sd_auth_lockout_seconds_for_code(3000), 3600)
source = EPGSource.objects.create(
name='SD Expired Lockout',
source_type='schedules_direct',
username='u',
password='p',
)
frozen_now = timezone.now()
with patch('apps.epg.sd_utils.timezone.now', return_value=frozen_now):
sd_save_auth_lockout(source, 4001, 'u', 'p')
source.refresh_from_db()
self.assertEqual(source.custom_properties.get('sd_auth_lockout_code'), 4001)
until = parse_datetime(source.custom_properties['sd_auth_lockout_until'])
self.assertAlmostEqual(
(until - frozen_now).total_seconds(),
24 * 3600,
delta=2,
)
@patch('requests.post')
def test_sd_authenticate_http_400_with_4003_locks_out(self, mock_post):
"""Lineup/form auth must persist lockout on HTTP 400 + code 4003."""
import requests
from apps.epg.api_views import EPGSourceViewSet
mock_resp = MagicMock(
status_code=400,
json=MagicMock(return_value={
'code': 4003,
'message': 'Invalid user or password',
}),
)
mock_resp.raise_for_status.side_effect = requests.exceptions.HTTPError(
'400 Client Error', response=mock_resp
)
mock_post.return_value = mock_resp
source = EPGSource.objects.create(
name='SD Form Auth Fail',
source_type='schedules_direct',
username='baduser',
password='badpass',
)
view = EPGSourceViewSet()
token, error = view._sd_authenticate(source)
self.assertIsNone(token)
self.assertEqual(error.status_code, 401)
self.assertIn('invalid username or password', error.data['error'].lower())
source.refresh_from_db()
self.assertTrue((source.custom_properties or {}).get('sd_auth_lockout'))
# Second call must not hit /token.
mock_post.reset_mock()
token, error = view._sd_authenticate(source)
self.assertIsNone(token)
mock_post.assert_not_called()
class FetchSchedulesDirectAuthCodeTests(TestCase):
"""Token response codes must map to idle vs error correctly."""
@patch('apps.epg.sd_tasks.requests.post')
def test_auth_service_offline_sets_idle_status(self, mock_post):
"""Token code 3000 (SERVICE_OFFLINE) must stop as idle, not a credential error."""
mock_post.return_value = MagicMock(
status_code=200,
json=MagicMock(return_value={
'code': 3000,
'message': 'Invalid credentials',
'message': 'Server offline for maintenance.',
}),
)
from apps.epg.tasks import fetch_schedules_direct
source = EPGSource.objects.create(
name='SD Auth Offline',
source_type='schedules_direct',
username='user',
password='pass',
)
with patch('apps.epg.sd_tasks.send_epg_update'):
fetch_schedules_direct(source)
source.refresh_from_db()
self.assertEqual(source.status, EPGSource.STATUS_IDLE)
self.assertIn('offline', source.last_message.lower())
@patch('apps.epg.sd_tasks.requests.post')
def test_auth_failure_sets_error_status(self, mock_post):
"""A non-zero credential error code must set STATUS_ERROR on the source."""
mock_post.return_value = MagicMock(
status_code=200,
json=MagicMock(return_value={
'code': 4003,
'message': 'Invalid username or password.',
}),
)
@ -182,13 +420,124 @@ class FetchSchedulesDirectAuthTests(TestCase):
password='badpass',
)
with patch('apps.epg.tasks.send_epg_update'):
with patch('apps.epg.sd_tasks.send_epg_update'):
fetch_schedules_direct(source)
source.refresh_from_db()
self.assertEqual(source.status, EPGSource.STATUS_ERROR)
self.assertIn('invalid username or password', source.last_message.lower())
self.assertTrue((source.custom_properties or {}).get('sd_auth_lockout'))
@patch('apps.epg.tasks.requests.post')
@patch('apps.epg.sd_tasks.requests.post')
def test_auth_4003_http_400_persists_lockout_without_raise(self, mock_post):
"""HTTP 400 + JSON code 4003 must hard-stop before raise_for_status."""
import requests
mock_resp = MagicMock(
status_code=400,
json=MagicMock(return_value={
'code': 4003,
'message': 'Invalid user or password',
}),
)
mock_resp.raise_for_status.side_effect = requests.exceptions.HTTPError(
'400 Client Error', response=mock_resp
)
mock_post.return_value = mock_resp
from apps.epg.tasks import fetch_schedules_direct
source = EPGSource.objects.create(
name='SD Auth HTTP 400',
source_type='schedules_direct',
username='baduser',
password='badpass',
)
with patch('apps.epg.sd_tasks.send_epg_update'):
fetch_schedules_direct(source)
source.refresh_from_db()
self.assertEqual(source.status, EPGSource.STATUS_ERROR)
self.assertIn('invalid username or password', source.last_message.lower())
self.assertTrue((source.custom_properties or {}).get('sd_auth_lockout'))
mock_resp.raise_for_status.assert_not_called()
@patch('apps.epg.sd_tasks.requests.post')
def test_auth_lockout_skips_token_until_credentials_change(self, mock_post):
"""Persisted 4003 lockout must not call /token again until credentials change."""
from apps.epg.sd_utils import sd_save_auth_lockout
from apps.epg.tasks import fetch_schedules_direct
source = EPGSource.objects.create(
name='SD Auth Lockout',
source_type='schedules_direct',
username='baduser',
password='badpass',
)
sd_save_auth_lockout(source, 4003, 'baduser', 'badpass')
with patch('apps.epg.sd_tasks.send_epg_update'):
fetch_schedules_direct(source)
mock_post.assert_not_called()
source.refresh_from_db()
self.assertEqual(source.status, EPGSource.STATUS_ERROR)
self.assertIn('not retrying', source.last_message.lower())
# Changing credentials clears the fingerprint lockout and allows /token.
source.password = 'newpass'
source.save(update_fields=['password'])
mock_post.return_value = MagicMock(
status_code=200,
json=MagicMock(return_value={
'code': 0,
'token': 'new-token',
'tokenExpires': time.time() + 86400,
}),
raise_for_status=MagicMock(),
)
with patch('apps.epg.sd_tasks.send_epg_update'), \
patch('apps.epg.sd_tasks.requests.get') as mock_get:
mock_get.return_value = MagicMock(
status_code=200,
json=MagicMock(return_value={
'systemStatus': [{'status': 'Offline'}],
}),
raise_for_status=MagicMock(),
)
fetch_schedules_direct(source)
mock_post.assert_called_once()
source.refresh_from_db()
self.assertFalse((source.custom_properties or {}).get('sd_auth_lockout'))
@patch('apps.epg.sd_tasks.requests.post')
def test_auth_too_many_ips_sets_error_status(self, mock_post):
"""Code 4010 must surface a clear multi-IP message and stop."""
mock_post.return_value = MagicMock(
status_code=200,
json=MagicMock(return_value={
'code': 4010,
'message': 'Exceeded maximum number of unique IP addresses in 24 hours.',
}),
)
from apps.epg.tasks import fetch_schedules_direct
source = EPGSource.objects.create(
name='SD Too Many IPs',
source_type='schedules_direct',
username='user',
password='pass',
)
with patch('apps.epg.sd_tasks.send_epg_update'):
fetch_schedules_direct(source)
source.refresh_from_db()
self.assertEqual(source.status, EPGSource.STATUS_ERROR)
self.assertIn('unique IP', source.last_message)
@patch('apps.epg.sd_tasks.requests.post')
def test_network_error_sets_error_status(self, mock_post):
"""A network-level exception must set STATUS_ERROR and not crash."""
import requests as req_lib
@ -202,7 +551,7 @@ class FetchSchedulesDirectAuthTests(TestCase):
password='pass',
)
with patch('apps.epg.tasks.send_epg_update'):
with patch('apps.epg.sd_tasks.send_epg_update'):
fetch_schedules_direct(source) # Must not raise
source.refresh_from_db()
@ -212,9 +561,9 @@ class FetchSchedulesDirectAuthTests(TestCase):
class FetchSchedulesDirectStationsOnlyTests(TestCase):
"""stations_only fetch must signal channel parsing completion to the frontend."""
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
@patch('apps.epg.sd_tasks.send_epg_update')
@patch('apps.epg.sd_tasks.requests.get')
@patch('apps.epg.sd_tasks.requests.post')
def test_stations_only_sends_parsing_channels_complete(
self, mock_post, mock_get, mock_send_epg_update
):
@ -492,9 +841,9 @@ class SDScheduleDeltaIntegrationTests(TestCase):
)
@patch('apps.epg.tasks.SD_DAYS_TO_FETCH', 3)
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
@patch('apps.epg.sd_tasks.send_epg_update')
@patch('apps.epg.sd_tasks.requests.get')
@patch('apps.epg.sd_tasks.requests.post')
def test_md5_api_only_requests_mapped_stations(
self, mock_post, mock_get, mock_send_epg_update,
):
@ -574,9 +923,9 @@ class SDScheduleDeltaIntegrationTests(TestCase):
self.assertEqual(source.status, EPGSource.STATUS_SUCCESS)
@patch('apps.epg.tasks.SD_DAYS_TO_FETCH', 3)
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
@patch('apps.epg.sd_tasks.send_epg_update')
@patch('apps.epg.sd_tasks.requests.get')
@patch('apps.epg.sd_tasks.requests.post')
def test_newly_mapped_station_fetches_despite_stale_cache(
self, mock_post, mock_get, mock_send_epg_update,
):
@ -652,9 +1001,9 @@ class SDScheduleDeltaIntegrationTests(TestCase):
)
@patch('apps.epg.tasks.SD_DAYS_TO_FETCH', 3)
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
@patch('apps.epg.sd_tasks.send_epg_update')
@patch('apps.epg.sd_tasks.requests.get')
@patch('apps.epg.sd_tasks.requests.post')
def test_orphan_program_data_removed_on_post_refresh(
self, mock_post, mock_get, mock_send_epg_update,
):
@ -793,7 +1142,7 @@ class SDDispatchProgramRefreshTests(TestCase):
url='http://example.com/epg.xml',
)
@patch('apps.epg.tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.sd_tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.tasks.parse_programs_for_tvg_id.delay')
def test_xmltv_still_uses_parse_programs_per_id(
self, mock_parse_delay, mock_batch_delay,
@ -813,7 +1162,7 @@ class SDDispatchProgramRefreshTests(TestCase):
mock_parse_delay.assert_called_once_with(epg.id)
mock_batch_delay.assert_not_called()
@patch('apps.epg.tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.sd_tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.tasks.parse_programs_for_tvg_id.delay')
def test_sd_below_threshold_uses_per_epg_tasks(
self, mock_parse_delay, mock_batch_delay,
@ -836,7 +1185,7 @@ class SDDispatchProgramRefreshTests(TestCase):
self.assertEqual(mock_parse_delay.call_count, 2)
mock_batch_delay.assert_not_called()
@patch('apps.epg.tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.sd_tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.tasks.parse_programs_for_tvg_id.delay')
def test_sd_at_threshold_uses_batched_fetch(
self, mock_parse_delay, mock_batch_delay,
@ -859,7 +1208,7 @@ class SDDispatchProgramRefreshTests(TestCase):
mock_batch_delay.assert_called_once_with(source.id)
mock_parse_delay.assert_not_called()
@patch('apps.epg.tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.sd_tasks.fetch_sd_mapped_guide_batch.delay')
@patch('apps.epg.tasks.parse_programs_for_tvg_id.delay')
def test_sd_skips_when_program_data_exists(
self, mock_parse_delay, mock_batch_delay,
@ -901,9 +1250,9 @@ class SDGuideFetchCoordinationTests(TestCase):
password='sdpass',
)
@patch('apps.epg.tasks.fetch_schedules_direct')
@patch('apps.epg.tasks.acquire_task_lock', return_value=False)
@patch('apps.epg.tasks.fetch_sd_mapped_guide_batch.apply_async')
@patch('apps.epg.sd_tasks.fetch_schedules_direct')
@patch('apps.epg.sd_tasks.acquire_task_lock', return_value=False)
@patch('apps.epg.sd_tasks.fetch_sd_mapped_guide_batch.apply_async')
def test_batch_fetch_defers_when_lock_held(
self, mock_apply_async, mock_acquire, mock_fetch,
):
@ -923,9 +1272,9 @@ class SDGuideFetchCoordinationTests(TestCase):
)
mock_fetch.assert_not_called()
@patch('apps.epg.tasks.fetch_schedules_direct')
@patch('apps.epg.tasks.acquire_task_lock', return_value=False)
@patch('apps.epg.tasks.fetch_sd_mapped_guide_batch.apply_async')
@patch('apps.epg.sd_tasks.fetch_schedules_direct')
@patch('apps.epg.sd_tasks.acquire_task_lock', return_value=False)
@patch('apps.epg.sd_tasks.fetch_sd_mapped_guide_batch.apply_async')
def test_batch_fetch_stops_after_max_defer_retries(
self, mock_apply_async, mock_acquire, mock_fetch,
):
@ -938,12 +1287,12 @@ class SDGuideFetchCoordinationTests(TestCase):
mock_apply_async.assert_not_called()
mock_fetch.assert_not_called()
@patch('apps.epg.tasks.fetch_schedules_direct')
@patch('apps.epg.tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.tasks.release_task_lock')
@patch('apps.epg.tasks.TaskLockRenewer')
@patch('apps.epg.tasks.is_task_lock_held', return_value=True)
@patch('apps.epg.tasks.fetch_sd_guide_for_epg.apply_async')
@patch('apps.epg.sd_tasks.fetch_schedules_direct')
@patch('apps.epg.sd_tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.sd_tasks.release_task_lock')
@patch('apps.epg.sd_tasks.TaskLockRenewer')
@patch('apps.epg.sd_tasks.is_task_lock_held', return_value=True)
@patch('apps.epg.sd_tasks.fetch_sd_guide_for_epg.apply_async')
def test_single_epg_defers_while_batch_running(
self, mock_apply_async, mock_batch_held, mock_renewer,
mock_release, mock_acquire, mock_fetch,
@ -971,12 +1320,12 @@ class SDGuideFetchCoordinationTests(TestCase):
mock_fetch.assert_not_called()
mock_acquire.assert_not_called()
@patch('apps.epg.tasks.fetch_schedules_direct')
@patch('apps.epg.tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.tasks.release_task_lock')
@patch('apps.epg.tasks.TaskLockRenewer')
@patch('apps.epg.tasks.is_task_lock_held', return_value=True)
@patch('apps.epg.tasks.fetch_sd_guide_for_epg.apply_async')
@patch('apps.epg.sd_tasks.fetch_schedules_direct')
@patch('apps.epg.sd_tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.sd_tasks.release_task_lock')
@patch('apps.epg.sd_tasks.TaskLockRenewer')
@patch('apps.epg.sd_tasks.is_task_lock_held', return_value=True)
@patch('apps.epg.sd_tasks.fetch_sd_guide_for_epg.apply_async')
def test_single_epg_proceeds_after_max_batch_deferrals(
self, mock_apply_async, mock_batch_held, mock_renewer,
mock_release, mock_acquire, mock_fetch,
@ -1032,9 +1381,9 @@ class SDSingleEpgFetchTests(TestCase):
)
raise AssertionError(f'Unexpected GET URL: {url}')
@patch('apps.epg.tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.tasks.release_task_lock')
@patch('apps.epg.tasks.TaskLockRenewer')
@patch('apps.epg.sd_tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.sd_tasks.release_task_lock')
@patch('apps.epg.sd_tasks.TaskLockRenewer')
def test_fetch_sd_guide_skips_when_program_data_exists(
self, mock_renewer, mock_release, mock_acquire,
):
@ -1055,15 +1404,15 @@ class SDSingleEpgFetchTests(TestCase):
tvg_id=epg.tvg_id,
)
with patch('apps.epg.tasks.fetch_schedules_direct') as mock_fetch:
with patch('apps.epg.sd_tasks.fetch_schedules_direct') as mock_fetch:
result = fetch_sd_guide_for_epg(epg.id)
self.assertEqual(result, 'Guide data already present')
mock_fetch.assert_not_called()
@patch('apps.epg.tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.tasks.release_task_lock')
@patch('apps.epg.tasks.TaskLockRenewer')
@patch('apps.epg.sd_tasks.acquire_task_lock', return_value=True)
@patch('apps.epg.sd_tasks.release_task_lock')
@patch('apps.epg.sd_tasks.TaskLockRenewer')
def test_parse_programs_for_tvg_id_delegates_to_sd_fetch(
self, mock_renewer, mock_release, mock_acquire,
):
@ -1083,9 +1432,9 @@ class SDSingleEpgFetchTests(TestCase):
self.assertEqual(result, 'SD guide fetch complete')
@patch('apps.epg.tasks.SD_DAYS_TO_FETCH', 3)
@patch('apps.epg.tasks.send_epg_update')
@patch('apps.epg.tasks.requests.get')
@patch('apps.epg.tasks.requests.post')
@patch('apps.epg.sd_tasks.send_epg_update')
@patch('apps.epg.sd_tasks.requests.get')
@patch('apps.epg.sd_tasks.requests.post')
def test_single_epg_fetch_skips_lineup_sync_and_updated_at(
self, mock_post, mock_get, mock_send_epg_update,
):
@ -1355,3 +1704,380 @@ class SDPosterSelectionTests(TestCase):
_sd_pick_poster_url(images, 'square_iconic'),
'assets/landscape_primary.jpg',
)
# ---------------------------------------------------------------------------
# SD helpers and poster proxy error handling
# ---------------------------------------------------------------------------
class SDUtilsTests(TestCase):
"""Unit tests for apps.epg.sd_utils helpers."""
def test_headers_include_routeto_when_extra_debugging_enabled(self):
from apps.epg.sd_utils import sd_headers_for_source
source = EPGSource.objects.create(
name='SD Debug Headers',
source_type='schedules_direct',
username='u',
password='p',
custom_properties={'sd_extra_debugging': True},
)
headers = sd_headers_for_source(source, token='tok', content_type=None)
self.assertEqual(headers.get('RouteTo'), 'debug')
self.assertEqual(headers.get('token'), 'tok')
self.assertNotIn('Content-Type', headers)
def test_headers_omit_routeto_by_default(self):
from apps.epg.sd_utils import sd_headers_for_source
source = EPGSource.objects.create(
name='SD No Debug',
source_type='schedules_direct',
username='u',
password='p',
)
headers = sd_headers_for_source(source)
self.assertNotIn('RouteTo', headers)
def test_2055_disables_extra_debugging(self):
from apps.epg.sd_utils import sd_handle_2055
source = EPGSource.objects.create(
name='SD 2055',
source_type='schedules_direct',
username='u',
password='p',
custom_properties={'sd_extra_debugging': True},
)
self.assertTrue(sd_handle_2055(source, {
'response': 'INVALID_PARAMETER:DEBUG',
'code': 2055,
'message': 'Unexpected debug connection from client.',
}))
source.refresh_from_db()
self.assertFalse(source.custom_properties.get('sd_extra_debugging'))
def test_image_limit_lockout_persists_until_midnight_utc(self):
from apps.epg.sd_utils import (
sd_image_limit_active,
sd_next_midnight_utc,
sd_save_image_limit_lockout,
)
source = EPGSource.objects.create(
name='SD Limit',
source_type='schedules_direct',
username='u',
password='p',
)
sd_save_image_limit_lockout(source, 5002)
source.refresh_from_db()
active, reason = sd_image_limit_active(source)
self.assertTrue(active)
self.assertIn('5002', reason)
self.assertEqual(
source.custom_properties.get('sd_image_limit_reset_at'),
sd_next_midnight_utc().isoformat(),
)
def test_token_cache_round_trip(self):
from apps.epg.sd_utils import (
sd_clear_cached_token,
sd_get_cached_token,
sd_set_cached_token,
)
source_id = 4242
user, password = 'sduser', 'sdpass'
sd_clear_cached_token(source_id)
self.assertIsNone(sd_get_cached_token(source_id, user, password))
self.assertTrue(
sd_set_cached_token(
source_id, 'tok-abc', time.time() + 3600,
username=user, password=password,
)
)
self.assertEqual(sd_get_cached_token(source_id, user, password), 'tok-abc')
sd_clear_cached_token(source_id)
self.assertIsNone(sd_get_cached_token(source_id, user, password))
def test_token_cache_ignores_near_expiry(self):
from apps.epg.sd_utils import (
sd_clear_cached_token,
sd_get_cached_token,
sd_set_cached_token,
)
source_id = 4243
user, password = 'sduser', 'sdpass'
sd_clear_cached_token(source_id)
# Within skew window: set should refuse or get should miss.
self.assertFalse(
sd_set_cached_token(
source_id, 'tok-soon', time.time() + 30,
username=user, password=password,
)
)
self.assertIsNone(sd_get_cached_token(source_id, user, password))
def test_token_cache_misses_after_username_change(self):
"""Cached token must not be reused after switching SD accounts."""
from apps.epg.sd_utils import (
sd_clear_cached_token,
sd_get_cached_token,
sd_set_cached_token,
)
source_id = 4244
sd_clear_cached_token(source_id)
self.assertTrue(
sd_set_cached_token(
source_id, 'tok-account-a', time.time() + 3600,
username='account-a', password='pass-a',
)
)
self.assertEqual(
sd_get_cached_token(source_id, 'account-a', 'pass-a'),
'tok-account-a',
)
self.assertIsNone(
sd_get_cached_token(source_id, 'account-b', 'pass-a')
)
# Mismatch clears the stale entry so account B cannot resurrect it.
self.assertIsNone(
sd_get_cached_token(source_id, 'account-a', 'pass-a')
)
def test_serializer_username_change_clears_token_cache(self):
"""Updating an EPG source username must drop the Redis SD token."""
from apps.epg.serializers import EPGSourceSerializer
from apps.epg.sd_utils import (
sd_clear_cached_token,
sd_get_cached_token,
sd_set_cached_token,
)
source = EPGSource.objects.create(
name='SD Token Clear',
source_type='schedules_direct',
username='account-a',
password='pass-a',
)
sd_clear_cached_token(source.id)
self.assertTrue(
sd_set_cached_token(
source.id, 'tok-a', time.time() + 3600,
username='account-a', password='pass-a',
)
)
serializer = EPGSourceSerializer(
source, data={'username': 'account-b'}, partial=True
)
self.assertTrue(serializer.is_valid(), serializer.errors)
serializer.save()
self.assertIsNone(
sd_get_cached_token(source.id, 'account-a', 'pass-a')
)
self.assertIsNone(
sd_get_cached_token(source.id, 'account-b', 'pass-a')
)
class SDPosterProxyErrorHandlingTests(TestCase):
"""Poster proxy must honor SD image error codes so accounts are not blocked."""
def setUp(self):
from apps.epg.api_views import ProgramViewSet
from apps.epg.sd_utils import sd_clear_cached_token
from rest_framework.test import APIClient
ProgramViewSet._sd_poster_error_cache.clear()
self.client = APIClient()
self.source = EPGSource.objects.create(
name='SD Poster Source',
source_type='schedules_direct',
username='sduser',
password='sdpass',
)
sd_clear_cached_token(self.source.id)
self.epg = EPGData.objects.create(
tvg_id='station1',
name='Station 1',
epg_source=self.source,
)
self.program = ProgramData.objects.create(
epg=self.epg,
title='Show',
start_time=timezone.now(),
end_time=timezone.now() + timedelta(hours=1),
program_id='EP123456789012',
custom_properties={
'sd_icon': 'https://json.schedulesdirect.org/20141201/image/assets/test.jpg',
},
)
self.url = f'/api/epg/programs/{self.program.id}/poster/'
def tearDown(self):
from apps.epg.api_views import ProgramViewSet
from apps.epg.sd_utils import sd_clear_cached_token
ProgramViewSet._sd_poster_error_cache.clear()
sd_clear_cached_token(self.source.id)
def _auth_ok(self):
return MagicMock(
status_code=200,
json=MagicMock(return_value={
'code': 0,
'token': 'poster-tok',
'tokenExpires': time.time() + 86400,
}),
)
def _json_response(self, status_code, payload, content_type='application/json'):
import json as json_mod
body = json_mod.dumps(payload).encode('utf-8')
resp = MagicMock()
resp.status_code = status_code
resp.headers = {'Content-Type': content_type}
resp.content = body
resp.json = MagicMock(return_value=payload)
return resp
@patch('requests.get')
@patch('requests.post')
def test_http_200_json_5002_locks_until_midnight_and_blocks_retry(
self, mock_post, mock_get
):
mock_post.return_value = self._auth_ok()
mock_get.return_value = self._json_response(200, {
'response': 'MAX_IMAGE_DOWNLOADS',
'code': 5002,
'message': 'Maximum image downloads reached.',
})
first = self.client.get(self.url)
self.assertEqual(first.status_code, 429)
self.source.refresh_from_db()
self.assertTrue(self.source.custom_properties.get('sd_image_limit_hit'))
mock_get.reset_mock()
second = self.client.get(self.url)
self.assertEqual(second.status_code, 503)
mock_get.assert_not_called()
@patch('requests.get')
@patch('requests.post')
def test_http_200_json_5003_locks_out(self, mock_post, mock_get):
mock_post.return_value = self._auth_ok()
mock_get.return_value = self._json_response(200, {
'response': 'MAX_IMAGE_DOWNLOADS_TRIAL',
'code': 5003,
'message': 'Maximum image downloads for trial user reached.',
})
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, 429)
self.source.refresh_from_db()
self.assertTrue(self.source.custom_properties.get('sd_image_limit_hit'))
self.assertIn('5003', self.source.custom_properties.get('sd_image_limit_reason', ''))
@patch('requests.get')
@patch('requests.post')
def test_http_404_json_5000_clears_sd_icon(self, mock_post, mock_get):
mock_post.return_value = self._auth_ok()
mock_get.return_value = self._json_response(404, {
'response': 'IMAGE_NOT_FOUND',
'code': 5000,
'message': 'Could not find requested image.',
})
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, 404)
self.program.refresh_from_db()
cp = self.program.custom_properties or {}
self.assertNotIn('sd_icon', cp)
self.assertTrue(cp.get('sd_icon_missing'))
mock_get.reset_mock()
again = self.client.get(self.url)
self.assertEqual(again.status_code, 404)
mock_get.assert_not_called()
@patch('requests.get')
@patch('requests.post')
def test_bare_http_404_does_not_clear_sd_icon(self, mock_post, mock_get):
"""Transient CDN/S3 404 without SD code 5000 must not blacklist the URI."""
mock_post.return_value = self._auth_ok()
bare_404 = MagicMock()
bare_404.status_code = 404
bare_404.headers = {'Content-Type': 'text/plain'}
bare_404.content = b'Not Found'
bare_404.json = MagicMock(side_effect=ValueError('not json'))
mock_get.return_value = bare_404
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, 404)
self.program.refresh_from_db()
cp = self.program.custom_properties or {}
self.assertIn('sd_icon', cp)
self.assertFalse(cp.get('sd_icon_missing'))
@patch('requests.get')
@patch('requests.post')
def test_2055_disables_extra_debugging_on_auth(self, mock_post, mock_get):
self.source.custom_properties = {'sd_extra_debugging': True}
self.source.save(update_fields=['custom_properties'])
mock_post.return_value = self._json_response(200, {
'response': 'INVALID_PARAMETER:DEBUG',
'code': 2055,
'message': 'Unexpected debug connection from client.',
})
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, 400)
self.source.refresh_from_db()
self.assertFalse(self.source.custom_properties.get('sd_extra_debugging'))
mock_get.assert_not_called()
auth_headers = mock_post.call_args.kwargs.get('headers') or mock_post.call_args[1].get('headers')
self.assertEqual(auth_headers.get('RouteTo'), 'debug')
@patch('requests.get')
@patch('requests.post')
def test_successful_image_pass_through(self, mock_post, mock_get):
mock_post.return_value = self._auth_ok()
img = MagicMock()
img.status_code = 200
img.headers = {'Content-Type': 'image/jpeg'}
img.content = b'\xff\xd8\xffjpeg-bytes'
img.json = MagicMock(side_effect=ValueError('not json'))
mock_get.return_value = img
resp = self.client.get(self.url)
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], 'image/jpeg')
self.assertEqual(resp.content, b'\xff\xd8\xffjpeg-bytes')
self.assertEqual(resp['Cache-Control'], 'public, max-age=86400')
@patch('requests.get')
@patch('requests.post')
def test_second_poster_request_reuses_cached_token(self, mock_post, mock_get):
mock_post.return_value = self._auth_ok()
img = MagicMock()
img.status_code = 200
img.headers = {'Content-Type': 'image/jpeg'}
img.content = b'\xff\xd8\xffjpeg-bytes'
img.json = MagicMock(side_effect=ValueError('not json'))
mock_get.return_value = img
self.assertEqual(self.client.get(self.url).status_code, 200)
mock_post.reset_mock()
self.assertEqual(self.client.get(self.url).status_code, 200)
mock_post.assert_not_called()
self.assertEqual(mock_get.call_count, 2)

41
apps/epg/tests/test_serializers.py
View file

@ -484,6 +484,34 @@ class ProgramDataSerializerDescriptionFallbackTests(TestCase):
self.assertEqual(data["episode"], 1)
class SDPosterProxyPathTests(TestCase):
"""Cache-bust query on SD poster proxy URLs."""
def test_stable_uri_stable_bust(self):
from apps.epg.utils import sd_poster_cache_bust, sd_poster_proxy_path
uri = 'https://json.schedulesdirect.org/20141201/image/assets/a.jpg'
self.assertEqual(sd_poster_cache_bust(uri), sd_poster_cache_bust(uri))
path = sd_poster_proxy_path(42, uri)
self.assertTrue(path.startswith('/api/epg/programs/42/poster/?v='))
self.assertIn(sd_poster_cache_bust(uri), path)
def test_uri_change_changes_bust(self):
from apps.epg.utils import sd_poster_cache_bust
a = 'https://json.schedulesdirect.org/20141201/image/assets/a.jpg'
b = 'https://json.schedulesdirect.org/20141201/image/assets/b.jpg'
self.assertNotEqual(sd_poster_cache_bust(a), sd_poster_cache_bust(b))
def test_empty_uri_has_no_query(self):
from apps.epg.utils import sd_poster_proxy_path
self.assertEqual(
sd_poster_proxy_path(7, None),
'/api/epg/programs/7/poster/',
)
class ProgramDetailSerializerTests(TestCase):
"""Tests for ProgramDetailSerializer — rich field extraction from custom_properties."""
@ -670,3 +698,16 @@ class ProgramDetailSerializerTests(TestCase):
self.assertEqual(slim["is_live"], detail["is_live"])
self.assertEqual(slim["is_premiere"], detail["is_premiere"])
self.assertEqual(slim["is_finale"], detail["is_finale"])
def test_poster_url_includes_cache_bust_from_sd_icon(self):
from apps.epg.utils import sd_poster_proxy_path
uri = 'https://json.schedulesdirect.org/20141201/image/assets/p.jpg'
program = self._create_program(custom_properties={"sd_icon": uri})
data = ProgramDetailSerializer(program).data
self.assertEqual(data["poster_url"], sd_poster_proxy_path(program.id, uri))
def test_poster_url_none_without_sd_icon(self):
program = self._create_program(custom_properties={})
data = ProgramDetailSerializer(program).data
self.assertIsNone(data["poster_url"])

59
apps/epg/utils.py
View file

@ -1,12 +1,17 @@
"""
Shared EPG utilities — season/episode extraction.
Shared EPG utilities.
These live here (rather than in serializers.py or tasks.py) to avoid circular imports:
serializers → tasks and channels/tasks → serializers both need these functions.
Season/episode extraction, WebSocket progress updates, and SD poster proxy URL
helpers live here so serializers, XMLTV output, and tasks can import without
circular dependencies.
"""
import gc
import hashlib
import re
from core.utils import send_websocket_update
# Matches patterns like "S12 E6", "S3E21", "S8 E8 P2/2"
_ONSCREEN_RE = re.compile(r'S(\d+)\s*E(\d+)', re.IGNORECASE)
@ -21,6 +26,34 @@ _DESC_SE_PATTERNS = [
re.compile(r'^[\s\-:]*(\d+)x(\d{2,})[\s\-:.]*'),
]
_SD_POSTER_CACHE_BUST_LEN = 12
def sd_poster_cache_bust(sd_icon_url):
"""
Short content hash of an SD poster URI for nginx cache busting.
Same URI keeps the same ``v`` (long-lived nginx cache). A new artwork URI
after refresh gets a new ``v`` so clients do not keep stale bytes.
"""
if not sd_icon_url:
return ''
return hashlib.sha256(sd_icon_url.encode('utf-8')).hexdigest()[:_SD_POSTER_CACHE_BUST_LEN]
def sd_poster_proxy_path(program_id, sd_icon_url):
"""
Relative proxy path for a program poster.
Includes ``?v=`` when ``sd_icon_url`` is set so nginx cache keys change with
the upstream SD URI. The poster endpoint ignores ``v``; nginx keys on full URI.
"""
path = f'/api/epg/programs/{program_id}/poster/'
bust = sd_poster_cache_bust(sd_icon_url)
if not bust:
return path
return f'{path}?v={bust}'
def extract_season_episode_from_description(desc):
"""
@ -59,3 +92,23 @@ def extract_season_episode(cp, description=None):
if episode is None:
episode = d_episode
return season, episode
def send_epg_update(source_id, action, progress, **kwargs):
"""Send WebSocket update about EPG download/parsing progress."""
data = {
"progress": progress,
"type": "epg_refresh",
"source": source_id,
"action": action,
}
data.update(kwargs)
# High-frequency program parsing needs more aggressive memory management
collect_garbage = action == "parsing_programs" and progress % 10 == 0
send_websocket_update('updates', 'update', data, collect_garbage=collect_garbage)
data = None
if action == "parsing_programs" and progress % 50 == 0:
gc.collect()

11
apps/output/epg.py
View file

@ -20,6 +20,7 @@ from django.utils import timezone as django_timezone
from apps.channels.models import Channel, ChannelProfile, Stream
from apps.channels.utils import format_channel_number
from apps.epg.models import ProgramData
from apps.epg.utils import sd_poster_proxy_path
from apps.output.streaming_chunk_cache import stream_cached_response
from core.utils import build_absolute_uri_with_port, log_system_event
@ -1379,7 +1380,7 @@ def generate_epg(request, profile_name=None, user=None, *, xc_catchup_prev_days=
chunk_size = _EPG_PROGRAM_DB_CHUNK_SIZE
last_epg_id = 0
last_id = 0
_poster_url_base = build_absolute_uri_with_port(request, "/api/epg/programs/")
_poster_site_origin = build_absolute_uri_with_port(request, '')
def flush_pending():
nonlocal program_batch, pending
@ -1592,7 +1593,13 @@ def generate_epg(request, profile_name=None, user=None, *, xc_catchup_prev_days=
if "icon" in custom_data:
program_xml.append(f' <icon src="{html.escape(custom_data["icon"])}" />')
elif "sd_icon" in custom_data:
program_xml.append(f' <icon src="{html.escape(_poster_url_base)}{prog["id"]}/poster/" />')
poster_src = (
f'{_poster_site_origin}'
f'{sd_poster_proxy_path(prog["id"], custom_data["sd_icon"])}'
)
program_xml.append(
f' <icon src="{html.escape(poster_src)}" />'
)
# Add special flags as proper tags with enhanced handling
if custom_data.get("previously_shown", False):

9
core/tests/test_core.py
View file

@ -283,6 +283,15 @@ class DispatcharrUserAgentTests(TestCase):
{'User-Agent': 'Dispatcharr/1.2.3'},
)
@patch('version.__version__', '1.2.3')
def test_dispatcharr_http_headers_with_route_to(self):
from core.utils import dispatcharr_http_headers
headers = dispatcharr_http_headers(content_type=None, route_to='debug')
self.assertEqual(headers, {
'User-Agent': 'Dispatcharr/1.2.3',
'RouteTo': 'debug',
})
class ProgrammeIndexRebuildTests(TestCase):
def test_startup_rebuild_does_not_lock_out_queued_build_task(self):

6
core/utils.py
View file

@ -33,18 +33,22 @@ def dispatcharr_dvr_user_agent(recording_id):
return f'Dispatcharr-DVR/recording-{recording_id}'
def dispatcharr_http_headers(*, token=None, content_type='application/json'):
def dispatcharr_http_headers(*, token=None, content_type='application/json', route_to=None):
"""
Build HTTP headers for outbound Dispatcharr requests.
content_type=None omits Content-Type (e.g. simple GET proxies).
token is included when authenticating with Schedules Direct.
route_to is an optional Schedules Direct load-balancer steer value
(e.g. "debug"); only set when coordinated with Schedules Direct support.
"""
headers = {'User-Agent': dispatcharr_user_agent()}
if content_type:
headers['Content-Type'] = content_type
if token:
headers['token'] = token
if route_to:
headers['RouteTo'] = route_to
return headers

15
docker/init/03-init-dispatcharr.sh
View file

@ -1,5 +1,16 @@
#!/bin/bash
# TEMPORARY: one-time move of the nginx logo proxy cache to the new layout.
# Removable once installs have had time to upgrade past the /data/cache/ paths.
if [ -d /data/logo_cache ] && [ ! -e /data/cache/logos ]; then
mkdir -p /data/cache
if mv /data/logo_cache /data/cache/logos; then
echo "Moved nginx logo cache from /data/logo_cache to /data/cache/logos"
else
echo "⚠️ Warning: could not move /data/logo_cache to /data/cache/logos"
fi
fi
# Define directories that need to exist and be owned by PUID:PGID.
# DATA_DIRS may reside on external mounts (NFS, SMB/CIFS, FUSE) where
# mkdir and chown can fail. Failures are collected and reported as a
@ -7,7 +18,9 @@
DATA_DIRS=(
"/data/backups"
"/data/logos"
"/data/logo_cache"
"/data/cache"
"/data/cache/logos"
"/data/cache/sd_posters"
"/data/recordings"
"/data/uploads/m3us"
"/data/uploads/epgs"

11
docker/nginx.conf
View file

@ -1,6 +1,10 @@
proxy_cache_path /data/logo_cache levels=1:2 keys_zone=logo_cache:10m
# Channel/VOD logos and SD program posters share /data/cache/ with separate zones.
proxy_cache_path /data/cache/logos levels=1:2 keys_zone=logo_cache:10m
inactive=24h use_temp_path=off;
proxy_cache_path /data/cache/sd_posters levels=1:2 keys_zone=sd_poster_cache:10m
inactive=14d use_temp_path=off;
server {
listen NGINX_PORT;
listen [::]:NGINX_PORT;
@ -58,11 +62,12 @@ server {
proxy_cache_use_stale error timeout updating; # Serve stale if Django is slow
}
# SD program posters: 14d nginx cache. Clients pass ?v=<hash of sd_icon>
location ~ ^/api/epg/programs/(?<prog_id>\d+)/poster/ {
proxy_pass http://127.0.0.1:5656;
proxy_cache logo_cache;
proxy_cache sd_poster_cache;
proxy_cache_key "$scheme$request_uri";
proxy_cache_valid 200 24h;
proxy_cache_valid 200 14d;
proxy_cache_use_stale error timeout updating;
}

70
frontend/src/components/forms/EPG.jsx
View file

@ -170,6 +170,9 @@ const SDSettings = ({ sourceId, customProperties }) => {
const [posterStyle, setPosterStyle] = useState(
resolvedCp.poster_style || 'sd_recommended'
);
const [extraDebugging, setExtraDebugging] = useState(
!!resolvedCp.sd_extra_debugging
);
const [saving, setSaving] = useState(false);
// Sync from store (preferred) or parent props when the form opens or settings save
@ -178,6 +181,7 @@ const SDSettings = ({ sourceId, customProperties }) => {
setLogoStyle(newCp.logo_style || 'dark');
setFetchPosters(!!newCp.fetch_posters);
setPosterStyle(newCp.poster_style || 'sd_recommended');
setExtraDebugging(!!newCp.sd_extra_debugging);
}, [storeCustomProps, customProperties]);
const saveSetting = async (key, value) => {
@ -205,6 +209,11 @@ const SDSettings = ({ sourceId, customProperties }) => {
saveSetting('poster_style', style);
};
const handleExtraDebuggingToggle = (checked) => {
setExtraDebugging(checked);
saveSetting('sd_extra_debugging', checked);
};
return (
<Box>
<Text size="sm" fw={500} mb="xs">
@ -260,17 +269,11 @@ const SDSettings = ({ sourceId, customProperties }) => {
))}
</Group>
<AutoApplyEpgLogosSwitch
sourceId={sourceId}
customProperties={customProperties}
description="When enabled, matched channels are updated to use the SD logo on each refresh. When disabled, logos are still fetched into EPG data and can be applied manually via Set Logo from EPG."
/>
<Divider my="sm" />
<Switch
label="Fetch Program Posters"
description="WARNING: USES ADDITIONAL API REQUESTS. Poster artwork is fetched during EPG refresh; image bytes are cached by nginx on first view (24h). Initial fetch and viewing new programs consume API requests against your Schedules Direct rate limit."
description="Stores poster links on refresh. Images download on first view (nginx caches ~14 days) and count toward your daily SD image limit."
checked={fetchPosters}
onChange={(e) => handlePosterToggle(e.currentTarget.checked)}
disabled={saving}
@ -290,6 +293,17 @@ const SDSettings = ({ sourceId, customProperties }) => {
allowDeselect={false}
/>
)}
<Divider my="sm" />
<Switch
label="Extra Schedules Direct Debugging"
description="Only enable if Schedules Direct support asks you to. Turns off automatically if debug access is not allowed for your account."
checked={extraDebugging}
onChange={(e) => handleExtraDebuggingToggle(e.currentTarget.checked)}
disabled={saving}
size="sm"
/>
</Box>
);
};
@ -422,18 +436,27 @@ const SDLineupManager = ({ sourceId }) => {
setRemovingLineup(lineup.lineup);
try {
const result = await API.deleteSDLineup(sourceId, lineup.lineup);
if (result && result.code === 0) {
if (!result) return;
if (
result.changes_remaining !== undefined &&
result.changes_remaining !== null
) {
setChangesRemaining(result.changes_remaining);
}
if (result.error === 'daily_limit_reached') {
setChangesRemaining(0);
await fetchActiveLineups();
return;
}
if (result.code === 0) {
showNotification({
title: 'Lineup removed',
message: lineup.name,
color: 'blue',
});
if (
result.changes_remaining !== undefined &&
result.changes_remaining !== null
) {
setChangesRemaining(result.changes_remaining);
}
await fetchActiveLineups();
setSearchResults([]);
}
@ -495,6 +518,7 @@ const SDLineupManager = ({ sourceId }) => {
variant="subtle"
leftSection={<Trash2 size={12} />}
loading={removingLineup === lineup.lineup}
disabled={changesRemaining === 0}
onClick={() => handleRemove(lineup)}
>
Remove
@ -527,8 +551,9 @@ const SDLineupManager = ({ sourceId }) => {
mb="xs"
icon={<TriangleAlert size={14} />}
>
You have reached your daily Schedules Direct lineup addition limit. SD
allows 6 adds per 24-hour period.{' '}
You have reached your daily Schedules Direct lineup change limit. SD
allows 6 add or delete operations per 24-hour period. Adds and removes
are both blocked until the limit resets.{' '}
{changesResetAt && (
<span>
Limit resets at{' '}
@ -536,7 +561,7 @@ const SDLineupManager = ({ sourceId }) => {
</span>
)}
{!changesResetAt && (
<span>Limit resets 24 hours after the first add of the day. </span>
<span>Limit resets at midnight UTC. </span>
)}
<a href={SD_DOCS_URL} target="_blank" rel="noopener noreferrer">
Learn more
@ -551,8 +576,9 @@ const SDLineupManager = ({ sourceId }) => {
mb="xs"
icon={<TriangleAlert size={14} />}
>
You have <strong>1 lineup addition remaining</strong> today. Use it
carefully — Schedules Direct limits adds to 6 per 24-hour period.{' '}
You have <strong>1 lineup change remaining</strong> today. Use it
carefully. Schedules Direct limits adds and deletes to 6 per 24-hour
period.{' '}
<a href={SD_DOCS_URL} target="_blank" rel="noopener noreferrer">
Learn more
</a>
@ -566,8 +592,8 @@ const SDLineupManager = ({ sourceId }) => {
mb="xs"
icon={<TriangleAlert size={14} />}
>
You have <strong>2 lineup additions remaining</strong> today.
Schedules Direct limits adds to 6 per 24-hour period.{' '}
You have <strong>2 lineup changes remaining</strong> today. Schedules
Direct limits adds and deletes to 6 per 24-hour period.{' '}
<a href={SD_DOCS_URL} target="_blank" rel="noopener noreferrer">
Learn more
</a>
@ -935,7 +961,7 @@ const EPG = ({ epg = null, isOpen, onClose }) => {
key={form.key('priority')}
/>
{sourceType === 'xmltv' && savedEpgId && (
{savedEpgId && (
<AutoApplyEpgLogosSwitch
sourceId={savedEpgId}
customProperties={sdCustomProps}