Assetmanager/app/backup.py

from __future__ import annotations

import json
import os
import shutil
import subprocess
import tempfile
import threading
import zipfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

from .version import APP_VERSION

BACKUP_DIR = Path(os.getenv("BACKUP_DIR", "/data/backups"))
CONFIG_PATH = Path(os.getenv("APP_CONFIG", "/app/config/config.json"))
APPINFO_PATH = Path(os.getenv("APPINFO_PATH", "/app/config/APPINFO.json"))
UPLOAD_DIR = Path(os.getenv("UPLOAD_DIR", "/app/app/static/uploads"))
BACKUP_INTERVAL_HOURS = max(1, int(os.getenv("BACKUP_INTERVAL_HOURS", "8")))
BACKUP_RETENTION_DAYS = max(1, int(os.getenv("BACKUP_RETENTION_DAYS", "3")))
BACKUP_PREFIX = "assetmanager-backup-"
BACKUP_SUFFIX = ".zip"
_LOCK = threading.Lock()


def _database_url() -> str:
    value = os.getenv("DATABASE_URL", "")
    if value.startswith("postgresql+psycopg://"):
        value = "postgresql://" + value[len("postgresql+psycopg://"):]
    elif value.startswith("postgres+psycopg://"):
        value = "postgresql://" + value[len("postgres+psycopg://"):]
    return value


def _safe_database_url() -> str:
    value = _database_url()
    try:
        parsed = urlsplit(value)
        host = parsed.hostname or ""
        port = f":{parsed.port}" if parsed.port else ""
        username = parsed.username or ""
        auth = f"{username}@" if username else ""
        return urlunsplit((parsed.scheme, f"{auth}{host}{port}", parsed.path, "", ""))
    except Exception:
        return "postgresql://"


def ensure_backup_dir() -> Path:
    BACKUP_DIR.mkdir(parents=True, exist_ok=True)
    return BACKUP_DIR




def _is_mount_point(path: Path) -> bool:
    """Best-effort check whether path is a distinct container mount."""
    try:
        resolved = path.resolve()
        if resolved.is_mount():
            return True
        mountinfo = Path("/proc/self/mountinfo")
        if mountinfo.is_file():
            target = str(resolved)
            for line in mountinfo.read_text(encoding="utf-8", errors="replace").splitlines():
                parts = line.split()
                if len(parts) > 4 and parts[4].replace("\\040", " ") == target:
                    return True
    except OSError:
        pass
    return False


def _directory_status(path: Path, create: bool = False) -> dict:
    error = ""
    writable = False
    exists = path.exists()
    try:
        if create:
            path.mkdir(parents=True, exist_ok=True)
            exists = True
        if exists and path.is_dir():
            with tempfile.NamedTemporaryFile(prefix=".assetmanager-write-test-", dir=path, delete=True):
                writable = True
    except Exception as exc:
        error = f"{type(exc).__name__}: {exc}"
    return {
        "path": str(path),
        "exists": exists,
        "writable": writable,
        "is_mount": _is_mount_point(path) if exists else False,
        "error": error,
    }


def system_storage_information(log_dir: Path | None = None) -> dict:
    backups = list_backups()
    total_size = sum(int(item.get("size") or 0) for item in backups)
    latest = backups[0] if backups else None
    return {
        "database_connected": True,
        "config": _directory_status(CONFIG_PATH.parent),
        "uploads": _directory_status(UPLOAD_DIR),
        "logs": _directory_status(log_dir or Path(os.getenv("LOG_DIR", "/app/data/logs"))),
        "backups": _directory_status(BACKUP_DIR, create=True),
        "backup_count": len(backups),
        "backup_total_size": total_size,
        "latest_backup": latest,
    }

def validate_backup_name(name: str) -> str:
    candidate = Path(name).name
    if candidate != name or not candidate.startswith(BACKUP_PREFIX) or not candidate.endswith(BACKUP_SUFFIX):
        raise ValueError("invalid backup name")
    return candidate


def backup_path(name: str) -> Path:
    return ensure_backup_dir() / validate_backup_name(name)


def _run(command: list[str]) -> None:
    process = subprocess.run(command, capture_output=True, text=True)
    if process.returncode != 0:
        detail = (process.stderr or process.stdout or "command failed").strip()
        raise RuntimeError(detail[-2000:])


def create_backup(created_by: str = "system", reason: str = "manual") -> dict:
    with _LOCK:
        directory = ensure_backup_dir()
        now = datetime.now(timezone.utc)
        timestamp = now.strftime("%Y%m%d-%H%M%S")
        final_path = directory / f"{BACKUP_PREFIX}{timestamp}{BACKUP_SUFFIX}"
        sequence = 1
        while final_path.exists():
            final_path = directory / f"{BACKUP_PREFIX}{timestamp}-{sequence}{BACKUP_SUFFIX}"
            sequence += 1

        with tempfile.TemporaryDirectory(prefix="assetmanager-backup-") as temp_name:
            temp = Path(temp_name)
            dump_path = temp / "database.dump"
            database_url = _database_url()
            if not database_url:
                raise RuntimeError("DATABASE_URL is not configured")
            _run(["pg_dump", "--format=custom", "--no-owner", "--no-privileges", "--file", str(dump_path), database_url])

            included: list[str] = ["database.dump"]
            files_dir = temp / "files"
            files_dir.mkdir()
            if CONFIG_PATH.is_file():
                shutil.copy2(CONFIG_PATH, files_dir / "config.json")
                included.append("files/config.json")
            if APPINFO_PATH.is_file():
                shutil.copy2(APPINFO_PATH, files_dir / "APPINFO.json")
                included.append("files/APPINFO.json")
            if UPLOAD_DIR.is_dir():
                target = files_dir / "uploads"
                shutil.copytree(UPLOAD_DIR, target)
                included.append("files/uploads/")

            metadata = {
                "format": 1,
                "application": "AssetManager",
                "version": _read_version(),
                "created_at": now.isoformat(),
                "created_by": created_by or "system",
                "reason": reason,
                "database": _safe_database_url(),
                "included": included,
            }
            (temp / "metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
            (temp / "VERSION").write_text(metadata["version"] + "\n", encoding="utf-8")
            included.append("VERSION")

            temporary_zip = final_path.with_suffix(".zip.part")
            try:
                with zipfile.ZipFile(temporary_zip, "w", zipfile.ZIP_DEFLATED, allowZip64=True) as archive:
                    for path in sorted(temp.rglob("*")):
                        if path.is_file():
                            archive.write(path, path.relative_to(temp))
                temporary_zip.replace(final_path)
            finally:
                temporary_zip.unlink(missing_ok=True)

        cleanup_backups()
        return describe_backup(final_path)


def _read_version() -> str:
    for candidate in (Path("/app/VERSION"), Path(__file__).resolve().parent.parent / "VERSION", Path(__file__).resolve().parent / "VERSION"):
        try:
            value = candidate.read_text(encoding="utf-8").strip()
            if value and value.casefold() != "unknown":
                return value
        except OSError:
            pass
    return APP_VERSION


def _archive_version(archive: zipfile.ZipFile, metadata: dict) -> str:
    value = str(metadata.get("version") or "").strip()
    if value and value.casefold() != "unknown":
        return value
    names = set(archive.namelist())
    for name in ("VERSION", "files/VERSION", "assetmanager/VERSION"):
        if name in names:
            try:
                value = archive.read(name).decode("utf-8", errors="replace").strip()
                if value and value.casefold() != "unknown":
                    return value
            except Exception:
                pass
    for name in ("files/APPINFO.json", "APPINFO.json"):
        if name in names:
            try:
                value = str(json.loads(archive.read(name).decode("utf-8")).get("version") or "").strip()
                if value and value.casefold() != "unknown":
                    return value
            except Exception:
                pass
    return ""


def _metadata(path: Path) -> dict:
    try:
        with zipfile.ZipFile(path) as archive:
            metadata = json.loads(archive.read("metadata.json").decode("utf-8"))
            if not isinstance(metadata, dict):
                return {}
            metadata = dict(metadata)
            metadata["version"] = _archive_version(archive, metadata)
            return metadata
    except Exception:
        return {}


def describe_backup(path: Path) -> dict:
    stat = path.stat()
    metadata = _metadata(path)
    return {
        "name": path.name,
        "size": stat.st_size,
        "modified_at": datetime.fromtimestamp(stat.st_mtime, timezone.utc),
        "created_at": metadata.get("created_at", ""),
        "created_by": metadata.get("created_by", ""),
        "reason": metadata.get("reason", ""),
        "version": metadata.get("version", ""),
        "valid": bool(metadata),
    }


def list_backups() -> list[dict]:
    directory = ensure_backup_dir()
    items = [describe_backup(path) for path in directory.glob(f"{BACKUP_PREFIX}*{BACKUP_SUFFIX}") if path.is_file()]
    return sorted(items, key=lambda item: item["modified_at"], reverse=True)


def cleanup_backups() -> int:
    threshold = datetime.now(timezone.utc) - timedelta(days=BACKUP_RETENTION_DAYS)
    removed = 0
    for item in list_backups():
        if item["modified_at"] < threshold:
            backup_path(item["name"]).unlink(missing_ok=True)
            removed += 1
    return removed


def delete_backup(name: str) -> None:
    backup_path(name).unlink()


def store_uploaded_backup(filename: str, content: bytes) -> dict:
    ensure_backup_dir()
    if len(content) < 100:
        raise ValueError("backup file is empty")
    with tempfile.NamedTemporaryFile(prefix="backup-upload-", suffix=".zip", delete=False, dir=BACKUP_DIR) as handle:
        temp_path = Path(handle.name)
        handle.write(content)
    try:
        with zipfile.ZipFile(temp_path) as archive:
            names = set(archive.namelist())
            if "metadata.json" not in names or "database.dump" not in names:
                raise ValueError("unsupported backup archive")
            metadata = json.loads(archive.read("metadata.json").decode("utf-8"))
            if metadata.get("application") != "AssetManager":
                raise ValueError("unsupported backup archive")
        now = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
        target = BACKUP_DIR / f"{BACKUP_PREFIX}{now}-upload{BACKUP_SUFFIX}"
        counter = 1
        while target.exists():
            target = BACKUP_DIR / f"{BACKUP_PREFIX}{now}-upload-{counter}{BACKUP_SUFFIX}"
            counter += 1
        temp_path.replace(target)
        return describe_backup(target)
    finally:
        temp_path.unlink(missing_ok=True)


def restore_backup(name: str) -> dict:
    path = backup_path(name)
    with _LOCK, tempfile.TemporaryDirectory(prefix="assetmanager-restore-") as temp_name:
        temp = Path(temp_name)
        with zipfile.ZipFile(path) as archive:
            archive.extractall(temp)
        dump_path = temp / "database.dump"
        if not dump_path.is_file():
            raise ValueError("database dump missing")
        database_url = _database_url()
        _run(["pg_restore", "--clean", "--if-exists", "--no-owner", "--no-privileges", "--dbname", database_url, str(dump_path)])

        files = temp / "files"
        if (files / "config.json").is_file():
            CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(files / "config.json", CONFIG_PATH)
        if (files / "APPINFO.json").is_file():
            APPINFO_PATH.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(files / "APPINFO.json", APPINFO_PATH)
        if (files / "uploads").is_dir():
            UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
            for child in UPLOAD_DIR.iterdir():
                if child.is_dir():
                    shutil.rmtree(child)
                else:
                    child.unlink()
            shutil.copytree(files / "uploads", UPLOAD_DIR, dirs_exist_ok=True)
        return _metadata(path)


def automatic_backup_loop(stop_event: threading.Event) -> None:
    # Wait after application start; do not create a backup during every restart.
    while not stop_event.wait(BACKUP_INTERVAL_HOURS * 3600):
        try:
            create_backup(created_by="system", reason="automatic")
        except Exception:
            # Main application logger records route errors; scheduler must stay alive.
            pass