feat: implementar Fase 2 del roadmap (Modo Replica Exacta / Mirroring)
This commit is contained in:
@@ -45,12 +45,28 @@ def is_file_stable(filepath: Path, min_stable_seconds: int = 60, sample_interval
|
||||
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", min_stable_seconds: int = 60):
|
||||
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]:
|
||||
@@ -61,15 +77,20 @@ class DirectoryScanner:
|
||||
matched_files: List[Path] = []
|
||||
|
||||
if self.source_path.is_file():
|
||||
if self._matches_patterns(self.source_path.name):
|
||||
matched_files.append(self.source_path)
|
||||
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, _, files in os.walk(self.source_path):
|
||||
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):
|
||||
if self._matches_patterns(file) and not self._matches_exclude(file):
|
||||
full_path = Path(root) / file
|
||||
matched_files.append(full_path)
|
||||
if not self._should_skip_by_attributes(full_path):
|
||||
matched_files.append(full_path)
|
||||
|
||||
return matched_files
|
||||
|
||||
@@ -80,3 +101,29 @@ class DirectoryScanner:
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user