"""General-purpose noisy-romaji rescue for the Windows hybrid fast path. This module is a *generic* recovery route, not a per-input dictionary patch. It is invoked by ``infer_fast.py`` only after the existing exact/segment/fuzzy routes and the generic kana fallback all decline, and strictly before the neural model fallback. Because every currently-passing acceptance gate resolves through an earlier route (model count 0), inserting this stage there cannot change a passing gate row: it only ever steals work from the neural fallback. Three cooperating pieces: 1. ``canonicalize_romaji_variants`` -- deterministic romaji canonicalization: IME small-tsu (``xtu``/``ltu`` + consonant -> gemination), repeated-character run collapse (vowel-aware so ``ou``/``oo``/``ei``/``ee`` long vowels survive), plus an extra long-vowel-reduced variant tried *in addition* (never as the sole mutation). Style differences (wapuro ``sy`` vs Hepburn ``sh`` etc.) are handled per-piece at match time via ``canon_style`` so string positions stay aligned with the kana-fill layer. 2. ``general_phrase_rescue`` -- an anchor-and-fill beam Viterbi over the general reading lexicon plus a small reusable colloquial/particle layer. High confidence dictionary anchors are found by exact or budgeted-fuzzy matching; spans that no anchor can cover are bridged by a lenient kana fill that may drop at most a couple of stray consonants. This is the generic upgrade of the old greedy, exact-only, all-or-nothing ``generic_romaji_fallback`` walk. 3. A confidence gate -- the result is emitted only when coverage, edit budget, ambiguity margin and output plausibility all pass; otherwise the function abstains (returns ``None``) and the caller proceeds to the neural model. The worst case is therefore today's behaviour, never a confident wrong answer. """ from __future__ import annotations import math import re from collections import Counter from romaji_kana import ( GENERIC_PHRASES, romaji_to_hiragana, ) GENERAL_PHRASE_VERSION = "general-phrase-v4-exactonly-prefuzzy-functionrun-20260614" # --------------------------------------------------------------------------- # # 1. Romaji canonicalization # --------------------------------------------------------------------------- # # IME small-tsu marker before a consonant geminates that *following* consonant. # moxtute -> mo + t + te -> motte ; ixtukai -> i + k + kai -> ikkai _SOKUON_RE = re.compile(r"(?:x|l)ts?u(?=([bcdfghjkmpqrstvwyz]))") # A run of the same letter, length >= 3. _RUN_RE = re.compile(r"([a-z])\1{2,}") # A doubled vowel (for the optional long-vowel-reduced variant only). _DOUBLE_VOWEL_RE = re.compile(r"([aiueo])\1") # A common stray-w typo around te-form progressive input: # mottewruu / mottewru -> motteru _TE_WRU_RE = re.compile(r"tewruu?") _VOWELS = frozenset("aiueo") _CONSONANTS = frozenset("bcdfghjklmnpqrstvwxyz") # Wapuro / Hepburn -> single canonical romaji style. Applied identically to both # dictionary keys (at index build) and input pieces (at match time), so any # consistent target works; we collapse toward kunrei-ish forms. Order matters. _STYLE_SUBS: tuple[tuple[str, str], ...] = ( ("tsu", "tu"), ("shi", "si"), ("sh", "sy"), ("chi", "ti"), ("ch", "ty"), ("jy", "zy"), ("ji", "zi"), ("j", "zy"), ("fu", "hu"), ("cy", "ty"), ) def canon_style(s: str) -> str: """Collapse wapuro/Hepburn spelling differences to one canonical style.""" for old, new in _STYLE_SUBS: if old in s: s = s.replace(old, new) return s def _geminate_sokuon(s: str) -> str: return _SOKUON_RE.sub(lambda m: m.group(1), s) def _collapse_runs(s: str) -> str: return _RUN_RE.sub(lambda m: m.group(1) * (2 if m.group(1) in _VOWELS else 1), s) def canonicalize_romaji_variants(inp: str) -> list[str]: """Return ordered, de-duplicated canonical candidate strings. The first element is always the untouched input so that the rescue never *only* sees an aggressively rewritten form. """ base = _collapse_runs(_geminate_sokuon(inp)) reduced = _DOUBLE_VOWEL_RE.sub(r"\1", base) candidates = [inp, base, reduced] # Additive, gated variant only. This does not rewrite arbitrary "wruu"; # it only repairs the reusable te-form/progressive shape "...tewru(u)". for v in (inp, base, reduced): fixed = _TE_WRU_RE.sub("teru", v) if fixed != v: candidates.append(fixed) out: list[str] = [] for v in candidates: if v and v not in out: out.append(v) return out # --------------------------------------------------------------------------- # # 2. Reusable colloquial / particle layer (generic building blocks, NOT # memorized input->output sentences). Keys are wapuro-ish; canon_style makes # them style-agnostic. These compose via the Viterbi rather than matching a # whole utterance. # --------------------------------------------------------------------------- # COLLOQUIAL: dict[str, str] = { # particles and connectives "no": "の", # の "wa": "は", # は (topic; spelled wa) "o": "を", # を (object; spelled o) "ga": "が", # が "ni": "に", # に "de": "で", # で "to": "と", # と "mo": "も", # も "ne": "ね", # ね "yo": "よ", # よ "na": "な", # な "ya": "や", # や "demo": "でも", # でも (connective; overrides general デモ in rescue) "kedo": "けど", # けど "node": "ので", # ので "kara": "から", # から "made": "まで", # まで "toka": "とか", # とか "nara": "なら", # なら "noni": "のに", # のに "yone": "よね", # よね "dayone": "だよね", # だよね "dane": "だね", # だね "kana": "かな", # かな "desu": "です", # です "masu": "ます", # ます # common verb/adjective endings (te-form, progressive, volitional helpers) "teru": "てる", # てる "teiru": "ている", # ている "teta": "てた", # てた "chau": "ちゃう", # ちゃう "chatta": "ちゃった", # ちゃった "toku": "とく", # とく "naide": "ないで", # ないで "nakya": "なきゃ", # なきゃ "naito": "ないと", # ないと "tai": "たい", # たい "tara": "たら", # たら "tari": "たり", # たり # high-frequency spoken content chunks as reusable units "motteru": "持ってる", # 持ってる "motteiru": "持っている", # 持っている "motte": "持って", # 持って "imamotteru": "今持ってる", # 今持ってる "imamotteiru": "今持っている", # 今持っている "imamotte": "今持って", # 今持って "haninara": "範囲なら", # 範囲なら "haninaraba": "範囲ならば", # 範囲ならば "hanide": "範囲で", # 範囲で "ittan": "一旦", # 一旦 "itannsyuuryou": "一旦終了", # common extra-n typo, only in this context "itannshuuryou": "一旦終了", # Hepburn-ish style variant before canon "syuuryou": "終了", # 終了 "shuuryou": "終了", # 終了 "iiyo": "いいよ", # いいよ "ii": "いい", # いい } # These chunks are allowed, but a long island made only from them is risky in # unsegmented text: e.g. "watara" can otherwise be parsed as "wa"+"tara". FUNCTION_COLLOQUIAL_KEYS = frozenset({ "no", "wa", "o", "ga", "ni", "de", "to", "mo", "ne", "yo", "na", "ya", "demo", "kedo", "node", "kara", "made", "toka", "nara", "noni", "yone", "dayone", "dane", "kana", "desu", "masu", "teru", "teiru", "teta", "chau", "chatta", "toku", "naide", "nakya", "naito", "tai", "tara", "tari", }) CONTENT_COLLOQUIAL_KEYS = frozenset({ "iiyo", "ii", }) # --------------------------------------------------------------------------- # # 3. Index # --------------------------------------------------------------------------- # def _has_kanji(text: str) -> bool: for ch in text: o = ord(ch) if 0x3400 <= o <= 0x9FFF or 0xF900 <= o <= 0xFAFF: return True return False def _char_grams(text: str) -> set[str]: # Local copy of infer_fast.char_grams to avoid an import cycle at module # load. Must stay behaviourally identical. if len(text) <= 3: return {text} width = 2 if len(text) <= 10 else 3 return {text[i:i + width] for i in range(0, len(text) - width + 1)} def _tier_cost(key: str, value: str, *, colloquial: bool) -> float: if colloquial: return 0.05 if len(key) <= 8 else 0.12 if len(key) <= 1: return 0.45 # Longer dictionary keys are more confident; bias the search toward them. return max(0.12, 0.34 - 0.02 * len(key)) def build_general_phrase_index(general_lexicon: dict[str, str]) -> dict: """Build the matching index once (lazily, on first rescue).""" table: dict[str, str] = {} colloquial_keys: set[str] = set() for k, v in general_lexicon.items(): if k: table.setdefault(k, v) for k, v in GENERIC_PHRASES.items(): table[k] = v for k, v in COLLOQUIAL.items(): table[k] = v colloquial_keys.add(k) canon: dict[str, list[tuple[str, str, float, bool, bool]]] = {} gram: dict[str, list[str]] = {} by_len: dict[int, list[str]] = {} for k, v in table.items(): ck = canon_style(k) cost = _tier_cost(k, v, colloquial=k in colloquial_keys) is_content = _has_kanji(v) or k in CONTENT_COLLOQUIAL_KEYS is_function = k in FUNCTION_COLLOQUIAL_KEYS canon.setdefault(ck, []).append((k, v, cost, is_content, is_function)) for ck in canon: by_len.setdefault(len(ck), []).append(ck) for g in _char_grams(ck): gram.setdefault(g, []).append(ck) # Collapse each canonical key to its single best (cheapest) entry; record # whether the canonical key is value-ambiguous (multiple distinct outputs). best: dict[str, tuple[str, float, bool, bool, bool]] = {} for ck, entries in canon.items(): entries.sort(key=lambda e: (e[2], -len(e[0]))) value = entries[0][1] cost = entries[0][2] is_content = entries[0][3] is_function = entries[0][4] distinct_values = {e[1] for e in entries} ambiguous = len(distinct_values) > 1 best[ck] = (value, cost, is_content, ambiguous, is_function) return { "best": best, "gram": gram, "by_len": sorted(by_len.keys()), "max_key_len": max((len(ck) for ck in best), default=1), "version": GENERAL_PHRASE_VERSION, } # --------------------------------------------------------------------------- # # 4. Piece matching + lenient kana fill # --------------------------------------------------------------------------- # def _piece_budget(length: int) -> int: if length <= 2: return 0 if length <= 5: return 1 return 2 # Cap fuzzy candidates per piece (ranked by shared-gram overlap) so the search # stays a few-ms operation even against a 16k-entry general lexicon. Mirrors the # bounded-candidate strategy already used by infer_fast.fuzzy_lexicon_match. FUZZY_CANDIDATE_LIMIT = 24 _WEIGHTED_EDIT_DISTANCE = None def _wed(a: str, b: str, max_dist: float) -> float: """Weighted edit distance, importing infer_fast's implementation once. The import is deferred to first call to avoid an import cycle at module load (infer_fast imports this module at top level).""" global _WEIGHTED_EDIT_DISTANCE if _WEIGHTED_EDIT_DISTANCE is None: from infer_fast import weighted_edit_distance _WEIGHTED_EDIT_DISTANCE = weighted_edit_distance return _WEIGHTED_EDIT_DISTANCE(a, b, max_dist=max_dist) def _match_piece(piece: str, index: dict, budget: int, cache: dict | None = None): """Return (value, cost, dist, is_content, is_function) or None.""" cp = canon_style(piece) ckey = (cp, budget) if cache is not None and ckey in cache: return cache[ckey] best = index["best"] hit = best.get(cp) if hit is not None: value, cost, is_content, _ambiguous, is_function = hit res = (value, cost, 0, is_content, is_function) if cache is not None: cache[ckey] = res return res if budget <= 0: if cache is not None: cache[ckey] = None return None gram = index["gram"] counts: Counter = Counter() for g in _char_grams(cp): for ck in gram.get(g, ()): # canonical keys sharing a gram if abs(len(ck) - len(cp)) <= budget: counts[ck] += 1 best_dist = None best_cost = None best_value = None best_is_content = False tie_values: set[str] = set() for ck, _shared in counts.most_common(FUZZY_CANDIDATE_LIMIT): dist = _wed(cp, ck, float(budget)) if dist > budget: continue value, cost, is_content, _ambiguous, is_function = best[ck] cand = (round(dist, 6), cost) if best_dist is None or cand < (best_dist, best_cost): best_dist, best_cost = cand best_value, best_is_content = value, (is_content, is_function) tie_values = {value} elif cand == (best_dist, best_cost): tie_values.add(value) if best_value is None or len(tie_values) > 1: # No match, or a genuinely ambiguous fuzzy repair: do not guess. res = None else: # Return the raw tier; the caller adds segment / fuzzy / length costs. is_content, is_function = best_is_content res = (best_value, best_cost, best_dist, is_content, is_function) if cache is not None: cache[ckey] = res return res def _lenient_kana_fill(span: str, max_drop: int): """Convert a noisy romaji span to kana, optionally dropping <= max_drop stray consonants. Returns (kana, drops) or None.""" direct = romaji_to_hiragana(span) if direct is not None and direct: return (direct, 0) if max_drop <= 0 or len(span) < 2: return None for i, ch in enumerate(span): if ch in _CONSONANTS: trimmed = span[:i] + span[i + 1:] if not trimmed: continue kana = romaji_to_hiragana(trimmed) if kana is not None and kana: return (kana, 1) return None # --------------------------------------------------------------------------- # # 5. Anchor-and-fill beam Viterbi + confidence gate # --------------------------------------------------------------------------- # # Tunable thresholds. Conservative by design: prefer abstaining (-> neural # model) over emitting a low-confidence answer. MIN_LEN = 8 MAX_LEN = 200 MAX_PIECE = 16 MAX_FILL_SPAN = 12 BEAM_WIDTH = 16 # Cost model. Each dictionary segment costs a small base plus its tier (cheap # for particles/colloquial units, dearer for content words), so the natural # segmentation -- e.g. の + 範囲 rather than a single fuzzy 模範 -- wins, while a # mild base still discourages over-fragmentation. Fuzzy and fill edges cost # strictly more so exact dictionary anchors are preferred. SEG_BASE = 0.15 # base cost per dictionary segment (anti-fragmentation) LEN_BONUS = 0.02 # per-char discount: prefer longest match, breaks ties MIN_DICT_EDGE_COST = 0.03 # long reusable chunks must never create negative cost FUZZY_PENALTY = 0.30 # extra cost for using a fuzzy (non-exact) anchor FUZZY_DIST_WEIGHT = 0.40 FILL_COST_PER_CHAR = 0.50 # kana fill is dearer per char than a dict anchor DROP_PENALTY = 0.40 # per dropped stray consonant in a fill ANCHOR_MIN_RATIO = 0.5 # >= this fraction of chars covered by dict anchors FILL_MAX_RATIO = 0.5 # <= this fraction covered by kana fill MIN_AVG_SEG_LEN = 1.5 # anchor chars / dict segments; blocks char-by-char MAX_DROPS = 2 MAX_FUNCTION_RUN = 5 # blocks wa+tara-style long function-only islands EDIT_RATIO = 0.25 MAX_AVG_COST = 0.4 # total cost / chars; blocks heavy fuzzy/fill parses OUTPUT_MIN_RATIO = 0.12 OUTPUT_MAX_RATIO = 1.3 # Canon-style collapse already removes true homophones from the lattice, so a # near-tie here is usually a particle-vs-content re-parse where the cheapest # (rank 1) reading is the intended one. Abstain only on a genuine dead heat. TIGHT_MARGIN = 0.12 # abstain only if a *different* output is this close class _State: __slots__ = ( "cost", "edits", "fill", "anchor", "content", "drops", "segs", "out", "func_run", "max_func_run", ) def __init__( self, cost, edits, fill, anchor, content, drops, segs, out, func_run=0, max_func_run=0, ): self.cost = cost self.edits = edits self.fill = fill self.anchor = anchor self.content = content self.drops = drops self.segs = segs self.out = out self.func_run = func_run self.max_func_run = max_func_run def _anchor_starts(s: str, index: dict) -> list[bool]: n = len(s) best = index["best"] max_len = min(index["max_key_len"], MAX_PIECE) starts = [False] * (n + 1) for i in range(n): for length in range(1, min(max_len, n - i) + 1): if canon_style(s[i:i + length]) in best: starts[i] = True break return starts def _run_beam(s: str, index: dict, *, aggressive: bool, allow_fuzzy: bool = True): """Run the anchor-and-fill beam; return ranked distinct-output states.""" n = len(s) anchor_start = _anchor_starts(s, index) max_edits = math.ceil(EDIT_RATIO * n) match_cache: dict = {} fill_cache: dict = {} beams: list[list[_State]] = [[] for _ in range(n + 1)] beams[0] = [_State(0.0, 0, 0, 0, 0, 0, 0, "")] for i in range(n): bucket = beams[i] if not bucket: continue # prune: keep cheapest per distinct output best_by_out: dict[str, _State] = {} for st in bucket: cur = best_by_out.get(st.out) if cur is None or st.cost < cur.cost: best_by_out[st.out] = st pruned = sorted(best_by_out.values(), key=lambda st: st.cost)[:BEAM_WIDTH] beams[i] = pruned for st in pruned: # dictionary edges (exact or budgeted fuzzy) max_j = min(i + MAX_PIECE, n) for j in range(i + 1, max_j + 1): length = j - i budget = _piece_budget(length) if allow_fuzzy else 0 m = _match_piece(s[i:j], index, budget, match_cache) if m is None: continue value, tier, dist, is_content, is_function = m new_edits = st.edits + int(round(dist)) if new_edits > max_edits: continue edge = max(MIN_DICT_EDGE_COST, SEG_BASE + tier - LEN_BONUS * length) if dist > 0: edge += FUZZY_PENALTY + FUZZY_DIST_WEIGHT * dist func_run = st.func_run + length if is_function else 0 max_func_run = max(st.max_func_run, func_run) beams[j].append(_State( st.cost + edge, new_edits, st.fill, st.anchor + length, st.content + (1 if is_content else 0), st.drops, st.segs + 1, st.out + value, func_run, max_func_run, )) # lenient kana-fill edges: bridge noise to the next anchor or to end max_fill_j = min(i + MAX_FILL_SPAN, n) for j in range(i + 1, max_fill_j + 1): if j != n and not anchor_start[j]: continue fkey = (i, j, MAX_DROPS - st.drops) if fkey in fill_cache: filled = fill_cache[fkey] else: filled = _lenient_kana_fill(s[i:j], MAX_DROPS - st.drops) fill_cache[fkey] = filled if filled is None: continue kana, drops = filled length = j - i beams[j].append(_State( st.cost + FILL_COST_PER_CHAR * length + DROP_PENALTY * drops, st.edits, st.fill + length, st.anchor, st.content, st.drops + drops, st.segs, st.out + kana, 0, st.max_func_run, )) finals = beams[n] if not finals: return [] by_out: dict[str, _State] = {} for st in finals: cur = by_out.get(st.out) if cur is None or st.cost < cur.cost: by_out[st.out] = st return sorted(by_out.values(), key=lambda st: st.cost) def _solve_variant(s: str, index: dict, *, aggressive: bool, allow_fuzzy: bool = True): n = len(s) anchor_min_ratio = ANCHOR_MIN_RATIO - (0.1 if aggressive else 0.0) max_edits = math.ceil(EDIT_RATIO * n) ranked = _run_beam(s, index, aggressive=aggressive, allow_fuzzy=allow_fuzzy) if not ranked: return None best = ranked[0] # --- confidence gate --- if best.content < 1: return None if best.anchor < anchor_min_ratio * n: return None if best.fill > FILL_MAX_RATIO * n: return None if best.drops > MAX_DROPS: return None if best.max_func_run > MAX_FUNCTION_RUN: return None if best.edits > max_edits: return None if best.cost > MAX_AVG_COST * n: return None if best.segs > 0 and best.anchor / best.segs < MIN_AVG_SEG_LEN: return None # degenerate char-by-char dictionary spam out_len = len(best.out) if not (OUTPUT_MIN_RATIO * n <= out_len <= OUTPUT_MAX_RATIO * n): return None if len(ranked) >= 2 and (ranked[1].cost - best.cost) < TIGHT_MARGIN: return None # a different output is nearly as cheap: genuinely ambiguous return { "output": best.out, "cost": best.cost, "cost_per_char": best.cost / max(1, n), "anchor_ratio": best.anchor / max(1, n), "fill_ratio": best.fill / max(1, n), "edits": best.edits, "drops": best.drops, "segs": best.segs, "max_function_run": best.max_func_run, "allow_fuzzy": allow_fuzzy, } def debug_parses( s: str, index: dict, *, aggressive: bool = False, allow_fuzzy: bool = True, topk: int = 8, ): """Return the top-k full-cover parses (pre-gate) for diagnostics.""" ranked = _run_beam(s, index, aggressive=aggressive, allow_fuzzy=allow_fuzzy) n = max(1, len(s)) out = [] for st in ranked[:topk]: out.append({ "output": st.out, "cost": round(st.cost, 4), "cost_per_char": round(st.cost / n, 4), "anchor": st.anchor, "anchor_ratio": round(st.anchor / n, 3), "fill": st.fill, "segs": st.segs, "content": st.content, "edits": st.edits, "drops": st.drops, "max_function_run": st.max_func_run, }) return out def general_phrase_rescue( inp: str, index: dict, *, aggressive: bool = False, exact_only: bool = False, ): """Generic noisy-romaji rescue. Returns (output, meta) or None. ``inp`` must already be normalized by ``normalize_input`` (lowercase, no spaces/soft separators) -- it is passed through unchanged from the fast path. """ if not inp or any(ch.isdigit() for ch in inp): return None if not (MIN_LEN <= len(inp) <= MAX_LEN): return None variants = [v for v in canonicalize_romaji_variants(inp) if MIN_LEN <= len(v) <= MAX_LEN] best_result = None # Most practical noise becomes exact after deterministic canonicalization. # Try that cheap lattice first; only pay fuzzy WED costs if every exact-only # parse abstains. for allow_fuzzy in ((False,) if exact_only else (False, True)): for variant in variants: res = _solve_variant(variant, index, aggressive=aggressive, allow_fuzzy=allow_fuzzy) if res is None: continue if not allow_fuzzy and (res["fill_ratio"] > 0 or res["drops"] > 0): continue if best_result is None or res["cost_per_char"] < best_result["cost_per_char"]: best_result = res if best_result is not None: break if best_result is None: return None return best_result["output"], best_result