feat(m3u): Add stream count summary and parsing result handling to M3U refresh process. Introduce helper functions for better readability and maintainability. Update frontend notification to display detailed stream processing outcomes, including counts for created, updated, stale, and removed streams.

CHANGELOG.md

@@ -29,6 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **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.
 
 ## [0.27.1] - 2026-06-25

apps/m3u/tasks.py

@@ -1034,6 +1034,26 @@ def _bulk_update_stream_refresh_batches(changed_streams, touch_streams, *, batch
         )
 
 
+def _batch_stream_count_message(created, updated, unchanged):
+    """Human-readable batch summary; unchanged = last_seen touch only."""
+    return (
+        f"{created} created, {updated} updated, {unchanged} unchanged."
+    )
+
+
+def _parse_batch_stream_counts(result):
+    """Extract (created, updated, unchanged) from a batch processing result string."""
+    if not isinstance(result, str):
+        return 0, 0, 0
+    try:
+        created = int(re.search(r"(\d+) created", result).group(1))
+        updated = int(re.search(r"(\d+) updated", result).group(1))
+        unchanged = int(re.search(r"(\d+) unchanged", result).group(1))
+        return created, updated, unchanged
+    except (AttributeError, ValueError):
+        return 0, 0, 0
+
+
 def process_xc_category_direct(account_id, batch, groups, hash_keys):
     from django.db import connections
 
@@ -1213,8 +1233,12 @@ def process_xc_category_direct(account_id, batch, groups, hash_keys):
             logger.error(f"Bulk operation failed for XC streams: {str(e)}")
 
         retval = (
-            f"Batch processed: {len(streams_to_create)} created, "
-            f"{len(streams_to_update) + len(streams_to_touch)} updated."
+            "Batch processed: "
+            + _batch_stream_count_message(
+                len(streams_to_create),
+                len(streams_to_update),
+                len(streams_to_touch),
+            )
         )
 
     except Exception as e:
@@ -1425,8 +1449,12 @@ def process_m3u_batch_direct(account_id, batch, groups, hash_keys, compiled_filt
         logger.error(f"Bulk operation failed: {str(e)}")
 
     retval = (
-        f"M3U account: {account_id}, Batch processed: {len(streams_to_create)} created, "
-        f"{len(streams_to_update) + len(streams_to_touch)} updated."
+        f"M3U account: {account_id}, Batch processed: "
+        + _batch_stream_count_message(
+            len(streams_to_create),
+            len(streams_to_update),
+            len(streams_to_touch),
+        )
     )
 
     # Clean up database connections for threading
@@ -3359,6 +3387,8 @@ def _refresh_single_m3u_account_impl(account_id):
     start_time = time.time()  # For tracking elapsed time as float
     streams_created = 0
     streams_updated = 0
+    streams_unchanged = 0
+    streams_stale = 0
     streams_deleted = 0
 
     try:
@@ -3540,6 +3570,7 @@ def _refresh_single_m3u_account_impl(account_id):
         # Initialize stream counters
         streams_created = 0
         streams_updated = 0
+        streams_unchanged = 0
 
         if account.account_type == M3UAccount.Types.STADNARD:
             logger.debug(
@@ -3582,17 +3613,13 @@ def _refresh_single_m3u_account_impl(account_id):
                         completed_batches += 1
 
                         # Extract stream counts from result
-                        if isinstance(result, str):
-                            try:
-                                created_match = re.search(r"(\d+) created", result)
-                                updated_match = re.search(r"(\d+) updated", result)
-                                if created_match and updated_match:
-                                    created_count = int(created_match.group(1))
-                                    updated_count = int(updated_match.group(1))
-                                    streams_created += created_count
-                                    streams_updated += updated_count
-                            except (AttributeError, ValueError):
-                                pass
+                        created_count, updated_count, unchanged_count = (
+                            _parse_batch_stream_counts(result)
+                        )
+                        if created_count or updated_count or unchanged_count:
+                            streams_created += created_count
+                            streams_updated += updated_count
+                            streams_unchanged += unchanged_count
 
                         # Send progress update
                         progress = int((completed_batches / total_batches) * 100)
@@ -3610,7 +3637,7 @@ def _refresh_single_m3u_account_impl(account_id):
                             progress,
                             elapsed_time=current_elapsed,
                             time_remaining=time_remaining,
-                            streams_processed=streams_created + streams_updated,
+                            streams_processed=streams_created + streams_updated + streams_unchanged,
                         )
 
                         logger.debug(f"Thread batch {completed_batches}/{total_batches} completed")
@@ -3708,17 +3735,13 @@ def _refresh_single_m3u_account_impl(account_id):
                             completed_batches += 1
 
                             # Extract stream counts from result
-                            if isinstance(result, str):
-                                try:
-                                    created_match = re.search(r"(\d+) created", result)
-                                    updated_match = re.search(r"(\d+) updated", result)
-                                    if created_match and updated_match:
-                                        created_count = int(created_match.group(1))
-                                        updated_count = int(updated_match.group(1))
-                                        streams_created += created_count
-                                        streams_updated += updated_count
-                                except (AttributeError, ValueError):
-                                    pass
+                            created_count, updated_count, unchanged_count = (
+                                _parse_batch_stream_counts(result)
+                            )
+                            if created_count or updated_count or unchanged_count:
+                                streams_created += created_count
+                                streams_updated += updated_count
+                                streams_unchanged += unchanged_count
 
                             # Send progress update
                             progress = int((completed_batches / total_batches) * 100)
@@ -3736,7 +3759,7 @@ def _refresh_single_m3u_account_impl(account_id):
                                 progress,
                                 elapsed_time=current_elapsed,
                                 time_remaining=time_remaining,
-                                streams_processed=streams_created + streams_updated,
+                                streams_processed=streams_created + streams_updated + streams_unchanged,
                             )
 
                             logger.debug(f"XC thread batch {completed_batches}/{total_batches} completed")
@@ -3762,11 +3785,11 @@ def _refresh_single_m3u_account_impl(account_id):
         ).exists()  # This will never find anything but ensures DB sync
 
         # Mark streams that weren't seen in this refresh as stale (pending deletion)
-        stale_stream_count = Stream.objects.filter(
+        streams_stale = Stream.objects.filter(
             m3u_account=account,
             last_seen__lt=refresh_start_timestamp
         ).update(is_stale=True)
-        logger.info(f"Marked {stale_stream_count} streams as stale for account {account_id}")
+        logger.info(f"Marked {streams_stale} streams as stale for account {account_id}")
 
         # Mark group relationships that weren't seen in this refresh as stale (pending deletion)
         stale_group_count = ChannelGroupM3UAccount.objects.filter(
@@ -3826,13 +3849,14 @@ def _refresh_single_m3u_account_impl(account_id):
         elapsed_time = time.time() - start_time
 
         # Calculate total streams processed
-        streams_processed = streams_created + streams_updated
+        streams_processed = streams_created + streams_updated + streams_unchanged
 
         # Set status to success and update timestamp BEFORE sending the final update
         account.status = M3UAccount.Status.SUCCESS
         account.last_message = (
             f"Processing completed in {elapsed_time:.1f} seconds. "
-            f"Streams: {streams_created} created, {streams_updated} updated, {streams_deleted} removed. "
+            f"Streams: {streams_created} created, {streams_updated} updated, "
+            f"{streams_stale} marked stale, {streams_deleted} removed. "
             f"Total processed: {streams_processed}.{auto_sync_message}"
         )
         account.updated_at = timezone.now()
@@ -3845,6 +3869,7 @@ def _refresh_single_m3u_account_impl(account_id):
             elapsed_time=round(elapsed_time, 2),
             streams_created=streams_created,
             streams_updated=streams_updated,
+            streams_stale=streams_stale,
             streams_deleted=streams_deleted,
             total_processed=streams_processed,
         )
@@ -3860,6 +3885,7 @@ def _refresh_single_m3u_account_impl(account_id):
             streams_processed=streams_processed,
             streams_created=streams_created,
             streams_updated=streams_updated,
+            streams_stale=streams_stale,
             streams_deleted=streams_deleted,
             # Structured auto-sync counts so the frontend can render a
             # warning card when anything failed, without parsing the

frontend/src/components/M3URefreshNotification.jsx

@@ -52,6 +52,22 @@ const M3uSetupSuccess = ({ data }) => {
 };
 
 // One-line outcome summary for the notification body.
+const buildStreamSummary = (data) => {
+  if (data.streams_processed == null && data.streams_created == null) {
+    return null;
+  }
+  const created = data.streams_created || 0;
+  const updated = data.streams_updated || 0;
+  const stale = data.streams_stale || 0;
+  const removed = data.streams_deleted || 0;
+  const processed = data.streams_processed || 0;
+  return (
+    `Streams: ${created} created, ${updated} updated, ` +
+    `${stale} marked stale, ${removed} removed. ` +
+    `Total processed: ${processed}.`
+  );
+};
+
 const buildAutoSyncSummary = (data) => {
   const created = data.channels_created || 0;
   const updated = data.channels_updated || 0;
@@ -211,21 +227,29 @@ export default function M3URefreshNotification() {
 
     let body = message;
     let autoClose = 2000;
-    // Surface auto-sync counts attached to the parsing-complete event
-    // so the channel-side outcome appears in the notification body.
+    // Surface stream and auto-sync counts attached to the parsing-complete
+    // event so the outcome appears in the notification body.
     if (data.progress == 100 && data.action === 'parsing') {
+      const streamSummary = buildStreamSummary(data);
       const autoSyncSummary = buildAutoSyncSummary(data);
       const failed = data.channels_failed || 0;
       const failedDetails = Array.isArray(data.failed_stream_details)
         ? data.failed_stream_details
         : [];
-      if (autoSyncSummary) {
+      if (streamSummary || autoSyncSummary) {
         body = (
           <Stack gap={4}>
             <Text size="sm">{message}</Text>
-            <Text size="xs" c="dimmed">
-              {autoSyncSummary}
-            </Text>
+            {streamSummary && (
+              <Text size="xs" c="dimmed">
+                {streamSummary}
+              </Text>
+            )}
+            {autoSyncSummary && (
+              <Text size="xs" c="dimmed">
+                {autoSyncSummary}
+              </Text>
+            )}
             {failed > 0 && failedDetails.length > 0 && (
               <Button
                 size="xs"

frontend/src/components/__tests__/M3URefreshNotification.test.jsx

@@ -851,4 +851,37 @@ describe('M3URefreshNotification', () => {
       expect(call[0].autoClose).toBe(12000);
     });
   });
+
+  describe('Stream count rendering on parsing complete', () => {
+    it('inlines stream summary including marked stale count', async () => {
+      mockPlaylistsStore.refreshProgress = {
+        1: {
+          account: 1,
+          action: 'parsing',
+          progress: 100,
+          status: 'success',
+          streams_created: 2,
+          streams_updated: 5,
+          streams_stale: 18,
+          streams_deleted: 3,
+          streams_processed: 1200,
+        },
+      };
+
+      renderWithProviders(<M3URefreshNotification />);
+
+      await waitFor(() => {
+        expect(showNotification).toHaveBeenCalled();
+      });
+      const call = showNotification.mock.calls.find(
+        (c) => typeof c[0]?.message === 'object'
+      );
+      expect(call).toBeDefined();
+      const { container } = render(<>{call[0].message}</>);
+      expect(container.textContent).toContain('Stream parsing complete!');
+      expect(container.textContent).toContain('18 marked stale');
+      expect(container.textContent).toContain('3 removed');
+      expect(container.textContent).toContain('Total processed: 1200');
+    });
+  });
 });