40-validation/validate-temporal-context.py

#!/usr/bin/env python3
import argparse
import datetime as dt
import importlib.util
import json
from pathlib import Path

ROOT = Path(__file__).parent
SPEC = importlib.util.spec_from_file_location("semantic_manifest", ROOT / "validate-semantic-manifest.py")
MANIFEST = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MANIFEST)
TEMPORAL_TYPES = {"supersedes", "evolves_from", "valid_during"}

def finding(severity, code, location, message):
    return {"severity": severity, "code": code, "location": location, "message": message}

def parse_date(value, location, findings):
    if value is None:
        return None
    if not isinstance(value, str):
        findings.append(finding("error", "INVALID_VALIDITY_DATE", location, "Zeitangabe muss ISO-8601-Datum sein.")); return None
    try:
        return dt.date.fromisoformat(value)
    except ValueError:
        findings.append(finding("error", "INVALID_VALIDITY_DATE", location, "Zeitangabe muss ISO-8601-Datum sein.")); return None

def validate(data):
    findings = []
    structural = MANIFEST.validate(data)
    for error in structural:
        findings.append(finding("error", "INVALID_SEMANTIC_MANIFEST", "semanticManifest", error))
    units = data.get("semanticUnits", []) if isinstance(data, dict) else []
    ids = {unit.get("id") for unit in units if isinstance(unit, dict) and isinstance(unit.get("id"), str)}
    status_by_id = {unit.get("id"): unit.get("validity", {}).get("status") for unit in units if isinstance(unit, dict) and isinstance(unit.get("validity"), dict)}
    for index, unit in enumerate(units):
        if not isinstance(unit, dict):
            continue
        location = f"semanticUnits[{index}]"
        validity = unit.get("validity", {})
        if not isinstance(validity, dict):
            continue
        start = parse_date(validity.get("from"), location + ".validity.from", findings)
        end = parse_date(validity.get("until"), location + ".validity.until", findings)
        if start and end and end < start:
            findings.append(finding("error", "INVALID_VALIDITY_RANGE", location + ".validity", "validity.until darf nicht vor validity.from liegen."))
        relationships = unit.get("relationships", [])
        if not isinstance(relationships, list):
            continue
        for rel_index, relation in enumerate(relationships):
            if not isinstance(relation, dict) or relation.get("type") not in TEMPORAL_TYPES:
                continue
            rel_location = f"{location}.relationships[{rel_index}]"
            target = relation.get("id")
            if not relation.get("timeContext"):
                findings.append(finding("error", "MISSING_RELATIONSHIP_TIME_CONTEXT", rel_location + ".timeContext", "Zeitliche Beziehung benötigt einen Zeit- oder Plateaukontext."))
            if relation["type"] == "supersedes" and not relation.get("evidence"):
                findings.append(finding("review", "MISSING_SUPERSEDES_EVIDENCE", rel_location + ".evidence", "Eine Ablösung benötigt einen fachlich zu prüfenden Nachweis."))
            if target not in ids:
                findings.append(finding("review", "UNRESOLVED_TEMPORAL_TARGET", rel_location + ".id", "Ziel der zeitlichen Beziehung ist im geprüften Manifest nicht auflösbar."))
            elif relation["type"] == "supersedes" and status_by_id.get(target) == "active":
                findings.append(finding("review", "SUPERSEDED_TARGET_STILL_ACTIVE", rel_location + ".id", "Als abgelöst deklarierter Zielstand ist weiterhin aktiv; Gültigkeit fachlich prüfen."))
    status = "FAILED" if any(item["severity"] == "error" for item in findings) else ("REVIEW_REQUIRED" if findings else "PASSED")
    return {"schemaVersion": "1.0", "status": status, "contentTruthAssessed": False, "findings": findings}

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("manifest", type=Path)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    try:
        report = validate(json.loads(args.manifest.read_text(encoding="utf-8")))
    except Exception as exc:
        report = {"schemaVersion":"1.0", "status":"FAILED", "contentTruthAssessed":False, "findings":[finding("error", "INVALID_INPUT", str(args.manifest), str(exc))]}
    text = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(text, encoding="utf-8")
    else:
        print(text, end="")
    raise SystemExit(0 if report["status"] in {"PASSED", "REVIEW_REQUIRED"} else 1)

if __name__ == "__main__":
    main()