Копирование файлов с повреждённых флешек

Исходный код (Python)

Ниже приведён полный исходный код утилиты. Код открыт и может быть адаптирован под конкретные задачи.

import os
import json
import time
from pathlib import Path

CONFIG_FILE = "config.json"
ERRORS_FILE = "errors.json"
RETRY_LOG = "retry_log.txt"


def load_config():
    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
        return json.load(f)


def load_errors():
    if not os.path.exists(ERRORS_FILE):
        return set()
    try:
        with open(ERRORS_FILE, "r", encoding="utf-8") as f:
            data = f.read().strip()
            if not data:  # пустой файл
                return set()
            return set(json.loads(data))
    except Exception as e:
        print(f"[ОШИБКА] Не удалось загрузить errors.json: {e}")
        return set()


def log(message):
    print(message)
    with open(RETRY_LOG, "a", encoding="utf-8") as f:
        f.write(message + "\n")


def safe_exists(path_str: str) -> bool:
    """Безопасная проверка существования файла/папки"""
    try:
        return os.path.exists(path_str)
    except OSError as e:
        log(f"[ПРЕДУПРЕЖДЕНИЕ] Не удалось проверить {path_str}: {e}")
        return False


def safe_copy(src, dst, chunk_size=4096, delay=0.01, retries=3):
    """
    src       – путь к исходному файлу
    dst       – путь назначения
    chunk_size – размер блока (по умолчанию 4КБ)
    delay     – задержка между блоками
    retries   – количество повторных попыток
    """
    os.makedirs(os.path.dirname(dst), exist_ok=True)

    for attempt in range(1, retries + 1):
        try:
            with open(src, "rb") as fsrc, open(dst, "wb") as fdst:
                while True:
                    chunk = fsrc.read(chunk_size)
                    if not chunk:
                        break
                    fdst.write(chunk)
                    time.sleep(delay)  # замедление для стабильности

            log(f"[УСПЕХ] {src} → {dst} (попытка {attempt})")
            return True
        except Exception as e:
            log(f"[ОШИБКА] {src} (попытка {attempt}): {e}")
            time.sleep(2)  # пауза перед повтором
    return False


def retry_errors():
    config = load_config()
    errors = load_errors()

    if not errors:
        log("Нет файлов для повторного копирования.")
        return

    source = Path(config["source_drive"])
    target = Path(config["target_drive"])

    still_bad = []

    for file_path in errors:
        src = Path(file_path)

        if not safe_exists(str(src)):
            log(f"[Пропуск] {src} не найден или недоступен")
            continue

        try:
            rel_path = src.relative_to(source)
        except ValueError:
            log(f"[ПРЕДУПРЕЖДЕНИЕ] {src} не относится к исходной папке {source}")
            still_bad.append(str(src))
            continue

        dst = target / rel_path

        if not safe_copy(str(src), str(dst)):
            log(f"[НЕ УДАЛОСЬ] {src}")
            still_bad.append(str(src))

    # обновляем errors.json (оставляем только реально неудачные)
    with open(ERRORS_FILE, "w", encoding="utf-8") as f:
        json.dump(still_bad, f, ensure_ascii=False, indent=2)

    log("=== Повторная попытка завершена ===")


if __name__ == "__main__":
    retry_errors()