from __future__ import annotations import argparse import json import re from collections import Counter from pathlib import Path from typing import Any, Iterable from urllib.parse import quote from normative_roles import extract_question_frame from utils import normalize_for_search, query_terms PILOT_SCHEMA = "MCKF-SemanticGraphRuntime-v1.2" BASE_URI = "https://mitranlil.ai/mckf/" RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" MVP_CORE = [ "document_ingestion", "semantic_normative_unit_parser", "mckf_builder", "hybrid_retrieval", "evidence_sufficiency_gate", "source_locked_generation", "citation_renderer", "deterministic_decision_runtime", "canonical_query_planner", "expert_reviewed_semantic_addresses", "shared_normative_status_protocol", "temporal_scope_guard", ] POST_PILOT = [ "advanced_rule_authoring", "wide_graph_traversal", "exception_precedence_inference", "strict_rdf_compliance", ] QUERY_INTENTS = { "obligation_lookup", "permission_lookup", "procedure_lookup", "deadline_lookup", "exception_lookup", "definition_lookup", "unknown", } CONCEPT_TYPES = { "obligation_rule", "permission_rule", "procedure_rule", "deadline_rule", "exception_rule", "definition_rule", } UNIT_TYPES = { "obligation", "permission", "prohibition", "condition", "exception", "procedure", "deadline", "definition", } RDF_CLASS_BY_CONCEPT_TYPE = { "obligation_rule": "ObligationRule", "permission_rule": "PermissionRule", "procedure_rule": "ProcedureRule", "deadline_rule": "DeadlineRule", "exception_rule": "ExceptionRule", "definition_rule": "DefinitionRule", } RDF_CLASS_BY_UNIT_TYPE = { "obligation": "ObligationUnit", "permission": "PermissionUnit", "prohibition": "ProhibitionUnit", "condition": "ConditionUnit", "exception": "ExceptionUnit", "procedure": "ProcedureUnit", "deadline": "DeadlineUnit", "definition": "DefinitionUnit", } def pilot_architecture_metadata() -> dict[str, Any]: return { "schema": PILOT_SCHEMA, "status": "canonical_runtime_graph_and_contract_validation_enabled", "goal": "plan_the_request_resolve_reviewed_normative_units_decide_or_fail_closed_with_sources", "mvp_core": MVP_CORE, "post_pilot": POST_PILOT, "runtime_contract": { "source_lock": True, "citation_required": True, "external_rdfox_required": False, "runtime_graph_backend": "RDFLib SPARQL with RDFox-compatible exports", "shacl_validation_required": True, "rdfox_export_paths": [ "data/mckf/rdfox_pilot_manifest.json", "data/mckf/rdfox_pilot_triples.nt", "data/mckf/rdfox_pilot_rules.dlog", ], }, } def analyze_pilot_query(question: str) -> dict[str, Any]: normalized = normalize_for_search(question or "") frame = extract_question_frame(question or "") intent = _intent_from_query(normalized, frame) terms = sorted(query_terms(question or "")) temporal_signals = [ term for term in ("saat", "gun", "ay", "yil", "sure", "tarih", "takvim", "hafta", "donem", "kredi") if term in normalized ] return { "raw_query": question or "", "normalized_query": normalized, "intent": intent, "actors": frame.get("actor", []) or [], "actions": frame.get("action", []) or [], "concepts": terms[:12], "temporal_signals": temporal_signals, "question_frame": frame, } def micro_rule_score( query_analysis: dict[str, Any] | None, evidence: dict[str, Any], parent_clause: dict[str, Any] | None = None, concept: dict[str, Any] | None = None, ) -> tuple[float, list[str]]: if not query_analysis: return 0.5, [] intent = query_analysis.get("intent", "unknown") frame = evidence.get("semantic_frame", {}) or {} roles = set(evidence.get("semantic_roles", []) or []) if parent_clause: roles.update(parent_clause.get("semantic_roles", []) or []) if concept: roles.update(concept.get("semantic_roles", []) or []) hits: list[str] = [] score = 0.0 if intent == "deadline_lookup" and (frame.get("temporal_constraint") or "sure" in roles or _has_deadline_signal(evidence)): score += 0.75 hits.append("deadline_priority") if intent == "procedure_lookup" and ("usul" in roles or _has_procedure_signal(evidence)): score += 0.7 hits.append("procedure_chain") if intent == "exception_lookup" and (frame.get("exception") or "istisna" in roles): score += 0.75 hits.append("exception_warning") if intent == "definition_lookup" and (frame.get("definition_term") or "tanim" in roles): score += 0.7 hits.append("definition_attach") if _has_cross_reference(evidence.get("source_text", "")): score += 0.25 hits.append("cross_reference_follow") actor_overlap = _overlap(query_analysis.get("actors", []), frame.get("actor", []) or []) action_overlap = _overlap(query_analysis.get("actions", []), frame.get("action", []) or []) score += 0.25 * actor_overlap score += 0.20 * action_overlap return min(1.0, score), hits def enrich_selected_evidence( query_analysis: dict[str, Any] | None, evidence_spans: list[dict[str, Any]], ) -> list[dict[str, Any]]: if not query_analysis or not evidence_spans: return evidence_spans if any("structural_context" in (item.get("_retrieval_channels", []) or []) for item in evidence_spans): return evidence_spans intent = query_analysis.get("intent", "unknown") if intent not in {"deadline_lookup", "procedure_lookup", "exception_lookup", "definition_lookup"}: return evidence_spans def priority(evidence: dict[str, Any]) -> tuple[int, str, str, int]: frame = evidence.get("semantic_frame", {}) or {} roles = set(evidence.get("semantic_roles", []) or []) score = 0 if intent == "deadline_lookup" and (frame.get("temporal_constraint") or "sure" in roles or _has_deadline_signal(evidence)): score -= 2 if intent == "procedure_lookup" and ("usul" in roles or _has_procedure_signal(evidence)): score -= 2 if intent == "exception_lookup" and (frame.get("exception") or "istisna" in roles): score -= 2 if intent == "definition_lookup" and (frame.get("definition_term") or "tanim" in roles): score -= 2 span_start = int((evidence.get("source_span", {}) or {}).get("char_start", 0) or 0) return ( score, evidence.get("document_id", ""), evidence.get("article_id", ""), span_start, ) return sorted(evidence_spans, key=priority) def pilot_answer_warnings( query_analysis: dict[str, Any] | None, evidence_spans: list[dict[str, Any]], ) -> list[str]: if not query_analysis or not evidence_spans: return [] warnings: list[str] = [] if any(_has_exception_signal(evidence) for evidence in evidence_spans): warnings.append("Seçili kaynakta istisna veya saklı hüküm sinyali var; cevap bu sınırlama ile birlikte okunmalıdır.") if query_analysis.get("intent") == "deadline_lookup" and not any( (e.get("semantic_frame", {}) or {}).get("temporal_constraint") for e in evidence_spans ): warnings.append("Seçili kaynak açık bir süre ifadesi taşımıyorsa süre çıkarımı yapılmadı.") return _dedupe(warnings) def apply_pilot_answer_micro_rules( answer: str, query_analysis: dict[str, Any] | None, evidence_spans: list[dict[str, Any]], ) -> str: warnings = pilot_answer_warnings(query_analysis, evidence_spans) if not warnings: return answer warning_text = " ".join(warnings) if warning_text in answer: return answer return f"{answer.rstrip()}\n\nNot: {warning_text}" def build_pilot_manifest(ontology: dict[str, Any]) -> dict[str, Any]: evidence_spans = ontology.get("evidence_spans", []) or _flatten_evidence_spans(ontology.get("clauses", []) or []) concepts = ontology.get("concepts", []) or [] unit_type_counts = Counter(_unit_type_from_evidence(evidence) for evidence in evidence_spans) concept_type_counts = Counter(_concept_type(concept) for concept in concepts) return { **pilot_architecture_metadata(), "data_contract": { "MVPConcept": { "source": "data/mckf/corpus_mckf_ontology.json::concepts", "fields": [ "concept_id", "document_id", "article_id", "concept_type", "title", "aliases", "source_unit_ids", "actors", "actions", "conditions", "exceptions", "temporal_constraints", "related_articles", "source_text", "source_policy", ], }, "MVPSemanticNormativeUnit": { "source": "data/mckf/corpus_mckf_ontology.json::evidence_spans", "fields": [ "unit_id", "document_id", "article_id", "clause_path", "raw_text", "normalized_text", "unit_type", "actors", "actions", "conditions", "exceptions", "temporal_constraints", "cross_references", "extraction_confidence", ], }, "MCKFDecisionContract": { "source": "data/mckf/corpus_mckf_ontology.json::decision_contracts", "fields": ["contract_id", "version", "decision_type", "source_refs", "input_schema", "rules", "judgment_requirements", "review_status"], }, }, "retrieval_channels": ["bm25", "lsa_dense", "semantic_frame", "sparql_graph", "exact_reference"], "fusion": "weighted_reciprocal_rank_fusion", "micro_rules": [ { "rule_id": "deadline_priority", "effect": "deadline units receive a retrieval bonus for duration/date questions", }, { "rule_id": "procedure_chain", "effect": "procedure evidence is ordered by source span when procedure intent is detected", }, { "rule_id": "exception_warning", "effect": "selected exception evidence adds a visible caution to the answer", }, { "rule_id": "cross_reference_follow", "effect": "cross-document/article references are exported as RDF edges and scored as retrieval hints", }, { "rule_id": "definition_attach", "effect": "definition units are boosted for term-definition questions", }, ], "stats": { "document_count": len(ontology.get("documents", []) or []), "concept_count": len(concepts), "semantic_normative_unit_count": len(evidence_spans), "unit_type_counts": dict(sorted(unit_type_counts.items())), "concept_type_counts": dict(sorted(concept_type_counts.items())), "cross_document_edge_count": len(ontology.get("cross_document_edges", []) or []), "decision_contract_count": len(ontology.get("decision_contracts", []) or []), }, } def write_rdfox_pilot_outputs(ontology: dict[str, Any], output_dir: str | Path) -> dict[str, Any]: out_dir = Path(output_dir) out_dir.mkdir(parents=True, exist_ok=True) manifest_path = out_dir / "rdfox_pilot_manifest.json" triples_path = out_dir / "rdfox_pilot_triples.nt" rules_path = out_dir / "rdfox_pilot_rules.dlog" triple_count = _write_triples(ontology, triples_path) rules_text = render_rdfox_pilot_rules() rules_path.write_text(rules_text, encoding="utf-8") manifest = build_pilot_manifest(ontology) manifest["outputs"] = { "manifest": str(manifest_path.as_posix()), "triples": str(triples_path.as_posix()), "rules": str(rules_path.as_posix()), "triple_count": triple_count, } manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") return manifest def render_rdfox_pilot_rules() -> str: return """# MCKF RDFox pilot micro-rules. # These rules are a portable MVP contract for RDFox/Datalog migration. # Runtime uses equivalent Python micro-rules in rdfox_pilot.py. PREFIX mckf: PREFIX rdf: # deadline_priority mckf:retrievalHint[?unit, "deadline_priority"] :- ?unit rdf:type mckf:DeadlineUnit . # procedure_chain mckf:retrievalHint[?unit, "procedure_chain"] :- ?unit rdf:type mckf:ProcedureUnit . # exception_warning mckf:answerWarning[?unit, "exception_warning"] :- ?unit rdf:type mckf:ExceptionUnit . # cross_reference_follow mckf:followReference[?unit, ?target] :- ?unit mckf:crossReferences ?target . # definition_attach mckf:retrievalHint[?unit, "definition_attach"] :- ?unit rdf:type mckf:DefinitionUnit . """ def _write_triples(ontology: dict[str, Any], triples_path: Path) -> int: count = 0 with triples_path.open("w", encoding="utf-8") as file: for triple in _iter_triples(ontology): file.write(triple + "\n") count += 1 return count def _iter_triples(ontology: dict[str, Any]) -> Iterable[str]: for document in ontology.get("documents", []) or []: doc_uri = _uri("document", document.get("document_id", "")) yield _triple(doc_uri, _uri_pred("schema"), _literal("Document")) yield _triple(doc_uri, RDF_TYPE, _uri_class("Document")) yield _triple(doc_uri, _uri_pred("documentId"), _literal(document.get("document_id", ""))) yield _triple(doc_uri, _uri_pred("title"), _literal(document.get("title", ""))) yield _triple(doc_uri, _uri_pred("versionId"), _literal(document.get("version_id", ""))) yield _triple(doc_uri, _uri_pred("sourceSha256"), _literal(document.get("source_sha256", ""))) yield _triple(doc_uri, _uri_pred("authorityLevel"), _literal(document.get("authority_level", ""))) yield _triple(doc_uri, _uri_pred("validityStatus"), _literal(document.get("validity_status", ""))) yield _triple(doc_uri, _uri_pred("sourceAuthority"), _literal(document.get("source_authority", ""))) yield _triple(doc_uri, _uri_pred("sourceSnapshotDate"), _literal(document.get("source_snapshot_date", ""))) yield _triple(doc_uri, _uri_pred("temporalCoverage"), _literal(document.get("temporal_coverage", ""))) yield _triple(doc_uri, _uri_pred("jurisdiction"), _literal(document.get("jurisdiction", ""))) yield _triple(doc_uri, _uri_pred("authorityRank"), _literal(str(document.get("authority_rank", "")))) for tag in document.get("domain_tags", []) or []: yield _triple(doc_uri, _uri_pred("domainTag"), _literal(tag)) concepts_by_id = {c.get("concept_id", ""): c for c in ontology.get("concepts", []) or []} for concept in concepts_by_id.values(): concept_uri = _uri("concept", concept.get("concept_id", "")) concept_type = _concept_type(concept) yield _triple(concept_uri, RDF_TYPE, _uri_class(RDF_CLASS_BY_CONCEPT_TYPE[concept_type])) yield _triple(concept_uri, _uri_pred("conceptType"), _literal(concept_type)) yield _triple(concept_uri, _uri_pred("document"), _uri("document", concept.get("document_id", ""))) yield _triple(concept_uri, _uri_pred("articleId"), _literal(concept.get("article_id", ""))) yield _triple(concept_uri, _uri_pred("title"), _literal(concept.get("title", ""))) yield _triple(concept_uri, _uri_pred("sourceLock"), _literal_bool(True)) yield _triple(concept_uri, _uri_pred("citationRequired"), _literal_bool(True)) metadata = concept.get("normative_metadata", {}) or {} yield _triple(concept_uri, _uri_pred("reviewStatus"), _literal(metadata.get("review_status", ""))) if metadata.get("approved_summary"): yield _triple(concept_uri, _uri_pred("approvedSummary"), _literal(metadata.get("approved_summary", ""))) for alias in concept.get("aliases", []) or []: yield _triple(concept_uri, _uri_pred("alias"), _literal(alias)) for actor in concept.get("actors", []) or []: yield _triple(concept_uri, _uri_pred("hasActor"), _literal(actor)) for related_article in concept.get("related_articles", []) or []: yield _triple(concept_uri, _uri_pred("relatedArticle"), _literal(related_article)) for clause_id in concept.get("clause_ids", []) or []: yield _triple(concept_uri, _uri_pred("sourceUnit"), _uri("unit", clause_id)) clauses_by_id = {c.get("clause_id", ""): c for c in ontology.get("clauses", []) or []} evidence_spans = ontology.get("evidence_spans", []) or _flatten_evidence_spans(ontology.get("clauses", []) or []) for evidence in evidence_spans: unit_id = evidence.get("evidence_id", "") if not unit_id: continue unit_type = _unit_type_from_evidence(evidence) unit_uri = _uri("unit", unit_id) parent_clause_id = evidence.get("parent_clause_id", "") parent_clause = clauses_by_id.get(parent_clause_id, {}) concept_id = parent_clause.get("parent_concept_id", "") yield _triple(unit_uri, RDF_TYPE, _uri_class(RDF_CLASS_BY_UNIT_TYPE[unit_type])) yield _triple(unit_uri, _uri_pred("unitType"), _literal(unit_type)) yield _triple(unit_uri, _uri_pred("evidenceId"), _literal(unit_id)) yield _triple(unit_uri, _uri_pred("document"), _uri("document", evidence.get("document_id", ""))) yield _triple(unit_uri, _uri_pred("articleId"), _literal(evidence.get("article_id", ""))) yield _triple(unit_uri, _uri_pred("parentClause"), _uri("unit", parent_clause_id)) yield _triple(unit_uri, _uri_pred("sourceText"), _literal(evidence.get("source_text", ""))) if concept_id: yield _triple(unit_uri, _uri_pred("belongsToConcept"), _uri("concept", concept_id)) for role in evidence.get("semantic_roles", []) or []: yield _triple(unit_uri, _uri_pred("semanticRole"), _literal(role)) frame = evidence.get("semantic_frame", {}) or {} for actor in frame.get("actor", []) or []: yield _triple(unit_uri, _uri_pred("hasActor"), _literal(actor)) for action in frame.get("action", []) or []: yield _triple(unit_uri, _uri_pred("hasAction"), _literal(action)) for obj in frame.get("object", []) or []: yield _triple(unit_uri, _uri_pred("hasObject"), _literal(obj)) for authority in frame.get("competent_authority", []) or []: yield _triple(unit_uri, _uri_pred("hasCompetentAuthority"), _literal(authority)) for beneficiary in frame.get("beneficiary", []) or []: yield _triple(unit_uri, _uri_pred("hasBeneficiary"), _literal(beneficiary)) for remedy in frame.get("remedy", []) or []: yield _triple(unit_uri, _uri_pred("hasRemedy"), _literal(remedy)) for sanction in frame.get("sanction", []) or []: yield _triple(unit_uri, _uri_pred("hasSanction"), _literal(sanction)) yield _triple(unit_uri, _uri_pred("modality"), _literal(frame.get("deontic_modality") or frame.get("modality", ""))) yield _triple(unit_uri, _uri_pred("normCategory"), _literal(frame.get("norm_category", ""))) yield _triple(unit_uri, _uri_pred("extractionConfidence"), _literal(str(frame.get("confidence", 0.0)))) for condition in _compact_values(frame.get("condition", []) or []): yield _triple(unit_uri, _uri_pred("hasCondition"), _literal(condition)) for exception in _compact_values(frame.get("exception", []) or []): yield _triple(unit_uri, _uri_pred("hasException"), _literal(exception)) for temporal in _compact_values(frame.get("temporal_constraint", []) or []): yield _triple(unit_uri, _uri_pred("hasTemporalConstraint"), _literal(temporal)) for reference in _cross_references(evidence.get("source_text", "")): yield _triple(unit_uri, _uri_pred("crossReferences"), _literal(reference["ref_text"])) for edge in ontology.get("cross_document_edges", []) or []: edge_uri = _uri("edge", edge.get("edge_id", "")) yield _triple(edge_uri, RDF_TYPE, _uri_class("CrossDocumentEdge")) yield _triple(edge_uri, _uri_pred("relationType"), _literal(edge.get("relation_type", ""))) yield _triple(edge_uri, _uri_pred("sourceDocument"), _uri("document", edge.get("source_document_id", ""))) yield _triple(edge_uri, _uri_pred("sourceArticle"), _literal(edge.get("source_article_id", ""))) yield _triple(edge_uri, _uri_pred("targetDocument"), _uri("document", edge.get("target_document_id", ""))) yield _triple(edge_uri, _uri_pred("targetArticle"), _literal(edge.get("target_article_id", ""))) yield _triple(edge_uri, _uri_pred("reviewStatus"), _literal(edge.get("review_status", "derived_from_source_reference"))) if edge.get("approved_interpretation"): yield _triple(edge_uri, _uri_pred("approvedInterpretation"), _literal(edge.get("approved_interpretation", ""))) for contract in ontology.get("decision_contracts", []) or []: contract_id = str(contract.get("contract_id", "") or "") contract_uri = _uri("decision-contract", contract_id) yield _triple(contract_uri, RDF_TYPE, _uri_class("DecisionContract")) yield _triple(contract_uri, _uri_pred("contractId"), _literal(contract_id)) yield _triple(contract_uri, _uri_pred("decisionType"), _literal(contract.get("decision_type", ""))) yield _triple(contract_uri, _uri_pred("version"), _literal(contract.get("version", ""))) yield _triple(contract_uri, _uri_pred("reviewStatus"), _literal(contract.get("review_status", ""))) for source in contract.get("source_refs", []) or []: yield _triple(contract_uri, _uri_pred("sourceDocument"), _uri("document", source.get("document_id", ""))) yield _triple(contract_uri, _uri_pred("sourceArticle"), _literal(source.get("article_id", ""))) for requirement in contract.get("judgment_requirements", []) or []: requirement_uri = _uri("judgment-requirement", f"{contract_id}__{requirement.get('fact', '')}") yield _triple(requirement_uri, RDF_TYPE, _uri_class("JudgmentRequirement")) yield _triple(requirement_uri, _uri_pred("belongsToDecisionContract"), contract_uri) yield _triple(requirement_uri, _uri_pred("judgmentFact"), _literal(requirement.get("fact", ""))) yield _triple(requirement_uri, _uri_pred("judgmentReason"), _literal(requirement.get("reason", ""))) yield _triple(contract_uri, _uri_pred("requiresJudgment"), requirement_uri) for rule in contract.get("rules", []) or []: rule_uri = _uri("decision-rule", f"{contract_id}__{rule.get('rule_id', '')}") yield _triple(rule_uri, RDF_TYPE, _uri_class("DecisionRule")) yield _triple(rule_uri, _uri_pred("ruleId"), _literal(rule.get("rule_id", ""))) yield _triple(rule_uri, _uri_pred("belongsToDecisionContract"), contract_uri) yield _triple(rule_uri, _uri_pred("decisionOutcome"), _literal(json.dumps(rule.get("outcome"), ensure_ascii=False, sort_keys=True))) yield _triple(rule_uri, _uri_pred("priority"), _literal(str(rule.get("priority", 0)))) yield _triple(contract_uri, _uri_pred("decisionRule"), rule_uri) def _intent_from_query(normalized: str, frame: dict[str, Any]) -> str: if any(term in normalized for term in ("kac saat", "kac gun", "kac ay", "kac yil", "sure", "suresi", "hangi tarihe", "ne zamana kadar")): return "deadline_lookup" if any(term in normalized for term in ("istisna", "haric", "sakli", "ancak", "disinda")): return "exception_lookup" if any(term in normalized for term in ("nasil", "basvuru", "adim", "surec", "prosedur", "islem", "izlemeliyim")): return "procedure_lookup" if any(term in normalized for term in ("nedir", "ne demek", "tanim", "ifade eder")): return "definition_lookup" if frame.get("modality") == "permission" or any(term in normalized for term in ("hak", "izin", "mumkun", "yapabilir", "yararlanabilir")): return "permission_lookup" if frame.get("modality") == "obligation" or any(term in normalized for term in ("zorunlu", "zorunda", "gerekir", "yukumluluk", "sart")): return "obligation_lookup" return "unknown" def _concept_type(concept: dict[str, Any]) -> str: roles = set(concept.get("semantic_roles", []) or []) text = normalize_for_search(" ".join([ concept.get("title", ""), concept.get("source_text", "")[:600], ])) if "sure" in roles or any(term in text for term in ("saat", "sure", "gun", "ay", "yil", "tarih")): return "deadline_rule" if "istisna" in roles: return "exception_rule" if "tanim" in roles: return "definition_rule" if "usul" in roles or any(term in text for term in ("usul", "basvuru", "islem", "surec")): return "procedure_rule" if "yetki" in roles: return "permission_rule" return "obligation_rule" def _unit_type_from_evidence(evidence: dict[str, Any]) -> str: frame = evidence.get("semantic_frame", {}) or {} roles = set(evidence.get("semantic_roles", []) or []) modality = frame.get("modality", "") if frame.get("temporal_constraint") or "sure" in roles or _has_deadline_signal(evidence): return "deadline" if frame.get("exception") or "istisna" in roles: return "exception" if "sart" in roles or frame.get("condition"): return "condition" if "usul" in roles or _has_procedure_signal(evidence): return "procedure" if modality == "definition" or "tanim" in roles: return "definition" if modality == "prohibition": return "prohibition" if modality in {"permission", "power"} or "yetki" in roles: return "permission" return "obligation" def _has_procedure_signal(evidence: dict[str, Any]) -> bool: text = normalize_for_search(evidence.get("source_text", "")) return any(term in text for term in ("basvuru", "usul", "esas", "islem", "surec", "oneri", "karar")) def _has_deadline_signal(evidence: dict[str, Any]) -> bool: text = normalize_for_search(evidence.get("source_text", "")) return bool( re.search(r"\b\d+\s*(saat|gun|ay|yil|hafta|donem)\b", text) or any(term in text for term in ("azami", "en cok", "en az", "sure", "suresi", "tarih")) ) def _has_exception_signal(evidence: dict[str, Any]) -> bool: frame = evidence.get("semantic_frame", {}) or {} roles = set(evidence.get("semantic_roles", []) or []) text = normalize_for_search(evidence.get("source_text", "")) return bool(frame.get("exception") or "istisna" in roles or any(term in text for term in ("ancak", "haric", "sakli", "istisna"))) def _has_cross_reference(text: str) -> bool: return bool(_cross_references(text)) def _cross_references(text: str) -> list[dict[str, str]]: refs = [] for match in re.finditer(r"\b(?:\d{3,5}\s+sayılı\s+)?(?:Kanunun\s+)?(?:Ek\s+)?Madde\s+\d+[A-ZÇĞİÖŞÜçğıöşü/]*\b", text or "", flags=re.IGNORECASE): refs.append({"ref_text": " ".join(match.group(0).split()), "resolution_status": "unresolved"}) for match in re.finditer(r"\b\d{4}\s+sayılı\s+Kanun\b", text or "", flags=re.IGNORECASE): refs.append({"ref_text": " ".join(match.group(0).split()), "resolution_status": "unresolved"}) return _dedupe_dicts(refs, "ref_text")[:8] def _overlap(left: list[Any], right: list[Any]) -> float: if not left: return 0.5 if not right: return 0.0 left_norm = {normalize_for_search(str(item)) for item in left if item} right_norm = {normalize_for_search(str(item)) for item in right if item} if not left_norm: return 0.5 hits = 0 for item in left_norm: if item in right_norm or any(item in candidate or candidate in item for candidate in right_norm): hits += 1 return hits / max(1, len(left_norm)) def _compact_values(values: list[Any], limit: int = 180) -> list[str]: compacted = [] for value in values: text = " ".join(str(value or "").split()) if not text: continue if len(text) > limit: text = text[:limit].rsplit(" ", 1)[0] + "..." compacted.append(text) return _dedupe(compacted) def _flatten_evidence_spans(clauses: list[dict[str, Any]]) -> list[dict[str, Any]]: flattened = [] for clause in clauses: flattened.extend(clause.get("evidence_spans", []) or []) return flattened def _uri(kind: str, identifier: str) -> str: value = quote(str(identifier or "unknown"), safe="") return f"{BASE_URI}{kind}/{value}" def _uri_pred(name: str) -> str: return f"{BASE_URI}{name}" def _uri_class(name: str) -> str: return f"{BASE_URI}{name}" def _literal(value: Any) -> str: text = " ".join(str(value or "").split()) text = text.replace("\\", "\\\\").replace('"', '\\"') return f'"{text}"' def _literal_bool(value: bool) -> str: return f'"{str(bool(value)).lower()}"^^' def _triple(subject: str, predicate: str, obj: str) -> str: object_term = obj if obj.startswith("<") or obj.startswith('"') else f"<{obj}>" return f"<{subject}> <{predicate}> {object_term} ." def _dedupe(values: list[str]) -> list[str]: seen = set() result = [] for value in values: if value in seen: continue seen.add(value) result.append(value) return result def _dedupe_dicts(values: list[dict[str, str]], key: str) -> list[dict[str, str]]: seen = set() result = [] for value in values: marker = value.get(key, "") if marker in seen: continue seen.add(marker) result.append(value) return result def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Build RDFox-compatible MCKF pilot artifacts") parser.add_argument("--ontology", default="data/mckf/corpus_mckf_ontology.json") parser.add_argument("--output-dir", default="data/mckf") args = parser.parse_args(argv) ontology_path = Path(args.ontology) if not ontology_path.exists(): print(f"FAIL: ontology not found: {ontology_path}") return 1 ontology = json.loads(ontology_path.read_text(encoding="utf-8")) manifest = write_rdfox_pilot_outputs(ontology, args.output_dir) outputs = manifest.get("outputs", {}) print( "RDFox pilot artifacts written " f"units={manifest.get('stats', {}).get('semantic_normative_unit_count', 0)} " f"triples={outputs.get('triple_count', 0)}" ) return 0 if __name__ == "__main__": raise SystemExit(main())