import os import time import fnmatch from pathlib import Path from typing import List, Tuple, Optional def is_file_locked(filepath: Path) -> bool: """ Checks if a file is locked exclusively by another process (e.g. SQL Server writing .bak). Attempts to open the file with read-shared permissions. """ if not filepath.exists() or not filepath.is_file(): return True try: # On Windows, try opening in append/read mode to detect exclusive write lock with open(filepath, "rb") as f: f.seek(0, os.SEEK_END) return False except (PermissionError, IOError, OSError): return True def is_file_stable(filepath: Path, min_stable_seconds: int = 60, sample_interval_seconds: float = 0.5) -> bool: """ Ensures that a file is not actively growing or being modified. Verifies that modification timestamp and size are stable. """ if is_file_locked(filepath): return False try: stat_initial = filepath.stat() initial_size = stat_initial.st_size initial_mtime = stat_initial.st_mtime # Check if the file was modified very recently compared to current time current_time = time.time() if (current_time - initial_mtime) < min_stable_seconds: # File was modified less than min_stable_seconds ago; perform sample check time.sleep(sample_interval_seconds) stat_second = filepath.stat() if stat_second.st_size != initial_size or stat_second.st_mtime != initial_mtime: return False return True except Exception: return False class DirectoryScanner: """Scans Windows source paths for files matching specific backup patterns.""" def __init__(self, source_path: str, file_patterns: str = "*.bak,*.mdf", min_stable_seconds: int = 60): self.source_path = Path(source_path).resolve() self.patterns = [p.strip() for p in file_patterns.split(",") if p.strip()] self.min_stable_seconds = min_stable_seconds def scan(self) -> List[Path]: """Returns list of all matching files that are stable and ready for backup.""" if not self.source_path.exists(): return [] matched_files: List[Path] = [] if self.source_path.is_file(): if self._matches_patterns(self.source_path.name): matched_files.append(self.source_path) return matched_files for root, _, files in os.walk(self.source_path): for file in files: if self._matches_patterns(file): full_path = Path(root) / file matched_files.append(full_path) return matched_files def _matches_patterns(self, filename: str) -> bool: if not self.patterns or "*" in self.patterns: return True for pattern in self.patterns: if fnmatch.fnmatch(filename.lower(), pattern.lower()): return True return False