"""Validasi skema JSON mimpi dan normalisasi koneksi / escape HTML.""" from __future__ import annotations import html import json import re from typing import Any from jsonschema import Draft202012Validator, ValidationError MOOD_VALID = frozenset({"mysterious", "joyful", "anxious", "surreal", "peaceful"}) ROLE_VALID = frozenset({"self", "known", "stranger", "creature", "archetype"}) TIPE_TEMPAT_VALID = frozenset({"real", "distorted", "fantastical", "transitional"}) TIPE_KONEKSI_VALID = frozenset({"tension", "harmony", "pursuit", "transformation"}) SKHEMA_MIMPI: dict[str, Any] = { "type": "object", "required": ["title", "mood", "core_theme"], "properties": { "title": {"type": "string"}, "mood": {"type": "string"}, "core_theme": {"type": "string"}, "characters": { "type": "array", "items": { "type": "object", "required": ["name"], "properties": { "name": {"type": "string"}, "role": {"type": "string"}, "emotion": {"type": "string"}, }, }, }, "places": { "type": "array", "items": { "type": "object", "required": ["name"], "properties": { "name": {"type": "string"}, "type": {"type": "string"}, "feeling": {"type": "string"}, }, }, }, "symbols": { "type": "array", "items": { "type": "object", "required": ["name"], "properties": { "name": {"type": "string"}, "appearance": {"type": "string"}, "significance": {"type": "string"}, }, }, }, "connections": { "type": "array", "items": { "type": "object", "required": ["from", "to"], "properties": { "from": {"type": "string"}, "to": {"type": "string"}, "type": {"type": "string"}, }, }, }, }, } _validator_skema = Draft202012Validator(SKHEMA_MIMPI) def escape_html(teks: str) -> str: """Escape string sebelum dimasukkan ke SVG (gr.HTML).""" return html.escape(str(teks), quote=True) def validasi_skema_json(data: dict) -> str | None: """Kembalikan pesan error jsonschema atau None jika valid.""" try: _validator_skema.validate(data) return None except ValidationError as e: return e.message def entitas_memadai(entities: dict) -> bool: """Minimal ada satu entitas bernama atau tema bermakna (gate 85%).""" if entities.get("title") == "Parse Failed": return False jumlah = ( len(entities.get("characters") or []) + len(entities.get("places") or []) + len(entities.get("symbols") or []) ) return jumlah >= 1 or bool(str(entities.get("core_theme", "")).strip()) def _normalisasi_mood(nilai: Any) -> str: if not isinstance(nilai, str): return "mysterious" mood = nilai.strip().lower() return mood if mood in MOOD_VALID else "mysterious" def _kata_kata(nama: str) -> list[str]: return re.findall(r"[a-z0-9]+", nama.lower()) def _daftar_nama_entitas(entities: dict) -> list[str]: nama: list[str] = [] for kunci in ("characters", "places", "symbols"): for item in entities.get(kunci) or []: if isinstance(item, dict) and item.get("name"): nama.append(str(item["name"]).strip()) return nama def _cocokkan_nama(teks: str, daftar_nama: list[str]) -> str | None: """Exact match dulu, lalu kata utuh — hindari substring pendek salah.""" if not teks or not daftar_nama: return None teks_lower = teks.strip().lower() if not teks_lower: return None for nama in daftar_nama: if nama.lower() == teks_lower: return nama if len(teks_lower) < 3: return None for nama in daftar_nama: if teks_lower in _kata_kata(nama): return nama if len(teks_lower) >= 4: cocok_awalan = [n for n in daftar_nama if n.lower().startswith(teks_lower)] if len(cocok_awalan) == 1: return cocok_awalan[0] return None def _deduplikasi_nama_entitas(entities: dict) -> dict: """Tandai nama duplikat persis (case-insensitive) dengan suffix.""" terlihat: dict[str, int] = {} def perbaiki_nama(nama: str) -> str: kunci = nama.lower() if kunci not in terlihat: terlihat[kunci] = 0 return nama terlihat[kunci] += 1 return f"{nama} ({terlihat[kunci] + 1})" for kunci in ("characters", "places", "symbols"): daftar = [] for item in entities.get(kunci) or []: if isinstance(item, dict) and item.get("name"): salinan = dict(item) salinan["name"] = perbaiki_nama(str(item["name"]).strip()) daftar.append(salinan) entities[kunci] = daftar return entities def normalisasi_koneksi(entities: dict) -> dict: """Perbaiki from/to, buang edge invalid, deduplikasi.""" daftar_nama = _daftar_nama_entitas(entities) if not daftar_nama: entities["connections"] = [] return entities koneksi_baru = [] sudah_ada: set[tuple[str, str, str]] = set() for edge in entities.get("connections") or []: if not isinstance(edge, dict): continue asal = _cocokkan_nama(str(edge.get("from", "")), daftar_nama) tujuan = _cocokkan_nama(str(edge.get("to", "")), daftar_nama) if not asal or not tujuan or asal == tujuan: continue tipe = str(edge.get("type", "harmony")).lower() if tipe not in TIPE_KONEKSI_VALID: tipe = "harmony" kunci = (asal, tujuan, tipe) if kunci in sudah_ada: continue sudah_ada.add(kunci) koneksi_baru.append({"from": asal, "to": tujuan, "type": tipe}) entities["connections"] = koneksi_baru return entities def _perbaiki_item_entitas( item: dict, field_peran: str | None, enum_peran: frozenset[str], field_kedua: str, ) -> dict: nama = str(item.get("name", "Unknown")).strip() or "Unknown" hasil = {"name": nama} if field_peran: peran = str(item.get(field_peran, "")).lower() hasil[field_peran] = peran if peran in enum_peran else "stranger" nilai_kedua = item.get(field_kedua, "") hasil[field_kedua] = str(nilai_kedua).strip()[:80] if nilai_kedua else "" return hasil def validasi_dan_perbaiki(entities: dict) -> dict: """Enum mood/role, default array kosong, dedupe nama, normalisasi koneksi.""" hasil: dict[str, Any] = { "title": str(entities.get("title", "Untitled Dream"))[:60], "mood": _normalisasi_mood(entities.get("mood")), "characters": [], "places": [], "symbols": [], "connections": entities.get("connections") or [], "core_theme": str(entities.get("core_theme", ""))[:200], } for item in entities.get("characters") or []: if isinstance(item, dict): hasil["characters"].append( _perbaiki_item_entitas(item, "role", ROLE_VALID, "emotion") ) for item in entities.get("places") or []: if isinstance(item, dict): tipe = str(item.get("type", "")).lower() hasil["places"].append( { "name": str(item.get("name", "Place")).strip() or "Place", "type": tipe if tipe in TIPE_TEMPAT_VALID else "distorted", "feeling": str(item.get("feeling", ""))[:80], } ) for item in entities.get("symbols") or []: if isinstance(item, dict): hasil["symbols"].append( { "name": str(item.get("name", "Symbol")).strip() or "Symbol", "appearance": str(item.get("appearance", ""))[:80], "significance": str(item.get("significance", ""))[:80], } ) hasil = _deduplikasi_nama_entitas(hasil) return normalisasi_koneksi(hasil) def parse_json_dari_teks(raw: str) -> tuple[dict | None, str | None]: """Parse JSON dari output model; bracket-aware, bukan regex greedy.""" teks = re.sub(r"```json\s*|\s*```", "", raw.strip()).strip() try: data = json.loads(teks) if isinstance(data, dict): return data, None except json.JSONDecodeError: pass idx = teks.find("{") while idx != -1: try: data, akhir = json.JSONDecoder().raw_decode(teks, idx) if isinstance(data, dict): return data, None except json.JSONDecodeError: pass idx = teks.find("{", idx + 1) return None, "Tidak ada objek JSON valid ditemukan"