scripts/backup.py

"""Consistent SQLite snapshot + immutable files; PostgreSQL dump support.

Restore always targets a new directory, never the running database.
"""

import argparse
import hashlib
import json
import os
import re
import shutil
import sqlite3
import subprocess
import sys
import tempfile
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit, unquote

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "backend"))


def backup(output):
    from app.config import DATABASE_URL
    from app.database import SessionLocal
    from app.models import File, TrashFile
    from app.storage import read
    from sqlalchemy import select

    output = Path(output).resolve()
    output.parent.mkdir(parents=True, exist_ok=True)
    if output.exists():
        raise ValueError("Escolha um arquivo de backup que ainda não existe.")
    with tempfile.TemporaryDirectory() as temp:
        work = Path(temp)
        manifest = {
            "format": "carla-backup-v1",
            "created_at": datetime.now(timezone.utc).isoformat(),
            "files": [],
        }
        if DATABASE_URL.startswith("sqlite:///"):
            source = sqlite3.connect(DATABASE_URL.removeprefix("sqlite:///"))
            snapshot = sqlite3.connect(work / "database.sqlite3")
            source.backup(snapshot)
            source.close()
            manifest["database"] = "database.sqlite3"
            manifest["provider"] = "sqlite"
            rows = snapshot.execute("SELECT key,checksum FROM files").fetchall()
            if snapshot.execute(
                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='trash_files'"
            ).fetchone():
                rows += snapshot.execute(
                    "SELECT key,checksum FROM trash_files"
                ).fetchall()
            rows = list(dict(rows).items())
            snapshot.close()
        else:
            if not shutil.which("pg_dump"):
                raise RuntimeError(
                    "PostgreSQL requer pg_dump instalado para exportar o banco."
                )
            # Prevent cleanup of unlinked objects during the dump. Stop worker for PG backups.
            parsed = urlsplit(
                DATABASE_URL.replace("postgresql+psycopg://", "postgresql://")
            )
            env = os.environ.copy()
            env["PGPASSWORD"] = unquote(parsed.password or "")
            args = [
                "pg_dump",
                "--no-owner",
                "--no-privileges",
                "-h",
                parsed.hostname or "localhost",
                "-p",
                str(parsed.port or 5432),
                "-U",
                unquote(parsed.username or ""),
                "-d",
                parsed.path.lstrip("/"),
                "-f",
                str(work / "database.sql"),
            ]
            if "sslmode=" in parsed.query:
                env["PGSSLMODE"] = dict(
                    part.split("=", 1) for part in parsed.query.split("&")
                ).get("sslmode", "require")
            subprocess.run(args, env=env, check=True)
            manifest["database"] = "database.sql"
            manifest["provider"] = "postgresql"
            with SessionLocal() as db:
                rows = list(
                    dict(
                        [(f.key, f.checksum) for f in db.scalars(select(File))]
                        + [(f.key, f.checksum) for f in db.scalars(select(TrashFile))]
                    ).items()
                )
        with zipfile.ZipFile(output, "x", compression=zipfile.ZIP_DEFLATED) as archive:
            archive.write(work / manifest["database"], manifest["database"])
            for key, checksum in rows:
                data = read(key)
                if hashlib.sha256(data).hexdigest() != checksum:
                    raise RuntimeError(f"Integridade inválida: {key}")
                archive.writestr("files/" + key, data)
                manifest["files"].append({"key": key, "checksum": checksum})
            manifest["database_checksum"] = hashlib.sha256(
                (work / manifest["database"]).read_bytes()
            ).hexdigest()
            archive.writestr(
                "manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2)
            )
    return manifest


def verify(archive_path):
    with zipfile.ZipFile(archive_path) as archive:
        manifest = json.loads(archive.read("manifest.json"))
        if manifest.get("format") != "carla-backup-v1":
            raise ValueError("Formato de backup desconhecido.")
        if manifest["database"] not in {"database.sqlite3", "database.sql"}:
            raise ValueError("Banco inválido.")
        if (
            hashlib.sha256(archive.read(manifest["database"])).hexdigest()
            != manifest["database_checksum"]
        ):
            raise ValueError("Checksum do banco inválido.")
        for file in manifest["files"]:
            if not re.fullmatch(r"[a-f0-9-]{36}\.[a-z0-9]+", file["key"]):
                raise ValueError("Chave de arquivo inválida.")
            if (
                hashlib.sha256(archive.read("files/" + file["key"])).hexdigest()
                != file["checksum"]
            ):
                raise ValueError("Checksum de arquivo inválido.")
    return manifest


def restore(archive_path, destination):
    manifest = verify(archive_path)
    destination = Path(destination).resolve()
    if destination.exists():
        raise ValueError(
            "A restauração requer uma pasta nova, para preservar os dados existentes."
        )
    destination.mkdir(parents=True)
    with zipfile.ZipFile(archive_path) as archive:
        (destination / manifest["database"]).write_bytes(
            archive.read(manifest["database"])
        )
        (destination / "files").mkdir()
        for file in manifest["files"]:
            (destination / "files" / file["key"]).write_bytes(
                archive.read("files/" + file["key"])
            )
        (destination / "manifest.json").write_text(
            json.dumps(manifest, ensure_ascii=False, indent=2)
        )
    if manifest["provider"] == "sqlite":
        with sqlite3.connect(destination / "database.sqlite3") as db:
            assert db.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
            assert db.execute("PRAGMA foreign_key_check").fetchall() == []
            references = db.execute("SELECT COUNT(*) FROM revision_files").fetchone()[0]
        print(f"Restauração verificada: {references} vínculos de arquivos preservados.")
    else:
        print(
            "Dump PostgreSQL restaurado na pasta; importe database.sql em um banco NOVO usando psql -v ON_ERROR_STOP=1."
        )
    return manifest


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Backup, verificação e recuperação da CARLA"
    )
    parser.add_argument("action", choices=["create", "verify", "restore"])
    parser.add_argument("path", help="Caminho do arquivo ZIP")
    parser.add_argument("--destination", help="Pasta nova para restauração")
    args = parser.parse_args()
    try:
        if args.action == "create":
            result = backup(args.path)
        elif args.action == "verify":
            result = verify(args.path)
        else:
            if not args.destination:
                parser.error("restore requer --destination")
            result = restore(args.path, args.destination)
        print(
            f'Concluído: {len(result["files"])} arquivos íntegros; formato {result["provider"]}.'
        )
    except Exception as exc:
        print(f"Falha: {exc}", file=sys.stderr)
        sys.exit(1)