rm-dups.py

Go to the documentation of this file.

import os
import hashlib
from datetime import datetime
 
 
# перевод в timestamp
def format_date(timestamp):
    """Преобразует timestamp в читаемую дату."""
    return datetime.fromtimestamp(timestamp).strftime('%d.%m.%Y %H:%M:%S')
 
 
# перевод в байты 
def format_size(size_bytes):
    for unit in ['Б', 'КБ', 'МБ', 'ГБ', 'ТБ']:
        if size_bytes < 1024:
            return f"{size_bytes:.2f} {unit}"
        size_bytes /= 1024
    return f"{size_bytes:.2f} ПБ"
 
 
# вычисляет хэш файла
def get_file_hash(filepath):
    """Вычисляет SHA-256 хэш файла по частям."""
    hasher = hashlib.sha256()
    try:
        with open(filepath, 'rb') as f:
            # Читаем по 1МБ, чтобы не забивать ОЗУ
            while chunk := f.read(1024*1024):
                hasher.update(chunk)
        return hasher.hexdigest()
    except (PermissionError, OSError) as e:
        print(f"[!] Ошибка доступа к файлу {filepath}: {e}")
        return None
 
 
def clean_database_duplicates(directory):
    if not os.path.isdir(directory):
        print('Указанная директория не найдена')
        return
    
    print(f"Сканирование директории: {directory}...")
 
    # Группировка по размеру (первичный фильтр)
    files_by_size = {}
    for root, _, filenames in os.walk(directory):
        for filename in filenames:
            path = os.path.join(root, filename)
            try:
                # ВАЖНО: проверка размера теперь внутри цикла по filenames
                size = os.path.getsize(path)
                files_by_size.setdefault(size, []).append(path)
            except OSError:
                continue
 
    # Хэшируем только те файлы, размер которых совпал
    files_by_hash = {}
    for size, paths in files_by_size.items():
        if len(paths) > 1:
            for path in paths:
                f_hash = get_file_hash(path)
                if f_hash:
                    try:
                        stat = os.stat(path)
                        file_info = {
                            'path': path,
                            'mtime': stat.st_mtime,
                            'size': stat.st_size
                        }                    
                        files_by_hash.setdefault(f_hash, []).append(file_info)
                    except OSError:
                        continue
    
    to_delete = []
 
    print("\n" + "="*80)
    print("АНАЛИЗ ДУБЛИКАТОВ")
    print("="*80)
 
    for f_hash, group in files_by_hash.items():
        if len(group) > 1:
            group.sort(key = lambda x: x['mtime'], reverse=True)
 
            keep_file = group[0]
            delete_files = group[1:]
 
            print(f'Оставляем самые свежие файлы : {keep_file["path"]}')
            print(f"    Дата: {format_date(keep_file['mtime'])} | Размер: {format_size(keep_file['size'])}")
 
            print(f"  [БУДУТ УДАЛЕНЫ]:")
            for duplicates in delete_files:
                print(f" -- {duplicates['path']}")
                print(f" Дата: ({format_date(duplicates['mtime'])}) | Размер: ({format_size(duplicates['size'])})")
                to_delete.append(duplicates['path'])
    if not to_delete:
        print("\n[+] Дубликаты не обнаружены. Все дампы уникальны.")
        return
 
    print("\n" + "="*80)
    print(f"ИТОГО: Найдено {len(to_delete)} дубликатов.")
    confirm = input("Удалить выбранные файлы? (да/нет): ").strip().lower()
 
    if confirm == 'да':
        for path in to_delete:
            try:
                os.remove(path)
                print(f"[УДАЛЕНО]: {path}")
            except Exception as e:
                print(f"[ОШИБКА]: Не удалось удалить {path}: {e}")
        print("\n[+] Очистка завершена успешно.")
    else:
        print("\n[!] Операция отменена пользователем.")
 
 
 
if __name__ == "__main__":
    target_dir = input("Введите путь к папке с дампами: ").strip()
    clean_database_duplicates(target_dir)