Bug Fix/Enhancement:

- EPG refresh tasks (`refresh_epg_data`) were being killed mid-transaction on large EPG sources. The `soft_time_limit=1700s` introduced in v0.21.0 raised `SoftTimeLimitExceeded`, a subclass of `Exception`, which was swallowed by the existing `except Exception` handler in `parse_programs_for_source`, leaving the database in a partial state with no logged error. The underlying cause is that the HTML entity preprocessing step added in v0.21.0 adds 2–10+ minutes on multi-gigabyte EPG files, pushing total task duration past the 1700s limit for large sources. `soft_time_limit` has been removed from `refresh_epg_data` and `time_limit` raised to 14400s (4 hours) as a true last-resort ceiling; the existing `TaskLockRenewer` daemon thread continues to renew the Redis lock every 120s for legitimately long-running tasks.
- HTML entity preprocessing (`_resolve_html_entities`) now performs a fast binary pre-scan before any file read or write. The pre-scan reads the file in 4 MB chunks without text decoding and exits immediately on the first non-XML named entity found. For EPG sources that contain no HTML entities — the common case — the full line-by-line read+rewrite pass (which takes 2–10+ minutes on a 2.5 GB file) is skipped entirely.

CHANGELOG.md

@@ -66,6 +66,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - **Program ID instability**: `parse_programs_for_source()` deletes and recreates all `ProgramData` rows with new auto-increment IDs on every EPG refresh. The dedup set used these IDs, so it never matched after a refresh. Deduplication now uses a stable `(tvg_id, start_time, end_time)` composite key sourced from `Recording.custom_properties.program`.
   - **Secondary guard using wrong times**: The DB guard compared unadjusted program times against offset-adjusted `Recording.start_time`/`end_time`, so it never matched when any DVR pre/post offset was configured. It now queries `custom_properties__program__start_time/end_time` (the original, unadjusted program times stored at recording creation).
   - **No concurrency guard**: Each EPG source refresh fired `evaluate_series_rules.delay()` independently. Concurrent tasks loaded the dedup set before others committed, allowing races. Evaluation is now serialized with `acquire_task_lock` (reusing the existing EPG task pattern). Gracefully degrades if Redis is unavailable — the primary and secondary dedup guards still protect.
+- EPG refresh tasks (`refresh_epg_data`) were being killed mid-transaction on large EPG sources. The `soft_time_limit=1700s` introduced in v0.21.0 raised `SoftTimeLimitExceeded`, a subclass of `Exception`, which was swallowed by the existing `except Exception` handler in `parse_programs_for_source`, leaving the database in a partial state with no logged error. The underlying cause is that the HTML entity preprocessing step added in v0.21.0 adds 2–10+ minutes on multi-gigabyte EPG files, pushing total task duration past the 1700s limit for large sources. `soft_time_limit` has been removed from `refresh_epg_data` and `time_limit` raised to 14400s (4 hours) as a true last-resort ceiling; the existing `TaskLockRenewer` daemon thread continues to renew the Redis lock every 120s for legitimately long-running tasks.
+- HTML entity preprocessing (`_resolve_html_entities`) now performs a fast binary pre-scan before any file read or write. The pre-scan reads the file in 4 MB chunks without text decoding and exits immediately on the first non-XML named entity found. For EPG sources that contain no HTML entities — the common case — the full line-by-line read+rewrite pass (which takes 2–10+ minutes on a 2.5 GB file) is skipped entirely.
 
 ## [0.21.1] - 2026-03-18
 

apps/epg/tasks.py

@@ -18,7 +18,7 @@ import zipfile
 
 from celery import shared_task
 from django.conf import settings
-from django.db import transaction
+from django.db import connection, transaction
 from django.utils import timezone
 from apps.channels.models import Channel
 from core.models import UserAgent, CoreSettings
@@ -36,8 +36,11 @@ logger = logging.getLogger(__name__)
 # é are silently dropped in recovery mode.  This substitution
 # converts them to Unicode before the file reaches the parser.
 _XML_ENTITIES = frozenset({'amp', 'lt', 'gt', 'quot', 'apos'})
+_XML_ENTITIES_BYTES = frozenset(n.encode() for n in _XML_ENTITIES)
 _XML_SPECIAL_CHARS = frozenset('&<>\'"')
 _NAMED_ENTITY_RE = re.compile(r'&([a-zA-Z][a-zA-Z0-9]*);')
+# Binary version used by the fast pre-scan (no text decoding needed)
+_BINARY_ENTITY_RE = re.compile(rb'&([a-zA-Z][a-zA-Z0-9]*);')
 
 
 def _replace_html_entity(match):
@@ -79,6 +82,32 @@ def _detect_xml_encoding(header):
     return 'utf-8'
 
 
+def _file_needs_entity_resolution(file_path):
+    """Binary pre-scan: return True only if file contains non-XML named HTML entities.
+
+    Reads in 4 MB chunks with a small overlap to catch entities that straddle
+    a chunk boundary.  Avoids all text-decode overhead — this is effectively a
+    raw memory scan and exits as soon as the first HTML entity is found.
+    """
+    CHUNK = 4 * 1024 * 1024  # 4 MB
+    OVERLAP = 32              # longer than any valid entity name + surrounding &;
+    try:
+        with open(file_path, 'rb') as f:
+            prev = b''
+            while True:
+                chunk = f.read(CHUNK)
+                if not chunk:
+                    return False
+                data = prev + chunk
+                for m in _BINARY_ENTITY_RE.finditer(data):
+                    if m.group(1).lower() not in _XML_ENTITIES_BYTES:
+                        return True
+                prev = data[-OVERLAP:]
+    except OSError:
+        return False
+    return False
+
+
 def _resolve_html_entities(file_path):
     """Replace HTML named entities in an XMLTV file with Unicode characters.
 
@@ -89,6 +118,14 @@ def _resolve_html_entities(file_path):
     This function is strictly additive — on any error the original file is
     left untouched so lxml can handle it as it always did.
     """
+    # Fast binary pre-scan: if the file contains no non-XML named entities
+    # there is nothing to do and we can skip the full read+write entirely.
+    # For large files (e.g. 2.5 GB) this avoids significant I/O time when
+    # the EPG source is already clean (the common case).
+    if not _file_needs_entity_resolution(file_path):
+        logger.debug(f"No HTML entities found in {file_path}, skipping entity resolution")
+        return
+    logger.info(f"HTML entities detected in {file_path}, resolving...")
     # Read the first 200 bytes to detect encoding from the XML declaration
     with open(file_path, 'rb') as f:
         header = f.read(200)
@@ -245,7 +282,7 @@ def refresh_all_epg_data():
     return "EPG data refreshed."
 
 
-@shared_task(time_limit=1800, soft_time_limit=1700)
+@shared_task(time_limit=14400)
 def refresh_epg_data(source_id):
     if not acquire_task_lock('refresh_epg_data', source_id):
         logger.debug(f"EPG refresh for {source_id} already running")
@@ -499,42 +536,41 @@ def fetch_xmltv(source):
 
             # Download to temporary file
             with open(temp_download_path, 'wb') as f:
-                for chunk in response.iter_content(chunk_size=16384):  # Increased chunk size for better performance
-                    if chunk:
-                        f.write(chunk)
+                for chunk in response.iter_content(chunk_size=16384):
+                    f.write(chunk)
 
-                        downloaded += len(chunk)
-                        elapsed_time = time.time() - start_time
+                    downloaded += len(chunk)
+                    elapsed_time = time.time() - start_time
 
-                        # Calculate download speed in KB/s
-                        speed = downloaded / elapsed_time / 1024 if elapsed_time > 0 else 0
+                    # Calculate download speed in KB/s
+                    speed = downloaded / elapsed_time / 1024 if elapsed_time > 0 else 0
 
-                        # Calculate progress percentage
-                        if total_size and total_size > 0:
-                            progress = min(100, int((downloaded / total_size) * 100))
-                        else:
-                            # If no content length header, estimate progress
-                            progress = min(95, int((downloaded / (10 * 1024 * 1024)) * 100))  # Assume 10MB if unknown
+                    # Calculate progress percentage
+                    if total_size and total_size > 0:
+                        progress = min(100, int((downloaded / total_size) * 100))
+                    else:
+                        # If no content length header, estimate progress
+                        progress = min(95, int((downloaded / (10 * 1024 * 1024)) * 100))  # Assume 10MB if unknown
 
-                        # Time remaining (in seconds)
-                        time_remaining = (total_size - downloaded) / (speed * 1024) if speed > 0 and total_size > 0 else 0
+                    # Time remaining (in seconds)
+                    time_remaining = (total_size - downloaded) / (speed * 1024) if speed > 0 and total_size > 0 else 0
 
-                        # Only send updates at specified intervals to avoid flooding
-                        current_time = time.time()
-                        if current_time - last_update_time >= update_interval and progress > 0:
-                            last_update_time = current_time
-                            send_epg_update(
-                                source.id,
-                                "downloading",
-                                progress,
-                                speed=round(speed, 2),
-                                elapsed_time=round(elapsed_time, 1),
-                                time_remaining=round(time_remaining, 1),
-                                downloaded=f"{downloaded / (1024 * 1024):.2f} MB"
-                            )
+                    # Only send updates at specified intervals to avoid flooding
+                    current_time = time.time()
+                    if current_time - last_update_time >= update_interval and progress > 0:
+                        last_update_time = current_time
+                        send_epg_update(
+                            source.id,
+                            "downloading",
+                            progress,
+                            speed=round(speed, 2),
+                            elapsed_time=round(elapsed_time, 1),
+                            time_remaining=round(time_remaining, 1),
+                            downloaded=f"{downloaded / (1024 * 1024):.2f} MB"
+                        )
 
-                        # Explicitly delete the chunk to free memory immediately
-                        del chunk
+                    # Explicitly delete the chunk to free memory immediately
+                    del chunk
 
             # Send completion notification
             send_epg_update(source.id, "downloading", 100)
@@ -1763,6 +1799,10 @@ def parse_programs_for_source(epg_source, tvg_id=None):
         batch_size = 1000
         try:
             with transaction.atomic():
+                # Kill any individual statement that hangs longer than 10 minutes.
+                # SET LOCAL automatically resets when this transaction ends (commit or rollback).
+                with connection.cursor() as cursor:
+                    cursor.execute("SET LOCAL statement_timeout = '10min'")
                 # Delete existing programs for mapped EPGs
                 deleted_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()[0]
                 logger.debug(f"Deleted {deleted_count} existing programs")