| 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: <https://mitranlil.ai/mckf/>
|
| PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
|
|
|
| # 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()}"^^<http://www.w3.org/2001/XMLSchema#boolean>'
|
|
|
|
|
| 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())
|
|
|
