bin/activate-gap.py

#!/usr/bin/env python3
"""Validiert und aktiviert ein GAP-ZIP atomar mit versioniertem Vorgängerarchiv."""
from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import re
import shutil
import tempfile
import zipfile
from pathlib import Path

VERSION_RE = re.compile(r"^[0-9]+(?:\.[0-9]+){1,3}(?:[-+][A-Za-z0-9._-]+)?$")


class GapActivationError(RuntimeError):
    pass


TIMESTAMPED_GAP_PREFIX_RE = re.compile(r"^[0-9]{8}-[0-9]{6}-")


def _safe_members(archive: zipfile.ZipFile) -> list[zipfile.ZipInfo]:
    members = archive.infolist()
    if not members:
        raise GapActivationError("Das GAP-ZIP ist leer.")
    for member in members:
        path = Path(member.filename)
        if path.is_absolute() or ".." in path.parts:
            raise GapActivationError(f"Unsicherer ZIP-Pfad: {member.filename}")
    return members


def inspect_gap(zip_path: Path) -> tuple[str, str]:
    if not zip_path.is_file():
        raise GapActivationError(f"GAP-ZIP nicht gefunden: {zip_path}")
    if not zipfile.is_zipfile(zip_path):
        raise GapActivationError(f"Keine gültige ZIP-Datei: {zip_path}")

    with zipfile.ZipFile(zip_path) as archive:
        members = _safe_members(archive)
        manifest_candidates = [m for m in members if Path(m.filename).name == "manifest.json"]
        if len(manifest_candidates) != 1:
            raise GapActivationError(
                "Das GAP-ZIP muss genau ein manifest.json enthalten; "
                f"gefunden: {len(manifest_candidates)}"
            )
        manifest_member = manifest_candidates[0]
        try:
            manifest = json.loads(archive.read(manifest_member).decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise GapActivationError(f"manifest.json ist ungültig: {exc}") from exc

    repository = manifest.get("repository")
    if not isinstance(repository, dict):
        raise GapActivationError("manifest.json enthält kein repository-Objekt.")
    repository_id = repository.get("id")
    version = repository.get("version")
    if not isinstance(repository_id, str) or not repository_id.strip():
        raise GapActivationError("manifest.json enthält keine Repository-ID.")
    if not isinstance(version, str) or not VERSION_RE.fullmatch(version):
        raise GapActivationError(f"Ungültige GAP-Version im Manifest: {version!r}")
    return repository_id, version


def resolve_candidate(requested: Path, processed_dir: Path) -> Path:
    """Resolve an existing candidate or exactly one importer-prefixed copy.

    The resolver deliberately never chooses the newest candidate. A caller who
    supplies a non-existing name may only be redirected to a single file named
    ``YYYYMMDD-HHMMSS-<requested-name>`` in the processed download directory.
    """
    expanded = requested.expanduser()
    if expanded.is_file():
        return expanded.resolve()

    processed = processed_dir.expanduser()
    basename = expanded.name
    candidates = sorted(
        path
        for path in processed.iterdir()
        if path.is_file()
        and TIMESTAMPED_GAP_PREFIX_RE.match(path.name)
        and path.name.endswith(f"-{basename}")
    ) if processed.is_dir() else []

    if not candidates:
        raise GapActivationError(
            f"GAP-ZIP nicht gefunden: {expanded}; "
            f"kein zeitgestempelter Kandidat in {processed}"
        )
    if len(candidates) > 1:
        listed = ", ".join(str(path) for path in candidates)
        raise GapActivationError(
            "GAP-ZIP ist mehrdeutig; bitte einen exakten Pfad angeben: " + listed
        )
    return candidates[0].resolve()


def _timestamp(now: dt.datetime | None = None) -> str:
    current = now or dt.datetime.now(dt.timezone.utc)
    return current.astimezone(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")


def activate_gap(candidate: Path, active: Path, archive_dir: Path) -> dict[str, str | None]:
    repository_id, version = inspect_gap(candidate)
    active.parent.mkdir(parents=True, exist_ok=True)
    archive_dir.mkdir(parents=True, exist_ok=True)

    previous_version: str | None = None
    archived_path: Path | None = None
    if active.exists():
        _, previous_version = inspect_gap(active)
        archived_path = archive_dir / f"gap-{previous_version}-{_timestamp()}.zip"
        if archived_path.exists():
            raise GapActivationError(f"Archivziel existiert bereits: {archived_path}")

    temp_fd, temp_name = tempfile.mkstemp(prefix=".gap-", suffix=".zip", dir=active.parent)
    os.close(temp_fd)
    temp_path = Path(temp_name)
    try:
        shutil.copy2(candidate, temp_path)
        inspect_gap(temp_path)

        if active.exists() and archived_path is not None:
            os.replace(active, archived_path)
        try:
            os.replace(temp_path, active)
        except Exception:
            if archived_path is not None and archived_path.exists() and not active.exists():
                os.replace(archived_path, active)
            raise
    finally:
        temp_path.unlink(missing_ok=True)

    return {
        "repositoryId": repository_id,
        "activatedVersion": version,
        "previousVersion": previous_version,
        "activePath": str(active),
        "archivedPath": str(archived_path) if archived_path else None,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("candidate", type=Path, help="Neu heruntergeladenes GAP-ZIP")
    parser.add_argument(
        "--active",
        type=Path,
        default=Path.home() / "netzwerksolution/workspace/downloads/current-gap/gap.zip",
        help="Aktiver GAP-Pfad",
    )
    parser.add_argument(
        "--archive-dir",
        type=Path,
        default=None,
        help="Archivverzeichnis; Standard: <active-parent>/old",
    )
    parser.add_argument(
        "--processed-dir",
        type=Path,
        default=Path.home() / "netzwerksolution/workspace/inbox/downloads/processed",
        help="Importer-Ziel für zeitgestempelte GAP-ZIPs",
    )
    args = parser.parse_args()
    archive_dir = args.archive_dir or args.active.parent / "old"
    try:
        requested = args.candidate.expanduser()
        resolved = resolve_candidate(requested, args.processed_dir)
        result = activate_gap(resolved, args.active.expanduser(), archive_dir.expanduser())
        result["requestedPath"] = str(requested)
        result["resolvedPath"] = str(resolved)
    except (GapActivationError, OSError) as exc:
        parser.exit(2, f"GAP-Aktivierung fehlgeschlagen: {exc}\n")
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())