40-validation/validate-canonical-sources.py

#!/usr/bin/env python3
"""Prüft genau eine aktive kanonische Quelle je semantischer Einheit."""
import argparse
import json
from pathlib import Path

REFERENCE_TYPES = {"references-canonical", "reference-to-canonical"}


def validate(inputs):
    units = []
    findings = []
    for repository, path in inputs:
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except Exception as exc:
            findings.append({"severity": "error", "code": "INVALID_MANIFEST", "location": f"{repository}:{path}", "message": str(exc)})
            continue
        for index, unit in enumerate(data.get("semanticUnits", [])):
            units.append({**unit, "repository": repository, "location": f"{repository}:{path}#semanticUnits[{index}]"})

    by_id = {unit.get("id"): unit for unit in units if isinstance(unit.get("id"), str)}
    canonical = {}
    historical = []
    references = []
    review = []
    for unit in units:
        status = unit.get("validity", {}).get("status")
        if status == "historical":
            historical.append({"id": unit.get("id"), "location": unit["location"], "canonicalSource": unit.get("canonicalSource")})
            continue
        rels = unit.get("relationships", [])
        refs = [rel for rel in rels if rel.get("type") in REFERENCE_TYPES]
        if refs:
            for rel in refs:
                target = by_id.get(rel.get("id"))
                if target is None or target.get("validity", {}).get("status") == "historical":
                    review.append({"code": "UNRESOLVED_CANONICAL_REFERENCE", "id": unit.get("id"), "target": rel.get("id"), "location": unit["location"]})
                else:
                    references.append({"id": unit.get("id"), "target": rel.get("id"), "location": unit["location"]})
            continue
        source = unit.get("canonicalSource")
        if not isinstance(source, str) or not source.strip():
            review.append({"code": "CANONICAL_SOURCE_UNDECLARED", "id": unit.get("id"), "location": unit["location"]})
            continue
        canonical.setdefault(unit.get("id"), []).append({"source": source, "location": unit["location"]})

    for identity, declarations in canonical.items():
        unique_sources = sorted({item["source"] for item in declarations})
        if len(declarations) > 1 or len(unique_sources) > 1:
            findings.append({"severity": "error", "code": "MULTIPLE_CANONICAL_SOURCES", "id": identity, "sources": unique_sources, "locations": [item["location"] for item in declarations]})

    status = "FAILED" if findings else ("REVIEW_REQUIRED" if review else "PASSED")
    return {
        "schemaVersion": "1.0", "workItem": "SE-0025", "status": status,
        "summary": {"manifests": len(inputs), "semanticUnits": len(units), "canonicalUnits": len(canonical), "references": len(references), "historical": len(historical), "reviewCases": len(review)},
        "canonicalSources": canonical, "references": references, "historicalSources": historical,
        "reviewCases": review, "findings": findings,
    }


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", action="append", required=True, help="repository=path")
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    inputs = []
    for value in args.input:
        repository, separator, path = value.partition("=")
        if not separator:
            parser.error("--input benötigt repository=path")
        inputs.append((repository, Path(path)))
    result = validate(inputs)
    text = json.dumps(result, 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 result["status"] in {"PASSED", "REVIEW_REQUIRED"} else 1)


if __name__ == "__main__":
    main()