From b928de7d50cf53ba8b26d8c1afe17458a7ac6e5d Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 7 Jun 2026 09:56:25 -0500 Subject: [PATCH] fix(epg): resolve EPG channel list refresh issue after parsing completion - Addressed a bug where the EPG channel list did not refresh after a source finished parsing channels due to the `epg_refresh` WebSocket handler using a stale snapshot. The handler now correctly reads the current store state and ensures `fetchEPGData()` is called upon completion of channel parsing. - Updated tests to verify that the frontend receives the correct parsing completion signals. --- CHANGELOG.md | 1 + apps/epg/tasks.py | 2 - apps/epg/tests/test_schedules_direct.py | 71 ++++++++++++- frontend/src/WebSocket.jsx | 133 ++++++++++++------------ 4 files changed, 140 insertions(+), 67 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a340006c..f2c1a6d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **EPG channel list did not refresh after a source finished parsing channels.** The `epg_refresh` WebSocket handler closed over a stale `epgs` snapshot from when the socket connected. When a newly created source was not found in that snapshot, the handler updated progress and returned early without calling `fetchEPGData()`, so the channel picker and EPG assignment UI stayed empty until a full page reload. The handler now reads the current store via `getState()` and still calls `fetchEPGData()` when `parsing_channels` completes. - **DVR recordings no longer stop immediately when FFmpeg exits mid-stream.** If FFmpeg crashes, stalls, or loses the source before the scheduled end time, `run_recording` now restarts it in-process instead of going straight to HLS→MKV concat and marking the recording complete. Each restart reuses the same HLS working directory and continues segment numbering (`append_list`, `-start_number`) so the final MKV is a single continuous file. Retries are bounded by a per-outage time window matching live-proxy client tolerance (`STREAM_TIMEOUT` + `FAILOVER_GRACE_PERIOD`, default 80s); the window resets whenever new segments resume, so a long recording can survive multiple separate outages. Reconnect pacing between attempts follows the same `min(0.25 × attempt, 3s)` backoff used by the live-proxy `StreamManager`. Input demuxing also uses `-err_detect ignore_err` to tolerate minor TS corruption without aborting the process. If the outage window expires without recovery, the recording is saved as `interrupted` with `ffmpeg_outage_window_exhausted` rather than falsely reported as `completed`. (Closes #1170) - **DVR HLS→MKV concat now tolerates timestamp splices and corrupt segments.** The finalize step (and startup recovery concat) previously used a bare `ffmpeg -f concat -c copy`, which could fail on truncated tail segments or PTS discontinuities from FFmpeg restarts mid-recording. Concat now uses a shared `_dvr_build_hls_concat_cmd` helper with `-fflags +genpts+igndts+discardcorrupt`, `-err_detect ignore_err`, and `-avoid_negative_ts make_zero`; the existing MP4-intermediate fallback path uses the same tolerant input flags. - **DVR FFmpeg restart no longer skips HLS segment numbers.** On restart, `-start_number` was set to the existing segment count while `append_list` also incremented the internal sequence for each segment reloaded from `index.m3u8`, so the first post-restart segment could land at roughly double the expected index (e.g. `seg_00028.ts` after 14 segments). Restarts now pass `-start_number 0` when the playlist already lists segments and let `append_list` continue numbering; loose `.ts` files without a playlist still seed from the highest filename index + 1. diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 55b7d9b0..d868dfeb 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -1360,8 +1360,6 @@ def parse_channels_only(source): channels_count=processed_channels ) - send_websocket_update('updates', 'update', {"success": True, "type": "epg_channels"}) - logger.info(f"Finished parsing channel info. Found {processed_channels} channels.") from apps.channels.utils import maybe_auto_apply_epg_logos diff --git a/apps/epg/tests/test_schedules_direct.py b/apps/epg/tests/test_schedules_direct.py index 16899dac..77334112 100644 --- a/apps/epg/tests/test_schedules_direct.py +++ b/apps/epg/tests/test_schedules_direct.py @@ -18,7 +18,7 @@ from unittest.mock import MagicMock, patch from django.test import TestCase from django.utils import timezone -from apps.epg.models import EPGSource +from apps.epg.models import EPGSource, EPGData from apps.epg.serializers import EPGSourceSerializer @@ -201,6 +201,75 @@ class FetchSchedulesDirectAuthTests(TestCase): self.assertEqual(source.status, EPGSource.STATUS_ERROR) +class FetchSchedulesDirectStationsOnlyTests(TestCase): + """stations_only fetch must signal channel parsing completion to the frontend.""" + + @patch('apps.epg.tasks.send_epg_update') + @patch('apps.epg.tasks.requests.get') + @patch('apps.epg.tasks.requests.post') + def test_stations_only_sends_parsing_channels_complete( + self, mock_post, mock_get, mock_send_epg_update + ): + from apps.epg.tasks import fetch_schedules_direct + + mock_post.return_value = MagicMock( + status_code=200, + json=MagicMock(return_value={'code': 0, 'token': 'tok123'}), + ) + + def get_side_effect(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': '10001', + 'name': 'Test Station', + 'callsign': 'TEST', + }], + }), + ) + raise AssertionError(f'Unexpected GET URL: {url}') + + mock_get.side_effect = get_side_effect + + source = EPGSource.objects.create( + name='SD Stations Only', + source_type='schedules_direct', + username='sduser', + password='sdpass', + ) + + fetch_schedules_direct(source, stations_only=True) + + source.refresh_from_db() + self.assertEqual(source.status, EPGSource.STATUS_SUCCESS) + self.assertEqual(EPGData.objects.filter(epg_source=source).count(), 1) + + parsing_channel_complete = [ + c + for c in mock_send_epg_update.call_args_list + if c[0][1] == 'parsing_channels' and c[0][2] == 100 + ] + self.assertEqual(len(parsing_channel_complete), 1) + complete_call = parsing_channel_complete[0] + self.assertEqual(complete_call[0][0], source.id) + self.assertEqual(complete_call[1]['status'], 'success') + self.assertEqual(complete_call[1]['channels_count'], 1) + + # --------------------------------------------------------------------------- # parse_schedules_direct_time tests # --------------------------------------------------------------------------- diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index 294e879e..0e70a46c 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -50,8 +50,6 @@ export const WebsocketProvider = ({ children }) => { const epgs = useEPGsStore((s) => s.epgs); const updateEPG = useEPGsStore((s) => s.updateEPG); - const updateEPGProgress = useEPGsStore((s) => s.updateEPGProgress); - const fetchEPGsForProgress = useEPGsStore((s) => s.fetchEPGs); const updatePlaylist = usePlaylistsStore((s) => s.updatePlaylist); @@ -641,80 +639,87 @@ export const WebsocketProvider = ({ children }) => { } break; - case 'epg_refresh': - // If we have source/account info, check if EPG exists before processing - if (parsedEvent.data.source || parsedEvent.data.account) { - const sourceId = - parsedEvent.data.source || parsedEvent.data.account; - const epg = epgs[sourceId]; + case 'epg_refresh': { + const sourceId = + parsedEvent.data.source || parsedEvent.data.account; + if (!sourceId) break; - // If EPG not in store yet (e.g. newly created and refresh started - // before the store was updated), fetch the EPG list and update progress. - if (!epg) { - fetchEPGsForProgress().then(() => { - updateEPGProgress(parsedEvent.data); - }); - break; + // Read from the store directly. connectWebSocket closes over a stale + // epgs snapshot, so a newly created source is missed and the old early- + // return path never reached fetchEPGData on parsing_channels completion. + let { epgs: epgsState, updateEPG, updateEPGProgress, fetchEPGs, fetchEPGData } = + useEPGsStore.getState(); + + if (!epgsState[sourceId]) { + try { + await fetchEPGs(); + } catch (e) { + console.warn( + 'Failed to refresh EPG sources for progress update:', + e + ); } + epgsState = useEPGsStore.getState().epgs; + } - // Update the store with progress information - updateEPGProgress(parsedEvent.data); + updateEPGProgress(parsedEvent.data); - if (epg) { - // Check for any indication of an error (either via status or error field) - const hasError = - parsedEvent.data.status === 'error' || - !!parsedEvent.data.error || - (parsedEvent.data.message && - parsedEvent.data.message.toLowerCase().includes('error')); + const epg = epgsState[sourceId]; + if (!epg) break; - if (hasError) { - // Handle error state - const errorMessage = - parsedEvent.data.error || - parsedEvent.data.message || - 'Unknown error occurred'; + const hasError = + parsedEvent.data.status === 'error' || + !!parsedEvent.data.error || + (parsedEvent.data.message && + parsedEvent.data.message.toLowerCase().includes('error')); - updateEPG({ - ...epg, - status: 'error', - last_message: errorMessage, - }); + if (hasError) { + const errorMessage = + parsedEvent.data.error || + parsedEvent.data.message || + 'Unknown error occurred'; - // Show notification for the error - notifications.show({ - title: 'EPG Refresh Error', - message: errorMessage, - color: 'red.5', - }); - } - // Update status on completion only if no errors - else if (parsedEvent.data.progress === 100) { - updateEPG({ - ...epg, - status: parsedEvent.data.status || 'success', - last_message: - parsedEvent.data.message || epg.last_message, - // Use the timestamp from the backend if provided - ...(parsedEvent.data.updated_at && { - updated_at: parsedEvent.data.updated_at, - }), - }); + updateEPG({ + ...epg, + status: 'error', + last_message: errorMessage, + }); - // Only show success notification if we've finished parsing programs and had no errors - if (parsedEvent.data.action === 'parsing_programs') { - notifications.show({ - title: 'EPG Processing Complete', - message: 'EPG data has been updated successfully', - color: 'green.5', - }); + notifications.show({ + title: 'EPG Refresh Error', + message: errorMessage, + color: 'red.5', + }); + } else if (parsedEvent.data.progress === 100) { + updateEPG({ + ...epg, + status: parsedEvent.data.status || 'success', + last_message: + parsedEvent.data.message || epg.last_message, + ...(parsedEvent.data.updated_at && { + updated_at: parsedEvent.data.updated_at, + }), + }); - fetchEPGData(); - } - } + if (parsedEvent.data.action === 'parsing_channels') { + notifications.show({ + message: 'EPG channels updated!', + color: 'green.5', + }); + + await fetchEPGData(); + } else if (parsedEvent.data.action === 'parsing_programs') { + notifications.show({ + title: 'EPG Processing Complete', + message: 'EPG data has been updated successfully', + color: 'green.5', + }); + + await fetchEPGData(); } } break; + } case 'epg_sources_changed': // A plugin or backend process signaled that the EPG sources changed