pdf-quiz-generator/backend/scripts/migrate_uploads_to_s3.py

"""Copy the existing uploads volume into object storage.

Verification order matters: every file is copied and read back *before* anything
is deleted, and deletion only happens when explicitly asked for. A migration that
removes the source before proving the destination is readable has no way back.

    docker compose exec backend python -m scripts.migrate_uploads_to_s3
    docker compose exec backend python -m scripts.migrate_uploads_to_s3 --apply
    docker compose exec backend python -m scripts.migrate_uploads_to_s3 --verify
    docker compose exec backend python -m scripts.migrate_uploads_to_s3 --apply --delete-local
"""
import hashlib
import sys
from pathlib import Path

from app.config import settings
from app.services import storage_service


def walk(root: Path):
    """Every file under the uploads directory, keyed by its relative path."""
    for path in sorted(root.rglob("*")):
        if path.is_file() and not path.name.startswith("."):
            yield path.relative_to(root).as_posix(), path


def human(size: int) -> str:
    for unit in ("B", "KB", "MB", "GB"):
        if size < 1024 or unit == "GB":
            return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}"
        size /= 1024
    return f"{size:.1f} GB"


def main():
    apply_changes = "--apply" in sys.argv
    verify_only = "--verify" in sys.argv
    delete_local = "--delete-local" in sys.argv

    if not storage_service.using_s3():
        print("STORAGE_BACKEND is not 's3'; nothing to migrate into.")
        return 1

    root = Path(settings.UPLOAD_DIR)
    if not root.is_dir():
        print(f"No uploads directory at {root}")
        return 1

    storage_service.ensure_bucket()
    files = list(walk(root))
    total_bytes = sum(path.stat().st_size for _key, path in files)
    print(f"  files on the volume : {len(files)} ({human(total_bytes)})")

    copied, already, failed, verified, mismatched = 0, 0, [], 0, []
    for key, path in files:
        data = path.read_bytes()
        digest = hashlib.sha256(data).hexdigest()

        # Strictly S3: the fallback in `load` would compare a file to itself.
        if verify_only or storage_service.s3_object(key) is not None:
            stored = storage_service.s3_object(key)
            if stored is None:
                failed.append(key)
            elif hashlib.sha256(stored).hexdigest() != digest:
                mismatched.append(key)
            else:
                verified += 1
                already += 1
            continue

        if not apply_changes:
            copied += 1
            continue

        try:
            storage_service.save(key, data)
            # Read back before trusting it; a write that cannot be read is not a copy.
            stored = storage_service.s3_object(key)
            if stored is None or hashlib.sha256(stored).hexdigest() != digest:
                mismatched.append(key)
            else:
                copied += 1
                verified += 1
        except Exception as error:
            failed.append(f"{key}: {error}")

    print(f"  copied              : {copied}")
    print(f"  already in storage  : {already}")
    print(f"  read back and match : {verified}")
    if mismatched:
        print(f"  MISMATCHED          : {len(mismatched)}")
        for key in mismatched[:10]:
            print(f"    {key}")
    if failed:
        print(f"  FAILED              : {len(failed)}")
        for key in failed[:10]:
            print(f"    {key}")

    if delete_local:
        if mismatched or failed:
            print("\n  Refusing to delete: not every file verified.")
            return 1
        if not apply_changes:
            print("\n  --delete-local needs --apply.")
            return 1
        removed = 0
        for key, path in files:
            # Belt and braces: prove the object reads before removing the source.
            if storage_service.s3_object(key) is not None:
                path.unlink()
                removed += 1
        print(f"  local files removed : {removed}")

    if not apply_changes and not verify_only:
        print("\n  Re-run with --apply to copy, then --verify to check.")
    return 0


if __name__ == "__main__":
    sys.exit(main())