mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-17 16:50:53 +00:00
Enhance Schedules Direct integration with per-channel guide fetch and improved EPG handling
- Implemented a targeted guide fetch for Schedules Direct when mapping a channel, ensuring immediate data availability without redundant API calls. - Updated EPG signals to queue guide fetches for newly mapped channels, streamlining the process and improving user experience. - Refactored existing logic to skip unnecessary program refreshes for dummy sources and ensure proper handling of Schedules Direct data. - Added unit tests to validate the new fetching behavior and ensure robustness of the integration.
This commit is contained in:
SergeantPanda
parent
bbff67899f
commit
c274bd7fbb
6 changed files with 466 additions and 210 deletions
|
|
@ -51,6 +51,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
|||
- **Schedules Direct lineup search country dropdown.** The country list was fetched directly from the SD API in the browser, which failed due to CORS and silently fell back to a 14-country hardcoded list. Countries are now included in the `GET sd-lineups` response (server-side fetch with proper User-Agent) so the full SD country list populates correctly. — Thanks [@sethwv](https://github.com/sethwv)
|
||||
- **Schedules Direct program poster proxy omitted User-Agent on image requests.** The poster endpoint authenticated with SD correctly but fetched images with only the `token` header, which violates SD's API requirements (error 1003). Image requests now include the standard `Dispatcharr/{version}` User-Agent via `dispatcharr_http_headers()`.
|
||||
- **Schedules Direct guide data missing after mapping a channel.** Lineup refreshes cached schedule MD5s for all stations, but `ProgramData` was only written for mapped channels. Mapping a channel later could leave MD5s looking unchanged so schedules were skipped and the channel had no guide until dates rolled outside the cached window. Refreshes now backfill fetch-window dates that lack `ProgramData` (including newly mapped stations with stale cache), fetch program metadata when no local `ProgramData` exists for a `programID`, and clean up orphaned `ProgramData` on unmapped `EPGData` entries. — Thanks [@sethwv](https://github.com/sethwv)
|
||||
- **Schedules Direct guide fetch on channel map.** Mapping a channel (or bulk-assigning EPG) now triggers a targeted guide fetch for that `EPGData` entry—mirroring XMLTV's `parse_programs_for_tvg_id` flow—without waiting for the next full source refresh. Skips the fetch when `ProgramData` already exists so additional channels sharing the same `tvg-id` do not trigger redundant API calls.
|
||||
|
||||
## [0.26.0] - 2026-06-07
|
||||
|
||||
|
|
|
|||
|
|
@ -2163,22 +2163,14 @@ class ChannelViewSet(viewsets.ModelViewSet):
|
|||
|
||||
epg_data = EPGData.objects.get(pk=epg_data_id)
|
||||
|
||||
# Set the EPG data and save
|
||||
# Set the EPG data and save. refresh_epg_programs (post_save) queues
|
||||
# parse_programs_for_tvg_id for non-dummy sources — no second dispatch here.
|
||||
channel.epg_data = epg_data
|
||||
channel.save(update_fields=["epg_data"])
|
||||
|
||||
# Only trigger program refresh for non-dummy EPG sources
|
||||
status_message = None
|
||||
if epg_data.epg_source.source_type != 'dummy':
|
||||
# Explicitly trigger program refresh for this EPG
|
||||
from apps.epg.tasks import parse_programs_for_tvg_id
|
||||
|
||||
task_result = parse_programs_for_tvg_id.delay(epg_data.id)
|
||||
|
||||
# Prepare response with task status info
|
||||
status_message = "EPG refresh queued"
|
||||
if task_result.result == "Task already running":
|
||||
status_message = "EPG refresh already in progress"
|
||||
|
||||
# Build response message
|
||||
message = f"EPG data set to {epg_data.tvg_id} for channel {channel.name}"
|
||||
|
|
@ -2374,7 +2366,7 @@ class ChannelViewSet(viewsets.ModelViewSet):
|
|||
continue
|
||||
|
||||
source_type = epg_data.epg_source.source_type if epg_data.epg_source else None
|
||||
if source_type not in ('dummy', 'schedules_direct'):
|
||||
if source_type != 'dummy':
|
||||
parse_programs_for_tvg_id.delay(epg_id)
|
||||
programs_refreshed += 1
|
||||
|
||||
|
|
|
|||
|
|
@ -327,7 +327,7 @@ def _dispatch_program_parse_for_epg_assignments(changed_associations):
|
|||
dispatched = 0
|
||||
for epg in EPGData.objects.filter(id__in=epg_ids).select_related("epg_source"):
|
||||
source_type = epg.epg_source.source_type if epg.epg_source else None
|
||||
if source_type in ("dummy", "schedules_direct"):
|
||||
if source_type == "dummy":
|
||||
continue
|
||||
parse_programs_for_tvg_id.delay(epg.id)
|
||||
dispatched += 1
|
||||
|
|
|
|||
|
|
@ -201,18 +201,13 @@ def refresh_epg_programs(sender, instance, created, **kwargs):
|
|||
if not created and kwargs.get('update_fields') and 'epg_data' in kwargs['update_fields']:
|
||||
logger.info(f"Channel {instance.id} ({instance.name}) EPG data updated, refreshing program data")
|
||||
if instance.epg_data:
|
||||
# Skip XMLTV program parser for SD sources — program data is handled
|
||||
# by fetch_schedules_direct() directly.
|
||||
if instance.epg_data.epg_source and instance.epg_data.epg_source.source_type == 'schedules_direct':
|
||||
logger.info(f"Skipping XMLTV parse for SD source {instance.epg_data.tvg_id}")
|
||||
if instance.epg_data.epg_source and instance.epg_data.epg_source.source_type == 'dummy':
|
||||
return
|
||||
logger.info(f"Triggering EPG program refresh for {instance.epg_data.tvg_id}")
|
||||
parse_programs_for_tvg_id.delay(instance.epg_data.id)
|
||||
# For new channels with EPG data, also refresh
|
||||
elif created and instance.epg_data:
|
||||
# Skip XMLTV program parser for SD sources
|
||||
if instance.epg_data.epg_source and instance.epg_data.epg_source.source_type == 'schedules_direct':
|
||||
logger.info(f"Skipping XMLTV parse for SD source {instance.epg_data.tvg_id}")
|
||||
if instance.epg_data.epg_source and instance.epg_data.epg_source.source_type == 'dummy':
|
||||
return
|
||||
logger.info(f"New channel {instance.id} ({instance.name}) created with EPG data, refreshing program data")
|
||||
parse_programs_for_tvg_id.delay(instance.epg_data.id)
|
||||
|
|
|
|||
|
|
@ -1496,14 +1496,11 @@ def parse_channels_only(source):
|
|||
|
||||
@shared_task(time_limit=3600, soft_time_limit=3500)
|
||||
def parse_programs_for_tvg_id(epg_id, force=False):
|
||||
# Skip XMLTV file parsing for Schedules Direct sources. Program data is
|
||||
# fetched and persisted directly by fetch_schedules_direct().
|
||||
try:
|
||||
from apps.epg.models import EPGData
|
||||
epg_obj = EPGData.objects.select_related('epg_source').filter(id=epg_id).first()
|
||||
if epg_obj and epg_obj.epg_source and epg_obj.epg_source.source_type == 'schedules_direct':
|
||||
logger.info(f"Skipping XMLTV parse for SD EPGData id={epg_id} (source: {epg_obj.epg_source.name})")
|
||||
return "Skipped (Schedules Direct source)"
|
||||
return fetch_sd_guide_for_epg(epg_id, force=force)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not check EPG source type for id={epg_id}: {e}")
|
||||
|
||||
|
|
@ -2139,6 +2136,73 @@ def parse_programs_for_source(epg_source, tvg_id=None):
|
|||
process = None
|
||||
|
||||
|
||||
def _sd_setup_single_epg_fetch(source, epg_id, token, sd_headers_fn):
|
||||
"""Build station_map / epg_id_map for a single mapped EPG entry."""
|
||||
epg = EPGData.objects.filter(id=epg_id, epg_source=source).first()
|
||||
if not epg or not epg.tvg_id:
|
||||
msg = f"Schedules Direct EPG entry {epg_id} not found or missing station ID."
|
||||
logger.error(msg)
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['last_message'])
|
||||
send_epg_update(source.id, "parsing_programs", 100, status="error", error=msg)
|
||||
return None
|
||||
|
||||
sd_lineup_country = None
|
||||
try:
|
||||
lineups_response = requests.get(
|
||||
f"{SD_BASE_URL}/lineups",
|
||||
headers=sd_headers_fn(token),
|
||||
timeout=30,
|
||||
)
|
||||
if lineups_response.ok:
|
||||
for lineup in lineups_response.json().get('lineups', []):
|
||||
lid = lineup.get('lineupID') or lineup.get('lineup') or ''
|
||||
if '-' in lid:
|
||||
sd_lineup_country = lid.split('-')[0]
|
||||
break
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.warning(f"Could not fetch lineups for country code: {e}")
|
||||
|
||||
send_epg_update(
|
||||
source.id, "parsing_programs", 15,
|
||||
message=f"Fetching guide data for {epg.name or epg.tvg_id}...",
|
||||
)
|
||||
station_map = {epg.tvg_id: {'name': epg.name or epg.tvg_id}}
|
||||
epg_id_map = {epg.tvg_id: epg.id}
|
||||
return station_map, epg_id_map, sd_lineup_country, epg
|
||||
|
||||
|
||||
@shared_task(time_limit=3600, soft_time_limit=3500)
|
||||
def fetch_sd_guide_for_epg(epg_id, force=False):
|
||||
"""
|
||||
Fetch Schedules Direct guide data for one mapped EPG entry (channel map flow).
|
||||
|
||||
Skips when ProgramData already exists so additional channels sharing the
|
||||
same EPGData / tvg_id do not trigger redundant API calls.
|
||||
"""
|
||||
epg = EPGData.objects.select_related('epg_source').filter(id=epg_id).first()
|
||||
if not epg or not epg.epg_source or epg.epg_source.source_type != 'schedules_direct':
|
||||
return "Not a Schedules Direct EPG entry"
|
||||
|
||||
if not force and ProgramData.objects.filter(epg_id=epg_id).exists():
|
||||
logger.info(f"SD guide fetch skipped for EPG {epg_id}: ProgramData already present")
|
||||
return "Guide data already present"
|
||||
|
||||
if not acquire_task_lock('parse_epg_programs', epg_id):
|
||||
logger.info(f"SD guide fetch for EPG {epg_id} already in progress, skipping duplicate task")
|
||||
return "Task already running"
|
||||
|
||||
lock_renewer = TaskLockRenewer('parse_epg_programs', epg_id)
|
||||
lock_renewer.start()
|
||||
try:
|
||||
logger.info(f"Fetching Schedules Direct guide for EPG {epg_id} ({epg.tvg_id})")
|
||||
fetch_schedules_direct(epg.epg_source, epg_id_only=epg_id, force=force)
|
||||
return "SD guide fetch complete"
|
||||
finally:
|
||||
lock_renewer.stop()
|
||||
release_task_lock('parse_epg_programs', epg_id)
|
||||
|
||||
|
||||
@shared_task(bind=True)
|
||||
def fetch_schedules_direct_stations(self, source_id):
|
||||
"""
|
||||
|
|
@ -2154,7 +2218,7 @@ def fetch_schedules_direct_stations(self, source_id):
|
|||
fetch_schedules_direct(source, stations_only=True)
|
||||
|
||||
|
||||
def fetch_schedules_direct(source, stations_only=False, force=False):
|
||||
def fetch_schedules_direct(source, stations_only=False, force=False, epg_id_only=None):
|
||||
"""
|
||||
Fetch EPG data from the Schedules Direct JSON API and persist it to the
|
||||
EPGData / ProgramData models.
|
||||
|
|
@ -2185,8 +2249,15 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
import hashlib
|
||||
from datetime import date
|
||||
|
||||
single_epg_fetch = epg_id_only is not None
|
||||
|
||||
logger.info(f"Fetching Schedules Direct data for source: {source.name}")
|
||||
if single_epg_fetch:
|
||||
logger.info(
|
||||
f"Fetching Schedules Direct guide for EPG {epg_id_only} "
|
||||
f"(source: {source.name})"
|
||||
)
|
||||
else:
|
||||
logger.info(f"Fetching Schedules Direct data for source: {source.name}")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Validate credentials
|
||||
|
|
@ -2213,7 +2284,7 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
# updated_at, which would otherwise incorrectly trigger this guard). Always
|
||||
# allow the first full refresh through so guide data is immediately available.
|
||||
# -------------------------------------------------------------------------
|
||||
if not stations_only and not force and source.updated_at:
|
||||
if not stations_only and not force and not single_epg_fetch and source.updated_at:
|
||||
from apps.epg.models import SDScheduleMD5 as _SDScheduleMD5
|
||||
has_prior_full_refresh = _SDScheduleMD5.objects.filter(epg_source=source).exists()
|
||||
if has_prior_full_refresh:
|
||||
|
|
@ -2234,7 +2305,7 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
return
|
||||
else:
|
||||
logger.info(f"SD source {source.id}: No prior full refresh detected, skipping 2-hour guard for first full fetch.")
|
||||
elif force and not stations_only:
|
||||
elif force and not stations_only and not single_epg_fetch:
|
||||
logger.info(f"SD source {source.id}: Force flag set, bypassing 2-hour refresh guard.")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
|
|
@ -2406,9 +2477,10 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
# This is a requirement of the Schedules Direct API specification, not an
|
||||
# architectural choice.
|
||||
# -------------------------------------------------------------------------
|
||||
source.status = EPGSource.STATUS_FETCHING
|
||||
source.last_message = "Authenticating with Schedules Direct..."
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
if not single_epg_fetch:
|
||||
source.status = EPGSource.STATUS_FETCHING
|
||||
source.last_message = "Authenticating with Schedules Direct..."
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "parsing_programs", 2, message="Authenticating with Schedules Direct...")
|
||||
|
||||
try:
|
||||
|
|
@ -2491,191 +2563,201 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
except requests.exceptions.RequestException as e:
|
||||
logger.warning(f"Could not fetch SD system status, proceeding anyway: {e}")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Step 3: Fetch subscribed lineups and build station map
|
||||
# -------------------------------------------------------------------------
|
||||
send_epg_update(source.id, "parsing_programs", 10, message="Fetching subscribed lineups...")
|
||||
try:
|
||||
lineups_response = requests.get(
|
||||
f"{SD_BASE_URL}/lineups",
|
||||
headers=_sd_headers(token),
|
||||
timeout=30,
|
||||
)
|
||||
# SD returns 400 with code 4102 when no lineups are configured.
|
||||
# This is a valid account state. The user needs to add lineups via
|
||||
# the Manage Lineups UI. Treat as idle rather than error.
|
||||
if lineups_response.status_code == 400:
|
||||
sd_data = lineups_response.json()
|
||||
if sd_data.get('code') == 4102:
|
||||
station_map = None
|
||||
epg_id_map = None
|
||||
sd_lineup_country = None
|
||||
|
||||
if epg_id_only is not None:
|
||||
setup = _sd_setup_single_epg_fetch(source, epg_id_only, token, _sd_headers)
|
||||
if setup is None:
|
||||
return
|
||||
station_map, epg_id_map, sd_lineup_country, _single_epg = setup
|
||||
else:
|
||||
# -------------------------------------------------------------------------
|
||||
# Step 3: Fetch subscribed lineups and build station map
|
||||
# -------------------------------------------------------------------------
|
||||
send_epg_update(source.id, "parsing_programs", 10, message="Fetching subscribed lineups...")
|
||||
try:
|
||||
lineups_response = requests.get(
|
||||
f"{SD_BASE_URL}/lineups",
|
||||
headers=_sd_headers(token),
|
||||
timeout=30,
|
||||
)
|
||||
# SD returns 400 with code 4102 when no lineups are configured.
|
||||
# This is a valid account state. The user needs to add lineups via
|
||||
# the Manage Lineups UI. Treat as idle rather than error.
|
||||
if lineups_response.status_code == 400:
|
||||
sd_data = lineups_response.json()
|
||||
if sd_data.get('code') == 4102:
|
||||
msg = "No lineups configured. Use the Manage Lineups option in the EPG source settings to add a lineup."
|
||||
logger.warning(f"SD source {source.id}: no lineups configured on account (4102).")
|
||||
source.status = EPGSource.STATUS_IDLE
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "refresh", 100, status="idle", message=msg)
|
||||
return
|
||||
lineups_response.raise_for_status()
|
||||
lineups_data = lineups_response.json()
|
||||
lineups = [l for l in lineups_data.get('lineups', []) if not l.get('isDeleted', False)]
|
||||
if not lineups:
|
||||
msg = "No lineups configured. Use the Manage Lineups option in the EPG source settings to add a lineup."
|
||||
logger.warning(f"SD source {source.id}: no lineups configured on account (4102).")
|
||||
logger.warning(f"SD source {source.id}: no active lineups found.")
|
||||
source.status = EPGSource.STATUS_IDLE
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "refresh", 100, status="idle", message=msg)
|
||||
return
|
||||
lineups_response.raise_for_status()
|
||||
lineups_data = lineups_response.json()
|
||||
lineups = [l for l in lineups_data.get('lineups', []) if not l.get('isDeleted', False)]
|
||||
if not lineups:
|
||||
msg = "No lineups configured. Use the Manage Lineups option in the EPG source settings to add a lineup."
|
||||
logger.warning(f"SD source {source.id}: no active lineups found.")
|
||||
source.status = EPGSource.STATUS_IDLE
|
||||
logger.info(f"Found {len(lineups)} lineup(s) in SD account.")
|
||||
|
||||
# Extract country from lineup IDs (format: "USA-NJ29486-X", "GBR-...", etc.)
|
||||
sd_lineup_country = None
|
||||
for l in lineups:
|
||||
lid = l.get('lineupID') or l.get('lineup') or ''
|
||||
if '-' in lid:
|
||||
sd_lineup_country = lid.split('-')[0]
|
||||
break
|
||||
logger.debug(f"SD lineup country: {sd_lineup_country}")
|
||||
except requests.exceptions.RequestException as e:
|
||||
msg = f"Failed to fetch Schedules Direct lineups: {e}"
|
||||
logger.error(msg, exc_info=True)
|
||||
source.status = EPGSource.STATUS_ERROR
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "refresh", 100, status="idle", message=msg)
|
||||
send_epg_update(source.id, "refresh", 100, status="error", error=msg)
|
||||
return
|
||||
logger.info(f"Found {len(lineups)} lineup(s) in SD account.")
|
||||
|
||||
# Extract country from lineup IDs (format: "USA-NJ29486-X", "GBR-...", etc.)
|
||||
sd_lineup_country = None
|
||||
for l in lineups:
|
||||
lid = l.get('lineupID') or l.get('lineup') or ''
|
||||
if '-' in lid:
|
||||
sd_lineup_country = lid.split('-')[0]
|
||||
break
|
||||
logger.debug(f"SD lineup country: {sd_lineup_country}")
|
||||
except requests.exceptions.RequestException as e:
|
||||
msg = f"Failed to fetch Schedules Direct lineups: {e}"
|
||||
logger.error(msg, exc_info=True)
|
||||
source.status = EPGSource.STATUS_ERROR
|
||||
source.last_message = msg
|
||||
|
||||
# Build station metadata map: stationID -> {name, callsign, logo_url}
|
||||
station_map = {}
|
||||
send_epg_update(source.id, "parsing_programs", 18, message=f"Fetching station metadata for {len(lineups)} lineup(s)...")
|
||||
for lineup in lineups:
|
||||
lineup_id = lineup.get('lineupID') or lineup.get('lineup')
|
||||
if not lineup_id:
|
||||
continue
|
||||
try:
|
||||
detail_response = requests.get(
|
||||
f"{SD_BASE_URL}/lineups/{lineup_id}",
|
||||
headers=_sd_headers(token),
|
||||
timeout=30,
|
||||
)
|
||||
detail_response.raise_for_status()
|
||||
detail_data = detail_response.json()
|
||||
for station in detail_data.get('stations', []):
|
||||
sid = station.get('stationID')
|
||||
if not sid:
|
||||
continue
|
||||
logo_url = None
|
||||
logos = station.get('stationLogo') or station.get('logo') or []
|
||||
if isinstance(logos, list) and logos:
|
||||
# Read preferred logo style from source settings; default to 'dark'
|
||||
logo_style = (source.custom_properties or {}).get('logo_style', 'dark')
|
||||
preferred = next((l for l in logos if l.get('category') == logo_style), logos[0])
|
||||
logo_url = preferred.get('URL') or preferred.get('url')
|
||||
elif isinstance(logos, dict):
|
||||
logo_url = logos.get('URL') or logos.get('url')
|
||||
station_map[sid] = {
|
||||
'name': station.get('name', sid),
|
||||
'callsign': station.get('callsign', ''),
|
||||
'logo_url': logo_url,
|
||||
}
|
||||
logger.debug(f"Fetched {len(detail_data.get('stations', []))} stations from lineup {lineup_id}")
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.warning(f"Failed to fetch lineup details for {lineup_id}: {e}")
|
||||
|
||||
if not station_map:
|
||||
msg = "No stations found across all Schedules Direct lineups."
|
||||
logger.warning(msg)
|
||||
source.status = EPGSource.STATUS_ERROR
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "refresh", 100, status="error", error=msg)
|
||||
return
|
||||
|
||||
logger.info(f"Built station map with {len(station_map)} stations.")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Step 4: Persist station metadata to EPGData
|
||||
# -------------------------------------------------------------------------
|
||||
source.status = EPGSource.STATUS_PARSING
|
||||
source.last_message = f"Syncing {len(station_map)} stations..."
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "refresh", 100, status="error", error=msg)
|
||||
return
|
||||
|
||||
# Build station metadata map: stationID -> {name, callsign, logo_url}
|
||||
station_map = {}
|
||||
send_epg_update(source.id, "parsing_programs", 18, message=f"Fetching station metadata for {len(lineups)} lineup(s)...")
|
||||
for lineup in lineups:
|
||||
lineup_id = lineup.get('lineupID') or lineup.get('lineup')
|
||||
if not lineup_id:
|
||||
continue
|
||||
try:
|
||||
detail_response = requests.get(
|
||||
f"{SD_BASE_URL}/lineups/{lineup_id}",
|
||||
headers=_sd_headers(token),
|
||||
timeout=30,
|
||||
send_epg_update(source.id, "parsing_programs", 28, message=f"Syncing {len(station_map)} stations to database...")
|
||||
|
||||
existing_epg_map = {
|
||||
epg.tvg_id: epg
|
||||
for epg in EPGData.objects.filter(epg_source=source)
|
||||
}
|
||||
|
||||
epgs_to_create = []
|
||||
epgs_to_update = []
|
||||
icon_max_length = EPGData._meta.get_field('icon_url').max_length
|
||||
name_max_length = EPGData._meta.get_field('name').max_length
|
||||
|
||||
for sid, info in station_map.items():
|
||||
display_name = (info['name'] or sid)[:name_max_length]
|
||||
logo = info['logo_url']
|
||||
if logo and len(logo) > icon_max_length:
|
||||
logo = None
|
||||
|
||||
if sid in existing_epg_map:
|
||||
epg_obj = existing_epg_map[sid]
|
||||
needs_update = False
|
||||
if epg_obj.name != display_name:
|
||||
epg_obj.name = display_name
|
||||
needs_update = True
|
||||
if epg_obj.icon_url != logo:
|
||||
epg_obj.icon_url = logo
|
||||
needs_update = True
|
||||
if needs_update:
|
||||
epgs_to_update.append(epg_obj)
|
||||
else:
|
||||
epgs_to_create.append(EPGData(
|
||||
tvg_id=sid,
|
||||
name=display_name,
|
||||
icon_url=logo,
|
||||
epg_source=source,
|
||||
))
|
||||
|
||||
if epgs_to_create:
|
||||
EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True)
|
||||
logger.info(f"Created {len(epgs_to_create)} new EPGData entries.")
|
||||
if epgs_to_update:
|
||||
EPGData.objects.bulk_update(epgs_to_update, ['name', 'icon_url'])
|
||||
logger.info(f"Updated {len(epgs_to_update)} existing EPGData entries.")
|
||||
|
||||
gc.collect()
|
||||
|
||||
# Rebuild map with fresh DB ids for all stations
|
||||
epg_id_map = {
|
||||
epg.tvg_id: epg.id
|
||||
for epg in EPGData.objects.filter(epg_source=source, tvg_id__in=list(station_map.keys()))
|
||||
}
|
||||
|
||||
# Station sync complete. Send progress update before continuing into programs phase.
|
||||
# We deliberately do NOT send parsing_channels at 100 with status=success here
|
||||
# because that would cause the frontend to mark the source as complete and
|
||||
# stop rendering progress updates for the subsequent program fetch phases.
|
||||
send_epg_update(source.id, "parsing_programs", 30,
|
||||
message=f"Stations synced ({len(station_map)} stations). Preparing schedule fetch...")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Stations-only mode. Used on initial source creation.
|
||||
# Stop here so the user can run Auto-match EPG before the full program fetch.
|
||||
# -------------------------------------------------------------------------
|
||||
if stations_only:
|
||||
success_msg = (
|
||||
f"{len(station_map)} stations loaded from Schedules Direct. "
|
||||
f"Run Auto-match EPG to map your channels, then use the Refresh "
|
||||
f"button to populate guide data."
|
||||
)
|
||||
detail_response.raise_for_status()
|
||||
detail_data = detail_response.json()
|
||||
for station in detail_data.get('stations', []):
|
||||
sid = station.get('stationID')
|
||||
if not sid:
|
||||
continue
|
||||
logo_url = None
|
||||
logos = station.get('stationLogo') or station.get('logo') or []
|
||||
if isinstance(logos, list) and logos:
|
||||
# Read preferred logo style from source settings; default to 'dark'
|
||||
logo_style = (source.custom_properties or {}).get('logo_style', 'dark')
|
||||
preferred = next((l for l in logos if l.get('category') == logo_style), logos[0])
|
||||
logo_url = preferred.get('URL') or preferred.get('url')
|
||||
elif isinstance(logos, dict):
|
||||
logo_url = logos.get('URL') or logos.get('url')
|
||||
station_map[sid] = {
|
||||
'name': station.get('name', sid),
|
||||
'callsign': station.get('callsign', ''),
|
||||
'logo_url': logo_url,
|
||||
}
|
||||
logger.debug(f"Fetched {len(detail_data.get('stations', []))} stations from lineup {lineup_id}")
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.warning(f"Failed to fetch lineup details for {lineup_id}: {e}")
|
||||
|
||||
if not station_map:
|
||||
msg = "No stations found across all Schedules Direct lineups."
|
||||
logger.warning(msg)
|
||||
source.status = EPGSource.STATUS_ERROR
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "refresh", 100, status="error", error=msg)
|
||||
return
|
||||
|
||||
logger.info(f"Built station map with {len(station_map)} stations.")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Step 4: Persist station metadata to EPGData
|
||||
# -------------------------------------------------------------------------
|
||||
source.status = EPGSource.STATUS_PARSING
|
||||
source.last_message = f"Syncing {len(station_map)} stations..."
|
||||
source.save(update_fields=['status', 'last_message'])
|
||||
send_epg_update(source.id, "parsing_programs", 28, message=f"Syncing {len(station_map)} stations to database...")
|
||||
|
||||
existing_epg_map = {
|
||||
epg.tvg_id: epg
|
||||
for epg in EPGData.objects.filter(epg_source=source)
|
||||
}
|
||||
|
||||
epgs_to_create = []
|
||||
epgs_to_update = []
|
||||
icon_max_length = EPGData._meta.get_field('icon_url').max_length
|
||||
name_max_length = EPGData._meta.get_field('name').max_length
|
||||
|
||||
for sid, info in station_map.items():
|
||||
display_name = (info['name'] or sid)[:name_max_length]
|
||||
logo = info['logo_url']
|
||||
if logo and len(logo) > icon_max_length:
|
||||
logo = None
|
||||
|
||||
if sid in existing_epg_map:
|
||||
epg_obj = existing_epg_map[sid]
|
||||
needs_update = False
|
||||
if epg_obj.name != display_name:
|
||||
epg_obj.name = display_name
|
||||
needs_update = True
|
||||
if epg_obj.icon_url != logo:
|
||||
epg_obj.icon_url = logo
|
||||
needs_update = True
|
||||
if needs_update:
|
||||
epgs_to_update.append(epg_obj)
|
||||
else:
|
||||
epgs_to_create.append(EPGData(
|
||||
tvg_id=sid,
|
||||
name=display_name,
|
||||
icon_url=logo,
|
||||
epg_source=source,
|
||||
))
|
||||
|
||||
if epgs_to_create:
|
||||
EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True)
|
||||
logger.info(f"Created {len(epgs_to_create)} new EPGData entries.")
|
||||
if epgs_to_update:
|
||||
EPGData.objects.bulk_update(epgs_to_update, ['name', 'icon_url'])
|
||||
logger.info(f"Updated {len(epgs_to_update)} existing EPGData entries.")
|
||||
|
||||
gc.collect()
|
||||
|
||||
# Rebuild map with fresh DB ids for all stations
|
||||
epg_id_map = {
|
||||
epg.tvg_id: epg.id
|
||||
for epg in EPGData.objects.filter(epg_source=source, tvg_id__in=list(station_map.keys()))
|
||||
}
|
||||
|
||||
# Station sync complete. Send progress update before continuing into programs phase.
|
||||
# We deliberately do NOT send parsing_channels at 100 with status=success here
|
||||
# because that would cause the frontend to mark the source as complete and
|
||||
# stop rendering progress updates for the subsequent program fetch phases.
|
||||
send_epg_update(source.id, "parsing_programs", 30,
|
||||
message=f"Stations synced ({len(station_map)} stations). Preparing schedule fetch...")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Stations-only mode. Used on initial source creation.
|
||||
# Stop here so the user can run Auto-match EPG before the full program fetch.
|
||||
# -------------------------------------------------------------------------
|
||||
if stations_only:
|
||||
success_msg = (
|
||||
f"{len(station_map)} stations loaded from Schedules Direct. "
|
||||
f"Run Auto-match EPG to map your channels, then use the Refresh "
|
||||
f"button to populate guide data."
|
||||
)
|
||||
source.status = EPGSource.STATUS_SUCCESS
|
||||
source.last_message = success_msg
|
||||
source.updated_at = timezone.now()
|
||||
source.save(update_fields=['status', 'last_message', 'updated_at'])
|
||||
send_epg_update(source.id, "parsing_channels", 100, status="success",
|
||||
message=success_msg, channels_count=len(station_map))
|
||||
logger.info(f"Stations-only fetch complete for source: {source.name} ({len(station_map)} stations)")
|
||||
return
|
||||
|
||||
source.status = EPGSource.STATUS_SUCCESS
|
||||
source.last_message = success_msg
|
||||
source.updated_at = timezone.now()
|
||||
source.save(update_fields=['status', 'last_message', 'updated_at'])
|
||||
send_epg_update(source.id, "parsing_channels", 100, status="success",
|
||||
message=success_msg, channels_count=len(station_map))
|
||||
logger.info(f"Stations-only fetch complete for source: {source.name} ({len(station_map)} stations)")
|
||||
return
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Step 5: MD5-delta schedule fetch
|
||||
# Only mapped channels need guide data. Fetch MD5 hashes and schedules for
|
||||
|
|
@ -2720,6 +2802,12 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
if not mapped_station_ids:
|
||||
logger.info("No channels mapped to this SD source; skipping schedule MD5 check and downloads.")
|
||||
_sd_post_refresh_tasks(mapped_epg_ids, {}, today)
|
||||
if single_epg_fetch:
|
||||
msg = "No mapped channel found for this EPG entry; guide fetch skipped."
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['last_message'])
|
||||
send_epg_update(source.id, "parsing_programs", 100, status="idle", message=msg)
|
||||
return
|
||||
success_msg = (
|
||||
f"{len(station_map)} lineup stations synced. "
|
||||
"Map channels to EPG entries, then refresh to populate guide data."
|
||||
|
|
@ -2828,6 +2916,12 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
if not changed_by_station:
|
||||
logger.info("No schedule changes detected, skipping schedule and program downloads.")
|
||||
_sd_post_refresh_tasks(mapped_epg_ids, {}, today)
|
||||
if single_epg_fetch:
|
||||
msg = "No schedule updates needed; guide data is up to date."
|
||||
source.last_message = msg
|
||||
source.save(update_fields=['last_message'])
|
||||
send_epg_update(source.id, "parsing_programs", 100, status="success", message=msg)
|
||||
return
|
||||
send_epg_update(source.id, "parsing_programs", 100, status="success",
|
||||
message="No schedule changes detected since last refresh. Guide data is up to date.")
|
||||
source.status = EPGSource.STATUS_SUCCESS
|
||||
|
|
@ -3383,6 +3477,25 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
|
|||
# -------------------------------------------------------------------------
|
||||
# Done
|
||||
# -------------------------------------------------------------------------
|
||||
if single_epg_fetch:
|
||||
epg_label = EPGData.objects.filter(id=epg_id_only).values_list('name', flat=True).first()
|
||||
success_msg = (
|
||||
f"Fetched {total_programs:,} programs for "
|
||||
f"{epg_label or epg_id_only} from Schedules Direct."
|
||||
)
|
||||
source.last_message = success_msg
|
||||
source.save(update_fields=['last_message'])
|
||||
send_epg_update(source.id, "parsing_programs", 100, status="success", message=success_msg)
|
||||
log_system_event(
|
||||
event_type='epg_refresh',
|
||||
source_name=source.name,
|
||||
programs=total_programs,
|
||||
channels=1,
|
||||
skipped_programs=skipped_unmapped,
|
||||
)
|
||||
logger.info(f"Schedules Direct single-EPG fetch complete for source: {source.name}")
|
||||
return
|
||||
|
||||
success_msg = (
|
||||
f"Successfully fetched {total_programs:,} programs for "
|
||||
f"{len(mapped_tvg_ids)} mapped stations from Schedules Direct "
|
||||
|
|
|
|||
|
|
@ -9,7 +9,8 @@ Covers:
|
|||
- fetch_schedules_direct: graceful error handling on auth failure
|
||||
- fetch_schedules_direct: schedule MD5 delta, backfill, and cache invalidation
|
||||
- parse_schedules_direct_time: correct UTC parsing
|
||||
- EPG signals: SD sources skip the XMLTV program parser
|
||||
- fetch_sd_guide_for_epg: per-channel guide fetch on map
|
||||
- EPG signals: SD sources queue guide fetch when a channel is mapped
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
|
|
@ -772,13 +773,167 @@ class ParseSchedulesDirectTimeTests(TestCase):
|
|||
# Signal tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class SDSingleEpgFetchTests(TestCase):
|
||||
"""Per-channel SD guide fetch on map (epg_id_only path)."""
|
||||
|
||||
MAPPED_STATION = '10001'
|
||||
|
||||
def _make_sd_source(self, updated_at=None):
|
||||
source = EPGSource.objects.create(
|
||||
name='SD Single EPG',
|
||||
source_type='schedules_direct',
|
||||
username='sduser',
|
||||
password='sdpass',
|
||||
)
|
||||
if updated_at is not None:
|
||||
EPGSource.objects.filter(id=source.id).update(updated_at=updated_at)
|
||||
source.refresh_from_db()
|
||||
return source
|
||||
|
||||
def _lineup_get_side_effect(self, url, **kwargs):
|
||||
if url.endswith('/status'):
|
||||
return MagicMock(
|
||||
status_code=200,
|
||||
json=MagicMock(return_value={'systemStatus': [{'status': 'Online'}]}),
|
||||
)
|
||||
if url.endswith('/lineups'):
|
||||
return MagicMock(
|
||||
status_code=200,
|
||||
json=MagicMock(return_value={'lineups': [{'lineupID': 'USA-TEST-X'}]}),
|
||||
)
|
||||
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')
|
||||
def test_fetch_sd_guide_skips_when_program_data_exists(
|
||||
self, mock_renewer, mock_release, mock_acquire,
|
||||
):
|
||||
from apps.epg.tasks import fetch_sd_guide_for_epg
|
||||
|
||||
source = self._make_sd_source()
|
||||
epg = EPGData.objects.create(
|
||||
tvg_id=self.MAPPED_STATION,
|
||||
name='Mapped',
|
||||
epg_source=source,
|
||||
)
|
||||
start = timezone.now()
|
||||
ProgramData.objects.create(
|
||||
epg=epg,
|
||||
start_time=start,
|
||||
end_time=start + timedelta(hours=1),
|
||||
title='Existing',
|
||||
tvg_id=epg.tvg_id,
|
||||
)
|
||||
|
||||
with patch('apps.epg.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')
|
||||
def test_parse_programs_for_tvg_id_delegates_to_sd_fetch(
|
||||
self, mock_renewer, mock_release, mock_acquire,
|
||||
):
|
||||
from apps.epg.tasks import parse_programs_for_tvg_id
|
||||
|
||||
source = self._make_sd_source()
|
||||
epg = EPGData.objects.create(
|
||||
tvg_id=self.MAPPED_STATION,
|
||||
name='Mapped',
|
||||
epg_source=source,
|
||||
)
|
||||
|
||||
with patch('apps.epg.tasks.fetch_sd_guide_for_epg', return_value='SD guide fetch complete') as mock_sd:
|
||||
result = parse_programs_for_tvg_id(epg.id)
|
||||
|
||||
mock_sd.assert_called_once_with(epg.id, force=False)
|
||||
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')
|
||||
def test_single_epg_fetch_skips_lineup_sync_and_updated_at(
|
||||
self, mock_post, mock_get, mock_send_epg_update,
|
||||
):
|
||||
from apps.epg.tasks import fetch_schedules_direct
|
||||
|
||||
prior_updated = timezone.now() - timedelta(hours=1)
|
||||
source = self._make_sd_source(updated_at=prior_updated)
|
||||
epg = EPGData.objects.create(
|
||||
tvg_id=self.MAPPED_STATION,
|
||||
name='Mapped',
|
||||
epg_source=source,
|
||||
)
|
||||
Channel.objects.create(name='Mapped Ch', epg_data=epg)
|
||||
|
||||
date_list = [
|
||||
(date.today() + timedelta(days=i)).strftime('%Y-%m-%d')
|
||||
for i in range(3)
|
||||
]
|
||||
mock_get.side_effect = self._lineup_get_side_effect
|
||||
|
||||
schedule_payload = [{
|
||||
'stationID': self.MAPPED_STATION,
|
||||
'metadata': {'startDate': date_list[0], 'md5': 'md5-new', 'modified': '2026-06-11T00:00:00Z'},
|
||||
'programs': [{
|
||||
'programID': 'EP000000000001',
|
||||
'airDateTime': f'{date_list[0]}T12:00:00Z',
|
||||
'duration': 3600,
|
||||
'md5': 'prog-md5-1',
|
||||
}],
|
||||
}]
|
||||
program_payload = [{
|
||||
'programID': 'EP000000000001',
|
||||
'titles': [{'title120': 'Test Show'}],
|
||||
}]
|
||||
|
||||
def post_side_effect(url, **kwargs):
|
||||
if url.endswith('/token'):
|
||||
return MagicMock(
|
||||
status_code=200,
|
||||
json=MagicMock(return_value={'code': 0, 'token': 'tok'}),
|
||||
)
|
||||
if url.endswith('/schedules/md5'):
|
||||
return MagicMock(
|
||||
status_code=200,
|
||||
json=MagicMock(return_value={
|
||||
self.MAPPED_STATION: {
|
||||
ds: {'code': 0, 'md5': f'md5-{ds}', 'lastModified': '2026-06-11T00:00:00Z'}
|
||||
for ds in date_list
|
||||
},
|
||||
}),
|
||||
)
|
||||
if url.endswith('/schedules'):
|
||||
return MagicMock(status_code=200, json=MagicMock(return_value=schedule_payload))
|
||||
if url.endswith('/programs'):
|
||||
return MagicMock(status_code=200, json=MagicMock(return_value=program_payload))
|
||||
raise AssertionError(f'Unexpected POST URL: {url}')
|
||||
|
||||
mock_post.side_effect = post_side_effect
|
||||
|
||||
fetch_schedules_direct(source, epg_id_only=epg.id)
|
||||
|
||||
lineup_detail_calls = [
|
||||
c for c in mock_get.call_args_list
|
||||
if '/lineups/' in c[0][0] and not c[0][0].endswith('/lineups')
|
||||
]
|
||||
self.assertEqual(lineup_detail_calls, [])
|
||||
|
||||
source.refresh_from_db()
|
||||
self.assertEqual(source.updated_at, prior_updated)
|
||||
self.assertEqual(ProgramData.objects.filter(epg=epg).count(), 1)
|
||||
|
||||
|
||||
class SDSourceSignalTests(TestCase):
|
||||
"""SD EPG sources must skip the XMLTV program parser signal."""
|
||||
"""SD EPG sources queue per-EPG guide fetch when a channel is mapped."""
|
||||
|
||||
@patch('apps.channels.signals.parse_programs_for_tvg_id')
|
||||
def test_sd_source_skips_xmltv_parse_on_channel_create(self, mock_parse):
|
||||
"""Creating a channel linked to an SD EPG source must not trigger
|
||||
the XMLTV program parser — SD data is handled by fetch_schedules_direct."""
|
||||
def test_sd_source_queues_guide_fetch_on_channel_create(self, mock_parse):
|
||||
from apps.epg.models import EPGData
|
||||
from apps.channels.models import Channel
|
||||
|
||||
|
|
@ -799,7 +954,7 @@ class SDSourceSignalTests(TestCase):
|
|||
epg_data=epg_data,
|
||||
)
|
||||
|
||||
mock_parse.delay.assert_not_called()
|
||||
mock_parse.delay.assert_called_once_with(epg_data.id)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue