from __future__ import annotations
from html import escape
from typing import Any
from config import OFFICIAL_SOURCE_URL, MAX_CONTEXT_CHARS, EVAL_CONTEXT_MAX_CHARS
from utils import compact_text, dedupe_repeated_text
SEMANTIC_UNITS: dict[str, dict[str, Any]] = {}
ARTICLE_SOURCES: dict[str, dict[str, Any]] = {}
ARTICLE_SOURCE_COLLISIONS: set[str] = set()
_MISSING_FULL_SOURCE_TEMPLATE = (
"[Kaynak madde metni yüklenemedi: {source_label}. "
"Bu cevap için tam kanun maddesi ARTICLE_SOURCES içinde bulunamadı. "
"Dar clause/evidence metni resmi kaynak maddesi gibi gösterilmedi.]"
)
_SOURCE_DISCLOSURE_STYLE = """"""
def set_article_sources(article_sources: dict[str, dict]) -> None:
"""Register full article texts for source rendering.
Source panels must render the full official article text, not the short evidence span
selected for answer grounding. To make this safe for the upcoming multi-document
architecture, records are indexed by both `document_id::article_id` and, when not
ambiguous, by `article_id`.
"""
global ARTICLE_SOURCES, ARTICLE_SOURCE_COLLISIONS
ARTICLE_SOURCES, ARTICLE_SOURCE_COLLISIONS = _normalize_article_sources(article_sources or {})
def set_semantic_units(units: dict[str, dict]) -> None:
global SEMANTIC_UNITS
SEMANTIC_UNITS = units or {}
def source_key(meta: dict) -> tuple[str, str, str]:
return (
_metadata_document_id(meta),
meta.get("article", "") or meta.get("article_id", ""),
meta.get("title", "") or meta.get("article_title", ""),
)
def source_display_text(node_with_score) -> str:
"""Return context text for retrieval/LLM grounding.
This function is allowed to fall back to node text because it is used for model context.
UI source panels use the stricter full-article functions below and intentionally do not
fall back to evidence/span text.
"""
meta = getattr(node_with_score.node, "metadata", {}) or {}
article = meta.get("article", "") or meta.get("article_id", "")
document_id = _metadata_document_id(meta)
unit = _lookup_semantic_unit(document_id, article)
full_text = unit.get("source_text")
if full_text:
return dedupe_repeated_text(full_text)
text = getattr(node_with_score.node, "text", "") or ""
if "\n\n" in text:
text = text.split("\n\n", 1)[1]
return dedupe_repeated_text(text)
def format_context(source_nodes) -> str:
blocks = []
for idx, node_with_score in enumerate(source_nodes, start=1):
meta = getattr(node_with_score.node, "metadata", {}) or {}
score = node_with_score.score
score_text = f"{score:.3f}" if score is not None else "yok"
article = meta.get("article") or meta.get("article_id") or ""
title = meta.get("title") or meta.get("article_title") or ""
document_id = _metadata_document_id(meta)
source_header = " | ".join(part for part in [document_id, article, title] if part)
blocks.append(
f"[Kaynak {idx}] {source_header} | Skor: {score_text}\n"
f"{compact_text(source_display_text(node_with_score), MAX_CONTEXT_CHARS)}"
)
return "\n\n".join(blocks)
def format_expandable_sources(source_nodes, question: str = "", answer: str = "") -> str:
"""Render full article sources for legacy node-based source lists.
Fail-closed behavior: if the full article cannot be resolved from ARTICLE_SOURCES or
SEMANTIC_UNITS, do not display node/evidence text as if it were the official source.
"""
seen = set()
blocks = []
for node_with_score in source_nodes:
meta = getattr(node_with_score.node, "metadata", {}) or {}
document_id = _metadata_document_id(meta)
article = meta.get("article", "") or meta.get("article_id", "")
title = meta.get("title", "") or meta.get("article_title", "")
key = _dedupe_key(document_id, article, title)
if key in seen:
continue
seen.add(key)
article_source = _lookup_article_source(document_id, article)
if article_source:
title = article_source.get("title") or title
body_text = article_source.get("source_text", "")
else:
unit = _lookup_semantic_unit(document_id, article)
title = unit.get("title") or title
body_text = unit.get("source_text", "")
if not body_text:
body_text = _missing_source_message(document_id, article)
label = _source_label(document_id, article, title, article_source.get("document_title", "") if article_source else "")
blocks.append(_render_source_block(label, body_text))
official = _official_source_link()
return "\n\n".join([_SOURCE_DISCLOSURE_STYLE] + blocks + [official])
def _extract_text_article_sources(text: str) -> list[dict[str, str]]:
import re
sources = []
text_val = text or ""
# Pattern 1: 2547/2914/2809 ... Madde 11 / Ek Madde 1 / Geçici Madde 2
pattern1 = r"\b(2547|2914|2809)(?:\s+say[ıi]l[ıi](?:\s+kanun(?:un|daki|da)?)?)?\s+(Ge[cç]ici\s+Madde\s+\d+|Ek\s+Madde\s+\d+|Madde\s+\d+(?:/[A-Za-zÇĞİÖŞÜçğıöşü])?)\b"
for match in re.finditer(pattern1, text_val, flags=re.IGNORECASE):
code = match.group(1)
doc_id = f"TR-KANUN-{code}"
raw_art = match.group(2).strip()
art_norm = re.sub(r"\s+", " ", raw_art).title()
if art_norm.lower().startswith("madde "):
art_norm = f"Madde {art_norm[6:].upper()}"
elif art_norm.lower().startswith("ek madde "):
art_norm = f"Ek Madde {art_norm[9:]}"
elif art_norm.lower().startswith("geçici madde ") or art_norm.lower().startswith("gecici madde "):
art_norm = f"Geçici Madde {art_norm[13:].upper()}"
sources.append({"document_id": doc_id, "article_id": art_norm})
# Pattern 2: 2547/2914/2809 ... 11. madde(si/sinde/sindeki/ye/den)
pattern2 = r"\b(2547|2914|2809)\s+say[ıi]l[ıi]\s+kanun(?:un|daki|da)?\s+(?:(Ek|Ge[cç]ici)\s+)?(\d+)\.\s+madde(?:si|sinde|sindeki|ye|den|siyle|siyle|de)?\b"
for match in re.finditer(pattern2, text_val, flags=re.IGNORECASE):
code = match.group(1)
doc_id = f"TR-KANUN-{code}"
prefix = match.group(2)
num = match.group(3)
if prefix and prefix.lower() in {"ek"}:
art_norm = f"Ek Madde {num}"
elif prefix and prefix.lower() in {"geçici", "gecici"}:
art_norm = f"Geçici Madde {num}"
else:
art_norm = f"Madde {num}"
sources.append({"document_id": doc_id, "article_id": art_norm})
return sources
def format_clause_sources(clauses: list[dict], question: str = "", answer: str = "") -> str:
"""Render full article sources for clause/evidence-based answers.
The answer engine selects small evidence spans for grounding, but the UI source panel
must show the full related article. Therefore this function never uses
`clause['source_text']` as a fallback. That field may contain only a clause, evidence
span, or generated/derived text. If the full article is missing, the UI displays an
explicit diagnostic message instead of pretending the evidence span is the source.
"""
blocks = []
seen_articles = set()
all_clauses = list(clauses or [])
if answer:
all_clauses.extend(_extract_text_article_sources(answer))
for clause in all_clauses:
document_id = _clause_document_id(clause)
article = str(clause.get("article_id", "") or clause.get("article", "")).strip()
if not article:
continue
key = _dedupe_key(document_id, article)
if key in seen_articles:
continue
seen_articles.add(key)
article_source = _lookup_article_source(document_id, article)
title = ""
body_text = ""
if article_source:
title = str(article_source.get("title", "") or "").strip()
body_text = str(article_source.get("source_text", "") or "").strip()
if not body_text:
# Accept explicit full article text only. Do not fall back to clause['source_text'].
explicit_full_article = str(clause.get("article_source_text", "") or "").strip()
if explicit_full_article:
body_text = explicit_full_article
title = title or _source_title_from_clause(clause)
if not body_text:
body_text = _missing_source_message(document_id, article)
title = title or _source_title_from_clause(clause)
label = _source_label(document_id, article, title, article_source.get("document_title", "") if article_source else clause.get("document_title", ""))
blocks.append(_render_source_block(label, body_text))
official = _official_source_link()
return "\n\n".join([_SOURCE_DISCLOSURE_STYLE] + blocks + [official])
def get_article_source(document_id: str, article_id: str) -> dict[str, Any]:
"""Return the canonical full-article source already loaded by the runtime."""
return dict(_lookup_article_source(document_id, article_id) or {})
def _normalize_article_sources(article_sources: dict[str, dict]) -> tuple[dict[str, dict[str, Any]], set[str]]:
normalized: dict[str, dict[str, Any]] = {}
plain_key_documents: dict[str, set[str]] = {}
for original_key, raw_record in (article_sources or {}).items():
if not isinstance(raw_record, dict):
continue
record = dict(raw_record)
article_id = str(
record.get("article_id")
or record.get("article")
or _article_from_key(str(original_key))
or ""
).strip()
document_id = str(
record.get("document_id")
or record.get("document")
or record.get("law_id")
or _document_from_key(str(original_key))
or ""
).strip()
if article_id:
record.setdefault("article_id", article_id)
if document_id:
record.setdefault("document_id", document_id)
# Preserve the original key for full backward compatibility.
normalized[str(original_key)] = record
if document_id and article_id:
normalized[_compound_key(document_id, article_id)] = record
plain_key_documents.setdefault(article_id, set()).add(document_id)
if article_id:
# Plain article key is safe in single-document mode. Collision handling below
# prevents ambiguous multi-document keys from being used silently.
normalized.setdefault(article_id, record)
collisions = {article for article, docs in plain_key_documents.items() if len(docs) > 1}
for article in collisions:
normalized.pop(article, None)
return normalized, collisions
def _lookup_article_source(document_id: str, article_id: str) -> dict[str, Any]:
document_id = (document_id or "").strip()
article_id = (article_id or "").strip()
if document_id and article_id:
found = ARTICLE_SOURCES.get(_compound_key(document_id, article_id))
if found:
return found
if article_id and article_id not in ARTICLE_SOURCE_COLLISIONS:
found = ARTICLE_SOURCES.get(article_id)
if found:
return found
return {}
def _lookup_semantic_unit(document_id: str, article_id: str) -> dict[str, Any]:
if document_id and article_id:
found = SEMANTIC_UNITS.get(_compound_key(document_id, article_id))
if found:
return found
if article_id:
return SEMANTIC_UNITS.get(article_id, {})
return {}
def _compound_key(document_id: str, article_id: str) -> str:
return f"{document_id}::{article_id}"
def _dedupe_key(document_id: str, article_id: str, title: str = "") -> tuple[str, str, str]:
return ((document_id or "").strip(), (article_id or "").strip(), (title or "").strip())
def _metadata_document_id(meta: dict) -> str:
return str(
meta.get("document_id")
or meta.get("document")
or meta.get("law_id")
or ""
).strip()
def _clause_document_id(clause: dict) -> str:
return str(
clause.get("document_id")
or clause.get("document")
or clause.get("law_id")
or ""
).strip()
def _article_from_key(key: str) -> str:
if "::" in key:
return key.split("::", 1)[1]
return key
def _document_from_key(key: str) -> str:
if "::" in key:
return key.split("::", 1)[0]
return ""
def _source_label(document_id: str, article: str, title: str = "", document_title: str = "") -> str:
heading = (document_title or document_id or "").strip()
article_part = (article or "").strip()
title_part = (title or "").strip()
if heading and article_part:
label = f"{heading} — {article_part}"
else:
label = heading or article_part
if title_part:
label = f"{label} | {title_part}" if label else title_part
return escape(label)
def _render_source_block(label: str, body_text: str) -> str:
"""Render one source as a collapsed disclosure block.
The full article text is available for inspection, but the source block starts
collapsed. The ``open`` attribute is intentionally not used.
"""
body = escape(body_text or "")
return (
''
f'Kaynak: {label}
'
''
f"{body}"
"
"
" "
)
def _official_source_link() -> str:
return (
f'Resmi mevzuat kaynağı
'
)
def _missing_source_message(document_id: str, article: str) -> str:
label = " | ".join(part for part in [document_id, article] if part) or "bilinmeyen kaynak"
return _MISSING_FULL_SOURCE_TEMPLATE.format(source_label=label)
def _source_title_from_clause(clause: dict) -> str:
title = str(clause.get("article_title", "") or clause.get("title", "")).strip()
if title:
return title
norm_type = str(clause.get("norm_type", "") or "").strip()
return norm_type
def evaluation_context_text(node_with_score) -> str:
meta = getattr(node_with_score.node, "metadata", {}) or {}
text = source_display_text(node_with_score)
if len(text) > EVAL_CONTEXT_MAX_CHARS:
text = text[:EVAL_CONTEXT_MAX_CHARS].rsplit(" ", 1)[0] + "..."
article = meta.get("article", "") or meta.get("article_id", "")
title = meta.get("title", "") or meta.get("article_title", "")
document_id = _metadata_document_id(meta)
header = " | ".join(part for part in [document_id, article, title] if part)
return f"{header}\n{text}"