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 # ---------------------------------------------------------------------------