refactor: improve file watcher code quality and event processing

Enhance file watcher service with improved event processing logic and code quality fixes.

Event Processing Improvements:
- Unified path checking logic for all event types
- Move events now exclusively check dest_path (no fallback to src_path)
- Cleaner conditional flow with single path validation call
- Better error handling with consistent exception management
- Reduced code complexity from 35 to 23 lines

Code Quality Improvements:
- Fix all f-string logging warnings (W1203) using lazy % formatting
- Add docstrings to fallback classes for better documentation
- Fix multiple-statements warnings by splitting single-line methods
- Move imports to top-level to follow PEP 8 conventions
- Disable docstring warnings for stub methods with appropriate comment

This improves upon PR #17's move event handling by removing the potential
for inconsistent behavior when dest_path validation fails, while also
significantly improving code quality and Pylint compliance.

Author: johnhuang316

src/code_index_mcp/services/file_watcher_service.py

@@ -5,8 +5,11 @@
 trigger index rebuilds when relevant files are modified, created, or deleted.
 It uses the watchdog library for cross-platform file system event monitoring.
 """
+# pylint: disable=missing-function-docstring  # Fallback stub methods don't need docstrings
 
 import logging
+import os
+import traceback
 from threading import Timer
 from typing import Optional, Callable
 from pathlib import Path
@@ -18,17 +21,27 @@
 except ImportError:
     # Fallback classes for when watchdog is not available
     class Observer:
-        def __init__(self): pass
-        def schedule(self, *args, **kwargs): pass
-        def start(self): pass
-        def stop(self): pass
-        def join(self, *args, **kwargs): pass
-        def is_alive(self): return False
+        """Fallback Observer class when watchdog library is not available."""
+        def __init__(self):
+            pass
+        def schedule(self, *args, **kwargs):
+            pass
+        def start(self):
+            pass
+        def stop(self):
+            pass
+        def join(self, *args, **kwargs):
+            pass
+        def is_alive(self):
+            return False
 
     class FileSystemEventHandler:
-        def __init__(self): pass
+        """Fallback FileSystemEventHandler class when watchdog library is not available."""
+        def __init__(self):
+            pass
 
     class FileSystemEvent:
+        """Fallback FileSystemEvent class when watchdog library is not available."""
         def __init__(self):
             self.is_directory = False
             self.src_path = ""
@@ -111,8 +124,8 @@ def start_monitoring(self, rebuild_callback: Callable) -> bool:
 
             # Log detailed Observer setup
             watch_path = str(self.base_path)
-            self.logger.debug(f"Scheduling Observer for path: {watch_path}")
-            
+            self.logger.debug("Scheduling Observer for path: %s", watch_path)
+
             self.observer.schedule(
                 self.event_handler,
                 watch_path,
@@ -124,11 +137,10 @@ def start_monitoring(self, rebuild_callback: Callable) -> bool:
             self.observer.start()
             self.is_monitoring = True
             self.restart_attempts = 0
-            
+
             # Log Observer thread info
             if hasattr(self.observer, '_thread'):
-                thread_info = f"Observer thread: {self.observer._thread}"
-                self.logger.debug(thread_info)
+                self.logger.debug("Observer thread: %s", self.observer._thread)
 
             # Verify observer is actually running
             if self.observer.is_alive():
@@ -140,18 +152,17 @@ def start_monitoring(self, rebuild_callback: Callable) -> bool:
                         "supported_extensions": len(SUPPORTED_EXTENSIONS)
                     }
                 )
-                
+
                 # Add diagnostic test - create a test event to verify Observer works
-                import os
-                self.logger.debug(f"Observer thread is alive: {self.observer.is_alive()}")
-                self.logger.debug(f"Monitored path exists: {os.path.exists(str(self.base_path))}")
-                self.logger.debug(f"Event handler is set: {self.event_handler is not None}")
-                
+                self.logger.debug("Observer thread is alive: %s", self.observer.is_alive())
+                self.logger.debug("Monitored path exists: %s", os.path.exists(str(self.base_path)))
+                self.logger.debug("Event handler is set: %s", self.event_handler is not None)
+
                 # Log current directory for comparison
                 current_dir = os.getcwd()
-                self.logger.debug(f"Current working directory: {current_dir}")
-                self.logger.debug(f"Are paths same: {os.path.normpath(current_dir) == os.path.normpath(str(self.base_path))}")
-                
+                self.logger.debug("Current working directory: %s", current_dir)
+                self.logger.debug("Are paths same: %s", os.path.normpath(current_dir) == os.path.normpath(str(self.base_path)))
+
                 return True
             else:
                 self.logger.error("File watcher failed to start - Observer not alive")
@@ -165,7 +176,7 @@ def start_monitoring(self, rebuild_callback: Callable) -> bool:
     def stop_monitoring(self) -> None:
         """
         Stop file system monitoring and cleanup all resources.
-        
+
         This method ensures complete cleanup of:
         - Observer thread
         - Event handler
@@ -175,43 +186,43 @@ def stop_monitoring(self) -> None:
         if not self.observer and not self.is_monitoring:
             # Already stopped or never started
             return
-            
+
         self.logger.info("Stopping file watcher monitoring...")
-        
+
         try:
             # Step 1: Stop the observer first
             if self.observer:
                 self.logger.debug("Stopping observer...")
                 self.observer.stop()
-                
+
                 # Step 2: Cancel any active debounce timer
                 if self.event_handler and self.event_handler.debounce_timer:
                     self.logger.debug("Cancelling debounce timer...")
                     self.event_handler.debounce_timer.cancel()
-                
+
                 # Step 3: Wait for observer thread to finish (with timeout)
                 self.logger.debug("Waiting for observer thread to finish...")
                 self.observer.join(timeout=5.0)
-                
+
                 # Step 4: Check if thread actually finished
                 if self.observer.is_alive():
                     self.logger.warning("Observer thread did not stop within timeout")
                 else:
                     self.logger.debug("Observer thread stopped successfully")
-            
+
             # Step 5: Clear all references
             self.observer = None
             self.event_handler = None
             self.rebuild_callback = None
             self.is_monitoring = False
-            
+
             self.logger.info("File watcher stopped and cleaned up successfully")
             print("STOPPED: File watcher stopped")
-            
+
         except Exception as e:
             self.logger.error("Error stopping file watcher: %s", e)
             print(f"WARNING: Error stopping file watcher: {e}")
-            
+
             # Force cleanup even if there were errors
             self.observer = None
             self.event_handler = None
@@ -340,21 +351,17 @@ def on_any_event(self, event: FileSystemEvent) -> None:
             event: The file system event
         """
         # Always log events for debugging
-        event_info = f"Raw event: {event.event_type} - {event.src_path} (is_directory: {event.is_directory})"
-        self.logger.debug(event_info)
-        
+        self.logger.debug("Raw event: %s - %s (is_directory: %s)", event.event_type, event.src_path, event.is_directory)
+
         # Test event processing
         should_process = self.should_process_event(event)
-        process_info = f"Should process: {should_process}"
-        self.logger.debug(process_info)
-        
+        self.logger.debug("Should process: %s", should_process)
+
         if should_process:
-            process_msg = f"Processing file system event: {event.event_type} - {event.src_path}"
-            self.logger.debug(process_msg)
+            self.logger.debug("Processing file system event: %s - %s", event.event_type, event.src_path)
             self.reset_debounce_timer()
         else:
-            filter_msg = f"Event filtered out: {event.event_type} - {event.src_path}"
-            self.logger.debug(filter_msg)
+            self.logger.debug("Event filtered out: %s - %s", event.event_type, event.src_path)
 
     def should_process_event(self, event: FileSystemEvent) -> bool:
         """
@@ -368,7 +375,7 @@ def should_process_event(self, event: FileSystemEvent) -> bool:
         """
         # Skip directory events
         if event.is_directory:
-            self.logger.debug(f"Skipping directory event: {event.src_path}")
+            self.logger.debug("Skipping directory event: %s", event.src_path)
             return False
 
         # Select path to check: dest_path for moves, src_path for others
@@ -377,17 +384,17 @@ def should_process_event(self, event: FileSystemEvent) -> bool:
                 self.logger.debug("Move event missing dest_path")
                 return False
             target_path = event.dest_path
-            self.logger.debug(f"Move event: checking destination path {target_path}")
+            self.logger.debug("Move event: checking destination path %s", target_path)
         else:
             target_path = event.src_path
-            self.logger.debug(f"{event.event_type} event: checking source path {target_path}")
-        
+            self.logger.debug("%s event: checking source path %s", event.event_type, target_path)
+
         # Unified path checking
         try:
             path = Path(target_path)
             return self._should_process_path(path)
         except Exception as e:
-            self.logger.debug(f"Path conversion failed for {target_path}: {e}")
+            self.logger.debug("Path conversion failed for %s: %s", target_path, e)
             return False
 
     def _should_process_path(self, path: Path) -> bool:
@@ -401,27 +408,27 @@ def _should_process_path(self, path: Path) -> bool:
             True if path should trigger rebuild, False otherwise
         """
         # Log detailed filtering steps
-        self.logger.debug(f"Checking path: {path}")
-        
+        self.logger.debug("Checking path: %s", path)
+
         # Skip excluded paths
         is_excluded = self.is_excluded_path(path)
-        self.logger.debug(f"Is excluded path: {is_excluded}")
+        self.logger.debug("Is excluded path: %s", is_excluded)
         if is_excluded:
             return False
 
         # Only process supported file types
         is_supported = self.is_supported_file_type(path)
-        self.logger.debug(f"Is supported file type: {is_supported} (extension: {path.suffix})")
+        self.logger.debug("Is supported file type: %s (extension: %s)", is_supported, path.suffix)
         if not is_supported:
             return False
 
         # Skip temporary files
         is_temp = self.is_temporary_file(path)
-        self.logger.debug(f"Is temporary file: {is_temp}")
+        self.logger.debug("Is temporary file: %s", is_temp)
         if is_temp:
             return False
 
-        self.logger.debug(f"Event will be processed: {path}")
+        self.logger.debug("Event will be processed: %s", path)
         return True
 
     def is_excluded_path(self, path: Path) -> bool:
@@ -494,15 +501,14 @@ def reset_debounce_timer(self) -> None:
             self.debounce_timer.cancel()
             self.logger.debug("Previous debounce timer cancelled")
 
-        timer_msg = f"Starting debounce timer for {self.debounce_seconds} seconds"
-        self.logger.debug(timer_msg)
-        
+        self.logger.debug("Starting debounce timer for %s seconds", self.debounce_seconds)
+
         self.debounce_timer = Timer(
             self.debounce_seconds,
             self.trigger_rebuild
         )
         self.debounce_timer.start()
-        
+
         self.logger.debug("Debounce timer started successfully")
 
     def trigger_rebuild(self) -> None:
@@ -512,19 +518,14 @@ def trigger_rebuild(self) -> None:
 
         if self.rebuild_callback:
             try:
-                callback_msg = "Calling rebuild callback..."
-                self.logger.debug(callback_msg)
-                
+                self.logger.debug("Calling rebuild callback...")
+
                 result = self.rebuild_callback()
-                
-                result_msg = f"Rebuild callback completed with result: {result}"
-                self.logger.debug(result_msg)
+
+                self.logger.debug("Rebuild callback completed with result: %s", result)
             except Exception as e:
-                error_msg = f"Rebuild callback failed: {e}"
-                self.logger.error(error_msg)
-                import traceback
+                self.logger.error("Rebuild callback failed: %s", e)
                 traceback_msg = traceback.format_exc()
                 self.logger.error("Traceback: %s", traceback_msg)
         else:
-            no_callback_msg = "No rebuild callback configured"
-            self.logger.warning(no_callback_msg)
+            self.logger.warning("No rebuild callback configured")