#!/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()