| 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 = """<style>
|
| .mitranlil-source {
|
| border: 1px solid #3a3a3f;
|
| border-radius: 8px;
|
| margin-top: 12px;
|
| background: #18181b;
|
| }
|
| details.mitranlil-source > summary.mitranlil-source-title {
|
| font-weight: 700;
|
| padding: 10px 12px;
|
| cursor: pointer;
|
| user-select: none;
|
| }
|
| details.mitranlil-source[open] > summary.mitranlil-source-title {
|
| border-bottom: 1px solid #3a3a3f;
|
| }
|
| details.mitranlil-source:not([open]) > summary.mitranlil-source-title {
|
| border-bottom: none;
|
| }
|
| details.mitranlil-source:not([open]) > .mitranlil-source-body {
|
| display: none !important;
|
| }
|
| .mitranlil-source-body {
|
| max-height: 340px;
|
| overflow-y: auto;
|
| white-space: pre-wrap;
|
| padding: 12px;
|
| line-height: 1.55;
|
| }
|
| </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 "" |
| |
| 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}) |
|
|
| |
| 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: |
| |
| 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)
|
|
|
|
|
| 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:
|
|
|
|
|
| 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 (
|
| '<details class="mitranlil-source">'
|
| f'<summary class="mitranlil-source-title">Kaynak: {label}</summary>'
|
| '<div class="mitranlil-source-body">'
|
| f"{body}"
|
| "</div>"
|
| "</details>"
|
| )
|
|
|
|
|
| def _official_source_link() -> str:
|
| return (
|
| f'<p class="mitranlil-official-source"><a href="{escape(OFFICIAL_SOURCE_URL)}" '
|
| 'target="_blank">Resmi mevzuat kaynağı</a></p>'
|
| )
|
|
|
|
|
| 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}"
|
|
|
