bin/activate-latest-gap-and-prepare.py
#!/usr/bin/env python3
"""Aktiviert das neueste verarbeitete GAP und erzeugt das nächste Uploadpaket."""
from __future__ import annotations
import argparse, hashlib, importlib.util, json, subprocess, sys
from pathlib import Path
class WorkflowError(RuntimeError): pass
def newest_gap(directory:Path)->Path:
candidates=[p for p in directory.glob("*semantic-gap*.zip") if p.is_file()]
if not candidates: raise WorkflowError(f"Kein Semantic-GAP-ZIP gefunden: {directory}")
return max(candidates,key=lambda p:(p.stat().st_mtime_ns,p.name))
def sha256(path:Path)->str:
h=hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda:stream.read(1024*1024),b""):h.update(block)
return h.hexdigest()
def run(args):
candidate=newest_gap(args.processed.expanduser().resolve());digest=sha256(candidate)
if args.expected_sha256 and digest.lower()!=args.expected_sha256.lower():raise WorkflowError(f"SHA-256 stimmt nicht: erwartet {args.expected_sha256}, gefunden {digest}")
runtime=Path(__file__).resolve().parents[1];activate=runtime/"bin/activate-gap.py";prepare=runtime/"bin/prepare-work-package.py"
spec=importlib.util.spec_from_file_location("activate_gap",activate);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
repository,version=module.inspect_gap(candidate)
selected={"candidate":str(candidate),"sha256":digest,"repositoryId":repository,"version":version}
print(json.dumps({"selectedGap":selected},ensure_ascii=False,indent=2))
if args.dry_run:return selected
subprocess.run([sys.executable,str(activate),str(candidate)],check=True)
subprocess.run([sys.executable,str(prepare)],check=True)
return selected
def main():
p=argparse.ArgumentParser();p.add_argument("--processed",type=Path,default=Path.home()/"netzwerksolution/workspace/inbox/downloads/processed");p.add_argument("--expected-sha256");p.add_argument("--dry-run",action="store_true");a=p.parse_args()
try:run(a)
except (WorkflowError,OSError,subprocess.CalledProcessError) as e:p.exit(2,f"Roundtrip-Vorbereitung fehlgeschlagen: {e}\n")
return 0
if __name__=="__main__":raise SystemExit(main())