refactor(tasks): Optimize M3U stream processing by pre-compiling filters for batch workers, reducing redundant regex evaluations. Update related functions to improve performance and memory management during account refreshes.

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

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")

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."""

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()