779 lines
32 KiB
Python
779 lines
32 KiB
Python
#!/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() |