mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-30 05:20:12 +00:00
2a5ab5c268
This commit refactors the Schedules Direct integration by centralizing the WebSocket progress update functionality into the `utils.py` module. The `send_epg_update` function is now defined in `utils.py`, allowing for cleaner imports across the codebase and reducing circular dependencies. Additionally, references to the old `send_epg_update` function in `sd_tasks.py` and tests have been updated accordingly to ensure consistent functionality.
3311 lines
139 KiB
Python
3311 lines
139 KiB
Python
# apps/epg/tasks.py
|
|
|
|
import logging
|
|
import gzip
|
|
import html.entities
|
|
import lzma
|
|
import os
|
|
import uuid
|
|
import requests
|
|
import time # Add import for tracking download progress
|
|
from datetime import datetime, timedelta, timezone as dt_timezone
|
|
import gc # Add garbage collection module
|
|
import json
|
|
import re
|
|
from lxml import etree # Using lxml exclusively
|
|
import psutil # Add import for memory tracking
|
|
import zipfile
|
|
|
|
from celery import shared_task
|
|
from django.conf import settings
|
|
from django.db import connection, transaction
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
from apps.channels.models import Channel
|
|
from core.models import UserAgent, CoreSettings
|
|
|
|
from asgiref.sync import async_to_sync
|
|
from channels.layers import get_channel_layer
|
|
|
|
from .models import EPGSource, EPGSourceIndex, EPGData, ProgramData, SDScheduleMD5, SDProgramMD5
|
|
from apps.epg.utils import (
|
|
_ONSCREEN_RE,
|
|
extract_season_episode_from_description,
|
|
send_epg_update,
|
|
)
|
|
from apps.epg.sd_utils import SD_BASE_URL
|
|
from apps.epg.sd_tasks import (
|
|
SD_BULK_GUIDE_FETCH_THRESHOLD,
|
|
SD_DAYS_TO_FETCH,
|
|
SD_MAPPED_GUIDE_BATCH_DEFER_SECONDS,
|
|
SD_MAPPED_GUIDE_FETCH_DEFER_MAX_RETRIES,
|
|
SD_POSTER_STYLE_DEFAULT,
|
|
SD_PROGRAM_BATCH_SIZE,
|
|
_sd_backfill_schedule_dates_without_data,
|
|
_sd_compute_schedule_changes_from_md5,
|
|
_sd_pick_poster_url,
|
|
_sd_programs_needing_metadata,
|
|
fetch_schedules_direct,
|
|
fetch_schedules_direct_stations,
|
|
fetch_sd_guide_for_epg,
|
|
fetch_sd_mapped_guide_batch,
|
|
parse_schedules_direct_time,
|
|
)
|
|
from core.utils import (
|
|
RedisClient,
|
|
acquire_task_lock,
|
|
is_task_lock_held,
|
|
release_task_lock,
|
|
TaskLockRenewer,
|
|
cleanup_memory,
|
|
log_system_event,
|
|
_is_celery_worker_context,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_EPG_SOURCE_FILE_LOCK = 'epg_source_file'
|
|
_EPG_REFRESH_DEFER_SECONDS = 15
|
|
_EPG_TVG_PARSE_DEFER_SECONDS = 5
|
|
_EPG_PARSE_DEFER_MAX = 480 # 15s x 480 = 2h for refresh; 5s x 480 = 40m for index build
|
|
|
|
|
|
def _source_tvg_parse_count_key(source_id):
|
|
return f'epg_source_tvg_parse_{source_id}'
|
|
|
|
|
|
def _incr_source_tvg_parse_count(source_id):
|
|
client = RedisClient.get_client()
|
|
if client:
|
|
client.incr(_source_tvg_parse_count_key(source_id))
|
|
|
|
|
|
def _decr_source_tvg_parse_count(source_id):
|
|
client = RedisClient.get_client()
|
|
if not client:
|
|
return
|
|
key = _source_tvg_parse_count_key(source_id)
|
|
value = client.decr(key)
|
|
if value is not None and value <= 0:
|
|
client.delete(key)
|
|
|
|
|
|
def _source_tvg_parse_count(source_id):
|
|
client = RedisClient.get_client()
|
|
if not client:
|
|
return 0
|
|
return int(client.get(_source_tvg_parse_count_key(source_id)) or 0)
|
|
|
|
|
|
def _defer_refresh_epg_data(source, force, file_defer_retry, reason):
|
|
if file_defer_retry >= _EPG_PARSE_DEFER_MAX:
|
|
msg = f"EPG refresh blocked for too long ({reason}); retry later"
|
|
logger.error(f"Cannot refresh {source.name}: {msg}")
|
|
source.status = EPGSource.STATUS_ERROR
|
|
source.last_message = msg
|
|
source.save(update_fields=['status', 'last_message'])
|
|
return True
|
|
logger.info(
|
|
f"Deferring refresh for {source.name} for {_EPG_REFRESH_DEFER_SECONDS}s: {reason}"
|
|
)
|
|
refresh_epg_data.apply_async(
|
|
args=[source.id],
|
|
kwargs={'force': force, '_file_defer_retry': file_defer_retry + 1},
|
|
countdown=_EPG_REFRESH_DEFER_SECONDS,
|
|
)
|
|
return True
|
|
|
|
|
|
_NON_TERMINAL_REFRESH_STATUSES = frozenset({
|
|
EPGSource.STATUS_FETCHING,
|
|
EPGSource.STATUS_PARSING,
|
|
})
|
|
|
|
|
|
def _release_task_db_connection():
|
|
"""Return the Celery worker's DB connection to a clean state after ORM errors."""
|
|
if not _is_celery_worker_context():
|
|
return
|
|
from django.db import close_old_connections
|
|
close_old_connections()
|
|
|
|
|
|
def _db_query_with_retry(fn, *, label="DB query", max_retries=2):
|
|
"""
|
|
Run an ORM read with one connection reset + retry on transient failures.
|
|
|
|
Poisoned Celery worker connections often surface as OperationalError or as
|
|
``IndexError: list index out of range`` inside Django's row converters.
|
|
"""
|
|
from django.db import DatabaseError, InterfaceError, OperationalError
|
|
|
|
transient_errors = (OperationalError, InterfaceError, IndexError, DatabaseError)
|
|
for attempt in range(max_retries):
|
|
try:
|
|
return fn()
|
|
except transient_errors as exc:
|
|
if attempt + 1 >= max_retries:
|
|
raise
|
|
logger.warning(
|
|
"%s failed (%s), resetting DB connection (%s/%s)",
|
|
label,
|
|
exc,
|
|
attempt + 1,
|
|
max_retries,
|
|
)
|
|
_release_task_db_connection()
|
|
|
|
|
|
def _get_epg_source(source_id):
|
|
return _db_query_with_retry(
|
|
lambda: EPGSource.objects.get(id=source_id),
|
|
label=f"load EPG source {source_id}",
|
|
)
|
|
|
|
|
|
def _set_epg_source_status(
|
|
source_id,
|
|
status,
|
|
last_message=None,
|
|
*,
|
|
notify_error=False,
|
|
ws_action="refresh",
|
|
ws_error=None,
|
|
):
|
|
"""Update source status using a fresh connection (safe after DB failures)."""
|
|
_release_task_db_connection()
|
|
update = {"status": status}
|
|
if last_message is not None:
|
|
update["last_message"] = last_message
|
|
try:
|
|
EPGSource.objects.filter(id=source_id).update(**update)
|
|
if notify_error:
|
|
send_epg_update(
|
|
source_id,
|
|
ws_action,
|
|
100,
|
|
status="error",
|
|
error=ws_error or last_message,
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Failed to set EPG source {source_id} status to {status}: {e}"
|
|
)
|
|
|
|
|
|
def _ensure_epg_refresh_terminal_status(source_id):
|
|
"""Mark refresh as failed when the task exits while still in progress."""
|
|
_release_task_db_connection()
|
|
try:
|
|
current_status = (
|
|
EPGSource.objects.filter(id=source_id)
|
|
.values_list("status", flat=True)
|
|
.first()
|
|
)
|
|
if current_status in _NON_TERMINAL_REFRESH_STATUSES:
|
|
message = "Refresh did not complete successfully"
|
|
EPGSource.objects.filter(id=source_id).update(
|
|
status=EPGSource.STATUS_ERROR,
|
|
last_message=message,
|
|
)
|
|
send_epg_update(
|
|
source_id, "refresh", 100, status="error", error=message
|
|
)
|
|
except Exception as e:
|
|
logger.debug(
|
|
f"Could not verify terminal refresh status for EPG source {source_id}: {e}"
|
|
)
|
|
|
|
|
|
# DOCTYPE internal subset for XMLTV files. Declares all 252 HTML 4 named
|
|
# entities so lxml/libxml2 can resolve references like é correctly
|
|
# instead of silently dropping them in recovery mode.
|
|
# The 5 XML-predefined entities (amp, lt, gt, quot, apos) are always
|
|
# recognised by the XML spec and must not be redeclared.
|
|
_XML_ENTITIES = frozenset({'amp', 'lt', 'gt', 'quot', 'apos'})
|
|
|
|
|
|
def _build_html_entity_doctype() -> bytes:
|
|
"""Build a DOCTYPE internal subset declaring all HTML 4 named entities."""
|
|
lines = [b'<!DOCTYPE tv [\n']
|
|
for name, codepoint in sorted(html.entities.name2codepoint.items()):
|
|
if name not in _XML_ENTITIES:
|
|
# Numeric character references are always valid XML regardless of codepoint.
|
|
lines.append(f'<!ENTITY {name} "&#x{codepoint:X};">\n'.encode('ascii'))
|
|
lines.append(b']>\n')
|
|
return b''.join(lines)
|
|
|
|
|
|
_HTML_ENTITY_DOCTYPE = _build_html_entity_doctype()
|
|
|
|
|
|
def _parse_programme_element(element_bytes):
|
|
"""Parse a single <programme> element, prepending the HTML-entity DOCTYPE
|
|
so references like é in the text resolve instead of failing."""
|
|
parser = etree.XMLParser(resolve_entities=True, load_dtd=True, no_network=True)
|
|
return etree.fromstring(_HTML_ENTITY_DOCTYPE + element_bytes, parser)
|
|
|
|
|
|
class _PrependStream:
|
|
"""Wraps an open binary file and prepends a bytes prefix to its content.
|
|
|
|
Used by _open_xmltv_file to inject a DOCTYPE entity block before the
|
|
file content reaches lxml's iterparse, with zero disk I/O.
|
|
"""
|
|
|
|
__slots__ = ('_prefix', '_prefix_pos', '_file')
|
|
|
|
def __init__(self, prefix: bytes, file_obj):
|
|
self._prefix = prefix
|
|
self._prefix_pos = 0
|
|
self._file = file_obj
|
|
|
|
def read(self, size=-1):
|
|
prefix_len = len(self._prefix)
|
|
if self._prefix_pos >= prefix_len:
|
|
return self._file.read(size)
|
|
remaining = prefix_len - self._prefix_pos
|
|
if size < 0:
|
|
chunk = self._prefix[self._prefix_pos:] + self._file.read()
|
|
self._prefix_pos = prefix_len
|
|
return chunk
|
|
if size <= remaining:
|
|
chunk = self._prefix[self._prefix_pos:self._prefix_pos + size]
|
|
self._prefix_pos += size
|
|
return chunk
|
|
chunk = self._prefix[self._prefix_pos:]
|
|
self._prefix_pos = prefix_len
|
|
return chunk + self._file.read(size - remaining)
|
|
|
|
def close(self):
|
|
self._file.close()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
self.close()
|
|
|
|
|
|
def _open_xmltv_file(file_path: str):
|
|
"""Open an XMLTV file for lxml iterparse, injecting an HTML entity DOCTYPE.
|
|
|
|
Prepends a <!DOCTYPE tv [...]> block that declares all 252 HTML 4 named
|
|
entities so lxml/libxml2 resolves references like é correctly
|
|
instead of silently dropping them in recovery mode. This involves zero
|
|
disk I/O (the DOCTYPE is streamed in-memory before the file content).
|
|
|
|
If the file already contains a <!DOCTYPE> declaration the file is returned
|
|
unchanged; a second DOCTYPE would be invalid XML.
|
|
|
|
The caller is responsible for closing the returned object.
|
|
"""
|
|
f = open(file_path, 'rb')
|
|
start = f.read(512)
|
|
|
|
# Do not inject if the file already declares a DOCTYPE.
|
|
if b'<!DOCTYPE' in start or b'<!doctype' in start.lower():
|
|
f.seek(0)
|
|
return f
|
|
|
|
# Insert the DOCTYPE after the XML declaration if one is present.
|
|
xml_pos = start.find(b'<?xml')
|
|
if xml_pos >= 0:
|
|
decl_end = start.find(b'?>', xml_pos)
|
|
if decl_end >= 0:
|
|
xml_decl = start[:decl_end + 2]
|
|
f.seek(decl_end + 2)
|
|
return _PrependStream(xml_decl + b'\n' + _HTML_ENTITY_DOCTYPE, f)
|
|
|
|
# No XML declaration found; insert DOCTYPE at the very start of the file.
|
|
f.seek(0)
|
|
return _PrependStream(_HTML_ENTITY_DOCTYPE, f)
|
|
|
|
|
|
def validate_icon_url_fast(icon_url, max_length=None):
|
|
"""
|
|
Fast validation for icon URLs during parsing.
|
|
Returns None if URL is too long, original URL otherwise.
|
|
If max_length is None, gets it dynamically from the EPGData model field.
|
|
"""
|
|
if max_length is None:
|
|
# Get max_length dynamically from the model field
|
|
max_length = EPGData._meta.get_field('icon_url').max_length
|
|
|
|
if icon_url and len(icon_url) > max_length:
|
|
logger.warning(f"Icon URL too long ({len(icon_url)} > {max_length}), skipping: {icon_url[:100]}...")
|
|
return None
|
|
return icon_url
|
|
|
|
|
|
MAX_EXTRACT_CHUNK_SIZE = 65536 # 64kb (base2)
|
|
|
|
|
|
def delete_epg_refresh_task_by_id(epg_id):
|
|
"""
|
|
Delete the periodic task associated with an EPG source ID.
|
|
Can be called directly or from the post_delete signal.
|
|
Returns True if a task was found and deleted, False otherwise.
|
|
"""
|
|
try:
|
|
task = None
|
|
task_name = f"epg_source-refresh-{epg_id}"
|
|
|
|
# Look for task by name
|
|
try:
|
|
from django_celery_beat.models import PeriodicTask, IntervalSchedule
|
|
task = PeriodicTask.objects.get(name=task_name)
|
|
logger.info(f"Found task by name: {task.id} for EPGSource {epg_id}")
|
|
except PeriodicTask.DoesNotExist:
|
|
logger.warning(f"No PeriodicTask found with name {task_name}")
|
|
return False
|
|
|
|
# Now delete the task and its interval
|
|
if task:
|
|
# Store interval info before deleting the task
|
|
interval_id = None
|
|
if hasattr(task, 'interval') and task.interval:
|
|
interval_id = task.interval.id
|
|
|
|
# Count how many TOTAL tasks use this interval (including this one)
|
|
tasks_with_same_interval = PeriodicTask.objects.filter(interval_id=interval_id).count()
|
|
logger.info(f"Interval {interval_id} is used by {tasks_with_same_interval} tasks total")
|
|
|
|
# Delete the task first
|
|
task_id = task.id
|
|
task.delete()
|
|
logger.info(f"Successfully deleted periodic task {task_id}")
|
|
|
|
# Now check if we should delete the interval
|
|
# We only delete if it was the ONLY task using this interval
|
|
if interval_id and tasks_with_same_interval == 1:
|
|
try:
|
|
interval = IntervalSchedule.objects.get(id=interval_id)
|
|
logger.info(f"Deleting interval schedule {interval_id} (not shared with other tasks)")
|
|
interval.delete()
|
|
logger.info(f"Successfully deleted interval {interval_id}")
|
|
except IntervalSchedule.DoesNotExist:
|
|
logger.warning(f"Interval {interval_id} no longer exists")
|
|
elif interval_id:
|
|
logger.info(f"Not deleting interval {interval_id} as it's shared with {tasks_with_same_interval-1} other tasks")
|
|
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Error deleting periodic task for EPGSource {epg_id}: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
|
|
@shared_task
|
|
def refresh_all_epg_data():
|
|
logger.info("Starting refresh_epg_data task.")
|
|
# Exclude dummy EPG sources from refresh - they don't need refreshing
|
|
active_sources = EPGSource.objects.filter(is_active=True).exclude(source_type='dummy')
|
|
logger.debug(f"Found {active_sources.count()} active EPGSource(s) (excluding dummy EPGs).")
|
|
|
|
for source in active_sources:
|
|
refresh_epg_data(source.id)
|
|
# Force garbage collection between sources
|
|
gc.collect()
|
|
|
|
logger.info("Finished refresh_epg_data task.")
|
|
return "EPG data refreshed."
|
|
|
|
|
|
@shared_task(time_limit=14400)
|
|
def refresh_epg_data(source_id, force=False, _file_defer_retry=0):
|
|
if not acquire_task_lock('refresh_epg_data', source_id):
|
|
logger.debug(f"EPG refresh for {source_id} already running")
|
|
return
|
|
|
|
lock_renewer = TaskLockRenewer('refresh_epg_data', source_id)
|
|
lock_renewer.start()
|
|
|
|
_release_task_db_connection()
|
|
|
|
try:
|
|
return _refresh_epg_data_impl(
|
|
source_id, force=force, _file_defer_retry=_file_defer_retry
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error in refresh_epg_data for source {source_id}: {e}",
|
|
exc_info=True,
|
|
)
|
|
_set_epg_source_status(
|
|
source_id,
|
|
EPGSource.STATUS_ERROR,
|
|
f"Error refreshing EPG data: {str(e)[:500]}",
|
|
notify_error=True,
|
|
ws_error=str(e)[:500],
|
|
)
|
|
finally:
|
|
_ensure_epg_refresh_terminal_status(source_id)
|
|
_release_task_db_connection()
|
|
gc.collect()
|
|
lock_renewer.stop()
|
|
release_task_lock('refresh_epg_data', source_id)
|
|
|
|
|
|
def _refresh_epg_data_impl(source_id, force=False, _file_defer_retry=0):
|
|
try:
|
|
source = _get_epg_source(source_id)
|
|
except EPGSource.DoesNotExist:
|
|
logger.warning(
|
|
f"EPG source with ID {source_id} not found, but task was triggered. "
|
|
"Cleaning up orphaned task."
|
|
)
|
|
|
|
if delete_epg_refresh_task_by_id(source_id):
|
|
logger.info(
|
|
f"Successfully cleaned up orphaned task for EPG source {source_id}"
|
|
)
|
|
else:
|
|
logger.info(f"No orphaned task found for EPG source {source_id}")
|
|
|
|
return f"EPG source {source_id} does not exist, task cleaned up"
|
|
|
|
if not source.is_active:
|
|
logger.info(f"EPG source {source_id} is not active. Skipping.")
|
|
return
|
|
|
|
if source.source_type == 'dummy':
|
|
logger.info(
|
|
f"Skipping refresh for dummy EPG source {source.name} (ID: {source_id})"
|
|
)
|
|
return
|
|
|
|
logger.info(f"Processing EPGSource: {source.name} (type: {source.source_type})")
|
|
if source.source_type == 'xmltv':
|
|
# Invalidate the byte-offset index before downloading the new file
|
|
# so stale offsets are never used during the refresh window.
|
|
EPGSourceIndex.objects.update_or_create(
|
|
source_id=source.id, defaults={'data': None}
|
|
)
|
|
|
|
if _source_tvg_parse_count(source.id) > 0:
|
|
_defer_refresh_epg_data(
|
|
source, force, _file_defer_retry, 'per-channel program parses in progress'
|
|
)
|
|
return
|
|
|
|
if not acquire_task_lock(_EPG_SOURCE_FILE_LOCK, source.id):
|
|
_defer_refresh_epg_data(
|
|
source, force, _file_defer_retry, 'EPG file in use (index build)'
|
|
)
|
|
return
|
|
|
|
file_lock_renewer = TaskLockRenewer(_EPG_SOURCE_FILE_LOCK, source.id)
|
|
file_lock_renewer.start()
|
|
try:
|
|
if not fetch_xmltv(source):
|
|
logger.error(f"Failed to fetch XMLTV for source {source.name}")
|
|
return
|
|
|
|
if not parse_channels_only(source):
|
|
logger.error(f"Failed to parse channels for source {source.name}")
|
|
return
|
|
|
|
if not parse_programs_for_source(source):
|
|
logger.error(f"Failed to parse programs for source {source.name}")
|
|
return
|
|
finally:
|
|
file_lock_renewer.stop()
|
|
release_task_lock(_EPG_SOURCE_FILE_LOCK, source.id)
|
|
|
|
# Index build runs after the file lock is released so it does not
|
|
# compete with download/parse for the same XML on disk.
|
|
build_programme_index_task.delay(source.id)
|
|
|
|
elif source.source_type == 'schedules_direct':
|
|
fetch_schedules_direct(source, force=force)
|
|
|
|
EPGSource.objects.filter(id=source.id).update(updated_at=timezone.now())
|
|
try:
|
|
from apps.channels.tasks import evaluate_series_rules
|
|
evaluate_series_rules.delay()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def fetch_xmltv(source):
|
|
# Handle cases with local file but no URL
|
|
if not source.url and source.file_path and os.path.exists(source.file_path):
|
|
logger.info(f"Using existing local file for EPG source: {source.name} at {source.file_path}")
|
|
|
|
# Check if the existing file is compressed and we need to extract it
|
|
if source.file_path.endswith(('.gz', '.zip', '.xz')) and not source.file_path.endswith('.xml'):
|
|
try:
|
|
# Define the path for the extracted file in the cache directory
|
|
cache_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg")
|
|
os.makedirs(cache_dir, exist_ok=True)
|
|
xml_path = os.path.join(cache_dir, f"{source.id}.xml")
|
|
|
|
# Extract to the cache location keeping the original
|
|
extracted_path = extract_compressed_file(source.file_path, xml_path, delete_original=False)
|
|
|
|
if extracted_path:
|
|
logger.info(f"Extracted mapped compressed file to: {extracted_path}")
|
|
# Update to use extracted_file_path instead of changing file_path
|
|
source.extracted_file_path = extracted_path
|
|
source.save(update_fields=['extracted_file_path'])
|
|
else:
|
|
logger.error(f"Failed to extract mapped compressed file. Using original file: {source.file_path}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to extract existing compressed file: {e}")
|
|
# Continue with the original file if extraction fails
|
|
|
|
# Set the status to success in the database
|
|
source.status = 'success'
|
|
source.save(update_fields=['status'])
|
|
|
|
# Send a download complete notification
|
|
send_epg_update(source.id, "downloading", 100, status="success")
|
|
|
|
# Return True to indicate successful fetch, processing will continue with parse_channels_only
|
|
return True
|
|
|
|
# Handle cases where no URL is provided and no valid file path exists
|
|
if not source.url:
|
|
# Update source status for missing URL
|
|
source.status = 'error'
|
|
source.last_message = "No URL provided and no valid local file exists"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(source.id, "downloading", 100, status="error", error="No URL provided and no valid local file exists")
|
|
return False
|
|
|
|
logger.info(f"Fetching XMLTV data from source: {source.name}")
|
|
try:
|
|
# Get default user agent from settings
|
|
stream_settings = CoreSettings.get_stream_settings()
|
|
user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:138.0) Gecko/20100101 Firefox/138.0" # Fallback default
|
|
default_user_agent_id = stream_settings.get('default_user_agent')
|
|
if default_user_agent_id:
|
|
try:
|
|
user_agent_obj = UserAgent.objects.filter(id=int(default_user_agent_id)).first()
|
|
if user_agent_obj and user_agent_obj.user_agent:
|
|
user_agent = user_agent_obj.user_agent
|
|
logger.debug(f"Using default user agent: {user_agent}")
|
|
except (ValueError, Exception) as e:
|
|
logger.warning(f"Error retrieving default user agent, using fallback: {e}")
|
|
|
|
headers = {
|
|
'User-Agent': user_agent
|
|
}
|
|
|
|
# Update status to fetching before starting download
|
|
source.status = 'fetching'
|
|
source.save(update_fields=['status'])
|
|
|
|
# Send initial download notification
|
|
send_epg_update(source.id, "downloading", 0)
|
|
|
|
# Use streaming response to track download progress
|
|
with requests.get(source.url, headers=headers, stream=True, timeout=60) as response:
|
|
# Handle 404 specifically
|
|
if response.status_code == 404:
|
|
logger.error(f"EPG URL not found (404): {source.url}")
|
|
# Update status to error in the database
|
|
source.status = 'error'
|
|
source.last_message = f"EPG source '{source.name}' returned 404 error - will retry on next scheduled run"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
|
|
# Notify users through the WebSocket about the EPG fetch failure
|
|
channel_layer = get_channel_layer()
|
|
async_to_sync(channel_layer.group_send)(
|
|
'updates',
|
|
{
|
|
'type': 'update',
|
|
'data': {
|
|
"success": False,
|
|
"type": "epg_fetch_error",
|
|
"source_id": source.id,
|
|
"source_name": source.name,
|
|
"error_code": 404,
|
|
"message": f"EPG source '{source.name}' returned 404 error - will retry on next scheduled run"
|
|
}
|
|
}
|
|
)
|
|
# Ensure we update the download progress to 100 with error status
|
|
send_epg_update(source.id, "downloading", 100, status="error", error="URL not found (404)")
|
|
return False
|
|
|
|
# For all other error status codes
|
|
if response.status_code >= 400:
|
|
error_message = f"HTTP error {response.status_code}"
|
|
user_message = f"EPG source '{source.name}' encountered HTTP error {response.status_code}"
|
|
|
|
# Update status to error in the database
|
|
source.status = 'error'
|
|
source.last_message = user_message
|
|
source.save(update_fields=['status', 'last_message'])
|
|
|
|
# Notify users through the WebSocket
|
|
channel_layer = get_channel_layer()
|
|
async_to_sync(channel_layer.group_send)(
|
|
'updates',
|
|
{
|
|
'type': 'update',
|
|
'data': {
|
|
"success": False,
|
|
"type": "epg_fetch_error",
|
|
"source_id": source.id,
|
|
"source_name": source.name,
|
|
"error_code": response.status_code,
|
|
"message": user_message
|
|
}
|
|
}
|
|
)
|
|
# Update download progress
|
|
send_epg_update(source.id, "downloading", 100, status="error", error=user_message)
|
|
return False
|
|
|
|
response.raise_for_status()
|
|
logger.debug("XMLTV data fetched successfully.")
|
|
|
|
# Define base paths for consistent file naming
|
|
cache_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg")
|
|
os.makedirs(cache_dir, exist_ok=True)
|
|
|
|
# Create temporary download file with .tmp extension
|
|
temp_download_path = os.path.join(cache_dir, f"{source.id}.tmp")
|
|
|
|
# Check if we have content length for progress tracking
|
|
total_size = int(response.headers.get('content-length', 0))
|
|
downloaded = 0
|
|
start_time = time.time()
|
|
last_update_time = start_time
|
|
update_interval = 0.5 # Only update every 0.5 seconds
|
|
|
|
# Download to temporary file
|
|
with open(temp_download_path, 'wb') as f:
|
|
for chunk in response.iter_content(chunk_size=16384):
|
|
f.write(chunk)
|
|
|
|
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 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
|
|
|
|
# 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
|
|
|
|
# Send completion notification
|
|
send_epg_update(source.id, "downloading", 100)
|
|
|
|
# Determine the appropriate file extension based on content detection
|
|
with open(temp_download_path, 'rb') as f:
|
|
content_sample = f.read(1024) # Just need the first 1KB to detect format
|
|
|
|
# Use our helper function to detect the format
|
|
format_type, is_compressed, file_extension = detect_file_format(
|
|
file_path=source.url, # Original URL as a hint
|
|
content=content_sample # Actual file content for detection
|
|
)
|
|
|
|
logger.debug(f"File format detection results: type={format_type}, compressed={is_compressed}, extension={file_extension}")
|
|
|
|
# Ensure consistent final paths
|
|
compressed_path = os.path.join(cache_dir, f"{source.id}{file_extension}" if is_compressed else f"{source.id}.compressed")
|
|
xml_path = os.path.join(cache_dir, f"{source.id}.xml")
|
|
|
|
# Clean up old files before saving new ones
|
|
if os.path.exists(compressed_path):
|
|
try:
|
|
os.remove(compressed_path)
|
|
logger.debug(f"Removed old compressed file: {compressed_path}")
|
|
except OSError as e:
|
|
logger.warning(f"Failed to remove old compressed file: {e}")
|
|
|
|
if os.path.exists(xml_path):
|
|
try:
|
|
os.remove(xml_path)
|
|
logger.debug(f"Removed old XML file: {xml_path}")
|
|
except OSError as e:
|
|
logger.warning(f"Failed to remove old XML file: {e}")
|
|
|
|
# Rename the temp file to appropriate final path
|
|
if is_compressed:
|
|
try:
|
|
os.rename(temp_download_path, compressed_path)
|
|
logger.debug(f"Renamed temp file to compressed file: {compressed_path}")
|
|
current_file_path = compressed_path
|
|
except OSError as e:
|
|
logger.error(f"Failed to rename temp file to compressed file: {e}")
|
|
current_file_path = temp_download_path # Fall back to using temp file
|
|
else:
|
|
try:
|
|
os.rename(temp_download_path, xml_path)
|
|
logger.debug(f"Renamed temp file to XML file: {xml_path}")
|
|
current_file_path = xml_path
|
|
except OSError as e:
|
|
logger.error(f"Failed to rename temp file to XML file: {e}")
|
|
current_file_path = temp_download_path # Fall back to using temp file
|
|
|
|
# Now extract the file if it's compressed
|
|
if is_compressed:
|
|
try:
|
|
logger.info(f"Extracting compressed file {current_file_path}")
|
|
send_epg_update(source.id, "extracting", 0, message="Extracting downloaded file")
|
|
|
|
# Always extract to the standard XML path - set delete_original to True to clean up
|
|
extracted = extract_compressed_file(current_file_path, xml_path, delete_original=True)
|
|
|
|
if extracted:
|
|
logger.info(f"Successfully extracted to {xml_path}, compressed file deleted")
|
|
send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully, temporary file removed")
|
|
# Update to store only the extracted file path since the compressed file is now gone
|
|
source.file_path = xml_path
|
|
source.extracted_file_path = None
|
|
else:
|
|
logger.error("Extraction failed, using compressed file")
|
|
send_epg_update(source.id, "extracting", 100, status="error", message="Extraction failed, using compressed file")
|
|
# Use the compressed file
|
|
source.file_path = current_file_path
|
|
source.extracted_file_path = None
|
|
except Exception as e:
|
|
logger.error(f"Error extracting file: {str(e)}", exc_info=True)
|
|
send_epg_update(source.id, "extracting", 100, status="error", message=f"Error during extraction: {str(e)}")
|
|
# Use the compressed file if extraction fails
|
|
source.file_path = current_file_path
|
|
source.extracted_file_path = None
|
|
else:
|
|
# It's already an XML file
|
|
source.file_path = current_file_path
|
|
source.extracted_file_path = None
|
|
|
|
# Update the source's file paths
|
|
source.save(update_fields=['file_path', 'status', 'extracted_file_path'])
|
|
|
|
# Update status to parsing
|
|
source.status = 'parsing'
|
|
source.save(update_fields=['status'])
|
|
|
|
logger.info(f"Cached EPG file saved to {source.file_path}")
|
|
|
|
return True
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
logger.error(f"HTTP Error fetching XMLTV from {source.name}: {e}", exc_info=True)
|
|
|
|
# Get error details
|
|
status_code = e.response.status_code if hasattr(e, 'response') and e.response else 'unknown'
|
|
error_message = str(e)
|
|
|
|
# Create a user-friendly message
|
|
user_message = f"EPG source '{source.name}' encountered HTTP error {status_code}"
|
|
|
|
# Add specific handling for common HTTP errors
|
|
if status_code == 404:
|
|
user_message = f"EPG source '{source.name}' URL not found (404) - will retry on next scheduled run"
|
|
elif status_code == 401 or status_code == 403:
|
|
user_message = f"EPG source '{source.name}' access denied (HTTP {status_code}) - check credentials"
|
|
elif status_code == 429:
|
|
user_message = f"EPG source '{source.name}' rate limited (429) - try again later"
|
|
elif status_code >= 500:
|
|
user_message = f"EPG source '{source.name}' server error (HTTP {status_code}) - will retry later"
|
|
|
|
# Update source status to error with the error message
|
|
source.status = 'error'
|
|
source.last_message = user_message
|
|
source.save(update_fields=['status', 'last_message'])
|
|
|
|
# Notify users through the WebSocket about the EPG fetch failure
|
|
channel_layer = get_channel_layer()
|
|
async_to_sync(channel_layer.group_send)(
|
|
'updates',
|
|
{
|
|
'type': 'update',
|
|
'data': {
|
|
"success": False,
|
|
"type": "epg_fetch_error",
|
|
"source_id": source.id,
|
|
"source_name": source.name,
|
|
"error_code": status_code,
|
|
"message": user_message,
|
|
"details": error_message
|
|
}
|
|
}
|
|
)
|
|
|
|
# Ensure we update the download progress to 100 with error status
|
|
send_epg_update(source.id, "downloading", 100, status="error", error=user_message)
|
|
return False
|
|
except requests.exceptions.ConnectionError as e:
|
|
# Handle connection errors separately
|
|
error_message = str(e)
|
|
user_message = f"Connection error: Unable to connect to EPG source '{source.name}'"
|
|
logger.error(f"Connection error fetching XMLTV from {source.name}: {e}", exc_info=True)
|
|
|
|
# Update source status
|
|
source.status = 'error'
|
|
source.last_message = user_message
|
|
source.save(update_fields=['status', 'last_message'])
|
|
|
|
# Send notifications
|
|
channel_layer = get_channel_layer()
|
|
async_to_sync(channel_layer.group_send)(
|
|
'updates',
|
|
{
|
|
'type': 'update',
|
|
'data': {
|
|
"success": False,
|
|
"type": "epg_fetch_error",
|
|
"source_id": source.id,
|
|
"source_name": source.name,
|
|
"error_code": "connection_error",
|
|
"message": user_message
|
|
}
|
|
}
|
|
)
|
|
send_epg_update(source.id, "downloading", 100, status="error", error=user_message)
|
|
return False
|
|
except requests.exceptions.Timeout as e:
|
|
# Handle timeout errors specifically
|
|
error_message = str(e)
|
|
user_message = f"Timeout error: EPG source '{source.name}' took too long to respond"
|
|
logger.error(f"Timeout error fetching XMLTV from {source.name}: {e}", exc_info=True)
|
|
|
|
# Update source status
|
|
source.status = 'error'
|
|
source.last_message = user_message
|
|
source.save(update_fields=['status', 'last_message'])
|
|
|
|
# Send notifications
|
|
send_epg_update(source.id, "downloading", 100, status="error", error=user_message)
|
|
return False
|
|
except Exception as e:
|
|
error_message = str(e)
|
|
logger.error(f"Error fetching XMLTV from {source.name}: {e}", exc_info=True)
|
|
|
|
# Update source status for general exceptions too
|
|
source.status = 'error'
|
|
source.last_message = f"Error: {error_message}"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
|
|
# Ensure we update the download progress to 100 with error status
|
|
send_epg_update(source.id, "downloading", 100, status="error", error=f"Error: {error_message}")
|
|
return False
|
|
|
|
|
|
def extract_compressed_file(file_path, output_path=None, delete_original=False):
|
|
"""
|
|
Extracts a compressed file (.gz, .zip, or .xz) to an XML file.
|
|
|
|
Args:
|
|
file_path: Path to the compressed file
|
|
output_path: Specific path where the file should be extracted (optional)
|
|
delete_original: Whether to delete the original compressed file after successful extraction
|
|
|
|
Returns:
|
|
Path to the extracted XML file, or None if extraction failed
|
|
"""
|
|
try:
|
|
if output_path is None:
|
|
base_path = os.path.splitext(file_path)[0]
|
|
extracted_path = f"{base_path}.xml"
|
|
else:
|
|
extracted_path = output_path
|
|
|
|
# Make sure the output path doesn't already exist
|
|
if os.path.exists(extracted_path):
|
|
try:
|
|
os.remove(extracted_path)
|
|
logger.info(f"Removed existing extracted file: {extracted_path}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to remove existing extracted file: {e}")
|
|
# If we can't delete the existing file and no specific output was requested,
|
|
# create a unique filename instead
|
|
if output_path is None:
|
|
base_path = os.path.splitext(file_path)[0]
|
|
extracted_path = f"{base_path}_{uuid.uuid4().hex[:8]}.xml"
|
|
|
|
# Use our detection helper to determine the file format instead of relying on extension
|
|
with open(file_path, 'rb') as f:
|
|
content_sample = f.read(4096) # Read a larger sample to ensure accurate detection
|
|
|
|
format_type, is_compressed, _ = detect_file_format(file_path=file_path, content=content_sample)
|
|
|
|
if format_type == 'gzip':
|
|
logger.debug(f"Extracting gzip file: {file_path}")
|
|
try:
|
|
# First check if the content is XML by reading a sample
|
|
with gzip.open(file_path, 'rb') as gz_file:
|
|
content_sample = gz_file.read(4096) # Read first 4KB for detection
|
|
detected_format, _, _ = detect_file_format(content=content_sample)
|
|
|
|
if detected_format != 'xml':
|
|
logger.warning(f"GZIP file does not appear to contain XML content: {file_path} (detected as: {detected_format})")
|
|
# Continue anyway since GZIP only contains one file
|
|
|
|
# Reset file pointer and extract the content
|
|
gz_file.seek(0)
|
|
with open(extracted_path, 'wb') as out_file:
|
|
while True:
|
|
chunk = gz_file.read(MAX_EXTRACT_CHUNK_SIZE)
|
|
if not chunk or len(chunk) == 0:
|
|
break
|
|
out_file.write(chunk)
|
|
except Exception as e:
|
|
logger.error(f"Error extracting GZIP file: {e}", exc_info=True)
|
|
return None
|
|
|
|
logger.info(f"Successfully extracted gzip file to: {extracted_path}")
|
|
|
|
# Delete original compressed file if requested
|
|
if delete_original:
|
|
try:
|
|
os.remove(file_path)
|
|
logger.info(f"Deleted original compressed file: {file_path}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to delete original compressed file {file_path}: {e}")
|
|
|
|
return extracted_path
|
|
|
|
elif format_type == 'xz':
|
|
logger.debug(f"Extracting xz file: {file_path}")
|
|
try:
|
|
# First check if the content is XML by reading a sample
|
|
with lzma.open(file_path, 'rb') as xz_file:
|
|
content_sample = xz_file.read(4096) # Read first 4KB for detection
|
|
detected_format, _, _ = detect_file_format(content=content_sample)
|
|
|
|
if detected_format != 'xml':
|
|
logger.warning(f"XZ file does not appear to contain XML content: {file_path} (detected as: {detected_format})")
|
|
# Continue anyway since XZ only contains one file
|
|
|
|
# Reset file pointer and extract the content
|
|
xz_file.seek(0)
|
|
with open(extracted_path, 'wb') as out_file:
|
|
while True:
|
|
chunk = xz_file.read(MAX_EXTRACT_CHUNK_SIZE)
|
|
if not chunk or len(chunk) == 0:
|
|
break
|
|
out_file.write(chunk)
|
|
except Exception as e:
|
|
logger.error(f"Error extracting XZ file: {e}", exc_info=True)
|
|
return None
|
|
|
|
logger.info(f"Successfully extracted xz file to: {extracted_path}")
|
|
|
|
# Delete original compressed file if requested
|
|
if delete_original:
|
|
try:
|
|
os.remove(file_path)
|
|
logger.info(f"Deleted original compressed file: {file_path}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to delete original compressed file {file_path}: {e}")
|
|
|
|
return extracted_path
|
|
|
|
elif format_type == 'zip':
|
|
logger.debug(f"Extracting zip file: {file_path}")
|
|
with zipfile.ZipFile(file_path, 'r') as zip_file:
|
|
# Find the first XML file in the ZIP archive
|
|
xml_files = [f for f in zip_file.namelist() if f.lower().endswith('.xml')]
|
|
|
|
if not xml_files:
|
|
logger.info("No files with .xml extension found in ZIP archive, checking content of all files")
|
|
# Check content of each file to see if any are XML without proper extension
|
|
for filename in zip_file.namelist():
|
|
if not filename.endswith('/'): # Skip directories
|
|
try:
|
|
# Read a sample of the file content
|
|
content_sample = zip_file.read(filename, 4096) # Read up to 4KB for detection
|
|
format_type, _, _ = detect_file_format(content=content_sample)
|
|
if format_type == 'xml':
|
|
logger.info(f"Found XML content in file without .xml extension: {filename}")
|
|
xml_files = [filename]
|
|
break
|
|
except Exception as e:
|
|
logger.warning(f"Error reading file {filename} from ZIP: {e}")
|
|
|
|
if not xml_files:
|
|
logger.error("No XML file found in ZIP archive")
|
|
return None
|
|
|
|
# Extract the first XML file
|
|
with open(extracted_path, 'wb') as out_file:
|
|
with zip_file.open(xml_files[0], "r") as xml_file:
|
|
while True:
|
|
chunk = xml_file.read(MAX_EXTRACT_CHUNK_SIZE)
|
|
if not chunk or len(chunk) == 0:
|
|
break
|
|
out_file.write(chunk)
|
|
|
|
logger.info(f"Successfully extracted zip file to: {extracted_path}")
|
|
|
|
# Delete original compressed file if requested
|
|
if delete_original:
|
|
try:
|
|
os.remove(file_path)
|
|
logger.info(f"Deleted original compressed file: {file_path}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to delete original compressed file {file_path}: {e}")
|
|
|
|
return extracted_path
|
|
|
|
else:
|
|
logger.error(f"Unsupported or unrecognized compressed file format: {file_path} (detected as: {format_type})")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error extracting {file_path}: {str(e)}", exc_info=True)
|
|
return None
|
|
|
|
_CHANNEL_PARSE_PROGRESS_START = 25
|
|
_CHANNEL_PARSE_PROGRESS_CAP = 98
|
|
_CHANNEL_PARSE_PROGRESS_SCALE = 0.98
|
|
|
|
|
|
def _channel_parse_progress(processed_channels, channel_estimate, had_db_baseline):
|
|
"""Map channel parsing onto 25-98%. Bump estimate when the XML has grown."""
|
|
if not had_db_baseline:
|
|
if processed_channels <= 0:
|
|
return _CHANNEL_PARSE_PROGRESS_START, channel_estimate
|
|
return (
|
|
min(
|
|
_CHANNEL_PARSE_PROGRESS_START + (processed_channels // 100),
|
|
_CHANNEL_PARSE_PROGRESS_CAP - 1,
|
|
),
|
|
channel_estimate,
|
|
)
|
|
|
|
if processed_channels > channel_estimate:
|
|
channel_estimate = processed_channels
|
|
if channel_estimate <= 0:
|
|
return _CHANNEL_PARSE_PROGRESS_START, channel_estimate
|
|
|
|
span = _CHANNEL_PARSE_PROGRESS_CAP - _CHANNEL_PARSE_PROGRESS_START
|
|
progress = _CHANNEL_PARSE_PROGRESS_START + int(
|
|
(processed_channels / channel_estimate) * span * _CHANNEL_PARSE_PROGRESS_SCALE
|
|
)
|
|
return min(_CHANNEL_PARSE_PROGRESS_CAP, progress), channel_estimate
|
|
|
|
|
|
def parse_channels_only(source):
|
|
# Use extracted file if available, otherwise use the original file path
|
|
file_path = source.extracted_file_path if source.extracted_file_path else source.file_path
|
|
if not file_path:
|
|
file_path = source.get_cache_file()
|
|
|
|
# Send initial parsing notification
|
|
send_epg_update(source.id, "parsing_channels", 0)
|
|
|
|
process = None
|
|
should_log_memory = False
|
|
|
|
try:
|
|
# Check if the file exists
|
|
if not os.path.exists(file_path):
|
|
logger.error(f"EPG file does not exist at path: {file_path}")
|
|
|
|
# Update the source's file_path to the default cache location
|
|
new_path = source.get_cache_file()
|
|
logger.info(f"Updating file_path from '{file_path}' to '{new_path}'")
|
|
source.file_path = new_path
|
|
source.save(update_fields=['file_path'])
|
|
|
|
# If the source has a URL, fetch the data before continuing
|
|
if source.url:
|
|
logger.info(f"Fetching new EPG data from URL: {source.url}")
|
|
fetch_success = fetch_xmltv(source) # Store the result
|
|
|
|
# Only proceed if fetch was successful AND file exists
|
|
if not fetch_success:
|
|
logger.error(f"Failed to fetch EPG data from URL: {source.url}")
|
|
# Update status to error
|
|
source.status = 'error'
|
|
source.last_message = f"Failed to fetch EPG data from URL"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
# Send error notification
|
|
send_epg_update(source.id, "parsing_channels", 100, status="error", error="Failed to fetch EPG data")
|
|
return False
|
|
|
|
# Verify the file was downloaded successfully
|
|
if not os.path.exists(source.file_path):
|
|
logger.error(f"Failed to fetch EPG data, file still missing at: {source.file_path}")
|
|
# Update status to error
|
|
source.status = 'error'
|
|
source.last_message = f"Failed to fetch EPG data, file missing after download"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(source.id, "parsing_channels", 100, status="error", error="File not found after download")
|
|
return False
|
|
|
|
# Update file_path with the new location
|
|
file_path = source.file_path
|
|
else:
|
|
logger.error(f"No URL provided for EPG source {source.name}, cannot fetch new data")
|
|
# Update status to error
|
|
source.status = 'error'
|
|
source.last_message = f"No URL provided, cannot fetch EPG data"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(
|
|
source.id,
|
|
"parsing_channels",
|
|
100,
|
|
status="error",
|
|
error="No URL provided",
|
|
)
|
|
return False
|
|
|
|
# Initialize process variable for memory tracking only in debug mode
|
|
try:
|
|
process = None
|
|
# Get current log level as a number
|
|
current_log_level = logger.getEffectiveLevel()
|
|
|
|
# Only track memory usage when log level is DEBUG (10) or more verbose
|
|
# This is more future-proof than string comparisons
|
|
should_log_memory = current_log_level <= logging.DEBUG or settings.DEBUG
|
|
|
|
if should_log_memory:
|
|
process = psutil.Process()
|
|
initial_memory = process.memory_info().rss / 1024 / 1024
|
|
logger.debug(f"[parse_channels_only] Initial memory usage: {initial_memory:.2f} MB")
|
|
except (ImportError, NameError):
|
|
process = None
|
|
should_log_memory = False
|
|
logger.warning("psutil not available for memory tracking")
|
|
|
|
# Replace full dictionary load with more efficient lookup set
|
|
existing_tvg_ids = set()
|
|
existing_epgs = {}
|
|
scanned_tvg_ids = set() # Track tvg_ids seen in the current scan for stale cleanup
|
|
last_id = 0
|
|
chunk_size = 5000
|
|
|
|
while True:
|
|
tvg_id_chunk = set(EPGData.objects.filter(
|
|
epg_source=source,
|
|
id__gt=last_id
|
|
).order_by('id').values_list('tvg_id', flat=True)[:chunk_size])
|
|
|
|
if not tvg_id_chunk:
|
|
break
|
|
|
|
existing_tvg_ids.update(tvg_id_chunk)
|
|
last_id = EPGData.objects.filter(tvg_id__in=tvg_id_chunk).order_by('-id')[0].id
|
|
# Update progress to show file read starting
|
|
send_epg_update(source.id, "parsing_channels", 10)
|
|
|
|
# Stream parsing instead of loading entire file at once
|
|
# This can be simplified since we now always have XML files
|
|
epgs_to_create = []
|
|
epgs_to_update = []
|
|
total_channels = 0
|
|
channel_estimate = 0
|
|
had_db_baseline = False
|
|
processed_channels = 0
|
|
batch_size = 500 # Process in batches to limit memory usage
|
|
progress = 0 # Initialize progress variable here
|
|
icon_url_max_length = EPGData._meta.get_field('icon_url').max_length # Get max length for icon_url field
|
|
name_max_length = EPGData._meta.get_field('name').max_length # Get max length for name field
|
|
|
|
# Track memory at key points
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory before opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
try:
|
|
# Attempt to count existing channels in the database
|
|
try:
|
|
total_channels = EPGData.objects.filter(epg_source=source).count()
|
|
channel_estimate = total_channels
|
|
had_db_baseline = total_channels > 0
|
|
logger.info(f"Found {total_channels} existing channels for this source")
|
|
except Exception as e:
|
|
logger.error(f"Error counting channels: {e}")
|
|
total_channels = 500 # Default estimate
|
|
channel_estimate = total_channels
|
|
had_db_baseline = True
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory after closing initial file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
# Update progress after counting
|
|
send_epg_update(source.id, "parsing_channels", 25, total_channels=total_channels)
|
|
|
|
# Open the file - no need to check file type since it's always XML now
|
|
logger.debug(f"Opening file for channel parsing: {file_path}")
|
|
source_file = _open_xmltv_file(file_path)
|
|
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory after opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
# Change iterparse to look for both channel and programme elements
|
|
logger.debug(f"Creating iterparse context for channels and programmes")
|
|
channel_parser = etree.iterparse(source_file, events=('end',), tag=('channel', 'programme'), remove_blank_text=True, recover=True)
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory after creating iterparse: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
channel_count = 0
|
|
total_elements_processed = 0 # Track total elements processed, not just channels
|
|
for _, elem in channel_parser:
|
|
total_elements_processed += 1
|
|
# Only process channel elements
|
|
if elem.tag == 'channel':
|
|
channel_count += 1
|
|
tvg_id = elem.get('id', '').strip()
|
|
if tvg_id:
|
|
scanned_tvg_ids.add(tvg_id)
|
|
display_name = None
|
|
icon_url = None
|
|
for child in elem:
|
|
if display_name is None and child.tag == 'display-name' and child.text:
|
|
display_name = child.text.strip()
|
|
elif child.tag == 'icon':
|
|
raw_icon_url = child.get('src', '').strip()
|
|
icon_url = validate_icon_url_fast(raw_icon_url, icon_url_max_length)
|
|
if display_name and icon_url:
|
|
break # No need to continue if we have both
|
|
|
|
if not display_name:
|
|
display_name = tvg_id
|
|
|
|
if display_name and len(display_name) > name_max_length:
|
|
logger.warning(f"EPG display name too long ({len(display_name)} > {name_max_length}), truncating: {display_name[:80]}...")
|
|
display_name = display_name[:name_max_length]
|
|
|
|
# Use lazy loading approach to reduce memory usage
|
|
if tvg_id in existing_tvg_ids:
|
|
# Only fetch the object if we need to update it and it hasn't been loaded yet
|
|
if tvg_id not in existing_epgs:
|
|
try:
|
|
# This loads the full EPG object from the database and caches it
|
|
existing_epgs[tvg_id] = EPGData.objects.get(tvg_id=tvg_id, epg_source=source)
|
|
except EPGData.DoesNotExist:
|
|
# Handle race condition where record was deleted
|
|
existing_tvg_ids.remove(tvg_id)
|
|
epgs_to_create.append(EPGData(
|
|
tvg_id=tvg_id,
|
|
name=display_name,
|
|
icon_url=icon_url,
|
|
epg_source=source,
|
|
))
|
|
logger.debug(f"[parse_channels_only] Added new channel to epgs_to_create 1: {tvg_id} - {display_name}")
|
|
processed_channels += 1
|
|
continue
|
|
|
|
# We use the cached object to check if the name or icon_url has changed
|
|
epg_obj = existing_epgs[tvg_id]
|
|
needs_update = False
|
|
if epg_obj.name != display_name:
|
|
epg_obj.name = display_name
|
|
needs_update = True
|
|
if epg_obj.icon_url != icon_url:
|
|
epg_obj.icon_url = icon_url
|
|
needs_update = True
|
|
|
|
if needs_update:
|
|
epgs_to_update.append(epg_obj)
|
|
logger.debug(f"[parse_channels_only] Added channel to update to epgs_to_update: {tvg_id} - {display_name}")
|
|
else:
|
|
# No changes needed, just clear the element
|
|
logger.debug(f"[parse_channels_only] No changes needed for channel {tvg_id} - {display_name}")
|
|
else:
|
|
# This is a new channel that doesn't exist in our database
|
|
epgs_to_create.append(EPGData(
|
|
tvg_id=tvg_id,
|
|
name=display_name,
|
|
icon_url=icon_url,
|
|
epg_source=source,
|
|
))
|
|
logger.debug(f"[parse_channels_only] Added new channel to epgs_to_create 2: {tvg_id} - {display_name}")
|
|
|
|
processed_channels += 1
|
|
|
|
# Batch processing
|
|
if len(epgs_to_create) >= batch_size:
|
|
logger.info(f"[parse_channels_only] Bulk creating {len(epgs_to_create)} EPG entries")
|
|
EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True)
|
|
if process:
|
|
logger.info(f"[parse_channels_only] Memory after bulk_create: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
del epgs_to_create # Explicit deletion
|
|
epgs_to_create = []
|
|
cleanup_memory(log_usage=should_log_memory, force_collection=True)
|
|
if process:
|
|
logger.info(f"[parse_channels_only] Memory after gc.collect(): {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
if len(epgs_to_update) >= batch_size:
|
|
logger.info(f"[parse_channels_only] Bulk updating {len(epgs_to_update)} EPG entries")
|
|
if process:
|
|
logger.info(f"[parse_channels_only] Memory before bulk_update: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
EPGData.objects.bulk_update(epgs_to_update, ["name", "icon_url"])
|
|
if process:
|
|
logger.info(f"[parse_channels_only] Memory after bulk_update: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
epgs_to_update = []
|
|
# Force garbage collection
|
|
cleanup_memory(log_usage=should_log_memory, force_collection=True)
|
|
|
|
# Periodically clear the existing_epgs cache to prevent memory buildup
|
|
if processed_channels % 1000 == 0:
|
|
logger.info(f"[parse_channels_only] Clearing existing_epgs cache at {processed_channels} channels")
|
|
existing_epgs.clear()
|
|
cleanup_memory(log_usage=should_log_memory, force_collection=True)
|
|
if process:
|
|
logger.info(f"[parse_channels_only] Memory after clearing cache: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
# Send progress updates (~20 ticks when the DB count still holds; else every 100)
|
|
estimate_exceeded = had_db_baseline and processed_channels > total_channels
|
|
interval = (
|
|
max(1, total_channels // 20)
|
|
if had_db_baseline and total_channels and not estimate_exceeded
|
|
else 100
|
|
)
|
|
if processed_channels % interval == 0:
|
|
progress, channel_estimate = _channel_parse_progress(
|
|
processed_channels,
|
|
channel_estimate,
|
|
had_db_baseline,
|
|
)
|
|
send_epg_update(
|
|
source.id,
|
|
"parsing_channels",
|
|
progress,
|
|
processed=processed_channels,
|
|
total=channel_estimate,
|
|
)
|
|
if had_db_baseline and processed_channels > total_channels:
|
|
logger.debug(
|
|
f"[parse_channels_only] Processed channel {tvg_id} - "
|
|
f"processed {processed_channels - total_channels} additional channels"
|
|
)
|
|
else:
|
|
logger.debug(
|
|
f"[parse_channels_only] Processed channel {tvg_id} - "
|
|
f"processed {processed_channels}/{channel_estimate or total_channels}"
|
|
)
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory before elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
# Clear memory
|
|
try:
|
|
# First clear the element's content
|
|
clear_element(elem)
|
|
|
|
except Exception as e:
|
|
# Just log the error and continue - don't let cleanup errors stop processing
|
|
logger.debug(f"[parse_channels_only] Non-critical error during XML element cleanup: {e}")
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory after elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
logger.debug(f"[parse_channels_only] Total elements processed: {total_elements_processed}")
|
|
|
|
else:
|
|
logger.trace(f"[parse_channels_only] Skipping non-channel element: {elem.get('channel', 'unknown')} - {elem.get('start', 'unknown')} {elem.tag}")
|
|
clear_element(elem)
|
|
continue
|
|
|
|
except (etree.XMLSyntaxError, Exception) as xml_error:
|
|
logger.error(f"[parse_channels_only] XML parsing failed: {xml_error}")
|
|
# Update status to error
|
|
source.status = 'error'
|
|
source.last_message = f"Error parsing XML file: {str(xml_error)}"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(xml_error))
|
|
return False
|
|
if process:
|
|
logger.info(f"[parse_channels_only] Processed {processed_channels} channels current memory: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
else:
|
|
logger.info(f"[parse_channels_only] Processed {processed_channels} channels")
|
|
|
|
if processed_channels > 0:
|
|
progress, channel_estimate = _channel_parse_progress(
|
|
processed_channels,
|
|
channel_estimate,
|
|
had_db_baseline,
|
|
)
|
|
send_epg_update(
|
|
source.id,
|
|
"parsing_channels",
|
|
progress,
|
|
processed=processed_channels,
|
|
total=channel_estimate,
|
|
)
|
|
|
|
# Process any remaining items
|
|
if epgs_to_create:
|
|
EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True)
|
|
logger.debug(f"[parse_channels_only] Created final batch of {len(epgs_to_create)} EPG entries")
|
|
|
|
if epgs_to_update:
|
|
EPGData.objects.bulk_update(epgs_to_update, ["name", "icon_url"])
|
|
logger.debug(f"[parse_channels_only] Updated final batch of {len(epgs_to_update)} EPG entries")
|
|
|
|
# Clean up stale EPGData: entries that existed before the scan but weren't seen, and aren't mapped to any channel.
|
|
# Use existing_tvg_ids - scanned_tvg_ids to avoid a full-table scan with a large EXCLUDE list.
|
|
potentially_stale = existing_tvg_ids - scanned_tvg_ids
|
|
if potentially_stale:
|
|
stale_qs = EPGData.objects.filter(epg_source=source, tvg_id__in=potentially_stale, channels__isnull=True)
|
|
deleted_count, _ = stale_qs.delete()
|
|
if deleted_count:
|
|
logger.info(f"[parse_channels_only] Cleaned up {deleted_count} stale EPG entries not in current scan and unmapped to any channel")
|
|
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory after final batch creation: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
|
|
# Update source status with channel count
|
|
source.status = 'success'
|
|
source.last_message = f"Successfully parsed {processed_channels} channels"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
|
|
# Send completion notification
|
|
send_epg_update(
|
|
source.id,
|
|
"parsing_channels",
|
|
100,
|
|
status="success",
|
|
channels_count=processed_channels
|
|
)
|
|
|
|
logger.info(f"Finished parsing channel info. Found {processed_channels} channels.")
|
|
|
|
from apps.channels.utils import maybe_auto_apply_epg_logos
|
|
maybe_auto_apply_epg_logos(source)
|
|
|
|
return True
|
|
|
|
except FileNotFoundError:
|
|
logger.error(f"EPG file not found at: {file_path}")
|
|
# Update status to error
|
|
source.status = 'error'
|
|
source.last_message = f"EPG file not found: {file_path}"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(source.id, "parsing_channels", 100, status="error", error="File not found")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Error reading EPG file {file_path}: {e}", exc_info=True)
|
|
# Update status to error
|
|
source.status = 'error'
|
|
source.last_message = f"Error parsing EPG file: {str(e)}"
|
|
source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(e))
|
|
return False
|
|
finally:
|
|
# Cleanup memory and close file
|
|
if process:
|
|
logger.debug(f"[parse_channels_only] Memory before cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
|
|
try:
|
|
# Output any errors in the channel_parser error log
|
|
if 'channel_parser' in locals() and hasattr(channel_parser, 'error_log') and len(channel_parser.error_log) > 0:
|
|
logger.debug(f"XML parser errors found ({len(channel_parser.error_log)} total):")
|
|
for i, error in enumerate(channel_parser.error_log):
|
|
logger.debug(f" Error {i+1}: {error}")
|
|
if 'channel_parser' in locals():
|
|
del channel_parser
|
|
if 'elem' in locals():
|
|
del elem
|
|
if 'parent' in locals():
|
|
del parent
|
|
|
|
if 'source_file' in locals():
|
|
source_file.close()
|
|
del source_file
|
|
# Clear remaining large data structures
|
|
existing_epgs.clear()
|
|
epgs_to_create.clear()
|
|
epgs_to_update.clear()
|
|
existing_epgs = None
|
|
epgs_to_create = None
|
|
epgs_to_update = None
|
|
if 'scanned_tvg_ids' in locals() and scanned_tvg_ids is not None:
|
|
scanned_tvg_ids.clear()
|
|
scanned_tvg_ids = None
|
|
cleanup_memory(log_usage=should_log_memory, force_collection=True)
|
|
except Exception as e:
|
|
logger.warning(f"Cleanup error: {e}")
|
|
|
|
try:
|
|
if process:
|
|
final_memory = process.memory_info().rss / 1024 / 1024
|
|
logger.debug(f"[parse_channels_only] Final memory usage: {final_memory:.2f} MB")
|
|
process = None
|
|
except:
|
|
pass
|
|
|
|
|
|
def _defer_parse_programs_for_tvg_id(epg_id, force, defer_retry, reason):
|
|
if defer_retry >= _EPG_PARSE_DEFER_MAX:
|
|
logger.warning(
|
|
f"Giving up on parse_programs_for_tvg_id({epg_id}) after "
|
|
f"{defer_retry} deferrals: {reason}"
|
|
)
|
|
return f"Deferred too many times: {reason}"
|
|
|
|
logger.info(
|
|
f"Deferring parse_programs_for_tvg_id({epg_id}) for "
|
|
f"{_EPG_REFRESH_DEFER_SECONDS}s: {reason}"
|
|
)
|
|
parse_programs_for_tvg_id.apply_async(
|
|
args=[epg_id],
|
|
kwargs={'force': force, '_defer_retry': defer_retry + 1},
|
|
countdown=_EPG_REFRESH_DEFER_SECONDS,
|
|
)
|
|
return "Deferred"
|
|
|
|
|
|
@shared_task(time_limit=3600, soft_time_limit=3500)
|
|
def parse_programs_for_tvg_id(epg_id, force=False, _defer_retry=0):
|
|
epg_obj = None
|
|
try:
|
|
epg_obj = EPGData.objects.select_related('epg_source').filter(id=epg_id).first()
|
|
if epg_obj and epg_obj.epg_source and epg_obj.epg_source.source_type == 'schedules_direct':
|
|
return fetch_sd_guide_for_epg(epg_id, force=force)
|
|
except Exception as e:
|
|
logger.warning(f"Could not check EPG source type for id={epg_id}: {e}")
|
|
|
|
if not epg_obj or not epg_obj.epg_source:
|
|
return "EPG not found"
|
|
|
|
epg_source = epg_obj.epg_source
|
|
if epg_source.source_type == 'dummy':
|
|
return
|
|
|
|
source_id = epg_source.id
|
|
if is_task_lock_held('refresh_epg_data', source_id):
|
|
return _defer_parse_programs_for_tvg_id(
|
|
epg_id, force, _defer_retry, 'source refresh in progress'
|
|
)
|
|
|
|
_incr_source_tvg_parse_count(source_id)
|
|
try:
|
|
if not acquire_task_lock('parse_epg_programs', epg_id):
|
|
mapped_channels = Channel.objects.filter(epg_data_id=epg_id).count()
|
|
logger.info(
|
|
f"Program parse for epg_id={epg_id} ({epg_obj.tvg_id}) already in progress, "
|
|
f"skipping duplicate task ({mapped_channels} channel(s) mapped to this EPG)"
|
|
)
|
|
return "Task already running"
|
|
|
|
lock_renewer = TaskLockRenewer('parse_epg_programs', epg_id)
|
|
lock_renewer.start()
|
|
|
|
source_file = None
|
|
program_parser = None
|
|
programs_to_create = []
|
|
programs_processed = 0
|
|
try:
|
|
# Add memory tracking only in trace mode or higher
|
|
try:
|
|
process = None
|
|
# Get current log level as a number
|
|
current_log_level = logger.getEffectiveLevel()
|
|
|
|
# Only track memory usage when log level is TRACE or more verbose or if running in DEBUG mode
|
|
should_log_memory = current_log_level <= 5 or settings.DEBUG
|
|
|
|
if should_log_memory:
|
|
process = psutil.Process()
|
|
initial_memory = process.memory_info().rss / 1024 / 1024
|
|
logger.info(f"[parse_programs_for_tvg_id] Initial memory usage: {initial_memory:.2f} MB")
|
|
mem_before = initial_memory
|
|
except ImportError:
|
|
process = None
|
|
should_log_memory = False
|
|
|
|
# Reuse the EPGData fetched at the top of the task (epg_source is already
|
|
# cached via select_related) instead of issuing a second query mid-task.
|
|
epg = epg_obj
|
|
|
|
if not force and not Channel.objects.filter(epg_data=epg).exists():
|
|
logger.info(f"No channels matched to EPG {epg.tvg_id}")
|
|
return
|
|
|
|
logger.info(f"Refreshing program data for tvg_id: {epg.tvg_id}")
|
|
|
|
file_path = epg_source.extracted_file_path if epg_source.extracted_file_path else epg_source.file_path
|
|
if not file_path:
|
|
file_path = epg_source.get_cache_file()
|
|
|
|
# Check if the file exists
|
|
if not os.path.exists(file_path):
|
|
logger.error(f"EPG file not found at: {file_path}")
|
|
|
|
if epg_source.url:
|
|
# Update the file path in the database
|
|
new_path = epg_source.get_cache_file()
|
|
logger.info(f"Updating file_path from '{file_path}' to '{new_path}'")
|
|
epg_source.file_path = new_path
|
|
epg_source.save(update_fields=['file_path'])
|
|
logger.info(f"Fetching new EPG data from URL: {epg_source.url}")
|
|
else:
|
|
logger.info(f"EPG source does not have a URL, using existing file path: {file_path} to rebuild cache")
|
|
|
|
# Fetch new data before continuing
|
|
if epg_source:
|
|
|
|
# Properly check the return value from fetch_xmltv
|
|
fetch_success = fetch_xmltv(epg_source)
|
|
|
|
# If fetch was not successful or the file still doesn't exist, abort
|
|
if not fetch_success:
|
|
logger.error(f"Failed to fetch EPG data, cannot parse programs for tvg_id: {epg.tvg_id}")
|
|
# Update status to error if not already set
|
|
epg_source.status = 'error'
|
|
epg_source.last_message = f"Failed to download EPG data, cannot parse programs"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="Failed to download EPG file")
|
|
return
|
|
|
|
# Also check if the file exists after download
|
|
if not os.path.exists(epg_source.file_path):
|
|
logger.error(f"Failed to fetch EPG data, file still missing at: {epg_source.file_path}")
|
|
epg_source.status = 'error'
|
|
epg_source.last_message = f"Failed to download EPG data, file missing after download"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="File not found after download")
|
|
return
|
|
|
|
# Update file_path with the new location
|
|
if epg_source.extracted_file_path:
|
|
file_path = epg_source.extracted_file_path
|
|
else:
|
|
file_path = epg_source.file_path
|
|
else:
|
|
logger.error(f"No URL provided for EPG source {epg_source.name}, cannot fetch new data")
|
|
# Update status to error
|
|
epg_source.status = 'error'
|
|
epg_source.last_message = f"No URL provided, cannot fetch EPG data"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="No URL provided")
|
|
return
|
|
|
|
# Use streaming parsing to reduce memory usage
|
|
# No need to check file type anymore since it's always XML
|
|
logger.debug(f"Parsing programs for tvg_id={epg.tvg_id} from {file_path}")
|
|
|
|
# Memory usage tracking
|
|
if process:
|
|
try:
|
|
mem_before = process.memory_info().rss / 1024 / 1024
|
|
logger.debug(f"[parse_programs_for_tvg_id] Memory before parsing {epg.tvg_id} - {mem_before:.2f} MB")
|
|
except Exception as e:
|
|
logger.warning(f"Error tracking memory: {e}")
|
|
mem_before = 0
|
|
|
|
# Accumulate all of this channel's programmes in memory and swap them in
|
|
# atomically at the end (delete-old + insert-new in one transaction), so a
|
|
# mid-parse failure leaves the previous guide data intact instead of the
|
|
# channel ending up with nothing. Per-channel volume is small enough
|
|
# (weeks of one channel's schedule) that this doesn't need batched flushing
|
|
# to the DB the way the whole-source bulk parse does.
|
|
programs_to_create = []
|
|
|
|
try:
|
|
# Open the file directly - no need to check compression
|
|
logger.debug(f"Opening file for parsing: {file_path}")
|
|
source_file = _open_xmltv_file(file_path)
|
|
|
|
# Stream parse the file using lxml's iterparse
|
|
program_parser = etree.iterparse(source_file, events=('end',), tag='programme', remove_blank_text=True, recover=True)
|
|
|
|
for _, elem in program_parser:
|
|
if elem.get('channel') == epg.tvg_id:
|
|
try:
|
|
start_time = parse_xmltv_time(elem.get('start'))
|
|
end_time = parse_xmltv_time(elem.get('stop'))
|
|
title = None
|
|
desc = None
|
|
sub_title = None
|
|
|
|
# Efficiently process child elements
|
|
for child in elem:
|
|
if child.tag == 'title':
|
|
title = child.text or 'No Title'
|
|
elif child.tag == 'desc':
|
|
desc = child.text or ''
|
|
elif child.tag == 'sub-title':
|
|
sub_title = child.text or ''
|
|
|
|
if not title:
|
|
title = 'No Title'
|
|
|
|
# Extract custom properties
|
|
custom_props = extract_custom_properties(elem)
|
|
custom_properties_json = None
|
|
|
|
if custom_props:
|
|
logger.trace(f"Number of custom properties: {len(custom_props)}")
|
|
custom_properties_json = custom_props
|
|
|
|
# Fallback: extract S/E from description when episode-num
|
|
# elements didn't provide them
|
|
if desc:
|
|
has_season = (custom_properties_json or {}).get('season') is not None
|
|
has_episode = (custom_properties_json or {}).get('episode') is not None
|
|
if not has_season or not has_episode:
|
|
d_season, d_episode, cleaned_desc = extract_season_episode_from_description(desc)
|
|
if d_season is not None and d_episode is not None:
|
|
if custom_properties_json is None:
|
|
custom_properties_json = {}
|
|
if not has_season:
|
|
custom_properties_json['season'] = d_season
|
|
if not has_episode:
|
|
custom_properties_json['episode'] = d_episode
|
|
custom_properties_json['season_episode_source'] = 'description'
|
|
desc = cleaned_desc
|
|
|
|
programs_to_create.append(ProgramData(
|
|
epg=epg,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
title=title[:255],
|
|
description=desc,
|
|
sub_title=sub_title,
|
|
tvg_id=epg.tvg_id,
|
|
custom_properties=custom_properties_json
|
|
))
|
|
programs_processed += 1
|
|
# Clear the element to free memory
|
|
clear_element(elem)
|
|
if programs_processed % 5000 == 0:
|
|
gc.collect()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing program for {epg.tvg_id}: {e}", exc_info=True)
|
|
else:
|
|
# Immediately clean up non-matching elements to reduce memory pressure
|
|
if elem is not None:
|
|
clear_element(elem)
|
|
continue
|
|
|
|
# Make sure to close the file and release parser resources
|
|
if source_file:
|
|
source_file.close()
|
|
source_file = None
|
|
|
|
if program_parser:
|
|
program_parser = None
|
|
|
|
gc.collect()
|
|
|
|
except zipfile.BadZipFile as zip_error:
|
|
logger.error(f"Bad ZIP file: {zip_error}")
|
|
raise
|
|
except etree.XMLSyntaxError as xml_error:
|
|
logger.error(f"XML syntax error parsing program data: {xml_error}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error parsing XML for programs: {e}", exc_info=True)
|
|
raise
|
|
finally:
|
|
# Ensure file is closed even if an exception occurs
|
|
if source_file:
|
|
source_file.close()
|
|
source_file = None
|
|
# Memory tracking after processing
|
|
if process:
|
|
try:
|
|
mem_after = process.memory_info().rss / 1024 / 1024
|
|
logger.info(f"[parse_programs_for_tvg_id] Memory after parsing 1 {epg.tvg_id} - {programs_processed} programs: {mem_after:.2f} MB (change: {mem_after-mem_before:.2f} MB)")
|
|
except Exception as e:
|
|
logger.warning(f"Error tracking memory: {e}")
|
|
|
|
# Swap old programmes for the newly parsed ones atomically. If the insert
|
|
# fails (including a poisoned-connection blip), the transaction rolls back
|
|
# and the previous guide data for this channel is left untouched.
|
|
with transaction.atomic():
|
|
deleted_count = ProgramData.objects.filter(epg=epg).delete()[0]
|
|
if programs_to_create:
|
|
for i in range(0, len(programs_to_create), _EPG_SWAP_BATCH_SIZE):
|
|
ProgramData.objects.bulk_create(
|
|
programs_to_create[i:i + _EPG_SWAP_BATCH_SIZE]
|
|
)
|
|
logger.debug(
|
|
f"Replaced {deleted_count} program(s) with {len(programs_to_create)} "
|
|
f"for {epg.tvg_id}"
|
|
)
|
|
programs_to_create = None
|
|
custom_props = None
|
|
custom_properties_json = None
|
|
|
|
logger.info(f"Completed program parsing for tvg_id={epg.tvg_id}.")
|
|
finally:
|
|
# Reset internal caches and pools that lxml might be keeping
|
|
try:
|
|
etree.clear_error_log()
|
|
except:
|
|
pass
|
|
# Explicit cleanup of all potentially large objects
|
|
if source_file:
|
|
try:
|
|
source_file.close()
|
|
except:
|
|
pass
|
|
source_file = None
|
|
program_parser = None
|
|
programs_to_create = None
|
|
|
|
epg_source = None
|
|
# Add comprehensive cleanup before releasing lock
|
|
cleanup_memory(log_usage=should_log_memory, force_collection=True)
|
|
# Memory tracking after processing
|
|
if process:
|
|
try:
|
|
mem_after = process.memory_info().rss / 1024 / 1024
|
|
logger.info(f"[parse_programs_for_tvg_id] Final memory usage {epg.tvg_id} - {programs_processed} programs: {mem_after:.2f} MB (change: {mem_after-mem_before:.2f} MB)")
|
|
except Exception as e:
|
|
logger.warning(f"Error tracking memory: {e}")
|
|
process = None
|
|
epg = None
|
|
programs_processed = None
|
|
lock_renewer.stop()
|
|
release_task_lock('parse_epg_programs', epg_id)
|
|
finally:
|
|
_decr_source_tvg_parse_count(source_id)
|
|
|
|
|
|
_EPG_PROGRAM_STAGING_TABLE = 'epg_program_staging'
|
|
# Parse batches bound Python memory during XML iterparse; swap batches bound each
|
|
# DELETE/INSERT statement inside the single atomic swap transaction.
|
|
_EPG_PARSE_BATCH_SIZE = 2500
|
|
_EPG_SWAP_BATCH_SIZE = 5000
|
|
|
|
|
|
def _epg_program_staging_supported():
|
|
return connection.vendor == 'postgresql'
|
|
|
|
|
|
def _prepare_epg_program_staging_table():
|
|
"""Create/truncate a session-scoped temp table for streaming EPG programme inserts."""
|
|
if not _epg_program_staging_supported():
|
|
return False
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
f"""
|
|
CREATE TEMP TABLE IF NOT EXISTS {_EPG_PROGRAM_STAGING_TABLE} (
|
|
epg_id bigint NOT NULL,
|
|
start_time timestamptz NOT NULL,
|
|
end_time timestamptz NOT NULL,
|
|
title varchar(255) NOT NULL,
|
|
sub_title text,
|
|
description text,
|
|
tvg_id varchar(255),
|
|
custom_properties jsonb
|
|
) ON COMMIT PRESERVE ROWS
|
|
"""
|
|
)
|
|
cursor.execute(f"TRUNCATE {_EPG_PROGRAM_STAGING_TABLE}")
|
|
return True
|
|
|
|
|
|
def _clear_epg_program_staging_table():
|
|
if not _epg_program_staging_supported():
|
|
return
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(f"TRUNCATE {_EPG_PROGRAM_STAGING_TABLE}")
|
|
|
|
|
|
def _flush_epg_program_staging_batch(programs_batch):
|
|
"""Insert a batch of unsaved ProgramData rows into the session staging table."""
|
|
if not programs_batch or not _epg_program_staging_supported():
|
|
return
|
|
|
|
values_sql = []
|
|
params = []
|
|
for program in programs_batch:
|
|
values_sql.append("(%s, %s, %s, %s, %s, %s, %s, %s)")
|
|
custom_properties = program.custom_properties
|
|
if custom_properties is not None and not isinstance(custom_properties, str):
|
|
custom_properties = json.dumps(custom_properties)
|
|
params.extend([
|
|
program.epg_id,
|
|
program.start_time,
|
|
program.end_time,
|
|
program.title,
|
|
program.sub_title,
|
|
program.description,
|
|
program.tvg_id,
|
|
custom_properties,
|
|
])
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
f"""
|
|
INSERT INTO {_EPG_PROGRAM_STAGING_TABLE} (
|
|
epg_id, start_time, end_time, title, sub_title, description, tvg_id, custom_properties
|
|
) VALUES {', '.join(values_sql)}
|
|
""",
|
|
params,
|
|
)
|
|
|
|
|
|
def _epg_ids_mapped_to_channels(epg_source):
|
|
"""EPGData ids currently assigned to at least one channel on this source."""
|
|
return set(
|
|
Channel.objects.filter(
|
|
epg_data__epg_source=epg_source,
|
|
epg_data__isnull=False,
|
|
).values_list('epg_data_id', flat=True)
|
|
)
|
|
|
|
|
|
def _delete_orphaned_epg_programs(epg_source):
|
|
"""
|
|
Remove programme rows for EPG entries that have no channel mapped now.
|
|
|
|
Uses live channel assignments, not the snapshot taken at bulk-parse start,
|
|
so channels matched mid-refresh are not treated as orphaned.
|
|
"""
|
|
currently_mapped = _epg_ids_mapped_to_channels(epg_source)
|
|
unmapped_epg_ids = list(
|
|
EPGData.objects.filter(epg_source=epg_source)
|
|
.exclude(id__in=currently_mapped)
|
|
.values_list('id', flat=True)
|
|
)
|
|
if not unmapped_epg_ids:
|
|
return 0
|
|
orphaned_count = ProgramData.objects.filter(epg_id__in=unmapped_epg_ids).delete()[0]
|
|
if orphaned_count > 0:
|
|
logger.info(
|
|
f"Cleaned up {orphaned_count} orphaned programs for "
|
|
f"{len(unmapped_epg_ids)} unmapped EPG entries"
|
|
)
|
|
return orphaned_count
|
|
|
|
|
|
def _dispatch_late_mapped_epg_parses(epg_source, bulk_parsed_epg_ids):
|
|
"""
|
|
Queue per-channel parses for EPG rows mapped after bulk parse began.
|
|
|
|
Per-channel tasks triggered during refresh defer until refresh finishes;
|
|
this ensures they still run promptly even if a deferred retry is pending.
|
|
"""
|
|
missed_epg_ids = _epg_ids_mapped_to_channels(epg_source) - bulk_parsed_epg_ids
|
|
if not missed_epg_ids:
|
|
return 0
|
|
logger.info(
|
|
f"Queueing per-channel program parse for {len(missed_epg_ids)} EPG row(s) "
|
|
f"mapped during bulk refresh of {epg_source.name}"
|
|
)
|
|
return dispatch_program_refresh_for_epg_ids(missed_epg_ids)
|
|
|
|
|
|
def _swap_staged_epg_programs(mapped_epg_ids, epg_source, batch_size=_EPG_SWAP_BATCH_SIZE):
|
|
"""
|
|
Atomically replace mapped programme rows with staged data.
|
|
Must be called inside transaction.atomic().
|
|
|
|
Staged rows are moved in batches (DELETE ... RETURNING + INSERT) so Postgres
|
|
does not need to materialize the entire catalogue in one statement.
|
|
"""
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SET LOCAL statement_timeout = '10min'")
|
|
|
|
deleted_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()[0]
|
|
logger.debug(f"Deleted {deleted_count} existing programs")
|
|
|
|
_delete_orphaned_epg_programs(epg_source)
|
|
|
|
if not _epg_program_staging_supported():
|
|
raise RuntimeError('_swap_staged_epg_programs requires PostgreSQL staging support')
|
|
|
|
program_table = ProgramData._meta.db_table
|
|
total_inserted = 0
|
|
while True:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
f"""
|
|
WITH moved AS (
|
|
DELETE FROM {_EPG_PROGRAM_STAGING_TABLE}
|
|
WHERE ctid IN (
|
|
SELECT ctid FROM {_EPG_PROGRAM_STAGING_TABLE} LIMIT %s
|
|
)
|
|
RETURNING
|
|
epg_id, start_time, end_time, title, sub_title,
|
|
description, tvg_id, custom_properties
|
|
)
|
|
INSERT INTO {program_table} (
|
|
epg_id, start_time, end_time, title, sub_title,
|
|
description, tvg_id, custom_properties
|
|
)
|
|
SELECT
|
|
epg_id, start_time, end_time, title, sub_title,
|
|
description, tvg_id, custom_properties
|
|
FROM moved
|
|
""",
|
|
[batch_size],
|
|
)
|
|
moved_count = cursor.rowcount
|
|
if moved_count == 0:
|
|
break
|
|
total_inserted += moved_count
|
|
|
|
logger.debug(f"Inserted {total_inserted} staged programs in batches of {batch_size}")
|
|
|
|
return deleted_count
|
|
|
|
|
|
def _swap_parsed_epg_programs(mapped_epg_ids, epg_source, programs_to_create, batch_size=_EPG_SWAP_BATCH_SIZE):
|
|
"""SQLite/dev fallback: atomic delete + bulk insert from an in-memory batch list."""
|
|
with transaction.atomic():
|
|
deleted_count = ProgramData.objects.filter(epg_id__in=mapped_epg_ids).delete()[0]
|
|
_delete_orphaned_epg_programs(epg_source)
|
|
for i in range(0, len(programs_to_create), batch_size):
|
|
ProgramData.objects.bulk_create(programs_to_create[i:i + batch_size])
|
|
return deleted_count
|
|
|
|
|
|
def parse_programs_for_source(epg_source, tvg_id=None):
|
|
"""
|
|
Parse programs for all MAPPED channels from an EPG source in a single pass.
|
|
|
|
This is an optimized version that:
|
|
1. Only processes EPG entries that are actually mapped to channels
|
|
2. Parses the XML file ONCE instead of once per channel
|
|
3. Skips programmes for unmapped channels entirely during parsing
|
|
|
|
This dramatically improves performance when an EPG source has many channels
|
|
but only a fraction are mapped.
|
|
"""
|
|
# Send initial programs parsing notification
|
|
send_epg_update(epg_source.id, "parsing_programs", 0)
|
|
should_log_memory = False
|
|
process = None
|
|
initial_memory = 0
|
|
source_file = None
|
|
|
|
# Add memory tracking only in trace mode or higher
|
|
try:
|
|
# Get current log level as a number
|
|
current_log_level = logger.getEffectiveLevel()
|
|
|
|
# Only track memory usage when log level is TRACE or more verbose
|
|
should_log_memory = current_log_level <= 5 or settings.DEBUG # Assuming TRACE is level 5 or lower
|
|
|
|
if should_log_memory:
|
|
process = psutil.Process()
|
|
initial_memory = process.memory_info().rss / 1024 / 1024
|
|
logger.info(f"[parse_programs_for_source] Initial memory usage: {initial_memory:.2f} MB")
|
|
except ImportError:
|
|
logger.warning("psutil not available for memory tracking")
|
|
process = None
|
|
should_log_memory = False
|
|
|
|
try:
|
|
# Only get EPG entries that are actually mapped to channels
|
|
mapped_epg_ids = set(
|
|
Channel.objects.filter(
|
|
epg_data__epg_source=epg_source,
|
|
epg_data__isnull=False
|
|
).values_list('epg_data_id', flat=True)
|
|
)
|
|
|
|
if not mapped_epg_ids:
|
|
total_epg_count = EPGData.objects.filter(epg_source=epg_source).count()
|
|
logger.info(f"No channels mapped to any EPG entries from source: {epg_source.name} "
|
|
f"(source has {total_epg_count} EPG entries, 0 mapped)")
|
|
# Update status - this is not an error, just no mapped entries
|
|
epg_source.status = 'success'
|
|
epg_source.last_message = f"No channels mapped to this EPG source ({total_epg_count} entries available)"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100, status="success")
|
|
return True
|
|
|
|
# Get the mapped EPG entries with their tvg_ids
|
|
mapped_epgs = EPGData.objects.filter(id__in=mapped_epg_ids).values('id', 'tvg_id')
|
|
tvg_id_to_epg_id = {epg['tvg_id']: epg['id'] for epg in mapped_epgs if epg['tvg_id']}
|
|
mapped_tvg_ids = set(tvg_id_to_epg_id.keys())
|
|
|
|
total_epg_count = EPGData.objects.filter(epg_source=epg_source).count()
|
|
mapped_count = len(mapped_tvg_ids)
|
|
|
|
logger.info(f"Parsing programs for {mapped_count} MAPPED channels from source: {epg_source.name} "
|
|
f"(skipping {total_epg_count - mapped_count} unmapped EPG entries)")
|
|
|
|
# Get the file path
|
|
file_path = epg_source.extracted_file_path if epg_source.extracted_file_path else epg_source.file_path
|
|
if not file_path:
|
|
file_path = epg_source.get_cache_file()
|
|
|
|
# Check if the file exists
|
|
if not os.path.exists(file_path):
|
|
logger.error(f"EPG file not found at: {file_path}")
|
|
|
|
if epg_source.url:
|
|
# Update the file path in the database
|
|
new_path = epg_source.get_cache_file()
|
|
logger.info(f"Updating file_path from '{file_path}' to '{new_path}'")
|
|
epg_source.file_path = new_path
|
|
epg_source.save(update_fields=['file_path'])
|
|
logger.info(f"Fetching new EPG data from URL: {epg_source.url}")
|
|
|
|
# Fetch new data before continuing
|
|
fetch_success = fetch_xmltv(epg_source)
|
|
|
|
if not fetch_success:
|
|
logger.error(f"Failed to fetch EPG data for source: {epg_source.name}")
|
|
epg_source.status = 'error'
|
|
epg_source.last_message = f"Failed to download EPG data"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="Failed to download EPG file")
|
|
return False
|
|
|
|
# Update file_path with the new location
|
|
file_path = epg_source.extracted_file_path if epg_source.extracted_file_path else epg_source.file_path
|
|
else:
|
|
logger.error(f"No URL provided for EPG source {epg_source.name}, cannot fetch new data")
|
|
epg_source.status = 'error'
|
|
epg_source.last_message = f"No URL provided, cannot fetch EPG data"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="No URL provided")
|
|
return False
|
|
|
|
# Stream parsed rows into a session temp table, then swap in a short transaction.
|
|
# This bounds Python memory (batched staging inserts) and Postgres memory (no
|
|
# long-lived transaction spanning the entire XML parse).
|
|
programs_by_channel = {tvg_id: 0 for tvg_id in mapped_tvg_ids}
|
|
total_programs = 0
|
|
skipped_programs = 0
|
|
last_progress_update = 0
|
|
parse_batch_size = _EPG_PARSE_BATCH_SIZE
|
|
swap_batch_size = _EPG_SWAP_BATCH_SIZE
|
|
programs_batch = []
|
|
deleted_count = 0
|
|
staging_prepared = False
|
|
use_staging = False
|
|
programs_accumulator = []
|
|
|
|
send_epg_update(epg_source.id, "parsing_programs", 10, message="Parsing programs...")
|
|
|
|
try:
|
|
staging_prepared = _prepare_epg_program_staging_table()
|
|
use_staging = staging_prepared
|
|
|
|
logger.debug(f"Opening file for streaming parse: {file_path}")
|
|
source_file = _open_xmltv_file(file_path)
|
|
try:
|
|
program_parser = etree.iterparse(
|
|
source_file,
|
|
events=('end',),
|
|
tag='programme',
|
|
remove_blank_text=True,
|
|
recover=True,
|
|
)
|
|
|
|
for _, elem in program_parser:
|
|
channel_id = elem.get('channel')
|
|
|
|
if channel_id not in mapped_tvg_ids:
|
|
skipped_programs += 1
|
|
clear_element(elem)
|
|
continue
|
|
|
|
try:
|
|
start_time = parse_xmltv_time(elem.get('start'))
|
|
end_time = parse_xmltv_time(elem.get('stop'))
|
|
title = None
|
|
desc = None
|
|
sub_title = None
|
|
|
|
for child in elem:
|
|
if child.tag == 'title':
|
|
title = child.text or 'No Title'
|
|
elif child.tag == 'desc':
|
|
desc = child.text or ''
|
|
elif child.tag == 'sub-title':
|
|
sub_title = child.text or ''
|
|
|
|
if not title:
|
|
title = 'No Title'
|
|
|
|
custom_props = extract_custom_properties(elem)
|
|
custom_properties_json = custom_props if custom_props else None
|
|
|
|
if desc:
|
|
has_season = (custom_properties_json or {}).get('season') is not None
|
|
has_episode = (custom_properties_json or {}).get('episode') is not None
|
|
if not has_season or not has_episode:
|
|
d_season, d_episode, cleaned_desc = extract_season_episode_from_description(desc)
|
|
if d_season is not None and d_episode is not None:
|
|
if custom_properties_json is None:
|
|
custom_properties_json = {}
|
|
if not has_season:
|
|
custom_properties_json['season'] = d_season
|
|
if not has_episode:
|
|
custom_properties_json['episode'] = d_episode
|
|
custom_properties_json['season_episode_source'] = 'description'
|
|
desc = cleaned_desc
|
|
|
|
epg_id = tvg_id_to_epg_id[channel_id]
|
|
programs_batch.append(ProgramData(
|
|
epg_id=epg_id,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
title=title[:255],
|
|
description=desc,
|
|
sub_title=sub_title,
|
|
tvg_id=channel_id,
|
|
custom_properties=custom_properties_json,
|
|
))
|
|
total_programs += 1
|
|
programs_by_channel[channel_id] += 1
|
|
clear_element(elem)
|
|
|
|
if len(programs_batch) >= parse_batch_size:
|
|
if use_staging:
|
|
_flush_epg_program_staging_batch(programs_batch)
|
|
programs_batch = []
|
|
else:
|
|
programs_accumulator.extend(programs_batch)
|
|
programs_batch = []
|
|
|
|
if total_programs - last_progress_update >= 5000:
|
|
last_progress_update = total_programs
|
|
progress = min(
|
|
85,
|
|
10 + int((total_programs / max(total_programs + 10000, 1)) * 75),
|
|
)
|
|
send_epg_update(
|
|
epg_source.id,
|
|
"parsing_programs",
|
|
progress,
|
|
processed=total_programs,
|
|
channels=mapped_count,
|
|
message=f"Staging programs... {total_programs:,}",
|
|
)
|
|
|
|
if total_programs % 5000 == 0:
|
|
gc.collect()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing program for {channel_id}: {e}", exc_info=True)
|
|
clear_element(elem)
|
|
continue
|
|
|
|
if programs_batch:
|
|
if use_staging:
|
|
_flush_epg_program_staging_batch(programs_batch)
|
|
else:
|
|
programs_accumulator.extend(programs_batch)
|
|
programs_batch = []
|
|
finally:
|
|
if source_file:
|
|
source_file.close()
|
|
source_file = None
|
|
|
|
try:
|
|
send_epg_update(epg_source.id, "parsing_programs", 90, message="Updating database...")
|
|
if use_staging:
|
|
with transaction.atomic():
|
|
deleted_count = _swap_staged_epg_programs(
|
|
mapped_epg_ids, epg_source, batch_size=swap_batch_size
|
|
)
|
|
else:
|
|
deleted_count = _swap_parsed_epg_programs(
|
|
mapped_epg_ids, epg_source, programs_accumulator, batch_size=swap_batch_size
|
|
)
|
|
programs_accumulator = []
|
|
|
|
logger.info(
|
|
f"Atomic update complete: deleted {deleted_count}, inserted {total_programs} programs"
|
|
)
|
|
except Exception as db_error:
|
|
logger.error(f"Database error during atomic update: {db_error}", exc_info=True)
|
|
epg_source.status = EPGSource.STATUS_ERROR
|
|
epg_source.last_message = f"Database error: {str(db_error)}"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(
|
|
epg_source.id, "parsing_programs", 100, status="error", message=str(db_error)
|
|
)
|
|
return False
|
|
|
|
except etree.XMLSyntaxError as xml_error:
|
|
logger.error(f"XML syntax error parsing program data: {xml_error}")
|
|
epg_source.status = EPGSource.STATUS_ERROR
|
|
epg_source.last_message = f"XML parsing error: {str(xml_error)}"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", message=str(xml_error))
|
|
return False
|
|
except Exception as parse_error:
|
|
logger.error(f"Error parsing programs from XML: {parse_error}", exc_info=True)
|
|
epg_source.status = EPGSource.STATUS_ERROR
|
|
epg_source.last_message = f"Error parsing programs: {str(parse_error)}"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(
|
|
epg_source.id, "parsing_programs", 100, status="error", message=str(parse_error)
|
|
)
|
|
return False
|
|
finally:
|
|
programs_batch = None
|
|
programs_accumulator = None
|
|
if staging_prepared:
|
|
try:
|
|
_clear_epg_program_staging_table()
|
|
except Exception:
|
|
pass
|
|
gc.collect()
|
|
|
|
# Count channels that actually got programs
|
|
channels_with_programs = sum(1 for count in programs_by_channel.values() if count > 0)
|
|
|
|
# Success message
|
|
epg_source.status = EPGSource.STATUS_SUCCESS
|
|
epg_source.last_message = (
|
|
f"Parsed {total_programs:,} programs for {channels_with_programs} channels "
|
|
f"(skipped {skipped_programs:,} programs for {total_epg_count - mapped_count} unmapped channels)"
|
|
)
|
|
epg_source.updated_at = timezone.now()
|
|
epg_source.save(update_fields=['status', 'last_message', 'updated_at'])
|
|
|
|
# Log system event for EPG refresh
|
|
log_system_event(
|
|
event_type='epg_refresh',
|
|
source_name=epg_source.name,
|
|
programs=total_programs,
|
|
channels=channels_with_programs,
|
|
skipped_programs=skipped_programs,
|
|
unmapped_channels=total_epg_count - mapped_count,
|
|
)
|
|
|
|
# Send completion notification with status
|
|
send_epg_update(epg_source.id, "parsing_programs", 100,
|
|
status="success",
|
|
message=epg_source.last_message,
|
|
updated_at=epg_source.updated_at.isoformat())
|
|
|
|
logger.info(f"Completed parsing programs for source: {epg_source.name} - "
|
|
f"{total_programs:,} programs for {channels_with_programs} channels, "
|
|
f"skipped {skipped_programs:,} programs for unmapped channels")
|
|
_dispatch_late_mapped_epg_parses(epg_source, mapped_epg_ids)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in parse_programs_for_source: {e}", exc_info=True)
|
|
# Update status to error
|
|
epg_source.status = EPGSource.STATUS_ERROR
|
|
epg_source.last_message = f"Error parsing programs: {str(e)}"
|
|
epg_source.save(update_fields=['status', 'last_message'])
|
|
send_epg_update(epg_source.id, "parsing_programs", 100,
|
|
status="error",
|
|
message=epg_source.last_message)
|
|
return False
|
|
finally:
|
|
# Final memory cleanup and tracking
|
|
if source_file:
|
|
try:
|
|
source_file.close()
|
|
except:
|
|
pass
|
|
source_file = None
|
|
|
|
# Explicitly release any remaining large data structures
|
|
programs_batch = None
|
|
programs_by_channel = None
|
|
mapped_epg_ids = None
|
|
mapped_tvg_ids = None
|
|
tvg_id_to_epg_id = None
|
|
gc.collect()
|
|
|
|
# Add comprehensive memory cleanup at the end
|
|
cleanup_memory(log_usage=should_log_memory, force_collection=True)
|
|
if process:
|
|
final_memory = process.memory_info().rss / 1024 / 1024
|
|
logger.info(f"[parse_programs_for_source] Final memory usage: {final_memory:.2f} MB difference: {final_memory - initial_memory:.2f} MB")
|
|
# Explicitly clear the process object to prevent potential memory leaks
|
|
process = None
|
|
|
|
|
|
def dispatch_program_refresh_for_epg_ids(epg_ids):
|
|
"""
|
|
Queue guide/program refresh for newly assigned EPGData rows.
|
|
|
|
XMLTV and other non-SD sources use parse_programs_for_tvg_id per id.
|
|
Schedules Direct uses per-EPG fetches for small batches and one batched
|
|
mapped-station fetch per source when the threshold is exceeded.
|
|
"""
|
|
if not epg_ids:
|
|
return 0
|
|
|
|
epg_ids = {eid for eid in epg_ids if eid}
|
|
if not epg_ids:
|
|
return 0
|
|
|
|
epgs = list(
|
|
EPGData.objects.filter(id__in=epg_ids).select_related('epg_source')
|
|
)
|
|
epgs_with_program_data = set(
|
|
ProgramData.objects.filter(epg_id__in=epg_ids)
|
|
.values_list('epg_id', flat=True)
|
|
.distinct()
|
|
)
|
|
|
|
non_sd_epg_ids = []
|
|
sd_by_source = {}
|
|
for epg in epgs:
|
|
if not epg.epg_source:
|
|
non_sd_epg_ids.append(epg.id)
|
|
continue
|
|
source_type = epg.epg_source.source_type
|
|
if source_type == 'dummy':
|
|
continue
|
|
if source_type == 'schedules_direct':
|
|
sd_by_source.setdefault(epg.epg_source_id, []).append(epg)
|
|
else:
|
|
non_sd_epg_ids.append(epg.id)
|
|
|
|
dispatched = 0
|
|
for epg_id in non_sd_epg_ids:
|
|
parse_programs_for_tvg_id.delay(epg_id)
|
|
dispatched += 1
|
|
|
|
for source_id, source_epgs in sd_by_source.items():
|
|
needs_fetch = [
|
|
epg for epg in source_epgs
|
|
if epg.id not in epgs_with_program_data
|
|
]
|
|
if not needs_fetch:
|
|
continue
|
|
if len(needs_fetch) >= SD_BULK_GUIDE_FETCH_THRESHOLD:
|
|
logger.info(
|
|
f"SD source {source_id}: {len(needs_fetch)} new mapping(s) exceed "
|
|
f"threshold ({SD_BULK_GUIDE_FETCH_THRESHOLD}); "
|
|
"queueing batched mapped guide fetch"
|
|
)
|
|
fetch_sd_mapped_guide_batch.delay(source_id)
|
|
dispatched += 1
|
|
else:
|
|
for epg in needs_fetch:
|
|
parse_programs_for_tvg_id.delay(epg.id)
|
|
dispatched += 1
|
|
|
|
return dispatched
|
|
|
|
|
|
def parse_xmltv_time(time_str):
|
|
try:
|
|
# Basic format validation
|
|
if len(time_str) < 14:
|
|
logger.warning(f"XMLTV timestamp too short: '{time_str}', using as-is")
|
|
dt_obj = datetime.strptime(time_str, '%Y%m%d%H%M%S')
|
|
return timezone.make_aware(dt_obj, timezone=dt_timezone.utc)
|
|
|
|
# Parse base datetime
|
|
dt_obj = datetime.strptime(time_str[:14], '%Y%m%d%H%M%S')
|
|
|
|
# Handle timezone if present
|
|
if len(time_str) >= 20: # Has timezone info
|
|
tz_sign = time_str[15]
|
|
tz_hours = int(time_str[16:18])
|
|
tz_minutes = int(time_str[18:20])
|
|
|
|
# Create a timezone object
|
|
if tz_sign == '+':
|
|
tz_offset = dt_timezone(timedelta(hours=tz_hours, minutes=tz_minutes))
|
|
elif tz_sign == '-':
|
|
tz_offset = dt_timezone(timedelta(hours=-tz_hours, minutes=-tz_minutes))
|
|
else:
|
|
tz_offset = dt_timezone.utc
|
|
|
|
# Make datetime aware with correct timezone
|
|
aware_dt = datetime.replace(dt_obj, tzinfo=tz_offset)
|
|
# Convert to UTC
|
|
aware_dt = aware_dt.astimezone(dt_timezone.utc)
|
|
|
|
logger.trace(f"Parsed XMLTV time '{time_str}' to {aware_dt}")
|
|
return aware_dt
|
|
else:
|
|
# No timezone info, assume UTC
|
|
aware_dt = timezone.make_aware(dt_obj, timezone=dt_timezone.utc)
|
|
logger.trace(f"Parsed XMLTV time without timezone '{time_str}' as UTC: {aware_dt}")
|
|
return aware_dt
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error parsing XMLTV time '{time_str}': {e}", exc_info=True)
|
|
raise
|
|
|
|
|
|
def extract_custom_properties(prog):
|
|
# Create a new dictionary for each call
|
|
custom_props = {}
|
|
|
|
# Extract categories with a single comprehension to reduce intermediate objects
|
|
categories = [cat.text.strip() for cat in prog.findall('category') if cat.text and cat.text.strip()]
|
|
if categories:
|
|
custom_props['categories'] = categories
|
|
|
|
# Extract keywords (new)
|
|
keywords = [kw.text.strip() for kw in prog.findall('keyword') if kw.text and kw.text.strip()]
|
|
if keywords:
|
|
custom_props['keywords'] = keywords
|
|
|
|
# Extract episode numbers
|
|
for ep_num in prog.findall('episode-num'):
|
|
system = ep_num.get('system', '')
|
|
if system == 'xmltv_ns' and ep_num.text:
|
|
# Parse XMLTV episode-num format (season.episode.part)
|
|
parts = ep_num.text.split('.')
|
|
if len(parts) >= 2:
|
|
if parts[0].strip() != '':
|
|
try:
|
|
season = int(parts[0]) + 1 # XMLTV format is zero-based
|
|
custom_props['season'] = season
|
|
except ValueError:
|
|
pass
|
|
if parts[1].strip() != '':
|
|
try:
|
|
episode = int(parts[1]) + 1 # XMLTV format is zero-based
|
|
custom_props['episode'] = episode
|
|
except ValueError:
|
|
pass
|
|
elif system == 'onscreen' and ep_num.text:
|
|
onscreen_text = ep_num.text.strip()
|
|
custom_props['onscreen_episode'] = onscreen_text
|
|
# Extract season/episode from onscreen format if not already set by xmltv_ns
|
|
if 'season' not in custom_props or 'episode' not in custom_props:
|
|
match = _ONSCREEN_RE.search(onscreen_text)
|
|
if match:
|
|
if 'season' not in custom_props:
|
|
custom_props['season'] = int(match.group(1))
|
|
if 'episode' not in custom_props:
|
|
custom_props['episode'] = int(match.group(2))
|
|
elif system == 'dd_progid' and ep_num.text:
|
|
# Store the dd_progid format
|
|
custom_props['dd_progid'] = ep_num.text.strip()
|
|
# Add support for other systems like thetvdb.com, themoviedb.org, imdb.com
|
|
elif system in ['thetvdb.com', 'themoviedb.org', 'imdb.com'] and ep_num.text:
|
|
custom_props[f'{system}_id'] = ep_num.text.strip()
|
|
|
|
# Extract ratings more efficiently
|
|
rating_elem = prog.find('rating')
|
|
if rating_elem is not None:
|
|
value_elem = rating_elem.find('value')
|
|
if value_elem is not None and value_elem.text:
|
|
custom_props['rating'] = value_elem.text.strip()
|
|
if rating_elem.get('system'):
|
|
custom_props['rating_system'] = rating_elem.get('system')
|
|
|
|
# Extract star ratings (new)
|
|
star_ratings = []
|
|
for star_rating in prog.findall('star-rating'):
|
|
value_elem = star_rating.find('value')
|
|
if value_elem is not None and value_elem.text:
|
|
rating_data = {'value': value_elem.text.strip()}
|
|
if star_rating.get('system'):
|
|
rating_data['system'] = star_rating.get('system')
|
|
star_ratings.append(rating_data)
|
|
if star_ratings:
|
|
custom_props['star_ratings'] = star_ratings
|
|
|
|
# Extract credits more efficiently
|
|
credits_elem = prog.find('credits')
|
|
if credits_elem is not None:
|
|
credits = {}
|
|
for credit_type in ['director', 'actor', 'writer', 'adapter', 'producer', 'composer', 'editor', 'presenter', 'commentator', 'guest']:
|
|
if credit_type == 'actor':
|
|
# Handle actors with roles and guest status
|
|
actors = []
|
|
for actor_elem in credits_elem.findall('actor'):
|
|
if actor_elem.text and actor_elem.text.strip():
|
|
actor_data = {'name': actor_elem.text.strip()}
|
|
if actor_elem.get('role'):
|
|
actor_data['role'] = actor_elem.get('role')
|
|
if actor_elem.get('guest') == 'yes':
|
|
actor_data['guest'] = True
|
|
actors.append(actor_data)
|
|
if actors:
|
|
credits['actor'] = actors
|
|
else:
|
|
names = [e.text.strip() for e in credits_elem.findall(credit_type) if e.text and e.text.strip()]
|
|
if names:
|
|
credits[credit_type] = names
|
|
if credits:
|
|
custom_props['credits'] = credits
|
|
|
|
# Extract other common program metadata
|
|
date_elem = prog.find('date')
|
|
if date_elem is not None and date_elem.text:
|
|
custom_props['date'] = date_elem.text.strip()
|
|
|
|
country_elem = prog.find('country')
|
|
if country_elem is not None and country_elem.text:
|
|
custom_props['country'] = country_elem.text.strip()
|
|
|
|
# Extract language information (new)
|
|
language_elem = prog.find('language')
|
|
if language_elem is not None and language_elem.text:
|
|
custom_props['language'] = language_elem.text.strip()
|
|
|
|
orig_language_elem = prog.find('orig-language')
|
|
if orig_language_elem is not None and orig_language_elem.text:
|
|
custom_props['original_language'] = orig_language_elem.text.strip()
|
|
|
|
# Extract length (new)
|
|
length_elem = prog.find('length')
|
|
if length_elem is not None and length_elem.text:
|
|
try:
|
|
length_value = int(length_elem.text.strip())
|
|
length_units = length_elem.get('units', 'minutes')
|
|
custom_props['length'] = {'value': length_value, 'units': length_units}
|
|
except ValueError:
|
|
pass
|
|
|
|
# Extract video information (new)
|
|
video_elem = prog.find('video')
|
|
if video_elem is not None:
|
|
video_info = {}
|
|
for video_attr in ['present', 'colour', 'aspect', 'quality']:
|
|
attr_elem = video_elem.find(video_attr)
|
|
if attr_elem is not None and attr_elem.text:
|
|
video_info[video_attr] = attr_elem.text.strip()
|
|
if video_info:
|
|
custom_props['video'] = video_info
|
|
|
|
# Extract audio information (new)
|
|
audio_elem = prog.find('audio')
|
|
if audio_elem is not None:
|
|
audio_info = {}
|
|
for audio_attr in ['present', 'stereo']:
|
|
attr_elem = audio_elem.find(audio_attr)
|
|
if attr_elem is not None and attr_elem.text:
|
|
audio_info[audio_attr] = attr_elem.text.strip()
|
|
if audio_info:
|
|
custom_props['audio'] = audio_info
|
|
|
|
# Extract subtitles information (new)
|
|
subtitles = []
|
|
for subtitle_elem in prog.findall('subtitles'):
|
|
subtitle_data = {}
|
|
if subtitle_elem.get('type'):
|
|
subtitle_data['type'] = subtitle_elem.get('type')
|
|
lang_elem = subtitle_elem.find('language')
|
|
if lang_elem is not None and lang_elem.text:
|
|
subtitle_data['language'] = lang_elem.text.strip()
|
|
if subtitle_data:
|
|
subtitles.append(subtitle_data)
|
|
|
|
if subtitles:
|
|
custom_props['subtitles'] = subtitles
|
|
|
|
# Extract reviews (new)
|
|
reviews = []
|
|
for review_elem in prog.findall('review'):
|
|
if review_elem.text and review_elem.text.strip():
|
|
review_data = {'content': review_elem.text.strip()}
|
|
if review_elem.get('type'):
|
|
review_data['type'] = review_elem.get('type')
|
|
if review_elem.get('source'):
|
|
review_data['source'] = review_elem.get('source')
|
|
if review_elem.get('reviewer'):
|
|
review_data['reviewer'] = review_elem.get('reviewer')
|
|
reviews.append(review_data)
|
|
if reviews:
|
|
custom_props['reviews'] = reviews
|
|
|
|
# Extract images (new)
|
|
images = []
|
|
for image_elem in prog.findall('image'):
|
|
if image_elem.text and image_elem.text.strip():
|
|
image_data = {'url': image_elem.text.strip()}
|
|
for attr in ['type', 'size', 'orient', 'system']:
|
|
if image_elem.get(attr):
|
|
image_data[attr] = image_elem.get(attr)
|
|
images.append(image_data)
|
|
if images:
|
|
custom_props['images'] = images
|
|
|
|
icon_elem = prog.find('icon')
|
|
if icon_elem is not None and icon_elem.get('src'):
|
|
custom_props['icon'] = icon_elem.get('src')
|
|
|
|
# Simpler approach for boolean flags - expanded list
|
|
for kw in ['previously-shown', 'premiere', 'new', 'live', 'last-chance']:
|
|
if prog.find(kw) is not None:
|
|
custom_props[kw.replace('-', '_')] = True
|
|
|
|
# Extract premiere and last-chance text content if available
|
|
premiere_elem = prog.find('premiere')
|
|
if premiere_elem is not None:
|
|
custom_props['premiere'] = True
|
|
if premiere_elem.text and premiere_elem.text.strip():
|
|
custom_props['premiere_text'] = premiere_elem.text.strip()
|
|
|
|
last_chance_elem = prog.find('last-chance')
|
|
if last_chance_elem is not None:
|
|
custom_props['last_chance'] = True
|
|
if last_chance_elem.text and last_chance_elem.text.strip():
|
|
custom_props['last_chance_text'] = last_chance_elem.text.strip()
|
|
|
|
# Extract previously-shown details
|
|
prev_shown_elem = prog.find('previously-shown')
|
|
if prev_shown_elem is not None:
|
|
custom_props['previously_shown'] = True
|
|
prev_shown_data = {}
|
|
if prev_shown_elem.get('start'):
|
|
prev_shown_data['start'] = prev_shown_elem.get('start')
|
|
if prev_shown_elem.get('channel'):
|
|
prev_shown_data['channel'] = prev_shown_elem.get('channel')
|
|
if prev_shown_data:
|
|
custom_props['previously_shown_details'] = prev_shown_data
|
|
|
|
return custom_props
|
|
|
|
|
|
def clear_element(elem):
|
|
"""Clear an XML element and its parent to free memory."""
|
|
try:
|
|
elem.clear()
|
|
parent = elem.getparent()
|
|
if parent is not None:
|
|
while elem.getprevious() is not None:
|
|
del parent[0]
|
|
parent.remove(elem)
|
|
except Exception as e:
|
|
logger.warning(f"Error clearing XML element: {e}", exc_info=True)
|
|
|
|
|
|
def detect_file_format(file_path=None, content=None):
|
|
"""
|
|
Detect file format by examining content or file path.
|
|
|
|
Args:
|
|
file_path: Path to file (optional)
|
|
content: Raw file content bytes (optional)
|
|
|
|
Returns:
|
|
tuple: (format_type, is_compressed, file_extension)
|
|
format_type: 'gzip', 'zip', 'xz', 'xml', or 'unknown'
|
|
is_compressed: Boolean indicating if the file is compressed
|
|
file_extension: Appropriate file extension including dot (.gz, .zip, .xz, .xml)
|
|
"""
|
|
# Default return values
|
|
format_type = 'unknown'
|
|
is_compressed = False
|
|
file_extension = '.tmp'
|
|
|
|
# First priority: check content magic numbers as they're most reliable
|
|
if content:
|
|
# We only need the first few bytes for magic number detection
|
|
header = content[:20] if len(content) >= 20 else content
|
|
|
|
# Check for gzip magic number (1f 8b)
|
|
if len(header) >= 2 and header[:2] == b'\x1f\x8b':
|
|
return 'gzip', True, '.gz'
|
|
|
|
# Check for zip magic number (PK..)
|
|
if len(header) >= 2 and header[:2] == b'PK':
|
|
return 'zip', True, '.zip'
|
|
|
|
# Check for xz magic number (fd 37 7a 58 5a 00)
|
|
if len(header) >= 6 and header[:6] == b'\xfd7zXZ\x00':
|
|
return 'xz', True, '.xz'
|
|
|
|
# Check for XML - either standard XML header or XMLTV-specific tag
|
|
if len(header) >= 5 and (b'<?xml' in header or b'<tv>' in header):
|
|
return 'xml', False, '.xml'
|
|
|
|
# Second priority: check file extension - focus on the final extension for compression
|
|
if file_path:
|
|
logger.debug(f"Detecting file format for: {file_path}")
|
|
|
|
# Handle compound extensions like .xml.gz - prioritize compression extensions
|
|
lower_path = file_path.lower()
|
|
|
|
# Check for compression extensions explicitly
|
|
if lower_path.endswith('.gz') or lower_path.endswith('.gzip'):
|
|
return 'gzip', True, '.gz'
|
|
elif lower_path.endswith('.zip'):
|
|
return 'zip', True, '.zip'
|
|
elif lower_path.endswith('.xz'):
|
|
return 'xz', True, '.xz'
|
|
elif lower_path.endswith('.xml'):
|
|
return 'xml', False, '.xml'
|
|
|
|
# Fallback to mimetypes only if direct extension check doesn't work
|
|
import mimetypes
|
|
mime_type, _ = mimetypes.guess_type(file_path)
|
|
logger.debug(f"Guessed MIME type: {mime_type}")
|
|
if mime_type:
|
|
if mime_type == 'application/gzip' or mime_type == 'application/x-gzip':
|
|
return 'gzip', True, '.gz'
|
|
elif mime_type == 'application/zip':
|
|
return 'zip', True, '.zip'
|
|
elif mime_type == 'application/x-xz':
|
|
return 'xz', True, '.xz'
|
|
elif mime_type == 'application/xml' or mime_type == 'text/xml':
|
|
return 'xml', False, '.xml'
|
|
|
|
# If we reach here, we couldn't reliably determine the format
|
|
return format_type, is_compressed, file_extension
|
|
|
|
|
|
def generate_dummy_epg(source):
|
|
"""
|
|
DEPRECATED: This function is no longer used.
|
|
|
|
Dummy EPG programs are now generated on-demand when they are requested
|
|
(during XMLTV export or EPG grid display), rather than being pre-generated
|
|
and stored in the database.
|
|
|
|
See: apps/output/views.py - generate_custom_dummy_programs()
|
|
|
|
This function remains for backward compatibility but should not be called.
|
|
"""
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Byte-offset programme index (ported from dev branch)
|
|
# These functions support fast current-program lookup for the CurrentPrograms
|
|
# API without doing a full DB query for every channel on every poll.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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\s*=\s*(?:"([^"]+)"|'([^']+)')""")
|
|
_PROGRAMME_TAG = b'<programme'
|
|
_PROGRAMME_TAG_LEN = len(_PROGRAMME_TAG)
|
|
_TAG_FOLLOW = b' \t\n\r>/'
|
|
_MAX_START_TAG = 4096 # generous upper bound for a start tag with namespaces/extra attrs
|
|
_OFFSET_CAP = 10 # max block-starts recorded per channel; exceeding this flags the channel as interleaved
|
|
|
|
|
|
def _decode_channel_id(raw):
|
|
"""Match how EPGData.tvg_id is stored: resolve XML entities and strip, so byte-level index keys equal the lxml-parsed channel ids."""
|
|
s = raw.decode('utf-8', errors='replace')
|
|
if '&' in s:
|
|
s = html.unescape(s)
|
|
return s.strip()
|
|
|
|
|
|
def _find_programme_tag(buf, start):
|
|
"""
|
|
Find the next <programme element in *buf* starting from *start*.
|
|
Returns (tag_pos, tag_end) or (-1, -1) if not found.
|
|
"""
|
|
pos = start
|
|
while True:
|
|
idx = buf.find(_PROGRAMME_TAG, pos)
|
|
if idx == -1:
|
|
return -1, -1
|
|
# Validate next byte is whitespace or '>'
|
|
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. <programmeXYZ), skip
|
|
continue
|
|
# Find the '>' 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] <programme> 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 <programme> 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.
|
|
Persists the result to the EPGSourceIndex table. Most XMLTV files group programmes
|
|
by channel, but some split a channel across multiple non-contiguous blocks, so we
|
|
record block starts up to _OFFSET_CAP and mark only channels that exceed the cap
|
|
as interleaved.
|
|
"""
|
|
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.debug(
|
|
f'[build_programme_index] Building byte-offset index for source {source_id} from {file_path}'
|
|
)
|
|
start = time.monotonic()
|
|
index = {}
|
|
prev_channel = None
|
|
interleaved_channels = set()
|
|
|
|
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 = _decode_channel_id(m.group(1) or m.group(2))
|
|
if channel_id not in index:
|
|
index[channel_id] = [abs_pos]
|
|
elif channel_id != prev_channel:
|
|
if len(index[channel_id]) < _OFFSET_CAP:
|
|
index[channel_id].append(abs_pos)
|
|
else:
|
|
interleaved_channels.add(channel_id)
|
|
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}'
|
|
+ (
|
|
f' ({len(interleaved_channels)} interleaved)'
|
|
if interleaved_channels
|
|
else ''
|
|
)
|
|
)
|
|
|
|
result = {
|
|
'channels': index,
|
|
'interleaved_channels': sorted(interleaved_channels),
|
|
}
|
|
EPGSourceIndex.objects.update_or_create(
|
|
source_id=source_id, defaults={'data': result}
|
|
)
|
|
|
|
|
|
@shared_task
|
|
def build_programme_index_task(source_id, _defer_retry=0):
|
|
"""Celery wrapper. Serializes index builds with other EPG file access on this source."""
|
|
if not acquire_task_lock(_EPG_SOURCE_FILE_LOCK, source_id):
|
|
if _defer_retry >= _EPG_PARSE_DEFER_MAX:
|
|
logger.warning(
|
|
f"Giving up on programme index build for source {source_id} "
|
|
f"after {_defer_retry} deferrals: file busy"
|
|
)
|
|
return
|
|
logger.debug(
|
|
f"EPG source file busy, deferring programme index build for source {source_id}"
|
|
)
|
|
build_programme_index_task.apply_async(
|
|
args=[source_id],
|
|
kwargs={'_defer_retry': _defer_retry + 1},
|
|
countdown=_EPG_TVG_PARSE_DEFER_SECONDS,
|
|
)
|
|
return
|
|
|
|
lock_renewer = TaskLockRenewer(_EPG_SOURCE_FILE_LOCK, source_id)
|
|
lock_renewer.start()
|
|
try:
|
|
build_programme_index(source_id)
|
|
finally:
|
|
lock_renewer.stop()
|
|
release_task_lock(_EPG_SOURCE_FILE_LOCK, source_id)
|
|
|
|
|
|
def find_current_program_for_tvg_id(epg_or_id):
|
|
"""
|
|
Look up the currently-airing program for an EPGData instance (or id) using
|
|
the byte-offset index. If no index exists yet, queue an async build and let
|
|
the caller retry rather than doing a blocking scan.
|
|
|
|
Returns dict, None, or "timeout".
|
|
"""
|
|
if isinstance(epg_or_id, EPGData):
|
|
epg = epg_or_id
|
|
else:
|
|
try:
|
|
epg = EPGData.objects.select_related('epg_source').get(id=epg_or_id)
|
|
except EPGData.DoesNotExist:
|
|
return None
|
|
|
|
source = epg.epg_source
|
|
if not source or source.source_type in ('dummy', 'schedules_direct'):
|
|
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
|
|
|
|
now = timezone.now()
|
|
# The property reads the EPGSourceIndex table fresh on each access, so a
|
|
# concurrent refresh invalidating/rebuilding the index can't serve stale state.
|
|
index = source.programme_index
|
|
|
|
if index is not None:
|
|
channels = index.get('channels', {})
|
|
if tvg_id not in channels:
|
|
# Channel has no programmes in the file
|
|
return None
|
|
offsets = channels[tvg_id]
|
|
if tvg_id in (index.get('interleaved_channels') or ()):
|
|
# Check all stored offsets first (cheap: one seek + one element parse each)
|
|
result = _read_programs_at_offsets(file_path, tvg_id, offsets, now)
|
|
if result is not None:
|
|
return result
|
|
# Current programme is beyond the stored offsets; scan forward from the
|
|
# last known position to avoid re-reading the already-checked portion
|
|
result = _scan_from_offset_for_tvg_id(file_path, tvg_id, offsets[-1], now)
|
|
if result == 'timeout':
|
|
logger.warning(
|
|
f'[find_current_program_for_tvg_id] Interleaved scan timed out for '
|
|
f'tvg_id={tvg_id} source={source.id}; index has {len(offsets)} offsets'
|
|
)
|
|
return None
|
|
return result
|
|
return _read_programs_at_offsets(file_path, tvg_id, offsets, now)
|
|
|
|
# No index yet: dispatch a background build and let the frontend retry.
|
|
# A sync scan can block a worker for ~10s on SMB-hosted EPGs.
|
|
build_programme_index_task.delay(source.id)
|
|
return 'timeout'
|
|
|
|
|
|
def _read_programs_at_offsets(file_path, tvg_id, offsets, now):
|
|
"""
|
|
Seek to each offset, extract <programme> elements for *tvg_id*, return the
|
|
first one currently airing. Chunk-based so it works on minified XML.
|
|
"""
|
|
PROG_CLOSE = b'</programme>'
|
|
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 = _decode_channel_id(m.group(1) or m.group(2))
|
|
if ch != tvg_id:
|
|
done = True # different channel, end of block
|
|
break
|
|
|
|
# Find the closing </programme> 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 = _parse_programme_element(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 _scan_from_offset_for_tvg_id(file_path, tvg_id, start_offset, now, timeout_sec=10):
|
|
"""
|
|
Scan forward from start_offset for tvg_id, skipping other channels rather than
|
|
stopping at a channel boundary. Used for interleaved/time-sorted XMLTV files where
|
|
a channel exceeded the stored offset cap.
|
|
Returns dict, None, or 'timeout'.
|
|
"""
|
|
PROG_CLOSE = b'</programme>'
|
|
CLOSE_LEN = len(PROG_CLOSE)
|
|
READ_SIZE = 2 * 1024 * 1024
|
|
deadline = time.monotonic() + timeout_sec
|
|
|
|
with open(file_path, 'rb') as f:
|
|
f.seek(start_offset)
|
|
buf = bytearray()
|
|
|
|
while True:
|
|
if time.monotonic() > deadline:
|
|
return 'timeout'
|
|
|
|
chunk = f.read(READ_SIZE)
|
|
if not chunk and not buf:
|
|
break
|
|
buf.extend(chunk)
|
|
search_from = 0
|
|
|
|
trim_to = 0
|
|
|
|
while True:
|
|
tag_start, tag_end = _find_programme_tag(buf, search_from)
|
|
if tag_start == -1:
|
|
trim_to = search_from
|
|
break
|
|
if tag_end == -1 and chunk:
|
|
trim_to = tag_start # keep incomplete tag for next read
|
|
break
|
|
|
|
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 = _decode_channel_id(m.group(1) or m.group(2))
|
|
if ch != tvg_id:
|
|
search_from = (
|
|
tag_end + 1 if tag_end != -1 else tag_start + _PROGRAMME_TAG_LEN
|
|
)
|
|
continue
|
|
|
|
close_pos = buf.find(
|
|
PROG_CLOSE, tag_end + 1 if tag_end != -1 else m.end()
|
|
)
|
|
if close_pos == -1:
|
|
trim_to = tag_start # keep incomplete element for next read
|
|
break
|
|
close_end = close_pos + CLOSE_LEN
|
|
|
|
element_bytes = bytes(buf[tag_start:close_end])
|
|
search_from = close_end
|
|
|
|
try:
|
|
prog = _parse_programme_element(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)
|
|
|
|
if trim_to > 0:
|
|
del buf[:trim_to]
|
|
|
|
if not chunk:
|
|
break
|
|
|
|
return None
|