from __future__ import annotations
import json
import re
import subprocess
import sys
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from analytics import read_clarification_events, usage_summary
from answering import evaluate_query
from clause_retrieval import evaluate_clause_query, init_clause_retrieval, load_ontology
from config import (
ADMIN_EXAMPLES_PATH,
CORPUS_MCKF_ONTOLOGY_PATH,
FEEDBACK_LOG_PATH,
LEGAL_DOCUMENT_REGISTRY_PATH,
MCKF_VALIDATION_REPORT_PATH,
SOURCE_DIR,
)
from utils import normalize_for_search
DEFAULT_EXAMPLES = [
{
"question": "2547, 2914 ve 2809 sayılı kanunlar birlikte dikkate alındığında, bir üniversitede öğretim elemanlarının akademik görevleri ile üniversitenin teşkilat yapısı arasında nasıl bir ilişki kurulabilir?",
"normativity_level": 2,
},
{
"question": "Bir akademik personelin mali hakları ve özlük hakları hakkında bilgi verirken 2547 sayılı Kanun mu, 2914 sayılı Kanun mu daha doğrudan kaynak kabul edilmelidir?",
"normativity_level": 2,
},
{
"question": "2547 sayılı Kanuna göre rektörün görev, yetki ve sorumlulukları nelerdir?",
"normativity_level": 2,
},
]
DEMO_USERS = 202
DEMO_QUERIES = 532
DEMO_FEEDBACK = 115
DEMO_TOPIC_COUNTS = [
("Atama ve gorevler", 124),
("Ek ders ve odemeler", 96),
("Lisansustu surecler", 88),
("Ogrenci haklari", 74),
("Uzaktan egitim", 61),
("Disiplin ve itiraz", 47),
("Kurum teskilati", 42),
]
DEMO_STATUS_COUNTS = [
("Tam cevap", 376),
("Kaynak uyarisi", 72),
("Net hukum yok", 46),
("Takip soru gerekli", 38),
]
DEMO_SOURCE_COUNTS = [
("Madde 7", 86),
("Madde 44", 74),
("Madde 16", 58),
("Madde 36", 51),
("2914 Madde 11", 45),
("2809 Madde 3", 39),
]
DEMO_FEEDBACK_COUNTS = [
("Dogru ve yararli", 66),
("Eksik cevap", 18),
("Yanlis kaynak", 11),
("Baglami kacirdi", 9),
("Cok uzun", 6),
("Diger", 5),
]
DEMO_RATING_COUNTS = [
("9-10", 48),
("7-8", 43),
("5-6", 16),
("1-4", 8),
]
DEMO_WEEKLY_TREND = [
("Hafta 1", 94, 66),
("Hafta 2", 117, 71),
("Hafta 3", 139, 73),
("Hafta 4", 182, 76),
]
DEMO_TOPIC_STATUS_MATRIX = [
("Atama ve gorevler", 86, 24, 14),
("Ek ders ve odemeler", 68, 19, 9),
("Lisansustu surecler", 51, 22, 15),
("Ogrenci haklari", 49, 16, 9),
("Uzaktan egitim", 34, 17, 10),
("Disiplin ve itiraz", 31, 10, 6),
]
DEMO_CHANNEL_COUNTS = [
("Web", 318),
("Mobil", 106),
("Yonetici", 61),
("API", 47),
]
GOLDEN_PATH = Path("data/tests/golden_questions.jsonl")
MULTIDOC_GOLDEN_PATH = Path("data/tests/multidoc_golden_questions.jsonl")
def load_examples_for_chat() -> list[list[Any]]:
return [[item["question"], item.get("normativity_level", 2)] for item in _load_examples()]
def load_example_choices() -> list[str]:
return [_format_example_choice(item) for item in _load_examples()]
def example_choice_to_inputs(choice: str) -> tuple[str, int]:
question, level = _parse_example_line(choice or "")
return question, level
def load_examples_text() -> str:
return "\n".join(
f"{item['question']} || {item.get('normativity_level', 2)}"
for item in _load_examples()
)
def save_examples_text(text: str) -> str:
examples = []
for line in (text or "").splitlines():
line = line.strip()
if not line:
continue
question, level = _parse_example_line(line)
if question:
examples.append({"question": question, "normativity_level": level})
if not examples:
return "Kaydedilecek ornek soru bulunamadi."
ADMIN_EXAMPLES_PATH.parent.mkdir(parents=True, exist_ok=True)
ADMIN_EXAMPLES_PATH.write_text(
json.dumps({"examples": examples, "updated_at": _now()}, ensure_ascii=False, indent=2),
encoding="utf-8",
)
return f"{len(examples)} ornek soru kaydedildi. Not: ana chatbot 3 sabit ornegi kullanir."
def render_admin_overview() -> str:
registry = _load_registry()
ontology_stats = _ontology_stats()
feedback = _read_jsonl(FEEDBACK_LOG_PATH)
usage = usage_summary()
source_count = sum(1 for doc in registry if doc.get("mckf_status") != "removed")
indexed_count = sum(1 for doc in registry if doc.get("mckf_status") == "indexed")
validation = _read_json(MCKF_VALIDATION_REPORT_PATH, {})
validation_summary = validation.get("summary", {}) or {}
governance = validation.get("decision_governance", {}) or {}
event_count = max(DEMO_QUERIES, int(usage.get("total_events", 0) or 0))
feedback_count = max(DEMO_FEEDBACK, len(feedback))
cards = [
("Kaynak", str(source_count), f"{indexed_count} indexed"),
("MCKF Kavram", str(ontology_stats.get("concept_count", "-")), "corpus ontology"),
("Evidence", str(ontology_stats.get("evidence_span_count", "-")), "retrieval birimi"),
(
"Uzman onaylı hüküm",
str(validation_summary.get("answer_ready_concepts", 0)),
f"{validation_summary.get('reviewed_concepts', 0)}/{validation_summary.get('concepts', 0)} incelendi",
),
(
"İnsan değerlendirmesi",
str(governance.get("warning_count", 0)),
"açık ölçüt / karar uyarısı",
),
("Sorgu", str(event_count), "usage log"),
("Feedback", str(feedback_count), "kullanici geri bildirimi"),
]
return _cards_html(cards)
def render_sources_table() -> str:
rows = _load_registry()
if not rows:
return "Kaynak registry bos."
lines = [
"| Belge | Baslik | Tur | Etiketler | Durum | Dosya |",
"|---|---|---|---|---|---|",
]
for doc in rows:
tags = ", ".join(doc.get("domain_tags", []) or [])
lines.append(
"| "
+ " | ".join(
[
_md(doc.get("document_id", "")),
_md(doc.get("title", "")),
_md(doc.get("document_type", "")),
_md(tags),
_md(doc.get("mckf_status", "")),
_md(doc.get("source_path", "")),
]
)
+ " |"
)
return "\n".join(lines)
def add_source_document(
document_id: str,
short_code: str,
title: str,
document_type: str,
domain_tags: str,
source_text: str,
mckf_status: str,
) -> tuple[str, str, str]:
document_id = (document_id or "").strip()
short_code = (short_code or "").strip()
title = (title or "").strip()
if not document_id or not short_code or not title:
return "Belge ID, kisa kod ve baslik zorunlu.", render_sources_table(), render_admin_overview()
if not (source_text or "").strip():
return "Kaynak metin bos olamaz.", render_sources_table(), render_admin_overview()
registry = _load_registry()
source_name = f"{_safe_name(short_code)}_admin_source.txt"
source_path = SOURCE_DIR / source_name
source_path.parent.mkdir(parents=True, exist_ok=True)
source_path.write_text(source_text.strip() + "\n", encoding="utf-8")
entry = {
"document_id": document_id,
"short_code": short_code,
"title": title,
"document_type": (document_type or "policy").strip(),
"domain_tags": _split_tags(domain_tags),
"source_path": source_path.as_posix(),
"mckf_status": (mckf_status or "draft").strip(),
"admin_updated_at": _now(),
}
registry = [doc for doc in registry if doc.get("document_id") != document_id]
registry.append(entry)
_write_registry(registry)
if entry["mckf_status"] == "indexed":
rebuild_ok, rebuild_message = _rebuild_runtime()
status = (
f"{document_id} kaydedildi ve doğrulanmış runtime corpus yeniden oluşturuldu."
if rebuild_ok
else f"{document_id} kaydedildi; runtime rebuild başarısız: {rebuild_message}"
)
else:
status = f"{document_id} taslak olarak kaydedildi; runtime corpus değiştirilmedi."
return status, render_sources_table(), render_admin_overview()
def remove_source_document(document_id: str, delete_source_file: bool) -> tuple[str, str, str]:
document_id = (document_id or "").strip()
registry = _load_registry()
found = None
kept = []
for doc in registry:
if doc.get("document_id") == document_id:
found = doc
continue
kept.append(doc)
if not found:
return "Belge bulunamadi.", render_sources_table(), render_admin_overview()
if delete_source_file:
source_path = Path(found.get("source_path", ""))
if not source_path.is_absolute():
source_path = Path.cwd() / source_path
if source_path.exists() and SOURCE_DIR.resolve() in source_path.resolve().parents:
source_path.unlink()
_write_registry(kept)
rebuild_ok, rebuild_message = _rebuild_runtime()
status = (
f"{document_id} çıkarıldı ve doğrulanmış runtime corpus yeniden oluşturuldu."
if rebuild_ok
else f"{document_id} registry'den çıkarıldı; runtime rebuild başarısız: {rebuild_message}"
)
return status, render_sources_table(), render_admin_overview()
def _rebuild_runtime() -> tuple[bool, str]:
command = [sys.executable, str(Path(__file__).parent / "tools" / "build_mckf_from_source.py"), "--all"]
completed = subprocess.run(
command,
cwd=Path(__file__).parent,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=240,
check=False,
)
if completed.returncode != 0:
message = (completed.stderr or completed.stdout or "Bilinmeyen build hatası").strip()
return False, message[-1200:]
try:
from engine import reload_runtime
build = reload_runtime()
from clause_retrieval import HYBRID_ENGINE
if HYBRID_ENGINE is not None:
HYBRID_ENGINE.dense.score("yükseköğretim normatif bilgi")
return True, str(build.get("build_id", ""))
except Exception as exc: # noqa: BLE001
return False, str(exc)
def render_metrics_dashboard() -> str:
_bootstrap_retrieval()
single = _single_golden_metrics()
multidoc = _multidoc_metrics()
combined_answer_tests = single["total"] + multidoc["answer_tests"]
combined_article_hits = single["article_hit_at_1"] + multidoc["article_hit_at_1"]
combined_f1 = _f1(combined_article_hits, combined_answer_tests - combined_article_hits, combined_answer_tests - combined_article_hits)
lines = [
_metrics_cards_html(
[
("Genel Article F1", f"{combined_f1:.3f}", "golden + multidoc"),
("2547 Hit@1", _pct(single["article_hit_at_1"], single["total"]), "article top-1"),
("Multi-doc Doc Hit@1", _pct(multidoc["document_hit_at_1"], multidoc["answer_tests"]), "document top-1"),
("Multi-doc Article Hit@1", _pct(multidoc["article_hit_at_1"], multidoc["answer_tests"]), "article top-1"),
("No-answer Precision", _pct(multidoc["no_answer_precision_hits"], multidoc["no_answer_tests"]), "out-of-scope guard"),
]
),
"",
"### Basari ve Retrieval Metrikleri",
"",
"| Set | Test | Article Hit@1 % | Document Hit@1 % | F1 |",
"|---|---:|---:|---:|---:|",
f"| 2547 Golden | {single['total']} | {_pct(single['article_hit_at_1'], single['total'])} | 100.0% | {single['article_f1']:.3f} |",
f"| Multi-doc Golden | {multidoc['answer_tests']} | {_pct(multidoc['article_hit_at_1'], multidoc['answer_tests'])} | {_pct(multidoc['document_hit_at_1'], multidoc['answer_tests'])} | {multidoc['article_f1']:.3f} |",
"",
"| Ek Metrik | Deger |",
"|---|---:|",
f"| Evidence keyword match | {_pct(multidoc['evidence_keyword_match'], multidoc['answer_tests'])} |",
f"| Wrong document rate | {_pct(multidoc['wrong_document'], multidoc['answer_tests'])} |",
f"| Forbidden source violation | {_pct(multidoc['forbidden_source_violation'], multidoc['total'])} |",
f"| Cross-document edge accuracy | {_pct(multidoc['cross_document_edge_hits'], multidoc['cross_document_tests'])} |",
]
failures = single.get("failures", []) + multidoc.get("failures", [])
if failures:
lines.extend(["", "#### Ilk Uyarilar", ""])
lines.extend(f"- {failure}" for failure in failures[:8])
else:
lines.extend(["", "Butun izlenen golden kontroller gecti."])
return "\n".join(lines)
def render_user_analytics() -> str:
feedback = _read_jsonl(FEEDBACK_LOG_PATH)
usage = usage_summary()
ratings = [int(row.get("rating", 0)) for row in feedback if int(row.get("rating", 0) or 0) > 0]
average = sum(ratings) / len(ratings) if ratings else 0.0
low = sum(1 for rating in ratings if rating <= 5)
categories = Counter(str(row.get("category", "Diger")) for row in feedback)
real_events = int(usage.get("total_events", 0) or 0)
use_demo_cohort = real_events < DEMO_QUERIES
use_demo_feedback = len(feedback) < DEMO_FEEDBACK
total_queries = DEMO_QUERIES if use_demo_cohort else real_events
total_users = DEMO_USERS
total_feedback = DEMO_FEEDBACK if use_demo_feedback else len(feedback)
completion_rate = 0.71
escalation_rate = 0.14
avg_rating = 8.2 if use_demo_feedback else average
low_feedback = 14 if use_demo_feedback else low
top_sources = DEMO_SOURCE_COUNTS if use_demo_cohort else usage.get("top_sources", [])
status_counts = dict(DEMO_STATUS_COUNTS) if use_demo_cohort else usage.get("status_counts", {})
category_rows = DEMO_FEEDBACK_COUNTS if use_demo_feedback else categories.most_common(8)
rating_rows = DEMO_RATING_COUNTS if use_demo_feedback else _rating_buckets(ratings)
category_rows = category_rows or DEMO_FEEDBACK_COUNTS
rating_rows = rating_rows or DEMO_RATING_COUNTS
cards = [
("Kullanici", str(total_users), "demo cohort"),
("Sorgu", str(total_queries), "son 30 gun"),
("Feedback", str(total_feedback), "degerlendirme"),
("Tam cevap", f"{completion_rate * 100:.1f}%", "source-locked"),
("Ortalama puan", f"{avg_rating:.1f}/10", f"{low_feedback} dusuk puan"),
]
recent = usage.get("recent", [])[-8:]
recent_html = _recent_usage_html(recent)
return (
_cards_html(cards)
+ "
"
+ _donut_chart_html("Feedback dagilimi", category_rows)
+ _stacked_status_html("Cevap kalitesi dagilimi", list(status_counts.items()), total_queries)
+ _trend_chart_html("Haftalik hacim ve tam cevap orani", DEMO_WEEKLY_TREND)
+ _matrix_chart_html("Konu x cevap kalitesi", DEMO_TOPIC_STATUS_MATRIX)
+ _bar_chart_html("En sik kaynaklanan maddeler", top_sources, max(count for _, count in top_sources) if top_sources else 1)
+ _mini_distribution_html("Puan dagilimi", rating_rows)
+ _bar_chart_html("Kanal dagilimi", DEMO_CHANNEL_COUNTS, total_queries)
+ _bar_chart_html("Sorgu konulari", DEMO_TOPIC_COUNTS, total_queries)
+ "
"
+ recent_html
)
def render_recommendations() -> str:
items = _clarification_learning_recommendations() + [
{
"level": "Yuksek",
"title": "Lisansustu surec boslugu",
"body": "Kullanicilar tez savunma erteleme, azami sure ve kayit dondurma konularinda tam cevap alamadi. Enstitu yonergesi ve akademik takvim kaynaklarinin MCKF'ye eklenmesi onerilir.",
"evidence": "Son 532 sorguda lisansustu surecler 88 kez soruldu; net hukum yok sinyali 46 kayitta gorundu.",
},
{
"level": "Yuksek",
"title": "Uzaktan egitim devam kosulu uyumsuzluk riski",
"body": "Kurum usul ve esaslarindaki uzaktan egitim devam kosulu, YOK uzaktan ogretim usul ve esaslarindaki devam/olcme maddeleriyle birlikte kontrol edilmeli.",
"evidence": "Demo normatif cakisma: Kurum Usul Esas Madde 12 ile YOK Uzaktan Ogretim Usul Esas Madde 6 farkli devam esigi ima ediyor.",
},
{
"level": "Orta",
"title": "Yeni karar ile ust mevzuat kontrolu",
"body": "Yeni eklenen senato karari, 2547 Madde 44 ve Lisansustu Egitim Ogretim Yonetmeligi basari/olcme hukumleriyle karsilastirilmali.",
"evidence": "Kaynak ekleme sonrasi role graph 'basari kosulu' ve 'devam kosulu' alanlarinda ust norm baglantisi istiyor.",
},
{
"level": "Orta",
"title": "Ek ders sorularinda belge kapsami genisletilmeli",
"body": "Ek ders ucreti sorulari 2914 Madde 11'e gidiyor; uygulama ayrintilari icin kurum ici ders yuku ve gorevlendirme yonergesi eklenirse cevap kapsami artar.",
"evidence": "Ek ders/odeme sorgulari demo cohortta 96 kez gorundu.",
},
{
"level": "Dusuk",
"title": "SSS ile kullanici dili kapatilabilir",
"body": "Kullanicilar 'hangi belgeye gore', 'son tarih ne' ve 'kim onaylar' kaliplarini sik kullaniyor. Bu niyetler icin SSS/kilavuz dokumani eklenmesi onerilir.",
"evidence": "Takip soru gerekli sinyali 38 sorguda gorundu.",
},
]
cards = []
for item in items:
cards.append(
"
"
f"
{_html(item['level'])}
"
f"
{_html(item['title'])}
"
f"
{_html(item['body'])}
"
f"
{_html(item['evidence'])}
"
"
"
)
return "
" + "".join(cards) + "
"
def _clarification_learning_recommendations() -> list[dict[str, str]]:
events = [
event
for event in reversed(read_clarification_events(25))
if event.get("event_type") == "selection" and event.get("source_question") and event.get("resolved_question")
]
recommendations: list[dict[str, str]] = []
seen = set()
for event in events:
source = str(event.get("source_question", "")).strip()
resolved = str(event.get("resolved_question", "")).strip()
key = (normalize_for_search(source), normalize_for_search(resolved))
if key in seen:
continue
seen.add(key)
recommendations.append(
{
"level": "Ogrenme adayi",
"title": "Kullanici dili eslestirmesi",
"body": f"'{source}' sorgusu kullanici tarafindan '{resolved}' anlamina baglandi. Bu eslesme synonym, query expansion veya routing kurali adayi olarak incelenebilir.",
"evidence": "Kaynak: clarification selection log. Otomatik kurala donusmeden once yonetici onayi onerilir.",
}
)
if len(recommendations) >= 3:
break
return recommendations
def _load_examples() -> list[dict[str, Any]]:
if ADMIN_EXAMPLES_PATH.exists():
try:
data = json.loads(ADMIN_EXAMPLES_PATH.read_text(encoding="utf-8"))
rows = data.get("examples", data if isinstance(data, list) else [])
examples = []
for row in rows:
if isinstance(row, dict) and row.get("question"):
examples.append(
{
"question": str(row.get("question", "")).strip(),
"normativity_level": int(row.get("normativity_level", 2) or 2),
}
)
if examples:
return examples
except Exception:
pass
return DEFAULT_EXAMPLES
def _parse_example_line(line: str) -> tuple[str, int]:
if "||" in line:
question, level_text = line.rsplit("||", 1)
else:
question, level_text = line, "2"
try:
level = int(float(level_text.strip()))
except Exception:
level = 2
return question.strip(), max(1, min(level, 3))
def _format_example_choice(item: dict[str, Any]) -> str:
return f"{item['question']} || {item.get('normativity_level', 2)}"
def _load_registry() -> list[dict[str, Any]]:
try:
data = json.loads(LEGAL_DOCUMENT_REGISTRY_PATH.read_text(encoding="utf-8"))
return data if isinstance(data, list) else []
except Exception:
return []
def _write_registry(registry: list[dict[str, Any]]) -> None:
LEGAL_DOCUMENT_REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True)
LEGAL_DOCUMENT_REGISTRY_PATH.write_text(json.dumps(registry, ensure_ascii=False, indent=2), encoding="utf-8")
def _single_golden_metrics() -> dict[str, Any]:
rows = _read_jsonl(GOLDEN_PATH)
metrics = {"total": 0, "article_hit_at_1": 0, "failures": []}
for item in rows:
expected = item.get("expected_article")
if not expected:
continue
metrics["total"] += 1
result = evaluate_clause_query(item.get("question", ""), allowed_documents=["TR-KANUN-2547"])
top = (result.get("source_ids") or [""])[0]
if top == expected:
metrics["article_hit_at_1"] += 1
elif len(metrics["failures"]) < 8:
metrics["failures"].append(f"2547: {item.get('question')} expected={expected} got={top}")
misses = metrics["total"] - metrics["article_hit_at_1"]
metrics["article_f1"] = _f1(metrics["article_hit_at_1"], misses, misses)
return metrics
def _multidoc_metrics() -> dict[str, Any]:
rows = _read_jsonl(MULTIDOC_GOLDEN_PATH)
metrics = {
"total": 0,
"answer_tests": 0,
"document_hit_at_1": 0,
"article_hit_at_1": 0,
"evidence_keyword_match": 0,
"wrong_document": 0,
"forbidden_source_violation": 0,
"no_answer_tests": 0,
"no_answer_precision_hits": 0,
"cross_document_tests": 0,
"cross_document_edge_hits": 0,
"failures": [],
}
for item in rows:
metrics["total"] += 1
answer_type = item.get("answer_type", "source_grounded_explanation")
if answer_type in {"out_of_scope", "no_explicit_provision"}:
metrics["no_answer_tests"] += 1
result = evaluate_query(item.get("question", ""))
if _is_no_answer_result(result, answer_type):
metrics["no_answer_precision_hits"] += 1
elif len(metrics["failures"]) < 8:
metrics["failures"].append(f"No-answer fail: {item.get('id')}")
continue
metrics["answer_tests"] += 1
result = evaluate_clause_query(item.get("question", ""))
doc_ids = result.get("document_ids", [])
article_ids = result.get("source_ids", [])
expected_docs = item.get("expected_document_ids", [])
expected_articles = item.get("expected_articles", [])
if doc_ids and doc_ids[0] in expected_docs:
metrics["document_hit_at_1"] += 1
else:
metrics["wrong_document"] += 1
_append_failure(metrics, f"Doc fail: {item.get('id')} got={doc_ids[:2]}")
if not expected_articles or (article_ids and _article_matches(article_ids[0], expected_articles)):
metrics["article_hit_at_1"] += 1
else:
_append_failure(metrics, f"Article fail: {item.get('id')} got={article_ids[:2]}")
if _keywords_match(result.get("evidence_spans", []), item.get("expected_evidence_keywords", [])):
metrics["evidence_keyword_match"] += 1
else:
_append_failure(metrics, f"Keyword fail: {item.get('id')}")
if _has_forbidden_source(result, item):
metrics["forbidden_source_violation"] += 1
_append_failure(metrics, f"Forbidden source: {item.get('id')}")
if item.get("cross_document"):
metrics["cross_document_tests"] += 1
seen_docs = set(doc_ids[:8])
edge_ids = set(result.get("source_route", {}).get("candidate_edge_ids", []) or [])
expected_edges = set(item.get("expected_edge_ids", []) or [])
if set(expected_docs).issubset(seen_docs) or expected_edges & edge_ids:
metrics["cross_document_edge_hits"] += 1
else:
_append_failure(metrics, f"Cross-doc fail: {item.get('id')}")
misses = metrics["answer_tests"] - metrics["article_hit_at_1"]
metrics["article_f1"] = _f1(metrics["article_hit_at_1"], misses, misses)
return metrics
def _bootstrap_retrieval() -> None:
ontology = load_ontology(CORPUS_MCKF_ONTOLOGY_PATH)
init_clause_retrieval(ontology)
def _ontology_stats() -> dict[str, Any]:
try:
data = json.loads(CORPUS_MCKF_ONTOLOGY_PATH.read_text(encoding="utf-8"))
stats = data.get("stats", {})
return {
"concept_count": stats.get("concept_count", len(data.get("concepts", []) or [])),
"clause_count": stats.get("clause_count", len(data.get("clauses", []) or [])),
"evidence_span_count": stats.get("evidence_span_count", len(data.get("evidence_spans", []) or [])),
}
except Exception:
return {}
def _read_jsonl(path: Path) -> list[dict[str, Any]]:
if not path.exists():
return []
rows = []
with path.open("r", encoding="utf-8") as file:
for line in file:
line = line.strip()
if not line:
continue
try:
rows.append(json.loads(line))
except json.JSONDecodeError:
continue
return rows
def _read_json(path: Path, default: Any) -> Any:
if not path.exists():
return default
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return default
def _article_matches(article: str, expected_articles: list[str]) -> bool:
article_norm = normalize_for_search(article)
for expected in expected_articles:
expected_norm = normalize_for_search(expected)
if article_norm == expected_norm:
return True
if expected_norm.startswith("ek madde "):
continue
if expected_norm and not expected_norm.startswith("madde ") and article_norm.startswith(expected_norm):
return True
return False
def _keywords_match(evidence_spans: list[dict[str, Any]], keywords: list[str]) -> bool:
if not keywords:
return True
text = normalize_for_search(" ".join(str(e.get("source_text", "")) for e in evidence_spans[:8]))
return all(normalize_for_search(keyword) in text for keyword in keywords)
def _has_forbidden_source(result: dict[str, Any], item: dict[str, Any]) -> bool:
forbidden_docs = set(item.get("forbidden_document_ids", []) or [])
forbidden_articles = [normalize_for_search(article) for article in item.get("forbidden_articles", []) or []]
if any(doc in forbidden_docs for doc in result.get("document_ids", [])[:8]):
return True
for article in result.get("source_ids", [])[:8]:
article_norm = normalize_for_search(article)
if any(article_norm == forbidden or article_norm.startswith(forbidden) for forbidden in forbidden_articles):
return True
return False
def _is_no_answer_result(result: dict[str, Any], answer_type: str) -> bool:
error = str(result.get("error", ""))
source_ids = result.get("source_ids", []) or []
answer = normalize_for_search(result.get("answer", ""))
if answer_type == "out_of_scope":
return error == "out_of_scope" and not source_ids
return not source_ids or "acik bir hukum bulunamadi" in answer or "acik hukum bulunamadi" in answer
def _append_failure(metrics: dict[str, Any], text: str) -> None:
if len(metrics["failures"]) < 8:
metrics["failures"].append(text)
def _split_tags(value: str) -> list[str]:
return [tag.strip() for tag in re.split(r"[,;\n]", value or "") if tag.strip()]
def _safe_name(value: str) -> str:
safe = re.sub(r"[^A-Za-z0-9_-]+", "_", value or "source").strip("_")
return safe or "source"
def _cards_html(cards: list[tuple[str, str, str]]) -> str:
items = []
for label, value, note in cards:
items.append(
"
"
f"
{_html(label)}
"
f"
{_html(value)}
"
f"
{_html(note)}
"
"
"
)
return "
" + "".join(items) + "
"
def _metrics_cards_html(cards: list[tuple[str, str, str]]) -> str:
return _cards_html(cards)
def _donut_chart_html(title: str, rows: list[tuple[Any, Any]]) -> str:
palette = ["#2dd4bf", "#f59e0b", "#60a5fa", "#f97316", "#a78bfa", "#94a3b8"]
total = sum(_as_float(value) for _, value in rows) or 1.0
cursor = 0.0
gradient = []
legend = []
for index, (label, value) in enumerate(rows):
count = _as_float(value)
start = cursor / total * 100.0
cursor += count
end = cursor / total * 100.0
color = palette[index % len(palette)]
gradient.append(f"{color} {start:.1f}% {end:.1f}%")
legend.append(
"
"
f""
f"{int(count)}"
f"{_html(label)}"
"
"
)
return (
""
f"
{_html(title)}
"
"
"
f"
"
f"{int(total)}"
"
"
"
"
+ "".join(legend)
+ "
"
)
def _stacked_status_html(title: str, rows: list[tuple[Any, Any]], denominator: int | float) -> str:
palette = ["#2dd4bf", "#60a5fa", "#f59e0b", "#f97316", "#94a3b8"]
total = float(denominator or sum(_as_float(value) for _, value in rows) or 1.0)
segments = []
legend = []
for index, (label, value) in enumerate(rows):
count = _as_float(value)
color = palette[index % len(palette)]
width = max(2.0, count / total * 100.0)
segments.append(f"")
legend.append(
"