Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-18 00:55:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(epg): extract EPG generation logic into dedicated module

Browse source 
- Moved all XMLTV output logic from `apps/output/views.py` to `apps/output/epg.py`, streamlining the codebase and maintaining the existing HTTP endpoint functionality.
- Updated related tests and references to ensure proper integration with the new module structure.
This commit is contained in:
SergeantPanda 2026-06-23 11:50:52 -05:00
parent 3868f02c45
commit bccee9ebc1
6 changed files with 1790 additions and 1748 deletions
Download patch file Download diff file Expand all files Collapse all files

2
CHANGELOG.md
View file

@ -15,6 +15,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- **EPG generation extracted into `apps/output/epg.py`.** All XMLTV output logic (`generate_epg`, `generate_dummy_programs`, `generate_custom_dummy_programs`, `generate_dummy_epg`, and supporting helpers) moved from `apps/output/views.py` into a dedicated module. `views.py` retains the thin HTTP endpoint wrappers and auth checks; `epg.py` handles all content generation. No behavior change.
- **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.
### Performance

1745
apps/output/epg.py Normal file
View file

File diff suppressed because it is too large Load diff

36
apps/output/epg_chunk_cache.py → apps/output/streaming_chunk_cache.py
View file

@ -1,4 +1,4 @@
"""Single-flight Redis chunk cache for XMLTV EPG streaming responses."""
"""Single-flight Redis chunk cache for large streaming HTTP responses."""
import logging
import time
@ -111,7 +111,7 @@ def _stream_build(redis, base_key, source, cache_ttl, lock_ttl):
try:
from django.core.cache import cache as django_cache
django_cache.delete(base_key) # legacy monolithic entry
django_cache.delete(base_key) # clear any non-chunked entry under this key
redis.delete(chunks_key, _ready_key(base_key))
redis.set(status_key, STATUS_BUILDING, ex=lock_ttl)
for chunk in source():
@ -123,9 +123,9 @@ def _stream_build(redis, base_key, source, cache_ttl, lock_ttl):
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))
logger.debug("Cached response in %s chunks", redis.llen(chunks_key))
except Exception:
logger.exception("EPG cache build failed for %s", base_key)
logger.exception("Chunk cache build failed for %s", base_key)
redis.delete(chunks_key)
redis.set(status_key, STATUS_ERROR, ex=60)
raise
@ -158,14 +158,14 @@ def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval,
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")
raise RuntimeError("Chunk 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)
logger.warning("Chunk 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)
logger.warning("Chunk cache follower timed out after partial read for %s", base_key)
break
lock_active = bool(redis.exists(lock_key))
@ -173,7 +173,7 @@ def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval,
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)
logger.warning("Chunk cache leader lost; rebuilding %s", base_key)
yield from _stream_build(redis, base_key, source, cache_ttl, lock_ttl)
return
else:
@ -182,10 +182,12 @@ def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval,
_poll_wait(poll_interval)
def stream_epg_response(
def stream_cached_response(
cache_key,
source,
*,
content_type="application/xml",
filename=None,
cache_ttl=DEFAULT_CACHE_TTL,
lock_ttl=DEFAULT_LOCK_TTL,
poll_interval=DEFAULT_POLL_INTERVAL,
@ -193,16 +195,17 @@ def stream_epg_response(
redis=None,
):
"""
Stream XMLTV EPG output with single-flight Redis chunk caching.
Stream a large response 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.
invokes it; concurrent followers replay chunks already written to Redis, so
the expensive ``source`` runs at most once per ``cache_key``.
"""
if redis is None:
redis = _get_redis()
if redis.get(_ready_key(cache_key)):
logger.debug("Serving EPG from chunk cache")
logger.debug("Serving response from chunk cache")
stream = _stream_ready(redis, cache_key)
else:
status = _get_status(redis, cache_key)
@ -210,10 +213,10 @@ def stream_epg_response(
_clear_build_keys(redis, cache_key)
if _try_acquire_lock(redis, cache_key, lock_ttl):
logger.debug("Building EPG (cache leader)")
logger.debug("Building response (cache leader)")
stream = _stream_build(redis, cache_key, source, cache_ttl, lock_ttl)
else:
logger.debug("Following in-flight EPG build")
logger.debug("Following in-flight cache build")
stream = _stream_follow(
redis,
cache_key,
@ -224,7 +227,8 @@ def stream_epg_response(
max_follower_wait,
)
response = StreamingHttpResponse(stream, content_type="application/xml")
response["Content-Disposition"] = 'attachment; filename="Dispatcharr.xml"'
response = StreamingHttpResponse(stream, content_type=content_type)
if filename:
response["Content-Disposition"] = f'attachment; filename="{filename}"'
response["Cache-Control"] = "no-cache"
return response

30
apps/output/test_epg_chunk_cache.py → apps/output/test_streaming_chunk_cache.py
View file

@ -2,14 +2,14 @@ import threading
import time
from unittest import TestCase
from apps.output.epg_chunk_cache import (
from apps.output.streaming_chunk_cache import (
STATUS_BUILDING,
STATUS_READY,
_chunks_key,
_lock_key,
_ready_key,
_status_key,
stream_epg_response,
stream_cached_response,
)
@ -74,7 +74,7 @@ def _consume(response):
return b"".join(response.streaming_content).decode("utf-8")
class EPGChunkCacheTests(TestCase):
class StreamingChunkCacheTests(TestCase):
def test_leader_caches_chunks_and_sets_ready(self):
redis = FakeRedis()
calls = []
@ -84,14 +84,14 @@ class EPGChunkCacheTests(TestCase):
yield "<tv>"
yield "</tv>"
body = _consume(stream_epg_response("epg:test", source, redis=redis))
body = _consume(stream_cached_response("cache:test", source, redis=redis))
self.assertEqual(body, "<tv></tv>")
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")))
self.assertEqual(redis.get(_ready_key("cache:test")), "1")
self.assertEqual(redis.get(_status_key("cache:test")), STATUS_READY)
self.assertEqual(redis.llen(_chunks_key("cache:test")), 2)
self.assertFalse(redis.exists(_lock_key("cache:test")))
def test_cache_hit_skips_source(self):
redis = FakeRedis()
@ -102,16 +102,16 @@ class EPGChunkCacheTests(TestCase):
yield "<tv>"
yield "</tv>"
_consume(stream_epg_response("epg:test", source, redis=redis))
_consume(stream_cached_response("cache:test", source, redis=redis))
calls.clear()
body = _consume(stream_epg_response("epg:test", source, redis=redis))
body = _consume(stream_cached_response("cache:test", source, redis=redis))
self.assertEqual(body, "<tv></tv>")
self.assertEqual(calls, [])
def test_follower_reads_leader_chunks_without_rebuilding(self):
redis = FakeRedis()
base = "epg:follow"
base = "cache:follow"
leader_started = threading.Event()
rebuild_calls = []
@ -128,7 +128,7 @@ class EPGChunkCacheTests(TestCase):
def leader():
_consume(
stream_epg_response(
stream_cached_response(
base,
slow_source,
redis=redis,
@ -140,7 +140,7 @@ class EPGChunkCacheTests(TestCase):
leader_thread.start()
leader_started.wait(timeout=5)
follower_body = _consume(
stream_epg_response(
stream_cached_response(
base,
forbidden_source,
redis=redis,
@ -165,8 +165,8 @@ class EPGChunkCacheTests(TestCase):
def worker():
barrier.wait()
results[threading.current_thread().name] = _consume(
stream_epg_response(
"epg:race",
stream_cached_response(
"cache:race",
source,
redis=redis,
poll_interval=0.01,

11
apps/output/tests.py
View file

@ -34,16 +34,18 @@ class OutputEndpointTestMixin:
"apps.output.views.network_access_allowed",
return_value=True,
)
self._epg_teardown_patch = patch("apps.output.views._epg_export_teardown")
self._epg_teardown_patch = patch("apps.output.epg._epg_export_teardown")
self._log_event_patch = patch("apps.output.views.log_system_event")
self._epg_log_event_patch = patch("apps.output.epg.log_system_event")
self._close_db_patch = patch("django.db.close_old_connections")
self._epg_cache_patch = patch(
"apps.output.views.stream_epg_response",
"apps.output.epg.stream_cached_response",
side_effect=_epg_response_without_redis,
)
self._network_patch.start()
self._epg_teardown_patch.start()
self._log_event_patch.start()
self._epg_log_event_patch.start()
self._close_db_patch.start()
self._epg_cache_patch.start()
@ -53,6 +55,7 @@ class OutputEndpointTestMixin:
cache.clear()
self._epg_cache_patch.stop()
self._close_db_patch.stop()
self._epg_log_event_patch.stop()
self._log_event_patch.stop()
self._epg_teardown_patch.stop()
self._network_patch.stop()
@ -339,14 +342,14 @@ class OutputEPGCustomDummyTest(TestCase):
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
from apps.output.epg 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
from apps.output.epg import _ceil_to_half_hour
dt = timezone.now().replace(minute=17, second=42, microsecond=123456)
aligned = _ceil_to_half_hour(dt)

1714
apps/output/views.py
View file

File diff suppressed because it is too large Load diff