Dalam dunia digital, duplikat file adalah hal yang sering terjadi — dan tanpa disadari bisa memenuhi ruang penyimpanan dengan cepat. File yang sama bisa tersebar di berbagai direktori, dengan nama berbeda, tapi isi identik. Di sinilah Python hadir sebagai "penendang saudara kembar".
Dengan pendekatan yang efisien dan sederhana, script Python berikut ini dirancang untuk:
- Memindai seluruh file dalam beberapa direktori sumber
- Mengidentifikasi file duplikat berdasarkan hash MD5
- Memindahkan file duplikat ke satu direktori pusat
- Melog semua aktivitas ke file
CSV
Berikut adalah implementasi Python-nya menggunakan multithreading untuk mempercepat proses hashing file:
import os
import hashlib
import shutil
import csv
from concurrent.futures import ThreadPoolExecutor, as_completed
SOURCE_DIRS = ["/mnt/data", "/mnt/video", "/mnt/photo", "/mnt/home"]
TARGET_DIR = "/mnt/cloudy/rara"
LOG_FILE = os.path.join(TARGET_DIR, "duplikat_log.csv")
EXTENSIONS = {'.jpg', '.png', '.mp4', '.rar', '.zip', '.img', '.psd', '.cdr', '.iso', '.exe', '.apk'}
MAX_WORKERS = 8 # jumlah thread paralel
def ensure_target_dir():
os.makedirs(TARGET_DIR, exist_ok=True)
def file_hash(path, block_size=65536):
hasher = hashlib.md5()
try:
with open(path, 'rb') as f:
while chunk := f.read(block_size):
hasher.update(chunk)
return hasher.hexdigest()
except Exception as e:
print(f"Gagal hash: {path} | {e}")
return None
def is_valid_extension(filename):
return os.path.splitext(filename)[1].lower() in EXTENSIONS
def collect_files():
all_files = []
for root_dir in SOURCE_DIRS:
for root, _, files in os.walk(root_dir):
for file in files:
full_path = os.path.join(root, file)
if is_valid_extension(file):
all_files.append(full_path)
return all_files
def process_files():
ensure_target_dir()
seen_hashes = {}
all_files = collect_files()
print(f"Total file: {len(all_files)}")
with open(LOG_FILE, "w", newline='') as log_file:
writer = csv.writer(log_file)
writer.writerow(["original_path", "duplicate_path", "md5", "size"])
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
future_to_file = {executor.submit(file_hash, file): file for file in all_files}
for idx, future in enumerate(as_completed(future_to_file)):
file = future_to_file[future]
try:
hash_val = future.result()
if not hash_val:
continue
except Exception as e:
print(f"Error file: {file} | {e}")
continue
file_size = os.path.getsize(file)
if hash_val in seen_hashes:
original = seen_hashes[hash_val]
relative_path = os.path.relpath(file, '/')
target_path = os.path.join(TARGET_DIR, relative_path)
os.makedirs(os.path.dirname(target_path), exist_ok=True)
try:
shutil.move(file, target_path)
writer.writerow([original, file, hash_val, file_size])
print(f"Duplikat dipindahkan: {file}")
except Exception as e:
print(f"Gagal memindahkan: {file} | {e}")
else:
seen_hashes[hash_val] = file
if idx % 100 == 0:
print(f"Proses {idx}/{len(all_files)} file selesai")
print(f"Selesai. Log di: {LOG_FILE}")
if __name__ == "__main__":
process_files()
Kamu dapat menyesuaikan jumlah thread melalui variabel
MAX_WORKERS, dan mengubah SOURCE_DIRS sesuai struktur direktori kamu.
Dengan script ini, saudara kembar digital tak diundang akan segera terdeteksi dan dipindahkan. Simpel, aman, dan cepat. Python pun berhasil menjalankan tugasnya sebagai penegak "ketertiban file" di jagat penyimpananmu.
Tidak ada komentar:
Posting Komentar