diff --git a/CHANGELOG.md b/CHANGELOG.md index 0517493d..3315456b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Performance + +- **XMLTV EPG export streams without holding the full guide in the worker.** `generate_epg()` streams incrementally; on a cache miss each chunk is pushed to a Redis list as it is yielded (no `''.join()` in the worker). Repeat requests within 300s stream chunks back one at a time from Redis. Post-export `malloc_trim` runs after cold builds. Real programmes use fast `(epg_id, id)` keyset pagination with a per-source `start_time` sort before emit. +- **EPG export no longer fetches the multi-MB `programme_index` per channel.** The channel query `select_related`s `epg_data__epg_source`, which pulled and JSON-parsed each source's byte-offset `programme_index` blob once per channel (~13s of the request on a ~2000-channel guide). It is now `defer()`red since EPG generation never reads it. +- **`ProgramData` composite index `(epg_id, id)`.** The EPG export scans hundreds of thousands of programmes with keyset pagination on `(epg_id, id)`; without a matching index PostgreSQL re-sorted every chunk. A composite index (created `CONCURRENTLY` on PostgreSQL so it does not lock the table) lets each chunk use an ordered index range scan. + ### Changed - **Channel list with nested streams loads faster.** `GET /api/channels/channels/?include_streams=true` (Channels UI and single-channel fetch) now builds nested stream payloads from the prefetched `channelstream_set` instead of issuing one extra streams M2M query per channel. diff --git a/apps/epg/migrations/0025_programdata_epg_id_index.py b/apps/epg/migrations/0025_programdata_epg_id_index.py new file mode 100644 index 00000000..1a350b20 --- /dev/null +++ b/apps/epg/migrations/0025_programdata_epg_id_index.py @@ -0,0 +1,40 @@ +from django.contrib.postgres.operations import AddIndexConcurrently +from django.db import migrations, models + + +class AddIndexConcurrentlyIfPostgres(AddIndexConcurrently): + """Create the index CONCURRENTLY on PostgreSQL (no table lock on large + tables), falling back to a normal blocking AddIndex on other backends + such as the sqlite dev/test fallback.""" + + def database_forwards(self, app_label, schema_editor, from_state, to_state): + if schema_editor.connection.vendor == 'postgresql': + super().database_forwards(app_label, schema_editor, from_state, to_state) + else: + migrations.AddIndex.database_forwards( + self, app_label, schema_editor, from_state, to_state + ) + + def database_backwards(self, app_label, schema_editor, from_state, to_state): + if schema_editor.connection.vendor == 'postgresql': + super().database_backwards(app_label, schema_editor, from_state, to_state) + else: + migrations.AddIndex.database_backwards( + self, app_label, schema_editor, from_state, to_state + ) + + +class Migration(migrations.Migration): + # CREATE INDEX CONCURRENTLY cannot run inside a transaction. + atomic = False + + dependencies = [ + ('epg', '0024_remove_epgsource_api_key_epgsource_password_and_more'), + ] + + operations = [ + AddIndexConcurrentlyIfPostgres( + model_name='programdata', + index=models.Index(fields=['epg', 'id'], name='epg_prog_epg_id_idx'), + ), + ] diff --git a/apps/epg/models.py b/apps/epg/models.py index 025945db..04c6072c 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -153,6 +153,11 @@ class ProgramData(models.Model): program_id = models.CharField(max_length=64, null=True, blank=True, help_text='Schedules Direct programID (e.g. EP123456789). Null for XMLTV sources.') custom_properties = models.JSONField(default=dict, blank=True, null=True) + class Meta: + indexes = [ + models.Index(fields=['epg', 'id'], name='epg_prog_epg_id_idx'), + ] + def __str__(self): return f"{self.title} ({self.start_time} - {self.end_time})" diff --git a/apps/output/epg_chunk_cache.py b/apps/output/epg_chunk_cache.py new file mode 100644 index 00000000..92ea6eeb --- /dev/null +++ b/apps/output/epg_chunk_cache.py @@ -0,0 +1,230 @@ +"""Single-flight Redis chunk cache for XMLTV EPG streaming responses.""" + +import logging +import time + +from django.http import StreamingHttpResponse + +logger = logging.getLogger(__name__) + +STATUS_BUILDING = "building" +STATUS_READY = "ready" +STATUS_ERROR = "error" + +DEFAULT_CACHE_TTL = 300 +DEFAULT_LOCK_TTL = 120 +DEFAULT_POLL_INTERVAL = 0.05 +DEFAULT_MAX_FOLLOWER_WAIT = 600 + + +def _chunks_key(base_key): + return f"{base_key}:chunks" + + +def _ready_key(base_key): + return f"{base_key}:ready" + + +def _status_key(base_key): + return f"{base_key}:status" + + +def _lock_key(base_key): + return f"{base_key}:lock" + + +def _decode_chunk(chunk): + if chunk is None: + return None + if isinstance(chunk, bytes): + return chunk.decode("utf-8") + return chunk + + +def _encode_chunk(chunk): + if isinstance(chunk, bytes): + return chunk + return chunk.encode("utf-8") + + +def _poll_wait(interval): + try: + from core.utils import _is_gevent_monkey_patched + + if _is_gevent_monkey_patched(): + import gevent + + gevent.sleep(interval) + return + except ImportError: + pass + time.sleep(interval) + + +def _get_redis(): + from django_redis import get_redis_connection + + return get_redis_connection("default") + + +def _get_status(redis, base_key): + raw = redis.get(_status_key(base_key)) + if raw is None: + return None + return _decode_chunk(raw) + + +def _clear_build_keys(redis, base_key): + redis.delete( + _chunks_key(base_key), + _status_key(base_key), + _ready_key(base_key), + _lock_key(base_key), + ) + + +def _try_acquire_lock(redis, base_key, lock_ttl): + return bool(redis.set(_lock_key(base_key), "1", nx=True, ex=lock_ttl)) + + +def _refresh_build_ttl(redis, base_key, lock_ttl): + redis.expire(_lock_key(base_key), lock_ttl) + redis.expire(_status_key(base_key), lock_ttl) + redis.expire(_chunks_key(base_key), lock_ttl) + + +def _stream_ready(redis, base_key): + offset = 0 + chunks_key = _chunks_key(base_key) + while True: + chunk = redis.lindex(chunks_key, offset) + if chunk is None: + break + yield _decode_chunk(chunk) + offset += 1 + + +def _stream_build(redis, base_key, source, cache_ttl, lock_ttl): + """Leader: stream to client and append each chunk to Redis.""" + chunks_key = _chunks_key(base_key) + status_key = _status_key(base_key) + try: + from django.core.cache import cache as django_cache + + django_cache.delete(base_key) # legacy monolithic entry + redis.delete(chunks_key, _ready_key(base_key)) + redis.set(status_key, STATUS_BUILDING, ex=lock_ttl) + for chunk in source(): + redis.rpush(chunks_key, _encode_chunk(chunk)) + _refresh_build_ttl(redis, base_key, lock_ttl) + yield chunk + redis.set(status_key, STATUS_READY) + redis.set(_ready_key(base_key), "1") + redis.expire(chunks_key, cache_ttl) + redis.expire(status_key, cache_ttl) + redis.expire(_ready_key(base_key), cache_ttl) + logger.debug("Cached EPG in %s chunks", redis.llen(chunks_key)) + except Exception: + logger.exception("EPG cache build failed for %s", base_key) + redis.delete(chunks_key) + redis.set(status_key, STATUS_ERROR, ex=60) + raise + finally: + redis.delete(_lock_key(base_key)) + + +def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval, max_follower_wait): + """Follower: read chunks as the leader writes them.""" + offset = 0 + deadline = time.monotonic() + max_follower_wait + idle_polls = 0 + chunks_key = _chunks_key(base_key) + lock_key = _lock_key(base_key) + + while True: + chunk = redis.lindex(chunks_key, offset) + if chunk is not None: + idle_polls = 0 + yield _decode_chunk(chunk) + offset += 1 + continue + + status = _get_status(redis, base_key) + if status == STATUS_READY: + break + + if status == STATUS_ERROR: + _clear_build_keys(redis, base_key) + if offset == 0 and _try_acquire_lock(redis, base_key, lock_ttl): + yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl) + return + raise RuntimeError("EPG cache build failed") + + if time.monotonic() >= deadline: + if offset == 0 and _try_acquire_lock(redis, base_key, lock_ttl): + logger.warning("EPG cache follower timed out; rebuilding %s", base_key) + yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl) + return + logger.warning("EPG cache follower timed out after partial read for %s", base_key) + break + + lock_active = bool(redis.exists(lock_key)) + if status != STATUS_BUILDING and not lock_active: + idle_polls += 1 + if offset == 0 and idle_polls >= max(1, int(1.0 / poll_interval)): + if _try_acquire_lock(redis, base_key, lock_ttl): + logger.warning("EPG cache leader lost; rebuilding %s", base_key) + yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl) + return + else: + idle_polls = 0 + + _poll_wait(poll_interval) + + +def stream_epg_response( + cache_key, + source, + *, + cache_ttl=DEFAULT_CACHE_TTL, + lock_ttl=DEFAULT_LOCK_TTL, + poll_interval=DEFAULT_POLL_INTERVAL, + max_follower_wait=DEFAULT_MAX_FOLLOWER_WAIT, + redis=None, +): + """ + Stream XMLTV EPG output with single-flight Redis chunk caching. + + ``source`` must be a callable returning a chunk iterator. Only the leader + invokes it; followers read chunks already written to Redis. + """ + if redis is None: + redis = _get_redis() + + if redis.get(_ready_key(cache_key)): + logger.debug("Serving EPG from chunk cache") + stream = _stream_ready(redis, cache_key) + else: + status = _get_status(redis, cache_key) + if status == STATUS_ERROR: + _clear_build_keys(redis, cache_key) + + if _try_acquire_lock(redis, cache_key, lock_ttl): + logger.debug("Building EPG (cache leader)") + stream = _stream_build(redis, cache_key, source, cache_ttl, lock_ttl) + else: + logger.debug("Following in-flight EPG build") + stream = _stream_follow( + redis, + cache_key, + source, + cache_ttl, + lock_ttl, + poll_interval, + max_follower_wait, + ) + + response = StreamingHttpResponse(stream, content_type="application/xml") + response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"' + response["Cache-Control"] = "no-cache" + return response diff --git a/apps/output/test_epg_chunk_cache.py b/apps/output/test_epg_chunk_cache.py new file mode 100644 index 00000000..6ca217af --- /dev/null +++ b/apps/output/test_epg_chunk_cache.py @@ -0,0 +1,187 @@ +import threading +import time +from unittest import TestCase + +from apps.output.epg_chunk_cache import ( + STATUS_BUILDING, + STATUS_READY, + _chunks_key, + _lock_key, + _ready_key, + _status_key, + stream_epg_response, +) + + +class FakeRedis: + """Minimal Redis stand-in for chunk-cache unit tests.""" + + def __init__(self): + self._strings = {} + self._lists = {} + self._expires_at = {} + + def _purge_expired(self): + now = time.monotonic() + expired = [key for key, deadline in self._expires_at.items() if deadline <= now] + for key in expired: + self._strings.pop(key, None) + self._lists.pop(key, None) + self._expires_at.pop(key, None) + + def get(self, key): + self._purge_expired() + return self._strings.get(key) + + def set(self, key, value, nx=False, ex=None): + self._purge_expired() + if nx and key in self._strings: + return None + self._strings[key] = value + if ex is not None: + self._expires_at[key] = time.monotonic() + ex + return True + + def delete(self, *keys): + for key in keys: + self._strings.pop(key, None) + self._lists.pop(key, None) + self._expires_at.pop(key, None) + + def exists(self, key): + self._purge_expired() + return key in self._strings or key in self._lists + + def expire(self, key, ttl): + if key in self._strings or key in self._lists: + self._expires_at[key] = time.monotonic() + ttl + return True + + def rpush(self, key, value): + self._lists.setdefault(key, []).append(value) + + def lindex(self, key, offset): + items = self._lists.get(key, []) + if offset < len(items): + return items[offset] + return None + + def llen(self, key): + return len(self._lists.get(key, [])) + + +def _consume(response): + return b"".join(response.streaming_content).decode("utf-8") + + +class EPGChunkCacheTests(TestCase): + def test_leader_caches_chunks_and_sets_ready(self): + redis = FakeRedis() + calls = [] + + def source(): + calls.append(1) + yield "" + yield "" + + body = _consume(stream_epg_response("epg:test", source, redis=redis)) + + self.assertEqual(body, "") + self.assertEqual(calls, [1]) + self.assertEqual(redis.get(_ready_key("epg:test")), "1") + self.assertEqual(redis.get(_status_key("epg:test")), STATUS_READY) + self.assertEqual(redis.llen(_chunks_key("epg:test")), 2) + self.assertFalse(redis.exists(_lock_key("epg:test"))) + + def test_cache_hit_skips_source(self): + redis = FakeRedis() + calls = [] + + def source(): + calls.append(1) + yield "" + yield "" + + _consume(stream_epg_response("epg:test", source, redis=redis)) + calls.clear() + body = _consume(stream_epg_response("epg:test", source, redis=redis)) + + self.assertEqual(body, "") + self.assertEqual(calls, []) + + def test_follower_reads_leader_chunks_without_rebuilding(self): + redis = FakeRedis() + base = "epg:follow" + leader_started = threading.Event() + rebuild_calls = [] + + def slow_source(): + rebuild_calls.append(1) + leader_started.set() + yield "a" + time.sleep(0.05) + yield "b" + + def forbidden_source(): + rebuild_calls.append(2) + yield "SHOULD_NOT_RUN" + + def leader(): + _consume( + stream_epg_response( + base, + slow_source, + redis=redis, + poll_interval=0.01, + ) + ) + + leader_thread = threading.Thread(target=leader) + leader_thread.start() + leader_started.wait(timeout=5) + follower_body = _consume( + stream_epg_response( + base, + forbidden_source, + redis=redis, + poll_interval=0.01, + ) + ) + leader_thread.join(timeout=5) + + self.assertEqual(follower_body, "ab") + self.assertEqual(rebuild_calls, [1]) + + def test_only_one_leader_when_two_clients_start_together(self): + redis = FakeRedis() + build_calls = [] + barrier = threading.Barrier(2) + results = {} + + def source(): + build_calls.append(threading.current_thread().name) + yield "x" + + def worker(): + barrier.wait() + results[threading.current_thread().name] = _consume( + stream_epg_response( + "epg:race", + source, + redis=redis, + poll_interval=0.01, + ) + ) + + threads = [ + threading.Thread(target=worker, name="t1"), + threading.Thread(target=worker, name="t2"), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + self.assertEqual(results["t1"], "x") + self.assertEqual(results["t2"], "x") + self.assertEqual(len(build_calls), 1) diff --git a/apps/output/tests.py b/apps/output/tests.py index b1c7d589..6e2ab395 100644 --- a/apps/output/tests.py +++ b/apps/output/tests.py @@ -1,31 +1,114 @@ -from django.test import TestCase, Client +from django.test import TestCase, Client, SimpleTestCase from django.urls import reverse -from apps.channels.models import Channel, ChannelGroup +from unittest.mock import patch +from uuid import uuid4 +from apps.channels.models import Channel, ChannelGroup, ChannelProfile, ChannelProfileMembership from apps.epg.models import EPGData, EPGSource import xml.etree.ElementTree as ET +from datetime import timedelta + + +def _response_text(response): + """Read body from HttpResponse or StreamingHttpResponse.""" + if getattr(response, "streaming", False): + return b"".join(response.streaming_content).decode() + return response.content.decode() + + +def _epg_response_without_redis(cache_key, source, **kwargs): + """Test helper: stream EPG directly without Redis chunk caching.""" + from django.http import StreamingHttpResponse + + response = StreamingHttpResponse(source(), content_type="application/xml") + response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"' + response["Cache-Control"] = "no-cache" + return response + + +class OutputEndpointTestMixin: + """Isolate HTTP endpoint tests from network ACL, logging, DB teardown, and Redis.""" -class OutputM3UTest(TestCase): def setUp(self): + super().setUp() + self._network_patch = patch( + "apps.output.views.network_access_allowed", + return_value=True, + ) + self._epg_teardown_patch = patch("apps.output.views._epg_export_teardown") + self._log_event_patch = patch("apps.output.views.log_system_event") + self._close_db_patch = patch("django.db.close_old_connections") + self._epg_cache_patch = patch( + "apps.output.views.stream_epg_response", + side_effect=_epg_response_without_redis, + ) + self._network_patch.start() + self._epg_teardown_patch.start() + self._log_event_patch.start() + self._close_db_patch.start() + self._epg_cache_patch.start() + + def tearDown(self): + from django.core.cache import cache + + cache.clear() + self._epg_cache_patch.stop() + self._close_db_patch.stop() + self._log_event_patch.stop() + self._epg_teardown_patch.stop() + self._network_patch.stop() + super().tearDown() + + def _create_isolated_profile(self, prefix): + """New profiles auto-include every channel via signal; clear that for tests.""" + profile = ChannelProfile.objects.create(name=f"{prefix}-{uuid4().hex[:8]}") + ChannelProfileMembership.objects.filter(channel_profile=profile).delete() + return profile + + def _add_channel_to_profile(self, profile, group, **kwargs): + channel = Channel.objects.create(channel_group=group, **kwargs) + ChannelProfileMembership.objects.create( + channel_profile=profile, + channel=channel, + enabled=True, + ) + return channel + + +class OutputM3UTest(OutputEndpointTestMixin, TestCase): + def setUp(self): + super().setUp() self.client = Client() - + self.group = ChannelGroup.objects.create(name=f"M3U Group {uuid4().hex[:8]}") + self.profile = self._create_isolated_profile("m3u") + self._add_channel_to_profile( + self.profile, + self.group, + channel_number=1.0, + name="Test M3U Channel", + ) + + def _m3u_url(self): + return reverse("output:m3u_endpoint", kwargs={"profile_name": self.profile.name}) + def test_generate_m3u_response(self): """ Test that the M3U endpoint returns a valid M3U file. """ - url = reverse('output:generate_m3u') - response = self.client.get(url) + response = self.client.get(self._m3u_url()) self.assertEqual(response.status_code, 200) - content = response.content.decode() + content = _response_text(response) self.assertIn("#EXTM3U", content) def test_generate_m3u_response_post_empty_body(self): """ Test that a POST request with an empty body returns 200 OK. """ - url = reverse('output:generate_m3u') - - response = self.client.post(url, data=None, content_type='application/x-www-form-urlencoded') - content = response.content.decode() + response = self.client.post( + self._m3u_url(), + data=None, + content_type="application/x-www-form-urlencoded", + ) + content = _response_text(response) self.assertEqual(response.status_code, 200, "POST with empty body should return 200 OK") self.assertIn("#EXTM3U", content) @@ -34,35 +117,40 @@ class OutputM3UTest(TestCase): """ Test that a POST request with a non-empty body returns 403 Forbidden. """ - url = reverse('output:generate_m3u') - - response = self.client.post(url, data={'evilstring': 'muhahaha'}) + response = self.client.post(self._m3u_url(), data={"evilstring": "muhahaha"}) self.assertEqual(response.status_code, 403, "POST with body should return 403 Forbidden") - self.assertIn("POST requests with body are not allowed, body is:", response.content.decode()) + self.assertIn("POST requests with body are not allowed", _response_text(response)) -class OutputEPGXMLEscapingTest(TestCase): +class OutputEPGXMLEscapingTest(OutputEndpointTestMixin, TestCase): """Test XML escaping of channel_id attributes in EPG generation""" def setUp(self): + super().setUp() self.client = Client() - self.group = ChannelGroup.objects.create(name="Test Group") + self.group = ChannelGroup.objects.create(name=f"Test Group {uuid4().hex[:8]}") + self.profile = self._create_isolated_profile("epg-xml") + + def _add_channel(self, **kwargs): + return self._add_channel_to_profile(self.profile, self.group, **kwargs) + + def _epg_url(self, query="tvg_id_source=tvg_id&days=0&prev_days=0"): + base = reverse("output:epg_endpoint", kwargs={"profile_name": self.profile.name}) + return f"{base}?{query}" def test_channel_id_with_ampersand(self): """Test channel ID with ampersand is properly escaped""" - channel = Channel.objects.create( + self._add_channel( channel_number=1.0, name="Test Channel", tvg_id="News & Sports", - channel_group=self.group ) - url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id' - response = self.client.get(url) + response = self.client.get(self._epg_url()) self.assertEqual(response.status_code, 200) - content = response.content.decode() + content = _response_text(response) # Should contain escaped ampersand self.assertIn('id="News & Sports"', content) @@ -76,17 +164,15 @@ class OutputEPGXMLEscapingTest(TestCase): def test_channel_id_with_angle_brackets(self): """Test channel ID with < and > characters""" - channel = Channel.objects.create( + self._add_channel( channel_number=2.0, name="HD Channel", tvg_id="Channel ", - channel_group=self.group ) - url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id' - response = self.client.get(url) + response = self.client.get(self._epg_url()) - content = response.content.decode() + content = _response_text(response) self.assertIn('id="Channel <HD>"', content) try: @@ -96,23 +182,28 @@ class OutputEPGXMLEscapingTest(TestCase): def test_channel_id_with_all_special_chars(self): """Test channel ID with all XML special characters""" - channel = Channel.objects.create( + expected_id = 'Test & "Special" ' + self._add_channel( channel_number=3.0, name="Complex Channel", - tvg_id='Test & "Special" ', - channel_group=self.group + tvg_id=expected_id, ) - url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id' - response = self.client.get(url) + response = self.client.get(self._epg_url()) - content = response.content.decode() + content = _response_text(response) self.assertIn('id="Test & "Special" <Chars>"', content) try: tree = ET.fromstring(content) - # Verify we can find the channel with correct ID in parsed tree - channel_elem = tree.find('.//channel[@id="Test & \\"Special\\" "]') + channel_elem = next( + ( + elem + for elem in tree.findall(".//channel") + if elem.get("id") == expected_id + ), + None, + ) self.assertIsNotNone(channel_elem) except ET.ParseError as e: self.fail(f"Generated EPG with all special chars is not valid XML: {e}") @@ -121,25 +212,144 @@ class OutputEPGXMLEscapingTest(TestCase): """Test that programme elements also have escaped channel attributes""" epg_source = EPGSource.objects.create(name="Test EPG", source_type="dummy") epg_data = EPGData.objects.create(name="Test EPG Data", epg_source=epg_source) - channel = Channel.objects.create( + self._add_channel( channel_number=4.0, name="Program Test", tvg_id="News & Sports", epg_data=epg_data, - channel_group=self.group ) - url = reverse('output:generate_epg') + '?tvg_id_source=tvg_id' - response = self.client.get(url) + response = self.client.get(self._epg_url()) - content = response.content.decode() + content = _response_text(response) # Check programme elements have escaped channel attributes self.assertIn('channel="News & Sports"', content) try: tree = ET.fromstring(content) - programmes = tree.findall('.//programme[@channel="News & Sports"]') + programmes = [ + programme + for programme in tree.findall(".//programme") + if programme.get("channel") == "News & Sports" + ] self.assertGreater(len(programmes), 0) except ET.ParseError as e: self.fail(f"Generated EPG with programme elements is not valid XML: {e}") + + def test_programmes_emitted_in_start_time_order(self): + """Programmes for a channel are emitted in start_time order, not insert order.""" + from django.utils import timezone + from apps.epg.models import ProgramData + + epg_source = EPGSource.objects.create(name="Real EPG", source_type="xmltv") + epg_data = EPGData.objects.create(name="Station", epg_source=epg_source, tvg_id="station1") + self._add_channel( + channel_number=149.0, + name="Food Network", + tvg_id="station1", + epg_data=epg_data, + ) + now = timezone.now() + # Insert out of chronological order so id order != start_time order. + ProgramData.objects.create( + epg=epg_data, + start_time=now + timedelta(days=3), + end_time=now + timedelta(days=3, hours=1), + title="Third", + tvg_id="station1", + ) + ProgramData.objects.create( + epg=epg_data, + start_time=now + timedelta(days=1), + end_time=now + timedelta(days=1, hours=1), + title="First", + tvg_id="station1", + ) + ProgramData.objects.create( + epg=epg_data, + start_time=now + timedelta(days=2), + end_time=now + timedelta(days=2, hours=1), + title="Second", + tvg_id="station1", + ) + + content = _response_text(self.client.get(self._epg_url("tvg_id_source=tvg_id&days=7"))) + + self.assertLess(content.find('First'), content.find('Second')) + self.assertLess(content.find('Second'), content.find('Third')) + + +class OutputEPGCustomDummyTest(TestCase): + """Custom dummy EPG must not fall back to default when pattern matched but event is outside window.""" + + def setUp(self): + self.group = ChannelGroup.objects.create(name="Sports Group") + + def test_custom_dummy_outside_window_fills_with_ended_programmes(self): + from django.utils import timezone + from apps.output.views import generate_dummy_programs + + epg_source = EPGSource.objects.create( + name="NHL Dummy", + source_type="dummy", + custom_properties={ + "title_pattern": r"(?.*)\s\d+:\s(?.*?)(?:\s+vs\s+)(?.*?)\s*@.*", + "time_pattern": r"(?\d{1,2}):(?\d{2})\s*(?AM|PM)", + "date_pattern": r"@ (?[A-Za-z]+)\s+(?\d{1,2})", + "timezone": "US/Eastern", + "program_duration": 180, + }, + ) + channel_name = ( + "NHL 01: Washington Capitals vs Philadelphia Flyers @ April 16 07:30 PM ET" + ) + now = timezone.now() + lookback = now - timedelta(days=7) + + programs = generate_dummy_programs( + channel_id="nhl01", + channel_name=channel_name, + num_days=7, + epg_source=epg_source, + export_lookback=lookback, + export_cutoff=now + timedelta(days=7), + ) + + self.assertGreater(len(programs), 0) + self.assertTrue( + all(p['end_time'] >= lookback for p in programs), + "All programmes should fall inside the export window", + ) + self.assertTrue( + any('Ended' in p['description'] for p in programs), + "Past events outside the window should still show ended filler", + ) + for program in programs: + start = program['start_time'] + self.assertEqual(start.second, 0) + self.assertEqual(start.microsecond, 0) + self.assertIn( + start.minute, (0, 30), + "Filler programmes should start on half-hour boundaries", + ) + self.assertGreaterEqual(programs[0]['start_time'], lookback) + + +class OutputEPGHelperTest(SimpleTestCase): + def test_ceil_to_half_hour_on_boundary(self): + from django.utils import timezone + from apps.output.views import _ceil_to_half_hour + + dt = timezone.now().replace(minute=30, second=0, microsecond=0) + self.assertEqual(_ceil_to_half_hour(dt), dt) + + def test_ceil_to_half_hour_rounds_up(self): + from django.utils import timezone + from apps.output.views import _ceil_to_half_hour + + dt = timezone.now().replace(minute=17, second=42, microsecond=123456) + aligned = _ceil_to_half_hour(dt) + self.assertEqual(aligned.minute, 30) + self.assertEqual(aligned.second, 0) + self.assertGreater(aligned, dt.replace(second=0, microsecond=0)) diff --git a/apps/output/views.py b/apps/output/views.py index dba35951..a0875c7f 100644 --- a/apps/output/views.py +++ b/apps/output/views.py @@ -25,9 +25,55 @@ from apps.proxy.utils import get_user_active_connections import regex from core.utils import log_system_event, build_absolute_uri_with_port import hashlib +from apps.output.epg_chunk_cache import stream_epg_response logger = logging.getLogger(__name__) +_EPG_CHANNEL_XML_BATCH_SIZE = 200 +_EPG_PROGRAM_YIELD_BATCH_SIZE = 1000 +_EPG_PROGRAM_DB_CHUNK_SIZE = 20000 + + +def _programme_overlaps_export_window(start_time, end_time, lookback_cutoff, cutoff_date): + if end_time < lookback_cutoff: + return False + if cutoff_date is not None and start_time >= cutoff_date: + return False + return True + + +def _ceil_to_half_hour(dt): + """Round a datetime up to the next :00 or :30 boundary.""" + dt = dt.replace(second=0, microsecond=0) + remainder = dt.minute % 30 + if remainder == 0: + return dt + return dt + timedelta(minutes=30 - remainder) + + +def _epg_export_teardown(): + from django.db import close_old_connections + + from core.utils import ( + _is_gevent_monkey_patched, + cleanup_memory, + trim_c_allocator_heap, + ) + + close_old_connections() + + def _run(): + cleanup_memory(force_collection=True) + trim_c_allocator_heap() + + if _is_gevent_monkey_patched(): + import gevent + + gevent.spawn(_run) + else: + _run() + + def get_client_identifier(request): """Get client information including IP, user agent, and a unique hash identifier @@ -112,6 +158,10 @@ def generate_m3u(request, profile_name=None, user=None): # Check if this is a POST request and the body is not empty (which we don't want to allow) logger.debug("Generating M3U for profile: %s, user: %s, method: %s", profile_name, user.username if user else "Anonymous", request.method) + if request.method == "POST" and request.body: + if request.body.decode() != '{}': + return HttpResponseForbidden("POST requests with body are not allowed.") + # Check cache for recent identical request (helps with double-GET from browsers) from django.core.cache import cache cache_params = f"{profile_name or 'all'}:{user.username if user else 'anonymous'}:{request.GET.urlencode()}" @@ -123,10 +173,6 @@ def generate_m3u(request, profile_name=None, user=None): response = HttpResponse(cached_content, content_type="audio/x-mpegurl") response["Content-Disposition"] = 'attachment; filename="channels.m3u"' return response - # Check if this is a POST request with data (which we don't want to allow) - if request.method == "POST" and request.body: - if request.body.decode() != '{}': - return HttpResponseForbidden("POST requests with body are not allowed.") if user is not None: if user.user_level < 10: @@ -409,7 +455,15 @@ def generate_fallback_programs(channel_id, channel_name, now, num_days, program_ return programs -def generate_dummy_programs(channel_id, channel_name, num_days=1, program_length_hours=4, epg_source=None): +def generate_dummy_programs( + channel_id, + channel_name, + num_days=1, + program_length_hours=4, + epg_source=None, + export_lookback=None, + export_cutoff=None, +): """ Generate dummy EPG programs for channels. @@ -435,29 +489,26 @@ def generate_dummy_programs(channel_id, channel_name, num_days=1, program_length if epg_source and epg_source.source_type == 'dummy' and epg_source.custom_properties: custom_programs = generate_custom_dummy_programs( channel_id, channel_name, now, num_days, - epg_source.custom_properties + epg_source.custom_properties, + export_lookback=export_lookback, + export_cutoff=export_cutoff, ) - # If custom generation succeeded, return those programs - # If it returned empty (pattern didn't match), check for custom fallback templates - if custom_programs: + if custom_programs is not None: return custom_programs - else: - logger.info(f"Custom pattern didn't match for '{channel_name}', checking for custom fallback templates") - # Check if custom fallback templates are provided - custom_props = epg_source.custom_properties - fallback_title = custom_props.get('fallback_title_template', '').strip() - fallback_description = custom_props.get('fallback_description_template', '').strip() + logger.info(f"Custom pattern didn't match for '{channel_name}', checking for custom fallback templates") - # If custom fallback templates exist, use them instead of default - if fallback_title or fallback_description: - logger.info(f"Using custom fallback templates for '{channel_name}'") - return generate_fallback_programs( - channel_id, channel_name, now, num_days, - program_length_hours, fallback_title, fallback_description - ) - else: - logger.info(f"No custom fallback templates found, using default dummy EPG") + custom_props = epg_source.custom_properties + fallback_title = custom_props.get('fallback_title_template', '').strip() + fallback_description = custom_props.get('fallback_description_template', '').strip() + + if fallback_title or fallback_description: + logger.info(f"Using custom fallback templates for '{channel_name}'") + return generate_fallback_programs( + channel_id, channel_name, now, num_days, + program_length_hours, fallback_title, fallback_description + ) + logger.info(f"No custom fallback templates found, using default dummy EPG") # Default humorous program descriptions based on time of day time_descriptions = { @@ -531,7 +582,15 @@ def generate_dummy_programs(channel_id, channel_name, num_days=1, program_length return programs -def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, custom_properties): +def generate_custom_dummy_programs( + channel_id, + channel_name, + now, + num_days, + custom_properties, + export_lookback=None, + export_cutoff=None, +): """ Generate programs using custom dummy EPG regex patterns. @@ -616,7 +675,7 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust if not title_pattern: logger.warning(f"No title_pattern in custom_properties, falling back to default") - return [] # Return empty, will use default + return None logger.debug(f"Title pattern from DB: {repr(title_pattern)}") @@ -633,7 +692,7 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust except Exception as e: logger.error(f"Invalid title regex pattern after conversion: {e}") logger.error(f"Pattern was: {repr(title_pattern)}") - return [] + return None time_regex = None if time_pattern: @@ -665,7 +724,7 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust title_match = title_regex.search(channel_name) if not title_match: logger.debug(f"Channel name '{channel_name}' doesn't match title pattern") - return [] # Return empty, will use default + return None groups = title_match.groupdict() logger.debug(f"Title pattern matched. Groups: {groups}") @@ -917,57 +976,93 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust iterations = num_days for day in range(iterations): - # Start from current time (like standard dummy) instead of midnight - # This ensures programs appear in the guide's current viewing window - day_start = now + timedelta(days=day) - day_end = day_start + timedelta(days=1) - - if time_info: - # We have an extracted event time - this is when the MAIN event starts - # The extracted time is in the SOURCE timezone (e.g., 8PM ET) - # We need to convert it to UTC for storage - - # Determine which date to use - if date_info: - # Use the extracted date from the channel title - current_date = datetime( - date_info['year'], - date_info['month'], - date_info['day'] - ).date() - logger.debug(f"Using extracted date: {current_date}") - else: - # No date extracted, use day offset from current time in SOURCE timezone - # This ensures we calculate "today" in the event's timezone, not UTC - # For example: 8:30 PM Central (1:30 AM UTC next day) for a 10 PM ET event - # should use today's date in ET, not tomorrow's date in UTC - now_in_source_tz = now.astimezone(source_tz) - current_date = (now_in_source_tz + timedelta(days=day)).date() - logger.debug(f"No date extracted, using day offset in {source_tz}: {current_date}") - - # Create a naive datetime (no timezone info) representing the event in source timezone + event_overlaps_window = True + if date_info and time_info: + current_date = datetime( + date_info['year'], + date_info['month'], + date_info['day'], + ).date() event_start_naive = datetime.combine( current_date, datetime.min.time().replace( hour=time_info['hour'], - minute=time_info['minute'] - ) + minute=time_info['minute'], + ), ) - - # Use pytz to localize the naive datetime to the source timezone - # This automatically handles DST! try: - event_start_local = source_tz.localize(event_start_naive) - # Convert to UTC - event_start_utc = event_start_local.astimezone(pytz.utc) - logger.debug(f"Converted {event_start_local} to UTC: {event_start_utc}") + event_start_utc = source_tz.localize(event_start_naive).astimezone(pytz.utc) except Exception as e: logger.error(f"Error localizing time to {source_tz}: {e}") - # Fallback: treat as UTC event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc) - event_end_utc = event_start_utc + timedelta(minutes=program_duration) + lookback = export_lookback if export_lookback is not None else now + event_overlaps_window = _programme_overlaps_export_window( + event_start_utc, event_end_utc, lookback, export_cutoff + ) + if not event_overlaps_window: + logger.debug( + "Custom dummy event outside export window; filling window only: %s", + channel_name, + ) + event_happened = event_end_utc < lookback + day_start = _ceil_to_half_hour(lookback) + if export_cutoff is not None: + day_end = export_cutoff + else: + day_end = now + timedelta(days=num_days if num_days > 0 else 3) + else: + day_start = source_tz.localize( + datetime.combine(current_date, datetime.min.time()) + ).astimezone(pytz.utc) + day_end = day_start + timedelta(days=1) + if export_lookback is not None: + day_start = max(day_start, export_lookback) + if export_cutoff is not None: + day_end = min(day_end, export_cutoff) + else: + day_start = now + timedelta(days=day) + day_end = day_start + timedelta(days=1) + if export_lookback is not None: + day_start = max(day_start, export_lookback) + if export_cutoff is not None: + day_end = min(day_end, export_cutoff) + + if day_start >= day_end: + continue + + if time_info: + if not date_info: + now_in_source_tz = now.astimezone(source_tz) + current_date = (now_in_source_tz + timedelta(days=day)).date() + logger.debug(f"No date extracted, using day offset in {source_tz}: {current_date}") + + event_start_naive = datetime.combine( + current_date, + datetime.min.time().replace( + hour=time_info['hour'], + minute=time_info['minute'], + ), + ) + try: + event_start_local = source_tz.localize(event_start_naive) + event_start_utc = event_start_local.astimezone(pytz.utc) + logger.debug(f"Converted {event_start_local} to UTC: {event_start_utc}") + except Exception as e: + logger.error(f"Error localizing time to {source_tz}: {e}") + event_start_utc = django_timezone.make_aware(event_start_naive, pytz.utc) + + event_end_utc = event_start_utc + timedelta(minutes=program_duration) + + lookback = export_lookback if export_lookback is not None else now + if not _programme_overlaps_export_window( + event_start_utc, event_end_utc, lookback, export_cutoff + ): + continue + else: + logger.debug(f"Using extracted date: {current_date}") + # Pre-generate the main event title and description for reuse if title_template: main_event_title = format_template(title_template, all_groups) @@ -994,15 +1089,13 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust # Determine if this day is before, during, or after the event - # Event only happens on day 0 (first day) - is_event_day = (day == 0) + # Event only happens on day 0 (first day) when it falls inside the window + is_event_day = (day == 0) and event_overlaps_window if is_event_day and not event_happened: - # This is THE day the event happens - # Fill programs BEFORE the event current_time = day_start - while current_time < event_start_utc: + while current_time < event_start_utc and current_time < day_end: program_start_utc = current_time program_end_utc = min(current_time + timedelta(minutes=program_duration), event_start_utc) @@ -1084,8 +1177,8 @@ def generate_custom_dummy_programs(channel_id, channel_name, now, num_days, cust event_happened = True - # Fill programs AFTER the event until end of day - current_time = event_end_utc + # Fill programs AFTER the event until end of export day window + current_time = max(event_end_utc, day_start) while current_time < day_end: program_start_utc = current_time @@ -1325,18 +1418,10 @@ def generate_dummy_epg( def generate_epg(request, profile_name=None, user=None): """ - Dynamically generate an XMLTV (EPG) file using streaming response to handle keep-alives. + Dynamically generate an XMLTV (EPG) file using a streaming response. Since the EPG data is stored independently of Channels, we group programmes by their associated EPGData record. - This version filters data based on the 'days' parameter and sends keep-alives during processing. """ - # Check cache for recent identical request (helps with double-GET from browsers) - from django.core.cache import cache - # Resolve all effective parameter values once here so they are reused for both - # the cache key and inside epg_generator() via closure. - # The cache key is built from resolved values only — not from the raw query string — - # so equivalent requests (e.g. days=7 via URL param vs. user default of 7) share - # the same cache entry regardless of how the value was supplied. user_custom = (user.custom_properties or {}) if user else {} try: num_days = int(request.GET.get('days', user_custom.get('epg_days', 0))) @@ -1356,21 +1441,13 @@ def generate_epg(request, profile_name=None, user=None): ) content_cache_key = f"epg_content:{cache_params}" - cached_content = cache.get(content_cache_key) - if cached_content: - logger.debug("Serving EPG from cache") - response = HttpResponse(cached_content, content_type="application/xml") - response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"' - response["Cache-Control"] = "no-cache" - return response - def epg_generator(): """Generator function that yields EPG data with keep-alives during processing.""" - xml_lines = [] - xml_lines.append('') - xml_lines.append( - '' + yield '\n' + yield ( + '\n' ) # Get channels based on user/profile @@ -1416,14 +1493,19 @@ def generate_epg(request, profile_name=None, user=None): # Resolve effective values at SQL level and exclude hidden channels # so output ordering/display honors user overrides. from apps.channels.managers import with_effective_values - channels = ( + channels = list( with_effective_values(base_qs, select_related_fks=True) .exclude(hidden_from_output=True) .order_by("effective_channel_number") + # programme_index is a multi-MB JSON byte-offset index that EPG + # generation never reads; defer it so it isn't fetched and JSON-parsed + # once per channel (was ~13s of the request on large guides). + .defer("epg_data__epg_source__programme_index") .prefetch_related( Prefetch('streams', queryset=Stream.objects.order_by('channelstream__order')) ) ) + channel_count = len(channels) # For dummy EPG, use either the specified value or default to 3 days dummy_days = num_days if num_days > 0 else 3 @@ -1465,12 +1547,14 @@ def generate_epg(request, profile_name=None, user=None): _logo_url_prefix = _base_url + _logo_prefix_raw + "/" _logo_url_suffix = "/" + _logo_suffix_raw - dummy_epg_ids_for_program_check = set() + dummy_program_list = [] + real_epg_map = {} + channel_xml_batch = [] - # Process channels for the section for channel in channels: effective_name = channel.effective_name effective_epg_data = channel.effective_epg_data_obj + effective_epg_data_id = channel.effective_epg_data_id effective_logo = channel.effective_logo_obj effective_number = channel.effective_channel_number @@ -1492,8 +1576,6 @@ def generate_epg(request, profile_name=None, user=None): # Check if this is a custom dummy EPG with channel logo URL template if effective_epg_data and effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy': - if channel.effective_epg_data_id: - dummy_epg_ids_for_program_check.add(channel.effective_epg_data_id) epg_source = effective_epg_data.epg_source if epg_source.custom_properties: custom_props = epg_source.custom_properties @@ -1548,51 +1630,16 @@ def generate_epg(request, profile_name=None, user=None): tvg_logo = direct_logo else: tvg_logo = f"{_logo_url_prefix}{effective_logo.id}{_logo_url_suffix}" - display_name = effective_name - xml_lines.append(f' ') - xml_lines.append(f' {html.escape(display_name)}') - xml_lines.append(f' ') - xml_lines.append(" ") + channel_xml_batch.append(f' ') + channel_xml_batch.append(f' {html.escape(effective_name)}') + channel_xml_batch.append(f' ') + channel_xml_batch.append(" ") - # Send all channel definitions - channel_xml = '\n'.join(xml_lines) + '\n' - yield channel_xml - xml_lines = [] # Clear to save memory + if len(channel_xml_batch) >= _EPG_CHANNEL_XML_BATCH_SIZE * 4: + yield '\n'.join(channel_xml_batch) + '\n' + channel_xml_batch = [] - dummy_epg_with_programs = set() - if dummy_epg_ids_for_program_check: - dummy_epg_with_programs = set( - ProgramData.objects.filter(epg_id__in=dummy_epg_ids_for_program_check) - .values_list('epg_id', flat=True) - .distinct() - ) - - # Pre-pass: categorize channels into dummy and real EPG groups - dummy_program_list = [] # (channel_id, pattern_match_name, epg_source_or_None) - real_epg_map = {} # epg_data_id -> [channel_id, ...] - - for channel in channels: - effective_name = channel.effective_name - effective_epg_data = channel.effective_epg_data_obj - effective_epg_data_id = channel.effective_epg_data_id - effective_number = channel.effective_channel_number - - # Determine channel_id (same logic as channel section) - if tvg_id_source == 'tvg_id' and channel.effective_tvg_id: - channel_id = channel.effective_tvg_id - elif tvg_id_source == 'gracenote' and channel.effective_tvc_guide_stationid: - channel_id = channel.effective_tvc_guide_stationid - else: - if user is not None: - formatted_channel_number = channel_num_map[channel.id] - else: - formatted_channel_number = format_channel_number(effective_number) - channel_id = str(formatted_channel_number) if formatted_channel_number != "" else str(channel.id) - - display_name = effective_epg_data.name if effective_epg_data else effective_name pattern_match_name = effective_name - - # Check if we should use stream name instead of channel name if effective_epg_data and effective_epg_data.epg_source: epg_source = effective_epg_data.epg_source if epg_source.custom_properties: @@ -1619,48 +1666,19 @@ def generate_epg(request, profile_name=None, user=None): if not effective_epg_data: dummy_program_list.append((channel_id, pattern_match_name, None)) + elif effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy': + dummy_program_list.append((channel_id, pattern_match_name, effective_epg_data.epg_source)) else: - if effective_epg_data.epg_source and effective_epg_data.epg_source.source_type == 'dummy': - if effective_epg_data_id in dummy_epg_with_programs: - real_epg_map.setdefault(effective_epg_data_id, []).append(channel_id) - else: - dummy_program_list.append((channel_id, pattern_match_name, effective_epg_data.epg_source)) - continue - real_epg_map.setdefault(effective_epg_data_id, []).append(channel_id) - # Emit dummy programmes - for channel_id, pattern_match_name, epg_source in dummy_program_list: - program_length_hours = 4 - dummy_programs = generate_dummy_programs( - channel_id, pattern_match_name, - num_days=dummy_days, - program_length_hours=program_length_hours, - epg_source=epg_source - ) - for program in dummy_programs: - start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z") - stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z") - yield f' \n' - yield f" {html.escape(program['title'])}\n" - if program.get('sub_title'): - yield f" {html.escape(program['sub_title'])}\n" - yield f" {html.escape(program['description'])}\n" - custom_data = program.get('custom_properties', {}) - if 'categories' in custom_data: - for cat in custom_data['categories']: - yield f" {html.escape(cat)}\n" - if 'date' in custom_data: - yield f" {html.escape(custom_data['date'])}\n" - if custom_data.get('live', False): - yield f" \n" - if custom_data.get('new', False): - yield f" \n" - if 'icon' in custom_data: - yield f' \n' - yield f" \n" + if channel_xml_batch: + yield '\n'.join(channel_xml_batch) + '\n' + + del channels + del channel_num_map + + batch_size = _EPG_PROGRAM_YIELD_BATCH_SIZE - # Emit real programmes: single bulk query, chunked to avoid server-side cursor issues. all_epg_ids = list(real_epg_map.keys()) if all_epg_ids: if num_days > 0: @@ -1682,27 +1700,44 @@ def generate_epg(request, profile_name=None, user=None): current_epg_id = None channel_ids_for_epg = None - is_multi = False - multi_buffer = [] + pending = [] program_batch = [] - batch_size = 1000 - chunk_size = 5000 - # Keyset pagination: track last (epg_id, id) instead of OFFSET - # to avoid skipping/duplicating rows if the table changes mid-stream. + chunk_size = _EPG_PROGRAM_DB_CHUNK_SIZE last_epg_id = 0 last_id = 0 _poster_url_base = build_absolute_uri_with_port(request, "/api/epg/programs/") + def flush_pending(): + nonlocal program_batch, pending + if not pending: + return + pending.sort(key=lambda row: (row[0], row[1])) + escaped_primary = ( + html.escape(channel_ids_for_epg[0]) + if len(channel_ids_for_epg) > 1 else None + ) + for _, _, xml_text in pending: + program_batch.append(xml_text) + if escaped_primary: + for cid in channel_ids_for_epg[1:]: + program_batch.append(xml_text.replace( + f'channel="{escaped_primary}"', + f'channel="{html.escape(cid)}"', + 1, + )) + if len(program_batch) >= batch_size: + yield '\n'.join(program_batch) + '\n' + program_batch = [] + pending.clear() + while True: program_chunk = list( programs_base_qs.filter(epg_id__gte=last_epg_id) .exclude(epg_id=last_epg_id, id__lte=last_id)[:chunk_size] ) - if not program_chunk: break - # Advance keyset cursor to last row in this chunk last_row = program_chunk[-1] last_epg_id = last_row['epg_id'] last_id = last_row['id'] @@ -1710,31 +1745,19 @@ def generate_epg(request, profile_name=None, user=None): for prog in program_chunk: epg_id = prog['epg_id'] - # When epg_id changes, flush multi-channel buffer for previous group if epg_id != current_epg_id: - if is_multi and multi_buffer: - escaped_primary = html.escape(channel_ids_for_epg[0]) - for extra_cid in channel_ids_for_epg[1:]: - escaped_extra = html.escape(extra_cid) - for xml_text in multi_buffer: - program_batch.append(xml_text.replace( - f'channel="{escaped_primary}"', - f'channel="{escaped_extra}"', - 1, - )) - if len(program_batch) >= batch_size: - yield '\n'.join(program_batch) + '\n' - program_batch = [] - multi_buffer = [] - + yield from flush_pending() current_epg_id = epg_id channel_ids_for_epg = real_epg_map[epg_id] - is_multi = len(channel_ids_for_epg) > 1 - # Build programme XML for primary channel_id primary_cid = channel_ids_for_epg[0] - start_str = prog['start_time'].strftime("%Y%m%d%H%M%S %z") - stop_str = prog['end_time'].strftime("%Y%m%d%H%M%S %z") + # DB datetimes are UTC (USE_TZ=True, TIME_ZONE=UTC); format + # directly instead of strftime("%Y%m%d%H%M%S %z"), which is + # ~10x slower and dominates XML build over 750k rows. + st = prog['start_time'] + et = prog['end_time'] + start_str = f"{st.year:04d}{st.month:02d}{st.day:02d}{st.hour:02d}{st.minute:02d}{st.second:02d} +0000" + stop_str = f"{et.year:04d}{et.month:02d}{et.day:02d}{et.hour:02d}{et.minute:02d}{et.second:02d} +0000" program_xml = [f' '] program_xml.append(f' {html.escape(prog["title"])}') @@ -1932,67 +1955,101 @@ def generate_epg(request, profile_name=None, user=None): program_xml.append(" ") xml_text = '\n'.join(program_xml) - program_batch.append(xml_text) + pending.append((prog['start_time'], prog['id'], xml_text)) - if is_multi: - multi_buffer.append(xml_text) + del program_chunk - if len(program_batch) >= batch_size: - yield '\n'.join(program_batch) + '\n' - program_batch = [] - - # Final flush of multi-channel buffer - if is_multi and multi_buffer: - escaped_primary = html.escape(channel_ids_for_epg[0]) - for extra_cid in channel_ids_for_epg[1:]: - escaped_extra = html.escape(extra_cid) - for xml_text in multi_buffer: - program_batch.append(xml_text.replace( - f'channel="{escaped_primary}"', - f'channel="{escaped_extra}"', - 1, - )) + yield from flush_pending() if program_batch: yield '\n'.join(program_batch) + '\n' - # Send final closing tag and completion message + del real_epg_map + + for channel_id, pattern_match_name, epg_source in dummy_program_list: + program_length_hours = 4 + dummy_programs = generate_dummy_programs( + channel_id, pattern_match_name, + num_days=dummy_days, + program_length_hours=program_length_hours, + epg_source=epg_source, + export_lookback=lookback_cutoff, + export_cutoff=cutoff_date, + ) + if not dummy_programs: + continue + dummy_batch = [] + for program in dummy_programs: + start_str = program['start_time'].strftime("%Y%m%d%H%M%S %z") + stop_str = program['end_time'].strftime("%Y%m%d%H%M%S %z") + lines = [ + f' ', + f" {html.escape(program['title'])}", + ] + if program.get('sub_title'): + lines.append(f" {html.escape(program['sub_title'])}") + lines.append(f" {html.escape(program['description'])}") + custom_data = program.get('custom_properties', {}) + if 'categories' in custom_data: + for cat in custom_data['categories']: + lines.append(f" {html.escape(cat)}") + if 'date' in custom_data: + lines.append(f" {html.escape(custom_data['date'])}") + if custom_data.get('live', False): + lines.append(" ") + if custom_data.get('new', False): + lines.append(" ") + if 'icon' in custom_data: + lines.append(f' ') + lines.append(" ") + dummy_batch.append('\n'.join(lines)) + if len(dummy_batch) >= batch_size: + yield '\n'.join(dummy_batch) + '\n' + dummy_batch = [] + del dummy_programs + if dummy_batch: + yield '\n'.join(dummy_batch) + '\n' + + del dummy_program_list + yield "\n" - # Log system event for EPG download after streaming completes (with deduplication based on client) client_id, client_ip, user_agent = get_client_identifier(request) event_cache_key = f"epg_download:{user.username if user else 'anonymous'}:{profile_name or 'all'}:{client_id}" - if not cache.get(event_cache_key): - # `len()` reuses the queryset's iteration cache populated above; - # `count()` would issue a separate SELECT COUNT(*). - log_system_event( - event_type='epg_download', - profile=profile_name or 'all', - user=user.username if user else 'anonymous', - channels=len(channels), - client_ip=client_ip, - user_agent=user_agent, - ) - cache.set(event_cache_key, True, 2) # Prevent duplicate events for 2 seconds - # Wrapper generator that collects content for caching - def caching_generator(): - collected_content = [] - for chunk in epg_generator(): - collected_content.append(chunk) - yield chunk - # After streaming completes, cache the full content - full_content = ''.join(collected_content) - cache.set(content_cache_key, full_content, 300) - logger.debug("Cached EPG content (%d bytes)", len(full_content)) + def _log_epg_download(): + from django.core.cache import cache as event_cache - response = StreamingHttpResponse( - streaming_content=caching_generator(), - content_type="application/xml" - ) - response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"' - response["Cache-Control"] = "no-cache" - return response + if not event_cache.get(event_cache_key): + log_system_event( + event_type='epg_download', + profile=profile_name or 'all', + user=user.username if user else 'anonymous', + channels=channel_count, + client_ip=client_ip, + user_agent=user_agent, + ) + event_cache.set(event_cache_key, True, 2) + + try: + from core.utils import _is_gevent_monkey_patched + + if _is_gevent_monkey_patched(): + import gevent + + gevent.spawn(_log_epg_download) + else: + _log_epg_download() + except Exception: + _log_epg_download() + + def build_epg_stream(): + try: + yield from epg_generator() + finally: + _epg_export_teardown() + + return stream_epg_response(content_cache_key, build_epg_stream) def xc_get_user(request): diff --git a/core/tests.py b/core/tests.py index 7b2f1986..33475ae7 100644 --- a/core/tests.py +++ b/core/tests.py @@ -1,6 +1,6 @@ from unittest.mock import patch, MagicMock -from django.test import TestCase +from django.test import TestCase, SimpleTestCase from apps.epg.models import EPGSource from core.models import CoreSettings, DVR_SETTINGS_KEY, EPG_SETTINGS_KEY @@ -301,3 +301,14 @@ class DropDBCommandTlsTest(TestCase): host='localhost', port=5432, autocommit=True, ) + + +class MallocTrimTests(SimpleTestCase): + def test_trim_is_noop_when_libc_has_no_malloc_trim(self): + from core.utils import trim_c_allocator_heap + + fake_libc = MagicMock(spec=[]) + with patch('ctypes.util.find_library', return_value='libc.so.6'), patch( + 'ctypes.CDLL', return_value=fake_libc + ): + self.assertFalse(trim_c_allocator_heap()) diff --git a/core/utils.py b/core/utils.py index 29c613b3..89228088 100644 --- a/core/utils.py +++ b/core/utils.py @@ -546,6 +546,25 @@ def monitor_memory_usage(func): return result return wrapper +def trim_c_allocator_heap(): + """Return unused C heap pages to the OS where supported (glibc malloc_trim).""" + try: + import ctypes + import ctypes.util + + libc_name = ctypes.util.find_library("c") + if not libc_name: + return False + libc = ctypes.CDLL(libc_name) + if not hasattr(libc, "malloc_trim"): + return False + libc.malloc_trim(0) + return True + except Exception: + logger.debug("malloc_trim unavailable or failed", exc_info=True) + return False + + def cleanup_memory(log_usage=False, force_collection=True): """ Comprehensive memory cleanup function to reduce memory footprint