diff --git a/apps/epg/api_views.py b/apps/epg/api_views.py index 9e9d26f2..4fa6b478 100644 --- a/apps/epg/api_views.py +++ b/apps/epg/api_views.py @@ -18,7 +18,7 @@ from .serializers import ( EPGDataSerializer, ProgramSearchResultSerializer, ) -from .tasks import refresh_epg_data +from .tasks import refresh_epg_data, find_current_program_for_tvg_id from .query_utils import parse_text_query from apps.accounts.permissions import ( Authenticated, @@ -772,12 +772,14 @@ class CurrentProgramsAPIView(APIView): status=status.HTTP_400_BAD_REQUEST ) - from apps.epg.models import EPGData + # Limit to 50 IDs per request + epg_data_ids = epg_data_ids[:50] - epg_data_entries = EPGData.objects.filter(id__in=epg_data_ids) + epg_data_entries = EPGData.objects.select_related('epg_source').filter(id__in=epg_data_ids) current_programs = [] for epg_data in epg_data_entries: + # Check DB first (already-parsed channels) program = ProgramData.objects.filter( epg=epg_data, start_time__lte=now, end_time__gt=now ).first() @@ -786,20 +788,23 @@ class CurrentProgramsAPIView(APIView): program_data = ProgramDataSerializer(program).data program_data['epg_data_id'] = epg_data.id current_programs.append(program_data) - else: - # Check if ANY programs exist (vs none parsed yet) - has_any_programs = ProgramData.objects.filter(epg=epg_data).exists() - if not has_any_programs and epg_data.epg_source.source_type != 'dummy': - from core.utils import RedisClient - redis_client = RedisClient.get_client() - lock_id = f"task_lock_parse_epg_programs_{epg_data.id}" - if not redis_client.exists(lock_id): - from apps.epg.tasks import parse_programs_for_tvg_id - parse_programs_for_tvg_id.delay(epg_data.id, force=True) - current_programs.append({ - "epg_data_id": epg_data.id, - "parsing": True, - }) + continue + + # Skip dummy sources + if epg_data.epg_source and epg_data.epg_source.source_type == 'dummy': + continue + + # Fall back to byte-offset index lookup + result = find_current_program_for_tvg_id(epg_data.id) + + if result == "timeout": + current_programs.append({ + "epg_data_id": epg_data.id, + "parsing": True, + }) + elif result is not None: + result['epg_data_id'] = epg_data.id + current_programs.append(result) return Response(current_programs, status=status.HTTP_200_OK) diff --git a/apps/epg/apps.py b/apps/epg/apps.py index a6e7a8ed..2ddb7d18 100644 --- a/apps/epg/apps.py +++ b/apps/epg/apps.py @@ -1,5 +1,6 @@ from django.apps import AppConfig + class EpgConfig(AppConfig): default_auto_field = 'django.db.models.BigAutoField' name = 'apps.epg' diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index be6d5bfc..05291d4a 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -4,6 +4,7 @@ import logging import gzip import html.entities import os +import re import uuid import requests import time # Add import for tracking download progress @@ -313,6 +314,9 @@ def refresh_epg_data(source_id): gc.collect() return + # Build byte-offset index for preview lookups (runs before full program parse) + build_programme_index(source.id) + parse_programs_for_source(source) elif source.source_type == 'schedules_direct': @@ -2339,3 +2343,340 @@ def generate_dummy_epg(source): logger.warning(f"generate_dummy_epg() called for {source.name} but this function is deprecated. " f"Dummy EPG programs are now generated on-demand.") return True + + +# EPG program byte-offset index for channel preview lookups + + +def _resolve_source_file(epg_source): + """Resolve the XML file path for an EPG source.""" + file_path = epg_source.extracted_file_path or epg_source.file_path + if not file_path: + file_path = epg_source.get_cache_file() + return file_path + + +_CHANNEL_ATTR_RE = re.compile(rb"""channel=(?:"([^"]+)"|'([^']+)')""") +_PROGRAMME_TAG = b'/' +_MAX_START_TAG = 4096 # generous upper bound for a start tag with namespaces/extra attrs + + +def _find_programme_tag(buf, start): + """ + Find the next ' + follow = idx + _PROGRAMME_TAG_LEN + if follow >= len(buf): + return idx, -1 # need more data + if buf[follow: follow + 1] not in _TAG_FOLLOW: + pos = follow # false match (e.g. ' that closes the opening tag (scan up to _MAX_START_TAG bytes) + tag_end = buf.find(b'>', follow, idx + _MAX_START_TAG) + if tag_end == -1: + if len(buf) >= idx + _MAX_START_TAG: + logger.warning( + f'[_find_programme_tag] start tag exceeds {_MAX_START_TAG} bytes at offset {idx}, skipping' + ) + return -1, -1 + return idx, -1 # need more data + return idx, tag_end + + +def _programme_to_dict(elem, start_time, end_time): + """Convert a lxml element to a serializable dict.""" + title_el = elem.find('title') + desc_el = elem.find('desc') + sub_el = elem.find('sub-title') + return { + 'title': title_el.text if title_el is not None and title_el.text else '', + 'description': desc_el.text if desc_el is not None and desc_el.text else '', + 'sub_title': sub_el.text if sub_el is not None and sub_el.text else '', + 'start_time': start_time.isoformat(), + 'end_time': end_time.isoformat(), + } + + +def build_programme_index(source_id): + """ + Scan the XML file with raw binary I/O to build a {tvg_id: [byte_offset, ...]} map. + Stores the result in Redis as JSON. Most XMLTV files group programmes by channel, + but some split a channel across multiple non-contiguous blocks, so we record the + start offset of every block. + """ + from core.utils import RedisClient + + try: + source = EPGSource.objects.get(id=source_id) + except EPGSource.DoesNotExist: + logger.error(f'[build_programme_index] EPGSource {source_id} not found') + return + + file_path = _resolve_source_file(source) + if not file_path or not os.path.exists(file_path): + logger.warning( + f'[build_programme_index] File not found for source {source_id}: {file_path}' + ) + return + + logger.info( + f'[build_programme_index] Building byte-offset index for source {source_id} from {file_path}' + ) + start = time.monotonic() + index = {} + prev_channel = None + + CHUNK = 8 * 1024 * 1024 # 8MB + + with open(file_path, 'rb') as f: + buf = bytearray() + buf_offset = 0 # absolute file offset of buf[0] + + while True: + chunk = f.read(CHUNK) + if not chunk and not buf: + break + buf.extend(chunk) + search_from = 0 + + while True: + idx, tag_end = _find_programme_tag(buf, search_from) + if idx == -1: + break + if tag_end == -1 and chunk: + break # incomplete tag at buffer edge, need more data + + abs_pos = buf_offset + idx + m = _CHANNEL_ATTR_RE.search( + buf, idx, tag_end + 1 if tag_end != -1 else idx + _MAX_START_TAG + ) + if m: + channel_id = (m.group(1) or m.group(2)).decode( + 'utf-8', errors='replace' + ) + if channel_id not in index: + index[channel_id] = [abs_pos] + elif channel_id != prev_channel: + index[channel_id].append(abs_pos) + prev_channel = channel_id + + search_from = ( + (tag_end + 1) if tag_end != -1 else (idx + _PROGRAMME_TAG_LEN) + ) + + if not chunk: + break + + # Keep unprocessed tail for next iteration + keep_from = ( + max(search_from, len(buf) - _MAX_START_TAG) if chunk else len(buf) + ) + del buf[:keep_from] + buf_offset += keep_from + + elapsed = time.monotonic() - start + logger.info( + f'[build_programme_index] Indexed {len(index)} channels in {elapsed:.1f}s for source {source_id}' + ) + + redis_client = RedisClient.get_client() + redis_key = f'epg_programme_index_{source_id}' + redis_client.set(redis_key, json.dumps(index)) + + +@shared_task +def build_programme_index_task(source_id): + """Celery wrapper so build_programme_index can be dispatched async.""" + build_programme_index(source_id) + + +def find_current_program_for_tvg_id(epg_id): + """ + Look up the currently-airing program for an EPGData id using the + byte-offset index. Falls back to sequential scan if no index exists. + + Returns dict, None, or "timeout". + """ + from core.utils import RedisClient + + try: + epg = EPGData.objects.select_related('epg_source').get(id=epg_id) + except EPGData.DoesNotExist: + return None + + source = epg.epg_source + if not source or source.source_type == 'dummy': + return None + + tvg_id = epg.tvg_id + if not tvg_id: + return None + + file_path = _resolve_source_file(source) + if not file_path or not os.path.exists(file_path): + return None + + redis_client = RedisClient.get_client() + redis_key = f'epg_programme_index_{source.id}' + index_raw = redis_client.get(redis_key) + + now = timezone.now() + + if index_raw: + # Indexed lookup + try: + index = json.loads(index_raw) + except (json.JSONDecodeError, TypeError): + index = None + + if index and tvg_id in index: + offsets = index[tvg_id] + return _read_programs_at_offsets(file_path, tvg_id, offsets, now) + # tvg_id not in index - channel has no programmes in the file + return None + + # No index yet, fall back to sequential scan + result = _sequential_scan_for_tvg_id(file_path, tvg_id, now, timeout_sec=10) + if result == 'timeout': + # Trigger background index build (Redis lock prevents duplicates) + lock_key = f'building_programme_index_{source.id}' + if redis_client.set(lock_key, '1', nx=True, ex=300): + build_programme_index_task.delay(source.id) + return result + + +def _read_programs_at_offsets(file_path, tvg_id, offsets, now): + """ + Seek to each offset, extract elements for *tvg_id*, return the + first one currently airing. Chunk-based so it works on minified XML. + """ + PROG_CLOSE = b'' + CLOSE_LEN = len(PROG_CLOSE) + READ_SIZE = 2 * 1024 * 1024 # 2MB per read + + with open(file_path, 'rb') as f: + for offset in offsets: + f.seek(offset) + buf = bytearray() + done = False + + while not done: + chunk = f.read(READ_SIZE) + if not chunk and not buf: + break + buf.extend(chunk) + search_from = 0 + + while True: + tag_start, tag_end = _find_programme_tag(buf, search_from) + if tag_start == -1: + break + if tag_end == -1 and chunk: + break # incomplete tag, need more data + + # Check channel before searching for close tag + m = _CHANNEL_ATTR_RE.search( + buf, + tag_start, + tag_end + 1 if tag_end != -1 else tag_start + _MAX_START_TAG, + ) + if not m: + search_from = ( + (tag_end + 1) + if tag_end != -1 + else (tag_start + _PROGRAMME_TAG_LEN) + ) + continue + + ch = (m.group(1) or m.group(2)).decode('utf-8', errors='replace') + if ch != tvg_id: + done = True # different channel, end of block + break + + # Find the closing tag + close_pos = buf.find( + PROG_CLOSE, tag_end + 1 if tag_end != -1 else m.end() + ) + if close_pos == -1: + if not chunk: + done = True # EOF with no close tag + break # need more data + close_end = close_pos + CLOSE_LEN + + element_bytes = bytes(buf[tag_start:close_end]) + search_from = close_end + + try: + prog = etree.fromstring(element_bytes) + except etree.XMLSyntaxError: + continue + + start_str = prog.get('start') + stop_str = prog.get('stop') + if not start_str or not stop_str: + continue + start_time = parse_xmltv_time(start_str) + end_time = parse_xmltv_time(stop_str) + if start_time is None or end_time is None: + continue + if start_time <= now < end_time: + return _programme_to_dict(prog, start_time, end_time) + + # Trim processed bytes + if search_from > 0: + del buf[:search_from] + search_from = 0 + + if not chunk: + break + + return None + + +def _sequential_scan_for_tvg_id(file_path, tvg_id, now, timeout_sec=10): + """Stream through the file with iterparse for *tvg_id*. Returns dict, None, or "timeout".""" + deadline = time.monotonic() + timeout_sec + + try: + with open(file_path, 'rb') as f: + context = etree.iterparse(f, events=('end',), tag='programme') + for _, elem in context: + if time.monotonic() > deadline: + return 'timeout' + + if elem.get('channel') != tvg_id: + elem.clear() + continue + + start_str = elem.get('start') + stop_str = elem.get('stop') + if not start_str or not stop_str: + elem.clear() + continue + + start_time = parse_xmltv_time(start_str) + end_time = parse_xmltv_time(stop_str) + if start_time is None or end_time is None: + elem.clear() + continue + + if start_time <= now < end_time: + result = _programme_to_dict(elem, start_time, end_time) + elem.clear() + return result + + elem.clear() + except Exception as e: + logger.error(f'[_sequential_scan_for_tvg_id] Error scanning for {tvg_id}: {e}') + return None + + return None diff --git a/core/tasks.py b/core/tasks.py index 3d3c059b..2009d1e5 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -388,9 +388,39 @@ def scan_and_process_files(): } ) + # Rebuild EPG programme indices on first run (Redis is ephemeral) + if not _first_scan_completed: + _rebuild_programme_indices() + # Mark that the first scan is complete _first_scan_completed = True +def _rebuild_programme_indices(): + """Queue index builds for active EPG sources that are missing their Redis index.""" + try: + from apps.epg.tasks import build_programme_index_task + redis_client = RedisClient.get_client() + + sources = EPGSource.objects.filter( + is_active=True, + ).exclude(source_type='dummy') + + count = 0 + for source in sources: + redis_key = f"epg_programme_index_{source.id}" + if redis_client.exists(redis_key): + continue + lock_key = f"building_programme_index_{source.id}" + if redis_client.set(lock_key, "1", nx=True, ex=300): + build_programme_index_task.delay(source.id) + count += 1 + + if count: + logger.info(f"Queued programme index rebuild for {count} EPG source(s)") + except Exception as e: + logger.warning(f"Failed to queue programme index rebuilds: {e}") + + def fetch_channel_stats(): redis_client = RedisClient.get_client() diff --git a/dispatcharr/urls.py b/dispatcharr/urls.py index 9123f3da..71e2d3c6 100644 --- a/dispatcharr/urls.py +++ b/dispatcharr/urls.py @@ -3,6 +3,7 @@ from django.urls import path, include, re_path from django.conf import settings from django.conf.urls.static import static from django.views.generic import TemplateView, RedirectView +from django.views.static import serve from .routing import websocket_urlpatterns from apps.output.views import xc_player_api, xc_panel_api, xc_get, xc_xmltv from apps.proxy.live_proxy.views import stream_xc @@ -57,6 +58,18 @@ urlpatterns = [ path("admin/", admin.site.urls), # VOD proxy is now handled by the main proxy URLs above + # Frontend static assets from Vite build output + re_path( + r"^assets/(?P.*)$", + serve, + {"document_root": settings.BASE_DIR / "frontend" / "dist" / "assets"}, + ), + re_path( + r"^(?Plogo\.png)$", + serve, + {"document_root": settings.BASE_DIR / "frontend" / "dist"}, + ), + # Catch-all routes should always be last path("", TemplateView.as_view(template_name="index.html")), # React entry point path("", TemplateView.as_view(template_name="index.html")), diff --git a/frontend/src/components/forms/Channel.jsx b/frontend/src/components/forms/Channel.jsx index f3c63b1d..0a955f0d 100644 --- a/frontend/src/components/forms/Channel.jsx +++ b/frontend/src/components/forms/Channel.jsx @@ -474,31 +474,49 @@ const ChannelForm = ({ channel: channelProp = null, isOpen, onClose }) => { setIsLoadingProgram(true); setHasFetchedProgram(false); - const fetchWithRetry = async (retriesLeft = 3) => { - try { - const program = await API.getCurrentProgramForEpg(epgDataId); - if (cancelled) return; + const fetchWithRetry = async () => { + const maxRetries = 5; + const deadlineMs = 3 * 60 * 1000; // 3 minutes + const startTime = Date.now(); + let delay = 3000; // 3s initial - if (program && program.parsing && retriesLeft > 0) { - // Programs are being parsed, retry after delay - await new Promise((resolve) => setTimeout(resolve, 10000)); + for (let attempt = 0; attempt <= maxRetries; attempt++) { + if (cancelled || Date.now() - startTime > deadlineMs) break; + + try { + const program = await API.getCurrentProgramForEpg(epgDataId); if (cancelled) return; - return fetchWithRetry(retriesLeft - 1); - } - if (!cancelled) { - setCurrentProgram(program && !program.parsing ? program : null); - setIsLoadingProgram(false); - setHasFetchedProgram(true); - } - } catch (error) { - if (!cancelled) { - console.error('Failed to fetch current program:', error); - setCurrentProgram(null); - setIsLoadingProgram(false); - setHasFetchedProgram(true); + if (program && program.parsing && attempt < maxRetries) { + // Index still building, retry with backoff + await new Promise((resolve) => setTimeout(resolve, delay)); + delay = Math.min(delay * 1.5, 15000); // 1.5x multiplier, 15s cap + continue; + } + + if (!cancelled) { + setCurrentProgram(program && !program.parsing ? program : null); + setIsLoadingProgram(false); + setHasFetchedProgram(true); + } + return; + } catch (error) { + if (!cancelled) { + console.error('Failed to fetch current program:', error); + } + if (attempt < maxRetries) { + await new Promise((resolve) => setTimeout(resolve, delay)); + delay = Math.min(delay * 1.5, 15000); + } } } + + // Exhausted retries or deadline + if (!cancelled) { + setCurrentProgram(null); + setIsLoadingProgram(false); + setHasFetchedProgram(true); + } }; fetchWithRetry();