""" Pipeline Server — Orchestrator semua detektor NLP. Port default: 8000 Menggabungkan sembilan detektor dalam satu endpoint untuk frontend index.html: 1. PII Detector — data pribadi (NIK, email, telepon, SSN, dll.) 2. Word Quality — kata slang, alay, dan typo 3. Konten Berisiko — prompt injection, kecurangan akademik, konten berbahaya 4. NER — entitas bernama (orang, lokasi, organisasi, dll.) 5. Profanity Checker — kata kasar dan tidak pantas 6. Filler Checker — frasa basa-basi yang membuang token 7. Special Char Detector — karakter invisible dan tidak efektif (zero-width, full-width, dll.) 8. Syntax Checker — urutan kata janggal Bahasa Indonesia (perplexity IndoBERT) 9. Field-Fit Detector — isi yang tampaknya salah field (embedding ML) Sistem difokuskan ke Bahasa Indonesia. Dukungan teks Inggris hanya ada pada PII dan NER karena sifat datanya lintas bahasa (identifier dan entitas global). Detektor lain hanya menangani Bahasa Indonesia. Endpoint HTTP: GET /api/status — status tiap detektor (siap / tidak siap) POST /api/evaluate — evaluasi semua field prompt Input : {"fields": {"task": "...", "context": "...", ...}} Output: {"issues": [...], "final_prompt": str, "final_prompt_sections": [...]} Sub-server (dijalankan otomatis sebagai proses latar belakang): port 8001 — NER server (web/ner-test.html) port 8002 — PII server (web/pii-test.html) port 8003 — Word Quality server (web/word-quality-test.html) port 8004 — Konten Berisiko server (web/risky-content-test.html) port 8005 — Profanity server (web/profanity-test.html) port 8006 — Filler server (web/filler-test.html) port 8007 — Special Char server (web/special-char-test.html) port 8008 — Syntax server (deteksi urutan kata) port 8009 — Field-Fit server (web/field-fit-test.html) Cara menjalankan: python src/core/pipeline_server.py python src/core/pipeline_server.py --fast # mode cepat: tanpa model berat (rekomendasi) python src/core/pipeline_server.py --no-resp-ml # nonaktifkan lapis ML ringan (Word Quality & Profanity) python src/core/pipeline_server.py --no-ner-ml # NER rule-only, tanpa transformer ~600 MB python src/core/pipeline_server.py --no-ner # nonaktifkan NER sepenuhnya python src/core/pipeline_server.py --port 8080 # ubah port """ from __future__ import annotations import argparse import importlib import json import logging import mimetypes import os import re import subprocess import sys import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path # Direktori src/ — digunakan untuk menemukan script sub-server _SRC_DIR = Path(__file__).parent.parent # Direktori web/ — file statis frontend (index.html, assets). Disajikan langsung # oleh server ini agar frontend dan backend satu origin (tanpa CORS / split deploy). _WEB_DIR = _SRC_DIR.parent / "web" # Proses sub-server yang berjalan di latar belakang (untuk cleanup saat exit) _sub_processes: list[subprocess.Popen] = [] # Tambahkan direktori src/ ke path agar import antar-modul detektor berfungsi sys.path.insert(0, str(Path(__file__).parent.parent)) # Konfigurasi global. Diimpor setelah path diatur agar paket core dapat ditemukan. from core import config # noqa: E402 from core.lexicons import load_word_set # noqa: E402 logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger(__name__) # Konfigurasi Field Prompt # # Setiap field memiliki: # id — kunci unik yang digunakan frontend # label — nama tampilan untuk pengguna # required — True jika field wajib diisi # color — warna teks pada label (hex) # surface — warna latar belakang field (hex) _FIELDS: list[dict] = [ {"id": "task", "label": "Task", "required": True, "color": "#9a3412", "surface": "#fff7ed"}, {"id": "context", "label": "Context", "required": True, "color": "#0f766e", "surface": "#ecfdf5"}, {"id": "references", "label": "References", "required": True, "color": "#4338ca", "surface": "#eef2ff"}, {"id": "role", "label": "Role", "required": False, "color": "#7e22ce", "surface": "#faf5ff"}, {"id": "audience", "label": "Audience", "required": False, "color": "#0369a1", "surface": "#f0f9ff"}, {"id": "tone", "label": "Tone", "required": False, "color": "#0f766e", "surface": "#f0fdfa"}, {"id": "constraints", "label": "Constraints", "required": False, "color": "#9f1239", "surface": "#fff1f2"}, {"id": "outputFormat","label": "Output Format", "required": False, "color": "#065f46", "surface": "#f0fdf4"}, {"id": "example", "label": "Example", "required": False, "color": "#7c2d12", "surface": "#fff7ed"}, ] # Batas ukuran body request. Prompt normal hanya beberapa KB; batas ini # mencegah satu request raksasa menghabiskan memori server. _MAX_BODY_BYTES = config.MAX_BODY_BYTES # State Global Detektor # # Instance detektor disimpan sebagai variabel global agar hanya dimuat sekali # saat server start, bukan setiap kali ada request. _pii = None # PIIDetector _wq = None # WordQualityDetector _risk = None # RiskyContentChecker _ner = None # IndonesianNER _prof = None # ProfanityChecker _filler = None # FillerChecker _special_char = None # SpecialCharDetector _syntax = None # SyntaxChecker _field_fit = None # FieldFitDetector (deteksi salah field) # Flag startup (disimpan agar reload menggunakan flag yang sama) _no_resp_ml = False _no_ner = False _no_ner_ml = False _no_syntax_ml = False _no_fieldfit_ml = False _detectors_loaded = threading.Event() _auto_reload_enabled = True def _terminate_subservers(timeout: float = 2.0) -> None: """Hentikan sub-server test yang di-spawn oleh pipeline utama.""" for p in list(_sub_processes): try: if p.poll() is None: p.terminate() try: p.wait(timeout=timeout) except subprocess.TimeoutExpired: p.kill() except Exception: pass _sub_processes.clear() def _restart_process(reason: str) -> None: """Restart proses pipeline dengan argumen command line yang sama.""" logger.info("%s — restart pipeline otomatis...", reason) _terminate_subservers() os.execv(sys.executable, [sys.executable, *sys.argv]) # Hot-Reload Watcher # # Memantau perubahan file source detektor dan memuat ulang detektor yang # berubah tanpa perlu restart server. Berjalan sebagai background daemon thread. class DetectorWatcher: """ Watcher berbasis polling — tidak butuh dependensi eksternal. Setiap _interval detik, cek mtime semua file detektor. Jika ada yang berubah, reload modul dan inisialisasi ulang detektor-nya. """ # (path relatif dari src/, nama reload method) _WATCH: list[tuple[str, str]] = [ ("pii/pii_detector.py", "_reload_pii"), ("word_quality/word_quality_detector.py", "_reload_wq"), ("risky_content/risky_content_detector.py", "_reload_risk"), ("ner/ner_detector.py", "_reload_ner"), ("profanity/profanity_detector.py", "_reload_prof"), ("filler/filler_detector.py", "_reload_filler"), ("special_char/special_char_detector.py", "_reload_special_char"), ("syntax/syntax_detector.py", "_reload_syntax"), ("field_fit/field_fit_detector.py", "_reload_field_fit"), ("thesaurus/thesaurus.py", "_reload_thesaurus"), ] def __init__(self, src_dir: Path, interval: float = 1.5) -> None: self._src_dir = src_dir self._interval = interval self._mtimes: dict[str, float] = {} self._stop = threading.Event() def start(self) -> None: for rel, _ in self._WATCH: p = self._src_dir / rel if p.exists(): self._mtimes[rel] = p.stat().st_mtime t = threading.Thread(target=self._loop, daemon=True, name="DetectorWatcher") t.start() logger.info("DetectorWatcher aktif (polling tiap %.1fs).", self._interval) def _loop(self) -> None: while not self._stop.wait(self._interval): for rel, method in self._WATCH: p = self._src_dir / rel if not p.exists(): continue mtime = p.stat().st_mtime if mtime == self._mtimes.get(rel): continue self._mtimes[rel] = mtime logger.info("Perubahan: %s — memuat ulang detektor...", rel) try: getattr(self, method)() logger.info("Reload selesai: %s", rel) except Exception as exc: logger.error("Reload gagal (%s): %s", rel, exc) def _reload_pii(self) -> None: global _pii import pii.pii_detector as m importlib.reload(m) _pii = m.PIIDetector() def _reload_wq(self) -> None: global _wq import word_quality.word_quality_detector as m importlib.reload(m) # Reset singleton tesaurus agar ikut terbaca ulang m._thesaurus = None det = m.WordQualityDetector(use_ml=not _no_resp_ml) det.load() _wq = det def _reload_risk(self) -> None: global _risk import risky_content.risky_content_detector as m importlib.reload(m) checker = m.RiskyContentChecker() checker.load() _risk = checker def _reload_ner(self) -> None: global _ner if _no_ner: return import ner.ner_detector as m importlib.reload(m) ner_obj = m.IndonesianNER() if not _no_ner_ml: ner_obj.load() _ner = ner_obj def _reload_prof(self) -> None: global _prof import profanity.profanity_detector as m importlib.reload(m) checker = m.ProfanityChecker(use_ml=not _no_resp_ml) checker.load() _prof = checker def _reload_filler(self) -> None: global _filler import filler.filler_detector as m importlib.reload(m) checker = m.FillerChecker() checker.load() _filler = checker def _reload_special_char(self) -> None: global _special_char import special_char.special_char_detector as m importlib.reload(m) _special_char = m.SpecialCharDetector() def _reload_syntax(self) -> None: global _syntax import syntax.syntax_detector as m importlib.reload(m) checker = m.SyntaxChecker(use_ml=not _no_syntax_ml) checker.load() _syntax = checker def _reload_field_fit(self) -> None: global _field_fit import field_fit.field_fit_detector as m importlib.reload(m) det = m.FieldFitDetector(use_ml=not _no_fieldfit_ml) det.load() _field_fit = det def _reload_thesaurus(self) -> None: # Reset singleton di word_quality_detector agar terbaca ulang saat request berikutnya try: import word_quality.word_quality_detector as wqm wqm._thesaurus = None import thesaurus.thesaurus as m importlib.reload(m) m._instance = None logger.info("Tesaurus singleton di-reset.") except Exception as exc: logger.warning("Reset tesaurus gagal: %s", exc) class PipelineRestartWatcher: """ Watcher file inti pipeline. Perubahan di file ini tidak aman di-reload dengan importlib karena handler, routing, argumen startup, dan global state sudah dipakai server berjalan. Solusinya restart proses otomatis dengan argv yang sama. """ _WATCH: list[str] = [ "core/pipeline_server.py", "core/language.py", "core/lexicons.py", ] def __init__(self, src_dir: Path, interval: float = 1.5) -> None: self._src_dir = src_dir self._interval = interval self._mtimes: dict[str, float] = {} self._stop = threading.Event() def start(self) -> None: for rel in self._WATCH: p = self._src_dir / rel if p.exists(): self._mtimes[rel] = p.stat().st_mtime t = threading.Thread(target=self._loop, daemon=True, name="PipelineRestartWatcher") t.start() logger.info("PipelineRestartWatcher aktif (polling tiap %.1fs).", self._interval) def _loop(self) -> None: while not self._stop.wait(self._interval): for rel in self._WATCH: p = self._src_dir / rel if not p.exists(): continue mtime = p.stat().st_mtime if mtime == self._mtimes.get(rel): continue self._mtimes[rel] = mtime time.sleep(0.25) # beri editor waktu menyelesaikan write _restart_process(f"Perubahan file inti terdeteksi: {rel}") # Konverter Issue # # Setiap detektor mengembalikan dataclass-nya masing-masing. Fungsi-fungsi di bawah # mengonversi ke format dict seragam yang dipahami frontend. # # Field dict issue yang dikirim ke frontend: # source — nama detektor (dipakai untuk dedup, dihapus sebelum dikirim) # field_id — id field asal (mis. "task", "context") # field_label — nama tampilan field # css_class — kelas CSS untuk pewarnaan di frontend # severity — "HIGH" | "MEDIUM" | "LOW" (dipakai pengelompokan tampilan) # word — potongan teks yang bermasalah # start / end — offset karakter dalam teks field # reason — penjelasan singkat untuk pengguna # action — "replace" (ada saran terapkan) | "select" (hanya sorot) # replacement — teks pengganti; string kosong berarti hapus # recommendation — saran perbaikan kalimat (hanya pada temuan konten berisiko) # Tipe data pribadi diberi nama yang mudah dipahami guru untuk teks placeholder. _PII_PLACEHOLDER: dict[str, str] = { "NIK": "[NIK]", "NPWP_LAMA": "[NPWP]", "NPWP_BARU": "[NPWP]", "BPJS_KES": "[BPJS]", "BPJS_TK": "[BPJS]", "PASPOR": "[NOMOR PASPOR]", "SIM": "[NOMOR SIM]", "PLAT_NOMOR": "[PLAT NOMOR]", "EMAIL": "[EMAIL]", "TELEPON_HP": "[NOMOR HP]", "TELEPON_TETAP": "[NOMOR TELEPON]", "REKENING_BANK": "[NOMOR REKENING]", "KARTU_KREDIT": "[NOMOR KARTU]", "IP_V4": "[ALAMAT IP]", "IP_V6": "[ALAMAT IP]", "TANGGAL_LAHIR": "[TANGGAL LAHIR]", "ALAMAT": "[ALAMAT]", # Identifier internasional (lintas bahasa) "SSN_US": "[NOMOR SSN]", "NIN_UK": "[NOMOR NIN]", "IBAN": "[NOMOR IBAN]", "MAC_ADDRESS": "[ALAMAT MAC]", "TELEPON_US": "[NOMOR TELEPON]", } def _pii_issue(entity, field: dict) -> dict: """Konversi PIIEntity menjadi dict issue standar.""" placeholder = _PII_PLACEHOLDER.get(entity.label, "[DATA PRIBADI]") return { "source": "pii", "field_id": field["id"], "field_label": field["label"], "css_class": "pii", "severity": "HIGH", "word": entity.word, "start": entity.start, "end": entity.end, "reason": "Ada data pribadi di sini. Jangan kirim data asli ke AI. " f"Ganti dengan penanda {placeholder} atau data contoh.", "action": "replace", "replacement": placeholder, } def _wq_issue(issue, field: dict) -> dict: """Konversi WordIssue menjadi dict issue standar.""" # Peta tipe masalah ke kelas CSS dan tingkat keparahan. SLANG bernilai MEDIUM # karena kata informal menurunkan kualitas, bukan menimbulkan risiko (bagian 3.6). css_map = {"SLANG": "slang", "ALAY": "weak", "TYPO": "typo"} sev_map = {"SLANG": "MEDIUM", "ALAY": "MEDIUM", "TYPO": "LOW"} css = css_map.get(issue.issue_type, "quality") sev = sev_map.get(issue.issue_type, "LOW") return { "source": "word_quality", "field_id": field["id"], "field_label": field["label"], "css_class": css, "severity": sev, "word": issue.word, "start": issue.start, "end": issue.end, "reason": issue.reason, "action": "replace" if issue.suggestion else "select", "replacement": issue.suggestion, } def _risk_issue(finding, field: dict) -> dict: """Konversi RiskyContentFinding menjadi dict issue standar.""" # Offset evidence dihitung langsung oleh checker pada teks asli, sehingga tidak # perlu pencarian string ulang yang rawan meleset. Saran perbaikan dibawa # lewat recommendation untuk ditampilkan di popover (tanpa tombol terapkan). css = "quality" if finding.code == "SENSITIVE_TOPIC" else "risky_content" return { "source": "risky_content", "field_id": field["id"], "field_label": field["label"], "css_class": css, "severity": finding.severity, "word": finding.evidence, "start": finding.start, "end": finding.end, "reason": finding.message, "recommendation": finding.recommendation, "action": "select", "replacement": None, } def _profanity_issue(finding, field: dict) -> dict: """Konversi ProfanityFinding menjadi dict issue standar.""" # Temuan Layer 2 (model toksisitas) menilai keseluruhan teks, bukan satu kata, # sehingga hanya disorot sebagai saran (tanpa tombol hapus). Temuan leksikon # (Layer 1) tetap menawarkan penghapusan kata (replacement kosong). is_ml = getattr(finding, "layer", "lexicon") == "ml" return { "source": "profanity", "field_id": field["id"], "field_label": field["label"], "css_class": "profanity", "severity": finding.severity, "word": finding.word, "start": finding.start, "end": finding.end, "reason": finding.reason, "action": "select" if is_ml else "replace", "replacement": None if is_ml else "", } def _filler_issue(finding, field: dict) -> dict: """Konversi FillerFinding menjadi dict issue standar.""" return { "source": "filler", "field_id": field["id"], "field_label": field["label"], "css_class": "filler", "severity": "LOW", "word": finding.word, "start": finding.start, "end": finding.end, "reason": finding.reason, # Saran menghapus frasa basa-basi: replacement kosong menandai penghapusan. "action": "replace", "replacement": "", } def _special_char_issue(finding, field: dict) -> dict: """Konversi SpecialCharFinding menjadi dict issue standar.""" replacement = getattr(finding, "replacement", None) return { "source": "special_char", "field_id": field["id"], "field_label": field["label"], "css_class": "special_char", "severity": "HIGH" if finding.category in ( "ZERO_WIDTH", "CONTROL_CHAR", "BIDI_CONTROL", "UNICODE_TAG" ) else "MEDIUM", "word": finding.word, "start": finding.start, "end": finding.end, "reason": finding.reason, "action": "replace" if replacement is not None else "select", "replacement": replacement, } def _syntax_issue(finding, field: dict) -> dict: """Konversi SyntaxFinding menjadi dict issue standar.""" return { "source": "syntax", "field_id": field["id"], "field_label": field["label"], "css_class": "syntax", "severity": "LOW", "word": finding.sentence, "start": finding.start, "end": finding.end, "reason": finding.reason, "action": "select", "replacement": None, } def _apply_wq_norms(text: str, norms: list[tuple[int, int, str]]) -> str: """Terapkan koreksi Word Quality ke teks untuk normalisasi sebelum field_fit ML. Penggantian diterapkan dari akhir ke awal agar offset tidak bergeser. """ if not norms: return text chars = list(text) for start, end, repl in sorted(norms, key=lambda x: x[0], reverse=True): chars[start:end] = list(repl) return "".join(chars) def _field_fit_issue(finding, field: dict) -> dict: """Konversi FieldFitFinding menjadi dict issue standar (advice-only).""" target_label = next( (f["label"] for f in _FIELDS if f["id"] == finding.target_field), finding.target_field, ) return { "source": "field_fit", "field_id": field["id"], "field_label": field["label"], "css_class": "field_fit", "severity": "LOW", "word": finding.word, "start": finding.start, "end": finding.end, "reason": finding.reason, "recommendation": f"Pertimbangkan memindahkannya ke field {target_label}.", "action": "select", "replacement": None, } _NER_PROMPT_LABELS = {"ORANG", "LOKASI", "ORGANISASI", "FASILITAS"} # Placeholder penyamaran per label NER (analog dengan PII). _NER_PLACEHOLDER: dict[str, str] = { "ORANG": "[NAMA]", "LOKASI": "[LOKASI]", "ORGANISASI": "[ORGANISASI]", "FASILITAS": "[FASILITAS]", } # Dimuat dari resources/lexicons/ner/prompt_stopwords.txt agar mudah dirawat. _NER_PROMPT_STOPWORDS = load_word_set("ner", "prompt_stopwords.txt") _NER_TOKEN_RE = re.compile(r"[A-Za-zÀ-ÿ]+") def _is_prompt_ner_entity(entity) -> bool: """Filter NER untuk kebutuhan Prompt Builder: hanya entitas privasi relevan.""" if entity.label not in _NER_PROMPT_LABELS: return False word = (entity.word or "").strip() if not word: return False norm = word.lower().strip(" \t\r\n,;:!?.") if norm in _NER_PROMPT_STOPWORDS: return False tokens = _NER_TOKEN_RE.findall(word) # Entitas multi-token yang diawali stopword institusional (mis. # "Kementerian Pendidikan", "Universitas Indonesia") difilter karena # sudah jelas merupakan nama lembaga, bukan data pribadi. if tokens and tokens[0].lower() in _NER_PROMPT_STOPWORDS: return False if len(tokens) == 1: token = tokens[0] if entity.source == "ml" and token.islower(): return False return True def _ner_issue(entity, field: dict) -> dict | None: """Konversi NEREntity menjadi dict issue standar.""" if not _is_prompt_ner_entity(entity): return None placeholder = _NER_PLACEHOLDER.get(entity.label, "[ENTITAS]") return { "source": "ner", "field_id": field["id"], "field_label": field["label"], "css_class": "ner", "severity": "LOW", "word": entity.word, "start": entity.start, "end": entity.end, "reason": "Ada nama orang, tempat, atau lembaga nyata di sini. " f"Bila menyangkut data rahasia, samarkan dengan {placeholder} " "atau data contoh.", "action": "replace", "replacement": placeholder, } # Evaluasi Prompt def _assemble_prompt(fields: dict[str, str]) -> tuple[str, list[dict]]: """ Rakit isi field menjadi teks prompt final dan daftar seksi untuk frontend. Returns: (final_prompt, sections) di mana: - final_prompt: teks gabungan semua field berisi "Label: nilai" - sections: daftar dict berisi metadata tiap field untuk rendering frontend """ sections: list[dict] = [] parts: list[str] = [] for f in _FIELDS: val = fields.get(f["id"], "").strip() if not val: continue sections.append({ "label": f["label"], "value": val, "color": f["color"], "surface": f["surface"], }) parts.append(f"{f['label']}: {val}") return "\n\n".join(parts), sections def _analyze_field(f: dict, text: str) -> list[dict]: """Jalankan semua detektor pada satu field. Dipanggil dari thread pool.""" issues: list[dict] = [] try: from core.language import detect_language language = detect_language(text).language except Exception: language = "unknown" # Kebijakan bahasa: hanya PII, NER, dan Profanity yang lintas bahasa # (identifier, entitas, dan kata kasar berlaku untuk Indonesia + Inggris). # Detektor lain Indonesia-only — dilewati untuk teks yang terdeteksi Inggris. is_english = (language == "en") # Word Quality (Indonesia-only) dijalankan pertama agar koreksi (slang/typo) # tersedia sebagai norm_text untuk detektor yang membutuhkan teks ternormalisasi # (Risky Content, Syntax, Field-Fit). wq_norms: list[tuple[int, int, str]] = [] if _wq and not is_english: try: for issue in _wq.detect(text, language=language): issues.append(_wq_issue(issue, f)) if issue.suggestion: wq_norms.append((issue.start, issue.end, issue.suggestion)) except Exception as e: logger.warning("WordQuality error [%s]: %s", f["id"], e) norm_text = _apply_wq_norms(text, wq_norms) if wq_norms else None if _risk and not is_english: try: for finding in _risk.check(text, f["id"], language=language, norm_text=norm_text): issues.append(_risk_issue(finding, f)) except Exception as e: logger.warning("Konten Berisiko error [%s]: %s", f["id"], e) if _pii: try: for entity in _pii.detect(text, language=language): issues.append(_pii_issue(entity, f)) except Exception as e: logger.warning("PII error [%s]: %s", f["id"], e) if _ner: try: for entity in _ner.predict(text, language=language): issue = _ner_issue(entity, f) if issue: issues.append(issue) except Exception as e: logger.warning("NER error [%s]: %s", f["id"], e) if _prof: try: for finding in _prof.check(text, language=language): issues.append(_profanity_issue(finding, f)) except Exception as e: logger.warning("Profanity error [%s]: %s", f["id"], e) if _filler and not is_english: try: for finding in _filler.check(text, language=language, norm_text=norm_text): issues.append(_filler_issue(finding, f)) except Exception as e: logger.warning("Filler error [%s]: %s", f["id"], e) if _special_char and not is_english: try: for finding in _special_char.detect(text): issues.append(_special_char_issue(finding, f)) except Exception as e: logger.warning("Special Char error [%s]: %s", f["id"], e) # Syntax checker dijalankan pada semua field. Guard panjang kalimat minimum di # dalam checker mencegah false positive pada field yang berupa frasa pendek # (mis. tone atau outputFormat), sehingga pembatasan field tidak diperlukan. if _syntax and not is_english: try: for finding in _syntax.check(text, language=language, norm_text=norm_text): issues.append(_syntax_issue(finding, f)) except Exception as e: logger.warning("Syntax error [%s]: %s", f["id"], e) # Field-Fit: deteksi isi yang tampaknya milik field lain via embedding # lintas-field. Jika model nonaktif, detektor tidak mengembalikan temuan. if _field_fit and not is_english: try: for finding in _field_fit.detect(text, f["id"], language=language, norm_text=norm_text): issues.append(_field_fit_issue(finding, f)) except Exception as e: logger.warning("Field-Fit error [%s]: %s", f["id"], e) return issues # Urutan prioritas temuan level kata. Angka lebih kecil berarti prioritas lebih # tinggi dan menang saat span tumpang-tindih. # NER diberi prioritas lebih tinggi dari word_quality agar entitas bernama # (mis. "Maranatha" sebagai nama universitas) tidak tertimpa oleh badge TYPO # dari WQ yang tidak mengenal kata tersebut sebagai proper noun. _WORD_LEVEL_PRIORITY: dict[str, int] = { "pii": 0, "profanity": 1, "ner": 2, "filler": 3, "word_quality": 4, } def _dedup_word_level(issues: list[dict]) -> list[dict]: """ Selesaikan tumpang-tindih antar temuan level kata memakai urutan prioritas. Temuan dari sumber selain level kata (risky_content, syntax, special_char, missing) dikembalikan apa adanya. Untuk temuan level kata, bila dua temuan pada field yang sama saling tumpang-tindih, hanya yang prioritasnya tertinggi yang dipertahankan. Parameters: issues (list[dict]): Seluruh temuan sebelum dedup. Returns: list[dict]: Temuan setelah tumpang-tindih level kata diselesaikan. """ word_level = [i for i in issues if i["source"] in _WORD_LEVEL_PRIORITY] others = [i for i in issues if i["source"] not in _WORD_LEVEL_PRIORITY] # Urutkan dari prioritas tertinggi agar temuan penting diterima lebih dulu. word_level.sort(key=lambda i: _WORD_LEVEL_PRIORITY[i["source"]]) accepted: list[dict] = [] for issue in word_level: fid, start, end = issue["field_id"], issue["start"], issue["end"] overlaps = any( fid == kept["field_id"] and start < kept["end"] and end > kept["start"] for kept in accepted ) if not overlaps: accepted.append(issue) return others + accepted def _evaluate(fields: dict[str, str]) -> dict: """ Jalankan semua detektor pada setiap field prompt dan kembalikan hasilnya. Field dianalisis secara paralel (ThreadPoolExecutor) sehingga 9 field tidak perlu menunggu satu per satu. Urutan detektor dalam tiap field tetap terjaga (sequential di dalam _analyze_field). Returns: dict berisi "issues", "final_prompt", dan "final_prompt_sections". """ all_issues: list[dict] = [] # Field wajib yang masih kosong ditandai di frontend, bukan di sini, agar # badge "belum diisi" muncul tanpa menunggu respons server. # Analisis semua field yang terisi secara paralel. active = [(f, fields.get(f["id"], "").strip()) for f in _FIELDS if fields.get(f["id"], "").strip()] with ThreadPoolExecutor(max_workers=min(len(active), 9)) as pool: futures = {pool.submit(_analyze_field, f, text): f["id"] for f, text in active} for future in as_completed(futures): try: all_issues.extend(future.result()) except Exception as e: logger.warning("Field analysis error [%s]: %s", futures[future], e) # 3. Dedup temuan level kata: pada span yang tumpang-tindih di field yang sama, # yang prioritasnya tertinggi menang. Urutan PII > Profanity > Filler > # Word Quality > NER (Filler di atas Word Quality karena menghapus frasa lebih # berguna daripada mengoreksi kata yang akan ikut terhapus). Temuan level # kalimat (Konten Berisiko, Syntax) dan karakter (Special Char) tidak diadu karena # beda ranah, jadi dibiarkan berdampingan. deduped_issues = _dedup_word_level(all_issues) # Field source hanya dipakai untuk dedup internal, tidak diperlukan frontend. for issue in deduped_issues: issue.pop("source", None) final_prompt, sections = _assemble_prompt(fields) return { "issues": deduped_issues, "final_prompt": final_prompt, "final_prompt_sections": sections, } # HTTP Request Handler class Handler(BaseHTTPRequestHandler): """Handler HTTP sederhana untuk melayani request GET dan POST.""" # Helper privat def _cors(self) -> None: """ Tambahkan header CORS agar frontend bisa mengakses server dari browser. Pada mode produksi header ini dilewati karena frontend disajikan dari origin yang sama dengan backend, sehingga CORS tidak diperlukan. """ if not config.ENABLE_CORS: return self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") def _json(self, status: int, body: object) -> None: """Kirim respons JSON dengan status code tertentu.""" payload = json.dumps(body, ensure_ascii=False).encode() try: self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self._cors() self.end_headers() self.wfile.write(payload) except (ConnectionAbortedError, BrokenPipeError, ConnectionResetError): pass # klien memutus koneksi sebelum respons selesai dikirim def _body(self) -> dict | None: """Baca dan parse JSON dari request body. Kembalikan None jika tidak valid.""" try: n = int(self.headers.get("Content-Length", 0)) except (TypeError, ValueError): return None if not n: return {} try: return json.loads(self.rfile.read(n)) except (json.JSONDecodeError, ValueError): return None def log_message(self, fmt, *args): """Override agar log HTTP request menggunakan logger standar proyek.""" logger.info("%-6s %s", args[0] if args else "", args[1] if len(args) > 1 else "") def _serve_static(self) -> None: """ Sajikan file statis dari web/. Path "/" → index.html. Aman dari path traversal: path hasil resolve harus tetap di dalam web/. """ # Buang query string, normalkan, default ke index.html rel = self.path.split("?", 1)[0].split("#", 1)[0].lstrip("/") if not rel: rel = "index.html" target = (_WEB_DIR / rel).resolve() try: target.relative_to(_WEB_DIR.resolve()) # cegah ../ keluar dari web/ except ValueError: self._json(403, {"error": "Akses ditolak."}) return if not target.is_file(): self._json(404, {"error": "File tidak ditemukan."}) return ctype = mimetypes.guess_type(str(target))[0] or "application/octet-stream" try: data = target.read_bytes() self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self._cors() self.end_headers() self.wfile.write(data) except (ConnectionAbortedError, BrokenPipeError, ConnectionResetError): pass # Endpoint handler def do_OPTIONS(self) -> None: """Tangani preflight CORS request dari browser.""" self.send_response(204) self._cors() self.end_headers() def do_GET(self) -> None: """ GET /api/status — kembalikan status semua detektor. Response: {"ready": true, "detectors": {"pii": true, ...}} """ if self.path == "/api/status": self._json(200, { "ready": _detectors_loaded.is_set(), "detectors": { "pii": _pii is not None, "word_quality": _wq is not None, "risky_content": _risk is not None, "ner": _ner is not None, "profanity": _prof is not None, "filler": _filler is not None, "special_char": _special_char is not None, "syntax": _syntax is not None, "field_fit": _field_fit is not None, }, }) elif self.path.startswith("/api/"): self._json(404, {"error": "Endpoint tidak ditemukan."}) else: self._serve_static() def do_POST(self) -> None: """ POST /api/evaluate — evaluasi semua field prompt. Request body : {"fields": {"task": "...", "context": "...", ...}} Response body: {"issues": [...], "final_prompt": "...", ...} """ if self.path != "/api/evaluate": self._json(404, {"error": "Endpoint tidak ditemukan."}) return try: content_length = int(self.headers.get("Content-Length", 0)) except (TypeError, ValueError): content_length = 0 if content_length > _MAX_BODY_BYTES: self._json(413, {"error": "Ukuran permintaan melebihi batas 1 MB."}) return body = self._body() if body is None: self._json(400, {"error": "Body bukan JSON yang valid."}) return fields = body.get("fields", {}) if not isinstance(fields, dict): self._json(400, {"error": "Field 'fields' harus berupa objek JSON."}) return try: result = _evaluate(fields) self._json(200, result) except Exception as e: logger.exception("Evaluate error: %s", e) self._json(500, {"error": str(e)}) # Entry Point def _start_subservers(host: str, no_resp_ml: bool = False, no_ner: bool = False, no_ner_ml: bool = False, no_syntax_ml: bool = False) -> None: """ Jalankan semua server test individual sebagai proses latar belakang. Setiap server di-spawn sebagai subprocess terpisah sehingga: - Halaman test (web/*-test.html) bisa langsung digunakan tanpa menjalankan server secara manual. - Sub-server bersifat lazy: detektor hanya dimuat saat request pertama masuk. - Jika port sudah terpakai (mis. sudah ada server yang berjalan manual), error diabaikan dengan pesan peringatan. Args: host: Host yang digunakan (sama dengan pipeline server). no_resp_ml: Teruskan flag --no-resp-ml ke sub-server lapis ML ringan. no_ner: Jangan jalankan sub-server NER. no_ner_ml: Jangan jalankan sub-server NER karena server ini selalu memuat model ML. no_syntax_ml: Jangan jalankan sub-server Syntax karena server ini selalu memuat model ML. """ global _sub_processes # (script_relative_path, port, nama_tampilan, extra_args) configs: list[tuple[str, int, str, list[str]]] = [ ("pii/pii_server.py", 8002, "PII", []), ("word_quality/word_quality_server.py", 8003, "WordQuality", []), ("risky_content/risky_content_server.py", 8004, "KontenBerisiko", []), ("profanity/profanity_server.py", 8005, "Profanity", []), ("filler/filler_server.py", 8006, "Filler", []), ("special_char/special_char_server.py", 8007, "SpecialChar", []), ("field_fit/field_fit_server.py", 8009, "FieldFit", []), ] if not no_ner and not no_ner_ml: configs.insert(0, ("ner/ner_server.py", 8001, "NER", [])) else: logger.info("Sub-server NER dilewati (mode cepat / NER ML nonaktif).") if not no_syntax_ml: configs.append(("syntax/syntax_server.py", 8008, "Syntax", [])) else: logger.info("Sub-server Syntax dilewati (mode cepat / Syntax ML nonaktif).") logger.info("─" * 56) logger.info("Memulai sub-server halaman test...") for script, port, name, extra in configs: script_path = _SRC_DIR / script if not script_path.exists(): logger.warning("Sub-server %s: file tidak ditemukan (%s)", name, script_path) continue try: p = subprocess.Popen( [sys.executable, str(script_path), "--port", str(port), "--host", host, *extra], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) _sub_processes.append(p) logger.info(" %-12s → http://%s:%-5d (PID %d)", name, host, port, p.pid) except OSError as exc: logger.warning("Sub-server %s gagal (port %d mungkin sudah terpakai): %s", name, port, exc) except Exception as exc: logger.warning("Sub-server %s gagal: %s", name, exc) logger.info("─" * 56) def main() -> None: """ Muat semua detektor lalu jalankan server HTTP. Urutan pemuatan: 1. PII Detector (ringan, selalu berhasil) 2. Word Quality (unduh kamus ~4 MB jika belum ada) 3. Konten Berisiko (berbasis aturan, ringan) 4. NER (model transformer ~600 MB, unduh otomatis) 5. Profanity Checker (unduh lexicon ~50 KB jika belum ada) 6. Filler Checker (ringan, berbasis regex) 7. Special Char Detector (ringan, berbasis karakter) """ global _pii, _wq, _risk, _ner, _prof, _filler, _special_char global _no_resp_ml, _no_ner, _no_ner_ml, _no_syntax_ml, _no_fieldfit_ml, _auto_reload_enabled # Railway (dan platform cloud lain) mengeset $PORT secara otomatis. # Fallback ke 8000 untuk development lokal. _env_port = int(os.environ.get("PORT", 8000)) _env_host = os.environ.get("HOST", "127.0.0.1") parser = argparse.ArgumentParser(description="Pipeline NLP Server") parser.add_argument("--port", type=int, default=_env_port, help="Port server (default: $PORT atau 8000)") parser.add_argument("--host", default=_env_host, help="Host server (default: $HOST atau 127.0.0.1)") parser.add_argument("--no-resp-ml", action="store_true", help="Nonaktifkan lapis ML ringan (Word Quality & Profanity), lebih cepat") parser.add_argument("--no-ner", action="store_true", help="Nonaktifkan NER sepenuhnya (hemat ~600 MB RAM, cocok untuk hosting gratis)") parser.add_argument("--no-ner-ml", action="store_true", help="NER hanya rule-based (regex + nama), tanpa model transformer (~600 MB). " "Lebih cepat; masih mendeteksi lembaga, tanggal, nama orang, dll.") parser.add_argument("--no-syntax-ml", action="store_true", help="Nonaktifkan Syntax Checker (deteksi urutan kata, model IndoBERT ~420 MB).") parser.add_argument("--no-fieldfit-ml", action="store_true", help="Nonaktifkan Field-Fit berbasis embedding lintas-field (~470 MB).") parser.add_argument("--fast", action="store_true", help="Mode cepat: setara --no-resp-ml --no-ner-ml --no-syntax-ml " "--no-fieldfit-ml. Semua detektor ringan aktif, tanpa model berat.") parser.add_argument("--no-auto-reload", action="store_true", help="Nonaktifkan auto-reload/restart saat file pipeline berubah.") args = parser.parse_args() _no_resp_ml = args.no_resp_ml or args.fast _no_ner = args.no_ner _no_ner_ml = args.no_ner_ml or args.fast _no_syntax_ml = args.no_syntax_ml or args.fast _no_fieldfit_ml = args.no_fieldfit_ml or args.fast _auto_reload_enabled = not args.no_auto_reload # Load semua detektor di background thread # Server langsung mulai listen agar healthcheck Railway tidak timeout. # /api/status mengembalikan ready:false selama loading, ready:true setelah selesai. def _load_detectors(): global _pii, _wq, _risk, _ner, _prof, _filler, _special_char, _syntax global _field_fit # 1. PII Detector logger.info("Memuat PII Detector...") from pii.pii_detector import PIIDetector _pii = PIIDetector() logger.info("PII Detector siap.") # 2. Word Quality Detector logger.info("Memuat Word Quality Detector...") from word_quality.word_quality_detector import WordQualityDetector _wq = WordQualityDetector(use_ml=not _no_resp_ml) _wq.load() logger.info( "Word Quality Detector siap (kamus: %d entri, SymSpell: %s, ML: %s).", _wq.slang_dict_size, "aktif" if _wq.symspell_loaded else "nonaktif", "aktif" if _wq.ml_active else "nonaktif", ) # 3. Konten Berisiko (berbasis aturan; lapis ML dihapus) logger.info("Memuat detektor Konten Berisiko...") from risky_content.risky_content_detector import RiskyContentChecker _risk = RiskyContentChecker() _risk.load() logger.info("Detektor Konten Berisiko siap.") # 4. NER if args.no_ner: logger.info("NER dinonaktifkan (--no-ner).") _ner = None elif _no_ner_ml: logger.info("NER mode rule-only (--no-ner-ml / --fast): regex + lexicon, tanpa model transformer.") from ner.ner_detector import IndonesianNER _ner = IndonesianNER() # tidak memanggil .load() → ML tetap None, rules aktif logger.info("NER rule-only siap.") else: logger.info("Memuat NER model (proses ini mungkin memerlukan waktu)...") try: from ner.ner_detector import IndonesianNER _ner = IndonesianNER() _ner.load() logger.info("NER siap (model: %s).", _ner.loaded_model) except Exception as e: logger.error("NER gagal dimuat: %s — lanjut tanpa NER.", e) _ner = None # 5. Profanity Checker (Layer 1 leksikon + Layer 2 toksisitas ML opsional) logger.info("Memuat Profanity Checker...") from profanity.profanity_detector import ProfanityChecker _prof = ProfanityChecker(use_ml=not _no_resp_ml) _prof.load() logger.info("Profanity Checker siap (%d kata, ML: %s).", _prof.lexicon_size, "aktif" if _prof.ml_active else "nonaktif") # 6. Filler Checker logger.info("Memuat Filler Checker...") from filler.filler_detector import FillerChecker _filler = FillerChecker() _filler.load() logger.info("Filler Checker siap (%d pola).", _filler.pattern_count) # 7. Special Char Detector logger.info("Memuat Special Char Detector...") from special_char.special_char_detector import SpecialCharDetector _special_char = SpecialCharDetector() logger.info("Special Char Detector siap.") # 8. Syntax Checker (deteksi urutan kata, model IndoBERT ~420 MB) if _no_syntax_ml: logger.info("Syntax Checker dinonaktifkan (--no-syntax-ml / --fast).") _syntax = None else: logger.info("Memuat Syntax Checker (model IndoBERT, proses ini mungkin memerlukan waktu)...") try: from syntax.syntax_detector import SyntaxChecker _syntax = SyntaxChecker(use_ml=True) _syntax.load() logger.info("Syntax Checker siap (ML aktif: %s).", _syntax.ml_active) except Exception as e: logger.error("Syntax Checker gagal dimuat: %s — lanjut tanpa deteksi urutan kata.", e) _syntax = None # 9. Field-Fit Detector (deteksi salah field via ML embedding opsional). logger.info("Memuat Field-Fit Detector...") from field_fit.field_fit_detector import FieldFitDetector _field_fit = FieldFitDetector(use_ml=not _no_fieldfit_ml) _field_fit.load() logger.info("Field-Fit Detector siap (%d prototipe field, ML: %s).", _field_fit.marker_count, "aktif" if _field_fit.ml_active else "nonaktif") # 10. Hot-reload watcher. # Dimatikan pada mode produksi karena tidak ada perubahan kode saat # runtime dan restart proses otomatis tidak diinginkan di server publik. if _auto_reload_enabled and config.ENABLE_AUTO_RELOAD: PipelineRestartWatcher(_SRC_DIR).start() DetectorWatcher(_SRC_DIR).start() else: logger.info("Auto-reload dinonaktifkan.") # 11. Sub-server halaman test. # Tidak dijalankan pada mode produksi untuk menghemat memori; halaman # test per detektor hanya diperlukan saat pengembangan. if config.ENABLE_TEST_SUBSERVERS: _start_subservers(args.host, no_resp_ml=_no_resp_ml, no_ner=_no_ner, no_ner_ml=_no_ner_ml, no_syntax_ml=_no_syntax_ml) else: logger.info("Sub-server halaman test dilewati (mode produksi).") logger.info("Semua detektor siap. Pipeline aktif penuh.") _detectors_loaded.set() loader_thread = threading.Thread(target=_load_detectors, daemon=True) loader_thread.start() # Jalankan pipeline server (langsung, tidak menunggu detektor selesai) # ThreadingHTTPServer memproses tiap request di thread terpisah sehingga # satu evaluasi berat tidak memblokir pengguna lain. Inferensi model ML # tetap aman karena tiap checker memegang lock-nya sendiri. server = ThreadingHTTPServer((args.host, args.port), Handler) logger.info("Pipeline Server berjalan di http://%s:%d", args.host, args.port) logger.info("Buka web/index.html untuk memulai.") logger.info("Tekan Ctrl+C untuk menghentikan.") try: server.serve_forever() except KeyboardInterrupt: logger.info("Server dihentikan. Menghentikan sub-server…") _terminate_subservers() server.server_close() if __name__ == "__main__": main()