from __future__ import annotations import re from dataclasses import asdict, dataclass, field from datetime import date from typing import Any from normative_status import NormativeStatus from utils import TOPIC_STOPWORDS, normalize_for_search, search_terms_match from enum import Enum class RequestMode(str, Enum): KNOWLEDGE = "KNOWLEDGE" INVENTORY = "INVENTORY" COMPARISON = "COMPARISON" CASE_ASSESSMENT = "CASE_ASSESSMENT" DIRECT_SOURCE = "DIRECT_SOURCE" SOURCE_IDENTITY = "SOURCE_IDENTITY" @dataclass(frozen=True) class RuntimePlan: status: str mode: str question: str scopes: tuple[tuple[str, str], ...] = () confidence: float = 0.0 reasons: tuple[str, ...] = () decision_type: str = "" facts: dict[str, Any] = field(default_factory=dict) missing_facts: tuple[str, ...] = () as_of_date: str = "" topic: str = "" def to_dict(self) -> dict[str, Any]: return asdict(self) class CanonicalNormativeRuntime: """Corpus-compiled query planning and deterministic presentation layer. The runtime contains no statute/article routing table. It compiles the semantic addresses, expert-approved summaries, topic memberships and decision contracts published in the active MCKF package. Rebuilding a package is therefore sufficient to teach routing to the application. """ def __init__(self, corpus: dict[str, Any] | None = None) -> None: corpus = corpus or {} self.build_id = str(corpus.get("build_id", "") or "") self.documents = { str(item.get("document_id", "")): dict(item) for item in corpus.get("documents", []) or [] if item.get("document_id") } self.concepts: dict[tuple[str, str], dict[str, Any]] = {} for concept in corpus.get("concepts", []) or []: scope = ( str(concept.get("document_id", "") or ""), str(concept.get("article_id", "") or ""), ) if all(scope) and scope not in self.concepts: self.concepts[scope] = concept # Compile the operation vocabulary from the published package. The # runtime can then distinguish the same actor under different # institutional operations without a statute-specific routing table. self.operation_terms: set[str] = set() for concept in self.concepts.values(): metadata = concept.get("normative_metadata", {}) or {} for operation in metadata.get("legal_operations", []) or []: terms = [ term for term in _normative_text(str(operation)).split() if len(term) >= 3 and term not in TOPIC_STOPWORDS and not term.isdigit() ] if terms: self.operation_terms.add(_operation_key(terms[-1])) self.edges = [dict(item) for item in corpus.get("cross_document_edges", []) or []] self.contracts = [dict(item) for item in corpus.get("decision_contracts", []) or []] self.curated_scopes = { scope for scope, concept in self.concepts.items() if self._curated_and_source_valid(concept) } def plan( self, question: str, constrained_scopes: list[tuple[str, str]] | None = None, ) -> RuntimePlan: raw = (question or "").strip() normalized = normalize_for_search(raw) if not normalized: return RuntimePlan(NormativeStatus.UNKNOWN.value, RequestMode.KNOWLEDGE.value, raw) temporal = self._temporal_request(raw) if temporal and not self._supports_date(temporal): return RuntimePlan( NormativeStatus.OUT_OF_SCOPE.value, RequestMode.KNOWLEDGE.value, raw, reasons=("historical_version_not_published",), as_of_date=temporal, ) selected_scopes = [scope for scope in (constrained_scopes or []) if scope in self.concepts] direct_scopes = self._explicit_scopes(raw) if direct_scopes and self._direct_source_intent(normalized): return RuntimePlan( NormativeStatus.ANSWERED.value, RequestMode.DIRECT_SOURCE.value, raw, scopes=tuple(direct_scopes), confidence=1.0, reasons=("explicit_document_and_provision",), as_of_date=temporal, ) ambiguous_article_scopes = self._ambiguous_bare_article_scopes(raw) if ambiguous_article_scopes: return RuntimePlan( NormativeStatus.UNKNOWN.value, RequestMode.KNOWLEDGE.value, raw, scopes=tuple(ambiguous_article_scopes), reasons=("ambiguous_article_reference",), as_of_date=temporal, ) decision = self._decision_plan(raw, temporal) if decision is not None: return decision if selected_scopes: return RuntimePlan( NormativeStatus.ANSWERED.value, RequestMode.KNOWLEDGE.value, raw, scopes=tuple(selected_scopes), confidence=1.0, reasons=("user_selected_canonical_scope",), as_of_date=temporal, ) if len(direct_scopes) > 1 and self._comparison_intent(normalized, direct_scopes): requested = set(direct_scopes) exact_relation = next( ( edge for edge in self.edges if str(edge.get("review_status", "")) in {"human_reviewed", "expert_approved"} and { (str(edge.get("source_document_id", "")), str(edge.get("source_article_id", ""))), (str(edge.get("target_document_id", "")), str(edge.get("target_article_id", ""))), } == requested ), None, ) reason = ( f"approved_normative_relation:{exact_relation.get('edge_id', '')}" if exact_relation else "explicit_multi_scope_comparison" ) return RuntimePlan( NormativeStatus.ANSWERED.value, RequestMode.COMPARISON.value, raw, scopes=tuple(direct_scopes), confidence=1.0, reasons=(reason,), as_of_date=temporal, ) relation_edges = self._matching_relation_edges(raw) if relation_edges and self._comparison_intent(normalized, direct_scopes): edge = relation_edges[0] relation_scopes = [ (str(edge.get("source_document_id", "")), str(edge.get("source_article_id", ""))), (str(edge.get("target_document_id", "")), str(edge.get("target_article_id", ""))), ] relation_scopes = [scope for scope in relation_scopes if scope in self.concepts] return RuntimePlan( NormativeStatus.ANSWERED.value, RequestMode.COMPARISON.value, raw, scopes=tuple(relation_scopes), confidence=1.0, reasons=(f"approved_normative_relation:{edge.get('edge_id', '')}",), as_of_date=temporal, ) ranked = self.rank_scopes(raw, explicit_scopes=direct_scopes) inventory = self._inventory_intent(normalized) comparison = self._comparison_intent(normalized, direct_scopes) if not direct_scopes and self._source_identity_intent(normalized) and ranked: top_score, top_scope = ranked[0] runner_up = ranked[1][0] if len(ranked) > 1 else 0.0 if top_score >= 0.90 and top_score - runner_up >= 0.10: return RuntimePlan( NormativeStatus.ANSWERED.value, RequestMode.SOURCE_IDENTITY.value, raw, scopes=(top_scope,), confidence=round(top_score, 4), reasons=("canonical_heading_identity",), as_of_date=temporal, ) if not direct_scopes and self._direct_source_intent(normalized) and ranked: top_score, top_scope = ranked[0] runner_up = ranked[1][0] if len(ranked) > 1 else 0.0 if top_scope in self.curated_scopes and top_score >= 0.72 and top_score - runner_up >= 0.12: return RuntimePlan( NormativeStatus.ANSWERED.value, RequestMode.DIRECT_SOURCE.value, raw, scopes=(top_scope,), confidence=round(top_score, 4), reasons=("canonical_source_scope",), as_of_date=temporal, ) if self._unresolved_generic_actor(normalized): actor_scopes = self._generic_actor_scopes(normalized) return RuntimePlan( NormativeStatus.UNKNOWN.value, RequestMode.KNOWLEDGE.value, raw, scopes=tuple(actor_scopes), confidence=ranked[0][0] if ranked else 0.0, reasons=("ambiguous_institutional_actor",), as_of_date=temporal, ) # A topic name on its own is not a request to select the highest-ranked # provision. Preserve the distinction between a corpus inventory # ("hangi maddeler?") and an underspecified topic probe ("ne var?"). # The latter must continue through the evidence/clarification layer so # the user can identify the intended institutional relation. if self._broad_topic_intent(normalized) and not inventory and not comparison and not direct_scopes: topic_scopes = self._inventory_scopes(raw, ranked) return RuntimePlan( NormativeStatus.UNKNOWN.value, RequestMode.KNOWLEDGE.value, raw, scopes=tuple(topic_scopes), confidence=ranked[0][0] if ranked else 0.0, reasons=("missing_normative_relation",), as_of_date=temporal, topic=self._best_topic(raw, topic_scopes) if topic_scopes else "", ) if direct_scopes and comparison: scopes = list(direct_scopes) for scope in self._comparison_scopes(ranked): if scope not in scopes: scopes.append(scope) if len(scopes) >= 2: break confidence = min((score for score, scope in ranked if scope in scopes), default=1.0) elif direct_scopes: scopes = direct_scopes confidence = 1.0 elif inventory: scopes = self._inventory_scopes(raw, ranked) confidence = ranked[0][0] if ranked else 0.0 elif comparison: scopes = self._comparison_scopes(ranked) confidence = min((score for score, _scope in ranked[:2]), default=0.0) else: scopes = [scope for score, scope in ranked[:1] if score >= 0.38] confidence = ranked[0][0] if ranked else 0.0 if not scopes: return RuntimePlan( NormativeStatus.UNKNOWN.value, RequestMode.INVENTORY.value if inventory else RequestMode.KNOWLEDGE.value, raw, confidence=confidence, reasons=("no_canonical_scope_above_threshold",), as_of_date=temporal, ) mode = ( RequestMode.INVENTORY.value if inventory else RequestMode.COMPARISON.value if comparison and len(scopes) > 1 else RequestMode.KNOWLEDGE.value ) topic = self._best_topic(raw, scopes) if inventory else "" unresolved_conflicts = self._unresolved_conflicts(scopes) if unresolved_conflicts: return RuntimePlan( NormativeStatus.CONFLICT.value, mode, raw, scopes=tuple(scopes), confidence=round(confidence, 4), reasons=tuple(f"unresolved_conflict:{edge.get('edge_id', '')}" for edge in unresolved_conflicts), as_of_date=temporal, topic=topic, ) final_reasons = ["canonical_semantic_address"] if ( mode == RequestMode.KNOWLEDGE.value and len(scopes) == 1 and self._requires_relation_validation(raw, scopes[0]) ): final_reasons.append("requires_relation_validation") return RuntimePlan( NormativeStatus.ANSWERED.value, mode, raw, scopes=tuple(scopes), confidence=round(confidence, 4), reasons=tuple(final_reasons), as_of_date=temporal, topic=topic, ) def rank_scopes( self, question: str, explicit_scopes: list[tuple[str, str]] | None = None, ) -> list[tuple[float, tuple[str, str]]]: normalized = normalize_for_search(question) explicit_documents = set(self._explicit_document_ids(question)) explicit_scope_set = set(explicit_scopes or []) ranked: list[tuple[float, tuple[str, str]]] = [] for scope, concept in self.concepts.items(): if explicit_documents and scope[0] not in explicit_documents: continue metadata = concept.get("normative_metadata", {}) or {} score = self._field_overlap(normalized, metadata) # Official provision headings are part of the canonical address. # A named institution in the question (for example, a university) # must therefore outrank incidental mentions elsewhere in the # corpus. This is compiled from the published heading rather than # maintained as an application-side routing table. score = max( score, _official_heading_identity_score( normalized, str(metadata.get("display_heading", "") or concept.get("title", "")), ), _official_heading_identity_score( normalized, str(concept.get("title", "")), ), _named_institution_title_score( normalized, str(concept.get("title", "") or metadata.get("display_heading", "")), ), ) score *= self._operation_alignment(normalized, metadata) if scope in explicit_scope_set: score = 1.0 if score <= 0: continue if scope in self.curated_scopes: # Human review raises confidence in an already strong semantic # match; it must not operate as a fixed routing preference. # A broad, reviewed purpose provision would otherwise overtake # a structurally exact but not-yet-curated heading (for example # "öğrenci disiplini" -> Madde 54). score = min(1.0, score + (score * 0.12)) ranked.append((round(score, 4), scope)) ranked.sort(key=lambda item: (-item[0], item[1][0], _article_sort_key(item[1][1]))) return ranked def route(self, question: str) -> dict[str, Any]: ranked = self.rank_scopes(question, explicit_scopes=self._explicit_scopes(question)) document_scores: dict[str, float] = {doc_id: 0.0 for doc_id in self.documents} target_articles: dict[str, list[str]] = {} reasons: dict[str, list[str]] = {doc_id: [] for doc_id in self.documents} for score, (document_id, article_id) in ranked[:12]: document_scores[document_id] = max(document_scores.get(document_id, 0.0), score) if score >= 0.38: target_articles.setdefault(document_id, []).append(article_id) reasons.setdefault(document_id, []).append(f"canonical:{article_id}") explicit_documents = self._explicit_document_ids(question) normalized = normalize_for_search(question) matched_edges = self._matching_relation_edges(question) if self._comparison_intent(normalized, self._explicit_scopes(question)) else [] if matched_edges: candidate_documents = list(dict.fromkeys( str(edge.get(field, "")) for edge in matched_edges for field in ("source_document_id", "target_document_id") if edge.get(field) )) for edge in matched_edges: for document_field, article_field in ( ("source_document_id", "source_article_id"), ("target_document_id", "target_article_id"), ): document_id = str(edge.get(document_field, "") or "") article_id = str(edge.get(article_field, "") or "") if document_id and article_id: existing = target_articles.setdefault(document_id, []) target_articles[document_id] = [article_id, *[item for item in existing if item != article_id]] elif explicit_documents: candidate_documents = explicit_documents else: top = max(document_scores.values() or [0.0]) broad_candidates = [ doc_id for doc_id, score in document_scores.items() if score >= max(0.24, top - 0.18) ] if top else list(self.documents) ambiguous_reference = bool(re.search(r"\b(?:ek |gecici )?madde\s+\d+", normalized)) if self._unresolved_generic_actor(normalized) or ambiguous_reference: candidate_documents = broad_candidates elif top: candidate_documents = [max(document_scores, key=document_scores.get)] else: candidate_documents = broad_candidates edges = matched_edges or [ edge for edge in self.edges if edge.get("source_document_id") in candidate_documents and edge.get("target_document_id") in candidate_documents ] values = sorted(document_scores.values(), reverse=True) top_score = values[0] if values else 0.0 runner_up = values[1] if len(values) > 1 else 0.0 return { "candidate_document_ids": candidate_documents, "document_scores": document_scores, "top_document_id": max(document_scores, key=document_scores.get) if document_scores else "", "top_score": round(top_score, 4), "runner_up_score": round(runner_up, 4), "scope_confident": bool(len(candidate_documents) == 1 and top_score >= 0.38), "cross_document": bool(matched_edges), "candidate_edge_ids": [str(edge.get("edge_id", "")) for edge in edges], "candidate_edges": edges, "target_articles_by_document": { doc_id: list(dict.fromkeys(items)) for doc_id, items in target_articles.items() }, "reasons": reasons, } def render(self, plan: RuntimePlan) -> dict[str, Any]: if plan.status != NormativeStatus.ANSWERED.value or not plan.scopes: return {} if "requires_relation_validation" in plan.reasons: return {} concepts = [self.concepts.get(scope, {}) for scope in plan.scopes] if not concepts or any(not concept for concept in concepts): return {} if plan.mode != RequestMode.SOURCE_IDENTITY.value and any( scope not in self.curated_scopes for scope in plan.scopes ): return {} if plan.mode == RequestMode.SOURCE_IDENTITY.value: concept = concepts[0] document_code = str(concept.get("document_id", "")).rsplit("-", 1)[-1] title = str(concept.get("title", "") or "").strip() answer = ( f"**Sonuç — {document_code} sayılı Kanun {concept.get('article_id', '')}" f"{f' | {title}' if title else ''}:** Soruda belirtilen kurum veya başlık bu hükümde düzenlenir." ) return self._render_payload(answer, plan, concepts) if plan.mode == RequestMode.INVENTORY.value: heading = plan.topic or "sorulan konu" lines = [f"**Sonuç — {heading}:** Yayınlanmış corpus içinde bu konuya uzman tarafından bağlanmış hükümler şunlardır:"] for concept in concepts: metadata = concept.get("normative_metadata", {}) or {} lines.append( f"- **{concept.get('document_title', concept.get('document_id', ''))} — " f"{concept.get('article_id', '')}:** " f"{metadata.get('inventory_summary') or metadata.get('approved_summary') or metadata.get('regulates', '')}" ) return self._render_payload("\n".join(lines), plan, concepts) if plan.mode == RequestMode.COMPARISON.value: lines = ["**Sonuç:** İlgili hükümler aynı işlemin farklı aşamalarını veya koşullarını birlikte düzenler:"] for concept in concepts: metadata = concept.get("normative_metadata", {}) or {} lines.append( f"- **{concept.get('document_title', concept.get('document_id', ''))} — " f"{concept.get('article_id', '')}:** {metadata.get('approved_summary', '')}" ) relation = self._relation_for_scopes(plan.scopes) if relation: lines.extend(["", f"**Normatif bağ:** {relation}"]) return self._render_payload("\n".join(lines), plan, concepts) concept = concepts[0] metadata = concept.get("normative_metadata", {}) or {} document_code = str(concept.get("document_id", "")).rsplit("-", 1)[-1] title = str(metadata.get("display_heading") or concept.get("title", "")) lines = [ f"**Sonuç — {document_code} sayılı Kanun {concept.get('article_id', '')}" f"{f' | {title}' if title else ''}:** {metadata.get('approved_summary', '')}" ] points = metadata.get("approved_points", []) or [] if points: lines.append("") for point in points: if isinstance(point, dict): label = str(point.get("label", "") or "") statement = str(point.get("statement", "") or "") lines.append(f"- **{label}:** {statement}" if label else f"- {statement}") elif point: lines.append(f"- {point}") return self._render_payload("\n".join(lines), plan, concepts) def review_coverage(self) -> dict[str, Any]: total = len(self.concepts) reviewed = sum( 1 for concept in self.concepts.values() if str((concept.get("normative_metadata", {}) or {}).get("review_status", "")) in {"human_reviewed", "expert_approved"} ) curated = len(self.curated_scopes) return { "total_provisions": total, "reviewed_provisions": reviewed, "answer_ready_provisions": curated, "review_ratio": round(reviewed / total, 4) if total else 0.0, } def concept(self, scope: tuple[str, str]) -> dict[str, Any]: return dict(self.concepts.get(scope, {}) or {}) def clarification_choices(self, plan: RuntimePlan) -> list[dict[str, Any]]: reasons = set(plan.reasons) article_ambiguity = "ambiguous_article_reference" in reasons if not ( {"missing_normative_relation", "ambiguous_institutional_actor", "ambiguous_article_reference"} & reasons ): return [] choices = [] for scope in plan.scopes: concept = self.concepts.get(scope, {}) or {} metadata = concept.get("normative_metadata", {}) or {} if scope not in self.curated_scopes and not article_ambiguity: continue choices.append({ "document_id": scope[0], "document_title": concept.get("document_title", scope[0]), "article_id": scope[1], "title": metadata.get("display_heading") or concept.get("title", ""), "summary": metadata.get("inventory_summary") or metadata.get("approved_summary", ""), "clarification_type": "document_scope" if article_ambiguity else "canonical_topic_scope", }) return choices[:6] def _generic_actor_scopes(self, normalized: str) -> list[tuple[str, str]]: requested = "gorev" if "gorev" in normalized else "yetki" if "yetki" in normalized else "sorumluluk" scopes = [] for scope, concept in self.concepts.items(): if scope not in self.curated_scopes: continue metadata = concept.get("normative_metadata", {}) or {} heading = _normative_text(str(metadata.get("display_heading", "") or concept.get("title", ""))) if "kurul" in heading and requested in heading: scopes.append(scope) return sorted(scopes, key=lambda item: (item[0], _article_sort_key(item[1]))) def _render_payload(self, answer: str, plan: RuntimePlan, concepts: list[dict[str, Any]]) -> dict[str, Any]: return { "answer": answer, "plan": plan.to_dict(), "build_id": self.build_id, "sources": [ { "document_id": item.get("document_id", ""), "document_title": item.get("document_title", ""), "article_id": item.get("article_id", ""), "article_title": item.get("title", ""), "evidence_id": _first_evidence_id(item), } for item in concepts ], } def _curated_and_source_valid(self, concept: dict[str, Any]) -> bool: metadata = concept.get("normative_metadata", {}) or {} if metadata.get("review_status") not in {"human_reviewed", "expert_approved"}: return False if not str(metadata.get("approved_summary", "") or "").strip(): return False source = normalize_for_search(str(concept.get("source_text", "") or "")) if not source: return False for point in metadata.get("approved_points", []) or []: if not isinstance(point, dict): continue for term in point.get("evidence_terms", []) or []: if normalize_for_search(str(term)) not in source: return False return True def _field_overlap(self, query: str, metadata: dict[str, Any]) -> float: query = _normative_text(query) query_terms = _content_terms(query) if not query_terms: return 0.0 primary_values = [] for key in ( "query_aliases", "canonical_concepts", "regulated_situations", "topic_memberships", "legal_operations", "competent_authorities", ): primary_values.extend(metadata.get(key, []) or []) secondary_values = [] for variable_values in (metadata.get("normative_variables", {}) or {}).values(): secondary_values.extend(variable_values or []) primary_values.extend([metadata.get("regulates", ""), metadata.get("display_heading", "")]) primary_text = _normative_text(" ".join(str(value) for value in primary_values if value)) secondary_text = _normative_text(" ".join(str(value) for value in secondary_values if value)) # Phrase authority belongs only to reviewed semantic descriptions. A # competent-authority value such as "Cumhurbaşkanı" is an actor facet, # not a query alias; treating every facet as a phrase previously made # all provisions mentioning that actor tie at a misleadingly high # score. phrase_values = [ *(metadata.get("query_aliases", []) or []), *(metadata.get("canonical_concepts", []) or []), metadata.get("regulates", ""), metadata.get("display_heading", ""), ] aliases = [_normative_text(str(value)) for value in phrase_values if value] phrase_score = max( ( min(1.0, 0.58 + len(alias.split()) * 0.07) for alias in aliases if alias and len(_content_terms(alias)) >= 2 and re.search(rf"(?= 0.85 ), default=0.0, ) # Reviewed query aliases also authorize close paraphrases. This is a # bidirectional coverage test, so a long generic alias cannot win on a # single shared actor or operation. alias_overlap = max( ( min(_topic_overlap(alias, query), _topic_overlap(query, alias)) for alias in [ _normative_text(str(value)) for value in metadata.get("query_aliases", []) or [] if value ] if len(_content_terms(alias)) >= 2 ), default=0.0, ) if alias_overlap >= 0.85: phrase_score = max(phrase_score, min(0.94, 0.70 + alias_overlap * 0.24)) primary_terms = set(primary_text.split()) secondary_terms = set(secondary_text.split()) primary_matches = { term for term in query_terms if any(search_terms_match(term, candidate) for candidate in primary_terms) } secondary_matches = { term for term in query_terms - primary_matches if any(search_terms_match(term, candidate) for candidate in secondary_terms) } coverage = len(primary_matches) / len(query_terms) precision = len(primary_matches) / max(1, min(len(primary_terms), len(query_terms) + 4)) # Document-wide variables are useful recall hints, but may not turn a # provision that merely mentions an actor into a top semantic match. secondary_bonus = min(0.12, (len(secondary_matches) / len(query_terms)) * 0.18) score = max(phrase_score, min(1.0, coverage * 0.78 + precision * 0.22 + secondary_bonus)) exclusion_overlap = max( (_semantic_exclusion_overlap(str(value), query) for value in metadata.get("exclusions", []) or []), default=0.0, ) if exclusion_overlap >= 0.72: score *= 0.25 return score def _operation_alignment(self, query: str, metadata: dict[str, Any]) -> float: """Discount actor/topic matches that miss the requested operation. Natural-language questions commonly name the same actor across many provisions. Flattening actor and operation facets makes ``öğretim elemanı + ek ders ödenmesi`` tie with degree promotion. This factor is compiled entirely from MCKF ``legal_operations`` values and therefore remains portable to new institutional packages. """ query_terms = _content_terms(query) requested = { _operation_key(term) for term in query_terms if _operation_key(term) in self.operation_terms } if not requested: return 1.0 candidate_terms = set() for operation in metadata.get("legal_operations", []) or []: terms = [ term for term in _normative_text(str(operation)).split() if len(term) >= 3 and term not in TOPIC_STOPWORDS and not term.isdigit() ] if terms: candidate_terms.add(_operation_key(terms[-1])) if not candidate_terms: # Missing operation metadata is an uncovered facet, not evidence # of a mismatch. Keep a modest uncertainty discount while # allowing an exact actor/title match to remain competitive. return 0.85 matched = { term for term in requested if term in candidate_terms } coverage = len(matched) / len(requested) return 0.62 + coverage * 0.38 def _inventory_scopes( self, question: str, ranked: list[tuple[float, tuple[str, str]]], ) -> list[tuple[str, str]]: normalized = _normative_text(question) topic_candidates: list[tuple[int, str]] = [] for scope, concept in self.concepts.items(): if scope not in self.curated_scopes: continue metadata = concept.get("normative_metadata", {}) or {} for topic in metadata.get("topic_memberships", []) or []: topic_norm = _normative_text(str(topic)) if topic_norm and (topic_norm in normalized or _topic_overlap(topic_norm, normalized) >= 0.66): topic_candidates.append((len(topic_norm.split()), topic_norm)) if topic_candidates: topic = max(topic_candidates)[1] scopes = [ scope for scope, concept in self.concepts.items() if scope in self.curated_scopes and topic in { _normative_text(str(value)) for value in (concept.get("normative_metadata", {}) or {}).get("topic_memberships", []) or [] } ] explicit_documents = set(self._explicit_document_ids(question)) if explicit_documents: scopes = [scope for scope in scopes if scope[0] in explicit_documents] return sorted(scopes, key=lambda item: (item[0], _article_sort_key(item[1]))) return [scope for score, scope in ranked if score >= max(0.48, ranked[0][0] - 0.18)][:8] if ranked else [] def _comparison_scopes(self, ranked: list[tuple[float, tuple[str, str]]]) -> list[tuple[str, str]]: selected: list[tuple[str, str]] = [] for score, scope in ranked: if score < 0.34: continue if scope not in self.curated_scopes: continue if scope not in selected: selected.append(scope) if len(selected) >= 2: break return selected def _best_topic(self, question: str, scopes: list[tuple[str, str]]) -> str: normalized = _normative_text(question) candidates = [] for scope in scopes: metadata = (self.concepts.get(scope, {}).get("normative_metadata", {}) or {}) for topic in metadata.get("topic_memberships", []) or []: if ( _normative_text(str(topic)) in normalized or _topic_overlap(str(topic), normalized) >= 0.66 ): candidates.append(str(topic)) return max(candidates, key=len, default="sorulan konu") def _relation_for_scopes(self, scopes: tuple[tuple[str, str], ...]) -> str: scope_set = set(scopes) for edge in self.edges: source = (str(edge.get("source_document_id", "")), str(edge.get("source_article_id", ""))) target = (str(edge.get("target_document_id", "")), str(edge.get("target_article_id", ""))) if source in scope_set and target in scope_set: return str(edge.get("approved_interpretation") or edge.get("description") or "") return "" def _unresolved_conflicts(self, scopes: list[tuple[str, str]]) -> list[dict[str, Any]]: scope_set = set(scopes) return [ edge for edge in self.edges if str(edge.get("relation_type", "")) in {"conflicts_with", "contradicts"} and (str(edge.get("source_document_id", "")), str(edge.get("source_article_id", ""))) in scope_set and (str(edge.get("target_document_id", "")), str(edge.get("target_article_id", ""))) in scope_set and not edge.get("resolved_by") ] def _matching_relation_edges(self, question: str) -> list[dict[str, Any]]: query_terms = _content_terms(question) if len(query_terms) < 2: return [] explicit_documents = set(self._explicit_document_ids(question)) ranked: list[tuple[int, float, dict[str, Any]]] = [] for edge in self.edges: if str(edge.get("review_status", "")) not in {"human_reviewed", "expert_approved"}: continue endpoint_documents = { str(edge.get("source_document_id", "")), str(edge.get("target_document_id", "")), } if explicit_documents and not explicit_documents.issubset(endpoint_documents): continue relation_text = " ".join( " ".join(str(value) for value in edge.get(field, []) or []) if field == "query_aliases" else str(edge.get(field, "") or "") for field in ("query_aliases", "description", "approved_interpretation", "relation_type") ) relation_terms = _content_terms(relation_text) matched = { term for term in query_terms if any(search_terms_match(term, candidate) for candidate in relation_terms) } if len(matched) < 2: continue coverage = len(matched) / max(1, min(len(query_terms), len(relation_terms))) ranked.append((len(matched), coverage, edge)) ranked.sort(key=lambda item: (-item[0], -item[1], str(item[2].get("edge_id", "")))) if not ranked: return [] best_count, best_coverage, _edge = ranked[0] return [ edge for count, coverage, edge in ranked if count == best_count and coverage >= best_coverage - 0.05 ] def _decision_plan(self, question: str, as_of_date: str) -> RuntimePlan | None: normalized = normalize_for_search(question) for contract in self.contracts: aliases = [normalize_for_search(str(value)) for value in contract.get("query_aliases", []) or []] if not aliases or not any(alias and _topic_overlap(alias, normalized) >= 0.60 for alias in aliases): continue indicators = [normalize_for_search(str(value)) for value in contract.get("case_indicators", []) or []] if indicators and not any(value in normalized for value in indicators): continue facts = self._extract_facts(normalized, contract) fields = ((contract.get("input_schema", {}) or {}).get("fields", {}) or {}) required = [name for name, spec in fields.items() if (spec or {}).get("required")] judgment = [str(item.get("fact", "")) for item in contract.get("judgment_requirements", []) or []] missing = tuple(name for name in required + judgment if name and name not in facts) return RuntimePlan( NormativeStatus.UNKNOWN.value if missing else NormativeStatus.ANSWERED.value, RequestMode.CASE_ASSESSMENT.value, question, scopes=tuple( (str(item.get("document_id", "")), str(item.get("article_id", ""))) for item in contract.get("source_refs", []) or [] if item.get("document_id") and item.get("article_id") ), confidence=1.0, reasons=("published_decision_contract",), decision_type=str(contract.get("decision_type", "")), facts=facts, missing_facts=missing, as_of_date=as_of_date, ) return None def _extract_facts(self, normalized: str, contract: dict[str, Any]) -> dict[str, Any]: facts: dict[str, Any] = {} for fact, extractor in (contract.get("fact_extractors", {}) or {}).items(): for item in extractor.get("patterns", []) or []: pattern = str(item.get("pattern", "") or "") match = re.search(pattern, normalized) if pattern else None if not match: continue if "value" in item: facts[fact] = item.get("value") elif item.get("type") == "integer" and match.groups(): facts[fact] = int(match.group(1)) break return facts def _explicit_scopes(self, question: str) -> list[tuple[str, str]]: normalized = normalize_for_search(question) code_to_document = { normalize_for_search(str(document.get("short_code", "") or "")): document_id for document_id, document in self.documents.items() if document.get("short_code") } code_pattern = "|".join(re.escape(code) for code in sorted(code_to_document, key=len, reverse=True)) codes = re.findall(rf"\b({code_pattern})\b", normalized) if code_pattern else [] article_matches = list(re.finditer(r"\b(?:ek |gecici )?madde(?:si|sindeki|deki|nin)?\s+(\d+(?:/[a-z])?)", normalized)) scopes = [] for match in article_matches: prefix = normalized[max(0, match.start() - 50):match.start()] nearby_codes = re.findall(rf"\b({code_pattern})\b", prefix) if code_pattern else [] code = nearby_codes[-1] if nearby_codes else (codes[0] if len(set(codes)) == 1 else "") if not code: continue token = match.group(0) kind = "Ek Madde" if token.startswith("ek ") else "Geçici Madde" if token.startswith("gecici ") else "Madde" scope = (code_to_document.get(code, ""), f"{kind} {match.group(1).upper()}") if scope in self.concepts and scope not in scopes: scopes.append(scope) for match in re.finditer(r"\b(\d+(?:/[a-z])?)\s*\.?\s*madd(?:e|esi|esindeki|edeki)", normalized): prefix = normalized[max(0, match.start() - 70):match.start()] nearby_codes = re.findall(rf"\b({code_pattern})\b", prefix) if code_pattern else [] code = nearby_codes[-1] if nearby_codes else (codes[0] if len(set(codes)) == 1 else "") scope = (code_to_document.get(code, ""), f"Madde {match.group(1).upper()}") if code and scope in self.concepts and scope not in scopes: scopes.append(scope) return scopes def _explicit_document_ids(self, question: str) -> list[str]: normalized = normalize_for_search(question) return [ document_id for document_id, document in self.documents.items() if re.search(rf"\b{re.escape(normalize_for_search(str(document.get('short_code', ''))))}\b", normalized) ] def _ambiguous_bare_article_scopes(self, question: str) -> list[tuple[str, str]]: if self._explicit_document_ids(question): return [] normalized = normalize_for_search(question) references = [] for match in re.finditer(r"\b((?:ek |gecici )?madde)\s+(\d+(?:/[a-z])?)", normalized): prefix = match.group(1) kind = "Ek Madde" if prefix.startswith("ek ") else "Geçici Madde" if prefix.startswith("gecici ") else "Madde" article_id = f"{kind} {match.group(2).upper()}" if article_id not in references: references.append(article_id) if len(references) != 1: return [] scopes = [scope for scope in self.concepts if scope[1] == references[0]] title_matches = [] for scope in scopes: concept = self.concepts.get(scope, {}) or {} metadata = concept.get("normative_metadata", {}) or {} title = _normative_text(str(metadata.get("display_heading") or concept.get("title", ""))) if title and len(_content_terms(title)) >= 2 and title in normalized: title_matches.append(scope) if len(title_matches) == 1: return [] return sorted(scopes) if len({scope[0] for scope in scopes}) > 1 else [] @staticmethod def _direct_source_intent(normalized: str) -> bool: return any(value in normalized for value in ("ne diyor", "metni", "aynen", "tam madd")) @staticmethod def _source_identity_intent(normalized: str) -> bool: return bool( re.search(r"\bhangi maddede\b", normalized) or re.search(r"\bhangi madde duzenler\b", normalized) or re.search(r"\bhangi maddede duzenlen", normalized) ) @staticmethod def _inventory_intent(normalized: str) -> bool: relational_markers = ( "yukumlu mu", "yukumlu mudur", "zorunda mi", "yetkili mi", "midir", "olur mu", "yapabilir mi", "verebilir mi", "odenir mi", ) bare_inventory = bool(re.search(r"\bmadde(?:ler)? var mi\b", normalized)) return bool( re.search(r"\bhangi maddeler\b", normalized) or (bare_inventory and not any(marker in normalized for marker in relational_markers)) or "mevzuat envanteri" in normalized or ("hukum" in normalized and any(value in normalized for value in ("listele", "listeler", "sirala"))) ) def _requires_relation_validation(self, question: str, scope: tuple[str, str]) -> bool: normalized = _normative_text(question) markers = ( "yukumlu", "zorunda", "yetkili", "sorumlu", "midir", "mudur", "olur mu", "yapabilir mi", "verebilir mi", "odenir mi", ) if not any(marker in normalized for marker in markers): return False metadata = (self.concepts.get(scope, {}).get("normative_metadata", {}) or {}) for value in metadata.get("query_aliases", []) or []: alias = _normative_text(str(value)) if alias and len(_content_terms(alias)) >= 3 and alias in normalized: return False return True @staticmethod def _broad_topic_intent(normalized: str) -> bool: return bool( re.search(r"\bile ilgili (?:ne var|neler var)\b", normalized) or re.search(r"\bhakkinda (?:ne var|neler var|bilgi var mi)\b", normalized) or re.search(r"\bkonusunda (?:ne var|neler var)\b", normalized) ) @staticmethod def _unresolved_generic_actor(normalized: str) -> bool: if not re.search(r"\bkurul(?:un|unun)?\s+(?:gorev|yetki|sorumluluk)\w*", normalized): return False specific = ( "yuksekogretim kurulu", "yok", "denetleme kurulu", "universitelerarasi kurul", "universite yonetim kurulu", "fakulte kurulu", "enstitu kurulu", "senato", ) return not any(value in normalized for value in specific) def _comparison_intent(self, normalized: str, direct_scopes: list[tuple[str, str]]) -> bool: return len(direct_scopes) > 1 or len(set(self._explicit_document_ids(normalized))) > 1 or any( value in normalized for value in ( "birlikte", "tamamlar", "iliski", "karsilastir", "farki", "baglanti", "bag nedir", "nasil baglan", "hangi 2547", "hangi 2809", "hangi 2914", "tanima dayan", ) ) @staticmethod def _temporal_request(question: str) -> str: normalized = normalize_for_search(question) temporal_markers = ("tarihinde", "tarihte", "yururlukteydi", "gecerliydi") past_year_request = "yilinda" in normalized and any( marker in normalized for marker in ("neydi", "nasildi", "miydi", "muydu", "uygulaniyordu", "gecerliydi", "yururlukte") ) if not any(marker in normalized for marker in temporal_markers) and not past_year_request: return "" full_date = re.search(r"\b(20\d{2})-(\d{2})-(\d{2})\b", normalized) if full_date: return full_date.group(0) year = re.search(r"\b(19\d{2}|20\d{2})\b", normalized) return f"{year.group(1)}-12-31" if year else "" def _supports_date(self, value: str) -> bool: if not value: return True try: target = date.fromisoformat(value) except ValueError: return False coverage_dates = [] for document in self.documents.values(): raw = str(document.get("source_snapshot_date", "") or "") try: coverage_dates.append(date.fromisoformat(raw)) except ValueError: continue return bool(coverage_dates) and all(target == snapshot for snapshot in coverage_dates) def _content_terms(value: str) -> set[str]: return { term for term in _normative_text(value).split() if len(term) >= 3 and term not in TOPIC_STOPWORDS and not term.isdigit() } def _operation_key(term: str) -> str: """Return a compact action family for Turkish operation predicates. This stemmer is deliberately used only on the final predicate of a published ``legal_operations`` phrase. It may therefore equate ``kurulur`` with ``kurma`` without reintroducing the dangerous global ``kurul`` (governing body) / ``kurulmak`` (establishment) collision. """ value = _normative_text(term) families = ( (("kurul", "kurma"), "kurma"), (("oden", "odem", "ode"), "odeme"), (("atan", "atam"), "atama"), (("gorevlendir",), "gorevlendirme"), (("yukselt", "yuksel"), "yukseltme"), (("secil", "secim", "secme"), "secme"), (("belirle", "belirlen"), "belirleme"), (("duzenle", "duzenlen"), "duzenleme"), (("hesapla", "hesaplan"), "hesaplama"), (("planla", "planlan"), "planlama"), (("programla", "programlan"), "programlama"), (("basvur", "basvuru"), "basvuru"), (("kaydet", "kayit"), "kayit"), (("denetle", "denetim"), "denetim"), (("onayla", "onay"), "onay"), (("bildir", "bildirim"), "bildirim"), (("uygula", "uygulan"), "uygulama"), (("acil", "acma"), "acma"), (("kapat", "kapan"), "kapatma"), ) for prefixes, key in families: if any(value.startswith(prefix) for prefix in prefixes): return key return value def _normative_text(value: str) -> str: """Normalize common Turkish institutional compounds before comparison.""" # Preserve the YÖK acronym as an institutional entity before accent # folding. Otherwise it becomes Turkish ``yok`` (absence), which is a # stopword, and every provision containing only ``görev`` ties with the # actual Yükseköğretim Kurulu provision. prepared = re.sub( r"\byök(?:['’]?(?:ün|un|ın|in))?\b", "yokkurulu", str(value), flags=re.IGNORECASE, ) normalized = normalize_for_search(prepared) replacements = { "acikogretim": "acik ogretim", "acikogretimde": "acik ogretim", "acikogretimin": "acik ogretim", "uzaktan egitim": "uzaktan ogretim", "ortadogu": "orta dogu", "yok un": "yokkurulu", "yokun": "yokkurulu", } for source, target in replacements.items(): normalized = normalized.replace(source, target) return re.sub(r"\s+", " ", normalized).strip() def _named_institution_title_score(query: str, title: str) -> float: """Return a strong score for a named institution in an official heading. The matcher intentionally derives names from the query/title pair. It contains no institution catalogue, so newly published universities and analogous institutional provisions become routable after a corpus build. """ query_tokens = _normative_text(query).split() title_text = _normative_text(title) if not query_tokens or not title_text: return 0.0 generic = {"bir", "bu", "hangi", "yeni", "devlet", "vakif", "ilgili"} matches: list[tuple[int, str]] = [] for index, token in enumerate(query_tokens): if not token.startswith("universite"): continue for length in range(1, min(4, index) + 1): name_tokens = query_tokens[index - length:index] if all(value in generic for value in name_tokens): continue phrase = " ".join([*name_tokens, "universitesi"]) if phrase in title_text: matches.append((length, phrase)) if not matches: return 0.0 length, phrase = max(matches) if title_text == phrase: return 1.0 operation_terms = { "kurulmustur", "kurulur", "kurulmasi", "duzenlenir", "olusur", "kapatilir", "birlestirilir", "donusturulur", } query_operations = operation_terms & set(query_tokens) title_operations = operation_terms & set(title_text.split()) if query_operations & title_operations and len(title_text.split()) <= 14: return min(0.99, 0.92 + length * 0.02) # A long provision that merely mentions the institution remains a useful # recall candidate, but cannot tie the provision whose official heading is # the institution itself. compactness = max(0.0, 1.0 - max(0, len(title_text.split()) - len(phrase.split())) / 24) return min(0.90, 0.68 + length * 0.03 + compactness * 0.12) def _official_heading_identity_score(query: str, title: str) -> float: """Treat a provision heading as a canonical semantic address. Headings are often inflected in natural-language questions (``Dekan`` -> ``dekanlık``, ``Öğrencilerin disiplin işleri`` -> ``öğrenci disiplini``). Requiring a byte-like phrase match loses that authoritative signal and lets incidental mentions win. Strong term coverage is therefore enough, while one-word headings require an explicit role/status question so that generic words do not become universal routers. """ query_text = _normative_text(query) title_text = _normative_text(title) if not title_text: return 0.0 # Phrase identity must respect token boundaries. A short structural # heading such as ``Ek`` is not present merely because those characters # occur inside another word (for example ``dekanlık``). title_terms = _content_terms(title_text) query_terms = _content_terms(query_text) if not title_terms or not query_terms: return 0.0 exact_phrase = bool(re.search(rf"(?= 2 and re.search(r"\b(?:ek |gecici )?madde\s+\d+", query_text): return 1.0 matched = { term for term in title_terms if any(search_terms_match(term, candidate) for candidate in query_terms) } if len(title_terms) >= 2 and len(matched) >= 2 and len(matched) / len(title_terms) >= 0.66: title_coverage = len(matched) / len(title_terms) query_coverage = len(matched) / len(query_terms) # A heading that names only the actor (for example ``Öğretim # elemanları``) is a useful clue, but must not outrank a provision that # also matches the requested operation/result (``ek ders ücreti``). return min(0.94, 0.62 + title_coverage * 0.20 + query_coverage * 0.12) role_markers = { "rol", "makam", "gorev", "yetki", "sorumluluk", "atama", "atanma", "kimdir", "nedir", "nasil", } if ( len(title_terms) == 1 and len(next(iter(title_terms))) >= 5 and matched and any( any(search_terms_match(marker, candidate) for candidate in query_terms) for marker in role_markers ) ): return 0.88 return 0.0 def _topic_overlap(left: str, right: str) -> float: left_terms = _content_terms(left) right_terms = _content_terms(right) if not left_terms or not right_terms: return 0.0 matched = sum( 1 for term in left_terms if any(search_terms_match(term, other) for other in right_terms) ) return matched / len(left_terms) def _semantic_exclusion_overlap(left: str, right: str) -> float: """Match a reviewed exclusion only when it expresses a real distinction. One generic shared word (for example ``kapatma``) must never suppress a provision. Exclusions are boundary statements and need at least two content-term matches plus meaningful coverage on both sides. """ left_terms = _content_terms(left) right_terms = _content_terms(right) if len(left_terms) < 2 or len(right_terms) < 2: return 0.0 matched = { term for term in left_terms if any(search_terms_match(term, other) for other in right_terms) } if len(matched) < 2: return 0.0 return min(len(matched) / len(left_terms), len(matched) / len(right_terms)) def _article_sort_key(article_id: str) -> tuple[int, int, str]: normalized = normalize_for_search(article_id) kind = 0 if normalized.startswith("madde") else 1 if normalized.startswith("ek") else 2 match = re.search(r"\d+", normalized) return kind, int(match.group(0)) if match else 999999, normalized def _first_evidence_id(concept: dict[str, Any]) -> str: for clause in concept.get("clauses", []) or []: for evidence in clause.get("evidence_spans", []) or []: evidence_id = str(evidence.get("evidence_id", "") or "") if evidence_id: return evidence_id return "" RUNTIME = CanonicalNormativeRuntime() def init_normative_runtime(corpus: dict[str, Any] | None) -> None: global RUNTIME RUNTIME = CanonicalNormativeRuntime(corpus) def plan_normative_request( question: str, constrained_scopes: list[tuple[str, str]] | None = None, ) -> RuntimePlan: return RUNTIME.plan(question, constrained_scopes=constrained_scopes) def route_normative_question(question: str) -> dict[str, Any]: return RUNTIME.route(question) def render_normative_plan(plan: RuntimePlan) -> dict[str, Any]: return RUNTIME.render(plan) def normative_review_coverage() -> dict[str, Any]: return RUNTIME.review_coverage() def normative_clarification_choices(plan: RuntimePlan) -> list[dict[str, Any]]: return RUNTIME.clarification_choices(plan)