normative_contract.py

from __future__ import annotations

import hashlib
import json
from dataclasses import asdict, dataclass
from datetime import date
from pathlib import Path
from typing import Any


CORE_FRAME_FIELDS = {
    "actor",
    "action",
    "object",
    "condition",
    "exception",
    "temporal_constraint",
    "modality",
    "norm_category",
    "confidence",
}

DECISION_OPERATORS = {"eq", "ne", "gt", "gte", "lt", "lte", "in", "not_in", "truthy", "falsy"}


@dataclass(frozen=True)
class ContractIssue:
    severity: str
    code: str
    entity_id: str
    message: str


def validate_corpus(corpus: dict[str, Any]) -> dict[str, Any]:
    """Validate identity, provenance and referential integrity before runtime use."""
    issues: list[ContractIssue] = []
    documents = corpus.get("documents", []) or []
    concepts = corpus.get("concepts", []) or []
    clauses = corpus.get("clauses", []) or []
    evidence_spans = corpus.get("evidence_spans", []) or []
    decision_contracts = corpus.get("decision_contracts", []) or []
    relations = corpus.get("cross_document_edges", []) or []

    document_ids = _unique_ids(documents, "document_id", "document", issues)
    concept_ids = _unique_ids(concepts, "concept_id", "concept", issues)
    clause_ids = _unique_ids(clauses, "clause_id", "clause", issues)
    evidence_ids = _unique_ids(evidence_spans, "evidence_id", "evidence", issues)
    _unique_ids(decision_contracts, "contract_id", "decision_contract", issues)

    if not corpus.get("build_id"):
        _add(issues, "error", "missing_build_id", "corpus", "Corpus build_id alanı zorunludur.")

    for document in documents:
        entity_id = str(document.get("document_id", ""))
        for field in (
            "title", "short_code", "source_path", "source_sha256", "version_id",
            "authority_level", "validity_status", "source_authority", "source_snapshot_date",
            "temporal_coverage", "jurisdiction", "authority_rank",
        ):
            if not document.get(field):
                _add(issues, "error", f"missing_document_{field}", entity_id, f"Belgede {field} alanı yok.")
        snapshot = str(document.get("source_snapshot_date", "") or "")
        try:
            date.fromisoformat(snapshot)
        except ValueError:
            _add(issues, "error", "invalid_source_snapshot_date", entity_id, "Belge kaynak anlık görüntü tarihi ISO-8601 olmalıdır.")

    reviewed_concepts = 0
    answer_ready_concepts = 0
    allowed_review_statuses = {"derived_from_structure", "human_reviewed", "expert_approved", "human_rejected"}
    for concept in concepts:
        entity_id = str(concept.get("concept_id", ""))
        document_id = str(concept.get("document_id", ""))
        if document_id not in document_ids:
            _add(issues, "error", "unknown_concept_document", entity_id, "Concept bilinmeyen bir belgeye bağlı.")
        if not entity_id.startswith(_id_prefix(document_id)):
            _add(issues, "error", "unscoped_concept_id", entity_id, "Concept ID belge namespace'i taşımıyor.")
        policy = concept.get("source_policy", {}) or {}
        if policy.get("source_lock") is not True:
            _add(issues, "error", "source_lock_disabled", entity_id, "Concept source_lock=true olmalıdır.")
        metadata = concept.get("normative_metadata", {}) or {}
        for field in ("regulates", "domain_path", "search_text", "provenance"):
            if not metadata.get(field):
                _add(
                    issues,
                    "error",
                    f"missing_normative_metadata_{field}",
                    entity_id,
                    f"Concept normative_metadata.{field} alanını taşımalıdır.",
                )
        review_status = str(metadata.get("review_status", "") or "")
        if review_status not in allowed_review_statuses:
            _add(issues, "error", "invalid_review_status", entity_id, f"Bilinmeyen uzman inceleme durumu: {review_status}.")
        if review_status in {"human_reviewed", "expert_approved"}:
            reviewed_concepts += 1
        approved_summary = str(metadata.get("approved_summary", "") or "").strip()
        if approved_summary:
            if review_status != "expert_approved":
                _add(issues, "error", "unapproved_canonical_summary", entity_id, "Kanonik kısa sonuç yalnız expert_approved hükümde yayımlanabilir.")
            source = str(concept.get("source_text", "") or "").casefold()
            invalid_terms = []
            for point in metadata.get("approved_points", []) or []:
                if not isinstance(point, dict) or not str(point.get("statement", "") or "").strip():
                    _add(issues, "error", "invalid_approved_point", entity_id, "Onaylı sonuç birimi statement taşımalıdır.")
                    continue
                for term in point.get("evidence_terms", []) or []:
                    if str(term).casefold() not in source:
                        invalid_terms.append(str(term))
            if invalid_terms:
                _add(issues, "error", "approved_point_source_mismatch", entity_id, f"Onaylı sonuç kanıtı kaynakta yok: {', '.join(invalid_terms[:4])}.")
            else:
                answer_ready_concepts += 1

    for clause in clauses:
        entity_id = str(clause.get("clause_id", ""))
        document_id = str(clause.get("document_id", ""))
        parent_id = str(clause.get("parent_concept_id", ""))
        if parent_id not in concept_ids:
            _add(issues, "error", "orphan_clause", entity_id, "Clause parent concept bulunamadı.")
        if not entity_id.startswith(_id_prefix(document_id)):
            _add(issues, "error", "unscoped_clause_id", entity_id, "Clause ID belge namespace'i taşımıyor.")

    low_confidence = 0
    for evidence in evidence_spans:
        entity_id = str(evidence.get("evidence_id", ""))
        document_id = str(evidence.get("document_id", ""))
        parent_id = str(evidence.get("parent_clause_id", ""))
        if parent_id not in clause_ids:
            _add(issues, "error", "orphan_evidence", entity_id, "Evidence parent clause bulunamadı.")
        if not entity_id.startswith(_id_prefix(document_id)):
            _add(issues, "error", "unscoped_evidence_id", entity_id, "Evidence ID belge namespace'i taşımıyor.")
        if not str(evidence.get("source_text", "")).strip():
            _add(issues, "error", "empty_evidence", entity_id, "Evidence kaynak metni boş.")
        span = evidence.get("source_span", {}) or {}
        if not isinstance(span.get("char_start"), int) or not isinstance(span.get("char_end"), int):
            _add(issues, "error", "invalid_source_span", entity_id, "Evidence kaynak konumu geçersiz.")
        frame = evidence.get("semantic_frame", {}) or {}
        missing_frame = sorted(CORE_FRAME_FIELDS - set(frame))
        if missing_frame:
            _add(issues, "error", "incomplete_semantic_frame", entity_id, f"Eksik frame alanları: {', '.join(missing_frame)}")
        if float(frame.get("confidence", 0.0) or 0.0) < 0.45:
            low_confidence += 1

    if evidence_spans and low_confidence / len(evidence_spans) > 0.45:
        _add(
            issues,
            "warning",
            "high_low_confidence_ratio",
            "corpus",
            f"Evidence kayıtlarının {low_confidence}/{len(evidence_spans)} kadarı düşük çıkarım güvenine sahip.",
        )

    article_refs = {
        (str(item.get("document_id", "")), str(item.get("article_id", "")))
        for item in [*concepts, *clauses]
        if item.get("document_id") and item.get("article_id")
    }
    for contract in decision_contracts:
        _validate_decision_contract(contract, document_ids, article_refs, issues)

    for relation in relations:
        relation_id = str(relation.get("edge_id", "") or "")
        source_scope = (str(relation.get("source_document_id", "")), str(relation.get("source_article_id", "")))
        target_scope = (str(relation.get("target_document_id", "")), str(relation.get("target_article_id", "")))
        if source_scope not in article_refs or target_scope not in article_refs:
            _add(issues, "error", "invalid_relation_scope", relation_id, "Belge ilişkisi corpus içinde bulunmayan hükme bağlanıyor.")
        if relation.get("approved_interpretation") and relation.get("review_status") != "expert_approved":
            _add(issues, "error", "unapproved_relation_interpretation", relation_id, "İlişki yorumu expert_approved olmalıdır.")

    errors = [issue for issue in issues if issue.severity == "error"]
    warnings = [issue for issue in issues if issue.severity == "warning"]
    return {
        "schema": "MCKF-ValidationReport-v1.2",
        "build_id": corpus.get("build_id", ""),
        "conforms": not errors,
        "summary": {
            "documents": len(documents),
            "concepts": len(concepts),
            "clauses": len(clauses),
            "evidence_spans": len(evidence_spans),
            "decision_contracts": len(decision_contracts),
            "reviewed_concepts": reviewed_concepts,
            "answer_ready_concepts": answer_ready_concepts,
            "derived_concepts": max(0, len(concepts) - reviewed_concepts),
            "review_coverage": round(reviewed_concepts / len(concepts), 4) if concepts else 0.0,
            "errors": len(errors),
            "warnings": len(warnings),
            "low_confidence_evidence": low_confidence,
        },
        "issues": [asdict(issue) for issue in issues],
    }


def assert_valid_corpus(corpus: dict[str, Any]) -> dict[str, Any]:
    report = validate_corpus(corpus)
    if not report["conforms"]:
        preview = "; ".join(issue["code"] for issue in report["issues"][:8])
        raise ValueError(f"MCKF contract validation failed: {preview}")
    return report


def validate_rdf_graph(triples_path: Path, shapes_path: Path) -> dict[str, Any]:
    """Run SHACL when the semantic-web dependencies are installed."""
    try:
        from pyshacl import validate  # type: ignore
        from rdflib import Graph  # type: ignore
    except ImportError as exc:
        return {
            "status": "dependency_missing",
            "conforms": False,
            "message": str(exc),
        }

    data_graph = Graph()
    data_graph.parse(triples_path, format="nt")
    shapes_graph = Graph()
    shapes_graph.parse(shapes_path, format="turtle")
    conforms, results_graph, results_text = validate(
        data_graph=data_graph,
        shacl_graph=shapes_graph,
        inference="rdfs",
        abort_on_first=False,
        allow_infos=True,
        allow_warnings=True,
    )
    return {
        "status": "ok" if conforms else "fail",
        "conforms": bool(conforms),
        "triple_count": len(data_graph),
        "result_count": len(results_graph),
        "message": str(results_text)[-4000:],
    }


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def write_json_atomic(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    temporary.replace(path)


def _unique_ids(rows: list[dict[str, Any]], field: str, kind: str, issues: list[ContractIssue]) -> set[str]:
    seen: set[str] = set()
    for row in rows:
        entity_id = str(row.get(field, "") or "")
        if not entity_id:
            _add(issues, "error", f"missing_{kind}_id", kind, f"{kind} kimliği boş.")
            continue
        if entity_id in seen:
            _add(issues, "error", f"duplicate_{kind}_id", entity_id, f"Aynı {kind} kimliği birden fazla kullanılmış.")
        seen.add(entity_id)
    return seen


def _id_prefix(document_id: str) -> str:
    import re

    return re.sub(r"\W+", "_", document_id.lower(), flags=re.UNICODE).strip("_") + "__"


def _validate_decision_contract(
    contract: dict[str, Any],
    document_ids: set[str],
    article_refs: set[tuple[str, str]],
    issues: list[ContractIssue],
) -> None:
    contract_id = str(contract.get("contract_id", "") or "")
    for field in ("version", "title", "decision_type", "input_schema", "rules", "review_status"):
        if not contract.get(field):
            _add(issues, "error", f"missing_decision_{field}", contract_id, f"Karar sözleşmesinde {field} alanı yok.")
    if contract.get("review_status") != "expert_approved":
        _add(issues, "error", "decision_contract_not_expert_approved", contract_id, "Çalıştırılabilir karar sözleşmesi expert_approved olmalıdır.")
    if not contract.get("query_aliases") or not contract.get("fact_extractors"):
        _add(issues, "error", "decision_contract_not_queryable", contract_id, "Karar sözleşmesi sorgu eş anlamları ve olgu çıkarıcıları taşımalıdır.")

    source_refs = contract.get("source_refs", []) or []
    if not source_refs:
        _add(issues, "error", "missing_decision_source_refs", contract_id, "Karar sözleşmesi en az bir kaynak hükme bağlanmalıdır.")
    for source in source_refs:
        document_id = str(source.get("document_id", "") or "")
        article_id = str(source.get("article_id", "") or "")
        if document_id not in document_ids:
            _add(issues, "error", "unknown_decision_document", contract_id, f"Bilinmeyen karar kaynağı: {document_id}.")
        if article_id and (document_id, article_id) not in article_refs:
            _add(issues, "error", "unknown_decision_article", contract_id, f"Kaynak hüküm corpus içinde bulunamadı: {document_id}::{article_id}.")

    fields = ((contract.get("input_schema", {}) or {}).get("fields", {}) or {})
    if not isinstance(fields, dict) or not fields:
        _add(issues, "error", "empty_decision_input_schema", contract_id, "Karar sözleşmesinin olay veri şeması boş olamaz.")
        fields = {}

    rules = contract.get("rules", []) or []
    rule_ids: set[str] = set()
    for rule in rules:
        rule_id = str(rule.get("rule_id", "") or "")
        if not rule_id:
            _add(issues, "error", "missing_decision_rule_id", contract_id, "Karar kuralı kimliği boş.")
        elif rule_id in rule_ids:
            _add(issues, "error", "duplicate_decision_rule_id", contract_id, f"Kural kimliği tekrar ediyor: {rule_id}.")
        rule_ids.add(rule_id)
        if "outcome" not in rule:
            _add(issues, "error", "missing_decision_outcome", rule_id or contract_id, "Karar kuralında sonuç yok.")
        if not isinstance(rule.get("priority", 0), int):
            _add(issues, "error", "invalid_decision_priority", rule_id or contract_id, "Kural önceliği tam sayı olmalıdır.")
        conditions = rule.get("when", {}) or {}
        if not isinstance(conditions, dict) or not ({"all", "any"} & set(conditions)):
            _add(issues, "error", "invalid_decision_conditions", rule_id or contract_id, "Kural koşulları all veya any listesi taşımalıdır.")
            continue
        for mode in ("all", "any"):
            for condition in conditions.get(mode, []) or []:
                fact = str(condition.get("fact", "") or "")
                operator = str(condition.get("operator", "eq") or "eq")
                if fact not in fields:
                    _add(issues, "error", "unknown_decision_fact", rule_id or contract_id, f"Koşul şemada olmayan olguyu kullanıyor: {fact}.")
                if operator not in DECISION_OPERATORS:
                    _add(issues, "error", "unknown_decision_operator", rule_id or contract_id, f"Desteklenmeyen operatör: {operator}.")

    for requirement in contract.get("judgment_requirements", []) or []:
        fact = str(requirement.get("fact", "") or "")
        if fact not in fields:
            _add(issues, "error", "unknown_judgment_fact", contract_id, f"Değerlendirme alanı şemada yok: {fact}.")
        if not requirement.get("reason"):
            _add(issues, "error", "missing_judgment_reason", contract_id, "İnsan değerlendirmesi gereksinimi gerekçe taşımalıdır.")
        _add(issues, "warning", "requires_legal_judgment", contract_id, str(requirement.get("reason", "")))


def _add(issues: list[ContractIssue], severity: str, code: str, entity_id: str, message: str) -> None:
    issues.append(ContractIssue(severity, code, entity_id, message))