bin/prepare-work-package.py

#!/usr/bin/env python3
"""Erzeugt aktive GAP-Arbeitspakete oder ein Folgeprojekt-Startpaket."""
from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import zipfile
from pathlib import Path, PurePosixPath
from typing import Any


class PreparationError(RuntimeError):
    pass


def _unique_names(value: Any, field: str) -> list[str]:
    if not isinstance(value, list) or not value:
        raise PreparationError(f"{field} muss eine nicht leere Liste sein")
    result: list[str] = []
    for name in value:
        if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9._-]+", name):
            raise PreparationError(f"{field} enthält einen ungültigen Repositorynamen: {name!r}")
        if name in result:
            raise PreparationError(f"{field} enthält einen doppelten Repositorynamen: {name}")
        result.append(name)
    return result


def _frontmatter_list(text: str, field: str) -> list[str]:
    if not text.startswith("---"):
        raise PreparationError("Aktives Work Item besitzt kein Frontmatter")
    parts = text.split("---", 2)
    if len(parts) < 3:
        raise PreparationError("Frontmatter des aktiven Work Items ist unvollständig")
    for line in parts[1].splitlines():
        if line.startswith(f"{field}:"):
            try:
                return _unique_names(json.loads(line.split(":", 1)[1].strip()), field)
            except json.JSONDecodeError as exc:
                raise PreparationError(f"{field} im Work Item ist ungültig: {exc}") from exc
    raise PreparationError(f"Aktives Work Item deklariert {field} nicht")


def _require_same_repository_set(
    manifest_values: list[str], work_item_values: list[str], label: str
) -> None:
    """Validate repository membership without making declaration order semantic."""
    manifest_set = set(manifest_values)
    work_item_set = set(work_item_values)
    if manifest_set == work_item_set:
        return
    only_manifest = sorted(manifest_set - work_item_set)
    only_work_item = sorted(work_item_set - manifest_set)
    details: list[str] = []
    if only_manifest:
        details.append("nur Manifest: " + ", ".join(only_manifest))
    if only_work_item:
        details.append("nur Work Item: " + ", ".join(only_work_item))
    raise PreparationError(
        f"{label}-Repositories in GAP-Manifest und Work Item widersprechen sich "
        f"({'; '.join(details)})"
    )


def _read_gap_contract(path: Path) -> tuple[str, list[str], list[str]]:
    if not path.is_file() or not zipfile.is_zipfile(path):
        raise PreparationError(f"Aktives Semantic GAP ist kein gültiges ZIP: {path}")
    with zipfile.ZipFile(path) as archive:
        infos = archive.infolist()
        names = [item.filename for item in infos]
        if len(names) != len(set(names)):
            raise PreparationError("Semantic GAP enthält doppelte ZIP-Einträge")
        for name in names:
            candidate = Path(name)
            if candidate.is_absolute() or ".." in candidate.parts:
                raise PreparationError(f"Unsicherer GAP-ZIP-Pfad: {name}")
        manifests: list[tuple[dict[str, Any], str]] = []
        for item in infos:
            if Path(item.filename).name != "manifest.json":
                continue
            try:
                value = json.loads(archive.read(item).decode("utf-8"))
            except (UnicodeDecodeError, json.JSONDecodeError) as exc:
                raise PreparationError(f"Ungültiges GAP-Manifest: {exc}") from exc
            if isinstance(value, dict) and value.get("artifactType") == "semantic-gap":
                manifests.append((value, item.filename))
        if len(manifests) != 1:
            raise PreparationError(
                "Semantic GAP muss genau ein Manifest mit artifactType=semantic-gap enthalten; "
                f"gefunden: {len(manifests)}"
            )
        manifest, manifest_name = manifests[0]
        active = manifest.get("activeWorkItem")
        if not isinstance(active, dict):
            raise PreparationError("GAP-Manifest enthält kein aktives Work Item")
        work_item_path = active.get("path")
        if not isinstance(work_item_path, str):
            raise PreparationError("GAP-Manifest enthält keinen Work-Item-Pfad")
        root = Path(manifest_name).parent
        archive_path = (root / work_item_path).as_posix()
        if archive_path not in names:
            raise PreparationError(f"Aktives Work Item fehlt im GAP: {archive_path}")
        try:
            work_item_text = archive.read(archive_path).decode("utf-8")
        except UnicodeDecodeError as exc:
            raise PreparationError("Aktives Work Item ist nicht UTF-8") from exc
    work_item = active.get("id")
    if not isinstance(work_item, str) or not re.fullmatch(r"[A-Za-z0-9._-]+", work_item):
        raise PreparationError(f"Ungültige Work-Item-ID: {work_item!r}")
    sources = _unique_names(active.get("requiredSourceRepositories"), "requiredSourceRepositories")
    targets = _unique_names(active.get("allowedTargetRepositories"), "allowedTargetRepositories")
    work_item_sources = _frontmatter_list(work_item_text, "required_source_repositories")
    work_item_targets = _frontmatter_list(work_item_text, "allowed_target_repositories")
    _require_same_repository_set(sources, work_item_sources, "Source")
    _require_same_repository_set(targets, work_item_targets, "Target")
    return work_item, sources, targets


def _normalize_gap_archive_root(source: Path, output: Path) -> Path:
    """Normalize a historical GAP ZIP to the required semantic-gap-000 root."""
    canonical_root = PurePosixPath("semantic-gap-000")
    if not source.is_file() or not zipfile.is_zipfile(source):
        raise PreparationError(f"Aktives Semantic GAP ist kein gültiges ZIP: {source}")
    with zipfile.ZipFile(source) as archive:
        infos = [item for item in archive.infolist() if not item.is_dir()]
        manifests: list[zipfile.ZipInfo] = []
        for item in infos:
            if PurePosixPath(item.filename).name != "manifest.json":
                continue
            try:
                value = json.loads(archive.read(item).decode("utf-8"))
            except (UnicodeDecodeError, json.JSONDecodeError) as exc:
                raise PreparationError(f"Ungültiges GAP-Manifest: {exc}") from exc
            if isinstance(value, dict) and value.get("artifactType") == "semantic-gap":
                manifests.append(item)
        if len(manifests) != 1:
            raise PreparationError("Semantic GAP enthält kein eindeutiges Manifest für die Root-Normalisierung")
        source_root = PurePosixPath(manifests[0].filename).parent
        if source_root == canonical_root:
            return source
        entries: list[tuple[zipfile.ZipInfo, str]] = []
        for item in infos:
            path = PurePosixPath(item.filename)
            if path.is_absolute() or ".." in path.parts:
                raise PreparationError(f"Unsicherer GAP-ZIP-Pfad: {item.filename}")
            try:
                relative = path.relative_to(source_root)
            except ValueError as exc:
                raise PreparationError(
                    f"GAP-ZIP enthält mehrere Repository-Wurzeln: {item.filename}"
                ) from exc
            entries.append((item, (canonical_root / relative).as_posix()))
        output.parent.mkdir(parents=True, exist_ok=True)
        with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as normalized:
            for item, name in sorted(entries, key=lambda value: value[1].encode("utf-8")):
                info = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
                info.compress_type = zipfile.ZIP_DEFLATED
                info.external_attr = item.external_attr
                normalized.writestr(info, archive.read(item), compress_type=zipfile.ZIP_DEFLATED, compresslevel=9)
    return output


def _read_repository_map(path: Path) -> dict[str, Path]:
    try:
        config = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise PreparationError(f"Repository-Zuordnung ist ungültig: {path}: {exc}") from exc
    values = config.get("repositories")
    if not isinstance(values, dict):
        raise PreparationError("Repository-Zuordnung enthält kein repositories-Objekt")
    result: dict[str, Path] = {}
    for name, raw_path in values.items():
        if not isinstance(name, str) or not isinstance(raw_path, str):
            raise PreparationError("Repository-Zuordnung enthält einen ungültigen Eintrag")
        resolved = Path(raw_path).expanduser().resolve()
        result[name] = resolved
    return result


def _next_output(root: Path, work_item: str, now: dt.datetime | None = None) -> Path:
    timestamp = (now or dt.datetime.now().astimezone()).strftime("%Y%m%d-%H%M%S")
    base = root / f"{work_item}-{timestamp}"
    candidate = base
    suffix = 2
    while candidate.exists():
        candidate = root / f"{base.name}-{suffix:02d}"
        suffix += 1
    return candidate


def _run(command: list[str], description: str) -> None:
    result = subprocess.run(command, check=False)
    if result.returncode != 0:
        raise PreparationError(f"{description} fehlgeschlagen (Exit {result.returncode})")


FOLLOW_UP_REPOSITORIES = [
    "enterprise-architecture",
    "solution-architecture",
    "engineering-tools",
    "runtime",
]


def _write_bundle(path: Path, entries: dict[str, Path]) -> None:
    """Schreibt ein reproduzierbares ZIP mit bereits erzeugten Repository-ZIPs."""
    with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
        for name, source in entries.items():
            info = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
            info.compress_type = zipfile.ZIP_DEFLATED
            info.external_attr = 0o100644 << 16
            archive.writestr(info, source.read_bytes())


def prepare_follow_up_bootstrap(
    *,
    runtime_root: Path,
    gap: Path,
    repository_map: Path,
    output_root: Path,
) -> Path:
    """Erzeugt einen Work-Item-unabhaengigen Intake fuer ein Folgeprojekt.

    Das bestehende GAP bleibt als referenzierbare Eingabe erhalten; sein
    terminaler Zustand wird hier absichtlich nicht als Fehler interpretiert.
    """
    if not gap.is_file() or not zipfile.is_zipfile(gap):
        raise PreparationError(f"Referenz-GAP ist kein gültiges ZIP: {gap}")
    repositories = _read_repository_map(repository_map)
    missing = [name for name in FOLLOW_UP_REPOSITORIES if name not in repositories]
    if missing:
        raise PreparationError(f"Repository-Zuordnung fehlt für: {', '.join(missing)}")
    invalid = [f"{name}={repositories[name]}" for name in FOLLOW_UP_REPOSITORIES if not repositories[name].is_dir()]
    if invalid:
        raise PreparationError(f"Lokales Repository fehlt: {', '.join(invalid)}")
    repository_adapter = Path(
        os.environ.get("NS_REPOSITORY_SET_ADAPTER", str(runtime_root / "bin/create-repository-set.py"))
    ).expanduser().resolve()
    if not repository_adapter.is_file():
        raise PreparationError(f"Runtime-Adapter nicht gefunden: {repository_adapter}")

    output_root.mkdir(parents=True, exist_ok=True)
    final_output = _next_output(output_root, "FOLLOW-UP-BOOTSTRAP")
    staging = Path(tempfile.mkdtemp(prefix=".FOLLOW-UP-BOOTSTRAP-", dir=output_root))
    try:
        repository_output = staging / "repositories"
        command = [sys.executable, str(repository_adapter)]
        for name in FOLLOW_UP_REPOSITORIES:
            command.extend(["--repository", f"{name}={repositories[name]}"])
        command.extend(["--output-dir", str(repository_output)])
        _run(command, "Repository-Materialisierung")

        archives = {name: repository_output / f"{name}.zip" for name in FOLLOW_UP_REPOSITORIES}
        if not all(path.is_file() and zipfile.is_zipfile(path) for path in archives.values()):
            raise PreparationError("Repository-Materialisierung ist unvollständig")
        archives["gap"] = gap
        package = staging / "package"
        package.mkdir()
        source_entries = {f"{name}.zip": path for name, path in archives.items()}
        target_entries = {f"{name}.zip": path for name, path in archives.items()}
        _write_bundle(package / "sources.zip", source_entries)
        _write_bundle(package / "targets.zip", target_entries)
        manifest = {
            "schemaVersion": "1.0",
            "artifactType": "follow-up-workflow-bootstrap",
            "description": "Work-Item-unabhängiger Ausgangsstand für ein neues Folge-Workflow-Projekt.",
            "sourceRepositories": FOLLOW_UP_REPOSITORIES,
            "targetRepositories": FOLLOW_UP_REPOSITORIES,
            "referenceArtifacts": ["gap.zip"],
            "gapRole": "immutable-reference-and-copy-baseline",
        }
        (package / "package-manifest.json").write_text(
            json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
        )
        sums = []
        for name in ("sources.zip", "targets.zip", "package-manifest.json"):
            digest = hashlib.sha256((package / name).read_bytes()).hexdigest()
            sums.append(f"{digest}  {name}")
        (package / "SHA256SUMS.txt").write_text("\n".join(sums) + "\n", encoding="utf-8")
        if not zipfile.is_zipfile(package / "sources.zip") or not zipfile.is_zipfile(package / "targets.zip"):
            raise PreparationError("Folgeprojekt-Startpaket enthält ungültige ZIP-Dateien")
        os.replace(package, final_output)
    except Exception:
        if final_output.exists():
            shutil.rmtree(final_output, ignore_errors=True)
        raise
    finally:
        shutil.rmtree(staging, ignore_errors=True)
    return final_output


def prepare(
    *,
    runtime_root: Path,
    gap: Path,
    repository_map: Path,
    output_root: Path,
    filter_config: Path,
) -> tuple[str, Path]:
    work_item, source_names, target_names = _read_gap_contract(gap)
    repositories = _read_repository_map(repository_map)
    selected = source_names + [name for name in target_names if name not in source_names]
    missing = [name for name in selected if name not in repositories]
    if missing:
        raise PreparationError(f"Repository-Zuordnung fehlt für: {', '.join(missing)}")
    invalid = [f"{name}={repositories[name]}" for name in selected if not repositories[name].is_dir()]
    if invalid:
        raise PreparationError(f"Lokales Repository fehlt: {', '.join(invalid)}")
    if not filter_config.is_file():
        raise PreparationError(f"Uploadfilter nicht gefunden: {filter_config}")

    repository_adapter = Path(
        os.environ.get("NS_REPOSITORY_SET_ADAPTER", str(runtime_root / "bin/create-repository-set.py"))
    ).expanduser().resolve()
    package_adapter = Path(
        os.environ.get("NS_WORK_PACKAGE_ADAPTER", str(runtime_root / "bin/create-work-package.py"))
    ).expanduser().resolve()
    for adapter in (repository_adapter, package_adapter):
        if not adapter.is_file():
            raise PreparationError(f"Runtime-Adapter nicht gefunden: {adapter}")

    output_root.mkdir(parents=True, exist_ok=True)
    final_output = _next_output(output_root, work_item)
    staging = Path(tempfile.mkdtemp(prefix=f".{work_item}-", dir=output_root))
    try:
        normalized_gap = _normalize_gap_archive_root(gap, staging / "gap.zip")
        repository_output = staging / "repositories"
        command = [sys.executable, str(repository_adapter)]
        for name in selected:
            command.extend(["--repository", f"{name}={repositories[name]}"])
        command.extend(["--output-dir", str(repository_output)])
        _run(command, "Repository-Materialisierung")

        config = {
            "schemaVersion": "1.0",
            "tool": "../engineering-tools/60-packaging/gap_work_package.py",
            "filterConfig": str(filter_config),
            "gap": str(normalized_gap),
            "sources": [
                {"name": name, "archive": str(repository_output / f"{name}.zip")}
                for name in source_names
            ],
            "targets": [
                {"name": name, "archive": str(repository_output / f"{name}.zip")}
                for name in target_names
            ],
            "outputDirectory": str(staging / "package"),
        }
        config_path = staging / "work-package.json"
        config_path.write_text(json.dumps(config, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        _run(
            [
                sys.executable,
                str(package_adapter),
                "--config",
                str(config_path),
                "--runtime-root",
                str(runtime_root),
            ],
            "GAP-Arbeitspaketbildung",
        )
        package = staging / "package"
        required = [package / "sources.zip", package / "targets.zip", package / "package-manifest.json", package / "SHA256SUMS.txt"]
        if not all(path.is_file() for path in required):
            raise PreparationError("GAP-Arbeitspaket ist unvollständig")
        os.replace(package, final_output)
    except Exception:
        if final_output.exists():
            shutil.rmtree(final_output, ignore_errors=True)
        raise
    finally:
        shutil.rmtree(staging, ignore_errors=True)
    return work_item, final_output


def main() -> int:
    runtime_default = Path(__file__).resolve().parents[1]
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--runtime-root", type=Path, default=runtime_default)
    parser.add_argument(
        "--gap",
        type=Path,
        default=Path.home() / "netzwerksolution/workspace/downloads/current-gap/gap.zip",
    )
    parser.add_argument(
        "--repository-map",
        type=Path,
        default=None,
    )
    parser.add_argument(
        "--output-root",
        type=Path,
        default=Path.home() / "netzwerksolution/workspace/out/chatgpt-upload",
    )
    parser.add_argument(
        "--filter-config",
        type=Path,
        default=runtime_default.parent / "engineering-tools" / "60-packaging" / "upload-filters.json",
    )
    parser.add_argument(
        "--follow-up-bootstrap",
        action="store_true",
        help="Erzeugt einen Work-Item-unabhängigen Intake für ein neues Folgeprojekt.",
    )
    args = parser.parse_args()
    runtime_root = args.runtime_root.expanduser().resolve()
    repository_map = args.repository_map or runtime_root / "config/repository-map.json"
    try:
        if args.follow_up_bootstrap:
            output = prepare_follow_up_bootstrap(
                runtime_root=runtime_root,
                gap=args.gap.expanduser().resolve(),
                repository_map=repository_map.expanduser().resolve(),
                output_root=args.output_root.expanduser().resolve(),
            )
            work_item = "FOLLOW-UP-BOOTSTRAP"
        else:
            work_item, output = prepare(
                runtime_root=runtime_root,
                gap=args.gap.expanduser().resolve(),
                repository_map=repository_map.expanduser().resolve(),
                output_root=args.output_root.expanduser().resolve(),
                filter_config=args.filter_config.expanduser().resolve(),
            )
    except (OSError, PreparationError) as exc:
        parser.exit(2, f"FEHLER: {exc}\n")
    print()
    print(f"Fertig: {work_item}")
    print(f"sources.zip: {output / 'sources.zip'}")
    print(f"targets.zip: {output / 'targets.zip'}")
    return 0


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