60-packaging/compare_legacy_package.py
#!/usr/bin/env python3
"""Vergleicht ein Legacy-Uploadset mit einem neuen GAP-Arbeitspaket."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any, Iterable
from gap_work_package import (
WorkPackageError,
_inspect_repository,
_json_bytes,
_open_zip,
_parse_named_archives,
_read_json,
_root_file_entries,
validate_work_package,
)
CONSCIOUS_DIFFERENCES = [
"legacy-individual-repository-zips-vs-new-two-archive-model",
"new-prompt-and-semantic-gap-process-control",
"new-repository-roles-identities-and-allowed-target-scope",
"new-sha256-security-validation-and-atomic-publication",
"new-manifest-controlled-temporary-runtime-source-filter",
]
def _repository_files(data: bytes, name: str, label: str) -> dict[str, bytes]:
_inspect_repository(data, name, label)
archive, _, infos = _open_zip(data, label)
try:
return {info.filename: archive.read(info) for info in infos if not info.is_dir()}
finally:
archive.close()
def _legacy_entries(path: Path) -> tuple[dict[str, Any], dict[str, bytes]]:
if not path.is_dir():
raise WorkPackageError(f"Legacy-Ausgabeverzeichnis nicht gefunden: {path}")
actual = {item.name for item in path.iterdir()}
if "manifest.json" not in actual or any(name != "manifest.json" and not name.endswith(".zip") for name in actual):
raise WorkPackageError("Legacy-Ausgabe muss manifest.json und ausschließlich Repository-ZIPs enthalten")
if any(not (path / name).is_file() for name in actual):
raise WorkPackageError("Legacy-Ausgabe darf keine Unterverzeichnisse enthalten")
manifest = _read_json((path / "manifest.json").read_bytes(), "Legacy manifest.json")
if manifest.get("artifactType") != "chatgpt-upload-set" or manifest.get("version") != "2.0":
raise WorkPackageError("Legacy manifest.json muss chatgpt-upload-set Version 2.0 deklarieren")
archives = {name[:-4]: (path / name).read_bytes() for name in actual if name.endswith(".zip")}
if not archives:
raise WorkPackageError("Legacy-Ausgabe enthält keine Repository-ZIPs")
return manifest, archives
def _filter_evidence(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
result: dict[str, dict[str, Any]] = {}
for item in manifest.get("uploadFilters", []):
if not isinstance(item, dict) or not isinstance(item.get("repository"), str):
raise WorkPackageError("Neues Paket enthält ungültige Uploadfilter-Evidence")
result[item["repository"]] = item
return result
def compare_packages(
legacy_dir: Path,
new_package: Path,
gap_path: Path,
target_specs: Iterable[str],
) -> dict[str, Any]:
legacy_manifest, legacy_archives = _legacy_entries(legacy_dir)
new_manifest = validate_work_package(new_package)
source_entries = _root_file_entries((new_package / "sources.zip").read_bytes(), "sources.zip")
target_entries = _root_file_entries((new_package / "targets.zip").read_bytes(), "targets.zip")
new_sources = {name[:-4]: data for name, data in source_entries.items() if name.endswith(".zip")}
new_targets = {name[:-4]: data for name, data in target_entries.items() if name.endswith(".zip") and name != "gap.zip"}
if set(legacy_archives) != set(new_sources):
raise WorkPackageError(
f"Repositorymenge unterscheidet sich: Legacy={sorted(legacy_archives)}, Neu={sorted(new_sources)}"
)
evidence = _filter_evidence(new_manifest)
repository_results: list[dict[str, Any]] = []
for name in sorted(new_sources, key=lambda value: value.encode("utf-8")):
legacy_files = _repository_files(legacy_archives[name], name, f"Legacy {name}")
new_files = _repository_files(new_sources[name], name, f"Neue Source {name}")
common = set(legacy_files) & set(new_files)
changed = sorted(
(path for path in common if legacy_files[path] != new_files[path]),
key=lambda value: value.encode("utf-8"),
)
legacy_only = sorted(set(legacy_files) - set(new_files), key=lambda value: value.encode("utf-8"))
new_only = sorted(set(new_files) - set(legacy_files), key=lambda value: value.encode("utf-8"))
expected_legacy_only = evidence.get(name, {}).get("excludedPaths", [])
if not isinstance(expected_legacy_only, list):
raise WorkPackageError(f"Uploadfilter-Evidence für {name} enthält keine gültigen excludedPaths")
if changed or new_only or legacy_only != expected_legacy_only:
raise WorkPackageError(
f"Ungeklärte Inhaltsdifferenz in {name}: changed={changed}, newOnly={new_only}, "
f"legacyOnly={legacy_only}, erwartet={expected_legacy_only}"
)
repository_results.append({
"changedPaths": changed,
"commonFileCount": len(common),
"legacyFileCount": len(legacy_files),
"legacyOnlyPaths": legacy_only,
"name": name,
"newOnlyPaths": new_only,
"newSourceFileCount": len(new_files),
"result": "PASSED",
})
target_paths = _parse_named_archives(target_specs, "Target")
originals = {name: path.read_bytes() for name, path in target_paths}
if set(originals) != set(new_targets):
raise WorkPackageError(
f"Targetmenge unterscheidet sich: Eingabe={sorted(originals)}, Neu={sorted(new_targets)}"
)
target_results = []
for name in sorted(originals, key=lambda value: value.encode("utf-8")):
if originals[name] != new_targets[name]:
raise WorkPackageError(f"Target {name} ist nicht byteidentisch zur ursprünglichen Eingabe")
target_results.append({"name": name, "preservedByteIdentically": True})
if not gap_path.is_file() or target_entries.get("gap.zip") != gap_path.read_bytes():
raise WorkPackageError("gap.zip ist nicht byteidentisch zur ursprünglichen Eingabe")
return {
"artifactType": "legacy-package-equivalence-report",
"consciousDifferences": CONSCIOUS_DIFFERENCES,
"legacy": {
"artifactType": legacy_manifest["artifactType"],
"exportMode": legacy_manifest.get("exportMode"),
"roundtripId": legacy_manifest.get("roundtripId"),
"version": legacy_manifest["version"],
},
"new": {
"activeWorkItem": new_manifest["activeWorkItem"]["id"],
"artifactType": new_manifest["artifactType"],
"packageVersion": new_manifest["packageVersion"],
},
"repositories": repository_results,
"schemaVersion": "1.0",
"semanticGapPreservedByteIdentically": True,
"status": "PASSED",
"targets": target_results,
"unexplainedDifferences": [],
}
def write_report(report: dict[str, Any], output: Path) -> None:
output = output.expanduser().resolve()
if output.exists():
raise WorkPackageError(f"Bericht existiert bereits und wird nicht überschrieben: {output}")
output.parent.mkdir(parents=True, exist_ok=True)
temporary = output.with_name(f".{output.name}.tmp")
try:
temporary.write_bytes(_json_bytes(report))
temporary.replace(output)
finally:
if temporary.exists():
temporary.unlink()
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--legacy-dir", required=True, type=Path)
parser.add_argument("--new-package", required=True, type=Path)
parser.add_argument("--gap", required=True, type=Path)
parser.add_argument("--target", action="append", default=[], metavar="NAME=PFAD")
parser.add_argument("--report", required=True, type=Path)
args = parser.parse_args()
try:
report = compare_packages(args.legacy_dir, args.new_package, args.gap, args.target)
write_report(report, args.report)
print(f"Paketgleichwertigkeit bestätigt: {args.report}")
return 0
except (OSError, WorkPackageError, ValueError) as exc:
print(f"FEHLER: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())