diff --git a/CHANGELOG.md b/CHANGELOG.md index ce80b8e5..82058d39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Performance - **M3U/XC stream refresh is faster on large accounts.** Steady-state refreshes split `bulk_update` into a lightweight touch pass (`last_seen` / `is_stale` only) for unchanged streams and a full column update only when provider metadata or catch-up fields actually change. +- **M3U stream filters compile once per refresh.** Account filters are regex-compiled before batch workers start; accounts with no filters skip per-stream filter checks entirely. - **Celery workers return RSS after memory-intensive tasks.** `cleanup_memory()` accepts an optional `trim_heap` flag (glibc `malloc_trim`); Celery `task_postrun` enables it after `close_old_connections()` for M3U account/group refresh, EPG, VOD, and channel-matching tasks so worker memory drops back toward baseline instead of ratcheting across successive large jobs. ### Fixed diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index 7ef71dda..3818683a 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -975,6 +975,39 @@ def collect_xc_streams(account_id, enabled_groups): return all_streams +def _compile_m3u_stream_filters(filter_queryset): + """Compile account M3UFilter rows once per refresh for batch workers.""" + compiled = [] + for filter_obj in filter_queryset: + flags = ( + re.IGNORECASE + if (filter_obj.custom_properties or {}).get("case_sensitive", True) is False + else 0 + ) + compiled.append((re.compile(filter_obj.regex_pattern, flags), filter_obj)) + return compiled + + +def _stream_passes_m3u_filters(name, url, group_title, compiled_filters): + """Return False when the first matching filter excludes the stream.""" + for pattern, filter_obj in compiled_filters: + logger.trace("Checking filter pattern %s", pattern.pattern) + if filter_obj.filter_type == "url": + target = url + elif filter_obj.filter_type == "group": + target = group_title + else: + target = name + + if pattern.search(target or ""): + logger.debug( + "Stream %s - %s matches filter pattern %s", + name, url, filter_obj.regex_pattern, + ) + return not filter_obj.exclude + return True + + _STREAM_TOUCH_FIELDS = ("last_seen", "is_stale") _STREAM_CHANGED_FIELDS = ( "name", "url", "logo_url", "tvg_id", "custom_properties", "is_adult", @@ -1192,8 +1225,12 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys): return retval -def process_m3u_batch_direct(account_id, batch, groups, hash_keys): - """Processes a batch of M3U streams using bulk operations with thread-safe DB connections.""" +def process_m3u_batch_direct(account_id, batch, groups, hash_keys, compiled_filters=None): + """Processes a batch of M3U streams using bulk operations with thread-safe DB connections. + + ``compiled_filters`` should be pre-built once per account refresh and shared + across batch workers. Pass an empty list when the account has no filters. + """ from django.db import connections # Ensure clean database connections for threading @@ -1201,23 +1238,8 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys): account = M3UAccount.objects.get(id=account_id) - compiled_filters = [ - ( - re.compile( - f.regex_pattern, - ( - re.IGNORECASE - if (f.custom_properties or {}).get( - "case_sensitive", True - ) - == False - else 0 - ), - ), - f, - ) - for f in account.filters.order_by("order") - ] + if compiled_filters is None: + compiled_filters = _compile_m3u_stream_filters(account.filters.order_by("order")) streams_to_create = [] streams_to_update = [] @@ -1228,7 +1250,10 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys): logger.debug(f"Processing batch of {len(batch)} for M3U account {account_id}") if compiled_filters: - logger.debug(f"Using compiled filters: {[f[1].regex_pattern for f in compiled_filters]}") + logger.debug( + "Using compiled filters: %s", + [filter_obj.regex_pattern for _, filter_obj in compiled_filters], + ) for stream_info in batch: try: name, url = stream_info["name"], stream_info["url"] @@ -1249,25 +1274,12 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys): group_title = get_case_insensitive_attr( stream_info["attributes"], "group-title", "Default Group" ) - logger.debug(f"Processing stream: {name} - {url} in group {group_title}") - include = True - for pattern, filter in compiled_filters: - logger.trace(f"Checking filter pattern {pattern}") - target = name - if filter.filter_type == "url": - target = url - elif filter.filter_type == "group": - target = group_title + logger.trace("Processing stream: %s - %s in group %s", name, url, group_title) - if pattern.search(target or ""): - logger.debug( - f"Stream {name} - {url} matches filter pattern {filter.regex_pattern}" - ) - include = not filter.exclude - break - - if not include: - logger.debug(f"Stream excluded by filter, skipping.") + if compiled_filters and not _stream_passes_m3u_filters( + name, url, group_title, compiled_filters, + ): + logger.debug("Stream excluded by filter, skipping.") continue # Filter out disabled groups for this account @@ -3322,7 +3334,15 @@ def _refresh_single_m3u_account_impl(account_id): ) account = _get_active_m3u_account(account_id) - filters = list(account.filters.all()) + compiled_stream_filters = _compile_m3u_stream_filters( + account.filters.order_by("order") + ) + if compiled_stream_filters: + logger.debug( + "Account %s has %s stream filter(s) for this refresh", + account_id, + len(compiled_stream_filters), + ) # Check if VOD is enabled for this account vod_enabled = ensure_custom_properties_dict(account.custom_properties).get( @@ -3499,7 +3519,14 @@ def _refresh_single_m3u_account_impl(account_id): with ThreadPoolExecutor(max_workers=max_workers) as executor: # Submit batch processing tasks using direct functions (now thread-safe) future_to_batch = { - executor.submit(process_m3u_batch_direct, account_id, batch, existing_groups, hash_keys): i + executor.submit( + process_m3u_batch_direct, + account_id, + batch, + existing_groups, + hash_keys, + compiled_stream_filters, + ): i for i, batch in enumerate(batches) } @@ -3612,7 +3639,14 @@ def _refresh_single_m3u_account_impl(account_id): with ThreadPoolExecutor(max_workers=max_workers) as executor: # Submit stream batch processing tasks (reuse standard M3U processing) future_to_batch = { - executor.submit(process_m3u_batch_direct, account_id, batch, existing_groups, hash_keys): i + executor.submit( + process_m3u_batch_direct, + account_id, + batch, + existing_groups, + hash_keys, + compiled_stream_filters, + ): i for i, batch in enumerate(batches) } @@ -3819,6 +3853,8 @@ def _refresh_single_m3u_account_impl(account_id): del filtered_groups if 'channel_group_relationships' in locals(): del channel_group_relationships + if 'compiled_stream_filters' in locals(): + del compiled_stream_filters # Remove cache file after processing (success or failure) cache_path = os.path.join(m3u_dir, f"{account_id}.json") diff --git a/apps/m3u/tests/test_memory_cleanup.py b/apps/m3u/tests/test_memory_cleanup.py index bfc964d0..b29ca947 100644 --- a/apps/m3u/tests/test_memory_cleanup.py +++ b/apps/m3u/tests/test_memory_cleanup.py @@ -30,7 +30,7 @@ class ProcessM3UBatchCleanupTests(SimpleTestCase): mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123") with patch("django.db.connections") as mock_connections: - process_m3u_batch_direct(1, [], {}, ["name", "url"]) + process_m3u_batch_direct(1, [], {}, ["name", "url"], compiled_filters=[]) mock_connections.close_all.assert_called() @patch("apps.m3u.tasks.Stream") @@ -48,9 +48,29 @@ class ProcessM3UBatchCleanupTests(SimpleTestCase): mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123") with patch("gc.collect") as mock_gc, patch("django.db.connections"): - process_m3u_batch_direct(1, [], {}, ["name", "url"]) + process_m3u_batch_direct(1, [], {}, ["name", "url"], compiled_filters=[]) mock_gc.assert_called() + @patch("apps.m3u.tasks.Stream") + @patch("apps.m3u.tasks.M3UAccount") + def test_precompiled_empty_filters_skip_db_lookup( + self, mock_account_cls, mock_stream_cls, + ): + """When filters are precompiled as empty, batch workers must not query filters.""" + from apps.m3u.tasks import process_m3u_batch_direct + + mock_account = MagicMock() + mock_account_cls.objects.get.return_value = mock_account + mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = ( + [] + ) + mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123") + + with patch("django.db.connections"): + process_m3u_batch_direct(1, [], {}, ["name", "url"], compiled_filters=[]) + + mock_account.filters.order_by.assert_not_called() + class LockReleaseTests(SimpleTestCase): """Verify task lock is released on all exit paths.""" diff --git a/apps/m3u/tests/test_stream_filters.py b/apps/m3u/tests/test_stream_filters.py new file mode 100644 index 00000000..aba1a262 --- /dev/null +++ b/apps/m3u/tests/test_stream_filters.py @@ -0,0 +1,156 @@ +"""Tests for M3U stream filter compilation and batch application.""" +from unittest.mock import MagicMock, patch + +from django.test import SimpleTestCase + +from apps.m3u.tasks import ( + _compile_m3u_stream_filters, + _stream_passes_m3u_filters, + process_m3u_batch_direct, +) + + +class CompileM3UStreamFiltersTests(SimpleTestCase): + def test_compiles_case_insensitive_when_configured(self): + filter_obj = MagicMock() + filter_obj.regex_pattern = "news" + filter_obj.custom_properties = {"case_sensitive": False} + + compiled = _compile_m3u_stream_filters([filter_obj]) + + self.assertEqual(len(compiled), 1) + pattern, _ = compiled[0] + self.assertTrue(pattern.search("NEWS")) + + def test_compiles_case_sensitive_by_default(self): + filter_obj = MagicMock() + filter_obj.regex_pattern = "news" + filter_obj.custom_properties = {} + + compiled = _compile_m3u_stream_filters([filter_obj]) + + pattern, _ = compiled[0] + self.assertIsNone(pattern.search("NEWS")) + self.assertTrue(pattern.search("news")) + + +class StreamPassesM3UFiltersTests(SimpleTestCase): + def _compiled(self, *, filter_type="name", exclude=False, pattern="Adult"): + filter_obj = MagicMock() + filter_obj.filter_type = filter_type + filter_obj.exclude = exclude + filter_obj.regex_pattern = pattern + filter_obj.custom_properties = {} + return _compile_m3u_stream_filters([filter_obj]) + + def test_include_filter_passes_matching_stream(self): + compiled = self._compiled(exclude=False) + self.assertTrue( + _stream_passes_m3u_filters("Adult Channel", "http://x", "News", compiled) + ) + + def test_include_filter_passes_non_matching_stream(self): + """Non-matching streams still pass unless a matching exclude filter hits.""" + compiled = self._compiled(exclude=False, pattern="news") + self.assertTrue( + _stream_passes_m3u_filters("Sports", "http://x", "Sports", compiled) + ) + + def test_exclude_filter_rejects_matching_stream(self): + compiled = self._compiled(exclude=True, pattern="Adult") + self.assertFalse( + _stream_passes_m3u_filters("Adult Channel", "http://x", "News", compiled) + ) + + def test_url_filter_type_targets_url(self): + compiled = self._compiled(filter_type="url", exclude=True, pattern="blocked") + self.assertFalse( + _stream_passes_m3u_filters("OK", "http://blocked.example/live", "News", compiled) + ) + self.assertTrue( + _stream_passes_m3u_filters("blocked name", "http://ok.example/live", "News", compiled) + ) + + def test_group_filter_type_targets_group(self): + compiled = self._compiled(filter_type="group", exclude=True, pattern="Hidden") + self.assertFalse( + _stream_passes_m3u_filters("Channel", "http://x", "Hidden Group", compiled) + ) + + +class ProcessM3UBatchFilterTests(SimpleTestCase): + def _mock_stream_meta(self, mock_stream_cls, max_length=255): + mock_field = MagicMock() + mock_field.max_length = max_length + mock_stream_cls._meta.get_field.return_value = mock_field + + @patch("apps.m3u.tasks._bulk_update_stream_refresh_batches") + @patch("apps.m3u.tasks.Stream") + @patch("apps.m3u.tasks.M3UAccount") + def test_exclude_filter_skips_stream_import( + self, mock_account_cls, mock_stream_cls, mock_bulk_update, + ): + self._mock_stream_meta(mock_stream_cls) + mock_account = MagicMock() + mock_account.account_type = "STD" + mock_account_cls.objects.get.return_value = mock_account + mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = ( + [] + ) + mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123") + + filter_obj = MagicMock() + filter_obj.regex_pattern = "skip-me" + filter_obj.filter_type = "name" + filter_obj.exclude = True + filter_obj.custom_properties = {} + compiled = _compile_m3u_stream_filters([filter_obj]) + + batch = [{ + "name": "skip-me channel", + "url": "http://example/live", + "attributes": {"group-title": "News"}, + "vlc_opts": {}, + }] + + with patch("django.db.connections"): + result = process_m3u_batch_direct( + 1, batch, {"News": 1}, ["name", "url"], compiled_filters=compiled, + ) + + self.assertIn("0 created", result) + mock_stream_cls.objects.bulk_create.assert_not_called() + mock_bulk_update.assert_called_once_with([], [], batch_size=200) + + @patch("apps.m3u.tasks._bulk_update_stream_refresh_batches") + @patch("apps.m3u.tasks.Stream") + @patch("apps.m3u.tasks.M3UAccount") + def test_no_filters_imports_matching_stream( + self, mock_account_cls, mock_stream_cls, mock_bulk_update, + ): + self._mock_stream_meta(mock_stream_cls) + mock_account = MagicMock() + mock_account.account_type = "STD" + mock_account_cls.objects.get.return_value = mock_account + mock_stream_cls.objects.filter.return_value.select_related.return_value.only.return_value = ( + [] + ) + mock_stream_cls.generate_hash_key = MagicMock(return_value="hash123") + mock_stream_cls.objects.bulk_create.return_value = [] + + batch = [{ + "name": "News One", + "url": "http://example/live", + "attributes": {"group-title": "News"}, + "vlc_opts": {}, + }] + + with patch("django.db.connections"), patch( + "apps.m3u.tasks.transaction.atomic", + ): + result = process_m3u_batch_direct( + 1, batch, {"News": 1}, ["name", "url"], compiled_filters=[], + ) + + self.assertIn("1 created", result) + mock_stream_cls.objects.bulk_create.assert_called_once()