Spaces:
Sleeping
Sleeping
Guilherme Silberfarb Costa
Adiciona exclusão autorizada e lixeira privada com recuperação de trabalhos e rascunhos
ad3d3f0 | """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) | |