Phase 3: incremental duplicate scans, debounced scheduling, and metadata-safe title normalization

Author: crocodilestick

cps/cwa_functions.py

@@ -19,7 +19,7 @@
 from time import sleep
 
 import json
-from threading import Thread
+from threading import Thread, Lock, Timer
 import queue
 import os
 import tempfile
@@ -59,6 +59,10 @@
 LOG_ARCHIVE = "/config/log_archive"
 DIRS_JSON = "/app/calibre-web-automated/dirs.json"
 
+# Debounced duplicate scan timer (web process)
+_duplicate_scan_timer = None
+_duplicate_scan_lock = Lock()
+
 ##———————————————————————————END OF GLOBAL VARIABLES——————————————————————————##
 
 ##————————————————————————————————————————————————————————————————————————————##
@@ -346,6 +350,57 @@ def _enqueue_autosend():
         log.error(f"Internal auto-send schedule failed: {e}")
         return jsonify({"error": str(e)}), 400
 
+
+@csrf.exempt
+@cwa_internal.route('/cwa-internal/queue-duplicate-scan', methods=["POST"])
+def cwa_internal_queue_duplicate_scan():
+    """Debounce and queue an incremental duplicate scan in the web process.
+
+    Security: Limited to localhost callers (within container/host).
+    Payload JSON: {delay_seconds:int}
+    """
+    try:
+        remote = request.headers.get('X-Forwarded-For', request.remote_addr)
+        if remote not in (None, '127.0.0.1', '::1'):
+            abort(403)
+
+        db = CWA_DB()
+        enabled = bool(db.cwa_settings.get('duplicate_scan_enabled', 0))
+        frequency = db.cwa_settings.get('duplicate_scan_frequency', 'manual')
+
+        data = request.get_json(force=True, silent=True) or {}
+        default_delay = db.cwa_settings.get('duplicate_scan_debounce_seconds', 30)
+        delay_seconds = int(data.get('delay_seconds', default_delay))
+        delay_seconds = max(5, min(600, delay_seconds))
+
+        if not enabled or frequency != 'after_import':
+            return jsonify({"success": True, "skipped": True, "reason": "disabled_or_manual"}), 200
+
+        global _duplicate_scan_timer
+        with _duplicate_scan_lock:
+            if _duplicate_scan_timer is not None:
+                try:
+                    _duplicate_scan_timer.cancel()
+                except Exception:
+                    pass
+
+            def _enqueue_scan():
+                try:
+                    from .tasks.duplicate_scan import TaskDuplicateScan
+                    WorkerThread.add('System', TaskDuplicateScan(full_scan=False, trigger_type='after_import'), hidden=False)
+                    log.info("[cwa-duplicates] Debounced duplicate scan queued (after_import)")
+                except Exception as e:
+                    log.error("[cwa-duplicates] Failed to queue debounced duplicate scan: %s", str(e))
+
+            _duplicate_scan_timer = Timer(delay_seconds, _enqueue_scan)
+            _duplicate_scan_timer.daemon = True
+            _duplicate_scan_timer.start()
+
+        return jsonify({"success": True, "queued": True, "delay_seconds": delay_seconds}), 200
+    except Exception as e:
+        log.error("[cwa-duplicates] Failed to schedule debounced duplicate scan: %s", str(e))
+        return jsonify({"success": False, "error": str(e)}), 500
+
 @csrf.exempt
 @cwa_internal.route('/cwa-internal/schedule-convert-library', methods=["POST"])
 def cwa_internal_schedule_convert_library():
@@ -554,7 +609,7 @@ def set_cwa_settings():
     boolean_settings = []
     string_settings = []
     list_settings = []
-    integer_settings = ['ingest_timeout_minutes', 'auto_send_delay_minutes', 'duplicate_scan_hour', 'duplicate_scan_chunk_size']  # Special handling for integer settings
+    integer_settings = ['ingest_timeout_minutes', 'auto_send_delay_minutes', 'duplicate_scan_hour', 'duplicate_scan_chunk_size', 'duplicate_scan_debounce_seconds']  # Special handling for integer settings
     json_settings = ['metadata_provider_hierarchy', 'metadata_providers_enabled', 'duplicate_format_priority']  # Special handling for JSON settings
 
     for setting in cwa_default_settings:
@@ -639,18 +694,24 @@ def set_cwa_settings():
                             int_value = max(0, min(23, int_value))
                         elif setting == 'duplicate_scan_chunk_size':
                             int_value = max(500, min(50000, int_value))
+                        elif setting == 'duplicate_scan_debounce_seconds':
+                            int_value = max(5, min(600, int_value))
                         result[setting] = int_value
                     except (ValueError, TypeError):
                         # Use current value if conversion fails
                         if setting == 'ingest_timeout_minutes':
                             result[setting] = cwa_db.cwa_settings.get(setting, 15)  # Default to 15 minutes
                         elif setting == 'auto_send_delay_minutes':
                             result[setting] = cwa_db.cwa_settings.get(setting, 5)  # Default to 5 minutes
+                        elif setting == 'duplicate_scan_debounce_seconds':
+                            result[setting] = cwa_db.cwa_settings.get(setting, 30)
                 else:
                     if setting == 'ingest_timeout_minutes':
                         result[setting] = cwa_db.cwa_settings.get(setting, 15)  # Default to 15 minutes
                     elif setting == 'auto_send_delay_minutes':
                         result[setting] = cwa_db.cwa_settings.get(setting, 5)  # Default to 5 minutes
+                    elif setting == 'duplicate_scan_debounce_seconds':
+                        result[setting] = cwa_db.cwa_settings.get(setting, 30)
 
             # Handle JSON settings
             for setting in json_settings:

cps/duplicates.py

@@ -62,6 +62,21 @@ def generate_group_hash(title, author):
     return hashlib.md5(composite.encode('utf-8')).hexdigest()
 
 
+def normalize_title_for_duplicates(title, primary_author=None):
+    """Normalize title for duplicate detection.
+
+    If the title starts with the primary author (e.g., "Homer, the Iliad"),
+    strip the leading author prefix to avoid false negatives.
+    """
+    normalized = (title or "untitled").lower().strip()
+    if primary_author:
+        author_norm = str(primary_author).lower().strip()
+        author_prefix = f"{author_norm}, "
+        if normalized.startswith(author_prefix):
+            normalized = normalized[len(author_prefix):].strip()
+    return normalized
+
+
 def validate_resolution_strategy(strategy):
     """Validate that strategy is one of the allowed values"""
     valid_strategies = ['newest', 'highest_quality_format', 'most_metadata', 'largest_file_size']
@@ -406,7 +421,7 @@ def find_duplicate_books(include_dismissed=False, user_id=None):
     return duplicate_groups
 
 
-def find_duplicate_candidate_ids_sql(use_title, use_author, user_id=None):
+def find_duplicate_candidate_ids_sql(use_title, use_author, user_id=None, min_book_id=None):
     """SQL-based candidate prefilter for hybrid mode.
 
     Returns a set of book IDs that are likely part of duplicate groups.
@@ -425,11 +440,12 @@ def find_duplicate_candidate_ids_sql(use_title, use_author, user_id=None):
 
     print("[cwa-duplicates] Using SQL hybrid prefilter (candidate IDs)", flush=True)
 
+    # Note: these GROUP BY fields are evaluated by SQLite at query time; they are not cached
+    # groupings in memory. We only use this query to prefilter candidate IDs.
     group_by_fields = []
 
-    if use_title:
-        norm_title = func.lower(func.trim(func.coalesce(db.Books.title, 'untitled')))
-        group_by_fields.append(norm_title)
+    norm_title = None
+    primary_author = None
 
     if use_author:
         norm_author_sort = func.lower(func.trim(func.coalesce(db.Books.author_sort, 'unknown')))
@@ -440,15 +456,32 @@ def find_duplicate_candidate_ids_sql(use_title, use_author, user_id=None):
         )
         group_by_fields.append(primary_author)
 
+    if use_title:
+        norm_title = func.lower(func.trim(func.coalesce(db.Books.title, 'untitled')))
+        if primary_author is not None:
+            author_prefix = primary_author + ', '
+            norm_title = case(
+                (norm_title.like(author_prefix + '%'),
+                 func.trim(func.substr(norm_title, func.length(primary_author) + 3))),
+                else_=norm_title
+            )
+        group_by_fields.append(norm_title)
+
+    max_id_field = func.max(db.Books.id).label('max_book_id')
+
     query = (calibre_db.session.query(
                 func.count(func.distinct(db.Books.id)).label('book_count'),
-                func.group_concat(func.distinct(db.Books.id)).label('book_ids_str')
+                func.group_concat(func.distinct(db.Books.id)).label('book_ids_str'),
+                max_id_field
             )
             .select_from(db.Books)
             .filter(get_common_filters(user_id=user_id))
             .group_by(*group_by_fields)
             .having(func.count(func.distinct(db.Books.id)) > 1))
 
+    if min_book_id is not None:
+        query = query.having(max_id_field > int(min_book_id))
+
     try:
         results = query.all()
     except Exception as e:
@@ -695,21 +728,24 @@ def find_duplicate_books_python(use_title, use_author, use_language, use_series,
     for book in all_books:
         # Build key based on selected criteria
         key_parts = []
-        
-        if use_title:
-            # Handle potential None title
-            title = book.title if book.title else "untitled"
-            key_parts.append(title.lower().strip())
-        
+
+        primary_author = None
         if use_author:
             # Ensure authors are loaded and not empty
             if book.authors and len(book.authors) > 0:
                 # Get primary author (use Calibre-Web's standard approach)
                 book.ordered_authors = calibre_db.order_authors([book])
                 primary_author = book.ordered_authors[0].name if book.ordered_authors and len(book.ordered_authors) > 0 else "unknown"
-                key_parts.append(primary_author.lower().strip())
             else:
-                key_parts.append("unknown")
+                primary_author = "unknown"
+        
+        if use_title:
+            # Handle potential None title
+            title = book.title if book.title else "untitled"
+            key_parts.append(normalize_title_for_duplicates(title, primary_author))
+
+        if use_author:
+            key_parts.append(primary_author.lower().strip() if primary_author else "unknown")
 
         if use_language:
             # Get primary language code

cps/tasks/duplicate_scan.py

@@ -6,10 +6,12 @@
 # See CONTRIBUTORS for full list of authors.
 
 import sys
+from datetime import datetime
 from sqlalchemy import func
 from flask_babel import lazy_gettext as N_
 
 from cps import calibre_db, db, logger
+from cps.duplicates import find_duplicate_books_python, find_duplicate_candidate_ids_sql
 from cps.services.worker import CalibreTask, STAT_CANCELLED, STAT_ENDED
 from cps.ub import init_db_thread
 
@@ -54,9 +56,11 @@ def run(self, worker_thread):
             cwa_db = CWA_DB()
             cache_data = cwa_db.get_duplicate_cache() or {}
 
-            if not self.full_scan and not cache_data.get('scan_pending', True):
-                self._handleSuccess()
-                return
+            if not self.full_scan:
+                last_scanned_book_id = int(cache_data.get('last_scanned_book_id') or 0)
+                if last_scanned_book_id == 0:
+                    # No baseline yet; fall back to full scan
+                    self.full_scan = True
 
             self.progress = 0.1
             self.message = N_('Scanning for duplicates')
@@ -67,26 +71,120 @@ def run(self, worker_thread):
             if self.stat in (STAT_CANCELLED, STAT_ENDED):
                 return
 
-            duplicate_groups = find_duplicate_books(include_dismissed=False, user_id=self.user_id)
-            self.result_count = len(duplicate_groups)
-
-            if self.stat in (STAT_CANCELLED, STAT_ENDED):
-                return
-
-            # Update cache with full results (including dismissed groups)
-            all_groups = find_duplicate_books(include_dismissed=True, user_id=self.user_id)
-
-            max_book_id = 0
-            try:
-                max_id_result = calibre_db.session.query(func.max(db.Books.id)).scalar()
-                max_book_id = max_id_result if max_id_result is not None else 0
-            except Exception as ex:
-                log.warning("[cwa-duplicates] Could not get max book ID in TaskDuplicateScan: %s", str(ex))
-
-            cwa_db.update_duplicate_cache(all_groups, len(all_groups), max_book_id)
+            if self.full_scan:
+                duplicate_groups = find_duplicate_books(include_dismissed=False, user_id=self.user_id)
+                self.result_count = len(duplicate_groups)
+
+                if self.stat in (STAT_CANCELLED, STAT_ENDED):
+                    return
+
+                # Update cache with full results (including dismissed groups)
+                all_groups = find_duplicate_books(include_dismissed=True, user_id=self.user_id)
+
+                max_book_id = 0
+                try:
+                    max_id_result = calibre_db.session.query(func.max(db.Books.id)).scalar()
+                    max_book_id = max_id_result if max_id_result is not None else 0
+                except Exception as ex:
+                    log.warning("[cwa-duplicates] Could not get max book ID in TaskDuplicateScan: %s", str(ex))
+
+                cwa_db.update_duplicate_cache(all_groups, len(all_groups), max_book_id)
+                log.info("[cwa-duplicates] Duplicate cache updated (full scan): groups=%s max_book_id=%s",
+                         len(all_groups), max_book_id)
+            else:
+                # Incremental scan: only groups impacted by newly added books
+                settings = cwa_db.cwa_settings
+                use_title = settings.get('duplicate_detection_title', 1)
+                use_author = settings.get('duplicate_detection_author', 1)
+                use_language = settings.get('duplicate_detection_language', 1)
+                use_series = settings.get('duplicate_detection_series', 0)
+                use_publisher = settings.get('duplicate_detection_publisher', 0)
+                use_format = settings.get('duplicate_detection_format', 0)
+
+                last_scanned_book_id = int(cache_data.get('last_scanned_book_id') or 0)
+                candidate_ids = find_duplicate_candidate_ids_sql(use_title, use_author, user_id=self.user_id,
+                                                                min_book_id=last_scanned_book_id)
+
+                max_book_id = 0
+                try:
+                    max_id_result = calibre_db.session.query(func.max(db.Books.id)).scalar()
+                    max_book_id = max_id_result if max_id_result is not None else 0
+                except Exception as ex:
+                    log.warning("[cwa-duplicates] Could not get max book ID in TaskDuplicateScan: %s", str(ex))
+
+                if not candidate_ids:
+                    # No impacted groups; just bump last_scanned_book_id
+                    try:
+                        cwa_db.cur.execute("""
+                            UPDATE cwa_duplicate_cache
+                            SET last_scanned_book_id = ?, scan_pending = 0, scan_timestamp = ?
+                            WHERE id = 1
+                        """, (max_book_id, datetime.now().isoformat()))
+                        cwa_db.con.commit()
+                        log.info("[cwa-duplicates] Incremental scan: no candidates; cache timestamp updated (last_scanned_book_id=%s)",
+                                 max_book_id)
+                    except Exception:
+                        pass
+                    self.result_count = 0
+                else:
+                    # Identify affected groups in cache
+                    cached_groups = cache_data.get('duplicate_groups', [])
+                    candidate_set = set(candidate_ids)
+                    affected_hashes = {
+                        group.get('group_hash')
+                        for group in cached_groups
+                        if group.get('book_ids') and candidate_set.intersection(set(group.get('book_ids')))
+                    }
+
+                    # Recompute groups for candidate IDs (include dismissed for cache)
+                    recomputed_groups = find_duplicate_books_python(
+                        use_title, use_author, use_language, use_series, use_publisher, use_format,
+                        include_dismissed=True, user_id=self.user_id, candidate_ids=candidate_ids
+                    )
+
+                    # Serialize recomputed groups
+                    serialized_groups = []
+                    for group in recomputed_groups:
+                        serialized_groups.append({
+                            'title': group.get('title', ''),
+                            'author': group.get('author', ''),
+                            'count': group.get('count', 0),
+                            'group_hash': group.get('group_hash', ''),
+                            'book_ids': [book.id for book in group.get('books', [])]
+                        })
+
+                    # Merge cache: remove affected, then append recomputed
+                    kept_groups = [g for g in cached_groups if g.get('group_hash') not in affected_hashes]
+                    merged_groups = kept_groups + serialized_groups
+
+                    try:
+                        import json
+                        cwa_db.cur.execute("""
+                            UPDATE cwa_duplicate_cache
+                            SET scan_timestamp = ?,
+                                duplicate_groups_json = ?,
+                                total_count = ?,
+                                scan_pending = 0,
+                                last_scanned_book_id = ?
+                            WHERE id = 1
+                        """, (datetime.now().isoformat(), json.dumps(merged_groups), len(merged_groups), max_book_id))
+                        cwa_db.con.commit()
+                        log.info("[cwa-duplicates] Duplicate cache updated (incremental): merged_groups=%s max_book_id=%s",
+                                 len(merged_groups), max_book_id)
+                    except Exception as ex:
+                        log.warning("[cwa-duplicates] Failed to update incremental cache: %s", str(ex))
+
+                    # Result count is unresolved duplicates for this run (exclude dismissed)
+                    self.result_count = len(find_duplicate_books_python(
+                        use_title, use_author, use_language, use_series, use_publisher, use_format,
+                        include_dismissed=False, user_id=self.user_id, candidate_ids=candidate_ids
+                    ))
 
             self.progress = 1
-            self.message = N_('Duplicate scan completed: %(count)s groups', count=len(duplicate_groups))
+            if self.full_scan:
+                self.message = N_('Duplicate scan completed: %(count)s groups', count=self.result_count)
+            else:
+                self.message = N_('Duplicate scan completed: %(count)s new groups', count=self.result_count)
             self._handleSuccess()
         except Exception as ex:
             log.error("[cwa-duplicates] Duplicate scan task failed: %s", str(ex))