From d4ecf45bc29ebcf7ebbbb8214a35773d0f32e8c2 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 29 Mar 2026 19:51:57 -0500 Subject: [PATCH] =?UTF-8?q?Bug=20Fix/Enhancement:=20-=20EPG=20refresh=20ta?= =?UTF-8?q?sks=20(`refresh=5Fepg=5Fdata`)=20were=20being=20killed=20mid-tr?= =?UTF-8?q?ansaction=20on=20large=20EPG=20sources.=20The=20`soft=5Ftime=5F?= =?UTF-8?q?limit=3D1700s`=20introduced=20in=20v0.21.0=20raised=20`SoftTime?= =?UTF-8?q?LimitExceeded`,=20a=20subclass=20of=20`Exception`,=20which=20wa?= =?UTF-8?q?s=20swallowed=20by=20the=20existing=20`except=20Exception`=20ha?= =?UTF-8?q?ndler=20in=20`parse=5Fprograms=5Ffor=5Fsource`,=20leaving=20the?= =?UTF-8?q?=20database=20in=20a=20partial=20state=20with=20no=20logged=20e?= =?UTF-8?q?rror.=20The=20underlying=20cause=20is=20that=20the=20HTML=20ent?= =?UTF-8?q?ity=20preprocessing=20step=20added=20in=20v0.21.0=20adds=202?= =?UTF-8?q?=E2=80=9310+=20minutes=20on=20multi-gigabyte=20EPG=20files,=20p?= =?UTF-8?q?ushing=20total=20task=20duration=20past=20the=201700s=20limit?= =?UTF-8?q?=20for=20large=20sources.=20`soft=5Ftime=5Flimit`=20has=20been?= =?UTF-8?q?=20removed=20from=20`refresh=5Fepg=5Fdata`=20and=20`time=5Flimi?= =?UTF-8?q?t`=20raised=20to=2014400s=20(4=20hours)=20as=20a=20true=20last-?= =?UTF-8?q?resort=20ceiling;=20the=20existing=20`TaskLockRenewer`=20daemon?= =?UTF-8?q?=20thread=20continues=20to=20renew=20the=20Redis=20lock=20every?= =?UTF-8?q?=20120s=20for=20legitimately=20long-running=20tasks.=20-=20HTML?= =?UTF-8?q?=20entity=20preprocessing=20(`=5Fresolve=5Fhtml=5Fentities`)=20?= =?UTF-8?q?now=20performs=20a=20fast=20binary=20pre-scan=20before=20any=20?= =?UTF-8?q?file=20read=20or=20write.=20The=20pre-scan=20reads=20the=20file?= =?UTF-8?q?=20in=204=20MB=20chunks=20without=20text=20decoding=20and=20exi?= =?UTF-8?q?ts=20immediately=20on=20the=20first=20non-XML=20named=20entity?= =?UTF-8?q?=20found.=20For=20EPG=20sources=20that=20contain=20no=20HTML=20?= =?UTF-8?q?entities=20=E2=80=94=20the=20common=20case=20=E2=80=94=20the=20?= =?UTF-8?q?full=20line-by-line=20read+rewrite=20pass=20(which=20takes=202?= =?UTF-8?q?=E2=80=9310+=20minutes=20on=20a=202.5=20GB=20file)=20is=20skipp?= =?UTF-8?q?ed=20entirely.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 2 + apps/epg/tasks.py | 104 ++++++++++++++++++++++++++++++++-------------- 2 files changed, 74 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 249ea29b..e8bccb55 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 9683c85a..3126453d 100644 --- a/apps/epg/tasks.py +++ b/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")