From 14a6bf4473bc4636d679ad2bf7d2d11e8957dc85 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 26 Apr 2026 09:06:48 -0500 Subject: [PATCH] Initial DVR HLS refactor. --- apps/channels/api_urls.py | 5 + apps/channels/api_views.py | 105 +- apps/channels/tasks.py | 1126 +++++++++-------- .../tests/test_dvr_port_resolution.py | 88 +- docker/docker-compose.yml | 6 +- 5 files changed, 742 insertions(+), 588 deletions(-) diff --git a/apps/channels/api_urls.py b/apps/channels/api_urls.py index bd53ae45..b2af3167 100644 --- a/apps/channels/api_urls.py +++ b/apps/channels/api_urls.py @@ -49,6 +49,11 @@ urlpatterns = [ path('series-rules/bulk-remove/', BulkRemoveSeriesRecordingsAPIView.as_view(), name='bulk_remove_series_recordings'), path('series-rules//', DeleteSeriesRuleAPIView.as_view(), name='delete_series_rule'), path('recordings/bulk-delete-upcoming/', BulkDeleteUpcomingRecordingsAPIView.as_view(), name='bulk_delete_upcoming_recordings'), + path( + 'recordings//hls/', + RecordingViewSet.as_view({'get': 'hls'}), + name='recording-hls', + ), path('dvr/comskip-config/', ComskipConfigAPIView.as_view(), name='comskip_config'), ] diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 3e786ff1..2907c48d 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -2444,7 +2444,7 @@ class RecordingViewSet(viewsets.ModelViewSet): def get_permissions(self): # Allow unauthenticated playback of recording files (like other streaming endpoints) - if self.action == 'file': + if self.action in ('file', 'hls'): return [AllowAny()] try: return [perm() for perm in permission_classes_by_action[self.action]] @@ -2464,14 +2464,29 @@ class RecordingViewSet(viewsets.ModelViewSet): @action(detail=True, methods=["get"], url_path="file") def file(self, request, pk=None): - """Stream a recorded file with HTTP Range support for seeking.""" + """Stream a completed recording file with HTTP Range support for seeking. + + For in-progress recordings, file_url in custom_properties points to + /hls/index.m3u8. If a client hits this endpoint while the recording + is still running (or the MKV is not yet produced), it is redirected to + the HLS playlist endpoint. + """ recording = get_object_or_404(Recording, pk=pk) cp = recording.custom_properties or {} file_path = cp.get("file_path") file_name = cp.get("file_name") or "recording" - if not file_path or not os.path.exists(file_path): - raise Http404("Recording file not found") + if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0: + # Redirect to HLS if recording is still in progress + hls_dir = cp.get("_hls_dir") + if hls_dir and os.path.isdir(hls_dir): + from django.http import HttpResponseRedirect + hls_url = request.build_absolute_uri( + f"/api/channels/recordings/{pk}/hls/index.m3u8" + ) + return HttpResponseRedirect(hls_url) + if not file_path or not os.path.exists(file_path): + raise Http404("Recording file not found") # Guess content type ext = os.path.splitext(file_path)[1].lower() @@ -2532,6 +2547,73 @@ class RecordingViewSet(viewsets.ModelViewSet): response["Content-Disposition"] = f"inline; filename=\"{file_name}\"" return response + @action(detail=True, methods=["get"], url_path="hls/(?P.+)") + def hls(self, request, pk=None, seg_path="index.m3u8"): + """Serve HLS playlist and segment files for an in-progress (or completed) recording. + + Clients connecting during recording should use the m3u8 URL returned in + custom_properties.file_url. Segment URLs inside the playlist are rewritten + to route through this endpoint so authentication and path isolation are + preserved. + """ + recording = get_object_or_404(Recording, pk=pk) + cp = recording.custom_properties or {} + hls_dir = cp.get("_hls_dir") + + if not hls_dir or not os.path.isdir(hls_dir): + # HLS dir is gone, recording is likely complete. Redirect to the + # permanent MKV endpoint for .m3u8 requests so clients that still + # have the HLS URL bookmarked get a useful response. + cp = recording.custom_properties or {} + file_path = cp.get("file_path") + if seg_path.endswith(".m3u8") and file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0: + from django.http import HttpResponseRedirect + return HttpResponseRedirect( + request.build_absolute_uri(f"/api/channels/recordings/{pk}/file/") + ) + raise Http404("HLS content not available for this recording") + + # Security: prevent path traversal outside the HLS directory + safe_dir = os.path.realpath(hls_dir) + requested = os.path.realpath(os.path.join(hls_dir, seg_path)) + if not requested.startswith(safe_dir + os.sep) and requested != safe_dir: + return Response({"error": "Forbidden"}, status=403) + + if not os.path.isfile(requested): + raise Http404(f"HLS file not found: {seg_path}") + + if seg_path.endswith(".m3u8"): + # Rewrite relative segment lines to absolute URLs through this API + base_url = request.build_absolute_uri( + f"/api/channels/recordings/{pk}/hls/" + ) + lines = [] + with open(requested) as _f: + for line in _f: + stripped = line.strip() + if stripped and not stripped.startswith("#"): + lines.append(f"{base_url}{stripped}\n") + else: + lines.append(line) + from django.http import HttpResponse as _HR + return _HR("".join(lines), content_type="application/x-mpegURL") + + if seg_path.endswith(".ts"): + # Refresh the viewer heartbeat in Redis so the Celery task knows an + # active client is still fetching segments. TTL is 20 s, enough for + # three 4-second segments plus network margin. + try: + from core.utils import RedisClient + _rv = RedisClient.get_client(max_retries=1, retry_interval=0) + if _rv: + _rv.set(f"dvr:hls_viewer:{pk}", "1", ex=20) + except Exception: + pass + from django.http import FileResponse as _FR + return _FR(open(requested, "rb"), content_type="video/mp2t") + + raise Http404("Unsupported HLS file type") + @action(detail=True, methods=["post"], url_path="stop") def stop(self, request, pk=None): """Stop a recording early while retaining the partial content for playback.""" @@ -2838,7 +2920,7 @@ class RecordingViewSet(viewsets.ModelViewSet): cp = instance.custom_properties or {} rec_status = cp.get("status", "") file_path = cp.get("file_path") - temp_ts_path = cp.get("_temp_file_path") + hls_dir = cp.get("_hls_dir") channel_uuid = str(instance.channel.uuid) # 1. Delete the DB record (also fires post_delete → revoke_task_on_delete) @@ -2871,6 +2953,17 @@ class RecordingViewSet(viewsets.ModelViewSet): except Exception as ex: logger.warning(f"Failed to delete recording artifact {path}: {ex}") + def _safe_rmtree(path: str): + if not path or not isinstance(path, str): + return + try: + import shutil as _shutil + if any(path.startswith(root) for root in allowed_roots) and os.path.isdir(path): + _shutil.rmtree(path) + logger.info(f"Deleted recording HLS directory: {path}") + except Exception as ex: + logger.warning(f"Failed to delete HLS directory {path}: {ex}") + def _background_cancel(): # Only stop the DVR client if the recording was actively streaming. # Stopping for completed/upcoming recordings would kill an unrelated @@ -2888,7 +2981,7 @@ class RecordingViewSet(viewsets.ModelViewSet): # Best-effort file cleanup in case run_recording already exited # before the DB delete. _safe_remove(file_path) - _safe_remove(temp_ts_path) + _safe_rmtree(hls_dir) try: from django.db import connection as _conn diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 59122d9f..67bec477 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -8,6 +8,8 @@ import time import json import subprocess import signal +import threading +from collections import deque from zoneinfo import ZoneInfo from datetime import datetime, timedelta import gc @@ -1654,22 +1656,18 @@ def _build_output_paths(channel, program, start_time, end_time): final_path = os.path.normpath(final_path) # Avoid overwriting an existing file from a different recording. - # Check BOTH .mkv and .ts — a pre-restart TS segment may exist at - # the same base name even when the MKV is a 0-byte placeholder. + # Check the MKV file and its associated HLS working directory. base, ext = os.path.splitext(final_path) counter = 1 while True: candidate_base = final_path[:-len(ext)] # strip extension - ts_candidate = candidate_base + '.ts' + hls_dir_candidate = candidate_base + '_hls' try: mkv_occupied = os.stat(final_path).st_size > 0 except OSError: mkv_occupied = False - try: - ts_occupied = os.stat(ts_candidate).st_size > 0 - except OSError: - ts_occupied = False - if not mkv_occupied and not ts_occupied: + hls_occupied = os.path.isdir(hls_dir_candidate) + if not mkv_occupied and not hls_occupied: break counter += 1 final_path = f"{base}_{counter}{ext}" @@ -1677,37 +1675,36 @@ def _build_output_paths(channel, program, start_time, end_time): # Ensure directory exists os.makedirs(os.path.dirname(final_path), exist_ok=True) - # Derive temp TS path in same directory + # Derive HLS working directory alongside the final MKV base_no_ext = os.path.splitext(os.path.basename(final_path))[0] - temp_ts_path = os.path.join(os.path.dirname(final_path), f"{base_no_ext}.ts") - return final_path, temp_ts_path, os.path.basename(final_path) + hls_dir = os.path.join(os.path.dirname(final_path), f"{base_no_ext}_hls") + return final_path, hls_dir, os.path.basename(final_path) -def build_dvr_candidates(): - """Build ordered list of candidate base URLs for DVR TS streaming. +def get_dvr_stream_base_url(): + """Return the single correct base URL for DVR to reach the TS stream proxy. - Reads environment variables to determine which URLs to try: - - DISPATCHARR_INTERNAL_TS_BASE_URL: explicit override (first priority) - - DISPATCHARR_PORT: the external port (default 9191) - - DISPATCHARR_ENV/DISPATCHARR_DEBUG/REDIS_HOST: dev-mode detection - - DISPATCHARR_INTERNAL_API_BASE: override for the docker service URL + Priority: + 1. DISPATCHARR_INTERNAL_TS_BASE_URL — explicit override, always wins. + 2. Modular mode (DISPATCHARR_ENV=modular) — celery runs in a separate container + and must reach the web container by its Docker service name on DISPATCHARR_PORT. + Override the host with DISPATCHARR_WEB_HOST for non-standard compose setups. + 3. AIO / dev / debug — celery shares the container with uwsgi which binds on + port 5656; use 127.0.0.1 to avoid any nginx layer. """ explicit = os.environ.get('DISPATCHARR_INTERNAL_TS_BASE_URL') - dispatcharr_port = os.environ.get('DISPATCHARR_PORT', '9191') - is_dev = (os.environ.get('DISPATCHARR_ENV', '').lower() == 'dev') or \ - (os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true') or \ - (os.environ.get('REDIS_HOST', 'redis') in ('localhost', '127.0.0.1')) - candidates = [] if explicit: - candidates.append(explicit) - if is_dev: - # Debug container typically exposes API on 5656 (uwsgi internal port) - candidates.extend(['http://127.0.0.1:5656', f'http://127.0.0.1:{dispatcharr_port}']) - # Docker service name fallback — use DISPATCHARR_PORT so modular mode works with custom ports - candidates.append(os.environ.get('DISPATCHARR_INTERNAL_API_BASE', f'http://web:{dispatcharr_port}')) - # Last-resort localhost ports - candidates.extend(['http://localhost:5656', f'http://localhost:{dispatcharr_port}']) - return candidates + return explicit.rstrip('/') + + dispatcharr_env = os.environ.get('DISPATCHARR_ENV', 'aio').lower() + + if dispatcharr_env == 'modular': + host = os.environ.get('DISPATCHARR_WEB_HOST', 'web') + port = os.environ.get('DISPATCHARR_PORT', '9191') + return f'http://{host}:{port}' + + # AIO, dev, debug: celery and uwsgi share the container, reach uwsgi directly + return 'http://127.0.0.1:5656' @shared_task @@ -1759,11 +1756,10 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): start_time = datetime.fromisoformat(start_time_str) end_time = datetime.fromisoformat(end_time_str) - duration_seconds = int((end_time - start_time).total_seconds()) # Build output paths from templates (refined after loading Recording cp below) filename = None final_path = None - temp_ts_path = None + hls_dir = None channel_layer = get_channel_layer() @@ -1810,7 +1806,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): "started_at": str(datetime.now()), }) # Provide a predictable playback URL for the frontend - cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/" + cp["file_url"] = f"/api/channels/recordings/{recording_id}/hls/index.m3u8" cp["output_file_url"] = cp["file_url"] # Determine program info (may include id for deeper details) @@ -1825,10 +1821,10 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): program.update(epg_match) cp["program"] = program - final_path, temp_ts_path, filename = _build_output_paths(channel, program, start_time, end_time) + final_path, hls_dir, filename = _build_output_paths(channel, program, start_time, end_time) cp["file_name"] = filename cp["file_path"] = final_path - cp["_temp_file_path"] = temp_ts_path + cp["_hls_dir"] = hls_dir # Resolve poster art via the shared pipeline (EPG → VOD → TMDB/OMDb → # TVMaze/iTunes → direct program fields → Logo table → channel logo). @@ -1863,7 +1859,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # Merge only the keys explicitly set into the fresh copy for key in ("status", "started_at", "file_url", "output_file_url", - "file_name", "file_path", "_temp_file_path", + "file_name", "file_path", "_hls_dir", "program", "poster_logo_id", "poster_url"): if key in cp: fresh_cp[key] = cp[key] @@ -1886,15 +1882,8 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): interrupted_reason = None bytes_written = 0 - from requests.exceptions import ReadTimeout, ConnectionError as ReqConnectionError, ChunkedEncodingError - - candidates = build_dvr_candidates() - - chosen_base = None - last_error = None - bytes_written = 0 - interrupted = False - interrupted_reason = None + base_url = get_dvr_stream_base_url() + logger.info(f"DVR recording {recording_id}: using stream base URL {base_url}") def _check_recording_cancelled(rid): """Check if a recording was stopped by user or deleted. @@ -1912,232 +1901,331 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): pass return False, False, None - # --- Retry / reconnection constants --- - # Stream reconnection: retry the same TS proxy base on transient - # connectivity loss. Counter resets when data resumes. - _dvr_max_reconnects = 5 - _dvr_reconnect_delay = 2.0 # seconds - # DB save retry: exponential backoff (1s, 2s, 4s) for transient errors. + # --- DB retry constants --- _dvr_db_max_retries = 3 _dvr_db_retry_interval = 1 # seconds (base for exponential backoff) - # FFmpeg remux retry: covers transient I/O errors. - _dvr_remux_max_retries = 2 - _dvr_remux_retry_interval = 2 # seconds (base for exponential backoff) - for base in candidates: - test_url = f"{base.rstrip('/')}/proxy/ts/stream/{channel.uuid}" - logger.info(f"DVR recording {recording_id}: trying TS base {base}") + # Redis key used to signal that an HLS client is actively fetching segments. + # The hls endpoint refreshes this on every .ts request; we check it before + # deleting the HLS directory so in-flight downloads are not cut short. + _hls_viewer_key = f"dvr:hls_viewer:{recording_id}" - _reconnects = 0 - _file_mode = 'wb' - _stream_started_at = None - _done = False + # --- HLS recording pipeline --- + # Launch FFmpeg with the deterministic base URL resolved above. Wait up to + # _first_segment_timeout seconds for the first segment to appear before + # treating the stream as unavailable. + from django.utils import timezone as _tz - while True: # Reconnection loop for this base + last_error = None + ffmpeg_proc = None + hls_m3u8 = None + hls_seg_pattern = None + _stream_confirmed = False + + if hls_dir: + os.makedirs(hls_dir, exist_ok=True) + hls_m3u8 = os.path.join(hls_dir, "index.m3u8") + hls_seg_pattern = os.path.join(hls_dir, "seg_%05d.ts") + + base = base_url + + if not interrupted: + # Check for stop/delete before starting + should_exit, is_int, reason = _check_recording_cancelled(recording_id) + if should_exit: + interrupted = is_int + interrupted_reason = reason + + if not interrupted: + stream_url = f"{base}/proxy/ts/stream/{channel.uuid}" + logger.info(f"DVR recording {recording_id}: stream URL: {stream_url}") + + if not interrupted: + + # Continue segment numbering from any previous session (server-restart resume) + existing_segs = sorted( + f for f in os.listdir(hls_dir) if f.startswith("seg_") and f.endswith(".ts") + ) if hls_dir else [] + hls_start_number = len(existing_segs) + + ffmpeg_cmd = [ + "ffmpeg", "-y", + "-reconnect", "1", + "-reconnect_streamed", "1", + "-reconnect_delay_max", "5", + "-user_agent", f"Dispatcharr-DVR/recording-{recording_id}", + # Regenerate monotonic PTS to handle erratic/discontinuous timestamps + # from IPTV sources. + "-fflags", "+genpts", + "-i", stream_url, + "-c", "copy", + # Shift output timestamps so they start from 0, fixing negative PTS + # values that can prevent segment boundary detection in the HLS muxer. + "-avoid_negative_ts", "make_zero", + "-f", "hls", + "-hls_time", "4", + "-hls_list_size", "0", + "-hls_flags", "append_list+omit_endlist+independent_segments", + "-start_number", str(hls_start_number), + "-hls_segment_filename", hls_seg_pattern, + hls_m3u8, + ] + + logger.info(f"DVR recording {recording_id}: starting FFmpeg — stream URL: {stream_url}") + logger.info(f"DVR recording {recording_id}: HLS output dir: {hls_dir}") + logger.debug(f"DVR recording {recording_id}: FFmpeg command: {' '.join(str(a) for a in ffmpeg_cmd)}") + # Rolling tail of FFmpeg stderr lines for post-mortem diagnostics + _ffmpeg_stderr_tail = deque(maxlen=200) + try: + ffmpeg_proc = subprocess.Popen( + ffmpeg_cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + except Exception as _fe: + last_error = str(_fe) + logger.warning(f"DVR recording {recording_id}: failed to launch FFmpeg: {_fe}") + ffmpeg_proc = None + + # Drain FFmpeg stderr in a background thread to prevent the OS pipe + # buffer (~64 KB on Linux) from filling up, which would block FFmpeg's + # writes and cause it to silently stop demuxing/segmenting after a few + # minutes of normal progress output. Lines are logged to the Dispatcharr + # log and tail-buffered for diagnostics. + def _drain_ffmpeg_stderr(proc, rec_id, tail): + # FFmpeg emits progress lines terminated with \r (carriage return) + # so it can rewrite them in place; only non-progress messages end + # with \n. Split on either to capture both kinds. try: - with requests.get( - test_url, - headers={ - 'User-Agent': f'Dispatcharr-DVR/recording-{recording_id}', - }, - stream=True, - timeout=(10, 15), - ) as response: - response.raise_for_status() - - _test_window = 3.0 - _window_start = time.time() - _stop_poll_interval = 2.0 - _last_stop_poll = time.time() - - with open(temp_ts_path, _file_mode) as file: - if _stream_started_at is None: - _stream_started_at = time.time() - - for chunk in response.iter_content(chunk_size=8192): - if not chunk: - if not chosen_base and (time.time() - _window_start) > _test_window: - break - continue - - if not chosen_base: - chosen_base = base - - # Data received after reconnect — connection restored - if _reconnects > 0: - logger.info( - f"DVR recording {recording_id}: " - f"stream resumed after reconnect" - ) - _reconnects = 0 - - file.write(chunk) - bytes_written += len(chunk) - - elapsed = time.time() - _stream_started_at - if elapsed > duration_seconds: - break - - # Periodic DB poll: stop, delete, end_time extension - _now = time.time() - if _now - _last_stop_poll >= _stop_poll_interval: - _last_stop_poll = _now - try: - _sc = Recording.objects.filter( - id=recording_id - ).only("custom_properties", "end_time").first() - if _sc is None: - logger.info( - f"DVR recording {recording_id}: " - f"deleted — exiting stream loop" - ) - interrupted = False - break - if (_sc.custom_properties or {}).get("status") == "stopped": - logger.info( - f"DVR recording {recording_id}: " - f"stop requested — exiting stream loop" - ) - break - try: - new_end = _sc.end_time - if new_end is not None: - from django.utils import timezone as _tz - if _tz.is_naive(new_end): - new_end = _tz.make_aware(new_end) - _ref = start_time - if _tz.is_naive(_ref): - _ref = _tz.make_aware(_ref) - new_duration = int( - (new_end - _ref).total_seconds() - ) - if new_duration > duration_seconds: - logger.info( - f"DVR recording {recording_id}: " - f"end_time extended to {new_end}, " - f"new duration {new_duration}s" - ) - duration_seconds = new_duration - except Exception: - pass - except Exception: - pass - - # iter_content exhausted or loop exited normally - if bytes_written > 0: - logger.info( - f"DVR recording {recording_id}: " - f"stream complete, {bytes_written} bytes written" - ) - _done = True - else: - last_error = f"no_data_from_{base}" - logger.warning( - f"DVR recording {recording_id}: no data from " - f"{base} within {_test_window}s, trying next base" - ) - try: - if os.path.exists(temp_ts_path) and os.path.getsize(temp_ts_path) == 0: - os.remove(temp_ts_path) - except FileNotFoundError: - pass - break # Exit reconnection loop - - except (ReadTimeout, ReqConnectionError, ChunkedEncodingError) as e: - if bytes_written > 0: - # Active stream lost — check cancellation before reconnecting - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit: - interrupted = is_int - interrupted_reason = reason - if reason == "stopped_by_user": - logger.info( - f"DVR recording {recording_id}: " - f"stopped by user — ending stream" - ) - _done = True + buf = bytearray() + stream = proc.stderr + while True: + byte = stream.read(1) + if not byte: break - - _reconnects += 1 - if _reconnects <= _dvr_max_reconnects: - logger.warning( - f"DVR recording {recording_id}: connection lost " - f"({type(e).__name__}), reconnecting " - f"({_reconnects}/{_dvr_max_reconnects}) " - f"in {_dvr_reconnect_delay}s..." - ) - time.sleep(_dvr_reconnect_delay) - _file_mode = 'ab' + if byte in (b'\r', b'\n'): + if buf: + line = buf.decode('utf-8', errors='replace').strip() + buf.clear() + if line: + tail.append(line) + low = line.lower() + if 'error' in low or 'failed' in low or 'invalid' in low: + logger.warning(f"DVR recording {rec_id} ffmpeg: {line}") + else: + logger.debug(f"DVR recording {rec_id} ffmpeg: {line}") continue + buf.append(byte[0]) + # Safety net: flush absurdly long lines so a malformed + # stream can't grow the buffer without bound. + if len(buf) > 4096: + line = buf.decode('utf-8', errors='replace').strip() + buf.clear() + if line: + tail.append(line) + logger.debug(f"DVR recording {rec_id} ffmpeg: {line}") + # Flush any trailing content + if buf: + line = buf.decode('utf-8', errors='replace').strip() + if line: + tail.append(line) + logger.debug(f"DVR recording {rec_id} ffmpeg: {line}") + except Exception as _de: + logger.debug(f"DVR recording {rec_id}: stderr drain ended: {_de}") - logger.error( - f"DVR recording {recording_id}: max reconnects " - f"({_dvr_max_reconnects}) exceeded — ending recording" - ) - interrupted = True - interrupted_reason = ( - f"stream_interrupted: max reconnects exceeded ({e})" - ) - _done = True - break + if ffmpeg_proc is not None and ffmpeg_proc.stderr is not None: + _stderr_thread = threading.Thread( + target=_drain_ffmpeg_stderr, + args=(ffmpeg_proc, recording_id, _ffmpeg_stderr_tail), + daemon=True, + name=f"dvr-ffmpeg-stderr-{recording_id}", + ) + _stderr_thread.start() - # No data received yet — retry same base before moving on - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit: - interrupted = is_int - interrupted_reason = reason - _done = True - break - _reconnects += 1 - if _reconnects <= _dvr_max_reconnects: + end_timestamp = ( + end_time.timestamp() + if hasattr(end_time, 'timestamp') + else time.mktime(end_time.timetuple()) + ) + _stop_poll_interval = 2.0 + _last_stop_poll = time.time() + _ffmpeg_start = time.time() + _first_segment_timeout = 15.0 + _stall_timeout = 60.0 # seconds without new segments → source stream gone + _stream_confirmed = False + _last_seg_count = hls_start_number + _last_new_seg_time = time.time() + + while ffmpeg_proc.poll() is None: + time.sleep(0.5) + now = time.time() + + segs_now = [ + f for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ] if hls_dir else [] + + # Wait for the first segment to confirm data is flowing + if not _stream_confirmed: + if segs_now: + _stream_confirmed = True + _last_seg_count = len(segs_now) + _last_new_seg_time = now + logger.info( + f"DVR recording {recording_id}: first HLS segment written, stream confirmed" + ) + elif now - _ffmpeg_start > _first_segment_timeout: logger.warning( - f"DVR recording {recording_id}: initial connection " - f"to {base} failed ({type(e).__name__}), retrying " - f"({_reconnects}/{_dvr_max_reconnects}) " - f"in {_dvr_reconnect_delay}s..." + f"DVR recording {recording_id}: no HLS segments produced after " + f"{_first_segment_timeout}s from {base} — stream unavailable" ) - time.sleep(_dvr_reconnect_delay) - continue - last_error = str(e) - logger.warning( - f"DVR recording {recording_id}: base {base} exhausted " - f"retries ({_dvr_max_reconnects}): {e}" - ) - break - - except Exception as e: - last_error = str(e) - logger.warning(f"DVR recording {recording_id}: base {base} failed: {e}") - if bytes_written > 0: - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit and reason == "stopped_by_user": - interrupted = False - logger.info( - f"DVR recording {recording_id}: " - f"stopped by user — ending stream" - ) - else: - interrupted = True - interrupted_reason = f"stream_interrupted: {e}" - _done = True + ffmpeg_proc.kill() + try: + ffmpeg_proc.wait(timeout=5) + except subprocess.TimeoutExpired: + pass + last_error = f"no_segments_in_{_first_segment_timeout}s_from_{base}" + ffmpeg_proc = None break - should_exit, is_int, reason = _check_recording_cancelled(recording_id) - if should_exit: - interrupted = is_int - interrupted_reason = reason - _done = True + else: + # Stream was confirmed, detect source stalls (e.g. proxy ghost-kills the client) + if len(segs_now) > _last_seg_count: + _last_seg_count = len(segs_now) + _last_new_seg_time = now + else: + # Also treat a recently-modified segment file as activity. + # With erratic source timestamps, FFmpeg may buffer data inside + # a partially-written segment for longer than hls_time, so the + # segment count won't increase even though data is flowing. + try: + if segs_now and hls_dir: + _newest_mtime = max( + os.path.getmtime(os.path.join(hls_dir, f)) + for f in segs_now + ) + if _newest_mtime > _last_new_seg_time: + _last_new_seg_time = _newest_mtime + except Exception: + pass + if now - _last_new_seg_time > _stall_timeout: + logger.warning( + f"DVR recording {recording_id}: no new HLS segments for " + f"{_stall_timeout:.0f}s — source stream stalled, stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + break + + # Duration check — SIGINT lets FFmpeg write #EXT-X-ENDLIST cleanly + if now >= end_timestamp: + logger.info( + f"DVR recording {recording_id}: scheduled end time reached, stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() break - if _done: - break + # Periodic DB poll: stop, delete, end_time extension + if now - _last_stop_poll >= _stop_poll_interval: + _last_stop_poll = now + try: + _sc = Recording.objects.filter( + id=recording_id + ).only("custom_properties", "end_time").first() + if _sc is None: + logger.info( + f"DVR recording {recording_id}: deleted — stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + interrupted = False + break + if (_sc.custom_properties or {}).get("status") == "stopped": + logger.info( + f"DVR recording {recording_id}: stop requested — stopping FFmpeg" + ) + ffmpeg_proc.send_signal(signal.SIGINT) + try: + ffmpeg_proc.wait(timeout=10) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + break + # Handle end_time extension — just update the deadline, no restart needed + try: + new_end = _sc.end_time + if new_end is not None: + if _tz.is_naive(new_end): + new_end = _tz.make_aware(new_end) + new_ts = new_end.timestamp() + if new_ts > end_timestamp: + logger.info( + f"DVR recording {recording_id}: end_time extended to {new_end}" + ) + end_timestamp = new_ts + except Exception: + pass + except Exception: + pass - if chosen_base is None and bytes_written == 0: + # If FFmpeg exited after confirming the stream, log the exit code so we know why it stopped. + if _stream_confirmed and ffmpeg_proc is not None: + _exit_code = ffmpeg_proc.poll() + if _exit_code is not None and _exit_code != 0: + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited unexpectedly (rc={_exit_code}) " + f"after stream was confirmed — source stream likely disconnected" + ) + + # If FFmpeg exited without the stream being confirmed, log the tail of + # captured stderr for diagnosis. Lines were already logged live by the + # drain thread; this surfaces the recent context in a single message. + elif not _stream_confirmed and ffmpeg_proc is not None: + try: + _exit_code = ffmpeg_proc.poll() + last_error = f"rc={_exit_code} from {base}" + _tail_text = "\n".join(_ffmpeg_stderr_tail) + if _tail_text: + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) " + f"for {base} without producing segments.\nFFmpeg stderr tail:\n{_tail_text[-1000:]}" + ) + else: + logger.warning( + f"DVR recording {recording_id}: FFmpeg exited (rc={_exit_code}) " + f"for {base} without producing segments (no stderr output)" + ) + except Exception: + pass + + # Ensure FFmpeg has fully exited (covers cases where the loop broke early) + if ffmpeg_proc is not None and ffmpeg_proc.poll() is None: + try: + ffmpeg_proc.wait(timeout=5) + except subprocess.TimeoutExpired: + ffmpeg_proc.kill() + + if not _stream_confirmed and not interrupted: interrupted = True - interrupted_reason = f"no_stream_data: {last_error or 'all_bases_failed'}" + interrupted_reason = f"no_stream_data: {last_error or 'ffmpeg_failed'}" - # If no bytes were written at all, check whether this was a deliberate stop or a - # genuine failure. The exception handler above already sets interrupted=False when - # it detects "stopped" status, but do not override that decision here. + # Measure bytes from HLS segment files + try: + bytes_written = sum( + os.path.getsize(os.path.join(hls_dir, f)) + for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ) if hls_dir and os.path.isdir(hls_dir) else 0 + except Exception: + bytes_written = 0 if bytes_written == 0 and not interrupted: _deliberately_stopped = False try: @@ -2178,19 +2266,19 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # After the loop, the file and response are closed automatically. logger.info(f"Finished recording for channel {channel.name}") - # Log system event for recording end - try: - from core.utils import log_system_event - log_system_event( - 'recording_end', - channel_id=channel.uuid, - channel_name=channel.name, - recording_id=recording_id, - interrupted=interrupted, - bytes_written=bytes_written - ) - except Exception as e: - logger.error(f"Could not log recording end event: {e}") + # Log system event for recording end + try: + from core.utils import log_system_event + log_system_event( + 'recording_end', + channel_id=channel.uuid, + channel_name=channel.name, + recording_id=recording_id, + interrupted=interrupted, + bytes_written=bytes_written + ) + except Exception as e: + logger.error(f"Could not log recording end event: {e}") # If the Recording was deleted (cancelled by user), skip post-processing recording_cancelled = not Recording.objects.filter(id=recording_id).exists() @@ -2199,261 +2287,164 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # Clean up all artifacts for the cancelled recording, # including any pre-restart .ts segments from server recovery. # Use the in-memory recording_obj since the DB row is already deleted. - _cancel_cleanup = [temp_ts_path, final_path] - _cancel_cp = (recording_obj.custom_properties or {}) if recording_obj else {} - _cancel_cleanup.extend(_cancel_cp.get("_pre_restart_ts_paths", [])) - for _cleanup_path in _cancel_cleanup: - if not _cleanup_path: - continue + import shutil as _shutil + if hls_dir and os.path.isdir(hls_dir): try: - os.remove(_cleanup_path) - logger.info(f"Cleaned up cancelled recording artifact: {_cleanup_path}") - except FileNotFoundError: - pass + _shutil.rmtree(hls_dir) + logger.info(f"Cleaned up cancelled recording HLS directory: {hls_dir}") + except Exception as _e: + logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}") + if final_path: + try: + if os.path.exists(final_path): + os.remove(final_path) + logger.info(f"Cleaned up cancelled recording MKV placeholder: {final_path}") except Exception: pass return - # Concatenate pre-restart .ts segments with the current segment. - # Instead of creating an intermediate combined.ts and then remuxing to - # MKV (which loses timestamp boundary info and causes playback freezes - # at the splice point), go directly from the concat list → MKV. - # This lets ffmpeg's MKV muxer see each segment boundary and write - # correct cue points / clusters for seamless seeking. - _concat_did_remux = False - try: - _rec_obj_for_concat = Recording.objects.filter(id=recording_id).only("custom_properties").first() - _concat_cp = (_rec_obj_for_concat.custom_properties or {}) if _rec_obj_for_concat else {} - pre_restart_segments = _concat_cp.get("_pre_restart_ts_paths", []) - # Filter to segments that still exist on disk and have data - def _has_data(p): - try: - return os.stat(p).st_size > 0 - except OSError: - return False - pre_restart_segments = [p for p in pre_restart_segments if p and _has_data(p)] - if pre_restart_segments and temp_ts_path and os.path.exists(temp_ts_path): - all_segments = pre_restart_segments + [temp_ts_path] - concat_list_path = temp_ts_path + ".concat.txt" - try: - with open(concat_list_path, "w") as cl: - for seg in all_segments: - cl.write(f"file '{seg}'\n") + # --- Post-processing: concat HLS segments → final MKV --- + remux_success = False + hls_m3u8 = os.path.join(hls_dir, "index.m3u8") if hls_dir else None - # Direct concat → MKV in a single pass. - # -reset_timestamps 1 tells the concat demuxer to reset - # timestamps at each segment boundary, eliminating the - # discontinuity that causes playback to freeze at the - # splice point. - concat_result = subprocess.run( - [ - "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", - "-err_detect", "ignore_err", - "-f", "concat", "-safe", "0", - "-segment_time_metadata", "1", - "-i", concat_list_path, - "-reset_timestamps", "1", - "-map", "0", - "-c", "copy", - final_path, - ], - stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - ) - if concat_result.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0: - _concat_did_remux = True - # Clean up individual TS segments (including current) - for seg in all_segments: - try: - os.remove(seg) - except OSError: - pass - logger.info( - f"DVR recording {recording_id}: concat→MKV succeeded — " - f"{len(all_segments)} segments → {os.path.basename(final_path)} " - f"({os.path.getsize(final_path):,} bytes)" - ) - else: - logger.warning( - f"DVR recording {recording_id}: direct concat→MKV failed " - f"(rc={concat_result.returncode}), falling back to " - f"normal remux with current segment only. " - f"stderr: {(concat_result.stderr or '')[:500]}" - ) - finally: - try: - os.remove(concat_list_path) - except OSError: - pass - # Clear the pre-restart paths from custom_properties - if _rec_obj_for_concat: - _ccp = _rec_obj_for_concat.custom_properties or {} - _ccp.pop("_pre_restart_ts_paths", None) - _ccp.pop("interrupted_reason", None) - _rec_obj_for_concat.custom_properties = _ccp - _rec_obj_for_concat.save(update_fields=["custom_properties"]) - except Exception as e: - logger.warning( - f"DVR recording {recording_id}: segment concatenation error " - f"({type(e).__name__}: {e}), proceeding with current segment only." - ) - - # Remux TS to MKV container with retry for transient I/O errors - # (Skip if concat already produced the final MKV directly.) - remux_success = _concat_did_remux - existing_mkv_size = 0 - try: - if final_path and os.path.exists(final_path): - existing_mkv_size = os.path.getsize(final_path) - except OSError: - pass - for _remux_attempt in range(_dvr_remux_max_retries): - if remux_success: - break + def _get_hls_segments(m3u8_path, seg_dir): + """Return ordered segment paths from an HLS m3u8 playlist.""" + segs = [] try: - if temp_ts_path and os.path.exists(temp_ts_path): - # First attempt: Direct TS to MKV remux - result = subprocess.run([ + with open(m3u8_path) as _f: + for _line in _f: + _line = _line.strip() + if _line and not _line.startswith('#'): + sp = os.path.join(seg_dir, _line) if not os.path.isabs(_line) else _line + if os.path.exists(sp): + segs.append(sp) + except Exception as _e: + logger.warning(f"DVR recording {recording_id}: failed to parse m3u8: {_e}") + return segs + + segments = ( + _get_hls_segments(hls_m3u8, hls_dir) + if (hls_m3u8 and os.path.exists(hls_m3u8)) + else [] + ) + if not segments and hls_dir and os.path.isdir(hls_dir): + # Fallback: sort all segment files by name if m3u8 is missing or empty + try: + segments = sorted( + os.path.join(hls_dir, f) + for f in os.listdir(hls_dir) + if f.startswith("seg_") and f.endswith(".ts") + ) + except Exception: + segments = [] + + if segments: + concat_list_path = os.path.join(hls_dir, "concat.txt") + try: + with open(concat_list_path, "w") as _cl: + for seg in segments: + _cl.write(f"file '{seg}'\n") + + concat_result = subprocess.run( + [ "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", # Regenerate timestamps, ignore DTS - "-err_detect", "ignore_err", # Ignore minor stream errors - "-i", temp_ts_path, - "-map", "0", # Map all streams + "-f", "concat", "-safe", "0", + "-i", concat_list_path, "-c", "copy", - final_path - ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - # Check if FFmpeg succeeded (return code 0) and output file is valid - if result.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0: - remux_success = True - logger.info(f"Direct TS→MKV remux succeeded for {os.path.basename(final_path)}") - else: - # Direct remux failed - try fallback: TS → MP4 → MKV to fix timestamps - logger.warning(f"Direct TS→MKV remux failed (return code: {result.returncode}), trying fallback TS→MP4→MKV") - - # Clean up partial/failed MKV - try: - if os.path.exists(final_path): - os.remove(final_path) - except Exception: - pass - - # Step 1: TS → MP4 (MP4 container handles broken timestamps better) - temp_mp4_path = os.path.splitext(temp_ts_path)[0] + ".mp4" - result_mp4 = subprocess.run([ - "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", - "-err_detect", "ignore_err", - "-i", temp_ts_path, - "-map", "0", - "-c", "copy", - temp_mp4_path - ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - if result_mp4.returncode == 0 and os.path.exists(temp_mp4_path) and os.path.getsize(temp_mp4_path) > 0: - logger.info(f"TS→MP4 conversion succeeded, now converting MP4→MKV") - - # Step 2: MP4 → MKV (clean timestamps from MP4) - result_mkv = subprocess.run([ - "ffmpeg", "-y", - "-i", temp_mp4_path, - "-map", "0", - "-c", "copy", - final_path - ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - if result_mkv.returncode == 0 and os.path.exists(final_path) and os.path.getsize(final_path) > 0: - remux_success = True - logger.info(f"Fallback TS→MP4→MKV remux succeeded for {os.path.basename(final_path)}") - else: - logger.error(f"MP4→MKV conversion failed (return code: {result_mkv.returncode})") - - # Clean up temp MP4 - try: - if os.path.exists(temp_mp4_path): - os.remove(temp_mp4_path) - except Exception: - pass - else: - logger.error(f"TS→MP4 conversion failed (return code: {result_mp4.returncode})") - - # Sanity-check the remuxed file. Two checks: - # 1. If a pre-existing MKV was overwritten, reject a - # file that is drastically smaller (duplicate-task - # overwrite protection). - # 2. If the MKV is smaller than the .ts source, the - # remux likely produced a corrupt or truncated file. - if remux_success: - try: - new_size = os.path.getsize(final_path) - ts_size = os.path.getsize(temp_ts_path) if temp_ts_path and os.path.exists(temp_ts_path) else 0 - reject = False - if existing_mkv_size > 0 and new_size < existing_mkv_size * 0.5: - logger.error( - f"DVR recording {recording_id}: new MKV " - f"({new_size:,} bytes) is less than 50%% of " - f"the previous MKV ({existing_mkv_size:,} bytes) " - f"— refusing to overwrite. Keeping .ts for " - f"manual recovery." - ) - reject = True - elif ts_size > 0 and new_size < ts_size * 0.1: - logger.error( - f"DVR recording {recording_id}: remuxed MKV " - f"({new_size:,} bytes) is less than 10%% of " - f"the source TS ({ts_size:,} bytes) — likely " - f"corrupt. Keeping .ts for manual recovery." - ) - reject = True - if reject: - remux_success = False - try: - os.remove(final_path) - except OSError: - pass - except OSError: - pass - - # Clean up temp TS file only on successful remux - if remux_success: - try: - os.remove(temp_ts_path) - logger.debug(f"Cleaned up temp TS file: {temp_ts_path}") - except Exception as e: - logger.warning(f"Failed to remove temp TS file: {e}") - else: - # Keep TS file for debugging/manual recovery if remux failed - logger.warning(f"Remux failed - keeping temp TS file for recovery: {temp_ts_path}") - # Clean up any partial MKV - try: - if os.path.exists(final_path): - os.remove(final_path) - logger.debug(f"Cleaned up partial MKV file: {final_path}") - except Exception: - pass - break # Completed (success or deterministic failure) - - except (OSError, subprocess.SubprocessError) as e: - # Clean up partial output before potential retry - try: - if os.path.exists(final_path): - os.remove(final_path) - except Exception: - pass - if _remux_attempt + 1 < _dvr_remux_max_retries: - _wait = _dvr_remux_retry_interval * (2 ** _remux_attempt) - logger.warning( - f"DVR recording {recording_id}: remux failed " - f"({type(e).__name__}), retrying in {_wait}s " - f"({_remux_attempt + 1}/{_dvr_remux_max_retries})..." + final_path, + ], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + if ( + concat_result.returncode == 0 + and os.path.exists(final_path) + and os.path.getsize(final_path) > 0 + ): + remux_success = True + logger.info( + f"DVR recording {recording_id}: HLS→MKV concat succeeded — " + f"{len(segments)} segments → {os.path.basename(final_path)} " + f"({os.path.getsize(final_path):,} bytes)" ) - time.sleep(_wait) + # Update DB so *new* client requests go to /file/ (the final MKV) + # rather than the soon-to-be-removed /hls/ endpoint. Important: + # we do NOT clear _hls_dir here, active viewers still in the + # 20s heartbeat window need it to keep fetching .ts segments + # during the viewer-wait grace period below. _hls_dir is + # cleared only after the directory is actually removed. + try: + _pre_rmtree_rec = Recording.objects.filter(id=recording_id).first() + if _pre_rmtree_rec: + _pre_rmtree_cp = _pre_rmtree_rec.custom_properties or {} + _pre_rmtree_cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/" + _pre_rmtree_cp["output_file_url"] = _pre_rmtree_cp["file_url"] + _pre_rmtree_rec.custom_properties = _pre_rmtree_cp + _pre_rmtree_rec.save(update_fields=["custom_properties"]) + except Exception as _pre_e: + logger.warning(f"DVR recording {recording_id}: pre-rmtree DB update failed: {_pre_e}") + import shutil as _shutil + # Wait for active HLS viewers to finish downloading segments + # before deleting the directory. The hls endpoint refreshes + # _hls_viewer_key on every .ts request with a 20s TTL, so the + # key expiring naturally means no segment was fetched in the + # last 20 seconds (client has stopped). We loop until the key + # is gone with a 4-hour hard safety cap in case Redis gets stuck. + _viewer_wait_start = time.time() + _safety_timeout = 14400 # 4 hours absolute maximum + try: + from core.utils import RedisClient as _RC + _rv = _RC.get_client(max_retries=1, retry_interval=0) + if _rv and _rv.exists(_hls_viewer_key): + logger.info( + f"DVR recording {recording_id}: active HLS viewer detected, " + f"deferring HLS directory cleanup until client disconnects" + ) + while _rv.exists(_hls_viewer_key): + if time.time() - _viewer_wait_start > _safety_timeout: + logger.warning( + f"DVR recording {recording_id}: viewer wait safety timeout " + f"({_safety_timeout}s) reached, proceeding with HLS directory cleanup" + ) + break + time.sleep(2) + logger.info( + f"DVR recording {recording_id}: viewer wait complete after " + f"{time.time() - _viewer_wait_start:.1f}s" + ) + except Exception as _ve: + logger.debug(f"DVR recording {recording_id}: viewer wait check failed: {_ve}") + try: + _shutil.rmtree(hls_dir) + logger.debug(f"Cleaned up HLS directory: {hls_dir}") + except Exception as _e: + logger.warning(f"Failed to remove HLS directory {hls_dir}: {_e}") + # Now that the HLS dir is gone, clear _hls_dir from custom_properties + # so the hls endpoint will redirect (.m3u8) or 404 (.ts) cleanly. + try: + _post_rmtree_rec = Recording.objects.filter(id=recording_id).first() + if _post_rmtree_rec: + _post_rmtree_cp = _post_rmtree_rec.custom_properties or {} + if "_hls_dir" in _post_rmtree_cp: + _post_rmtree_cp.pop("_hls_dir", None) + _post_rmtree_rec.custom_properties = _post_rmtree_cp + _post_rmtree_rec.save(update_fields=["custom_properties"]) + except Exception as _post_e: + logger.warning(f"DVR recording {recording_id}: post-rmtree DB update failed: {_post_e}") else: - logger.warning( - f"DVR recording {recording_id}: remux failed " - f"after {_dvr_remux_max_retries} attempts: {e}. " - f"Keeping .ts for manual recovery: {temp_ts_path}" + logger.error( + f"DVR recording {recording_id}: HLS→MKV concat failed " + f"(rc={concat_result.returncode}). Keeping HLS segments for recovery. " + f"stderr: {(concat_result.stderr or '')[:500]}" ) + except Exception as _ce: + logger.error(f"DVR recording {recording_id}: concat exception: {_ce}") + finally: + try: + os.remove(concat_list_path) + except OSError: + pass + else: + logger.warning(f"DVR recording {recording_id}: no HLS segments found. Nothing to concat") # Persist final metadata to Recording (status, ended_at, and stream stats if available) try: @@ -2465,6 +2456,11 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): cp = recording_obj.custom_properties or {} cp["ended_at"] = str(datetime.now()) + # Restore file_url to the permanent MKV endpoint now that recording has ended. + # During recording it pointed to /hls/index.m3u8; clients should use /file/ hereafter. + cp["file_url"] = f"/api/channels/recordings/{recording_id}/file/" + cp["output_file_url"] = cp["file_url"] + # Final status priority: stopped > completed > interrupted. # "stopped" is set by the stop endpoint before stream teardown, so # refresh_from_db() above guarantees it is visible here. @@ -2591,7 +2587,7 @@ def recover_recordings_on_startup(): from django.utils import timezone from .models import Recording from core.utils import RedisClient - from .signals import schedule_recording_task + from .signals import schedule_recording_task, revoke_task redis = RedisClient.get_client() if redis: @@ -2637,14 +2633,12 @@ def recover_recordings_on_startup(): # Preserve the pre-restart .ts segment path so run_recording # can concatenate it with the resumed segment later. - old_ts = cp.get("_temp_file_path") - if old_ts and os.path.exists(old_ts) and os.path.getsize(old_ts) > 0: - prior_segments = cp.get("_pre_restart_ts_paths", []) - prior_segments.append(old_ts) - cp["_pre_restart_ts_paths"] = prior_segments + hls_dir_path = cp.get("_hls_dir") + if hls_dir_path and os.path.isdir(hls_dir_path): + existing_segs = [f for f in os.listdir(hls_dir_path) if f.endswith(".ts")] logger.info( f"recover_recordings_on_startup: recording {rec.id} — " - f"preserving pre-restart TS segment: {old_ts}" + f"HLS dir has {len(existing_segs)} existing segment(s), will resume" ) rec.custom_properties = cp @@ -2687,38 +2681,86 @@ def recover_recordings_on_startup(): for rec in expired: try: cp = rec.custom_properties or {} - ts_path = cp.get("_temp_file_path") + hls_dir_path = cp.get("_hls_dir") mkv_path = cp.get("file_path") - if ts_path and os.path.exists(ts_path) and os.path.getsize(ts_path) > 0 and mkv_path: - logger.info( - f"recover_recordings_on_startup: recording {rec.id} expired " - f"during downtime — remuxing partial TS ({os.path.getsize(ts_path):,} bytes)" - ) - os.makedirs(os.path.dirname(mkv_path), exist_ok=True) - result = subprocess.run( - [ - "ffmpeg", "-y", - "-fflags", "+genpts+igndts+discardcorrupt", - "-err_detect", "ignore_err", - "-i", ts_path, "-map", "0", "-c", "copy", mkv_path, - ], - stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - ) - if result.returncode == 0 and os.path.exists(mkv_path) and os.path.getsize(mkv_path) > 0: - cp["status"] = "interrupted" - cp["interrupted_reason"] = "server_restarted_after_end" - cp["remux_success"] = True + if hls_dir_path and os.path.isdir(hls_dir_path) and mkv_path: + import shutil as _shutil + # Parse m3u8 for ordered segment list; fall back to sorted filenames + _m3u8 = os.path.join(hls_dir_path, "index.m3u8") + _segs = [] + if os.path.exists(_m3u8): try: - os.remove(ts_path) - except OSError: + with open(_m3u8) as _f: + for _l in _f: + _l = _l.strip() + if _l and not _l.startswith('#'): + _sp = os.path.join(hls_dir_path, _l) if not os.path.isabs(_l) else _l + if os.path.exists(_sp): + _segs.append(_sp) + except Exception: pass - logger.info(f"recover_recordings_on_startup: recording {rec.id} remuxed successfully") + if not _segs: + _segs = sorted( + os.path.join(hls_dir_path, f) + for f in os.listdir(hls_dir_path) + if f.startswith("seg_") and f.endswith(".ts") + ) + + if _segs: + logger.info( + f"recover_recordings_on_startup: recording {rec.id} expired " + f"during downtime, concat {len(_segs)} HLS segment(s) \u2192 MKV" + ) + os.makedirs(os.path.dirname(mkv_path), exist_ok=True) + _concat_txt = os.path.join(hls_dir_path, "concat.txt") + try: + with open(_concat_txt, "w") as _cl: + for _s in _segs: + _cl.write(f"file '{_s}'\n") + _res = subprocess.run( + [ + "ffmpeg", "-y", + "-f", "concat", "-safe", "0", + "-i", _concat_txt, + "-c", "copy", mkv_path, + ], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + if _res.returncode == 0 and os.path.exists(mkv_path) and os.path.getsize(mkv_path) > 0: + cp["status"] = "interrupted" + cp["interrupted_reason"] = "server_restarted_after_end" + cp["remux_success"] = True + try: + _shutil.rmtree(hls_dir_path) + except OSError: + pass + logger.info( + f"recover_recordings_on_startup: recording {rec.id} HLS\u2192MKV concat succeeded" + ) + else: + cp["status"] = "interrupted" + cp["interrupted_reason"] = "server_restarted_after_end" + cp["remux_success"] = False + logger.warning( + f"recover_recordings_on_startup: recording {rec.id} concat failed, keeping HLS dir" + ) + except Exception as _ce: + cp["status"] = "interrupted" + cp["interrupted_reason"] = "server_restarted_after_end" + cp["remux_success"] = False + logger.warning( + f"recover_recordings_on_startup: recording {rec.id} concat error: {_ce}" + ) + finally: + try: + os.remove(_concat_txt) + except OSError: + pass else: cp["status"] = "interrupted" cp["interrupted_reason"] = "server_restarted_after_end" cp["remux_success"] = False - logger.warning(f"recover_recordings_on_startup: recording {rec.id} remux failed, keeping .ts") else: cp["status"] = "interrupted" cp["interrupted_reason"] = "server_restarted_after_end" diff --git a/apps/channels/tests/test_dvr_port_resolution.py b/apps/channels/tests/test_dvr_port_resolution.py index c7373959..61f74454 100644 --- a/apps/channels/tests/test_dvr_port_resolution.py +++ b/apps/channels/tests/test_dvr_port_resolution.py @@ -2,58 +2,68 @@ import os from django.test import SimpleTestCase from unittest.mock import patch -from apps.channels.tasks import build_dvr_candidates +from apps.channels.tasks import get_dvr_stream_base_url -class DVRPortResolutionTests(SimpleTestCase): +class DVRStreamBaseURLTests(SimpleTestCase): """ - Tests that DVR recording candidate URLs respect the DISPATCHARR_PORT - environment variable instead of hardcoding port 9191. + Tests that get_dvr_stream_base_url() returns the correct single URL + for each deployment mode. """ - @patch.dict(os.environ, {'REDIS_HOST': 'redis'}, clear=True) - def test_default_port_uses_9191(self): - """Without DISPATCHARR_PORT set, candidates default to 9191.""" - candidates = build_dvr_candidates() - self.assertIn('http://web:9191', candidates) - self.assertIn('http://localhost:9191', candidates) + @patch.dict(os.environ, {}, clear=True) + def test_aio_default_uses_localhost_5656(self): + """AIO mode (default) reaches uwsgi directly on loopback port 5656.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://127.0.0.1:5656') - @patch.dict(os.environ, {'DISPATCHARR_PORT': '8080', 'REDIS_HOST': 'redis'}, clear=True) - def test_custom_port_reflected_in_candidates(self): - """DISPATCHARR_PORT=8080 replaces all hardcoded 9191 references.""" - candidates = build_dvr_candidates() - self.assertIn('http://web:8080', candidates) - self.assertIn('http://localhost:8080', candidates) - self.assertNotIn('http://web:9191', candidates) - self.assertNotIn('http://localhost:9191', candidates) + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'aio'}, clear=True) + def test_aio_explicit_uses_localhost_5656(self): + """Explicit DISPATCHARR_ENV=aio also uses loopback port 5656.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://127.0.0.1:5656') + + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'dev'}, clear=True) + def test_dev_mode_uses_localhost_5656(self): + """Dev mode shares the container with uwsgi — uses loopback port 5656.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://127.0.0.1:5656') + + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'modular', 'DISPATCHARR_PORT': '9191'}, clear=True) + def test_modular_uses_web_service_name(self): + """Modular mode uses the 'web' Docker service name by default.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://web:9191') + + @patch.dict(os.environ, {'DISPATCHARR_ENV': 'modular', 'DISPATCHARR_PORT': '8080'}, clear=True) + def test_modular_custom_port(self): + """Modular mode respects DISPATCHARR_PORT.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://web:8080') @patch.dict(os.environ, { - 'DISPATCHARR_PORT': '7777', - 'DISPATCHARR_ENV': 'dev', - 'REDIS_HOST': 'redis', + 'DISPATCHARR_ENV': 'modular', + 'DISPATCHARR_PORT': '9191', + 'DISPATCHARR_WEB_HOST': 'dispatcharr_web', }, clear=True) - def test_dev_mode_includes_5656_and_custom_port(self): - """Dev mode includes both uwsgi internal port (5656) and custom port.""" - candidates = build_dvr_candidates() - self.assertIn('http://127.0.0.1:5656', candidates) - self.assertIn('http://127.0.0.1:7777', candidates) + def test_modular_custom_web_host(self): + """DISPATCHARR_WEB_HOST overrides the default 'web' service name.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://dispatcharr_web:9191') @patch.dict(os.environ, { 'DISPATCHARR_INTERNAL_TS_BASE_URL': 'http://custom:1234', - 'REDIS_HOST': 'redis', + 'DISPATCHARR_ENV': 'modular', }, clear=True) - def test_explicit_override_is_first(self): - """DISPATCHARR_INTERNAL_TS_BASE_URL should be the first candidate.""" - candidates = build_dvr_candidates() - self.assertEqual(candidates[0], 'http://custom:1234') + def test_explicit_override_always_wins(self): + """DISPATCHARR_INTERNAL_TS_BASE_URL takes priority over all other settings.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://custom:1234') @patch.dict(os.environ, { - 'DISPATCHARR_PORT': '3000', - 'DISPATCHARR_INTERNAL_API_BASE': 'http://myhost:4000', - 'REDIS_HOST': 'redis', + 'DISPATCHARR_INTERNAL_TS_BASE_URL': 'http://custom:1234/', }, clear=True) - def test_internal_api_base_overrides_web_fallback(self): - """DISPATCHARR_INTERNAL_API_BASE replaces the http://web:{port} default.""" - candidates = build_dvr_candidates() - self.assertIn('http://myhost:4000', candidates) - self.assertNotIn('http://web:3000', candidates) + def test_explicit_override_strips_trailing_slash(self): + """Trailing slash is stripped from DISPATCHARR_INTERNAL_TS_BASE_URL.""" + url = get_dvr_stream_base_url() + self.assertEqual(url, 'http://custom:1234') diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 547ac249..58b6e4f5 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -118,8 +118,12 @@ services: - DISPATCHARR_ENV=modular # Internal Service Communication - # Must match the web service port for DVR recording and internal API calls + # Celery uses these to reach the web container for DVR recording. + # DISPATCHARR_PORT must match the port exposed by the web service. + # DISPATCHARR_WEB_HOST defaults to "web" (the service name above). + # Only set DISPATCHARR_WEB_HOST if you rename the web service in this file. - DISPATCHARR_PORT=9191 + #- DISPATCHARR_WEB_HOST=web # PostgreSQL — must match web service settings - POSTGRES_HOST=db