#!/usr/bin/env python3 """ File System Caching for NFOGuard Provides LRU caching for expensive file system operations """ import os import time from functools import lru_cache from pathlib import Path from typing import Dict, List, Optional, Tuple, Any from datetime import datetime, timezone from core.logging import _log class FileSystemCache: """Smart caching for file system operations with mtime invalidation""" def __init__(self, max_cache_size: int = 1000): self.max_cache_size = max_cache_size self._dir_cache: Dict[str, Tuple[float, List[Path]]] = {} self._file_cache: Dict[str, Tuple[float, Any]] = {} self._mtime_cache: Dict[str, float] = {} self._cache_hits = 0 self._cache_misses = 0 def get_directory_contents(self, directory_path: Path, pattern: str = "*") -> List[Path]: """Get directory contents with caching and mtime invalidation""" cache_key = f"{directory_path}:{pattern}" if not directory_path.exists(): return [] # Check if directory mtime has changed current_mtime = directory_path.stat().st_mtime if cache_key in self._dir_cache: cached_mtime, cached_contents = self._dir_cache[cache_key] if cached_mtime == current_mtime: self._cache_hits += 1 return cached_contents # Cache miss - scan directory self._cache_misses += 1 contents = [] try: if pattern == "*": contents = list(directory_path.iterdir()) else: contents = list(directory_path.glob(pattern)) except (OSError, PermissionError) as e: _log("DEBUG", f"Error scanning directory {directory_path}: {e}") return [] # Cache the results self._dir_cache[cache_key] = (current_mtime, contents) self._cleanup_cache() return contents def find_video_files(self, directory_path: Path) -> List[Path]: """Find video files with caching""" video_extensions = {".mkv", ".mp4", ".avi", ".mov", ".m4v"} cache_key = f"videos:{directory_path}" if not directory_path.exists(): return [] current_mtime = directory_path.stat().st_mtime # Check cache if cache_key in self._file_cache: cached_mtime, cached_files = self._file_cache[cache_key] if cached_mtime == current_mtime: self._cache_hits += 1 return cached_files # Find video files self._cache_misses += 1 video_files = [] try: for file_path in directory_path.iterdir(): if file_path.is_file() and file_path.suffix.lower() in video_extensions: video_files.append(file_path) except (OSError, PermissionError) as e: _log("DEBUG", f"Error scanning for videos in {directory_path}: {e}") return [] # Cache results self._file_cache[cache_key] = (current_mtime, video_files) self._cleanup_cache() return video_files def get_file_mtime(self, file_path: Path) -> Optional[float]: """Get file modification time with caching""" cache_key = str(file_path) # Always check actual mtime for files (they change frequently) try: current_mtime = file_path.stat().st_mtime self._mtime_cache[cache_key] = current_mtime return current_mtime except (OSError, FileNotFoundError): self._mtime_cache.pop(cache_key, None) return None def file_exists(self, file_path: Path) -> bool: """Check if file exists with smart caching""" cache_key = f"exists:{file_path}" # For existence checks, we can cache briefly but need to be careful if cache_key in self._file_cache: cache_time, exists = self._file_cache[cache_key] if time.time() - cache_time < 30: # Cache for 30 seconds self._cache_hits += 1 return exists # Check actual existence self._cache_misses += 1 exists = file_path.exists() self._file_cache[cache_key] = (time.time(), exists) return exists def find_episode_files(self, season_dir: Path, season_num: int, episode_num: int) -> List[Path]: """Find episode files with pattern caching""" cache_key = f"episode:{season_dir}:S{season_num:02d}E{episode_num:02d}" if not season_dir.exists(): return [] current_mtime = season_dir.stat().st_mtime # Check cache if cache_key in self._file_cache: cached_mtime, cached_files = self._file_cache[cache_key] if cached_mtime == current_mtime: self._cache_hits += 1 return cached_files # Find episode files self._cache_misses += 1 episode_pattern = f"S{season_num:02d}E{episode_num:02d}" episode_files = [] video_extensions = {".mkv", ".mp4", ".avi", ".mov", ".m4v"} try: for file_path in season_dir.iterdir(): if (file_path.is_file() and file_path.suffix.lower() in video_extensions and episode_pattern.upper() in file_path.name.upper()): episode_files.append(file_path) except (OSError, PermissionError) as e: _log("DEBUG", f"Error finding episode files in {season_dir}: {e}") return [] # Cache results self._file_cache[cache_key] = (current_mtime, episode_files) self._cleanup_cache() return episode_files def _cleanup_cache(self): """Clean up cache when it gets too large""" if len(self._dir_cache) > self.max_cache_size: # Remove oldest 25% of entries remove_count = self.max_cache_size // 4 old_keys = list(self._dir_cache.keys())[:remove_count] for key in old_keys: del self._dir_cache[key] if len(self._file_cache) > self.max_cache_size: remove_count = self.max_cache_size // 4 old_keys = list(self._file_cache.keys())[:remove_count] for key in old_keys: del self._file_cache[key] def clear_cache(self): """Clear all caches""" self._dir_cache.clear() self._file_cache.clear() self._mtime_cache.clear() _log("INFO", "File system cache cleared") def get_cache_stats(self) -> Dict[str, Any]: """Get cache performance statistics""" total_requests = self._cache_hits + self._cache_misses hit_rate = (self._cache_hits / total_requests * 100) if total_requests > 0 else 0 return { "cache_hits": self._cache_hits, "cache_misses": self._cache_misses, "hit_rate_percent": round(hit_rate, 2), "dir_cache_size": len(self._dir_cache), "file_cache_size": len(self._file_cache), "mtime_cache_size": len(self._mtime_cache) } # Global cache instance fs_cache = FileSystemCache() # Cached utility functions @lru_cache(maxsize=500) def parse_episode_from_filename(filename: str) -> Optional[Tuple[int, int]]: """Parse season/episode from filename with LRU caching""" import re match = re.search(r"S(\d{1,2})E(\d{1,2})", filename, re.IGNORECASE) if match: return int(match.group(1)), int(match.group(2)) return None @lru_cache(maxsize=200) def extract_imdb_from_path(path_str: str) -> Optional[str]: """Extract IMDb ID from path with LRU caching""" import re match = re.search(r'\[imdb-([^]]+)\]', path_str, re.IGNORECASE) return match.group(1) if match else None def clear_all_caches(): """Clear all caches including LRU caches""" fs_cache.clear_cache() parse_episode_from_filename.cache_clear() extract_imdb_from_path.cache_clear() # Also clear XML caches try: from core.xml_cache import clear_xml_caches clear_xml_caches() except ImportError: pass _log("INFO", "All caches cleared")