diff --git a/nas-dedup.copy.py b/nas-dedup.copy.py new file mode 100644 index 0000000..1d2cec8 --- /dev/null +++ b/nas-dedup.copy.py @@ -0,0 +1,712 @@ +#!/usr/bin/env python3 +""" +nas-dedup-copy.py +Szybkie, deduplikujące kopiowanie z dysku źródłowego na NAS (przez zamontowany +katalog docelowy, np. NFS z TrueNAS). + +Realizuje: + 1. DEDUP W ŹRÓDLE: pliki grupowane po ROZMIARZE; hashujemy TYLKO te, których + rozmiar z czymś koliduje. Identyczna treść (rozmiar+hash) kopiowana raz. + Nazwy i ścieżki nie mają znaczenia — liczy się TREŚĆ. + 2. DEDUP WZGLĘDEM ISTNIEJĄCYCH DANYCH (--against DIR, można podać wiele razy): + jeśli treść pliku ze źródła już leży w którymś ze wskazanych katalogów + (np. na NAS-ie, pod dowolną nazwą), plik jest POMIJANY. + 3. FILTR: pomija pliki < MIN_SIZE (domyślnie 1 MiB), CHYBA że jpg/jpeg/png/gif. + 4. SPRAWDZENIE MIEJSCA: po dedup+filtrze liczy realny rozmiar do skopiowania + i porównuje z wolnym miejscem na woluminie docelowym. Bez miejsca -> stop. + 5. HEARTBEAT: co --heartbeat sekund (domyślnie 120) wypisuje na stderr znak + życia: etap, WSZYSTKIE równolegle przetwarzane pliki z postępem każdego, + postęp całości etapu, tempo i ETA. + 6. WIELOWĄTKOWOŚĆ: + --workers N (domyślnie 4) — wątki KOPIOWANIA i hashowania --against. + --hash-workers M (domyślnie 1) — wątki hashowania ŹRÓDŁA. + UWAGA: zwiększaj tylko gdy źródło to SSD. Na HDD + równoległe odczyty = seek-thrashing = WOLNIEJ. + 7. PLIK STANU (--state stan.json, OPCJONALNY): + - cache HASHY: każdy policzony hash ląduje w pliku stanu (klucz: ścieżka, + walidacja: rozmiar + mtime). Po przerwaniu i restarcie pliki niezmienione + NIE są hashowane ponownie — ani źródło, ani --against. + - rejestr SKOPIOWANYCH: ukończone kopie są odnotowane; restart pomija je + bez dotykania NFS (szybciej niż stat na tysiącach plików). + - zapis ATOMOWY (tmp + rename) co --state-interval sekund (domyślnie 30), + na końcu każdego etapu oraz przy SIGINT (Ctrl+C) i SIGTERM. + - plik przerwany w połowie KOPIOWANIA nie jest w rejestrze -> po restarcie + zostanie skopiowany od zera (niepełny plik w celu ma zły rozmiar, więc + złapie go też --skip-existing). Przerwany w połowie HASH liczy się od zera. + - stan jest wiązany z backendem hasha (xxh64/sha256); zmiana backendu + unieważnia cache hashy (rejestr kopii zostaje). + +Optymalizacja hashowania (kluczowa dla prędkości): + - Plik ŹRÓDŁA hashujemy tylko, gdy jego rozmiar koliduje z innym plikiem + źródła LUB występuje wśród istniejących danych (--against). + - Plik ISTNIEJĄCY (--against) hashujemy tylko, gdy jego rozmiar występuje + w zestawie do skopiowania. Reszty nie dotykamy (nie może być duplikatem). + - Wynik dedupu jest DETERMINISTYCZNY niezależnie od liczby wątków. + +Bezpieczeństwo: + - Źródło i katalogi --against są tylko CZYTANE. Skrypt NICZEGO nie usuwa. + - Domyślnie: analiza + raport, potem pyta o potwierdzenie (--yes pomija). + - --dry-run: sama analiza. --skip-existing (domyślnie ON): wznawianie po przerwaniu. + +Przykłady: + # pełny zestaw na noc: dedup vs NAS, 6 wątków, stan zapisywany: + ./nas-dedup-copy.py --source /mnt/source --dest /mnt/pve/truenas-nfs/import \\ + --against /mnt/pve/truenas-nfs --yes --workers 6 \\ + --state /root/import-stan.json + + # po przerwaniu: to samo polecenie — hashe i kopie wracają z pliku stanu. +""" + +import argparse +import json +import os +import signal +import sys +import stat +import shutil +import hashlib +import threading +import time +from collections import Counter, defaultdict +from concurrent.futures import ThreadPoolExecutor, as_completed + +# --- backend hasha: xxhash jeśli dostępny (szybki), inaczej sha256 --- +try: + import xxhash + def make_hasher(): + return xxhash.xxh64() + HASH_NAME = "xxh64" +except ImportError: + def make_hasher(): + return hashlib.sha256() + HASH_NAME = "sha256 (zainstaluj 'xxhash' dla szybszego dedupu)" + +HASH_BACKEND = HASH_NAME.split()[0] # "xxh64" / "sha256" +DEFAULT_KEEP_EXTS = {".jpg", ".jpeg", ".png", ".gif"} +MIB = 1 << 20 +COPY_CHUNK = 4 * MIB +HB_MAX_FILES = 4 # ile równoległych plików pokazywać w linii heartbeatu +STATE_VERSION = 1 + + +def human(n): + for unit in ("B", "KiB", "MiB", "GiB", "TiB"): + if abs(n) < 1024.0: + return f"{n:,.1f} {unit}" + n /= 1024.0 + return f"{n:,.1f} PiB" + + +# ====================== PLIK STANU ====================== + +class State: + """Trwały stan w JSON: cache hashy (walidacja size+mtime) i rejestr + ukończonych kopii. Zapis atomowy (tmp + os.replace). Thread-safe. + Gdy path=None, wszystkie operacje są no-op (tryb bez stanu).""" + + def __init__(self, path): + self.path = path + self.lock = threading.Lock() + self.dirty = False + self.hash_hits = 0 + self.copy_hits = 0 + self.data = { + "version": STATE_VERSION, + "hash_backend": HASH_BACKEND, + "hashes": {}, # abspath -> {"size": int, "mtime": float, "hash": str} + "copied": {}, # relpath -> {"size": int} + } + if path and os.path.exists(path): + self._load() + + def _load(self): + try: + with open(self.path, "r", encoding="utf-8") as f: + loaded = json.load(f) + except (OSError, json.JSONDecodeError) as e: + print(f"UWAGA: plik stanu nieczytelny ({e}) — zaczynam z pustym stanem.", + file=sys.stderr) + return + if loaded.get("version") != STATE_VERSION: + print("UWAGA: niezgodna wersja pliku stanu — zaczynam z pustym stanem.", + file=sys.stderr) + return + if loaded.get("hash_backend") != HASH_BACKEND: + print(f"UWAGA: plik stanu ma hashe {loaded.get('hash_backend')}, " + f"a działamy na {HASH_BACKEND} — cache hashy odrzucony, " + f"rejestr kopii zachowany.", file=sys.stderr) + loaded["hashes"] = {} + loaded["hash_backend"] = HASH_BACKEND + self.data = loaded + self.data.setdefault("hashes", {}) + self.data.setdefault("copied", {}) + print(f"Stan wczytany: {len(self.data['hashes']):,} hashy w cache, " + f"{len(self.data['copied']):,} plików odnotowanych jako skopiowane.", + file=sys.stderr) + + # --- hashe --- + def get_hash(self, path, size, mtime): + if not self.path: + return None + with self.lock: + e = self.data["hashes"].get(path) + if e and e["size"] == size and e["mtime"] == mtime: + self.hash_hits += 1 + return e["hash"] + return None + + def put_hash(self, path, size, mtime, digest): + if not self.path: + return + with self.lock: + self.data["hashes"][path] = {"size": size, "mtime": mtime, "hash": digest} + self.dirty = True + + # --- kopie --- + def is_copied(self, rel, size): + if not self.path: + return False + with self.lock: + e = self.data["copied"].get(rel) + if e and e["size"] == size: + self.copy_hits += 1 + return True + return False + + def put_copied(self, rel, size): + if not self.path: + return + with self.lock: + self.data["copied"][rel] = {"size": size} + self.dirty = True + + # --- zapis --- + def save(self, force=False): + if not self.path: + return + with self.lock: + if not self.dirty and not force: + return + tmp = self.path + ".tmp" + try: + with open(tmp, "w", encoding="utf-8") as f: + json.dump(self.data, f) + os.replace(tmp, self.path) # atomowo + self.dirty = False + except OSError as e: + print(f"UWAGA: nie mogę zapisać pliku stanu: {e}", file=sys.stderr) + + +STATE = State(None) # podmieniane w main() + + +def state_saver_loop(interval, stop_event): + while not stop_event.wait(interval): + STATE.save() + + +# ====================== HEARTBEAT ====================== + +class Progress: + """Współdzielony stan postępu. Wiele wątków-workerów pisze (każdy o swoim + pliku), wątek heartbeat czyta i drukuje zbiorczą linię.""" + + def __init__(self): + self.lock = threading.Lock() + self.phase = "start" + self.detail = "" + self.current = {} # path -> [done, total, op] — pliki w locie + self.items_done = 0 + self.items_total = 0 + self.bytes_done = 0 # bajty UKOŃCZONYCH plików w etapie + self.bytes_total = 0 + self.phase_start = time.monotonic() + self._hb_bytes = 0 + self._hb_time = time.monotonic() + + def set_phase(self, phase, items_total=0, bytes_total=0, detail=""): + with self.lock: + self.phase = phase + self.detail = detail + self.current = {} + self.items_done = 0 + self.items_total = items_total + self.bytes_done = 0 + self.bytes_total = bytes_total + self.phase_start = time.monotonic() + self._hb_bytes = 0 + self._hb_time = time.monotonic() + + def start_file(self, path, size, op): + with self.lock: + self.current[path] = [0, size, op] + + def advance(self, path, nbytes): + with self.lock: + cur = self.current.get(path) + if cur: + cur[0] += nbytes + + def finish_file(self, path): + with self.lock: + cur = self.current.pop(path, None) + if cur: + self.bytes_done += cur[1] + self.items_done += 1 + + def tick(self, items=1, add_bytes=0): + """Licznik dla pominięć/etapów bez czytania danych (np. cache hit).""" + with self.lock: + self.items_done += items + self.bytes_done += add_bytes + + def line(self): + with self.lock: + now = time.monotonic() + inflight = sum(c[0] for c in self.current.values()) + done_now = self.bytes_done + inflight + dt = max(now - self._hb_time, 1e-6) + rate = (done_now - self._hb_bytes) / dt + self._hb_bytes = done_now + self._hb_time = now + elapsed = int(now - self.phase_start) + + parts = [f"etap: {self.phase}"] + if self.detail: + parts.append(self.detail) + if self.items_total: + parts.append(f"pliki {self.items_done:,}/{self.items_total:,}") + elif self.items_done: + parts.append(f"przetworzono {self.items_done:,} plików") + if self.bytes_total: + pct = 100.0 * done_now / self.bytes_total + parts.append(f"dane {human(done_now)}/{human(self.bytes_total)} ({pct:.1f}%)") + + shown = 0 + for path, (done, total, op) in list(self.current.items()): + if shown >= HB_MAX_FILES: + parts.append(f"(+{len(self.current) - shown} innych w locie)") + break + if total: + fpct = 100.0 * done / total + parts.append(f"{op}: {path} [{human(done)}/{human(total)} = {fpct:.1f}%]") + else: + parts.append(f"{op}: {path}") + shown += 1 + + if rate > 1 and (self.current or self.bytes_total): + parts.append(f"tempo {human(rate)}/s") + if self.bytes_total and done_now < self.bytes_total: + eta = (self.bytes_total - done_now) / rate + parts.append(f"ETA etapu ~{int(eta // 60)}m{int(eta % 60):02d}s") + parts.append(f"czas etapu {elapsed // 60}m{elapsed % 60:02d}s") + return " | ".join(parts) + + +PROGRESS = Progress() + + +def heartbeat_loop(interval, stop_event): + while not stop_event.wait(interval): + ts = time.strftime("%H:%M:%S") + print(f"[ŻYJĘ {ts}] {PROGRESS.line()}", file=sys.stderr, flush=True) + + +# ====================== LOGIKA ====================== + +def hash_file(path, chunk_size=MIB): + h = make_hasher() + with open(path, "rb") as f: + for block in iter(lambda: f.read(chunk_size), b""): + h.update(block) + PROGRESS.advance(path, len(block)) + return h.hexdigest() + + +def cached_hash(path, size): + """Hash z cache stanu (walidacja size+mtime) albo policzony i zapisany.""" + try: + mtime = os.stat(path).st_mtime + except OSError: + raise + d = STATE.get_hash(path, size, mtime) + if d is not None: + PROGRESS.tick(items=0, add_bytes=size) # licz jako "zrobione bajty" bez czytania + return d + d = hash_file(path) + STATE.put_hash(path, size, mtime, d) + return d + + +def parallel_hash(files, workers, progress_every=500, label="hash"): + """Hashuje listę (abspath, relpath, size) w `workers` wątkach. + Zwraca dict abspath -> hexdigest (brak klucza = błąd odczytu). + Postęp per-plik raportowany do PROGRESS; hashe idą przez cache stanu.""" + digests = {} + done_counter = [0] + counter_lock = threading.Lock() + + def job(f): + p, _rel, size = f + PROGRESS.start_file(p, size, "hashuję") + try: + d = cached_hash(p, size) + except OSError: + d = None + PROGRESS.finish_file(p) + with counter_lock: + done_counter[0] += 1 + if done_counter[0] % progress_every == 0: + print(f" ... {label} {done_counter[0]}/{len(files)}", file=sys.stderr) + return p, d + + if workers <= 1: + for f in files: + p, d = job(f) + if d is not None: + digests[p] = d + return digests + + with ThreadPoolExecutor(max_workers=workers) as ex: + for fut in as_completed(ex.submit(job, f) for f in files): + p, d = fut.result() + if d is not None: + digests[p] = d + return digests + + +def scan(root_dir): + """Zwraca (abspath, relpath, size) dla zwykłych plików + liczbę błędów odczytu.""" + files = [] + errors = 0 + for root, _dirs, names in os.walk(root_dir, onerror=lambda e: None): + for name in names: + p = os.path.join(root, name) + try: + st = os.lstat(p) + except OSError: + errors += 1 + continue + if not stat.S_ISREG(st.st_mode): + continue # symlinki, urządzenia, potoki — pomijamy + files.append((p, os.path.relpath(p, root_dir), st.st_size)) + PROGRESS.tick() + return files, errors + + +def apply_filter(files, min_size, keep_exts): + kept, skipped, skipped_bytes = [], 0, 0 + for p, rel, size in files: + ext = os.path.splitext(p)[1].lower() + if size < min_size and ext not in keep_exts: + skipped += 1 + skipped_bytes += size + continue + kept.append((p, rel, size)) + return kept, skipped, skipped_bytes + + +def build_existing_index(existing_files, source_sizes, workers): + """Index treści istniejących danych: {(size, hash)}. + Hashuje tylko pliki, których rozmiar występuje w imporcie. Równolegle.""" + to_hash = [f for f in existing_files if f[2] in source_sizes] + PROGRESS.set_phase("hash istniejących (--against)", + items_total=len(to_hash), + bytes_total=sum(f[2] for f in to_hash), + detail=f"{workers} wątk.") + if to_hash: + print(f" Istniejące dane: {len(to_hash)} plików o pasującym rozmiarze do zahashowania", + file=sys.stderr) + digests = parallel_hash(to_hash, workers, label="hash istniejących") + STATE.save() + index = set() + for p, _rel, size in to_hash: + d = digests.get(p) + if d is not None: + index.add((size, d)) + return index + + +def dedup(kept, existing_index, existing_sizes, hash_workers): + """Dedup źródła + odsianie treści już obecnych w existing_index. + Hashe liczone RÓWNOLEGLE (przez cache stanu), selekcja SEKWENCYJNIE + w kolejności skanu. Zwraca (unique, dup_internal, dup_existing, hashed_src).""" + size_count = Counter(f[2] for f in kept) + need_hash = [f for f in kept if size_count[f[2]] > 1 or f[2] in existing_sizes] + + PROGRESS.set_phase("hash źródła (dedup)", + items_total=len(need_hash), + bytes_total=sum(f[2] for f in need_hash), + detail=f"hash tylko dla kolizji rozmiaru, {hash_workers} wątk.") + digests = parallel_hash(need_hash, hash_workers, label="hash źródła") + STATE.save() + + seen = set() + unique = [] + dup_internal = 0 + dup_existing = 0 + hashed_src = 0 + + for p, rel, size in kept: + needs = size_count[size] > 1 or size in existing_sizes + if needs: + hashed_src += 1 + d = digests.get(p) + key = (size, d) if d is not None else None # nieczytelny -> unikat, nie gub + else: + key = (size, None) # unikalny rozmiar i brak w istniejących = na pewno unikat + + if key is not None and key in existing_index: + dup_existing += 1 + continue + if key is not None and key in seen: + dup_internal += 1 + continue + if key is not None: + seen.add(key) + unique.append((p, rel, size)) + + return unique, dup_internal, dup_existing, hashed_src + + +def copy_file_with_progress(src, dst, chunk_size=COPY_CHUNK): + """Jak shutil.copy2, ale chunkami z raportowaniem postępu bieżącego pliku.""" + with open(src, "rb") as fsrc, open(dst, "wb") as fdst: + for block in iter(lambda: fsrc.read(chunk_size), b""): + fdst.write(block) + PROGRESS.advance(src, len(block)) + shutil.copystat(src, dst, follow_symlinks=False) + + +def do_copy(unique, dest, skip_existing, workers): + PROGRESS.set_phase("kopiowanie", + items_total=len(unique), + bytes_total=sum(f[2] for f in unique), + detail=f"{workers} wątk.") + n = len(unique) + state_lock = threading.Lock() + stats = {"copied": 0, "copied_bytes": 0, "skipped": 0, "done": 0} + + def job(f): + p, rel, size = f + # 1) rejestr stanu — bez dotykania NFS + if STATE.is_copied(rel, size): + with state_lock: + stats["skipped"] += 1 + stats["done"] += 1 + PROGRESS.tick(add_bytes=size) + return + dpath = os.path.join(dest, rel) + os.makedirs(os.path.dirname(dpath), exist_ok=True) + # 2) klasyczny skip-existing (stat na celu) + if skip_existing and os.path.exists(dpath): + try: + if os.path.getsize(dpath) == size: + STATE.put_copied(rel, size) + with state_lock: + stats["skipped"] += 1 + stats["done"] += 1 + PROGRESS.tick(add_bytes=size) + return + except OSError: + pass + PROGRESS.start_file(p, size, "kopiuję") + try: + copy_file_with_progress(p, dpath) + STATE.put_copied(rel, size) + with state_lock: + stats["copied"] += 1 + stats["copied_bytes"] += size + except OSError as e: + print(f" BŁĄD kopiowania {p}: {e}", file=sys.stderr) + PROGRESS.finish_file(p) + with state_lock: + stats["done"] += 1 + if stats["done"] % 200 == 0: + print(f" ... kopiuję {stats['done']}/{n} ({human(stats['copied_bytes'])})", + file=sys.stderr) + + if workers <= 1: + for f in unique: + job(f) + else: + with ThreadPoolExecutor(max_workers=workers) as ex: + list(ex.map(job, unique)) + + STATE.save() + return stats["copied"], stats["copied_bytes"], stats["skipped"] + + +def install_signal_handlers(): + """SIGINT/SIGTERM: zapisz stan i wyjdź. Wątki-workery są daemonami + ThreadPoolExecutora tylko pośrednio, więc kończymy twardo — plik w połowie + kopiowany zostanie dokończony przy następnym biegu (zły rozmiar w celu).""" + def handler(signum, _frame): + print(f"\nPrzerwano sygnałem {signal.Signals(signum).name} — zapisuję stan...", + file=sys.stderr) + STATE.save(force=True) + print("Stan zapisany. Uruchom ponownie z tym samym --state, aby wznowić.", + file=sys.stderr) + os._exit(130) + signal.signal(signal.SIGINT, handler) + signal.signal(signal.SIGTERM, handler) + + +def main(): + global STATE + ap = argparse.ArgumentParser( + description="Deduplikujące kopiowanie na NAS z filtrem, dedupem względem istniejących danych, plikiem stanu i sprawdzeniem miejsca.") + ap.add_argument("--source", required=True, help="Katalog źródłowy (dysk wpięty w host).") + ap.add_argument("--dest", required=True, help="Katalog docelowy (np. /mnt/pve/truenas-nfs/import).") + ap.add_argument("--against", action="append", default=[], + help="Katalog z ISTNIEJĄCYMI danymi (np. cały mount NAS). Treść już tam obecna nie jest kopiowana. Można podać wiele razy.") + ap.add_argument("--min-size", type=float, default=1.0, + help="Próg w MiB; pliki mniejsze pomijane (domyślnie 1). Wyjątki: --keep-exts.") + ap.add_argument("--keep-exts", default="jpg,jpeg,png,gif", + help="Rozszerzenia kopiowane niezależnie od rozmiaru (po przecinku).") + ap.add_argument("--reserve", type=float, default=1.0, + help="Margines wolnego miejsca w GiB do zostawienia na NAS (domyślnie 1).") + ap.add_argument("--workers", type=int, default=4, + help="Wątki kopiowania i hashowania --against (NFS). Domyślnie 4.") + ap.add_argument("--hash-workers", type=int, default=1, + help="Wątki hashowania ŹRÓDŁA. Domyślnie 1. Zwiększ tylko dla SSD; na HDD więcej = wolniej (seeki).") + ap.add_argument("--state", default=None, metavar="PLIK.json", + help="Plik stanu JSON: cache hashy + rejestr skopiowanych. Przerwany bieg wznawia się bez powtarzania pracy.") + ap.add_argument("--state-interval", type=float, default=30.0, + help="Co ile sekund autozapis pliku stanu (domyślnie 30).") + ap.add_argument("--dry-run", action="store_true", help="Tylko analiza + raport, bez kopiowania.") + ap.add_argument("--yes", action="store_true", help="Nie pytaj o potwierdzenie.") + ap.add_argument("--no-skip-existing", action="store_true", + help="Nie pomijaj plików już obecnych w celu po ścieżce (domyślnie pomija = wznawianie).") + ap.add_argument("--heartbeat", type=float, default=120.0, + help="Co ile sekund wypisywać znak życia na stderr (domyślnie 120; 0 = wyłącz).") + args = ap.parse_args() + + source = os.path.abspath(args.source) + dest = os.path.abspath(args.dest) + min_size = int(args.min_size * MIB) + keep_exts = {("." + e.strip().lower().lstrip(".")) for e in args.keep_exts.split(",") if e.strip()} + reserve = int(args.reserve * (1 << 30)) + skip_existing = not args.no_skip_existing + workers = max(1, args.workers) + hash_workers = max(1, args.hash_workers) + + if not os.path.isdir(source): + sys.exit(f"BŁĄD: źródło nie istnieje lub nie jest katalogiem: {source}") + os.makedirs(dest, exist_ok=True) + + STATE = State(os.path.abspath(args.state) if args.state else None) + install_signal_handlers() + + print(f"Źródło : {source}") + print(f"Cel : {dest}") + if args.against: + print(f"Dedup vs : {', '.join(os.path.abspath(a) for a in args.against)}") + print(f"Filtr : pomijaj < {human(min_size)} poza {sorted(keep_exts)}") + print(f"Hash : {HASH_NAME}") + print(f"Wątki : kopiowanie/against={workers}, hash źródła={hash_workers}") + print(f"Stan : {STATE.path or 'brak (bez wznawiania)'}") + if args.heartbeat > 0: + print(f"Puls : co {args.heartbeat:.0f} s na stderr\n") + else: + print("Puls : wyłączony\n") + + stop_hb = threading.Event() + if args.heartbeat > 0: + threading.Thread(target=heartbeat_loop, args=(args.heartbeat, stop_hb), daemon=True).start() + if STATE.path: + threading.Thread(target=state_saver_loop, + args=(max(5.0, args.state_interval), stop_hb), daemon=True).start() + + try: + print("[1/4] Skanuję źródło...", file=sys.stderr) + PROGRESS.set_phase("skanowanie źródła", detail=source) + all_files, scan_errors = scan(source) + total_scanned = len(all_files) + total_scanned_bytes = sum(f[2] for f in all_files) + + print("[2/4] Filtruję...", file=sys.stderr) + kept, skipped_small, skipped_bytes = apply_filter(all_files, min_size, keep_exts) + source_sizes = set(f[2] for f in kept) + + existing_index = set() + existing_sizes = set() + existing_total = 0 + if args.against: + print("[3/4] Skanuję istniejące dane i buduję index treści...", file=sys.stderr) + existing_files = [] + for d in args.against: + PROGRESS.set_phase("skanowanie --against", detail=os.path.abspath(d)) + ef, _err = scan(os.path.abspath(d)) + existing_files.extend(ef) + existing_total = len(existing_files) + existing_sizes = set(f[2] for f in existing_files) + existing_index = build_existing_index(existing_files, source_sizes, workers) + else: + print("[3/4] (pominięto — brak --against)", file=sys.stderr) + + print("[4/4] Deduplikuję źródło i odsiewam treści już obecne...", file=sys.stderr) + unique, dup_internal, dup_existing, hashed_src = dedup( + kept, existing_index, existing_sizes, hash_workers) + + PROGRESS.set_phase("raport / oczekiwanie") + + to_copy_bytes = sum(f[2] for f in unique) + usage = shutil.disk_usage(dest) + fits = (to_copy_bytes + reserve) <= usage.free + + # --- RAPORT --- + print("\n================ RAPORT ================") + print(f"Plików w źródle : {total_scanned:,} ({human(total_scanned_bytes)})") + print(f"Pominięte filtrem (<{human(min_size)}) : {skipped_small:,} ({human(skipped_bytes)})") + print(f"Po filtrze : {len(kept):,}") + if args.against: + print(f"Istniejące dane (skan) : {existing_total:,} plików") + print(f"Odsiane (już na NAS) : {dup_existing:,}") + print(f"Duplikaty w źródle : {dup_internal:,}") + print(f"Zahashowano plików źródła : {hashed_src:,}") + if STATE.path: + print(f"Hashe z cache stanu : {STATE.hash_hits:,}") + print(f"UNIKALNE do skopiowania : {len(unique):,} ({human(to_copy_bytes)})") + if scan_errors: + print(f"Błędy odczytu przy skanie : {scan_errors}") + print("----------------------------------------") + print(f"Wolne miejsce na celu : {human(usage.free)}") + print(f"Potrzeba (+margines {human(reserve)}): {human(to_copy_bytes + reserve)}") + print(f"MIEŚCI SIĘ : {'TAK' if fits else 'NIE'}") + print("========================================\n") + + if not fits: + STATE.save(force=True) + sys.exit("PRZERWANO: za mało miejsca na woluminie docelowym. " + "Zwolnij miejsce, zmień --dest, albo zaostrz filtr (--min-size).") + + if args.dry_run: + STATE.save(force=True) + print("Dry-run: nic nie skopiowano (hashe trafiły do pliku stanu).") + return + + if not args.yes: + ans = input(f"Skopiować {len(unique):,} plików ({human(to_copy_bytes)}) do {dest}? [t/N] ").strip().lower() + if ans not in ("t", "tak", "y", "yes"): + STATE.save(force=True) + print("Anulowano (stan zapisany).") + return + + print("\nKopiuję...", file=sys.stderr) + copied, copied_bytes, skipped_ex = do_copy(unique, dest, skip_existing, workers) + + print("\n================ KONIEC ================") + print(f"Skopiowano : {copied:,} plików ({human(copied_bytes)})") + if skip_existing: + print(f"Pominięto (były) : {skipped_ex:,}") + if STATE.path: + print(f" w tym z rejestru stanu: {STATE.copy_hits:,}") + print("========================================") + finally: + stop_hb.set() + STATE.save(force=True) + + +if __name__ == "__main__": + main() \ No newline at end of file