130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
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
|
|
|
|
import platform
|
|
import stat
|
|
|
|
class DirectoryScanner:
|
|
"""Scans Windows source paths for files matching specific backup patterns."""
|
|
|
|
def __init__(
|
|
self,
|
|
source_path: str,
|
|
file_patterns: str = "*.bak,*.mdf",
|
|
exclude_patterns: str = "",
|
|
skip_hidden: bool = False,
|
|
skip_system: bool = False,
|
|
skip_readonly: bool = False,
|
|
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.exclude_patterns = [p.strip() for p in exclude_patterns.split(",") if p.strip()]
|
|
self.skip_hidden = skip_hidden
|
|
self.skip_system = skip_system
|
|
self.skip_readonly = skip_readonly
|
|
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) and not self._matches_exclude(self.source_path.name):
|
|
if not self._should_skip_by_attributes(self.source_path):
|
|
matched_files.append(self.source_path)
|
|
return matched_files
|
|
|
|
for root, dirs, files in os.walk(self.source_path):
|
|
# Prune excluded directories from search in-place
|
|
dirs[:] = [d for d in dirs if not self._matches_exclude(d)]
|
|
|
|
for file in files:
|
|
if self._matches_patterns(file) and not self._matches_exclude(file):
|
|
full_path = Path(root) / file
|
|
if not self._should_skip_by_attributes(full_path):
|
|
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
|
|
|
|
def _matches_exclude(self, filename: str) -> bool:
|
|
for pattern in self.exclude_patterns:
|
|
if fnmatch.fnmatch(filename.lower(), pattern.lower()):
|
|
return True
|
|
return False
|
|
|
|
def _should_skip_by_attributes(self, filepath: Path) -> bool:
|
|
if platform.system() != "Windows":
|
|
return False
|
|
|
|
try:
|
|
st = filepath.stat()
|
|
attrs = getattr(st, "st_file_attributes", 0)
|
|
if not attrs:
|
|
return False
|
|
|
|
if self.skip_readonly and (attrs & 0x01): # FILE_ATTRIBUTE_READONLY
|
|
return True
|
|
if self.skip_hidden and (attrs & 0x02): # FILE_ATTRIBUTE_HIDDEN
|
|
return True
|
|
if self.skip_system and (attrs & 0x04): # FILE_ATTRIBUTE_SYSTEM
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|