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: logging.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) logging.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(): logging.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(): logging.warning(f"Avviso: {source_dir} non esiste") continue logging.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'] logging.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']: logging.warning("=" * 70) logging.warning("⚠️ ATTENZIONE: Più copie che hard links!") logging.warning("⚠️ Il backup sta OCCUPANDO PIÙ SPAZIO del necessario") logging.warning("⚠️ Causa probabile: Filesystem non supporta hard links (FAT32/exFAT)") logging.warning("⚠️ Soluzione: Formatta il disco come NTFS (Windows) o ext4 (Linux)") logging.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(): logging.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: logging.warning( f"Salto elemento con nome riservato Windows: '{bad}' nel percorso '{relative_path}'" ) return # Log di debug opzionale del file in analisi logging.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 logging.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) logging.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 logging.error(f"Errore processando {file_path} -> {dest_file}: {e}") except Exception: logging.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")