scripts/import_lexicons.py

from __future__ import annotations
import argparse
import csv
import io
import json
import re
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
LEXICON_ROOT = ROOT / "resources" / "lexicons"
URLS = {
    "kamus_alay": (
        "https://raw.githubusercontent.com/nasalsabila/"
        "kamus-alay/master/colloquial-indonesian-lexicon.csv"
    ),
    "okky_kamusalay": (
        "https://raw.githubusercontent.com/okkyibrohim/"
        "id-multi-label-hate-speech-and-abusive-language-detection/master/new_kamusalay.csv"
    ),
    "stopwords_id": (
        "https://raw.githubusercontent.com/stopwords-iso/"
        "stopwords-id/master/stopwords-id.txt"
    ),
    "kbbi_id": (
        "https://raw.githubusercontent.com/Wikidepia/"
        "indonesian_datasets/master/dictionary/wordlist/data/wordlist.txt"
    ),
    "okky_abusive": (
        "https://raw.githubusercontent.com/okkyibrohim/"
        "id-multi-label-hate-speech-and-abusive-language-detection/master/abusive.csv"
    ),
    "names_id": (
        "https://gist.githubusercontent.com/maulvi/"
        "e443e22b82a1dc24e14344b47f0a80ea/raw/nama.txt"
    ),
}

def fetch_bytes(url: str, timeout: int = 45) -> bytes:
    req = urllib.request.Request(url, headers={"User-Agent": "PromptBuilderLexiconImporter/1.0"})
    with urllib.request.urlopen(req, timeout=timeout) as response:
        return response.read()

def fetch_text(url: str, timeout: int = 45) -> str:
    return fetch_bytes(url, timeout=timeout).decode("utf-8", errors="replace")

def read_lines(path: Path, *, lower: bool = True) -> set[str]:
    if not path.exists():
        return set()
    words: set[str] = set()
    for raw in path.read_text(encoding="utf-8", errors="replace").splitlines():
        line = raw.strip().lstrip("\ufeff")
        if not line or line.startswith("#"):
            continue
        words.add(line.lower() if lower else line)
    return words

def read_mapping(path: Path) -> dict[str, str]:
    if not path.exists():
        return {}
    mapping: dict[str, str] = {}
    for raw in path.read_text(encoding="utf-8", errors="replace").splitlines():
        line = raw.strip().lstrip("\ufeff")
        if not line or line.startswith("#") or "\t" not in line:
            continue
        key, value = line.split("\t", 1)
        key = key.strip().lower()
        value = value.strip()
        if key and value and key != value.lower():
            mapping[key] = value
    return mapping

def write_lines(path: Path, words: set[str], *, lower: bool = True) -> int:
    path.parent.mkdir(parents=True, exist_ok=True)
    cleaned = set()
    for word in words:
        word = word.strip()
        if not word or word.startswith("#"):
            continue
        cleaned.add(word.lower() if lower else word)
    ordered = sorted(cleaned, key=lambda item: (item.lower(), item))
    path.write_text("\n".join(ordered) + "\n", encoding="utf-8")
    return len(ordered)

def write_mapping(path: Path, mapping: dict[str, str]) -> int:
    path.parent.mkdir(parents=True, exist_ok=True)
    cleaned = {
        key.strip().lower(): value.strip()
        for key, value in mapping.items()
        if key.strip() and value.strip() and key.strip().lower() != value.strip().lower()
    }
    lines = [f"{key}\t{cleaned[key]}" for key in sorted(cleaned)]
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return len(lines)

def simple_word(text: str) -> str | None:
    word = text.strip().lower().strip("\"'`.,;:!?()[]{}")
    if re.fullmatch(r"[a-z][a-z0-9@$-]{1,40}", word):
        return word
    return None

def import_kamus_alay() -> int:
    text = fetch_text(URLS["kamus_alay"])
    reader = csv.DictReader(io.StringIO(text))
    downloaded: dict[str, str] = {}
    for row in reader:
        slang = (row.get("slang") or "").strip().lower()
        formal = (row.get("formal") or "").strip()
        if slang and formal and slang != formal.lower():
            downloaded[slang] = formal
    target = LEXICON_ROOT / "word_quality" / "slang_id.tsv"
    merged = downloaded
    merged.update(read_mapping(target))
    return write_mapping(target, merged)

# Kata fungsi Bahasa Inggris — dipakai HANYA untuk MENYARING entri ber-rasa Inggris
# dari kamus okky (mis. "pls"→"please", "rn"→"right now") agar kamus tetap
# Indonesia-only. Bukan deteksi runtime; hanya filter saat impor.
_EN_MARKERS = {
    "the", "you", "your", "are", "was", "were", "please", "right", "now",
    "what", "which", "with", "this", "that", "they", "have", "has", "dont",
    "don't", "isnt", "really", "literally", "omg", "lol", "anyway", "because",
    "about", "would", "could", "should", "very", "just", "know", "want",
}

def _formal_looks_english(formal: str) -> bool:
    return any(tok in _EN_MARKERS for tok in re.findall(r"[a-z']+", formal.lower()))

def import_okky_kamusalay() -> int:
    """Kamus alay besar okkyibrohim (~15rb pasangan slang→baku, tanpa header)."""
    text = fetch_text(URLS["okky_kamusalay"])
    downloaded: dict[str, str] = {}
    for row in csv.reader(io.StringIO(text)):
        if len(row) < 2:
            continue
        slang = row[0].strip().lower()
        formal = row[1].strip()
        if not slang or not formal or slang == formal.lower():
            continue
        if slang in {"alay", "slang", "kataalay", "tidakbaku"}:  # baris header bila ada
            continue
        if _formal_looks_english(formal):   # buang slang Inggris (Indonesia-only)
            continue
        downloaded[slang] = formal
    target = LEXICON_ROOT / "word_quality" / "slang_id.tsv"
    merged = downloaded
    merged.update(read_mapping(target))   # entri terkurasi yang sudah ada tetap menang
    return write_mapping(target, merged)

def import_kbbi_id() -> int:
    """Kosakata baku Indonesia (wordlist Wikidepia/KBBI) — hanya kata tunggal.
    Dipakai sebagai daftar 'kata baku dikenal' agar kata formal langka TIDAK
    salah ditandai sebagai typo (menurunkan false positive).
    """
    text = fetch_text(URLS["kbbi_id"])
    words: set[str] = set()
    for raw in text.splitlines():
        w = raw.strip().lower()
        # Hanya kata tunggal huruf — buang idiom, frasa, dan entri ber-kurung.
        if re.fullmatch(r"[a-zà-ÿ]{2,30}", w):
            words.add(w)
    target = LEXICON_ROOT / "word_quality" / "kbbi_id.txt"
    return write_lines(target, read_lines(target) | words)

def import_stopwords_id() -> int:
    """Stopwords Indonesia (stopwords-iso — agregasi Sastrawi dkk., paling lengkap)."""
    text = fetch_text(URLS["stopwords_id"])
    words = {
        w.strip().lower()
        for w in text.splitlines()
        if w.strip() and not w.strip().startswith("#")
    }
    target = LEXICON_ROOT / "language" / "stopwords_id.txt"
    return write_lines(target, read_lines(target) | words)

def import_okky_abusive() -> int:
    text = fetch_text(URLS["okky_abusive"])
    words: set[str] = set()
    for row in csv.reader(io.StringIO(text)):
        if not row:
            continue
        word = simple_word(row[0])
        if word and word != "abusive":
            words.add(word)
    target = LEXICON_ROOT / "profanity" / "external_id.txt"
    return write_lines(target, read_lines(target) | words)

def import_names_id() -> int:
    """Nama depan Indonesia (daftar publik) — union dengan daftar terkurasi yang ada.
    Nama yang juga merupakan kata baku KBBI (mis. "bunga", "mawar", "cinta",
    "indah") DIBUANG. Tanpa penyaringan ini, booster nama rule-based di NER salah
    menandai frasa kapital biasa seperti "Beli Bunga Mawar" sebagai ORANG. Filter
    bersumber kamus baku KBBI yang sudah diimpor (deterministik, bukan daftar
    pengecualian buatan tangan); nama-kata semacam itu tetap dapat dikenali
    transformer dari konteks. Catatan: filter frekuensi (wordfreq) sengaja tidak
    dipakai karena nama umum justru berfrekuensi tinggi sehingga ikut terbuang.
    """
    text = fetch_text(URLS["names_id"])
    raw = {w for line in text.splitlines() if (w := simple_word(line)) and len(w) >= 3}
    kbbi_words = read_lines(LEXICON_ROOT / "word_quality" / "kbbi_id.txt")
    target = LEXICON_ROOT / "ner" / "names_id.txt"
    # Filter SELURUH gabungan (terkurasi + online) terhadap KBBI sehingga berkas
    # selalu bebas kata baku — termasuk membersihkan entri lama yang ambigu
    # (mis. "bunga", "cinta") penyebab false positive yang sudah ada sebelumnya.
    combined = (read_lines(target) | raw) - kbbi_words
    return write_lines(target, combined)

def write_report(report: dict[str, object]) -> None:
    report["generated_at"] = datetime.now(timezone.utc).isoformat()
    target = LEXICON_ROOT / "import_report.json"
    target.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")

def main() -> None:
    parser = argparse.ArgumentParser(description="Import online lexicons into resources/lexicons.")
    parser.add_argument("--all", action="store_true", help="Run all importers.")
    parser.add_argument("--slang-id", action="store_true", help="Import nasalsabila/kamus-alay (terkurasi).")
    parser.add_argument("--slang-id-big", action="store_true",
                        help="Import kamus alay besar okkyibrohim (~15rb entri, lebih lengkap).")
    parser.add_argument("--stopwords-id", action="store_true",
                        help="Import stopwords Indonesia (stopwords-iso, agregasi Sastrawi dkk.).")
    parser.add_argument("--kbbi-id", action="store_true",
                        help="Import kosakata baku KBBI/wordlist (kurangi false positive typo).")
    parser.add_argument("--profanity-id", action="store_true", help="Import Indonesian abusive lexicon.")
    parser.add_argument("--names-id", action="store_true",
                        help="Import daftar nama depan Indonesia (perkaya recall NER).")
    args = parser.parse_args()
    report: dict[str, object] = {}
    if args.all or args.slang_id:
        report["word_quality/slang_id.tsv (salsabila)"] = import_kamus_alay()
    if args.all or args.slang_id_big:
        report["word_quality/slang_id.tsv (okky_kamusalay)"] = import_okky_kamusalay()
    if args.all or args.stopwords_id:
        report["language/stopwords_id.txt"] = import_stopwords_id()
    if args.all or args.kbbi_id:
        report["word_quality/kbbi_id.txt"] = import_kbbi_id()
    if args.all or args.profanity_id:
        report["profanity/external_id.txt"] = import_okky_abusive()
    if args.all or args.names_id:
        report["ner/names_id.txt"] = import_names_id()
    if not report:
        parser.error("Pilih salah satu importer atau gunakan --all.")
    write_report(report)
    print(json.dumps(report, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()