""" Backup Engine - TimeMachineBackup Incremental backup system with hard-link deduplication Extracted from BackupSystem for integration with Scheduler """ import os import shutil import hashlib import json from datetime import datetime, timedelta from pathlib import Path import tempfile from loguru import logger import fnmatch class TimeMachineBackup: def __init__(self, backup_root): # Espandi ~ e variabili d'ambiente (es. %USERPROFILE%) if isinstance(backup_root, Path): backup_root = str(backup_root) backup_root = os.path.expandvars(os.path.expanduser(str(backup_root))) self.backup_root = Path(backup_root) self.backup_root.mkdir(parents=True, exist_ok=True) self.metadata_file = self.backup_root / "backup_metadata.json" self.lock_file = self.backup_root / ".backup.lock" self.store_dir = self.backup_root / "store" self.store_dir.mkdir(parents=True, exist_ok=True) self.load_metadata() def load_metadata(self): """Carica i metadati dei backup esistenti""" if self.metadata_file.exists(): try: with open(self.metadata_file, 'r', encoding='utf-8') as f: self.metadata = json.load(f) except Exception: self.metadata = {"backups": {}, "file_hashes": {}} else: self.metadata = {"backups": {}, "file_hashes": {}} # Ensure new keys for extended features if "path_cache" not in self.metadata: self.metadata["path_cache"] = {} def _windows_fs_type(self, path: Path): """Ritorna il tipo di filesystem (es. 'NTFS', 'exFAT') su Windows oppure None.""" try: if os.name != 'nt': return None root = Path(path).anchor or str(path) import ctypes from ctypes import wintypes GetVolumeInformationW = ctypes.windll.kernel32.GetVolumeInformationW GetVolumeInformationW.argtypes = [ wintypes.LPCWSTR, wintypes.LPWSTR, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD), ctypes.POINTER(wintypes.DWORD), ctypes.POINTER(wintypes.DWORD), wintypes.LPWSTR, wintypes.DWORD, ] ser = wintypes.DWORD() mcl = wintypes.DWORD() flags = wintypes.DWORD() fsname_buf = ctypes.create_unicode_buffer(255) ok = GetVolumeInformationW(root, None, 0, ctypes.byref(ser), ctypes.byref(mcl), ctypes.byref(flags), fsname_buf, 255) if ok: return fsname_buf.value or None except Exception: return None return None def fs_supports_hardlinks(self, path: Path): """Prova a capire se il FS supporta hard links. Ritorna (supports: bool|None, fsname: str|None). """ try: if os.name == 'nt': fsname = self._windows_fs_type(path) if fsname: return (fsname.upper() == 'NTFS', fsname) return (None, None) # Su POSIX non tentiamo rilevamenti sofisticati qui. return (None, None) except Exception: return (None, None) def iter_source_files(self, source_path: Path, exclude): """Itera i file di una sorgente con pruning veloce delle directory inutili. - Usa os.walk(topdown=True) per poter modificare `dirs` in-place ed evitare di scendere in alberi enormi come .git, node_modules, venv, ecc. - Applica i pattern di exclude sia alle directory (per il pruning) sia ai file. - Non segue symlink. """ try: import fnmatch as _fnm prune_dirnames = {".git", "node_modules", "venv", ".venv", "__pycache__", ".cache", "tmp", "Temp"} for root, dirs, files in os.walk(str(source_path), topdown=True, followlinks=False): # Pruning directory: rimuovi quelle note e quelle che matchano gli exclude pruned = [] for d in list(dirs): # Skip symlinked directories dpath = Path(root) / d if dpath.is_symlink(): pruned.append(d) continue if d in prune_dirnames: pruned.append(d) continue if exclude: rel_dir = str((Path(root) / d).relative_to(source_path)).replace("\\", "/") name = d for pat in exclude: if _fnm.fnmatch(rel_dir, pat) or _fnm.fnmatch(name, pat): pruned.append(d) break # effettua il pruning in-place for d in pruned: try: dirs.remove(d) except ValueError: pass # File nel livello corrente for f in files: fpath = Path(root) / f try: if fpath.is_symlink(): continue # Applica exclude if exclude: rel_file = str(fpath.relative_to(source_path)).replace("\\", "/") name = fpath.name skip = False for pat in exclude: if _fnm.fnmatch(rel_file, pat) or _fnm.fnmatch(name, pat): skip = True break if skip: continue yield fpath except Exception: # Qualsiasi errore di accesso: salta continue except Exception: # Fallback di sicurezza: se qualcosa va storto, non restituisce nulla return def _windows_reserved_segment(self, relative_path: Path): """Ritorna il segmento riservato di Windows trovato nel percorso relativo, altrimenti None. Su Windows non sono permessi nomi di dispositivo come: CON, PRN, AUX, NUL, COM1..COM9, LPT1..LPT9. Sono vietati anche con estensione (es. NUL.txt). """ try: if os.name != 'nt': return None reserved = { 'CON', 'PRN', 'AUX', 'NUL', 'COM1','COM2','COM3','COM4','COM5','COM6','COM7','COM8','COM9', 'LPT1','LPT2','LPT3','LPT4','LPT5','LPT6','LPT7','LPT8','LPT9' } for part in relative_path.parts: # Considera il nome fino al primo punto (NUL.txt -> NUL) base = part.split('.', 1)[0].strip().upper() if base in reserved: return part return None except Exception: return None def save_metadata(self): """Salva i metadati""" tmp = None try: fd, tmp = tempfile.mkstemp(prefix="metadata_", suffix=".json", dir=str(self.backup_root)) with os.fdopen(fd, 'w', encoding='utf-8') as f: json.dump(self.metadata, f, indent=2, ensure_ascii=False) os.replace(tmp, self.metadata_file) finally: if tmp and os.path.exists(tmp): try: os.remove(tmp) except OSError: pass def create_backup_log_file(self, backup_path, backup_info): """Crea un file di log leggibile per il backup corrente.""" log_file = backup_path / "BACKUP_INFO.txt" try: total_files = backup_info.get('files_linked', 0) + backup_info.get('files_copied', 0) files_in_store = backup_info.get('files_in_store', 0) files_linked = backup_info.get('files_linked', 0) files_copied = backup_info.get('files_copied', 0) total_size = backup_info.get('total_size', 0) timestamp = backup_info.get('timestamp', '') sources = backup_info.get('source_dirs', []) # Calcola dimensione apparente della snapshot (somma tutti i file) apparent_size = 0 file_count = 0 for item in backup_path.rglob('*'): if item.is_file() and item.name != "BACKUP_INFO.txt": try: apparent_size += item.stat().st_size file_count += 1 except (OSError, PermissionError): pass # Determina stato del backup status = "✓ OK - Hard links funzionanti" if files_copied > files_linked: status = "⚠ ATTENZIONE - Troppi file copiati (filesystem non supporta hard links?)" elif files_copied > 0: status = "⚠ PARZIALE - Alcuni file copiati invece di linked" # Calcola percentuale hard links hardlink_percentage = (files_linked * 100 // total_files) if total_files > 0 else 0 with open(log_file, 'w', encoding='utf-8') as f: f.write("=" * 70 + "\n") f.write(" RIEPILOGO BACKUP\n") f.write("=" * 70 + "\n\n") f.write(f"Data e ora: {timestamp}\n") f.write(f"Nome backup: {backup_path.name}\n") f.write(f"Stato: {status}\n\n") f.write("-" * 70 + "\n") f.write("STATISTICHE FILE\n") f.write("-" * 70 + "\n") f.write(f"File totali nella snapshot: {total_files:>10}\n") f.write(f" • Hard links (risparmiati): {files_linked:>10} ({hardlink_percentage}%)\n") f.write(f" • Copie fisiche: {files_copied:>10} ({100-hardlink_percentage}%)\n") f.write(f"Nuovi file nel content store: {files_in_store:>10}\n\n") f.write("-" * 70 + "\n") f.write("OCCUPAZIONE SPAZIO\n") f.write("-" * 70 + "\n") f.write(f"Dimensione apparente snapshot: {apparent_size:>15,} bytes ({apparent_size/1024/1024:>8.1f} MB)\n") f.write(f"Nuovi dati nel content store: {total_size:>15,} bytes ({total_size/1024/1024:>8.1f} MB)\n") if apparent_size > 0 and total_size > 0: if total_size < apparent_size: savings = 100 - (total_size * 100 / apparent_size) f.write(f"Risparmio spazio (hard links): {savings:>14.1f}%\n") f.write("\n") f.write("-" * 70 + "\n") f.write("DIRECTORY SORGENTI\n") f.write("-" * 70 + "\n") for i, src in enumerate(sources, 1): f.write(f"{i}. {src}\n") f.write("\n") if files_copied > files_linked: f.write("=" * 70 + "\n") f.write("⚠ AVVISO IMPORTANTE\n") f.write("=" * 70 + "\n") f.write("Questo backup contiene più COPIE che HARD LINKS!\n") f.write("Stai occupando PIÙ SPAZIO del necessario.\n\n") f.write("Causa probabile:\n") f.write(" • Filesystem non supporta hard links (FAT32, exFAT)\n\n") f.write("Soluzione:\n") f.write(" • Formatta il disco come NTFS (Windows) o ext4 (Linux)\n") f.write(" • Esegui un nuovo backup dopo la formattazione\n") f.write("=" * 70 + "\n") except Exception as e: logger.warning(f"Impossibile creare file di log: {e}") def calculate_file_hash(self, file_path): """Calcola l'hash SHA-256 di un file""" hasher = hashlib.sha256() try: with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): hasher.update(chunk) return hasher.hexdigest() except (OSError, PermissionError): return None def hash_to_store_path(self, file_hash: str) -> Path: """Percorso nel content store per un dato hash.""" a, b = file_hash[:2], file_hash[2:4] return self.backup_root / "store" / a / b / file_hash def acquire_lock(self): """Evita sovrapposizione backup (lock file semplice).""" try: fd = os.open(self.lock_file, os.O_CREAT | os.O_EXCL | os.O_WRONLY) with os.fdopen(fd, 'w', encoding='utf-8') as f: f.write(json.dumps({"pid": os.getpid(), "time": datetime.now().isoformat()})) return True except FileExistsError: print("Un altro backup sembra in esecuzione (lock presente).") return False def release_lock(self): try: if self.lock_file.exists(): self.lock_file.unlink() except OSError: pass def create_backup(self, source_dirs, backup_name=None, exclude=None, progress=None, progress_precount=True): """Crea un nuovo backup. - Se `progress` è None (CLI senza barra), niente pre‑scansione. - Se `progress` è fornito: * con `progress_precount=True` (default, usato dalla GUI): conta rapida per inizializzare il totale. * con `progress_precount=False` (CLI): evita la pre‑scansione, mostrando comunque avanzamento incrementale. """ if backup_name is None: backup_name = f"backup-{datetime.now().strftime('%Y-%m-%d-%H%M%S')}" backup_path = self.backup_root / backup_name backup_path.mkdir(exist_ok=True) logger.info(f"Creando backup: {backup_name}") backup_info = { "timestamp": datetime.now().isoformat(), "source_dirs": [], "files_in_store": 0, # Nuovi file aggiunti al content store "files_linked": 0, # File nella snapshot che sono hard links "files_copied": 0, # File copiati direttamente (fallback) "total_size": 0 # Dimensione totale dei nuovi dati fisici } # Se richiesto, esegui una conta veloce per avere un totale (usato dalla GUI) total = 0 if progress and progress_precount: for source_dir in source_dirs: source_path = Path(source_dir) if not source_path.exists(): logger.warning(f"Avviso: {source_dir} non esiste") continue for _ in self.iter_source_files(source_path, exclude): total += 1 try: progress({"phase": "start", "total": total}) except Exception: pass processed = 0 source_idx = -1 for source_dir in source_dirs: source_path = Path(source_dir) if not source_path.exists(): logger.warning(f"Avviso: {source_dir} non esiste") continue logger.info(f" Backup di: {source_dir}") dest_dir = backup_path / source_path.name dest_dir.mkdir(exist_ok=True) source_idx += 1 source_processed = 0 if progress: try: progress({ "phase": "source_start", "source": str(source_path), "source_index": source_idx, }) except Exception: pass # Enumerazione ottimizzata in passaggio unico (CLI) o secondo pass (GUI) for file_path in self.iter_source_files(source_path, exclude): self.process_file_v2(file_path, dest_dir, backup_info, source_path, exclude) processed += 1 source_processed += 1 if progress: try: progress({ "phase": "progress", "processed": processed, "total": total, "file": str(file_path), "source": str(source_path), "source_index": source_idx, "source_processed": source_processed, }) except Exception: pass backup_info["source_dirs"].append(str(source_dir)) if progress: try: progress({ "phase": "source_end", "source": str(source_path), "source_index": source_idx, "source_processed": source_processed, }) except Exception: pass # Salva metadati del backup self.metadata["backups"][backup_name] = backup_info self.save_metadata() # Crea file di log per questo backup self.create_backup_log_file(backup_path, backup_info) # Log dettagliato del risultato total_files = backup_info['files_linked'] + backup_info['files_copied'] logger.info( f"Backup completato: {total_files} file totali " f"({backup_info['files_linked']} hard links, {backup_info['files_copied']} copie), " f"{backup_info['files_in_store']} nuovi file nel content store" ) # Avviso se troppi file sono copie invece di hard links if backup_info['files_copied'] > backup_info['files_linked']: logger.warning("=" * 70) logger.warning("⚠️ ATTENZIONE: Più copie che hard links!") logger.warning("⚠️ Il backup sta OCCUPANDO PIÙ SPAZIO del necessario") logger.warning("⚠️ Causa probabile: Filesystem non supporta hard links (FAT32/exFAT)") logger.warning("⚠️ Soluzione: Formatta il disco come NTFS (Windows) o ext4 (Linux)") logger.warning("=" * 70) def process_file(self, file_path, dest_dir, backup_info, source_root): """Processa un singolo file (copia o crea hard link)""" try: relative_path = file_path.relative_to(source_root) dest_file = dest_dir / relative_path dest_file.parent.mkdir(parents=True, exist_ok=True) # Calcola hash del file file_hash = self.calculate_file_hash(file_path) if file_hash is None: return file_size = file_path.stat().st_size # Controlla se il file esiste già in altri backup if file_hash in self.metadata["file_hashes"]: existing_file = Path(self.metadata["file_hashes"][file_hash]) if existing_file.exists(): # Crea hard link al file esistente if dest_file.exists(): dest_file.unlink() try: os.link(str(existing_file), str(dest_file)) backup_info["files_linked"] += 1 return except OSError: pass # Fallback alla copia normale # Copia il file (nuovo o fallback) shutil.copy2(file_path, dest_file) backup_info["files_copied"] += 1 backup_info["total_size"] += file_size # Aggiorna metadati self.metadata["file_hashes"][file_hash] = str(dest_file) except Exception as e: print(f"Errore processando {file_path}: {e}") def list_backups(self): """Lista tutti i backup disponibili""" if not self.metadata["backups"]: print("Nessun backup trovato") return print("Backup disponibili:") for name, info in self.metadata["backups"].items(): date = datetime.fromisoformat(info["timestamp"]).strftime("%Y-%m-%d %H:%M:%S") print(f" {name} - {date}") # Mostra statistiche in base ai campi disponibili (retrocompatibilità) if "files_in_store" in info: total = info.get('files_linked', 0) + info.get('files_copied', 0) print(f" File totali: {total} ({info.get('files_linked', 0)} hard links, " f"{info.get('files_copied', 0)} copie)") print(f" Nuovi nel content store: {info.get('files_in_store', 0)}") else: # Vecchio formato (prima della correzione) print(f" File: {info.get('files_copied', 0)} copiati, " f"{info.get('files_linked', 0)} collegamenti") def restore_backup(self, backup_name, restore_path=None): """Ripristina un backup""" if backup_name not in self.metadata["backups"]: print(f"Backup {backup_name} non trovato") return backup_path = self.backup_root / backup_name if not backup_path.exists(): print(f"Cartella backup {backup_name} non trovata") return if restore_path is None: restore_path = Path.home() / "restored_backup" restore_path = Path(restore_path) restore_path.mkdir(exist_ok=True) print(f"Ripristinando backup {backup_name} in {restore_path}") # Copia tutto il contenuto del backup (dir e file a radice) for item in backup_path.iterdir(): if item.is_dir(): dest_dir = restore_path / item.name shutil.copytree(item, dest_dir, dirs_exist_ok=True) elif item.is_file(): shutil.copy2(item, restore_path / item.name) print(f"Ripristino completato in {restore_path}") def restore_file(self, backup_name, file_path, restore_to=None): """Ripristina un file specifico""" backup_path = self.backup_root / backup_name source_file = backup_path / file_path if not source_file.exists(): print(f"File {file_path} non trovato nel backup {backup_name}") return if restore_to is None: restore_to = Path.cwd() / Path(file_path).name else: restore_to = Path(restore_to) restore_to.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_file, restore_to) print(f"File ripristinato: {restore_to}") def restore_directory(self, backup_name, directory_path, restore_to=None): """Ripristina una directory specifica dal backup. directory_path è relativo alla radice del backup (es: 'Documents' o 'Documents/Progetti').""" backup_path = self.backup_root / backup_name source_dir = backup_path / directory_path if not source_dir.exists() or not source_dir.is_dir(): print(f"Directory {directory_path} non trovata nel backup {backup_name}") return if restore_to is None: restore_to = Path.cwd() / source_dir.name else: restore_to = Path(restore_to) restore_to.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(source_dir, restore_to, dirs_exist_ok=True) print(f"Directory ripristinata in: {restore_to}") def process_file_v2(self, file_path, dest_dir, backup_info, source_root, exclude): """Nuova versione: usa content store e cache path.""" try: # Symlink: salta if file_path.is_symlink(): logger.warning(f"Salto symlink: {file_path}") return relative_path = file_path.relative_to(source_root) dest_file = dest_dir / relative_path dest_file.parent.mkdir(parents=True, exist_ok=True) # Evita errori su nomi riservati di Windows (es. 'nul') bad = self._windows_reserved_segment(relative_path) if bad: logger.warning( f"Salto elemento con nome riservato Windows: '{bad}' nel percorso '{relative_path}'" ) return # Log di debug opzionale del file in analisi logger.debug(f"Analizzo: {file_path} -> {dest_file}") # Escludi pattern if exclude: rel_str = str(relative_path).replace("\\", "/") name = file_path.name for pat in exclude: if fnmatch.fnmatch(rel_str, pat) or fnmatch.fnmatch(name, pat): return # Cache per path st = file_path.stat() abs_path = str(file_path) cached = self.metadata.get("path_cache", {}).get(abs_path) if cached and cached.get("mtime_ns") == st.st_mtime_ns and cached.get("size") == st.st_size: file_hash = cached.get("hash") else: file_hash = self.calculate_file_hash(file_path) if file_hash is None: return self.metadata.setdefault("path_cache", {})[abs_path] = { "mtime_ns": st.st_mtime_ns, "size": st.st_size, "hash": file_hash, } file_size = st.st_size store_path = self.hash_to_store_path(file_hash) # Assicura nel content store newly_stored = False if not store_path.exists(): store_path.parent.mkdir(parents=True, exist_ok=True) tmp_fd, tmp_path = tempfile.mkstemp(prefix="obj_", dir=str(store_path.parent)) os.close(tmp_fd) try: shutil.copy2(file_path, tmp_path) os.replace(tmp_path, store_path) finally: if os.path.exists(tmp_path): try: os.remove(tmp_path) except OSError: pass # Nuovo file aggiunto al content store newly_stored = True backup_info.setdefault("files_in_store", 0) backup_info["files_in_store"] += 1 backup_info["total_size"] += file_size # Link allo snapshot if dest_file.exists(): dest_file.unlink() try: os.link(str(store_path), str(dest_file)) # Hard link creato con successo backup_info["files_linked"] += 1 except OSError as e: # Fallback: copia diretta (filesystem non supporta hard links) # Logga il primo errore per diagnostica winerr = getattr(e, 'winerror', None) if winerr != 1142 and not backup_info.get("hardlink_failed"): backup_info["hardlink_failed"] = True logger.warning(f"Hard link fallito (usando copia): {e}") # Se riusciamo a rilevare che il volume non supporta hard links, avvisa in modo accurato supports, fsname = self.fs_supports_hardlinks(self.backup_root) if supports is False: try: vol = str(Path(self.backup_root).anchor or self.backup_root) except Exception: vol = str(self.backup_root) logger.warning(f"Il volume di backup {vol} è '{fsname}' e non supporta hard links: verranno effettuate copie.") shutil.copy2(file_path, dest_file) backup_info["files_copied"] += 1 # Aggiungi size solo se non già aggiunto durante newly_stored if not newly_stored: backup_info["total_size"] += file_size # Metadati self.metadata.setdefault("file_hashes", {})[file_hash] = str(store_path) backup_info.setdefault("hashes", []) if file_hash not in backup_info["hashes"]: backup_info["hashes"].append(file_hash) except Exception as e: try: # Prova a includere anche il percorso di destinazione se disponibile logger.error(f"Errore processando {file_path} -> {dest_file}: {e}") except Exception: logger.error(f"Errore processando {file_path}: {e}") def prune_backups(self, keep=None, older_than_days=None): """Rimuove vecchi backup per conteggio o età.""" backups = self.metadata.get("backups", {}) if not backups: print("Nessun backup da rimuovere") return def parse_dt(info): try: return datetime.fromisoformat(info.get("timestamp")) except Exception: return datetime.min ordered = sorted(backups.items(), key=lambda kv: parse_dt(kv[1])) to_delete = [] if older_than_days is not None: cutoff = datetime.now() - timedelta(days=older_than_days) to_delete = [name for name, info in ordered if parse_dt(info) < cutoff] elif keep is not None: if keep < 0: keep = 0 if len(ordered) > keep: to_delete = [name for name, _ in ordered[:-keep]] else: print("Specifica --keep o --older-than") return for name in to_delete: path = self.backup_root / name try: if path.exists(): shutil.rmtree(path) backups.pop(name, None) print(f"Rimosso backup {name}") except Exception as e: print(f"Impossibile rimuovere {name}: {e}") self.save_metadata() def gc_store(self): """Rimuove oggetti nello store non referenziati dai backup (se possibile).""" backups = self.metadata.get("backups", {}) if not backups: print("Nessun backup presente; niente da fare") return for info in backups.values(): if "hashes" not in info: print("GC saltata: alcuni backup non hanno elenco 'hashes' (versione precedente)") return used = set() for info in backups.values(): used.update(info.get("hashes", [])) removed = 0 if self.store_dir.exists(): for root, dirs, files in os.walk(self.store_dir): for fname in files: if fname not in used: fpath = Path(root) / fname try: fpath.unlink() removed += 1 except Exception: pass print(f"GC store: rimossi {removed} oggetti non referenziati")