From bfa70193241cbfdafb916b46b7cbab22093fd3c7 Mon Sep 17 00:00:00 2001 From: Seth Van Niekerk Date: Wed, 10 Jun 2026 10:01:47 -0400 Subject: [PATCH 1/3] force-fetch SD schedules for newly mapped stations with no existing ProgramData --- apps/epg/tasks.py | 53 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index d868dfeb..6da11218 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -2665,6 +2665,49 @@ def fetch_schedules_direct(source, stations_only=False, force=False): send_epg_update(source.id, "parsing_programs", 38, message=f"MD5 check complete: {len(changed_by_station)} stations have schedule updates.") + # Force-fetch for mapped stations with no existing ProgramData. + # The MD5 cache stores hashes for all lineup stations, including unmapped ones. + # If a station was in the lineup (MD5 cached) but not yet mapped to a channel + # (no ProgramData created), a later mapping causes those cached dates to appear + # "unchanged", skipping the fetch and leaving the channel with no guide data. + mapped_epg_ids_early = set( + Channel.objects.filter( + epg_data__epg_source=source, epg_data__isnull=False + ).values_list('epg_data_id', flat=True) + ) + mapped_tvg_ids_early = set( + EPGData.objects.filter( + id__in=mapped_epg_ids_early, epg_source=source + ).values_list('tvg_id', flat=True) + ) + newly_mapped = set() + if mapped_tvg_ids_early: + stations_with_data = set( + ProgramData.objects.filter( + epg__epg_source=source, + tvg_id__in=mapped_tvg_ids_early, + ).values_list('tvg_id', flat=True).distinct() + ) + newly_mapped = mapped_tvg_ids_early - stations_with_data + if newly_mapped: + server_dates_by_sid = {} + for (sid, ds) in server_md5s: + if sid in newly_mapped: + server_dates_by_sid.setdefault(sid, set()).add(ds) + forced_count = 0 + for sid in newly_mapped: + available = server_dates_by_sid.get(sid, set()) + already_changing = set(changed_by_station.get(sid, [])) + force_dates = [ds for ds in date_list if ds in available and ds not in already_changing] + if force_dates: + changed_by_station.setdefault(sid, []).extend(force_dates) + forced_count += len(force_dates) + if forced_count: + logger.info( + f"Force-fetching {forced_count} station/date combinations for " + f"{len(newly_mapped)} stations with no existing ProgramData (newly mapped channels)." + ) + # schedules_by_station: stationID -> list of {programID, airDateTime, duration, ...} schedules_by_station = {sid: [] for sid in station_ids} program_ids_needed = set() @@ -2814,6 +2857,16 @@ def fetch_schedules_direct(source, stations_only=False, force=False): ).only('program_id', 'md5') } + # For newly mapped stations with no ProgramData, invalidate cached program MD5s so + # metadata gets re-fetched. Without this, programs whose MD5s were cached when the + # station was unmapped would be skipped, producing "No Title" records. + if newly_mapped: + for sid in newly_mapped: + for airing in schedules_by_station.get(sid, []): + pid = airing.get('programID') + if pid: + cached_prog_md5s.pop(pid, None) + # Only fetch programs where MD5 differs from our cached value programs_to_fetch = { pid for pid in program_ids_needed From 205deb150beb63c33849863002fe48937ca6cd1d Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 11 Jun 2026 12:31:30 -0500 Subject: [PATCH 2/3] Implement schedule MD5 delta detection and backfill logic for improved cache management - Added functions to compute schedule changes based on MD5 hashes and backfill missing dates for mapped stations. - Enhanced handling of orphaned ProgramData records for unmapped EPG entries during schedule fetch. - Introduced unit tests to validate MD5 comparison, backfill logic, and metadata fetching requirements. --- apps/epg/tasks.py | 309 +++++++++------ apps/epg/tests/test_schedules_direct.py | 474 +++++++++++++++++++++++- 2 files changed, 675 insertions(+), 108 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 6da11218..9d244e79 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -35,6 +35,73 @@ SD_BASE_URL = 'https://json.schedulesdirect.org/20141201' SD_DAYS_TO_FETCH = 20 SD_PROGRAM_BATCH_SIZE = 5000 + +def _sd_compute_schedule_changes_from_md5(server_md5s, cached_md5s, date_list): + """Return station_id -> [date_str] for dates whose schedule MD5 differs from cache.""" + changed_by_station = {} + for (sid, date_str), server_info in server_md5s.items(): + if date_str not in date_list: + continue + cached = cached_md5s.get((sid, date_str)) + if cached != server_info['md5']: + changed_by_station.setdefault(sid, []).append(date_str) + return changed_by_station + + +def _sd_backfill_schedule_dates_without_data( + changed_by_station, + server_md5s, + date_list, + mapped_station_ids, + epg_id_map, + dates_with_data, + cached_md5s, + stations_without_any_data, +): + """ + Add fetch-window dates that lack ProgramData to changed_by_station. + + Dates with a cached schedule MD5 are treated as already fetched (e.g. legitimately + empty airings). Stations with zero ProgramData still backfill all missing dates + so stale cache from unmapped lineup refreshes cannot block guide population. + """ + from datetime import date as date_type + + stations_without_any_data = set(stations_without_any_data) + backfilled_count = 0 + for sid in mapped_station_ids: + epg_db_id = epg_id_map.get(sid) + if not epg_db_id: + continue + force_despite_cache = sid in stations_without_any_data + already_changing = set(changed_by_station.get(sid, [])) + for ds in date_list: + if ds in already_changing or (sid, ds) not in server_md5s: + continue + if (epg_db_id, date_type.fromisoformat(ds)) in dates_with_data: + continue + if (sid, ds) in cached_md5s and not force_despite_cache: + continue + changed_by_station.setdefault(sid, []).append(ds) + backfilled_count += 1 + return backfilled_count + + +def _sd_programs_needing_metadata( + program_ids_needed, + schedule_program_md5s, + cached_prog_md5s, + programs_with_data, +): + """Return programIDs that need metadata download from Schedules Direct.""" + programs_with_data = set(programs_with_data) + return { + pid for pid in program_ids_needed + if schedule_program_md5s.get(pid) != cached_prog_md5s.get(pid) + or pid not in programs_with_data + } + + SD_POSTER_CATEGORIES = ( 'Iconic', 'Banner-L1', 'Banner-L2', 'Banner-L3', 'Banner', 'Staple', 'Poster Art', 'Box Art', @@ -2070,6 +2137,8 @@ def parse_programs_for_source(epg_source, tvg_id=None): logger.info(f"[parse_programs_for_source] Final memory usage: {final_memory:.2f} MB difference: {final_memory - initial_memory:.2f} MB") # Explicitly clear the process object to prevent potential memory leaks process = None + + @shared_task(bind=True) def fetch_schedules_direct_stations(self, source_id): """ @@ -2299,6 +2368,24 @@ def fetch_schedules_direct(source, stations_only=False, force=False): from apps.channels.utils import maybe_auto_apply_epg_logos maybe_auto_apply_epg_logos(source) + try: + unmapped_epg_ids = list( + EPGData.objects.filter(epg_source=source).exclude( + id__in=mapped_epg_ids, + ).values_list('id', flat=True) + ) + if unmapped_epg_ids: + orphaned_count = ProgramData.objects.filter( + epg_id__in=unmapped_epg_ids, + ).delete()[0] + if orphaned_count: + logger.info( + f"Cleaned up {orphaned_count} orphaned ProgramData records " + f"for {len(unmapped_epg_ids)} unmapped EPG entries." + ) + except Exception as prune_err: + logger.warning(f"Failed to clean up orphaned SD ProgramData: {prune_err}") + today_utc = datetime(today.year, today.month, today.day, tzinfo=dt_timezone.utc) try: expired_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids, end_time__lt=today_utc).delete()[0] @@ -2598,32 +2685,77 @@ def fetch_schedules_direct(source, stations_only=False, force=False): # ------------------------------------------------------------------------- # Step 5: MD5-delta schedule fetch - # First fetch MD5 hashes for all stations/dates. Compare against our - # locally cached hashes to determine which schedules have changed. - # Only download schedules that have actually changed; this minimises - # API calls against SD's rate-limited endpoints. + # Only mapped channels need guide data. Fetch MD5 hashes and schedules for + # mapped stations only; never cache schedule MD5s for unmapped lineup entries. # ------------------------------------------------------------------------- - from apps.epg.models import SDScheduleMD5 from django.utils.dateparse import parse_datetime - send_epg_update(source.id, "parsing_programs", 33, message=f"Checking schedule MD5s for {len(station_map)} stations over {SD_DAYS_TO_FETCH} days...") station_ids = list(station_map.keys()) today = date.today() date_list = [(today + timedelta(days=i)).strftime('%Y-%m-%d') for i in range(SD_DAYS_TO_FETCH)] - # Prune SDScheduleMD5 records whose dates have rolled off the fetch window. - # These accumulate one row per station per day and are never useful once past. - pruned_sched_md5_count = SDScheduleMD5.objects.filter(epg_source=source, date__lt=today).delete()[0] + mapped_epg_ids = set( + Channel.objects.filter( + epg_data__epg_source=source, + epg_data__isnull=False, + ).values_list('epg_data_id', flat=True) + ) + mapped_tvg_ids = set( + EPGData.objects.filter( + id__in=mapped_epg_ids, + epg_source=source, + ).values_list('tvg_id', flat=True) + ) + mapped_station_ids = [sid for sid in station_ids if sid in mapped_tvg_ids] + + # Prune expired schedule MD5s and drop cache for unmapped stations. + pruned_sched_md5_count = SDScheduleMD5.objects.filter( + epg_source=source, date__lt=today, + ).delete()[0] if pruned_sched_md5_count: logger.info(f"Pruned {pruned_sched_md5_count} expired SDScheduleMD5 records (before {today}).") - # Fetch MD5 hashes for all stations in batches of 5000 + if mapped_tvg_ids: + unmapped_cache_pruned = SDScheduleMD5.objects.filter( + epg_source=source, + ).exclude(station_id__in=mapped_tvg_ids).delete()[0] + else: + unmapped_cache_pruned = SDScheduleMD5.objects.filter(epg_source=source).delete()[0] + if unmapped_cache_pruned: + logger.info(f"Pruned {unmapped_cache_pruned} SDScheduleMD5 records for unmapped lineup stations.") + + 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) + success_msg = ( + f"{len(station_map)} lineup stations synced. " + "Map channels to EPG entries, then refresh 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_programs", 100, status="success", message=success_msg) + return + + send_epg_update( + source.id, "parsing_programs", 33, + message=f"Checking schedule MD5s for {len(mapped_station_ids)} mapped stations over {SD_DAYS_TO_FETCH} days...", + ) + + # Fetch MD5 hashes for mapped stations in batches of 5000 STATION_BATCH_SIZE = 5000 server_md5s = {} # (station_id, date) -> {md5, last_modified} - logger.info(f"Fetching schedule MD5s for {len(station_ids)} stations over {SD_DAYS_TO_FETCH} days.") + logger.info( + f"Fetching schedule MD5s for {len(mapped_station_ids)} mapped stations " + f"(of {len(station_ids)} lineup stations) over {SD_DAYS_TO_FETCH} days." + ) - station_batches = [station_ids[i:i + STATION_BATCH_SIZE] for i in range(0, len(station_ids), STATION_BATCH_SIZE)] + station_batches = [ + mapped_station_ids[i:i + STATION_BATCH_SIZE] + for i in range(0, len(mapped_station_ids), STATION_BATCH_SIZE) + ] for batch in station_batches: try: md5_response = requests.post( @@ -2644,84 +2776,65 @@ def fetch_schedules_direct(source, stations_only=False, force=False): except requests.exceptions.RequestException as e: logger.warning(f"Failed to fetch schedule MD5s: {e}") - # Load our cached MD5s from DB + # Load our cached MD5s from DB (mapped stations only) cached_md5s = { (r.station_id, r.date.strftime('%Y-%m-%d')): r.md5 - for r in SDScheduleMD5.objects.filter(epg_source=source, station_id__in=station_ids) + for r in SDScheduleMD5.objects.filter( + epg_source=source, station_id__in=mapped_station_ids, + ) } - # Determine which station/date combinations need downloading - changed_by_station = {} # station_id -> [date_str, ...] - for (sid, date_str), server_info in server_md5s.items(): - if date_str not in date_list: - continue - cached = cached_md5s.get((sid, date_str)) - if cached != server_info['md5']: - changed_by_station.setdefault(sid, []).append(date_str) + changed_by_station = _sd_compute_schedule_changes_from_md5( + server_md5s, cached_md5s, date_list, + ) + + window_start = datetime(today.year, today.month, today.day, tzinfo=dt_timezone.utc) + window_end = window_start + timedelta(days=SD_DAYS_TO_FETCH) + dates_with_data = set() + if mapped_epg_ids: + for epg_id, start_time in ProgramData.objects.filter( + epg_id__in=mapped_epg_ids, + start_time__gte=window_start, + start_time__lt=window_end, + ).values_list('epg_id', 'start_time'): + dates_with_data.add((epg_id, start_time.date())) + + stations_without_any_data = mapped_tvg_ids - set( + ProgramData.objects.filter(epg_id__in=mapped_epg_ids) + .values_list('tvg_id', flat=True).distinct() + ) + backfilled_count = _sd_backfill_schedule_dates_without_data( + changed_by_station, + server_md5s, + date_list, + mapped_station_ids, + epg_id_map, + dates_with_data, + cached_md5s, + stations_without_any_data, + ) + if backfilled_count: + logger.info( + f"Backfilling {backfilled_count} station/date combinations with no ProgramData " + f"in the {SD_DAYS_TO_FETCH}-day fetch window." + ) total_changed = sum(len(v) for v in changed_by_station.values()) - total_possible = len(station_ids) * len(date_list) - logger.info(f"Schedule MD5 check: {len(server_md5s)} hashes checked, {total_changed} station/date combinations changed (of {total_possible} possible).") + total_possible = len(mapped_station_ids) * len(date_list) + logger.info( + f"Schedule MD5 check: {len(server_md5s)} hashes checked, " + f"{total_changed} station/date combinations to fetch (of {total_possible} possible)." + ) send_epg_update(source.id, "parsing_programs", 38, message=f"MD5 check complete: {len(changed_by_station)} stations have schedule updates.") - # Force-fetch for mapped stations with no existing ProgramData. - # The MD5 cache stores hashes for all lineup stations, including unmapped ones. - # If a station was in the lineup (MD5 cached) but not yet mapped to a channel - # (no ProgramData created), a later mapping causes those cached dates to appear - # "unchanged", skipping the fetch and leaving the channel with no guide data. - mapped_epg_ids_early = set( - Channel.objects.filter( - epg_data__epg_source=source, epg_data__isnull=False - ).values_list('epg_data_id', flat=True) - ) - mapped_tvg_ids_early = set( - EPGData.objects.filter( - id__in=mapped_epg_ids_early, epg_source=source - ).values_list('tvg_id', flat=True) - ) - newly_mapped = set() - if mapped_tvg_ids_early: - stations_with_data = set( - ProgramData.objects.filter( - epg__epg_source=source, - tvg_id__in=mapped_tvg_ids_early, - ).values_list('tvg_id', flat=True).distinct() - ) - newly_mapped = mapped_tvg_ids_early - stations_with_data - if newly_mapped: - server_dates_by_sid = {} - for (sid, ds) in server_md5s: - if sid in newly_mapped: - server_dates_by_sid.setdefault(sid, set()).add(ds) - forced_count = 0 - for sid in newly_mapped: - available = server_dates_by_sid.get(sid, set()) - already_changing = set(changed_by_station.get(sid, [])) - force_dates = [ds for ds in date_list if ds in available and ds not in already_changing] - if force_dates: - changed_by_station.setdefault(sid, []).extend(force_dates) - forced_count += len(force_dates) - if forced_count: - logger.info( - f"Force-fetching {forced_count} station/date combinations for " - f"{len(newly_mapped)} stations with no existing ProgramData (newly mapped channels)." - ) - # schedules_by_station: stationID -> list of {programID, airDateTime, duration, ...} - schedules_by_station = {sid: [] for sid in station_ids} + schedules_by_station = {sid: [] for sid in mapped_station_ids} program_ids_needed = set() if not changed_by_station: logger.info("No schedule changes detected, skipping schedule and program downloads.") - from apps.channels.models import Channel as ChannelModel - mapped_epg_ids_no_change = set( - ChannelModel.objects.filter( - epg_data__epg_source=source, - epg_data__isnull=False, - ).values_list('epg_data_id', flat=True) - ) - _sd_post_refresh_tasks(mapped_epg_ids_no_change, {}, today) + _sd_post_refresh_tasks(mapped_epg_ids, {}, today) 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 @@ -2857,21 +2970,21 @@ def fetch_schedules_direct(source, stations_only=False, force=False): ).only('program_id', 'md5') } - # For newly mapped stations with no ProgramData, invalidate cached program MD5s so - # metadata gets re-fetched. Without this, programs whose MD5s were cached when the - # station was unmapped would be skipped, producing "No Title" records. - if newly_mapped: - for sid in newly_mapped: - for airing in schedules_by_station.get(sid, []): - pid = airing.get('programID') - if pid: - cached_prog_md5s.pop(pid, None) + programs_with_data = set() + if program_ids_needed: + programs_with_data = set( + ProgramData.objects.filter( + epg__epg_source=source, + program_id__in=program_ids_needed, + ).values_list('program_id', flat=True).distinct() + ) - # Only fetch programs where MD5 differs from our cached value - programs_to_fetch = { - pid for pid in program_ids_needed - if schedule_program_md5s.get(pid) != cached_prog_md5s.get(pid) - } + programs_to_fetch = _sd_programs_needing_metadata( + program_ids_needed, + schedule_program_md5s, + cached_prog_md5s, + programs_with_data, + ) logger.info( f"Program MD5 delta: {len(program_ids_needed)} programs in schedules, " @@ -2928,22 +3041,6 @@ def fetch_schedules_direct(source, stations_only=False, force=False): logger.info("Building program records...") send_epg_update(source.id, "parsing_programs", 80) - # Only process stations that are mapped to channels to match the existing - # XMLTV flow (parse_programs_for_source only processes mapped channels). - from apps.channels.models import Channel as ChannelModel - mapped_epg_ids = set( - ChannelModel.objects.filter( - epg_data__epg_source=source, - epg_data__isnull=False, - ).values_list('epg_data_id', flat=True) - ) - mapped_tvg_ids = set( - EPGData.objects.filter( - id__in=mapped_epg_ids, - epg_source=source, - ).values_list('tvg_id', flat=True) - ) - # Cache existing program data for unchanged programs BEFORE surgical delete. # When a station/date schedule MD5 changes, ALL airings are re-fetched, but only # programs with changed program MD5s get metadata re-downloaded. The surgical delete diff --git a/apps/epg/tests/test_schedules_direct.py b/apps/epg/tests/test_schedules_direct.py index 77334112..47860f92 100644 --- a/apps/epg/tests/test_schedules_direct.py +++ b/apps/epg/tests/test_schedules_direct.py @@ -7,19 +7,26 @@ Covers: - fetch_schedules_direct: credential validation - fetch_schedules_direct: SHA1 password hashing and token exchange - 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 """ import hashlib -from datetime import datetime +from datetime import date, datetime, timedelta, timezone as dt_timezone from unittest.mock import MagicMock, patch from django.test import TestCase from django.utils import timezone -from apps.epg.models import EPGSource, EPGData +from apps.channels.models import Channel +from apps.epg.models import EPGSource, EPGData, ProgramData, SDScheduleMD5 from apps.epg.serializers import EPGSourceSerializer +from apps.epg.tasks import ( + _sd_backfill_schedule_dates_without_data, + _sd_compute_schedule_changes_from_md5, + _sd_programs_needing_metadata, +) # --------------------------------------------------------------------------- @@ -270,6 +277,469 @@ class FetchSchedulesDirectStationsOnlyTests(TestCase): self.assertEqual(complete_call[1]['channels_count'], 1) +# --------------------------------------------------------------------------- +# Schedule MD5 delta, backfill, and cache tests +# --------------------------------------------------------------------------- + +class SDScheduleMd5DeltaTests(TestCase): + """Pure-function tests for schedule MD5 comparison.""" + + DATE_LIST = ['2026-06-11', '2026-06-12', '2026-06-13'] + + def test_detects_changed_md5(self): + server_md5s = { + ('10001', '2026-06-11'): {'md5': 'abc', 'last_modified': ''}, + ('10001', '2026-06-12'): {'md5': 'def', 'last_modified': ''}, + } + cached_md5s = { + ('10001', '2026-06-11'): 'abc', + ('10001', '2026-06-12'): 'old', + } + changed = _sd_compute_schedule_changes_from_md5( + server_md5s, cached_md5s, self.DATE_LIST, + ) + self.assertEqual(changed['10001'], ['2026-06-12']) + + def test_missing_cache_treated_as_changed(self): + server_md5s = { + ('10001', '2026-06-11'): {'md5': 'abc', 'last_modified': ''}, + } + changed = _sd_compute_schedule_changes_from_md5(server_md5s, {}, self.DATE_LIST) + self.assertEqual(changed['10001'], ['2026-06-11']) + + def test_ignores_dates_outside_fetch_window(self): + server_md5s = { + ('10001', '2026-06-01'): {'md5': 'abc', 'last_modified': ''}, + } + changed = _sd_compute_schedule_changes_from_md5(server_md5s, {}, self.DATE_LIST) + self.assertEqual(changed, {}) + + +class SDScheduleBackfillTests(TestCase): + """Backfill must fix stale-cache gaps without re-fetching empty cached days.""" + + DATE_LIST = ['2026-06-11', '2026-06-12', '2026-06-13'] + EPG_ID = 42 + EPG_ID_MAP = {'10001': 42} + + def _server_md5s(self): + return { + (sid, ds): {'md5': 'hash', 'last_modified': ''} + for sid in ('10001', '10002') + for ds in self.DATE_LIST + } + + def test_backfills_when_no_cache_and_no_program_data(self): + changed = {} + count = _sd_backfill_schedule_dates_without_data( + changed, + self._server_md5s(), + self.DATE_LIST, + ['10001'], + self.EPG_ID_MAP, + set(), + {}, + {'10001'}, + ) + self.assertEqual(count, 3) + self.assertEqual(len(changed['10001']), 3) + + def test_stale_cache_ignored_when_station_has_zero_program_data(self): + """Newly mapped channel: stale MD5 cache must not block backfill.""" + changed = {} + cached_md5s = { + (sid, ds): 'hash' + for sid in ('10001',) + for ds in self.DATE_LIST + } + count = _sd_backfill_schedule_dates_without_data( + changed, + self._server_md5s(), + self.DATE_LIST, + ['10001'], + self.EPG_ID_MAP, + set(), + cached_md5s, + {'10001'}, + ) + self.assertEqual(count, 3) + self.assertEqual(len(changed['10001']), 3) + + def test_skips_cached_empty_day_when_station_has_other_program_data(self): + """Legitimately empty schedule day must not be re-fetched every refresh.""" + changed = {} + cached_md5s = {('10001', '2026-06-12'): 'hash'} + dates_with_data = {(self.EPG_ID, date(2026, 6, 11))} + count = _sd_backfill_schedule_dates_without_data( + changed, + self._server_md5s(), + self.DATE_LIST, + ['10001'], + self.EPG_ID_MAP, + dates_with_data, + cached_md5s, + set(), + ) + self.assertEqual(count, 1) + self.assertEqual(changed['10001'], ['2026-06-13']) + + def test_does_not_duplicate_dates_already_marked_changed(self): + changed = {'10001': ['2026-06-11']} + count = _sd_backfill_schedule_dates_without_data( + changed, + self._server_md5s(), + self.DATE_LIST, + ['10001'], + self.EPG_ID_MAP, + set(), + {}, + {'10001'}, + ) + self.assertEqual(count, 2) + self.assertEqual(sorted(changed['10001']), self.DATE_LIST) + + +class SDProgramMetadataDeltaTests(TestCase): + def test_fetches_when_md5_changed(self): + needed = _sd_programs_needing_metadata( + {'EP0001'}, + {'EP0001': 'new'}, + {'EP0001': 'old'}, + {'EP0001'}, + ) + self.assertEqual(needed, {'EP0001'}) + + def test_fetches_when_no_local_program_data(self): + needed = _sd_programs_needing_metadata( + {'EP0001', 'EP0002'}, + {'EP0001': 'same', 'EP0002': 'same'}, + {'EP0001': 'same', 'EP0002': 'same'}, + {'EP0001'}, + ) + self.assertEqual(needed, {'EP0002'}) + + def test_skips_when_md5_matches_and_program_data_exists(self): + needed = _sd_programs_needing_metadata( + {'EP0001'}, + {'EP0001': 'same'}, + {'EP0001': 'same'}, + {'EP0001'}, + ) + self.assertEqual(needed, set()) + + +class SDScheduleDeltaIntegrationTests(TestCase): + """DB-backed tests for cache pruning and mapped-only MD5 API calls.""" + + MAPPED_STATION = '10001' + UNMAPPED_STATION = '10002' + + def _make_sd_source(self): + return EPGSource.objects.create( + name='SD Integration', + source_type='schedules_direct', + username='sduser', + password='sdpass', + ) + + 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'}]}), + ) + if '/lineups/USA-TEST-X' in url: + return MagicMock( + status_code=200, + json=MagicMock(return_value={ + 'stations': [ + { + 'stationID': self.MAPPED_STATION, + 'name': 'Mapped Station', + 'callsign': 'MAP', + }, + { + 'stationID': self.UNMAPPED_STATION, + 'name': 'Unmapped Station', + 'callsign': 'UNM', + }, + ], + }), + ) + raise AssertionError(f'Unexpected GET URL: {url}') + + def _build_date_list(self, days=3): + today = date.today() + return [(today + timedelta(days=i)).strftime('%Y-%m-%d') for i in range(days)] + + def _seed_full_window_program_data(self, epg, days=3): + today = date.today() + for i in range(days): + day = today + timedelta(days=i) + start = datetime(day.year, day.month, day.day, 12, 0, tzinfo=dt_timezone.utc) + ProgramData.objects.create( + epg=epg, + start_time=start, + end_time=start + timedelta(hours=1), + title='Show', + tvg_id=epg.tvg_id, + ) + + @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_md5_api_only_requests_mapped_stations( + self, mock_post, mock_get, mock_send_epg_update, + ): + from apps.epg.tasks import fetch_schedules_direct + + source = self._make_sd_source() + mapped_epg = EPGData.objects.create( + tvg_id=self.MAPPED_STATION, + name='Mapped', + epg_source=source, + ) + Channel.objects.create(name='Mapped Ch', epg_data=mapped_epg) + + date_list = self._build_date_list(3) + self._seed_full_window_program_data(mapped_epg, days=3) + + today = date.today() + for i, ds in enumerate(date_list): + SDScheduleMD5.objects.create( + epg_source=source, + station_id=self.MAPPED_STATION, + date=today + timedelta(days=i), + md5=f'md5-{ds}', + last_modified=timezone.now(), + ) + SDScheduleMD5.objects.create( + epg_source=source, + station_id=self.UNMAPPED_STATION, + date=today, + md5='unmapped-stale', + last_modified=timezone.now(), + ) + + mock_get.side_effect = self._lineup_get_side_effect + + md5_response_payload = { + self.MAPPED_STATION: { + ds: {'code': 0, 'md5': f'md5-{ds}', 'lastModified': '2026-06-11T00:00:00Z'} + for ds in date_list + }, + } + + 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=md5_response_payload), + ) + raise AssertionError(f'Unexpected POST URL: {url}') + + mock_post.side_effect = post_side_effect + + fetch_schedules_direct(source, force=True) + + md5_calls = [c for c in mock_post.call_args_list if c[0][0].endswith('/schedules/md5')] + self.assertEqual(len(md5_calls), 1) + request_body = md5_calls[0][1]['json'] + station_ids_in_request = {entry['stationID'] for entry in request_body} + self.assertEqual(station_ids_in_request, {self.MAPPED_STATION}) + + self.assertFalse( + SDScheduleMD5.objects.filter( + epg_source=source, + station_id=self.UNMAPPED_STATION, + ).exists() + ) + + schedule_calls = [c for c in mock_post.call_args_list if c[0][0].endswith('/schedules')] + self.assertEqual(len(schedule_calls), 0) + + source.refresh_from_db() + 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') + def test_newly_mapped_station_fetches_despite_stale_cache( + self, mock_post, mock_get, mock_send_epg_update, + ): + from apps.epg.tasks import fetch_schedules_direct + + source = self._make_sd_source() + mapped_epg = EPGData.objects.create( + tvg_id=self.MAPPED_STATION, + name='Mapped', + epg_source=source, + ) + Channel.objects.create(name='Mapped Ch', epg_data=mapped_epg) + + date_list = self._build_date_list(3) + today = date.today() + for i, ds in enumerate(date_list): + SDScheduleMD5.objects.create( + epg_source=source, + station_id=self.MAPPED_STATION, + date=today + timedelta(days=i), + md5=f'md5-{ds}', + last_modified=timezone.now(), + ) + + mock_get.side_effect = self._lineup_get_side_effect + + schedule_payload = [{ + 'stationID': self.MAPPED_STATION, + 'metadata': {'startDate': date_list[0], 'md5': 'md5-' + date_list[0], '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, force=True) + + schedule_calls = [c for c in mock_post.call_args_list if c[0][0].endswith('/schedules')] + self.assertGreaterEqual(len(schedule_calls), 1) + self.assertEqual( + ProgramData.objects.filter(epg=mapped_epg).count(), + 1, + ) + + @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_orphan_program_data_removed_on_post_refresh( + self, mock_post, mock_get, mock_send_epg_update, + ): + from apps.epg.tasks import fetch_schedules_direct + + source = self._make_sd_source() + mapped_epg = EPGData.objects.create( + tvg_id=self.MAPPED_STATION, + name='Mapped', + epg_source=source, + ) + orphan_epg = EPGData.objects.create( + tvg_id=self.UNMAPPED_STATION, + name='Orphan', + epg_source=source, + ) + Channel.objects.create(name='Mapped Ch', epg_data=mapped_epg) + + date_list = self._build_date_list(3) + self._seed_full_window_program_data(mapped_epg, days=3) + + start = timezone.now() + ProgramData.objects.create( + epg=orphan_epg, + start_time=start, + end_time=start + timedelta(hours=1), + title='Orphan Show', + tvg_id=orphan_epg.tvg_id, + ) + + today = date.today() + for i, ds in enumerate(date_list): + SDScheduleMD5.objects.create( + epg_source=source, + station_id=self.MAPPED_STATION, + date=today + timedelta(days=i), + md5=f'md5-{ds}', + last_modified=timezone.now(), + ) + + mock_get.side_effect = self._lineup_get_side_effect + + 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 + }, + }), + ) + raise AssertionError(f'Unexpected POST URL: {url}') + + mock_post.side_effect = post_side_effect + + fetch_schedules_direct(source, force=True) + + self.assertFalse(ProgramData.objects.filter(epg=orphan_epg).exists()) + + def test_stale_program_md5_fetched_when_no_program_data(self): + programs_with_data = set() + needed = _sd_programs_needing_metadata( + {'EP0001'}, + {'EP0001': 'cached-md5'}, + {'EP0001': 'cached-md5'}, + programs_with_data, + ) + self.assertEqual(needed, {'EP0001'}) + + def test_shared_program_skips_metadata_when_cached(self): + needed = _sd_programs_needing_metadata( + {'EP0001'}, + {'EP0001': 'cached-md5'}, + {'EP0001': 'cached-md5'}, + {'EP0001'}, + ) + self.assertEqual(needed, set()) + + # --------------------------------------------------------------------------- # parse_schedules_direct_time tests # --------------------------------------------------------------------------- From 8d59b0b27b830b3e31bb56169272326be00b27a4 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 11 Jun 2026 12:43:16 -0500 Subject: [PATCH 3/3] changelog: Update for SD PR. --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 49637717..f349d186 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Performance +- **Schedules Direct refresh only fetches guide data for mapped channels.** Schedule MD5 checks and schedule downloads now target mapped lineup stations instead of the entire lineup, and schedule MD5 cache for unmapped stations is pruned each refresh. Unmapped lineup entries no longer trigger wasted schedule API calls when their MD5 changes. - **EPG auto-match memory and throughput improvements.** - Single-channel matching streams active EPG rows and keeps only the best match plus the top 20 candidates in memory; ML validates at most 21 names per channel instead of embedding the full catalog. - Strong fuzzy matches (≥75% single channel, ≥80% bulk) skip ML entirely, avoiding a ~500MB PyTorch load when the fuzzy result is already reliable. @@ -49,6 +50,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Bulk auto-match completion no longer calls `batch-set-epg` from the WebSocket handler, which had been re-applying every match and queueing redundant `parse_programs_for_tvg_id` tasks even when assignments were unchanged. - **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) ## [0.26.0] - 2026-06-07