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))