feat(epg): Implement channel parsing progress tracking and update progress reporting logic in parse_channels_only function

CHANGELOG.md

@@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **EPG channel parsing progress no longer exceeds 100%.** During `parsing_channels`, progress used the pre-parse database channel count as the denominator while the numerator counted channels found in the XML. When the guide grew (or on sources where the file had more channels than the DB), the UI could show values well above 100% (e.g. ~300%). The estimate now bumps when the XML exceeds the DB baseline, progress is capped at 98% until final cleanup, and update frequency adapts to the known total (~20 ticks on steady-state refreshes; coarse every-100-channel updates for first import or when the file outgrows the estimate). A final in-loop update reports the true parsed count before completion.
 - **M3U refresh completion counts now reflect actual stream changes.** The "updated" count previously included every existing stream because the summary treated `last_seen` touch-only rows as updates; it now counts only streams whose provider metadata changed. "Total processed" includes unchanged streams separately. The completion message, WebSocket payload, and parsing notification now also report how many streams were **marked stale** (missing from this refresh, pending retention-gated deletion) versus **removed** (deleted this run).
 - **M3U group processing retries on poisoned DB connections.** `_db_query_with_retry` now treats psycopg desync errors (`DatabaseError`, e.g. `lost synchronization with server`) as transient; `process_groups` relationship loading uses it so a stale Celery worker connection resets once instead of failing the whole refresh.
 - **Auto Channel Sync's Find and Replace preview now matches the rename it performs.** The preview rendered the literal `$1` instead of substituting numbered capture groups, and compiled patterns with a stricter engine than the rename, so patterns like `^*` previewed a change the sync silently skipped. The live rename now uses the same `regex` engine as the preview (with a substitution timeout to bound catastrophic backtracking), both apply the `$1`→`\1` conversion through one shared helper, and a rename whose result exceeds the channel-name column length is truncated instead of aborting the whole sync. (Fixes #1332) — Thanks [@CodeBormen](https://github.com/CodeBormen)

apps/epg/tasks.py

@@ -1197,6 +1197,35 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False):
         logger.error(f"Error extracting {file_path}: {str(e)}", exc_info=True)
         return None
 
+_CHANNEL_PARSE_PROGRESS_START = 25
+_CHANNEL_PARSE_PROGRESS_CAP = 98
+_CHANNEL_PARSE_PROGRESS_SCALE = 0.98
+
+
+def _channel_parse_progress(processed_channels, channel_estimate, had_db_baseline):
+    """Map channel parsing onto 25-98%. Bump estimate when the XML has grown."""
+    if not had_db_baseline:
+        if processed_channels <= 0:
+            return _CHANNEL_PARSE_PROGRESS_START, channel_estimate
+        return (
+            min(
+                _CHANNEL_PARSE_PROGRESS_START + (processed_channels // 100),
+                _CHANNEL_PARSE_PROGRESS_CAP - 1,
+            ),
+            channel_estimate,
+        )
+
+    if processed_channels > channel_estimate:
+        channel_estimate = processed_channels
+    if channel_estimate <= 0:
+        return _CHANNEL_PARSE_PROGRESS_START, channel_estimate
+
+    span = _CHANNEL_PARSE_PROGRESS_CAP - _CHANNEL_PARSE_PROGRESS_START
+    progress = _CHANNEL_PARSE_PROGRESS_START + int(
+        (processed_channels / channel_estimate) * span * _CHANNEL_PARSE_PROGRESS_SCALE
+    )
+    return min(_CHANNEL_PARSE_PROGRESS_CAP, progress), channel_estimate
+
 
 def parse_channels_only(source):
     # Use extracted file if available, otherwise use the original file path
@@ -1309,6 +1338,8 @@ def parse_channels_only(source):
         epgs_to_create = []
         epgs_to_update = []
         total_channels = 0
+        channel_estimate = 0
+        had_db_baseline = False
         processed_channels = 0
         batch_size = 500  # Process in batches to limit memory usage
         progress = 0  # Initialize progress variable here
@@ -1323,10 +1354,14 @@ def parse_channels_only(source):
             # Attempt to count existing channels in the database
             try:
                 total_channels = EPGData.objects.filter(epg_source=source).count()
+                channel_estimate = total_channels
+                had_db_baseline = total_channels > 0
                 logger.info(f"Found {total_channels} existing channels for this source")
             except Exception as e:
                 logger.error(f"Error counting channels: {e}")
                 total_channels = 500  # Default estimate
+                channel_estimate = total_channels
+                had_db_baseline = True
             if process:
                 logger.debug(f"[parse_channels_only] Memory after closing initial file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
 
@@ -1453,20 +1488,36 @@ def parse_channels_only(source):
                         if process:
                             logger.info(f"[parse_channels_only] Memory after clearing cache: {process.memory_info().rss / 1024 / 1024:.2f} MB")
 
-                    # Send progress updates
-                    if processed_channels % 100 == 0 or processed_channels == total_channels:
-                        progress = 25 + int((processed_channels / total_channels) * 65) if total_channels > 0 else 90
+                    # Send progress updates (~20 ticks when the DB count still holds; else every 100)
+                    estimate_exceeded = had_db_baseline and processed_channels > total_channels
+                    interval = (
+                        max(1, total_channels // 20)
+                        if had_db_baseline and total_channels and not estimate_exceeded
+                        else 100
+                    )
+                    if processed_channels % interval == 0:
+                        progress, channel_estimate = _channel_parse_progress(
+                            processed_channels,
+                            channel_estimate,
+                            had_db_baseline,
+                        )
                         send_epg_update(
                             source.id,
                             "parsing_channels",
                             progress,
                             processed=processed_channels,
-                            total=total_channels
+                            total=channel_estimate,
+                        )
+                    if had_db_baseline and processed_channels > total_channels:
+                        logger.debug(
+                            f"[parse_channels_only] Processed channel {tvg_id} - "
+                            f"processed {processed_channels - total_channels} additional channels"
                         )
-                    if processed_channels > total_channels:
-                        logger.debug(f"[parse_channels_only] Processed channel {tvg_id} - processed {processed_channels - total_channels} additional channels")
                     else:
-                        logger.debug(f"[parse_channels_only] Processed channel {tvg_id} - processed {processed_channels}/{total_channels}")
+                        logger.debug(
+                            f"[parse_channels_only] Processed channel {tvg_id} - "
+                            f"processed {processed_channels}/{channel_estimate or total_channels}"
+                        )
                     if process:
                         logger.debug(f"[parse_channels_only] Memory before elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
                     # Clear memory
@@ -1499,6 +1550,21 @@ def parse_channels_only(source):
             logger.info(f"[parse_channels_only] Processed {processed_channels} channels current memory: {process.memory_info().rss / 1024 / 1024:.2f} MB")
         else:
             logger.info(f"[parse_channels_only] Processed {processed_channels} channels")
+
+        if processed_channels > 0:
+            progress, channel_estimate = _channel_parse_progress(
+                processed_channels,
+                channel_estimate,
+                had_db_baseline,
+            )
+            send_epg_update(
+                source.id,
+                "parsing_channels",
+                progress,
+                processed=processed_channels,
+                total=channel_estimate,
+            )
+
         # Process any remaining items
         if epgs_to_create:
             EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True)

apps/epg/tests/test_parsing_channels_progress.py

@@ -0,0 +1,44 @@
+from django.test import SimpleTestCase
+
+from apps.epg.tasks import (
+    _CHANNEL_PARSE_PROGRESS_CAP,
+    _CHANNEL_PARSE_PROGRESS_START,
+    _channel_parse_progress,
+)
+
+
+class ChannelParseProgressTests(SimpleTestCase):
+    def test_exceeding_estimate_recalculates_instead_of_going_over_cap(self):
+        """101/100 used to yield ~90%+; now bumps estimate to 101 so ratio stays <= 1."""
+        progress, estimate = _channel_parse_progress(100, 100, had_db_baseline=True)
+        self.assertEqual(estimate, 100)
+        self.assertLessEqual(progress, _CHANNEL_PARSE_PROGRESS_CAP)
+
+        progress, estimate = _channel_parse_progress(101, 100, had_db_baseline=True)
+        self.assertEqual(estimate, 101)
+        self.assertLessEqual(progress, _CHANNEL_PARSE_PROGRESS_CAP)
+        self.assertGreater(progress, 90)
+
+    def test_large_xml_growth_never_exceeds_cap(self):
+        """Previously 425 channels vs DB estimate of 100 could show ~300%."""
+        estimate = 100
+        for processed in (100, 200, 300, 425):
+            progress, estimate = _channel_parse_progress(
+                processed, estimate, had_db_baseline=True
+            )
+            self.assertLessEqual(progress, _CHANNEL_PARSE_PROGRESS_CAP)
+
+    def test_first_import_crawls_instead_of_flat_ninety(self):
+        progress, estimate = _channel_parse_progress(100, 0, had_db_baseline=False)
+        self.assertEqual(estimate, 0)
+        self.assertEqual(progress, _CHANNEL_PARSE_PROGRESS_START + 1)
+        self.assertLess(progress, 90)
+
+        progress, _ = _channel_parse_progress(5000, 0, had_db_baseline=False)
+        self.assertLess(progress, _CHANNEL_PARSE_PROGRESS_CAP)
+
+    def test_partial_progress_when_below_estimate(self):
+        progress, estimate = _channel_parse_progress(50, 100, had_db_baseline=True)
+        self.assertEqual(estimate, 100)
+        self.assertGreater(progress, _CHANNEL_PARSE_PROGRESS_START)
+        self.assertLess(progress, _CHANNEL_PARSE_PROGRESS_CAP)