diff --git a/nas-dedup.copy.py b/nas-dedup.copy.py deleted file mode 100644 index c25f56b..0000000 --- a/nas-dedup.copy.py +++ /dev/null @@ -1,779 +0,0 @@ -#!/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 - - -# ---- rejestr pominięć (thread-safe): kategoria -> lista (ścieżka, szczegół) ---- -_SKIP_LOCK = threading.Lock() -_SKIPPED = defaultdict(list) - -def add_skip(category, path, detail=""): - with _SKIP_LOCK: - _SKIPPED[category].append((path, detail)) - -def print_skipped(skipped_list_path=None): - """Wypisuje raport pominięć pogrupowany po kategorii. - Bez skipped_list_path: pełna lista na stdout. - Z skipped_list_path: pełna lista do pliku, na stdout tylko liczniki per kategoria.""" - with _SKIP_LOCK: - if not _SKIPPED: - return - total = sum(len(v) for v in _SKIPPED.values()) - print(f"\n============ POMINIĘTE PLIKI ({total:,}) ============") - if skipped_list_path: - try: - with open(skipped_list_path, "w", encoding="utf-8") as fh: - for cat in sorted(_SKIPPED): - items = _SKIPPED[cat] - fh.write(f"# {cat} ({len(items)})\n") - for path, detail in items: - fh.write(f"{path}\t{detail}\n" if detail else f"{path}\n") - fh.write("\n") - for cat in sorted(_SKIPPED): - print(f" {cat:<28}: {len(_SKIPPED[cat]):,}") - print(f" (pełna lista zapisana do: {skipped_list_path})") - except OSError as e: - print(f" BŁĄD zapisu {skipped_list_path}: {e} — wypisuję na stdout:") - skipped_list_path = None - if not skipped_list_path: - for cat in sorted(_SKIPPED): - items = _SKIPPED[cat] - print(f"\n {cat} ({len(items)}):") - for path, detail in items: - print(f" {path}" + (f" [{detail}]" if detail else "")) - print("========================================") - - -def apply_filter(files, min_size, keep_exts, only_exts=None): - kept, skipped, skipped_bytes = [], 0, 0 - skipped_ext = 0 - for p, rel, size in files: - ext = os.path.splitext(p)[1].lower() - if only_exts is not None and ext not in only_exts: - skipped_ext += 1 - add_skip("poza listą --only-exts", p, f"rozszerzenie '{ext or '(brak)'}'") - continue - if size < min_size and ext not in keep_exts: - skipped += 1 - skipped_bytes += size - add_skip("filtr rozmiaru", p, f"{human(size)} < próg {human(min_size)}") - continue - kept.append((p, rel, size)) - return kept, skipped, skipped_bytes, skipped_ext - - -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 - add_skip("już na NAS (--against)", p, f"treść identyczna, {human(size)}") - continue - if key is not None and key in seen: - dup_internal += 1 - add_skip("duplikat w źródle", p, f"kopia wcześniejszego pliku, {human(size)}") - 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 - add_skip("już skopiowany (rejestr stanu)", p, f"rel={rel}") - 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 - add_skip("już w celu (ten sam rozmiar)", p, f"cel={dpath}") - 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("--only-exts", default=None, metavar="EXT,EXT", - help="Kopiuj WYŁĄCZNIE pliki z jednym z podanych rozszerzeń (po przecinku, np. 'mkv,mp4,iso'). Reszta pomijana PRZED hashowaniem. Filtr --min-size dalej obowiązuje.") - ap.add_argument("--keep-exts", default="jpg,jpeg,png,gif", - help="Rozszerzenia kopiowane niezależnie od rozmiaru (po przecinku).") - ap.add_argument("--skipped-list", default=None, metavar="PLIK", - help="Zapisz pełną listę pominiętych plików do pliku; na stdout tylko liczniki per kategoria. Bez tej opcji pełna lista idzie na stdout.") - 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()} - only_exts = None - if args.only_exts: - only_exts = {("." + e.strip().lower().lstrip(".")) for e in args.only_exts.split(",") if e.strip()} - if not only_exts: - sys.exit("BŁĄD: --only-exts podane, ale lista rozszerzeń pusta.") - 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)}") - if only_exts: - print(f"Tylko ext : {sorted(only_exts)}") - 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, skipped_ext = apply_filter( - all_files, min_size, keep_exts, only_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)})") - if only_exts: - print(f"Pominięte (--only-exts) : {skipped_ext:,}") - 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).") - print_skipped(args.skipped_list) - 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("========================================") - print_skipped(args.skipped_list) - finally: - stop_hb.set() - STATE.save(force=True) - - -if __name__ == "__main__": - main() \ No newline at end of file