kfe.py

"""Trimmed khmer_frontend for the demo Space: normalize + numbers + chunking.

Vendored subset of the main repo's khmer_frontend (kept self-contained so the
Space needs no private code). Keep in sync when the main front-end changes.
"""

from __future__ import annotations

import re
import unicodedata

ZWNJ, ZWJ = "‌", "‍"

_CONFUSABLES = str.maketrans({"็": "៏", "ํ": "ំ"})
_UNSPOKEN = str.maketrans({c: " " for c in "–—‑()[]{}«»<>|/*_~"})


def normalize(text: str) -> str:
    if not text:
        return ""
    text = unicodedata.normalize("NFC", text)
    text = text.translate(_CONFUSABLES).translate(_UNSPOKEN)
    text = text.replace(ZWNJ, "").replace(ZWJ, "")
    return " ".join(text.split())


_UNITS = ["សូន្យ", "មួយ", "ពីរ", "បី", "បួន", "ប្រាំ",
          "ប្រាំមួយ", "ប្រាំពីរ", "ប្រាំបី", "ប្រាំបួន"]
_TENS = ["", "ដប់", "ម្ភៃ", "សាមសិប", "សែសិប", "ហាសិប",
         "ហុកសិប", "ចិតសិប", "ប៉ែតសិប", "កៅសិប"]
_KH_DIGITS = str.maketrans("០១២៣៤៥៦៧៨៩", "0123456789")


def int_to_khmer(n: int) -> str:
    if n < 0:
        return "ដក" + int_to_khmer(-n)
    if n < 10:
        return _UNITS[n]
    if n < 100:
        tens, unit = divmod(n, 10)
        return _TENS[tens] + (_UNITS[unit] if unit else "")
    for value, word in ((1_000_000, "លាន"), (100_000, "សែន"),
                        (10_000, "ម៉ឺន"), (1_000, "ពាន់"), (100, "រយ")):
        if n >= value:
            head, rest = divmod(n, value)
            return int_to_khmer(head) + word + (int_to_khmer(rest) if rest else "")
    return _UNITS[0]


def _read_number(token: str) -> str:
    token = token.translate(_KH_DIGITS).replace(",", "")
    suffix = ""
    if token.endswith("%"):
        token, suffix = token[:-1], "ភាគរយ"
    if "." in token:
        whole, frac = token.split(".", 1)
        words = int_to_khmer(int(whole or 0)) + "ក្បៀស" + \
            "".join(_UNITS[int(d)] for d in frac if d.isdigit())
    else:
        words = int_to_khmer(int(token)) if token else ""
    return words + suffix


_NUM_RE = re.compile(r"[0-9០-៩][0-9០-៩,]*(?:\.[0-9០-៩]+)?%?")


def normalize_numbers(text: str) -> str:
    return _NUM_RE.sub(lambda m: _read_number(m.group(0)), text).replace("៛", "រៀល")


_SENT_END = "។៕!?…"
_NEVER_BEFORE = ("ខែ", "ឆ្នាំ", "នាទី", "ម៉ោង", "រៀល", "ដុល្លារ", "នាក់", "ភាគរយ")
_GOOD_BEFORE = ("នៅ", "កាលពី", "ដោយ", "ដែល", "និង", "ព្រោះ", "ដើម្បី",
                "បន្ទាប់", "ក្នុង", "ចំពោះ", "តាម", "រួម")
_DEPENDENT = set("ាិីឹឺុូួើឿៀេែៃោៅំះៈ៉៊់៌៍៎៏័៑្")


def _split_syllables(chunk: str) -> list[str]:
    out, cur, i, n = [], "", 0, len(chunk)
    while i < n:
        ch = chunk[i]
        if ch == "្":
            cur += ch
            if i + 1 < n:
                cur += chunk[i + 1]
                i += 2
                continue
            i += 1
            continue
        if ch in _DEPENDENT:
            cur += ch
            i += 1
            continue
        if cur:
            out.append(cur)
        cur = ch
        i += 1
    if cur:
        out.append(cur)
    return out


def _safe_cut(text: str, limit: int) -> int:
    spaces = [i for i, c in enumerate(text[: limit + 1]) if c == " "]
    if spaces:
        def nxt(i):
            return text[i + 1: i + 12].lstrip()
        allowed = [i for i in spaces if not nxt(i).startswith(_NEVER_BEFORE)]
        if allowed:
            good = [i for i in allowed if i >= limit * 0.4
                    and nxt(i).startswith(_GOOD_BEFORE)]
            return max(good) if good else max(allowed)
        return max(spaces)
    n = 0
    for syl in _split_syllables(text):
        if n + len(syl) > limit and n:
            return n
        n += len(syl)
    return len(text)


def chunk_khmer_text(text: str, max_chars: int = 110) -> list[str]:
    text = normalize_numbers(normalize(text))
    sentences, cur = [], ""
    for ch in text:
        cur += ch
        if ch in _SENT_END:
            sentences.append(cur.strip())
            cur = ""
    if cur.strip():
        sentences.append(cur.strip())

    sentence_max = 250
    chunks, buf = [], ""
    for s in sentences:
        if len(s) > sentence_max:
            if buf:
                chunks.append(buf)
                buf = ""
            while len(s) > sentence_max:
                cut = _safe_cut(s, sentence_max)
                chunks.append(s[:cut].strip())
                s = s[cut:].strip()
            chunks.append(s)
            continue
        if buf and len(buf) + len(s) + 1 > max_chars:
            chunks.append(buf)
            buf = s
        else:
            buf = f"{buf} {s}".strip()
    if buf:
        chunks.append(buf)
    return [c for c in chunks if c] or [""]