| 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:
|
| 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)
|
| + "<div class='mitranlil-analytics-grid'>"
|
| + _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)
|
| + "</div>"
|
| + 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(
|
| "<div class='mitranlil-recommendation'>"
|
| f"<div class='mitranlil-rec-level'>{_html(item['level'])}</div>"
|
| f"<div class='mitranlil-rec-title'>{_html(item['title'])}</div>"
|
| f"<div class='mitranlil-rec-body'>{_html(item['body'])}</div>"
|
| f"<div class='mitranlil-rec-evidence'>{_html(item['evidence'])}</div>"
|
| "</div>"
|
| )
|
| return "<div class='mitranlil-rec-grid'>" + "".join(cards) + "</div>"
|
|
|
|
|
| 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(
|
| "<div class='mitranlil-admin-card'>"
|
| f"<div class='mitranlil-admin-label'>{_html(label)}</div>"
|
| f"<div class='mitranlil-admin-value'>{_html(value)}</div>"
|
| f"<div class='mitranlil-admin-note'>{_html(note)}</div>"
|
| "</div>"
|
| )
|
| return "<div class='mitranlil-admin-grid'>" + "".join(items) + "</div>"
|
|
|
|
|
| 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(
|
| "<div class='mitranlil-donut-legend-row'>"
|
| f"<span style='background:{color}'></span>"
|
| f"<strong>{int(count)}</strong>"
|
| f"<em>{_html(label)}</em>"
|
| "</div>"
|
| )
|
| return (
|
| "<section class='mitranlil-chart-card mitranlil-donut-card'>"
|
| f"<h3>{_html(title)}</h3>"
|
| "<div class='mitranlil-donut-wrap'>"
|
| f"<div class='mitranlil-donut' style='background: conic-gradient({', '.join(gradient)})'>"
|
| f"<span>{int(total)}</span>"
|
| "</div>"
|
| "<div class='mitranlil-donut-legend'>"
|
| + "".join(legend)
|
| + "</div></div></section>"
|
| )
|
|
|
|
|
| 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"<span style='width:{width:.1f}%; background:{color}'></span>")
|
| legend.append(
|
| "<div class='mitranlil-stack-legend-row'>"
|
| f"<span style='background:{color}'></span>"
|
| f"<em>{_html(label)}</em>"
|
| f"<strong>{int(count)}</strong>"
|
| "</div>"
|
| )
|
| return (
|
| "<section class='mitranlil-chart-card'>"
|
| f"<h3>{_html(title)}</h3>"
|
| "<div class='mitranlil-stacked-bar'>"
|
| + "".join(segments)
|
| + "</div><div class='mitranlil-stack-legend'>"
|
| + "".join(legend)
|
| + "</div>"
|
| f"<div class='mitranlil-chart-note'>Inceleme onerisi: {int(total * 0.14)} sorgu</div>"
|
| "</section>"
|
| )
|
|
|
|
|
| def _trend_chart_html(title: str, rows: list[tuple[str, int, int]]) -> str:
|
| width = 420
|
| height = 160
|
| pad_x = 28
|
| pad_y = 18
|
| max_query = max((query_count for _, query_count, _ in rows), default=1)
|
| points_query = []
|
| points_success = []
|
| labels = []
|
| for index, (label, query_count, success_rate) in enumerate(rows):
|
| x = pad_x + index * ((width - pad_x * 2) / max(1, len(rows) - 1))
|
| y_query = height - pad_y - (query_count / max_query) * (height - pad_y * 2)
|
| y_success = height - pad_y - (success_rate / 100.0) * (height - pad_y * 2)
|
| points_query.append(f"{x:.1f},{y_query:.1f}")
|
| points_success.append(f"{x:.1f},{y_success:.1f}")
|
| labels.append(f"<text x='{x:.1f}' y='{height - 2}' text-anchor='middle'>{_html(label.replace('Hafta ', 'H'))}</text>")
|
| return (
|
| "<section class='mitranlil-chart-card mitranlil-trend-card'>"
|
| f"<h3>{_html(title)}</h3>"
|
| f"<svg viewBox='0 0 {width} {height}' role='img'>"
|
| "<line x1='24' y1='142' x2='400' y2='142'></line>"
|
| f"<polyline class='query' points='{' '.join(points_query)}'></polyline>"
|
| f"<polyline class='success' points='{' '.join(points_success)}'></polyline>"
|
| + "".join(labels)
|
| + "</svg>"
|
| "<div class='mitranlil-trend-legend'><span class='query'></span>Sorgu hacmi <span class='success'></span>Tam cevap orani</div>"
|
| "</section>"
|
| )
|
|
|
|
|
| def _matrix_chart_html(title: str, rows: list[tuple[str, int, int, int]]) -> str:
|
| max_value = max((max(values) for _, *values in rows), default=1)
|
| lines = [
|
| "<section class='mitranlil-table-card mitranlil-matrix-card'>",
|
| f"<h3>{_html(title)}</h3>",
|
| "<table><thead><tr><th>Konu</th><th>Tam</th><th>Uyari</th><th>Bosluk</th></tr></thead><tbody>",
|
| ]
|
| for topic, full, warning, gap in rows:
|
| lines.append(
|
| "<tr>"
|
| f"<td>{_html(topic)}</td>"
|
| + _heat_cell(full, max_value, "#2dd4bf")
|
| + _heat_cell(warning, max_value, "#f59e0b")
|
| + _heat_cell(gap, max_value, "#f97316")
|
| + "</tr>"
|
| )
|
| lines.append("</tbody></table></section>")
|
| return "".join(lines)
|
|
|
|
|
| def _mini_distribution_html(title: str, rows: list[tuple[Any, Any]]) -> str:
|
| max_value = max((_as_float(value) for _, value in rows), default=1.0)
|
| columns = []
|
| for label, value in rows:
|
| count = _as_float(value)
|
| height = max(12.0, count / max_value * 100.0)
|
| columns.append(
|
| "<div class='mitranlil-mini-col'>"
|
| f"<div class='mitranlil-mini-value'>{int(count)}</div>"
|
| f"<div class='mitranlil-mini-bar' style='height:{height:.1f}%'></div>"
|
| f"<div class='mitranlil-mini-label'>{_html(label)}</div>"
|
| "</div>"
|
| )
|
| return (
|
| "<section class='mitranlil-chart-card mitranlil-mini-card'>"
|
| f"<h3>{_html(title)}</h3>"
|
| "<div class='mitranlil-mini-bars'>"
|
| + "".join(columns)
|
| + "</div></section>"
|
| )
|
|
|
|
|
| def _bar_chart_html(title: str, rows: list[tuple[Any, Any]], denominator: int | float) -> str:
|
| denominator = float(denominator or 1)
|
| bars = []
|
| for label, value in rows:
|
| try:
|
| count = float(value)
|
| except Exception:
|
| count = 0.0
|
| width = max(4.0, min(100.0, count / denominator * 100.0))
|
| bars.append(
|
| "<div class='mitranlil-bar-row'>"
|
| f"<div class='mitranlil-bar-label'>{_html(label)}</div>"
|
| "<div class='mitranlil-bar-track'>"
|
| f"<div class='mitranlil-bar-fill' style='width:{width:.1f}%'></div>"
|
| "</div>"
|
| f"<div class='mitranlil-bar-value'>{int(count)}</div>"
|
| "</div>"
|
| )
|
| return (
|
| "<section class='mitranlil-chart-card'>"
|
| f"<h3>{_html(title)}</h3>"
|
| + "".join(bars)
|
| + "</section>"
|
| )
|
|
|
|
|
| def _recent_usage_html(recent: list[dict[str, Any]]) -> str:
|
| if not recent:
|
| recent = [
|
| {"created_at": "2026-07-09T09:12:00", "status": "source_grounded_answer", "question": "Tez savunmasi ertelenebilir mi?", "top_source": "LEE Yonetmeligi"},
|
| {"created_at": "2026-07-09T09:19:00", "status": "no_clear_provision", "question": "Uzaktan derste devam zorunlulugu nasil hesaplanir?", "top_source": ""},
|
| {"created_at": "2026-07-09T09:27:00", "status": "source_grounded_answer", "question": "Ek ders ucreti kac saat odenir?", "top_source": "2914 Madde 11"},
|
| ]
|
| rows = []
|
| for item in reversed(recent[-8:]):
|
| rows.append(
|
| "<tr>"
|
| f"<td>{_html(str(item.get('created_at', ''))[:19])}</td>"
|
| f"<td>{_html(item.get('status', ''))}</td>"
|
| f"<td>{_html(item.get('question', ''))}</td>"
|
| f"<td>{_html(item.get('top_source', ''))}</td>"
|
| "</tr>"
|
| )
|
| return (
|
| "<section class='mitranlil-table-card'>"
|
| "<h3>Son sorgular</h3>"
|
| "<table><thead><tr><th>Zaman</th><th>Durum</th><th>Soru</th><th>Kaynak</th></tr></thead>"
|
| "<tbody>"
|
| + "".join(rows)
|
| + "</tbody></table></section>"
|
| )
|
|
|
|
|
| def _metric_line(label: str, value: float) -> str:
|
| return f"- **{label}:** {value:.3f}"
|
|
|
|
|
| def _ratio(numerator: int, denominator: int) -> float:
|
| return numerator / denominator if denominator else 1.0
|
|
|
|
|
| def _f1(tp: int, fp: int, fn: int) -> float:
|
| precision = tp / (tp + fp) if tp + fp else 1.0
|
| recall = tp / (tp + fn) if tp + fn else 1.0
|
| return 2 * precision * recall / (precision + recall) if precision + recall else 0.0
|
|
|
|
|
| def _pct(numerator: int, denominator: int) -> str:
|
| return f"{_ratio(numerator, denominator) * 100:.1f}%"
|
|
|
|
|
| def _rating_buckets(ratings: list[int]) -> list[tuple[str, int]]:
|
| return [
|
| ("9-10", sum(1 for rating in ratings if rating >= 9)),
|
| ("7-8", sum(1 for rating in ratings if 7 <= rating <= 8)),
|
| ("5-6", sum(1 for rating in ratings if 5 <= rating <= 6)),
|
| ("1-4", sum(1 for rating in ratings if 1 <= rating <= 4)),
|
| ]
|
|
|
|
|
| def _heat_cell(value: int, max_value: int, color: str) -> str:
|
| rgb_map = {
|
| "#2dd4bf": "45, 212, 191",
|
| "#f59e0b": "245, 158, 11",
|
| "#f97316": "249, 115, 22",
|
| }
|
| opacity = 0.18 + 0.62 * (value / max_value if max_value else 0)
|
| return f"<td style='background: rgba({rgb_map.get(color, '148, 163, 184')}, {opacity:.2f})'>{int(value)}</td>"
|
|
|
|
|
| def _as_float(value: Any) -> float:
|
| try:
|
| return float(value)
|
| except Exception:
|
| return 0.0
|
|
|
|
|
| def _md(value: Any) -> str:
|
| return str(value or "").replace("|", "\\|").replace("\n", " ")
|
|
|
|
|
| def _html(value: Any) -> str:
|
| return (
|
| str(value or "")
|
| .replace("&", "&")
|
| .replace("<", "<")
|
| .replace(">", ">")
|
| .replace('"', """)
|
| )
|
|
|
|
|
| def _now() -> str:
|
| return datetime.now(timezone.utc).isoformat()
|
|
|
