Spaces:
Running on Zero
Running on Zero
| """Production inference helpers for the BlueMagpie-TTS Space.""" | |
| from __future__ import annotations | |
| import math | |
| import operator | |
| import random | |
| import re | |
| import unicodedata | |
| from dataclasses import dataclass | |
| from datetime import date | |
| from decimal import Decimal, InvalidOperation, localcontext | |
| from typing import Sequence | |
| from urllib.parse import urlsplit | |
| import numpy as np | |
| import torch | |
| from opencc import OpenCC | |
| from pypinyin import Style, lazy_pinyin | |
| from torch import nn | |
| _SPACE_RE = re.compile(r"\s+") | |
| _PUNCT_NO_LEFT_SPACE_RE = re.compile(r"\s+([，。！？；：、,.!?;:])") | |
| _CJK_PUNCT_RIGHT_SPACE_RE = re.compile(r"([，。！？；：、])\s+") | |
| _BOPOMOFO_TONES = {"ˊ", "ˇ", "ˋ", "˙"} | |
| _BOPOMOFO_ASR_POPO_RE = re.compile( | |
| r"(?i)(?<![a-z0-9])(?:po|bo)\s*(?:po|bo)\s*(?:mo|摸)(?![a-z0-9])" | |
| ) | |
| _ASR_HOMOPHONE_TRANSLATION = str.maketrans( | |
| { | |
| "她": "他", | |
| "它": "他", | |
| "牠": "他", | |
| "祂": "他", | |
| "妳": "你", | |
| } | |
| ) | |
| _TERMINAL_PUNCTUATION = frozenset("。！？.!?") | |
| _NONTERMINAL_TRAILING_PUNCTUATION = frozenset("，,、；;：:") | |
| _TRAILING_CLOSERS = frozenset("\"'”’」』】）)]}") | |
| _BOPOMOFO_READINGS = { | |
| "ㄅ": "波", "ㄆ": "坡", "ㄇ": "摸", "ㄈ": "佛", "ㄉ": "得", | |
| "ㄊ": "特", "ㄋ": "呢", "ㄌ": "了", "ㄍ": "哥", "ㄎ": "科", | |
| "ㄏ": "喝", "ㄐ": "基", "ㄑ": "欺", "ㄒ": "希", "ㄓ": "知", | |
| "ㄔ": "吃", "ㄕ": "師", "ㄖ": "日", "ㄗ": "資", "ㄘ": "雌", | |
| "ㄙ": "思", "ㄚ": "啊", "ㄛ": "喔", "ㄜ": "鵝", "ㄝ": "欸", | |
| "ㄞ": "哀", "ㄟ": "欸", "ㄠ": "凹", "ㄡ": "歐", "ㄢ": "安", | |
| "ㄣ": "恩", "ㄤ": "昂", "ㄥ": "鞥", "ㄦ": "兒", "ㄧ": "衣", | |
| "ㄨ": "烏", "ㄩ": "迂", | |
| } | |
| _ZH_DIGITS = "零一二三四五六七八九" | |
| _ZH_SMALL_UNITS = ("", "十", "百", "千") | |
| _ZH_LARGE_UNITS = ("", "萬", "億", "兆") | |
| _SPOKEN_DATE_RE = re.compile( | |
| r"(?<![A-Za-z0-9/])(\d{4})([/-])(\d{1,2})\2(\d{1,2})(?![A-Za-z0-9/])" | |
| ) | |
| _SPOKEN_ZH_DATE_RE = re.compile( | |
| r"(?<![A-Za-z0-9])(\d{4})\s*年\s*(\d{1,2})\s*月\s*" | |
| r"(\d{1,2})\s*(?:日|號)(?![A-Za-z0-9])" | |
| ) | |
| _SPOKEN_TIME_RE = re.compile( | |
| r"(?<![A-Za-z0-9:])([01]?\d|2[0-3]):([0-5]\d)(?![A-Za-z0-9:])" | |
| ) | |
| _ASR_EXPLICIT_ZH_TIME_RE = re.compile( | |
| r"(?<![A-Za-z0-9])([01]?\d|2[0-3])[\u9ede\u70b9\u6642\u65f6]" | |
| r"([0-5]?\d)\u5206(?![A-Za-z0-9])" | |
| ) | |
| _ASR_BARE_ZH_TIME_RE = re.compile( | |
| r"(?<![A-Za-z0-9])([01]?\d|2[0-3])[\u9ede\u70b9\u6642\u65f6]" | |
| r"([0-5]?\d)(?![A-Za-z0-9]|\u5206|\u6beb?\u79d2|\u5fae\u79d2|" | |
| r"\u516c\u65a4|\u5343\u514b|\u516c\u514b|\u6beb\u514b|\u5ea6)" | |
| ) | |
| _ASR_FORMATTED_CLOCK_RE = re.compile( | |
| r"(?<![A-Za-z0-9])([01]?\d|2[0-3])\s*" | |
| r"[.:\u9ede\u70b9\u6642\u65f6]\s*([0-5]?\d)(?:\s*\u5206)?" | |
| r"(?![A-Za-z0-9])" | |
| ) | |
| _ASR_ARABIC_DIGIT_RUN_RE = re.compile( | |
| r"(?<![A-Za-z0-9])\d+(?![A-Za-z0-9])" | |
| ) | |
| _ASR_XIAOLAOSHU_ALIAS_RE = re.compile(r"(?i)xiaolaoshu") | |
| _ASR_STRUCTURED_NUMERIC_NEIGHBORS = frozenset( | |
| ".,:/\\-+_@#年月日時时分秒點点%％" | |
| ) | |
| _ASR_FREQUENCY_MEASUREMENT_RE = re.compile( | |
| r"(?i)(?<![A-Za-z0-9./:\\+_@#-])" | |
| r"([+]?\d+(?:\.\d+)?)\s*" | |
| r"(GHz|MHz|kHz|Hz|吉赫茲|兆赫茲|千赫茲|赫茲|吉赫兹|兆赫兹|千赫兹|赫兹)" | |
| r"(?![A-Za-z0-9])" | |
| ) | |
| _REJECT_ONLY_FREQUENCY_MEASUREMENT_RE = re.compile( | |
| r"(?i)([+]?\d+(?:\.\d+)?)\s*" | |
| r"(GHz|MHz|kHz|Hz|吉赫茲|兆赫茲|千赫茲|赫茲|吉赫兹|兆赫兹|千赫兹|赫兹)" | |
| r"(?![A-Za-z0-9])" | |
| ) | |
| _ZH_FREQUENCY_DIGITS = "零一二三四五六七八九" | |
| _ZH_FREQUENCY_INTEGER_CHARS = _ZH_FREQUENCY_DIGITS + "十百千萬億兆" | |
| _ZH_FREQUENCY_NUMBER_BOUNDARY_CLASS = ( | |
| "A-Za-z0-9" | |
| + re.escape( | |
| _ZH_FREQUENCY_INTEGER_CHARS + "點点正負负./:\\+_@#-" | |
| ) | |
| ) | |
| _CANONICAL_ZH_FREQUENCY_PATTERNS = tuple( | |
| ( | |
| re.compile( | |
| rf"(?<![{_ZH_FREQUENCY_NUMBER_BOUNDARY_CLASS}])" | |
| rf"(正?[{_ZH_FREQUENCY_INTEGER_CHARS}]+" | |
| rf"(?:點[{_ZH_FREQUENCY_DIGITS}]+)?)\s*{unit}" | |
| r"(?![A-Za-z0-9])" | |
| ), | |
| unit, | |
| ) | |
| # Every valid unit decomposition is retained below. Some frontend forms, | |
| # notably ``一兆赫茲``, are lexically ambiguous between a large cardinal | |
| # followed by Hz and a smaller cardinal followed by MHz. Such targets | |
| # must fail closed instead of relying on unit-order preference. | |
| for unit in ( | |
| "吉赫茲", | |
| "吉赫兹", | |
| "兆赫茲", | |
| "兆赫兹", | |
| "千赫茲", | |
| "千赫兹", | |
| "赫茲", | |
| "赫兹", | |
| ) | |
| ) | |
| _FREQUENCY_SCALE_HZ = { | |
| "hz": Decimal("1"), | |
| "khz": Decimal("1000"), | |
| "mhz": Decimal("1000000"), | |
| "ghz": Decimal("1000000000"), | |
| "赫茲": Decimal("1"), | |
| "千赫茲": Decimal("1000"), | |
| "兆赫茲": Decimal("1000000"), | |
| "吉赫茲": Decimal("1000000000"), | |
| "赫兹": Decimal("1"), | |
| "千赫兹": Decimal("1000"), | |
| "兆赫兹": Decimal("1000000"), | |
| "吉赫兹": Decimal("1000000000"), | |
| } | |
| _SPOKEN_PERCENT_RE = re.compile( | |
| r"(?<![A-Za-z0-9.])([+-]?\d+(?:\.\d+)?)\s*[%％](?![A-Za-z0-9])" | |
| ) | |
| _SPOKEN_UNIT_RE = re.compile( | |
| r"(?<![A-Za-z0-9.])([+-]?\d+(?:\.\d+)?)\s*" | |
| r"(km/h|m/s|ms|kHz|MHz|GHz|Hz|km|cm|mm|kg|mg|mL|ml|kW|°\s*[Cc]|℃|m|g|L|l|W|V)" | |
| r"(?![A-Za-z])" | |
| ) | |
| _SPOKEN_ZH_MEASURE_RE = re.compile( | |
| r"(?<![A-Za-z0-9.\u9ede\u70b9])([+-]?\d+(?:\.\d+)?)\s*" | |
| r"(公里|公尺|公分|公厘|厘米|毫米|公斤|千克|公克|毫克|公升|毫升|" | |
| r"攝氏度|千赫茲|兆赫茲|吉赫茲|赫茲|小時|分鐘|毫秒|米|克|升|度|秒|天|" | |
| r"個|人|次|張|台|件|份|元)" | |
| r"(?![A-Za-z0-9])" | |
| ) | |
| _NETWORK_NONASCII_BOUNDARY_PUNCTUATION = ( | |
| "，。！？；、：（）【】「」『』《》〈〉“”‘’［］｛｝" | |
| "…⋯—–．｡､〔〕〖〗〘〙〚〛﹁﹂＂＇︰﹐﹑﹒﹔﹕﹖﹗" | |
| ) | |
| _SPOKEN_URL_RE = re.compile( | |
| rf"(?i)(?<![A-Za-z0-9])(?:https?://|ftp://|www\.)" | |
| rf"[^\s<>\"{re.escape(_NETWORK_NONASCII_BOUNDARY_PUNCTUATION)}]+" | |
| ) | |
| _SPOKEN_EMAIL_RE = re.compile( | |
| r"(?i)(?<![A-Za-z0-9.!#$%&'*+/=?^_`{|}~-])" | |
| r"[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+@" | |
| r"[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)+" | |
| r"(?![A-Za-z0-9-])" | |
| ) | |
| _UNSUPPORTED_QUOTED_EMAIL_RE = re.compile( | |
| r'(?i)(?<![A-Za-z0-9])"(?:[^"\\]|\\.)*"@(?:' | |
| r"[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)+|\[[^\]\s]+\])" | |
| ) | |
| _SPOKEN_SEMVER_RE = re.compile( | |
| r"(?i)(?<![A-Za-z0-9.])(?:v|version\s+|版本\s*(?:v\s*)?)?" | |
| r"(\d+)\.(\d+)\.(\d+)" | |
| r"(?:-([0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*))?" | |
| r"(?:\+([0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*))?" | |
| r"(?![A-Za-z0-9.])" | |
| ) | |
| _CURRENCY_NUMBER_PATTERN = r"[+-]?(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d+)?" | |
| _CURRENCY_CODE_PATTERN = ( | |
| r"NT\$|TWD|NTD|US\$|USD|HK\$|HKD|CN¥|CNY|RMB|JPY|EUR|GBP|KRW|" | |
| r"\$|€|¥|￥|£|₩" | |
| ) | |
| _SPOKEN_CURRENCY_PREFIX_RE = re.compile( | |
| rf"(?i)(?<![A-Za-z0-9])({_CURRENCY_CODE_PATTERN})\s*" | |
| rf"({_CURRENCY_NUMBER_PATTERN})(?![\d,.])" | |
| ) | |
| _SPOKEN_CURRENCY_SUFFIX_RE = re.compile( | |
| rf"(?i)(?<![A-Za-z0-9.])({_CURRENCY_NUMBER_PATTERN})\s*" | |
| rf"(TWD|NTD|USD|HKD|CNY|RMB|JPY|EUR|GBP|KRW)(?![A-Za-z])" | |
| ) | |
| _MODEL_CODE_RE = re.compile( | |
| r"(?<![A-Za-z0-9])([A-Z]{2,8})[ -]*(\d{1,8})(?![A-Za-z0-9.])" | |
| ) | |
| _ASR_MODEL_CODE_RE = re.compile( | |
| r"(?<![A-Za-z0-9])([A-Za-z]{2,8})[ _-]*(\d{1,8})" | |
| r"(?![A-Za-z0-9.])" | |
| ) | |
| _UPPERCASE_ACRONYM_RE = re.compile(r"(?<![A-Za-z0-9])([A-Z]{2,8})(?![A-Za-z0-9])") | |
| _ZH_UNIT_READINGS = { | |
| "km/h": "公里每小時", | |
| "m/s": "公尺每秒", | |
| "ms": "毫秒", | |
| "khz": "千赫茲", | |
| "mhz": "兆赫茲", | |
| "ghz": "吉赫茲", | |
| "hz": "赫茲", | |
| "km": "公里", | |
| "cm": "公分", | |
| "mm": "毫米", | |
| "kg": "公斤", | |
| "mg": "毫克", | |
| "ml": "毫升", | |
| "kw": "千瓦", | |
| "°c": "攝氏度", | |
| "℃": "攝氏度", | |
| "m": "公尺", | |
| "g": "公克", | |
| "l": "公升", | |
| "w": "瓦", | |
| "v": "伏特", | |
| } | |
| _ZH_CURRENCY_READINGS = { | |
| "NT$": "新台幣", | |
| "TWD": "新台幣", | |
| "NTD": "新台幣", | |
| "US$": "美元", | |
| "USD": "美元", | |
| "$": "美元", | |
| "HK$": "港幣", | |
| "HKD": "港幣", | |
| "CN¥": "人民幣", | |
| "CNY": "人民幣", | |
| "RMB": "人民幣", | |
| "JPY": "日圓", | |
| "¥": "日圓", | |
| "￥": "日圓", | |
| "EUR": "歐元", | |
| "€": "歐元", | |
| "GBP": "英鎊", | |
| "£": "英鎊", | |
| "KRW": "韓元", | |
| "₩": "韓元", | |
| } | |
| _ASR_JPY_ZH_AMOUNT_PATTERN = ( | |
| r"[正負负]?[" | |
| + _ZH_DIGITS | |
| + r"十百千萬万億亿兆]+(?:[點点][" | |
| + _ZH_DIGITS | |
| + r"]+)?" | |
| ) | |
| _ASR_JPY_CURRENCY_AMOUNT_RE = re.compile( | |
| rf"(?P<currency>日[元圓圆])\s*" | |
| rf"(?P<amount>{_CURRENCY_NUMBER_PATTERN}|{_ASR_JPY_ZH_AMOUNT_PATTERN})" | |
| r"(?P<redundant_unit>元)?" | |
| rf"(?![\d,.{_ZH_DIGITS}十百千萬万億亿兆點点元])" | |
| ) | |
| _ASR_JPY_AMOUNT_TRANSLATION = str.maketrans( | |
| {"负": "負", "点": "點", "万": "萬", "亿": "億"} | |
| ) | |
| _ZH_NETWORK_SYMBOL_READINGS = { | |
| ".": "點", | |
| "/": "斜線", | |
| ":": "冒號", | |
| "?": "問號", | |
| "=": "等於", | |
| "&": "和", | |
| "#": "井號", | |
| "@": "小老鼠", | |
| "-": "橫線", | |
| "_": "底線", | |
| "%": "百分號", | |
| "+": "加號", | |
| "~": "波浪號", | |
| "!": "驚嘆號", | |
| "$": "錢字號", | |
| "'": "單引號", | |
| "(": "左括號", | |
| ")": "右括號", | |
| "*": "星號", | |
| ",": "逗號", | |
| ";": "分號", | |
| "[": "左方括號", | |
| "]": "右方括號", | |
| "^": "插入號", | |
| "`": "反引號", | |
| "{": "左大括號", | |
| "|": "豎線", | |
| "}": "右大括號", | |
| } | |
| _ZH_NETWORK_LETTER_READINGS = { | |
| "A": "欸", | |
| "B": "比", | |
| "C": "西", | |
| "D": "迪", | |
| "E": "伊", | |
| "F": "艾夫", | |
| "G": "吉", | |
| "H": "艾取", | |
| "I": "愛", | |
| "J": "傑", | |
| "K": "凱", | |
| "L": "艾爾", | |
| "M": "艾姆", | |
| "N": "恩", | |
| "O": "歐", | |
| "P": "批", | |
| "Q": "丘", | |
| "R": "阿爾", | |
| "S": "艾斯", | |
| "T": "踢", | |
| "U": "優", | |
| "V": "維", | |
| "W": "達不溜", | |
| "X": "艾克斯", | |
| "Y": "歪", | |
| "Z": "茲", | |
| } | |
| _ZH_NETWORK_READING_TO_LETTER = { | |
| reading: letter.casefold() | |
| for letter, reading in _ZH_NETWORK_LETTER_READINGS.items() | |
| } | |
| _ZH_NETWORK_READING_TO_LETTER.update( | |
| { | |
| "爱": "i", | |
| "杰": "j", | |
| "凯": "k", | |
| "艾尔": "l", | |
| "欧": "o", | |
| "阿尔": "r", | |
| "优": "u", | |
| "维": "v", | |
| "达不溜": "w", | |
| "兹": "z", | |
| } | |
| ) | |
| _T2S_CONVERTER = OpenCC("t2s") | |
| class StopHysteresisController(nn.Module): | |
| """Apply probability-threshold hysteresis around a legacy stop head. | |
| The pinned public model predates native ``stop_threshold`` and | |
| ``stop_consecutive`` arguments. This adapter preserves the learned logits | |
| and only changes the final stop decision used by legacy generation. | |
| """ | |
| def __init__( | |
| self, | |
| stop_head: nn.Module, | |
| threshold: float = 0.65, | |
| late_threshold: float = 0.50, | |
| consecutive: int = 2, | |
| late_start_ratio: float = 0.80, | |
| late_full_ratio: float = 1.00, | |
| ) -> None: | |
| super().__init__() | |
| self.stop_head = stop_head | |
| self.threshold = float(threshold) | |
| self.late_threshold = float(late_threshold) | |
| self.consecutive = max(1, int(consecutive)) | |
| self.late_start_ratio = max(0.0, float(late_start_ratio)) | |
| self.late_full_ratio = max( | |
| self.late_start_ratio + 1.0e-6, | |
| float(late_full_ratio), | |
| ) | |
| self._active = False | |
| self._min_len = 2 | |
| self._expected_steps = 0 | |
| self._hard_stop_steps = 0 | |
| self._step = 0 | |
| self._hits = 0 | |
| self.last_probabilities: list[float] = [] | |
| self.last_generated_steps = 0 | |
| self.last_stop_reason = "inactive" | |
| def begin( | |
| self, | |
| min_len: int, | |
| *, | |
| expected_steps: int = 0, | |
| hard_stop_steps: int = 0, | |
| ) -> None: | |
| self._active = True | |
| self._min_len = max(0, int(min_len)) | |
| self._expected_steps = max(0, int(expected_steps)) | |
| self._hard_stop_steps = max(0, int(hard_stop_steps)) | |
| self._step = 0 | |
| self._hits = 0 | |
| self.last_probabilities = [] | |
| self.last_generated_steps = 0 | |
| self.last_stop_reason = "running" | |
| def end(self) -> None: | |
| self._active = False | |
| def forward(self, hidden: torch.Tensor) -> torch.Tensor: | |
| logits = self.stop_head(hidden) | |
| if not self._active: | |
| return logits | |
| probability = float(torch.softmax(logits.float(), dim=-1)[0, 1].detach().cpu()) | |
| self.last_probabilities.append(probability) | |
| generated_steps = self._step + 1 | |
| threshold = self.threshold | |
| if self._expected_steps > 0: | |
| progress = generated_steps / float(self._expected_steps) | |
| if progress >= self.late_start_ratio: | |
| blend = min( | |
| 1.0, | |
| max( | |
| 0.0, | |
| (progress - self.late_start_ratio) | |
| / (self.late_full_ratio - self.late_start_ratio), | |
| ), | |
| ) | |
| threshold = self.threshold + blend * (self.late_threshold - self.threshold) | |
| eligible = self._step > self._min_len | |
| if eligible and probability >= threshold: | |
| self._hits += 1 | |
| else: | |
| self._hits = 0 | |
| should_stop = eligible and self._hits >= self.consecutive | |
| stop_reason = "stop_threshold" if should_stop else "running" | |
| if self._hard_stop_steps > 0 and generated_steps >= self._hard_stop_steps: | |
| if not should_stop: | |
| should_stop = True | |
| stop_reason = "hard_stop" | |
| self._step += 1 | |
| self.last_generated_steps = generated_steps | |
| self.last_stop_reason = stop_reason | |
| decision = torch.zeros_like(logits) | |
| decision[..., 1 if should_stop else 0] = 1.0 | |
| return decision | |
| def set_generation_seed(seed: int) -> None: | |
| random.seed(seed) | |
| np.random.seed(seed) | |
| torch.manual_seed(seed) | |
| if torch.cuda.is_available(): | |
| torch.cuda.manual_seed_all(seed) | |
| def _is_cjk(char: str) -> bool: | |
| codepoint = ord(char) | |
| return ( | |
| 0x3400 <= codepoint <= 0x4DBF | |
| or 0x4E00 <= codepoint <= 0x9FFF | |
| or 0xF900 <= codepoint <= 0xFAFF | |
| or 0x3040 <= codepoint <= 0x30FF | |
| or 0xAC00 <= codepoint <= 0xD7AF | |
| ) | |
| def _is_bopomofo(char: str) -> bool: | |
| codepoint = ord(char) | |
| return 0x3100 <= codepoint <= 0x312F or 0x31A0 <= codepoint <= 0x31BF or char in _BOPOMOFO_TONES | |
| def _zh_four_digit_section(value: int, *, suppress_leading_one: bool = True) -> str: | |
| """Read one non-negative, at-most-four-digit integer in Mandarin.""" | |
| output: list[str] = [] | |
| pending_zero = False | |
| for position in range(3, -1, -1): | |
| divisor = 10**position | |
| digit = value // divisor | |
| value %= divisor | |
| if digit: | |
| if pending_zero and output: | |
| output.append(_ZH_DIGITS[0]) | |
| if not (digit == 1 and position == 1 and not output and suppress_leading_one): | |
| output.append(_ZH_DIGITS[digit]) | |
| output.append(_ZH_SMALL_UNITS[position]) | |
| pending_zero = False | |
| elif output and value: | |
| pending_zero = True | |
| return "".join(output) | |
| def _zh_integer(value: int) -> str: | |
| if value == 0: | |
| return _ZH_DIGITS[0] | |
| if value < 0 or value >= 10**16: | |
| raise ValueError("Mandarin integer normalizer supports values from 0 to 10^16 - 1") | |
| sections: list[int] = [] | |
| while value: | |
| sections.append(value % 10000) | |
| value //= 10000 | |
| output: list[str] = [] | |
| pending_zero = False | |
| for section_index in range(len(sections) - 1, -1, -1): | |
| section = sections[section_index] | |
| if section == 0: | |
| if output and any(sections[:section_index]): | |
| pending_zero = True | |
| continue | |
| if output and (pending_zero or section < 1000): | |
| output.append(_ZH_DIGITS[0]) | |
| output.append(_zh_four_digit_section(section, suppress_leading_one=not output)) | |
| output.append(_ZH_LARGE_UNITS[section_index]) | |
| pending_zero = False | |
| return "".join(output) | |
| def _zh_number(number: str) -> str: | |
| """Read a validated decimal string as a Mandarin cardinal number.""" | |
| sign = "" | |
| if number.startswith(("+", "-")): | |
| sign = "正" if number[0] == "+" else "負" | |
| number = number[1:] | |
| integer, separator, fraction = number.partition(".") | |
| if not integer or not integer.isdigit() or separator and (not fraction or not fraction.isdigit()): | |
| raise ValueError("invalid decimal number") | |
| if len(integer) > 16: | |
| raise ValueError("number is too large for conservative normalization") | |
| spoken = _zh_integer(int(integer)) | |
| if separator: | |
| spoken += "點" + "".join(_ZH_DIGITS[int(digit)] for digit in fraction) | |
| return sign + spoken | |
| def _zh_clock_time(hour_text: str, minute_text: str) -> str: | |
| """Return the canonical Mandarin reading for a validated clock time.""" | |
| hour = int(hour_text) | |
| minute = int(minute_text) | |
| minute_spoken = "整" if minute == 0 else f"{_zh_integer(minute)}分" | |
| return f"{_zh_integer(hour)}點{minute_spoken}" | |
| def _zh_digit_sequence(digits: str) -> str: | |
| if not digits or not digits.isdigit(): | |
| raise ValueError("expected a non-empty digit sequence") | |
| return "".join(_ZH_DIGITS[int(digit)] for digit in digits) | |
| def _nonspace_neighbor(text: str, index: int, step: int) -> str | None: | |
| """Return the nearest non-space character on one side of ``index``.""" | |
| position = int(index) | |
| while 0 <= position < len(text): | |
| character = text[position] | |
| if not character.isspace(): | |
| return character | |
| position += int(step) | |
| return None | |
| def _semantic_neighbor(text: str, index: int, step: int) -> str | None: | |
| """Return a simplified alphanumeric/CJK context anchor.""" | |
| position = int(index) | |
| while 0 <= position < len(text): | |
| character = text[position] | |
| if character.isalnum() or _is_cjk(character): | |
| return _T2S_CONVERTER.convert(character).casefold() | |
| position += int(step) | |
| return None | |
| def _structured_numeric_neighbor(character: str | None) -> bool: | |
| if character is None: | |
| return False | |
| return bool( | |
| character in _ASR_STRUCTURED_NUMERIC_NEIGHBORS | |
| or character.isascii() and character.isalnum() | |
| ) | |
| def _target_numeric_proofs( | |
| normalized_target: str, | |
| reading: str, | |
| ) -> list[tuple[int, int]]: | |
| """Find unstructured, exact target occurrences of one derived reading.""" | |
| proofs: list[tuple[int, int]] = [] | |
| start = 0 | |
| while True: | |
| index = normalized_target.find(reading, start) | |
| if index < 0: | |
| return proofs | |
| end = index + len(reading) | |
| left = _nonspace_neighbor(normalized_target, index - 1, -1) | |
| right = _nonspace_neighbor(normalized_target, end, 1) | |
| if not ( | |
| _structured_numeric_neighbor(left) | |
| or _structured_numeric_neighbor(right) | |
| ): | |
| proofs.append((index, end)) | |
| start = index + 1 | |
| def _canonicalize_target_proven_arabic_digits( | |
| normalized_text: str, | |
| normalized_target: str, | |
| ) -> str: | |
| """Replace a bare ASR digit run only with its exact target-proven reading. | |
| Both possible Mandarin readings are derived from the run itself: a | |
| cardinal value (``1250`` -> ``一千二百五十``) and a digit sequence | |
| (``1250`` -> ``一二五零``). A derived reading must occur in the target at | |
| matching semantic left/right anchors. Structured numeric domains are | |
| deliberately excluded because dates, clocks, versions, network values, | |
| and model codes have their own stricter normalizers. | |
| """ | |
| matches = tuple(_ASR_ARABIC_DIGIT_RUN_RE.finditer(normalized_text)) | |
| if not matches: | |
| return normalized_text | |
| used_proofs: set[tuple[int, int, str]] = set() | |
| output: list[str] = [] | |
| cursor = 0 | |
| for match in matches: | |
| output.append(normalized_text[cursor : match.start()]) | |
| run = match.group(0) | |
| left = _nonspace_neighbor(normalized_text, match.start() - 1, -1) | |
| right = _nonspace_neighbor(normalized_text, match.end(), 1) | |
| replacement: str | None = None | |
| if not ( | |
| _structured_numeric_neighbor(left) | |
| or _structured_numeric_neighbor(right) | |
| ): | |
| readings: list[str] = [] | |
| try: | |
| readings.append(_zh_integer(int(run))) | |
| except ValueError: | |
| pass | |
| digit_reading = _zh_digit_sequence(run) | |
| if digit_reading not in readings: | |
| readings.append(digit_reading) | |
| text_context = ( | |
| _semantic_neighbor(normalized_text, match.start() - 1, -1), | |
| _semantic_neighbor(normalized_text, match.end(), 1), | |
| ) | |
| for reading in readings: | |
| for proof_start, proof_end in _target_numeric_proofs( | |
| normalized_target, | |
| reading, | |
| ): | |
| proof_key = (proof_start, proof_end, reading) | |
| if proof_key in used_proofs: | |
| continue | |
| target_context = ( | |
| _semantic_neighbor( | |
| normalized_target, | |
| proof_start - 1, | |
| -1, | |
| ), | |
| _semantic_neighbor(normalized_target, proof_end, 1), | |
| ) | |
| if target_context == text_context: | |
| replacement = reading | |
| used_proofs.add(proof_key) | |
| break | |
| if replacement is not None: | |
| break | |
| output.append(replacement if replacement is not None else run) | |
| cursor = match.end() | |
| output.append(normalized_text[cursor:]) | |
| return "".join(output) | |
| def _canonicalize_target_proven_structured_asr_forms( | |
| text: str, | |
| normalized_target: str, | |
| ) -> str: | |
| """Canonicalize Breeze-style clock/model renderings under exact proof. | |
| Breeze ASR 25 commonly writes a spoken clock as ``7.45 分`` and emits | |
| model identifiers in lowercase without their separator (``ax418``). | |
| Those are legitimate written renderings, but globally interpreting every | |
| decimal or lowercase alphanumeric run would hide real content errors. | |
| Each replacement is therefore derived from the candidate text, required | |
| verbatim in the already-normalized target, matched at the same semantic | |
| left/right anchors, and consumed at most once. | |
| """ | |
| used_proofs: set[tuple[int, int, str]] = set() | |
| def replace_matches( | |
| value: str, | |
| pattern: re.Pattern[str], | |
| canonicalizer: Callable[[re.Match[str]], str | None], | |
| ) -> str: | |
| matches = tuple(pattern.finditer(value)) | |
| if not matches: | |
| return value | |
| output: list[str] = [] | |
| cursor = 0 | |
| for match in matches: | |
| output.append(value[cursor : match.start()]) | |
| canonical = canonicalizer(match) | |
| replacement: str | None = None | |
| if canonical: | |
| text_context = ( | |
| _semantic_neighbor(value, match.start() - 1, -1), | |
| _semantic_neighbor(value, match.end(), 1), | |
| ) | |
| proof_start = 0 | |
| while True: | |
| proof_start = normalized_target.find(canonical, proof_start) | |
| if proof_start < 0: | |
| break | |
| proof_end = proof_start + len(canonical) | |
| proof_key = (proof_start, proof_end, canonical) | |
| target_context = ( | |
| _semantic_neighbor( | |
| normalized_target, | |
| proof_start - 1, | |
| -1, | |
| ), | |
| _semantic_neighbor(normalized_target, proof_end, 1), | |
| ) | |
| if proof_key not in used_proofs and target_context == text_context: | |
| replacement = canonical | |
| used_proofs.add(proof_key) | |
| break | |
| proof_start += 1 | |
| output.append(replacement if replacement is not None else match.group(0)) | |
| cursor = match.end() | |
| output.append(value[cursor:]) | |
| return "".join(output) | |
| def canonical_clock(match: re.Match[str]) -> str | None: | |
| try: | |
| return _zh_clock_time(match.group(1), match.group(2)) | |
| except (TypeError, ValueError): | |
| return None | |
| def canonical_model_code(match: re.Match[str]) -> str | None: | |
| letters, digits = match.groups() | |
| try: | |
| return f"{' '.join(letters.upper())} {_zh_digit_sequence(digits)}" | |
| except ValueError: | |
| return None | |
| canonicalized = replace_matches( | |
| text, | |
| _ASR_FORMATTED_CLOCK_RE, | |
| canonical_clock, | |
| ) | |
| return replace_matches( | |
| canonicalized, | |
| _ASR_MODEL_CODE_RE, | |
| canonical_model_code, | |
| ) | |
| def _canonical_jpy_amount_reading(amount: str) -> str | None: | |
| """Return one validated Mandarin amount reading for a JPY proof span.""" | |
| compact = re.sub(r"\s+", "", str(amount or "")) | |
| if re.fullmatch(_CURRENCY_NUMBER_PATTERN, compact): | |
| try: | |
| return _zh_number(compact.replace(",", "")) | |
| except ValueError: | |
| return None | |
| if re.fullmatch(_ASR_JPY_ZH_AMOUNT_PATTERN, compact): | |
| return compact.translate(_ASR_JPY_AMOUNT_TRANSLATION) | |
| return None | |
| def _canonicalize_target_proven_jpy_amounts( | |
| normalized_text: str, | |
| normalized_target: str, | |
| ) -> str: | |
| """Canonicalize a bounded JPY alias and amount only under an exact proof. | |
| Whisper commonly renders spoken ``日圓四千五百`` as ``日元4500`` or | |
| ``日元4500元``. A global ``圓``/``元`` homophone or trailing-unit rule | |
| would hide unrelated lexical errors, while the generic bare-number | |
| normalizer correctly refuses to borrow a proof across unequal currency | |
| anchors. This structured-domain path therefore requires a target span | |
| without the redundant unit, the same exact amount, and the same semantic | |
| characters immediately outside the complete candidate span. Every | |
| target range is consumable once and ambiguous matches stay unchanged. | |
| """ | |
| target_proofs: list[ | |
| tuple[int, int, str, tuple[str | None, str | None], str] | |
| ] = [] | |
| for match in _ASR_JPY_CURRENCY_AMOUNT_RE.finditer(normalized_target): | |
| if match.group("redundant_unit") is not None: | |
| continue | |
| reading = _canonical_jpy_amount_reading(match.group("amount")) | |
| if reading is None: | |
| continue | |
| target_proofs.append( | |
| ( | |
| match.start(), | |
| match.end(), | |
| reading, | |
| ( | |
| _semantic_neighbor( | |
| normalized_target, | |
| match.start() - 1, | |
| -1, | |
| ), | |
| _semantic_neighbor(normalized_target, match.end(), 1), | |
| ), | |
| match.group(0), | |
| ) | |
| ) | |
| if not target_proofs: | |
| return normalized_text | |
| used_proofs: set[int] = set() | |
| output: list[str] = [] | |
| cursor = 0 | |
| for match in _ASR_JPY_CURRENCY_AMOUNT_RE.finditer(normalized_text): | |
| output.append(normalized_text[cursor : match.start()]) | |
| reading = _canonical_jpy_amount_reading(match.group("amount")) | |
| text_context = ( | |
| _semantic_neighbor(normalized_text, match.start() - 1, -1), | |
| _semantic_neighbor(normalized_text, match.end(), 1), | |
| ) | |
| matching = [ | |
| index | |
| for index, (_, _, proof_reading, proof_context, _) in enumerate( | |
| target_proofs | |
| ) | |
| if ( | |
| index not in used_proofs | |
| and reading is not None | |
| and proof_reading == reading | |
| and proof_context == text_context | |
| ) | |
| ] | |
| replacement = match.group(0) | |
| if len(matching) == 1: | |
| proof_index = matching[0] | |
| used_proofs.add(proof_index) | |
| replacement = target_proofs[proof_index][4] | |
| output.append(replacement) | |
| cursor = match.end() | |
| output.append(normalized_text[cursor:]) | |
| return "".join(output) | |
| class _FrequencyMeasurement: | |
| """One explicitly unit-bearing frequency measurement.""" | |
| start: int | |
| end: int | |
| raw_text: str | |
| dimension: str | |
| base_unit_value: Decimal | |
| context_key: tuple[str, int, int] | |
| def _frequency_structured_spans(text: str) -> tuple[tuple[int, int], ...]: | |
| """Return numeric domains that must never authorize frequency scaling.""" | |
| patterns = ( | |
| _SPOKEN_URL_RE, | |
| _SPOKEN_EMAIL_RE, | |
| _SPOKEN_DATE_RE, | |
| _SPOKEN_ZH_DATE_RE, | |
| _SPOKEN_TIME_RE, | |
| _SPOKEN_PERCENT_RE, | |
| _SPOKEN_SEMVER_RE, | |
| _SPOKEN_CURRENCY_PREFIX_RE, | |
| _SPOKEN_CURRENCY_SUFFIX_RE, | |
| _MODEL_CODE_RE, | |
| ) | |
| return tuple( | |
| (match.start(), match.end()) | |
| for pattern in patterns | |
| for match in pattern.finditer(text) | |
| ) | |
| def _frequency_model_prefix(text: str, start: int) -> bool: | |
| """Reject a measurement whose number is serving as a model identifier.""" | |
| prefix = text[:start] | |
| clause_prefix = re.split(r"[，。！？；、,!?;\n]", prefix)[-1] | |
| return bool( | |
| re.search( | |
| r"(?:型號|型号|版本|\bmodel\b|\bversion\b)", | |
| clause_prefix, | |
| flags=re.IGNORECASE, | |
| ) | |
| or re.search(r"(?:\b(?:model|version)\s+|[A-Z]{2,8}[ -]*)$", prefix) | |
| or re.search(r"(?:[A-Z](?:\s+[A-Z]){1,7})\s*$", prefix) | |
| or re.search( | |
| r"(?:型號|型号|版本)\s*" | |
| r"(?:(?:[A-Za-z]\s*){2,8})?$", | |
| prefix, | |
| ) | |
| or re.search( | |
| r"\b(?:model|version)\s+" | |
| r"(?:(?:[A-Za-z]\s*){2,8})?$", | |
| prefix, | |
| flags=re.IGNORECASE, | |
| ) | |
| ) | |
| def _parse_canonical_zh_integer(spoken: str) -> int | None: | |
| """Invert only the exact integer language emitted by ``_zh_integer``.""" | |
| if spoken == "零": | |
| return 0 | |
| if not spoken: | |
| return None | |
| digit_values = {character: value for value, character in enumerate(_ZH_DIGITS)} | |
| small_units = {"十": 10, "百": 100, "千": 1000} | |
| large_units = {"萬": 10**4, "億": 10**8, "兆": 10**12} | |
| total = 0 | |
| section = 0 | |
| pending_digit: int | None = None | |
| for character in spoken: | |
| if character in digit_values: | |
| digit = digit_values[character] | |
| if pending_digit not in {None, 0}: | |
| return None | |
| pending_digit = digit | |
| continue | |
| if character in small_units: | |
| multiplier = 1 if pending_digit is None else pending_digit | |
| section += multiplier * small_units[character] | |
| pending_digit = None | |
| continue | |
| if character in large_units: | |
| if pending_digit is not None: | |
| section += pending_digit | |
| if section <= 0: | |
| return None | |
| total += section * large_units[character] | |
| section = 0 | |
| pending_digit = None | |
| continue | |
| return None | |
| if pending_digit is not None: | |
| section += pending_digit | |
| value = total + section | |
| if value >= 10**16 or _zh_integer(value) != spoken: | |
| return None | |
| return value | |
| def _parse_canonical_zh_number(spoken: str) -> Decimal | None: | |
| """Invert a non-negative canonical number and verify it by round trip.""" | |
| explicit_positive = spoken.startswith("正") | |
| unsigned = spoken[1:] if explicit_positive else spoken | |
| if unsigned.count("點") > 1: | |
| return None | |
| integer_spoken, separator, fraction_spoken = unsigned.partition("點") | |
| integer = _parse_canonical_zh_integer(integer_spoken) | |
| if integer is None or separator and not fraction_spoken: | |
| return None | |
| digit_values = {character: value for value, character in enumerate(_ZH_DIGITS)} | |
| if any(character not in digit_values for character in fraction_spoken): | |
| return None | |
| fraction = "".join(str(digit_values[character]) for character in fraction_spoken) | |
| number_text = ("+" if explicit_positive else "") + str(integer) | |
| if separator: | |
| number_text += "." + fraction | |
| try: | |
| if _zh_number(number_text) != spoken: | |
| return None | |
| value = Decimal(number_text) | |
| except (InvalidOperation, ValueError): | |
| return None | |
| return value if value.is_finite() and value >= 0 else None | |
| def _frequency_context_skeleton( | |
| text: str, | |
| measurement_spans: Sequence[tuple[int, int]], | |
| ) -> str: | |
| """Return a punctuation-insensitive, spoken-form sentence skeleton. | |
| Frequency spellings must be hidden before normalizing the rest of the | |
| sentence. Otherwise the very scale difference being proved (for example | |
| ``48 kHz`` versus ``48000Hz``) leaks into the context key. Normalizing the | |
| remaining text also lets equivalent raw/ASR unit readings such as | |
| ``10 km/h`` and ``10公里每小時`` provide the same conservative | |
| full-sentence context. | |
| """ | |
| output: list[str] = [] | |
| cursor = 0 | |
| for start, end in measurement_spans: | |
| output.append(text[cursor:start]) | |
| output.append("¤") | |
| cursor = end | |
| output.append(text[cursor:]) | |
| normalized = normalize_spoken_forms("".join(output)) | |
| simplified = _T2S_CONVERTER.convert(normalized).casefold() | |
| return "".join( | |
| character | |
| for character in simplified | |
| if character == "¤" or character.isalnum() or _is_cjk(character) | |
| ) | |
| def _extract_frequency_measurements( | |
| text: str, | |
| ) -> tuple[ | |
| tuple[_FrequencyMeasurement, ...], | |
| bool, | |
| dict[tuple[str, int, int], frozenset[Decimal]], | |
| ]: | |
| """Extract measurements, ambiguity, and exact base-Hz evidence sets.""" | |
| candidates_by_span: dict[ | |
| tuple[int, int], | |
| list[tuple[int, int, str, Decimal, int]], | |
| ] = {} | |
| structured_spans = _frequency_structured_spans(text) | |
| def is_blocked(start: int, end: int) -> bool: | |
| return bool( | |
| any( | |
| start < structured_end and end > structured_start | |
| for structured_start, structured_end in structured_spans | |
| ) | |
| or _frequency_model_prefix(text, start) | |
| ) | |
| def add_candidate( | |
| start: int, | |
| end: int, | |
| raw_text: str, | |
| number: Decimal, | |
| scale: Decimal, | |
| priority: int, | |
| ) -> None: | |
| if is_blocked(start, end): | |
| return | |
| with localcontext() as context: | |
| # Decimal arithmetic still observes a context precision. Size it | |
| # from the exact coefficient so multiplying by a power-of-ten | |
| # frequency scale cannot round a long fractional measurement. | |
| context.prec = max(64, len(number.as_tuple().digits) + 16) | |
| base_value = number * scale | |
| span = (start, end) | |
| candidates_by_span.setdefault(span, []).append( | |
| ( | |
| start, | |
| end, | |
| raw_text, | |
| base_value, | |
| priority, | |
| ) | |
| ) | |
| for match in _ASR_FREQUENCY_MEASUREMENT_RE.finditer(text): | |
| number_text = match.group(1) | |
| try: | |
| # Keep this proof path inside the same numeric domain supported by | |
| # the unchanged frontend normalizer. | |
| _zh_number(number_text) | |
| number = Decimal(number_text) | |
| except (InvalidOperation, ValueError): | |
| continue | |
| scale = _FREQUENCY_SCALE_HZ.get(match.group(2).casefold()) | |
| if scale is None or not number.is_finite() or number < 0: | |
| continue | |
| add_candidate( | |
| match.start(), | |
| match.end(), | |
| match.group(0), | |
| number, | |
| scale, | |
| -1, | |
| ) | |
| for priority, (pattern, unit) in enumerate(_CANONICAL_ZH_FREQUENCY_PATTERNS): | |
| scale = _FREQUENCY_SCALE_HZ[unit] | |
| for match in pattern.finditer(text): | |
| number = _parse_canonical_zh_number(match.group(1)) | |
| if number is None: | |
| continue | |
| add_candidate( | |
| match.start(), | |
| match.end(), | |
| match.group(0), | |
| number, | |
| scale, | |
| priority, | |
| ) | |
| if not candidates_by_span: | |
| return (), False, {} | |
| ordered_spans = sorted(candidates_by_span) | |
| measurement_spans = tuple(ordered_spans) | |
| skeleton = _frequency_context_skeleton(text, measurement_spans) | |
| measurement_count = len(ordered_spans) | |
| raw_matches: list[tuple[int, int, str, Decimal, int]] = [] | |
| evidence_by_context: dict[ | |
| tuple[str, int, int], | |
| frozenset[Decimal], | |
| ] = {} | |
| ambiguous = False | |
| context_by_span: dict[tuple[int, int], tuple[str, int, int]] = {} | |
| for ordinal, span in enumerate(ordered_spans): | |
| candidates = candidates_by_span[span] | |
| base_values = frozenset(candidate[3] for candidate in candidates) | |
| context_key = (skeleton, ordinal, measurement_count) | |
| context_by_span[span] = context_key | |
| evidence_by_context[context_key] = base_values | |
| if len(base_values) != 1: | |
| ambiguous = True | |
| continue | |
| raw_matches.append(min(candidates, key=lambda candidate: candidate[4])) | |
| if not raw_matches: | |
| return (), ambiguous, evidence_by_context | |
| raw_matches.sort(key=lambda item: item[0]) | |
| measurements: list[_FrequencyMeasurement] = [] | |
| for start, end, raw_text, base_value, _ in raw_matches: | |
| measurements.append( | |
| _FrequencyMeasurement( | |
| start=start, | |
| end=end, | |
| raw_text=raw_text, | |
| dimension="frequency", | |
| base_unit_value=base_value, | |
| context_key=context_by_span[(start, end)], | |
| ) | |
| ) | |
| return tuple(measurements), ambiguous, evidence_by_context | |
| def _raw_frequency_measurements(text: str) -> tuple[_FrequencyMeasurement, ...]: | |
| """Extract only unambiguous, context-bound frequency measurements.""" | |
| measurements, _, _ = _extract_frequency_measurements(text) | |
| return measurements | |
| def _has_ambiguous_canonical_frequency(text: str) -> bool: | |
| """Return whether frontend-normalized text lost frequency provenance.""" | |
| _, ambiguous, _ = _extract_frequency_measurements(text) | |
| return ambiguous | |
| def _frequency_measurements_contradict(text: str, target_text: str) -> bool: | |
| """Reject explicit frequency values whose exact base-Hz sets disagree.""" | |
| def reject_only_evidence( | |
| value: str, | |
| ) -> dict[tuple[str, int, int], frozenset[Decimal]]: | |
| raw_matches: list[tuple[int, int, Decimal]] = [] | |
| for match in _REJECT_ONLY_FREQUENCY_MEASUREMENT_RE.finditer(value): | |
| try: | |
| number = Decimal(match.group(1)) | |
| except InvalidOperation: | |
| continue | |
| scale = _FREQUENCY_SCALE_HZ.get(match.group(2).casefold()) | |
| if scale is None or not number.is_finite() or number < 0: | |
| continue | |
| with localcontext() as context: | |
| context.prec = max(64, len(number.as_tuple().digits) + 16) | |
| base_value = number * scale | |
| raw_matches.append((match.start(), match.end(), base_value)) | |
| if not raw_matches: | |
| return {} | |
| spans = tuple((start, end) for start, end, _ in raw_matches) | |
| skeleton = _frequency_context_skeleton(value, spans) | |
| count = len(raw_matches) | |
| return { | |
| (skeleton, ordinal, count): frozenset((base_value,)) | |
| for ordinal, (_, _, base_value) in enumerate(raw_matches) | |
| } | |
| target_reject_evidence = reject_only_evidence(target_text) | |
| text_reject_evidence = reject_only_evidence(text) | |
| for context_key, target_values in target_reject_evidence.items(): | |
| text_values = text_reject_evidence.get(context_key) | |
| if text_values is not None and target_values.isdisjoint(text_values): | |
| return True | |
| _, _, target_evidence = _extract_frequency_measurements(target_text) | |
| _, _, text_evidence = _extract_frequency_measurements(text) | |
| for context_key, target_values in target_evidence.items(): | |
| text_values = text_evidence.get(context_key) | |
| if text_values is not None and target_values.isdisjoint(text_values): | |
| return True | |
| return False | |
| def _canonicalize_target_proven_frequency_scales( | |
| text: str, | |
| target_text: str, | |
| ) -> str: | |
| """Rewrite ASR frequency scale only under an exact raw target proof.""" | |
| target_measurements, target_ambiguous, _ = _extract_frequency_measurements( | |
| target_text | |
| ) | |
| text_measurements, _, _ = _extract_frequency_measurements(text) | |
| if target_ambiguous or not target_measurements or not text_measurements: | |
| return text | |
| target_by_context = { | |
| measurement.context_key: measurement | |
| for measurement in target_measurements | |
| } | |
| output: list[str] = [] | |
| cursor = 0 | |
| for measurement in text_measurements: | |
| output.append(text[cursor : measurement.start]) | |
| proof = target_by_context.get(measurement.context_key) | |
| replacement = measurement.raw_text | |
| if ( | |
| proof is not None | |
| and proof.dimension == measurement.dimension | |
| and proof.base_unit_value == measurement.base_unit_value | |
| ): | |
| replacement = proof.raw_text | |
| output.append(replacement) | |
| cursor = measurement.end | |
| output.append(text[cursor:]) | |
| return "".join(output) | |
| def _zh_network_text(value: str) -> str: | |
| """Read lexical labels naturally while keeping typed atoms explicit.""" | |
| output: list[str] = [] | |
| buffer: list[str] = [] | |
| def flush() -> None: | |
| if not buffer: | |
| return | |
| token = "".join(buffer) | |
| if token.isdigit(): | |
| output.append(_zh_digit_sequence(token)) | |
| elif token.isascii() and token.isalpha(): | |
| if token.casefold() == "www" or (len(token) > 1 and token.isupper()): | |
| output.append(_zh_network_letters(token)) | |
| else: | |
| output.append(token) | |
| else: | |
| output.append(token) | |
| buffer.clear() | |
| for character in value: | |
| if character.isascii() and character.isalnum(): | |
| if buffer and character.isdigit() != buffer[-1].isdigit(): | |
| flush() | |
| buffer.append(character) | |
| continue | |
| flush() | |
| reading = _ZH_NETWORK_SYMBOL_READINGS.get(character) | |
| if reading: | |
| output.append(reading) | |
| elif not character.isspace() and not character.isalnum(): | |
| # Never let an accepted-but-unmapped ASCII symbol disappear when | |
| # comparison normalization removes punctuation. The explicit | |
| # Unicode codepoint reading is conservative and keeps exact | |
| # omission detection fail-closed for unusual URI/IRI characters. | |
| output.append(f"符號{_zh_digit_sequence(str(ord(character)))}") | |
| elif not character.isspace(): | |
| output.append(character) | |
| flush() | |
| return " ".join(output) | |
| def _zh_network_letters(value: str) -> str: | |
| """Return model-native spaced uppercase letters for typed ASCII atoms.""" | |
| return " ".join( | |
| character.upper() | |
| for character in value | |
| if not character.isspace() | |
| ) | |
| def _zh_version_qualifier(value: str) -> str: | |
| """Spell opaque prerelease/build labels without lexical network rules.""" | |
| output: list[str] = [] | |
| for character in value: | |
| if character.isascii() and character.isalpha(): | |
| output.append(character.upper()) | |
| elif character.isascii() and character.isdigit(): | |
| output.append(_ZH_DIGITS[int(character)]) | |
| else: | |
| reading = _ZH_NETWORK_SYMBOL_READINGS.get(character) | |
| if reading: | |
| output.append(reading) | |
| elif not character.isspace(): | |
| output.append(character) | |
| return " ".join(output) | |
| def _zh_network_domain(value: str) -> str: | |
| """Render an explicit URL-fallback domain without lexical guessing.""" | |
| head, separator, suffix = value.rpartition(".") | |
| if separator and head and suffix.isascii() and suffix.isalpha(): | |
| return f"{_zh_network_text(head)} 點 {_zh_network_letters(suffix)}" | |
| return _zh_network_text(value) | |
| def _zh_email_domain(value: str) -> str: | |
| """Render uniquely segmented email labels and explicit separators. | |
| Compound labels such as ``forestmail`` are substantially clearer to the | |
| acoustic model as ``forest mail``. Segmentation is allowed only when the | |
| fixed lexical dictionary proves one unique full-label decomposition; all | |
| opaque, ambiguous and typed labels retain the existing literal reading. | |
| The email verifier reconstructs adjacent lexical tokens before applying | |
| the exact raw-address proof, so this changes pronunciation without | |
| weakening identifier equality. | |
| """ | |
| head, separator, suffix = value.rpartition(".") | |
| if separator and head and suffix.isascii() and suffix.isalpha(): | |
| spoken_labels: list[str] = [] | |
| for label in head.split("."): | |
| segments = _unique_email_domain_label_segments(label) | |
| spoken_labels.append( | |
| " ".join(segments) | |
| if len(segments) > 1 | |
| else _zh_network_text(label) | |
| ) | |
| return ( | |
| f"{' 點 '.join(spoken_labels)} 點 " | |
| f"{_zh_network_letters(suffix)}" | |
| ) | |
| return _zh_network_text(value) | |
| class NetworkUrlRenderAtomProof: | |
| """One raw URL slice and its exact emitted spoken range.""" | |
| component: str | |
| rule: str | |
| raw_start: int | |
| raw_end: int | |
| spoken_start: int | |
| spoken_end: int | |
| raw_text: str | |
| spoken_text: str | |
| class NetworkUrlRenderingProof: | |
| """Complete reversible proof for one safely naturalized HTTP(S) URL.""" | |
| raw_identifier: str | |
| spoken_text: str | |
| atoms: tuple[NetworkUrlRenderAtomProof, ...] | |
| _NATURALIZED_URL_LEXICAL_WORDS = frozenset( | |
| { | |
| "audio", | |
| "coast", | |
| "data", | |
| "forest", | |
| "help", | |
| "island", | |
| "mail", | |
| "museum", | |
| "news", | |
| "radio", | |
| "search", | |
| "sound", | |
| "tour", | |
| "trail", | |
| "voice", | |
| "watch", | |
| "weather", | |
| } | |
| ) | |
| _EMAIL_DOMAIN_LEXICAL_WORDS = ( | |
| _NATURALIZED_URL_LEXICAL_WORDS | |
| | frozenset({"arts", "harbor"}) | |
| ) | |
| _NATURALIZED_URL_ATOM_COMPONENTS = frozenset( | |
| {"scheme", "host_label", "host_dot", "path_slash", "path_atom", "terminator"} | |
| ) | |
| _NATURALIZED_URL_ATOM_RULES = frozenset( | |
| { | |
| "https_scheme_cue", | |
| "http_scheme_cue", | |
| "lexical_host_label", | |
| "segmented_host_label", | |
| "host_dot", | |
| "spelled_host_suffix", | |
| "path_cue", | |
| "spelled_short_path", | |
| "url_terminator", | |
| } | |
| ) | |
| _NATURALIZED_URL_MAX_SPOKEN_UNITS = 36 | |
| def _unique_lexical_label_segments( | |
| label: str, | |
| lexical_words: frozenset[str], | |
| ) -> tuple[str, ...]: | |
| """Return one lexicon-proven split, otherwise retain the whole label.""" | |
| if ( | |
| not label.isascii() | |
| or not label.isalpha() | |
| or not label.islower() | |
| or len(label) < 6 | |
| ): | |
| return (label,) | |
| solutions: list[tuple[str, ...]] = [] | |
| def search(offset: int, path: tuple[str, ...]) -> None: | |
| if len(solutions) > 1: | |
| return | |
| if offset == len(label): | |
| if len(path) >= 2: | |
| solutions.append(path) | |
| return | |
| for end in range(offset + 3, len(label) + 1): | |
| word = label[offset:end] | |
| if word in lexical_words: | |
| search(end, (*path, word)) | |
| search(0, ()) | |
| return solutions[0] if len(solutions) == 1 else (label,) | |
| def _unique_naturalized_url_label_segments(label: str) -> tuple[str, ...]: | |
| """Return one URL-dictionary-proven split or retain the whole label.""" | |
| return _unique_lexical_label_segments( | |
| label, | |
| _NATURALIZED_URL_LEXICAL_WORDS, | |
| ) | |
| def _unique_email_domain_label_segments(label: str) -> tuple[str, ...]: | |
| """Return one email-dictionary-proven split or retain the whole label.""" | |
| return _unique_lexical_label_segments( | |
| label, | |
| _EMAIL_DOMAIN_LEXICAL_WORDS, | |
| ) | |
| def _build_naturalized_url_rendering_proof( | |
| value: str, | |
| ) -> NetworkUrlRenderingProof | None: | |
| """Build the one canonical rendering for the safe short-path grammar. | |
| Unsupported ports, credentials, queries, fragments, non-DNS hosts, long | |
| paths or non-canonical casing fall back to the existing fully explicit URL | |
| spelling. This unchecked builder is also the independent source of truth | |
| used by :func:`inverse_network_url_rendering`. | |
| """ | |
| raw = unicodedata.normalize("NFC", str(value or "")) | |
| if not raw or not raw.isascii(): | |
| return None | |
| try: | |
| parsed = urlsplit(raw) | |
| port = parsed.port | |
| except ValueError: | |
| return None | |
| del port | |
| scheme = parsed.scheme | |
| host = parsed.hostname or "" | |
| path = parsed.path | |
| if ( | |
| scheme not in {"http", "https"} | |
| or parsed.username is not None | |
| or parsed.password is not None | |
| or parsed.netloc != host | |
| or parsed.query | |
| or parsed.fragment | |
| or not re.fullmatch(r"[a-z]+(?:\.[a-z]+)+", host) | |
| # URL paths are case-sensitive. Accept one canonical casing only so | |
| # acoustically identical upper/lower paths can never share a proof. | |
| or not re.fullmatch(r"(?:/[a-z]{1,4})?", path) | |
| or raw != f"{scheme}://{host}{path}" | |
| ): | |
| return None | |
| pieces: list[str] = [] | |
| atoms: list[NetworkUrlRenderAtomProof] = [] | |
| spoken_cursor = 0 | |
| def emit( | |
| *, | |
| component: str, | |
| rule: str, | |
| raw_start: int, | |
| raw_end: int, | |
| spoken: str, | |
| ) -> None: | |
| nonlocal spoken_cursor | |
| pieces.append(spoken) | |
| spoken_end = spoken_cursor + len(spoken) | |
| atoms.append( | |
| NetworkUrlRenderAtomProof( | |
| component=component, | |
| rule=rule, | |
| raw_start=raw_start, | |
| raw_end=raw_end, | |
| spoken_start=spoken_cursor, | |
| spoken_end=spoken_end, | |
| raw_text=raw[raw_start:raw_end], | |
| spoken_text=spoken, | |
| ) | |
| ) | |
| spoken_cursor = spoken_end | |
| scheme_end = len(scheme) + 3 | |
| emit( | |
| component="scheme", | |
| rule=( | |
| "https_scheme_cue" | |
| if scheme == "https" | |
| else "http_scheme_cue" | |
| ), | |
| raw_start=0, | |
| raw_end=scheme_end, | |
| spoken=("安全網址是 " if scheme == "https" else "網址是 "), | |
| ) | |
| host_start = scheme_end | |
| raw_cursor = host_start | |
| labels = host.split(".") | |
| segmented_label_seen = False | |
| for label_index, label in enumerate(labels): | |
| if label_index: | |
| emit( | |
| component="host_dot", | |
| rule="host_dot", | |
| raw_start=raw_cursor, | |
| raw_end=raw_cursor + 1, | |
| spoken=" 點 ", | |
| ) | |
| raw_cursor += 1 | |
| if label_index + 1 == len(labels) and len(label) <= 3: | |
| for character_index, character in enumerate(label): | |
| emit( | |
| component="host_label", | |
| rule="spelled_host_suffix", | |
| raw_start=raw_cursor + character_index, | |
| raw_end=raw_cursor + character_index + 1, | |
| spoken=( | |
| character.upper() | |
| if character_index == 0 | |
| else f" {character.upper()}" | |
| ), | |
| ) | |
| else: | |
| segments = _unique_naturalized_url_label_segments(label) | |
| segmented_label_seen = segmented_label_seen or len(segments) > 1 | |
| segment_cursor = raw_cursor | |
| for segment_index, segment in enumerate(segments): | |
| emit( | |
| component="host_label", | |
| rule=( | |
| "segmented_host_label" | |
| if len(segments) > 1 | |
| else "lexical_host_label" | |
| ), | |
| raw_start=segment_cursor, | |
| raw_end=segment_cursor + len(segment), | |
| spoken=( | |
| segment | |
| if segment_index == 0 | |
| else f" {segment}" | |
| ), | |
| ) | |
| segment_cursor += len(segment) | |
| raw_cursor += len(label) | |
| if path: | |
| slash_start = host_start + len(host) | |
| emit( | |
| component="path_slash", | |
| rule="path_cue", | |
| raw_start=slash_start, | |
| raw_end=slash_start + 1, | |
| spoken="，路徑是 ", | |
| ) | |
| for character_index, character in enumerate(path[1:]): | |
| emit( | |
| component="path_atom", | |
| rule="spelled_short_path", | |
| raw_start=slash_start + 1 + character_index, | |
| raw_end=slash_start + 2 + character_index, | |
| spoken=( | |
| character.upper() | |
| if character_index == 0 | |
| else f" {character.upper()}" | |
| ), | |
| ) | |
| emit( | |
| component="terminator", | |
| rule="url_terminator", | |
| raw_start=len(raw), | |
| raw_end=len(raw), | |
| spoken="，網址輸入完畢", | |
| ) | |
| proof = NetworkUrlRenderingProof( | |
| raw_identifier=raw, | |
| spoken_text="".join(pieces), | |
| atoms=tuple(atoms), | |
| ) | |
| if not segmented_label_seen: | |
| return None | |
| # The natural phrase is intentionally indivisible. Long generic URLs keep | |
| # the legacy explicit spelling so the existing component planner can split | |
| # them rather than turning a valid request into a hard planning failure. | |
| if count_speech_units(proof.spoken_text) > _NATURALIZED_URL_MAX_SPOKEN_UNITS: | |
| return None | |
| return proof | |
| def _naturalized_url_rendering_proof( | |
| value: str, | |
| ) -> NetworkUrlRenderingProof | None: | |
| """Render and independently validate the safe short-path URL subset.""" | |
| proof = _build_naturalized_url_rendering_proof(value) | |
| if proof is not None and inverse_network_url_rendering(proof) != proof.raw_identifier: | |
| raise ValueError("naturalized URL rendering is not reversible") | |
| return proof | |
| def inverse_network_url_rendering(proof: NetworkUrlRenderingProof) -> str: | |
| """Validate every atom range and reconstruct the exact raw identifier.""" | |
| if not isinstance(proof, NetworkUrlRenderingProof): | |
| raise ValueError("URL rendering proof has an invalid type") | |
| raw = proof.raw_identifier | |
| spoken = proof.spoken_text | |
| if ( | |
| not raw | |
| or not raw.isascii() | |
| or not spoken | |
| or not isinstance(proof.atoms, tuple) | |
| or not proof.atoms | |
| ): | |
| raise ValueError("URL rendering proof is incomplete") | |
| raw_cursor = 0 | |
| spoken_cursor = 0 | |
| reconstructed: list[str] = [] | |
| terminator_seen = False | |
| for atom_index, atom in enumerate(proof.atoms): | |
| if ( | |
| not isinstance(atom, NetworkUrlRenderAtomProof) | |
| or atom.component not in _NATURALIZED_URL_ATOM_COMPONENTS | |
| or atom.rule not in _NATURALIZED_URL_ATOM_RULES | |
| or type(atom.raw_start) is not int | |
| or type(atom.raw_end) is not int | |
| or type(atom.spoken_start) is not int | |
| or type(atom.spoken_end) is not int | |
| or not 0 <= atom.raw_start <= atom.raw_end <= len(raw) | |
| or not 0 <= atom.spoken_start < atom.spoken_end <= len(spoken) | |
| or atom.spoken_start != spoken_cursor | |
| or raw[atom.raw_start : atom.raw_end] != atom.raw_text | |
| or spoken[atom.spoken_start : atom.spoken_end] != atom.spoken_text | |
| ): | |
| raise ValueError("URL rendering atom proof is inconsistent") | |
| if atom.component == "terminator": | |
| if ( | |
| terminator_seen | |
| or atom_index + 1 != len(proof.atoms) | |
| or atom.raw_start != len(raw) | |
| or atom.raw_end != len(raw) | |
| or atom.spoken_text != "，網址輸入完畢" | |
| ): | |
| raise ValueError("URL terminator proof is inconsistent") | |
| terminator_seen = True | |
| else: | |
| if terminator_seen or atom.raw_start != raw_cursor: | |
| raise ValueError("URL rendering raw ranges are not contiguous") | |
| reconstructed.append(atom.raw_text) | |
| raw_cursor = atom.raw_end | |
| spoken_cursor = atom.spoken_end | |
| if ( | |
| not terminator_seen | |
| or raw_cursor != len(raw) | |
| or spoken_cursor != len(spoken) | |
| or "".join(reconstructed) != raw | |
| ): | |
| raise ValueError("URL rendering proof does not cover its identifier") | |
| scheme = raw.partition("://")[0] | |
| first = proof.atoms[0] | |
| expected_scheme_rule = ( | |
| "https_scheme_cue" | |
| if scheme == "https" | |
| else "http_scheme_cue" | |
| ) | |
| expected_scheme_spoken = ( | |
| "安全網址是 " if scheme == "https" else "網址是 " | |
| ) | |
| if ( | |
| scheme not in {"http", "https"} | |
| or first.component != "scheme" | |
| or first.rule != expected_scheme_rule | |
| or first.spoken_text != expected_scheme_spoken | |
| ): | |
| raise ValueError("URL scheme cue does not match its raw proof") | |
| expected = _build_naturalized_url_rendering_proof(raw) | |
| if expected is None or proof != expected: | |
| raise ValueError("URL rendering proof is not the canonical raw mapping") | |
| return raw | |
| def naturalized_url_rendering_proof( | |
| value: str, | |
| ) -> NetworkUrlRenderingProof | None: | |
| """Return the validated reversible proof for the safe URL subset.""" | |
| return _naturalized_url_rendering_proof(value) | |
| _NATURALIZED_URL_SPOKEN_LABEL_PATTERN = ( | |
| r"(?:[a-z]+(?: [a-z]+)*|[A-Z](?: [A-Z]){0,2})" | |
| ) | |
| _NATURALIZED_URL_SPOKEN_RE = re.compile( | |
| rf"(?<!\w)(?P<cue>安全網址是|網址是) " | |
| rf"(?P<host>{_NATURALIZED_URL_SPOKEN_LABEL_PATTERN}" | |
| rf"(?: 點 {_NATURALIZED_URL_SPOKEN_LABEL_PATTERN})+)" | |
| r"(?:，路徑是 (?P<path>[A-Z](?: [A-Z]){0,3}))?" | |
| r"，網址輸入完畢(?!\w)" | |
| ) | |
| def _naturalized_url_proof_from_spoken( | |
| spoken: str, | |
| ) -> NetworkUrlRenderingProof | None: | |
| """Recover a proof only when ``spoken`` is one canonical renderer image.""" | |
| normalized = normalize_tts_text(spoken) | |
| match = _NATURALIZED_URL_SPOKEN_RE.fullmatch(normalized) | |
| if match is None: | |
| return None | |
| scheme = "https" if match.group("cue") == "安全網址是" else "http" | |
| labels: list[str] = [] | |
| for label in match.group("host").split(" 點 "): | |
| tokens = label.split() | |
| if tokens and all( | |
| token.isascii() and token.isalpha() and token.islower() | |
| for token in tokens | |
| ): | |
| labels.append("".join(tokens)) | |
| continue | |
| if ( | |
| tokens | |
| and len(tokens) <= 3 | |
| and all( | |
| len(token) == 1 | |
| and token.isascii() | |
| and token.isalpha() | |
| and token.isupper() | |
| for token in tokens | |
| ) | |
| ): | |
| labels.append("".join(tokens).lower()) | |
| continue | |
| return None | |
| path = match.group("path") | |
| path_tokens = path.split() if path else () | |
| raw = f"{scheme}://{'.'.join(labels)}" | |
| if path_tokens: | |
| raw += f"/{''.join(path_tokens).lower()}" | |
| proof = _build_naturalized_url_rendering_proof(raw) | |
| return proof if proof is not None and proof.spoken_text == normalized else None | |
| def _naturalized_url_spoken_records( | |
| text: str, | |
| ) -> tuple[tuple[int, int, str], ...]: | |
| """Locate canonical naturalized URLs without accepting free-form cue prose.""" | |
| source = normalize_tts_text(text) | |
| output: list[tuple[int, int, str]] = [] | |
| for match in _NATURALIZED_URL_SPOKEN_RE.finditer(source): | |
| spoken = match.group(0) | |
| if _naturalized_url_proof_from_spoken(spoken) is not None: | |
| output.append((match.start(), match.end(), spoken)) | |
| return tuple(output) | |
| def contains_naturalized_url_spoken_form(text: str) -> bool: | |
| """Return whether text contains one complete canonical naturalized URL.""" | |
| return bool(_naturalized_url_spoken_records(text)) | |
| def _strip_naturalized_url_prosodic_commas(text: str) -> str: | |
| """Drop only renderer-authored, non-lexical URL pause marks.""" | |
| def replace(match: re.Match[str]) -> str: | |
| pieces = [match.group(1)] | |
| if match.group(2): | |
| pieces.append(match.group(2)) | |
| pieces.append(match.group(3)) | |
| return " ".join(pieces) | |
| return re.sub( | |
| r"((?:安全網址是|網址是)\s+.*?)" | |
| r"(?:\s*[，,]?\s*(路徑是\s+.*?))?" | |
| r"\s*[，,]?\s*(網址輸入完畢)", | |
| replace, | |
| str(text or ""), | |
| ) | |
| def _zh_url(value: str) -> str: | |
| naturalized = _naturalized_url_rendering_proof(value) | |
| if naturalized is not None: | |
| return naturalized.spoken_text | |
| scheme, separator, remainder = value.partition("://") | |
| if separator: | |
| protocol = _zh_network_letters(scheme) | |
| prefix = f"{protocol} 冒號 斜線 斜線 " | |
| else: | |
| remainder = value | |
| prefix = "" | |
| boundary = re.search(r"[/?#]", remainder) | |
| split_at = boundary.start() if boundary is not None else len(remainder) | |
| host = remainder[:split_at] | |
| tail = remainder[split_at:] | |
| spoken_host = _zh_network_domain(host) | |
| spoken_tail = f" {_zh_network_text(tail)}" if tail else "" | |
| return f"{prefix}{spoken_host}{spoken_tail}" | |
| def _zh_email(value: str) -> str: | |
| local, domain = value.rsplit("@", 1) | |
| return f"{_zh_network_text(local)} 小老鼠 {_zh_email_domain(domain)}" | |
| def contains_network_identifier(text: str) -> bool: | |
| """Return whether raw input contains an explicit URL or email address.""" | |
| raw = unicodedata.normalize("NFC", str(text or "")) | |
| return bool(_SPOKEN_URL_RE.search(raw) or _SPOKEN_EMAIL_RE.search(raw)) | |
| def network_identifier_has_ambiguous_iri(text: str) -> bool: | |
| """Reject non-ASCII URL identifiers whose speech collides with ASCII.""" | |
| raw = unicodedata.normalize("NFC", str(text or "")) | |
| return any( | |
| not match.group(0).isascii() | |
| for match in _SPOKEN_URL_RE.finditer(raw) | |
| ) | |
| _NETWORK_BOUNDARY_PUNCTUATION = ( | |
| _NETWORK_NONASCII_BOUNDARY_PUNCTUATION + ",.!?;:()[]{}<>\"'" | |
| ) | |
| _NETWORK_ALIGNMENT_BOUNDARY_SENTINEL = "\ue200" | |
| _NETWORK_SPOKEN_SYMBOL_TOKENS = frozenset( | |
| (*_ZH_NETWORK_SYMBOL_READINGS.values(), *_ZH_NETWORK_LETTER_READINGS.values()) | |
| ) | |
| def _raw_network_spoken_spans(text: str) -> tuple[str, ...]: | |
| """Expand explicit raw URL/email spans in source order without overlap.""" | |
| raw = unicodedata.normalize("NFC", str(text or "")) | |
| matches: list[tuple[int, int, str]] = [] | |
| for match in _SPOKEN_URL_RE.finditer(raw): | |
| matched = match.group(0) | |
| if matched: | |
| matches.append((match.start(), match.end(), _zh_url(matched))) | |
| for match in _SPOKEN_EMAIL_RE.finditer(raw): | |
| matches.append((match.start(), match.end(), _zh_email(match.group(0)))) | |
| selected: list[tuple[int, int, str]] = [] | |
| for start, end, spoken in sorted(matches, key=lambda item: (item[0], -item[1])): | |
| if any(start < kept_end and end > kept_start for kept_start, kept_end, _ in selected): | |
| continue | |
| selected.append((start, end, spoken)) | |
| return tuple(spoken for _, _, spoken in sorted(selected)) | |
| def _is_expanded_network_token(token: str) -> bool: | |
| if token.isalnum(): | |
| return True | |
| if token.isascii() and token and all( | |
| character.isalnum() or character in _ZH_NETWORK_SYMBOL_READINGS | |
| for character in token | |
| ): | |
| return True | |
| if token and all(character in _ZH_DIGITS for character in token): | |
| return True | |
| if token.startswith("符號") and token[2:] and all( | |
| character in _ZH_DIGITS for character in token[2:] | |
| ): | |
| return True | |
| return ( | |
| token in _NETWORK_SPOKEN_SYMBOL_TOKENS | |
| or token in _SIMPLIFIED_NETWORK_SYMBOL_READINGS | |
| ) | |
| def _network_letter_token(token: str) -> str | None: | |
| if len(token) == 1 and token.isascii() and token.isalpha(): | |
| return token.casefold() | |
| return _ZH_NETWORK_READING_TO_LETTER.get(token) | |
| _SPOKEN_EMAIL_SYMBOL_TOKENS = { | |
| reading: symbol | |
| for symbol, reading in _ZH_NETWORK_SYMBOL_READINGS.items() | |
| if symbol in ".!#$%&'*+/=?^_`{|}~-@" | |
| } | |
| def _spoken_email_token_value(token: str) -> str | None: | |
| letter = _network_letter_token(token) | |
| if letter is not None: | |
| return letter | |
| if token.isascii() and token.isalnum(): | |
| return token.casefold() | |
| if token and all(character in _ZH_DIGITS for character in token): | |
| return "".join(str(_ZH_DIGITS.index(character)) for character in token) | |
| canonical = _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token) | |
| return _SPOKEN_EMAIL_SYMBOL_TOKENS.get(canonical) | |
| def _spoken_email_subspan_bounds(tokens: Sequence[str]) -> tuple[tuple[int, int], ...]: | |
| """Locate maximal syntactically valid spoken email token ranges.""" | |
| tokens = tuple(_SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token) for token in tokens) | |
| output: list[tuple[int, int]] = [] | |
| for at, token in enumerate(tokens): | |
| if token != "小老鼠": | |
| continue | |
| if any( | |
| tokens[index : index + 3] == ["冒號", "斜線", "斜線"] | |
| for index in range(max(0, at - 2)) | |
| ): | |
| continue | |
| www_end = None | |
| for start in range(at): | |
| if ( | |
| start + 1 < at | |
| and tokens[start].casefold() == "www" | |
| and tokens[start + 1] == "點" | |
| ): | |
| www_end = start + 2 | |
| break | |
| if ( | |
| start + 3 < at | |
| and [ | |
| _network_letter_token(part) | |
| for part in tokens[start : start + 3] | |
| ] | |
| == ["w", "w", "w"] | |
| and tokens[start + 3] == "點" | |
| ): | |
| www_end = start + 4 | |
| break | |
| if www_end is not None and any( | |
| part in {"斜線", "問號", "井號", "冒號"} | |
| for part in tokens[www_end:] | |
| ): | |
| continue | |
| left_floor = 0 | |
| for separator in range(at): | |
| if tokens[separator] == "和" and "小老鼠" in tokens[:separator]: | |
| left_floor = separator + 1 | |
| left = at | |
| while left > left_floor: | |
| value = _spoken_email_token_value(tokens[left - 1]) | |
| if value is None or value == "@": | |
| break | |
| left -= 1 | |
| right = at + 1 | |
| while right < len(tokens): | |
| value = _spoken_email_token_value(tokens[right]) | |
| if value is None or value == "@": | |
| break | |
| right += 1 | |
| matches: list[tuple[int, int, int, int]] = [] | |
| for start in range(left, at): | |
| for end in range(at + 2, right + 1): | |
| rendered = "".join( | |
| value | |
| for part in tokens[start:end] | |
| if (value := _spoken_email_token_value(part)) is not None | |
| ) | |
| if _SPOKEN_EMAIL_RE.fullmatch(rendered): | |
| matches.append((len(rendered), end - start, -start, end)) | |
| if matches: | |
| match = max(matches) | |
| output.append((-match[2], match[3])) | |
| return tuple(output) | |
| def _spoken_email_subspans(span: str) -> tuple[str, ...]: | |
| tokens = span.split() | |
| return tuple( | |
| " ".join(tokens[start:end]) | |
| for start, end in _spoken_email_subspan_bounds(tokens) | |
| ) | |
| _SIMPLIFIED_NETWORK_SYMBOL_READINGS = { | |
| "点": "點", | |
| "冒号": "冒號", | |
| "斜线": "斜線", | |
| "问号": "問號", | |
| "等于": "等於", | |
| "井号": "井號", | |
| "横线": "橫線", | |
| "底线": "底線", | |
| "百分号": "百分號", | |
| "加号": "加號", | |
| "波浪号": "波浪號", | |
| "惊叹号": "驚嘆號", | |
| "钱字号": "錢字號", | |
| "单引号": "單引號", | |
| "左括号": "左括號", | |
| "右括号": "右括號", | |
| "星号": "星號", | |
| "逗号": "逗號", | |
| "分号": "分號", | |
| "左方括号": "左方括號", | |
| "右方括号": "右方括號", | |
| "插入号": "插入號", | |
| "反引号": "反引號", | |
| "左大括号": "左大括號", | |
| "竖线": "豎線", | |
| "右大括号": "右大括號", | |
| } | |
| def _canonicalize_network_spoken_span(span: str) -> str: | |
| """Normalize opaque network tokens without applying prose unit rules.""" | |
| symbol_readings = frozenset(_ZH_NETWORK_SYMBOL_READINGS.values()) | |
| output: list[str] = [] | |
| for token in span.split(): | |
| letter = _network_letter_token(token) | |
| if letter is not None: | |
| output.append(_ZH_NETWORK_LETTER_READINGS[letter.upper()]) | |
| continue | |
| if token in symbol_readings: | |
| output.append(token) | |
| continue | |
| if token in _SIMPLIFIED_NETWORK_SYMBOL_READINGS: | |
| output.append(_SIMPLIFIED_NETWORK_SYMBOL_READINGS[token]) | |
| continue | |
| if token and token.isascii() and all( | |
| character.isalnum() or character in _ZH_NETWORK_SYMBOL_READINGS | |
| for character in token | |
| ): | |
| for character in token: | |
| if character.isdigit(): | |
| output.append(_ZH_DIGITS[int(character)]) | |
| elif character.isalpha(): | |
| output.append(_ZH_NETWORK_LETTER_READINGS[character.upper()]) | |
| else: | |
| output.append(_ZH_NETWORK_SYMBOL_READINGS[character]) | |
| continue | |
| output.append(token) | |
| return " ".join(output) | |
| def _expanded_network_spoken_span_records( | |
| text: str, | |
| ) -> tuple[tuple[int, int, str], ...]: | |
| """Recover protected spans after the request frontend has expanded them. | |
| The Space normalizes the whole request before chunking, so the quality | |
| path receives spoken forms rather than the original URL/email spelling. | |
| A maximal run is considered network data only when it contains an | |
| unambiguous scheme, email, or ``www`` anchor. | |
| """ | |
| source = normalize_tts_text(str(text or "")) | |
| records: list[tuple[str, int, int]] = [] | |
| boundary_pattern = f"([{re.escape(_NETWORK_BOUNDARY_PUNCTUATION)}])" | |
| source_cursor = 0 | |
| for segment in re.split(boundary_pattern, source): | |
| if not segment: | |
| continue | |
| if len(segment) == 1 and segment in _NETWORK_BOUNDARY_PUNCTUATION: | |
| position = source.find(segment, source_cursor) | |
| if position < 0: | |
| raise ValueError("network span tokenizer lost a boundary") | |
| records.append(("", position, position + 1)) | |
| source_cursor = position + 1 | |
| continue | |
| segment_start = source.find(segment, source_cursor) | |
| if segment_start < 0: | |
| raise ValueError("network span tokenizer lost a segment") | |
| records.extend( | |
| ( | |
| match.group(0), | |
| segment_start + match.start(), | |
| segment_start + match.end(), | |
| ) | |
| for match in re.finditer(r"\S+", segment) | |
| ) | |
| source_cursor = segment_start + len(segment) | |
| spans: list[tuple[int, int, str]] = [] | |
| index = 0 | |
| while index < len(records): | |
| if not _is_expanded_network_token(records[index][0]): | |
| index += 1 | |
| continue | |
| end = index | |
| while end < len(records) and _is_expanded_network_token(records[end][0]): | |
| end += 1 | |
| tokens = [token for token, _, _ in records[index:end]] | |
| comparison_tokens = [ | |
| _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token) | |
| for token in tokens | |
| ] | |
| protected_start: int | None = None | |
| for anchor in range(1, len(comparison_tokens) - 2): | |
| if comparison_tokens[anchor : anchor + 3] != ["冒號", "斜線", "斜線"]: | |
| continue | |
| for scheme in ("https", "http", "ftp"): | |
| length = len(scheme) | |
| start = anchor - length | |
| letters = ( | |
| [ | |
| _network_letter_token(token) | |
| for token in comparison_tokens[start:anchor] | |
| ] | |
| if start >= 0 | |
| else [] | |
| ) | |
| if letters and None not in letters and "".join(letters) == scheme: | |
| protected_start = start | |
| break | |
| if protected_start is not None: | |
| break | |
| email_bounds = ( | |
| _spoken_email_subspan_bounds(comparison_tokens) | |
| if protected_start is None | |
| else () | |
| ) | |
| if protected_start is None and not email_bounds: | |
| for start in range(max(0, len(comparison_tokens) - 3)): | |
| prefix = comparison_tokens[start] | |
| split_prefix = [ | |
| _network_letter_token(token) | |
| for token in comparison_tokens[start : start + 3] | |
| ] | |
| if ( | |
| ( | |
| prefix.casefold() == "www" | |
| and comparison_tokens[start + 1] == "點" | |
| ) | |
| or ( | |
| split_prefix == ["w", "w", "w"] | |
| and comparison_tokens[start + 3] == "點" | |
| ) | |
| ): | |
| protected_start = start | |
| break | |
| if protected_start is not None: | |
| span_start = records[index + protected_start][1] | |
| span_end = records[end - 1][2] | |
| spans.append((span_start, span_end, source[span_start:span_end])) | |
| elif email_bounds: | |
| for token_start, token_end in email_bounds: | |
| span_start = records[index + token_start][1] | |
| span_end = records[index + token_end - 1][2] | |
| spans.append((span_start, span_end, source[span_start:span_end])) | |
| index = max(end, index + 1) | |
| return tuple(spans) | |
| def _expanded_network_spoken_spans(text: str) -> tuple[str, ...]: | |
| return tuple( | |
| spoken | |
| for _, _, spoken in _expanded_network_spoken_span_records(text) | |
| ) | |
| def network_protected_spoken_spans(text: str) -> tuple[str, ...]: | |
| """Return exact spoken URL/email spans from raw or frontend-normalized text.""" | |
| raw_spans = _raw_network_spoken_spans(text) | |
| if raw_spans: | |
| return raw_spans | |
| records = ( | |
| *_naturalized_url_spoken_records(text), | |
| *_expanded_network_spoken_span_records(text), | |
| ) | |
| selected: list[tuple[int, int, str]] = [] | |
| for start, end, spoken in sorted(records, key=lambda item: (item[0], -item[1])): | |
| if any( | |
| start < kept_end and end > kept_start | |
| for kept_start, kept_end, _ in selected | |
| ): | |
| continue | |
| selected.append((start, end, spoken)) | |
| return tuple(spoken for _, _, spoken in selected) | |
| def _zh_semantic_version(match: re.Match[str]) -> str: | |
| major, minor, patch, prerelease, build = match.groups() | |
| try: | |
| core = "點".join(_zh_integer(int(part)) for part in (major, minor, patch)) | |
| except ValueError: | |
| return match.group(0) | |
| output = f"版本{core}" | |
| if prerelease: | |
| output += f" 預發布 {_zh_version_qualifier(prerelease)}" | |
| if build: | |
| output += f" 建置 {_zh_version_qualifier(build)}" | |
| return output | |
| def _zh_currency(code: str, number: str, original: str) -> str: | |
| reading = _ZH_CURRENCY_READINGS.get(code.upper(), _ZH_CURRENCY_READINGS.get(code)) | |
| if reading is None: | |
| return original | |
| try: | |
| spoken_number = _zh_number(number.replace(",", "")) | |
| except ValueError: | |
| return original | |
| return reading + spoken_number | |
| def normalize_spoken_forms(text: str, *, locale: str = "zh-TW") -> str: | |
| """Conservatively expand common written forms for TTS. | |
| Chinese locales expand validated dates, network addresses, semantic | |
| versions, currencies, percentages, numeric units, and standalone | |
| all-uppercase acronyms/model identifiers. Ordinary English words are | |
| deliberately left untouched. English locales only receive the existing | |
| Unicode/punctuation normalization. Unknown locales raise instead of | |
| silently applying the wrong pronunciation rules. | |
| """ | |
| raw = unicodedata.normalize("NFC", str(text or "")) | |
| locale_key = str(locale or "").replace("_", "-").lower() | |
| if locale_key in {"en", "en-us", "en-gb"}: | |
| return normalize_tts_text(raw) | |
| if locale_key not in {"zh", "zh-tw", "zh-hant", "zh-cn", "zh-hans"}: | |
| raise ValueError(f"unsupported spoken-form locale: {locale!r}") | |
| if _UNSUPPORTED_QUOTED_EMAIL_RE.search(raw): | |
| raise ValueError("quoted email local parts are not supported") | |
| if any( | |
| match.group(0).endswith(tuple(".,!?;:'")) | |
| for match in _SPOKEN_URL_RE.finditer(raw) | |
| ): | |
| raise ValueError( | |
| "URLs ending in ambiguous punctuation are not supported; " | |
| "percent-encode the final symbol" | |
| ) | |
| protected: list[str] = [] | |
| def protect(spoken: str) -> str: | |
| marker = f"\uf000{len(protected)}\uf001" | |
| protected.append(spoken) | |
| return marker | |
| def network_prefix_boundary(match: re.Match[str]) -> str: | |
| preceding = match.string[: match.start()].rstrip()[-1:] | |
| if not preceding or preceding in _NETWORK_BOUNDARY_PUNCTUATION: | |
| return "" | |
| return "，" | |
| def network_suffix_boundary(match: re.Match[str], *, trailing: str = "") -> str: | |
| """Keep expanded network data separate from following prose. | |
| Once a URL/email is expanded, whitespace alone cannot distinguish its | |
| final path/domain token from ordinary following English. A semantic | |
| comma is inaudible as lexical content but preserves that provenance | |
| for chunking and the exact protected-range verifier. | |
| """ | |
| if trailing: | |
| return "" | |
| remainder = match.string[match.end() :] | |
| following = remainder.lstrip()[:1] | |
| if not following or following in _NETWORK_BOUNDARY_PUNCTUATION: | |
| return "" | |
| return "，" | |
| def replace_url(match: re.Match[str]) -> str: | |
| matched = match.group(0) | |
| return ( | |
| network_prefix_boundary(match) | |
| + protect(_zh_url(matched)) | |
| + network_suffix_boundary(match) | |
| ) | |
| def replace_email(match: re.Match[str]) -> str: | |
| return ( | |
| network_prefix_boundary(match) | |
| + protect(_zh_email(match.group(0))) | |
| + network_suffix_boundary(match) | |
| ) | |
| def replace_date(match: re.Match[str]) -> str: | |
| year_text, _, month_text, day_text = match.groups() | |
| try: | |
| date(int(year_text), int(month_text), int(day_text)) | |
| except ValueError: | |
| return match.group(0) | |
| return ( | |
| f"{_zh_digit_sequence(year_text)}年" | |
| f"{_zh_integer(int(month_text))}月{_zh_integer(int(day_text))}日" | |
| ) | |
| def replace_zh_date(match: re.Match[str]) -> str: | |
| year_text, month_text, day_text = match.groups() | |
| try: | |
| date(int(year_text), int(month_text), int(day_text)) | |
| except ValueError: | |
| return match.group(0) | |
| return ( | |
| f"{_zh_digit_sequence(year_text)}年" | |
| f"{_zh_integer(int(month_text))}月{_zh_integer(int(day_text))}日" | |
| ) | |
| def replace_number(match: re.Match[str], suffix: str) -> str: | |
| try: | |
| return _zh_number(match.group(1)) + suffix | |
| except ValueError: | |
| return match.group(0) | |
| def replace_time(match: re.Match[str]) -> str: | |
| return _zh_clock_time(match.group(1), match.group(2)) | |
| def replace_percent(match: re.Match[str]) -> str: | |
| try: | |
| return "百分之" + _zh_number(match.group(1)) | |
| except ValueError: | |
| return match.group(0) | |
| def replace_unit(match: re.Match[str]) -> str: | |
| unit_key = re.sub(r"\s+", "", match.group(2)).lower() | |
| reading = _ZH_UNIT_READINGS.get(unit_key) | |
| if reading is None: | |
| return match.group(0) | |
| return replace_number(match, reading) | |
| def replace_model_code(match: re.Match[str]) -> str: | |
| letters, digits = match.groups() | |
| return f"{' '.join(letters)} {_zh_digit_sequence(digits)}" | |
| output = _SPOKEN_URL_RE.sub(replace_url, raw) | |
| output = _SPOKEN_EMAIL_RE.sub(replace_email, output) | |
| output = _SPOKEN_ZH_DATE_RE.sub(replace_zh_date, output) | |
| output = _SPOKEN_DATE_RE.sub(replace_date, output) | |
| output = _SPOKEN_TIME_RE.sub(replace_time, output) | |
| output = _SPOKEN_PERCENT_RE.sub(replace_percent, output) | |
| output = _SPOKEN_UNIT_RE.sub(replace_unit, output) | |
| output = _SPOKEN_ZH_MEASURE_RE.sub( | |
| lambda match: replace_number(match, match.group(2)), | |
| output, | |
| ) | |
| output = _SPOKEN_SEMVER_RE.sub(_zh_semantic_version, output) | |
| output = _SPOKEN_CURRENCY_PREFIX_RE.sub( | |
| lambda match: _zh_currency(match.group(1), match.group(2), match.group(0)), | |
| output, | |
| ) | |
| output = _SPOKEN_CURRENCY_SUFFIX_RE.sub( | |
| lambda match: _zh_currency(match.group(2), match.group(1), match.group(0)), | |
| output, | |
| ) | |
| output = _MODEL_CODE_RE.sub(replace_model_code, output) | |
| output = _UPPERCASE_ACRONYM_RE.sub(lambda match: " ".join(match.group(1)), output) | |
| for index, spoken in enumerate(protected): | |
| output = output.replace(f"\uf000{index}\uf001", spoken) | |
| return normalize_tts_text(output) | |
| def normalize_asr_spoken_forms( | |
| text: str, | |
| target_text: str, | |
| *, | |
| locale: str = "zh-TW", | |
| ) -> str: | |
| """Canonicalize only target-proven ASR numeric and frequency readings.""" | |
| raw_target = unicodedata.normalize("NFC", str(target_text or "")) | |
| raw_text = unicodedata.normalize("NFC", str(text or "")) | |
| normalized_target = normalize_spoken_forms(raw_target, locale=locale) | |
| locale_key = str(locale or "").replace("_", "-").lower() | |
| if locale_key not in {"zh", "zh-tw", "zh-hant", "zh-cn", "zh-hans"}: | |
| return normalize_spoken_forms(raw_text, locale=locale) | |
| # Preserve complete literal URL/email spans before fragment punctuation | |
| # canonicalization. Otherwise rewriting a dot first destroys the raw URL | |
| # regex and makes a correct path/query fail the protected exact-range gate. | |
| protected_network: list[str] = [] | |
| def network_comparison_key(value: str) -> str: | |
| canonical = normalize_tts_eval_text(value).casefold() | |
| for reading, letter in sorted( | |
| _ZH_NETWORK_READING_TO_LETTER.items(), | |
| key=lambda item: len(item[0]), | |
| reverse=True, | |
| ): | |
| canonical = canonical.replace(reading.casefold(), letter) | |
| canonical = _T2S_CONVERTER.convert(canonical) | |
| output: list[str] = [] | |
| for character in canonical: | |
| if character in _ZH_DIGITS: | |
| output.append(str(_ZH_DIGITS.index(character))) | |
| continue | |
| try: | |
| output.append(str(unicodedata.decimal(character))) | |
| except (TypeError, ValueError): | |
| if character.isalnum(): | |
| output.append(character) | |
| return "".join(output) | |
| target_network_spans = network_protected_spoken_spans(normalized_target) | |
| target_network_keys = { | |
| network_comparison_key(span) | |
| for span in target_network_spans | |
| } | |
| email_symbol_tokens = { | |
| reading: symbol | |
| for symbol, reading in _ZH_NETWORK_SYMBOL_READINGS.items() | |
| if symbol in ".!#$%&'*+/=?^_`{|}~-@" | |
| } | |
| def spoken_email_token_value(token: str) -> str | None: | |
| token = _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token) | |
| letter = _network_letter_token(token) | |
| if letter is not None: | |
| return letter | |
| if token.isascii() and token.isalnum(): | |
| return token.casefold() | |
| if token and all(character in _ZH_DIGITS for character in token): | |
| return "".join(str(_ZH_DIGITS.index(character)) for character in token) | |
| return email_symbol_tokens.get(token) | |
| def spoken_email_subspans(span: str) -> tuple[str, ...]: | |
| """Extract maximal syntactically valid emails around spoken ``@``.""" | |
| tokens = [ | |
| _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token) | |
| for token in span.split() | |
| ] | |
| output: list[str] = [] | |
| for at, token in enumerate(tokens): | |
| if token != "小老鼠": | |
| continue | |
| if any( | |
| tokens[index : index + 3] == ["冒號", "斜線", "斜線"] | |
| for index in range(max(0, at - 2)) | |
| ): | |
| continue | |
| www_end = None | |
| for start in range(at): | |
| if ( | |
| start + 1 < at | |
| and tokens[start].casefold() == "www" | |
| and tokens[start + 1] == "點" | |
| ): | |
| www_end = start + 2 | |
| break | |
| if ( | |
| start + 3 < at | |
| and [ | |
| _network_letter_token(part) | |
| for part in tokens[start : start + 3] | |
| ] | |
| == ["w", "w", "w"] | |
| and tokens[start + 3] == "點" | |
| ): | |
| www_end = start + 4 | |
| break | |
| if www_end is not None and any( | |
| part in {"斜線", "問號", "井號", "冒號"} | |
| for part in tokens[www_end:] | |
| ): | |
| continue | |
| left_floor = 0 | |
| for separator in range(at): | |
| if tokens[separator] == "和" and "小老鼠" in tokens[:separator]: | |
| left_floor = separator + 1 | |
| left = at | |
| while left > left_floor: | |
| value = spoken_email_token_value(tokens[left - 1]) | |
| if value is None or value == "@": | |
| break | |
| left -= 1 | |
| right = at + 1 | |
| while right < len(tokens): | |
| value = spoken_email_token_value(tokens[right]) | |
| if value is None or value == "@": | |
| break | |
| right += 1 | |
| matches: list[tuple[int, int, int, str]] = [] | |
| for start in range(left, at): | |
| for end in range(at + 2, right + 1): | |
| rendered = "".join( | |
| value | |
| for part in tokens[start:end] | |
| if (value := spoken_email_token_value(part)) is not None | |
| ) | |
| if _SPOKEN_EMAIL_RE.fullmatch(rendered): | |
| matches.append( | |
| ( | |
| len(rendered), | |
| end - start, | |
| -start, | |
| " ".join(tokens[start:end]), | |
| ) | |
| ) | |
| if matches: | |
| output.append(max(matches)[3]) | |
| return tuple(output) | |
| target_email_key_limits: dict[str, int] = {} | |
| target_email_value_limits: dict[str, int] = {} | |
| for span in target_network_spans: | |
| for email_span in spoken_email_subspans(span): | |
| key = network_comparison_key(email_span) | |
| if key: | |
| target_email_key_limits[key] = target_email_key_limits.get(key, 0) + 1 | |
| rendered = "".join( | |
| value | |
| for token in email_span.split() | |
| if (value := spoken_email_token_value(token)) is not None | |
| ) | |
| if _SPOKEN_EMAIL_RE.fullmatch(rendered): | |
| rendered = rendered.casefold() | |
| target_email_value_limits[rendered] = ( | |
| target_email_value_limits.get(rendered, 0) + 1 | |
| ) | |
| def protect_network(spoken: str) -> str: | |
| marker = f"\uf100{len(protected_network)}\uf101" | |
| protected_network.append(_canonicalize_network_spoken_span(spoken)) | |
| return marker | |
| # Pinned Whisper decoders sometimes romanize the spoken email separator | |
| # and remove the whitespace at the preceding CJK/ASCII boundary. Recover | |
| # only a complete email that is already proven by the normalized target; | |
| # every local-part, domain and TLD atom must match exactly and identifier | |
| # characters are forbidden on both outer edges. | |
| email_identifier_boundary = r"A-Za-z0-9.!#$%&'*+/=?^_`{|}~@-" | |
| for target_email, occurrence_limit in target_email_value_limits.items(): | |
| local, domain = target_email.rsplit("@", 1) | |
| alias_pattern = re.compile( | |
| rf"(?i)(?<![{email_identifier_boundary}])" | |
| rf"{re.escape(local)}\s*" | |
| rf"(?:xiaolaoshui|xiaolaoshu|xlouse)\s*" | |
| rf"{re.escape(domain)}" | |
| rf"(?![{email_identifier_boundary}])" | |
| ) | |
| raw_text = alias_pattern.sub( | |
| lambda _match, email=target_email: protect_network(_zh_email(email)), | |
| raw_text, | |
| count=occurrence_limit, | |
| ) | |
| def protect_url(match: re.Match[str]) -> str: | |
| matched = match.group(0) | |
| alias_in_or_after_match = ( | |
| _ASR_XIAOLAOSHU_ALIAS_RE.search(matched) is not None | |
| or re.match( | |
| r"\s*xiaolaoshu", | |
| match.string[match.end() :], | |
| flags=re.IGNORECASE, | |
| ) | |
| is not None | |
| ) | |
| if target_email_key_limits and alias_in_or_after_match: | |
| # A no-space ``www.fooXiaolaoshuexample.tw`` rendering is first | |
| # recognized by the broad schemeless-URL regex. Leave it visible | |
| # to the exact target-email alias proof below; wrong content still | |
| # remains literal and fails the protected-range comparison. | |
| return matched | |
| core = matched | |
| # Whisper commonly emits ASCII sentence punctuation directly after a | |
| # URL. Keep a trailing symbol inside the protected range only when the | |
| # normalized target proves that exact URL spelling; otherwise preserve | |
| # it outside the marker as ordinary punctuation. | |
| while core and core[-1] in ".,!?;:": | |
| if network_comparison_key(_zh_url(core)) in target_network_keys: | |
| break | |
| core = core[:-1] | |
| trailing = matched[len(core) :] | |
| return protect_network(_zh_url(core)) + trailing | |
| raw_text = _SPOKEN_URL_RE.sub(protect_url, raw_text) | |
| raw_text = _SPOKEN_EMAIL_RE.sub( | |
| lambda match: protect_network(_zh_email(match.group(0))), | |
| raw_text, | |
| ) | |
| target_network_tokens = { | |
| token | |
| for span in target_network_spans | |
| for token in span.split() | |
| } | |
| if {"冒號", "斜線"}.issubset(target_network_tokens): | |
| raw_text = re.sub( | |
| r"\s*:\s*/\s*/\s*", | |
| " 冒號 斜線 斜線 ", | |
| raw_text, | |
| ) | |
| # Whisper often formats a correctly spoken URL/email fragment back into | |
| # literal punctuation. Expand only separators proven by the target and | |
| # bounded by identifier characters. Missing/reordered letters still fail, | |
| # while sentence punctuation is deliberately outside the bounded pattern. | |
| for symbol, reading in _ZH_NETWORK_SYMBOL_READINGS.items(): | |
| alpha_side = r"(?:[A-Za-z]\s*)+" | |
| fragment_proven = re.search( | |
| rf"{alpha_side}{re.escape(reading)}\s*{alpha_side}", | |
| normalized_target, | |
| ) is not None | |
| if reading not in target_network_tokens and not fragment_proven: | |
| continue | |
| raw_text = re.sub( | |
| rf"(?:" | |
| rf"(?<=[A-Za-z0-9])\s*{re.escape(symbol)}\s*(?=[^\W_])" | |
| rf"|(?<=[^\W_])\s*{re.escape(symbol)}\s*(?=[A-Za-z0-9])" | |
| rf")", | |
| f" {reading} ", | |
| raw_text, | |
| ) | |
| # Whisper can render the spoken email separator ``小老鼠`` as its | |
| # Mandarin pinyin ``Xiaolaoshu``, including without spaces between opaque | |
| # labels. Try each occurrence independently and accept it only when the | |
| # replacement adds exact coverage of a complete target email span. This | |
| # keeps URL paths containing ``@``, wrong domains, missing labels, literal | |
| # labels, prose aliases, and already-covered targets fail-closed. | |
| def covered_target_email_counts(value: str) -> dict[str, int]: | |
| unprotected = value | |
| for index in range(len(protected_network)): | |
| unprotected = unprotected.replace(f"\uf100{index}\uf101", "，") | |
| counts = {key: 0 for key in target_email_key_limits} | |
| for span in protected_network: | |
| for email_span in spoken_email_subspans(span): | |
| key = network_comparison_key(email_span) | |
| limit = target_email_key_limits.get(key, 0) | |
| if limit and counts[key] < limit: | |
| counts[key] += 1 | |
| for span in network_protected_spoken_spans(unprotected): | |
| for email_span in spoken_email_subspans(span): | |
| key = network_comparison_key(email_span) | |
| limit = target_email_key_limits.get(key, 0) | |
| if limit and counts[key] < limit: | |
| counts[key] += 1 | |
| return counts | |
| alias_search_start = 0 | |
| while target_email_key_limits: | |
| alias_match = _ASR_XIAOLAOSHU_ALIAS_RE.search(raw_text, alias_search_start) | |
| if alias_match is None: | |
| break | |
| candidate = ( | |
| raw_text[: alias_match.start()] | |
| + " 小老鼠 " | |
| + raw_text[alias_match.end() :] | |
| ) | |
| before_counts = covered_target_email_counts(raw_text) | |
| after_counts = covered_target_email_counts(candidate) | |
| if any( | |
| after_counts[key] > before_counts[key] | |
| for key in target_email_key_limits | |
| ): | |
| raw_text = candidate | |
| alias_search_start = alias_match.start() + len(" 小老鼠 ") | |
| else: | |
| alias_search_start = alias_match.end() | |
| for spoken in network_protected_spoken_spans(raw_text): | |
| pattern = r"\s+".join(re.escape(token) for token in spoken.split()) | |
| match = re.search(pattern, raw_text) | |
| if match is None: | |
| continue | |
| raw_text = ( | |
| raw_text[: match.start()] | |
| + protect_network(spoken) | |
| + raw_text[match.end() :] | |
| ) | |
| placeholder_prefix = "藍鵲網路保護佔位符" | |
| while placeholder_prefix in raw_text or placeholder_prefix in raw_target: | |
| placeholder_prefix += "號" | |
| network_placeholders: list[tuple[str, str]] = [] | |
| for index, spoken in enumerate(protected_network): | |
| marker = f"\uf100{index}\uf101" | |
| placeholder = f"{placeholder_prefix}{_zh_integer(index + 1)}結束" | |
| raw_text = raw_text.replace(marker, placeholder) | |
| network_placeholders.append((placeholder, spoken)) | |
| def restore_network_placeholders(value: str) -> str: | |
| for placeholder, spoken in network_placeholders: | |
| value = value.replace(placeholder, spoken) | |
| return value | |
| raw_text = _canonicalize_target_proven_structured_asr_forms( | |
| raw_text, | |
| normalized_target, | |
| ) | |
| target_proven_text = _canonicalize_target_proven_frequency_scales( | |
| raw_text, | |
| raw_target, | |
| ) | |
| # Preserve numeric grouping long enough to prove ASR forms such as | |
| # ``日圓4,500元``. General punctuation normalization would otherwise | |
| # turn the thousands separator into a sentence comma before the bounded | |
| # JPY matcher sees it. The post-normalization pass remains necessary for | |
| # already-spoken Chinese amount forms. | |
| target_proven_text = _canonicalize_target_proven_jpy_amounts( | |
| target_proven_text, | |
| normalized_target, | |
| ) | |
| normalized_text = normalize_spoken_forms(target_proven_text, locale=locale) | |
| normalized_text = _canonicalize_target_proven_jpy_amounts( | |
| normalized_text, | |
| normalized_target, | |
| ) | |
| normalized_text = _canonicalize_target_proven_arabic_digits( | |
| normalized_text, | |
| normalized_target, | |
| ) | |
| # Only an unambiguous target reading may prove that Whisper's numeric | |
| # ``15點30(分)`` rendering means a clock time. In particular, do not | |
| # rewrite the target's own numeric form first: that would let ambiguous | |
| # decimals such as ``15點05分貝`` or scores self-authorize a clock | |
| # canonicalization. Colon times have already been expanded by | |
| # ``normalize_spoken_forms`` and fully-spoken clock targets already contain | |
| # the canonical phrase, so both remain valid proof sources. | |
| target_clock_text = normalized_target.translate( | |
| str.maketrans({"点": "點", "時": "點", "时": "點"}) | |
| ) | |
| if ( | |
| _ASR_EXPLICIT_ZH_TIME_RE.search(normalized_target) is not None | |
| or _ASR_BARE_ZH_TIME_RE.search(normalized_target) is not None | |
| ): | |
| # A numeric ``X點Y`` construction in the target is itself ambiguous. | |
| # Do not let a separate, fully-spoken clock elsewhere in the sentence | |
| # globally authorize rewriting that decimal, duration, or score. | |
| return restore_network_placeholders(normalized_text) | |
| def replace_target_proven_time(match: re.Match[str]) -> str: | |
| canonical = _zh_clock_time(match.group(1), match.group(2)) | |
| return canonical if canonical in target_clock_text else match.group(0) | |
| normalized_text = _ASR_EXPLICIT_ZH_TIME_RE.sub( | |
| replace_target_proven_time, | |
| normalized_text, | |
| ) | |
| normalized_text = _ASR_BARE_ZH_TIME_RE.sub( | |
| replace_target_proven_time, | |
| normalized_text, | |
| ) | |
| return restore_network_placeholders(normalize_tts_text(normalized_text)) | |
| def normalize_tts_text(text: str) -> str: | |
| """Normalize common punctuation and pronounce standalone Bopomofo symbols.""" | |
| raw = unicodedata.normalize("NFC", str(text or "")) | |
| has_cjk = any(_is_cjk(char) for char in raw) | |
| output: list[str] = [] | |
| index = 0 | |
| while index < len(raw): | |
| char = raw[index] | |
| if char.isspace() or char in {"\r", "\n", "\t", "\u3000"}: | |
| output.append(" ") | |
| index += 1 | |
| continue | |
| if unicodedata.category(char)[0] == "C": | |
| index += 1 | |
| continue | |
| if raw.startswith("...", index): | |
| output.append("。" if has_cjk else ".") | |
| index += 3 | |
| while index < len(raw) and raw[index] == ".": | |
| index += 1 | |
| continue | |
| if char in {"…", "⋯"}: | |
| output.append("。" if has_cjk else ".") | |
| index += 1 | |
| while index < len(raw) and raw[index] in {"…", "⋯"}: | |
| index += 1 | |
| continue | |
| if _is_bopomofo(char): | |
| while index < len(raw) and _is_bopomofo(raw[index]): | |
| reading = _BOPOMOFO_READINGS.get(raw[index]) | |
| if reading: | |
| output.append(reading) | |
| index += 1 | |
| continue | |
| previous = raw[index - 1] if index else "" | |
| following = raw[index + 1] if index + 1 < len(raw) else "" | |
| folded = unicodedata.normalize("NFKC", char) | |
| if folded.isascii() and folded.isalnum(): | |
| char = folded | |
| if char in {",", "，", "﹐", "､"}: | |
| output.append("," if previous.isdigit() and following.isdigit() else ("，" if has_cjk else ",")) | |
| elif char in {".", "。", "｡", "．"}: | |
| inside_ascii = previous.isascii() and previous.isalnum() and following.isascii() and following.isalnum() | |
| output.append("." if inside_ascii or not has_cjk else "。") | |
| elif char in {"?", "？", "﹖"}: | |
| output.append("？" if has_cjk else "?") | |
| elif char in {"!", "！", "﹗"}: | |
| output.append("！" if has_cjk else "!") | |
| elif char in {";", "；"}: | |
| output.append("；" if has_cjk else ";") | |
| elif char in {":", "："}: | |
| ascii_context = ( | |
| previous.isascii() | |
| and previous.isalnum() | |
| and following.isascii() | |
| and following.isalnum() | |
| ) | |
| output.append(":" if following == "/" or ascii_context else ("：" if has_cjk else ":")) | |
| elif char in {"“", "”", "„", "＂"}: | |
| output.append('"') | |
| elif char in {"‘", "’", "＇"}: | |
| output.append("'") | |
| elif char == "（": | |
| output.append("(") | |
| elif char == "）": | |
| output.append(")") | |
| else: | |
| output.append(char) | |
| index += 1 | |
| normalized = _SPACE_RE.sub(" ", "".join(output)) | |
| normalized = _PUNCT_NO_LEFT_SPACE_RE.sub(r"\1", normalized) | |
| normalized = _CJK_PUNCT_RIGHT_SPACE_RE.sub(r"\1", normalized) | |
| return normalized.strip() | |
| _UNSPOKEN_CJK_QUOTE_MARKS = frozenset("「」『』") | |
| def strip_unspoken_cjk_quotes_for_generation(text: str) -> str: | |
| """Remove layout-only CJK quote marks from ordinary generation text. | |
| This helper is intentionally generation-only and must not run on network | |
| requests: quotation marks remain useful raw boundaries around a URL or | |
| email. Lexical verification continues to use the original request, so no | |
| spoken unit is waived by this acoustic-model input cleanup. | |
| """ | |
| normalized = normalize_tts_text(text) | |
| if not any(mark in normalized for mark in _UNSPOKEN_CJK_QUOTE_MARKS): | |
| return normalized | |
| without_quotes = "".join( | |
| " " if character in _UNSPOKEN_CJK_QUOTE_MARKS else character | |
| for character in normalized | |
| ) | |
| return normalize_tts_text(without_quotes) | |
| def normalize_tts_eval_text(text: str) -> str: | |
| """Normalize only verified ASR-equivalent forms for semantic scoring. | |
| This remains separate from model-input normalization: Mandarin pronoun | |
| homophones are acoustically indistinguishable, but their written forms must | |
| stay untouched in the text sent to the TTS model. | |
| """ | |
| normalized = normalize_tts_text(text) | |
| normalized = normalized.translate(_ASR_HOMOPHONE_TRANSLATION) | |
| if "注音" in normalized or "符號" in normalized: | |
| normalized = _BOPOMOFO_ASR_POPO_RE.sub("波坡摸", normalized) | |
| return normalized | |
| def ensure_terminal_punctuation(text: str) -> str: | |
| """Give the acoustic model an explicit endpoint cue without changing words.""" | |
| normalized = normalize_tts_text(text) | |
| if not normalized: | |
| return "" | |
| split_at = len(normalized) | |
| while split_at > 0 and normalized[split_at - 1] in _TRAILING_CLOSERS: | |
| split_at -= 1 | |
| core = normalized[:split_at].rstrip() | |
| suffix = normalized[split_at:] | |
| if core and core[-1] in _TERMINAL_PUNCTUATION: | |
| return normalized | |
| while core and core[-1] in _NONTERMINAL_TRAILING_PUNCTUATION: | |
| core = core[:-1].rstrip() | |
| if not core: | |
| return normalized | |
| punctuation = "。" if any(_is_cjk(char) for char in core) else "." | |
| return f"{core}{punctuation}{suffix}" | |
| def count_speech_units(text: str) -> int: | |
| """Count CJK characters and compressed ASCII runs for pace control.""" | |
| units = 0 | |
| ascii_buffer: list[str] = [] | |
| def flush_ascii() -> None: | |
| nonlocal units | |
| if not ascii_buffer: | |
| return | |
| token = "".join(ascii_buffer) | |
| divisor = 2 if token.isdigit() else 4 | |
| units += max(1, math.ceil(len(token) / divisor)) | |
| ascii_buffer.clear() | |
| for char in normalize_tts_text(text).lower(): | |
| if char.isascii() and char.isalnum(): | |
| ascii_buffer.append(char) | |
| elif _is_cjk(char): | |
| flush_ascii() | |
| units += 1 | |
| else: | |
| flush_ascii() | |
| flush_ascii() | |
| return units | |
| def count_network_endpoint_duration_units(text: str) -> int: | |
| """Count conservative duration units for network-conditioned generation. | |
| Public pace, chunk planning and generation-work budgets intentionally keep | |
| using :func:`count_speech_units`. The acoustic model needs more native | |
| endpoint headroom for opaque URL/email labels, though, so this separate | |
| counter compresses every contiguous ASCII alphanumeric run by two instead | |
| of compressing alphabetic runs by four. It is valid only for the | |
| network-conditioned endpoint estimate and never imposes a minimum length. | |
| """ | |
| units = 0 | |
| ascii_run_length = 0 | |
| def flush_ascii() -> None: | |
| nonlocal units, ascii_run_length | |
| if ascii_run_length <= 0: | |
| return | |
| units += max(1, math.ceil(ascii_run_length / 2)) | |
| ascii_run_length = 0 | |
| for char in normalize_tts_text(text).lower(): | |
| if char.isascii() and char.isalnum(): | |
| ascii_run_length += 1 | |
| elif _is_cjk(char): | |
| flush_ascii() | |
| units += 1 | |
| else: | |
| flush_ascii() | |
| flush_ascii() | |
| return units | |
| def effective_generation_cfg( | |
| text: str, | |
| requested_cfg: float, | |
| *, | |
| short_text_unit_threshold: int = 6, | |
| short_text_min_cfg: float = 3.0, | |
| ) -> float: | |
| """Use stronger acoustic guidance only for empirically unstable short text.""" | |
| cfg = float(requested_cfg) | |
| units = count_speech_units(text) | |
| if 0 < units <= max(1, int(short_text_unit_threshold)): | |
| return max(cfg, float(short_text_min_cfg)) | |
| return cfg | |
| def estimate_step_seconds(model, sample_rate: int) -> float | None: | |
| patch_size = int( | |
| getattr(model, "patch_size", 0) | |
| or getattr(getattr(model, "config", None), "patch_size", 0) | |
| or 0 | |
| ) | |
| audio_vae = getattr(model, "audio_vae", None) | |
| decode_chunk = int( | |
| getattr(model, "_decode_chunk_size", 0) | |
| or getattr(audio_vae, "decode_chunk_size", 0) | |
| or getattr(audio_vae, "chunk_size", 0) | |
| or 0 | |
| ) | |
| if patch_size <= 0 or decode_chunk <= 0 or sample_rate <= 0: | |
| return None | |
| return float(patch_size * decode_chunk / sample_rate) | |
| def target_cps_min_len( | |
| text: str, | |
| target_cps: float, | |
| step_seconds: float | None, | |
| base_min_len: int = 2, | |
| stop_consecutive: int = 1, | |
| ) -> int: | |
| if target_cps <= 0.0 or not step_seconds or step_seconds <= 0.0: | |
| return int(base_min_len) | |
| units = count_speech_units(text) | |
| if units <= 0: | |
| return int(base_min_len) | |
| target_steps = math.ceil((units / target_cps) / step_seconds) | |
| earliest_stop_offset = max(1, int(stop_consecutive)) + 1 | |
| return max(int(base_min_len), max(0, target_steps - earliest_stop_offset)) | |
| def target_cps_steps(text: str, target_cps: float, step_seconds: float | None) -> int: | |
| if target_cps <= 0.0 or not step_seconds or step_seconds <= 0.0: | |
| return 0 | |
| units = count_speech_units(text) | |
| if units <= 0: | |
| return 0 | |
| return max(1, math.ceil((units / target_cps) / step_seconds)) | |
| def target_pace_speed( | |
| audio_samples: int, | |
| sample_rate: int, | |
| text: str, | |
| *, | |
| target_cps: float, | |
| min_speed: float = 0.80, | |
| ) -> float: | |
| """Return a pitch-preserving stretch rate without extending generation.""" | |
| units = count_speech_units(text) | |
| if audio_samples <= 0 or sample_rate <= 0 or units <= 0 or target_cps <= 0.0: | |
| return 1.0 | |
| actual_seconds = float(audio_samples) / float(sample_rate) | |
| target_seconds = float(units) / float(target_cps) | |
| if actual_seconds >= target_seconds: | |
| return 1.0 | |
| return min(1.0, max(float(min_speed), actual_seconds / target_seconds)) | |
| def active_pace_correction_speed( | |
| active_duration_seconds: float, | |
| text: str, | |
| *, | |
| target_cps: float, | |
| prior_speed: float = 1.0, | |
| min_total_speed: float = 0.80, | |
| ) -> float: | |
| """Return a second stretch rate from active-voice duration. | |
| ``prior_speed`` is the rate already applied by ``target_pace_speed``. | |
| Successive pitch-preserving stretch rates multiply, so the second rate is | |
| floored at ``min_total_speed / prior_speed``. Invalid evidence and audio | |
| that is already at or below the target active CPS fail safely to ``1.0``; | |
| this helper never speeds audio up. | |
| """ | |
| units = count_speech_units(text) | |
| try: | |
| active_seconds = float(active_duration_seconds) | |
| target = float(target_cps) | |
| previous_rate = float(prior_speed) | |
| total_floor = float(min_total_speed) | |
| except (TypeError, ValueError, OverflowError): | |
| return 1.0 | |
| if ( | |
| units <= 0 | |
| or not math.isfinite(active_seconds) | |
| or active_seconds <= 0.0 | |
| or not math.isfinite(target) | |
| or target <= 0.0 | |
| or not math.isfinite(previous_rate) | |
| or not 0.0 < previous_rate <= 1.0 | |
| or not math.isfinite(total_floor) | |
| or not 0.0 < total_floor <= 1.0 | |
| ): | |
| return 1.0 | |
| desired_rate = active_seconds * target / float(units) | |
| if not math.isfinite(desired_rate) or desired_rate >= 1.0: | |
| return 1.0 | |
| remaining_floor = total_floor / previous_rate | |
| if remaining_floor >= 1.0: | |
| return 1.0 | |
| return min(1.0, max(remaining_floor, desired_rate)) | |
| def duration_hard_stop_steps( | |
| expected_steps: int, | |
| *, | |
| ratio: float = 1.08, | |
| margin_steps: int = 3, | |
| fallback: int = 2000, | |
| ) -> int: | |
| expected_steps = max(0, int(expected_steps)) | |
| if expected_steps <= 0: | |
| return max(1, int(fallback)) | |
| return max( | |
| expected_steps + max(0, int(margin_steps)), | |
| math.ceil(expected_steps * max(1.0, float(ratio))), | |
| ) | |
| def endpoint_generation_plan( | |
| text: str, | |
| *, | |
| generation_cps: float, | |
| step_seconds: float | None, | |
| margin_steps: int = 1, | |
| add_terminal_punctuation: bool = True, | |
| duration_units: int | None = None, | |
| ) -> tuple[str, int, int]: | |
| """Plan a native-pace generation cap independently of output playback pace.""" | |
| model_text = ( | |
| ensure_terminal_punctuation(text) | |
| if add_terminal_punctuation | |
| else normalize_tts_text(text) | |
| ) | |
| if duration_units is None: | |
| expected_steps = target_cps_steps( | |
| model_text, | |
| generation_cps, | |
| step_seconds, | |
| ) | |
| else: | |
| if isinstance(duration_units, bool): | |
| raise ValueError("duration_units must be a non-negative integer") | |
| try: | |
| planned_units = operator.index(duration_units) | |
| except (TypeError, ValueError, OverflowError) as error: | |
| raise ValueError( | |
| "duration_units must be a non-negative integer" | |
| ) from error | |
| if planned_units < 0: | |
| raise ValueError("duration_units must be a non-negative integer") | |
| expected_steps = ( | |
| 0 | |
| if ( | |
| planned_units <= 0 | |
| or generation_cps <= 0.0 | |
| or not step_seconds | |
| or step_seconds <= 0.0 | |
| ) | |
| else max( | |
| 1, | |
| math.ceil( | |
| (planned_units / generation_cps) / step_seconds | |
| ), | |
| ) | |
| ) | |
| hard_stop_steps = duration_hard_stop_steps( | |
| expected_steps, | |
| ratio=1.0, | |
| margin_steps=margin_steps, | |
| ) | |
| return model_text, expected_steps, hard_stop_steps | |
| def select_generation_cps( | |
| text: str, | |
| *, | |
| cjk_cps: float = 5.2, | |
| ascii_cps: float = 4.6, | |
| ) -> float: | |
| """Reserve more generation time for ASCII words than compact CJK units.""" | |
| normalized = normalize_tts_text(text) | |
| if ( | |
| any(char.isascii() and char.isalnum() for char in normalized) | |
| or network_protected_spoken_spans(normalized) | |
| ): | |
| return float(ascii_cps) | |
| return float(cjk_cps) | |
| def finish_audio( | |
| audio: np.ndarray, | |
| sample_rate: int, | |
| *, | |
| fade_ms: float = 60.0, | |
| trailing_silence_ms: float = 180.0, | |
| ) -> np.ndarray: | |
| """Fade a forced endpoint and leave a short, unambiguous final pause.""" | |
| signal = np.asarray(audio, dtype=np.float32).reshape(-1).copy() | |
| fade_samples = min( | |
| signal.size, | |
| max(0, int(round(float(fade_ms) * int(sample_rate) / 1000.0))), | |
| ) | |
| if fade_samples > 0: | |
| phase = np.linspace( | |
| 0.0, | |
| np.pi / 2.0, | |
| fade_samples, | |
| dtype=np.float32, | |
| ) | |
| envelope = np.square(np.cos(phase)) | |
| envelope[-1] = 0.0 | |
| signal[-fade_samples:] *= envelope | |
| silence_samples = max( | |
| 0, | |
| int(round(float(trailing_silence_ms) * int(sample_rate) / 1000.0)), | |
| ) | |
| if silence_samples > 0: | |
| signal = np.pad(signal, (0, silence_samples)) | |
| return signal | |
| _SEMANTIC_STRONG_BREAKS = frozenset("。！？!?；;") | |
| _SEMANTIC_SOFT_BREAKS = frozenset("，,、：: ") | |
| _QUALITY_MEDIUM_BREAKS = frozenset("。！？!?；;，,、：:") | |
| _NETWORK_CONTEXTUAL_EMAIL_MIN_PROSE_UNITS = 7 | |
| _SEMANTIC_TRAILING_CLOSERS = frozenset("\"'”’」』】）)]}") | |
| _PROTECTED_ASCII_SPAN_RE = re.compile( | |
| r"(?i)[a-z0-9]+(?:[._+/@:#~%-]+[a-z0-9]+)+" | |
| r"|(?<![a-z0-9])(?:[a-z]\s+){2,}[a-z](?![a-z0-9])" | |
| ) | |
| _STRUCTURED_CLAUSE_BREAKS = frozenset("，,、：:") | |
| _STRUCTURED_CLAUSE_TARGET_UNITS = 32 | |
| _NETWORK_ATOMIC_MAX_UNITS = 36 | |
| _NETWORK_GENERATION_COMPONENT_AFTER = frozenset() | |
| _NETWORK_GENERATION_COMPONENT_BEFORE = frozenset( | |
| {"點", "小老鼠", "斜線", "問號", "井號", "冒號", "和"} | |
| ) | |
| _NORMALIZED_ZH_NUMBER_PATTERN = ( | |
| r"[正負]?[" + _ZH_DIGITS + r"十百千萬億兆]+" | |
| r"(?:點[" + _ZH_DIGITS + r"]+)?" | |
| ) | |
| _NORMALIZED_STRUCTURED_NUMERIC_CUE_RE = re.compile( | |
| rf"(?:" | |
| rf"百分之{_NORMALIZED_ZH_NUMBER_PATTERN}" | |
| rf"|(?:新台幣|美元|港幣|人民幣|日圓|歐元|英鎊|韓元)" | |
| rf"{_NORMALIZED_ZH_NUMBER_PATTERN}" | |
| rf"|{_NORMALIZED_ZH_NUMBER_PATTERN}(?:" | |
| rf"公里每小時|公尺每秒|攝氏度|千赫茲|兆赫茲|" | |
| rf"吉赫茲|赫茲|毫秒|公里|公尺|公分|公厘|" | |
| rf"公斤|公克|毫克|公升|毫升|千瓦|瓦|伏特" | |
| rf")" | |
| rf"|[" + _ZH_DIGITS + rf"]{{4}}年{_NORMALIZED_ZH_NUMBER_PATTERN}月" | |
| rf"{_NORMALIZED_ZH_NUMBER_PATTERN}日" | |
| rf"|{_NORMALIZED_ZH_NUMBER_PATTERN}點(?:整|{_NORMALIZED_ZH_NUMBER_PATTERN}分)" | |
| rf"|版本{_NORMALIZED_ZH_NUMBER_PATTERN}點{_NORMALIZED_ZH_NUMBER_PATTERN}" | |
| rf"點{_NORMALIZED_ZH_NUMBER_PATTERN}" | |
| rf")" | |
| ) | |
| def _semantic_sentence_units(text: str) -> list[str]: | |
| """Return strong-punctuation units, keeping closing quotes with the sentence.""" | |
| units: list[str] = [] | |
| start = 0 | |
| index = 0 | |
| while index < len(text): | |
| if text[index] not in _SEMANTIC_STRONG_BREAKS: | |
| index += 1 | |
| continue | |
| end = index + 1 | |
| while end < len(text) and text[end] in _SEMANTIC_TRAILING_CLOSERS: | |
| end += 1 | |
| unit = text[start:end].strip() | |
| if unit: | |
| units.append(unit) | |
| start = end | |
| index = end | |
| tail = text[start:].strip() | |
| if tail: | |
| units.append(tail) | |
| return units | |
| def _ends_at_strong_boundary(text: str) -> bool: | |
| core = str(text).rstrip("".join(_SEMANTIC_TRAILING_CLOSERS)) | |
| return bool(core and core[-1] in _SEMANTIC_STRONG_BREAKS) | |
| def _expanded_network_split_ranges(text: str) -> tuple[tuple[int, int], ...]: | |
| """Locate frontend-expanded network spans in normalized model text.""" | |
| normalized = normalize_tts_text(text) | |
| records = ( | |
| *_naturalized_url_spoken_records(normalized), | |
| *_expanded_network_spoken_span_records(normalized), | |
| ) | |
| ranges: list[tuple[int, int]] = [] | |
| for start, end, _ in sorted(records, key=lambda item: (item[0], -item[1])): | |
| if any(start < kept_end and end > kept_start for kept_start, kept_end in ranges): | |
| continue | |
| ranges.append((start, end)) | |
| return tuple(ranges) | |
| def _protected_split_offsets(text: str, max_units: int) -> set[int]: | |
| """Return offsets that would bisect one bounded ASCII/network identifier.""" | |
| forbidden: set[int] = set() | |
| for match in _PROTECTED_ASCII_SPAN_RE.finditer(text): | |
| if count_speech_units(match.group(0)) > max_units: | |
| raise ValueError("an indivisible ASCII identifier exceeds the chunk limit") | |
| forbidden.update(range(match.start() + 1, match.end())) | |
| for start, end in _expanded_network_split_ranges(text): | |
| if count_speech_units(text[start:end]) > max_units: | |
| raise ValueError("an indivisible network identifier exceeds the chunk limit") | |
| forbidden.update(range(start + 1, end)) | |
| return forbidden | |
| def _structured_numeric_cue_count(text: str) -> int: | |
| """Count bounded, already-expanded numeric frontend constructions.""" | |
| return sum(1 for _ in _NORMALIZED_STRUCTURED_NUMERIC_CUE_RE.finditer(text)) | |
| def _should_split_structured_clauses(text: str) -> bool: | |
| return bool(network_protected_spoken_spans(text)) or ( | |
| _structured_numeric_cue_count(text) >= 2 | |
| ) | |
| def _structured_clause_units(text: str, max_units: int) -> list[str]: | |
| """Split only on real clause punctuation outside protected identifiers.""" | |
| forbidden = _protected_split_offsets(text, max_units) | |
| clauses: list[str] = [] | |
| start = 0 | |
| for index, character in enumerate(text): | |
| end = index + 1 | |
| if character not in _STRUCTURED_CLAUSE_BREAKS or end in forbidden: | |
| continue | |
| clause = text[start:end].strip() | |
| if clause: | |
| clauses.append(clause) | |
| start = end | |
| tail = text[start:].strip() | |
| if tail: | |
| clauses.append(tail) | |
| return clauses | |
| def _pack_structured_clauses( | |
| clauses: Sequence[str], | |
| *, | |
| max_units: int, | |
| min_units: int, | |
| target_units: int = _STRUCTURED_CLAUSE_TARGET_UNITS, | |
| ) -> list[str]: | |
| """Pack natural clauses near a soft target while retaining hard bounds.""" | |
| maximum = max(1, int(max_units)) | |
| minimum = max(1, min(int(min_units), maximum)) | |
| target = max(minimum, min(int(target_units), maximum)) | |
| atoms: list[str] = [] | |
| for clause in clauses: | |
| normalized = normalize_tts_text(clause) | |
| if not normalized: | |
| continue | |
| if count_speech_units(normalized) > maximum: | |
| atoms.extend(_hard_split_text(normalized, maximum, minimum)) | |
| else: | |
| atoms.append(normalized) | |
| if not atoms: | |
| return [] | |
| # A short opening clause is an onset risk on its own. Merge it forward | |
| # before optimizing the remaining natural boundaries. | |
| while len(atoms) > 1 and count_speech_units(atoms[0]) < minimum: | |
| merged = f"{atoms[0]}{atoms[1]}" | |
| if count_speech_units(merged) > maximum: | |
| return _hard_split_text("".join(atoms), maximum, minimum) | |
| atoms[:2] = [merged] | |
| def build_plan(multi_limit: int) -> tuple[int, int, tuple[int, ...]] | None: | |
| best: list[tuple[int, int, tuple[int, ...]] | None] = [None] * ( | |
| len(atoms) + 1 | |
| ) | |
| best[len(atoms)] = (0, 0, ()) | |
| for start in range(len(atoms) - 1, -1, -1): | |
| for end in range(start + 1, len(atoms) + 1): | |
| chunk = "".join(atoms[start:end]) | |
| units = count_speech_units(chunk) | |
| if units > maximum: | |
| break | |
| if units < minimum: | |
| continue | |
| if units > multi_limit and end - start > 1: | |
| continue | |
| remainder = best[end] | |
| if remainder is None: | |
| continue | |
| candidate = ( | |
| 1 + remainder[0], | |
| abs(target - units) + remainder[1], | |
| (end,) + remainder[2], | |
| ) | |
| if best[start] is None or candidate[:2] < best[start][:2]: | |
| best[start] = candidate | |
| return best[0] | |
| # First hold multi-clause chunks to the 32-unit target. If an interior or | |
| # trailing short clause makes that impossible, permit at most one | |
| # minimum-sized unit of slack; protected single clauses remain indivisible. | |
| plan = build_plan(target) | |
| if plan is None: | |
| plan = build_plan(min(maximum, target + minimum)) | |
| if plan is None: | |
| return _hard_split_text("".join(atoms), maximum, minimum) | |
| output: list[str] = [] | |
| start = 0 | |
| for end in plan[2]: | |
| output.append("".join(atoms[start:end])) | |
| start = end | |
| return output | |
| def _hard_split_text(text: str, max_chars: int, min_chunk_chars: int) -> list[str]: | |
| """Deterministically split one overlong unit by speech units, not codepoints.""" | |
| remaining = text.strip() | |
| maximum = max(1, int(max_chars)) | |
| minimum = max(1, min(int(min_chunk_chars), maximum)) | |
| chunks: list[str] = [] | |
| while count_speech_units(remaining) > maximum: | |
| forbidden = _protected_split_offsets(remaining, maximum) | |
| candidates: list[tuple[int, int, int]] = [] | |
| for offset in range(1, len(remaining)): | |
| if offset in forbidden: | |
| continue | |
| head = remaining[:offset].rstrip() | |
| tail = remaining[offset:].lstrip() | |
| head_units = count_speech_units(head) | |
| tail_units = count_speech_units(tail) | |
| if not minimum <= head_units <= maximum: | |
| continue | |
| if 0 < tail_units < minimum: | |
| continue | |
| semantic = int( | |
| remaining[offset - 1] in _SEMANTIC_SOFT_BREAKS | |
| or remaining[offset].isspace() | |
| ) | |
| # Prefer a real soft boundary, then fill the chunk, then choose the | |
| # earliest raw offset for a deterministic tie break. | |
| candidates.append((semantic, head_units, -offset)) | |
| if not candidates: | |
| raise ValueError("text cannot be split within the speech-unit limits") | |
| _, _, negative_offset = max(candidates) | |
| cut = -negative_offset | |
| chunk = remaining[:cut].strip() | |
| if not chunk: | |
| raise ValueError("text splitter produced an empty chunk") | |
| chunks.append(chunk) | |
| remaining = remaining[cut:].lstrip() | |
| if remaining: | |
| chunks.append(remaining) | |
| return chunks | |
| def split_text_for_tts(text: str, max_chars: int = 80, min_chunk_chars: int = 12) -> list[str]: | |
| """Split at semantic boundaries under speech-unit limits. | |
| ``max_chars`` and ``min_chunk_chars`` retain their public names for API | |
| compatibility, but both limits are measured with ``count_speech_units``. | |
| This prevents expanded URLs and spaced acronyms from forcing a raw-string | |
| cut through an otherwise short spoken sentence. | |
| """ | |
| text = normalize_tts_text(text) | |
| if not text: | |
| return [] | |
| if max_chars <= 0: | |
| return [text] | |
| maximum = max(1, int(max_chars)) | |
| minimum = max(1, min(int(min_chunk_chars), maximum)) | |
| total_units = count_speech_units(text) | |
| sentence_units = _semantic_sentence_units(text) | |
| # Long multi-sentence requests benefit from keeping their natural endpoint | |
| # cues even when the combined text narrowly fits under the hard maximum. | |
| proactive_sentence_split = ( | |
| len(sentence_units) > 1 and total_units > maximum * 0.60 | |
| ) | |
| proactive_clause_split = any( | |
| _should_split_structured_clauses(unit) | |
| and len(_structured_clause_units(unit, maximum)) > 1 | |
| and count_speech_units(unit) > min( | |
| maximum, | |
| _STRUCTURED_CLAUSE_TARGET_UNITS, | |
| ) | |
| for unit in sentence_units | |
| ) | |
| if ( | |
| total_units <= maximum | |
| and not proactive_sentence_split | |
| and not proactive_clause_split | |
| ): | |
| return [text] if text else [] | |
| chunks: list[str] = [] | |
| for unit in sentence_units: | |
| clause_units = _structured_clause_units(unit, maximum) | |
| if ( | |
| _should_split_structured_clauses(unit) | |
| and len(clause_units) > 1 | |
| and count_speech_units(unit) | |
| > min(maximum, _STRUCTURED_CLAUSE_TARGET_UNITS) | |
| ): | |
| pieces = _pack_structured_clauses( | |
| clause_units, | |
| max_units=maximum, | |
| min_units=minimum, | |
| ) | |
| else: | |
| pieces = ( | |
| _hard_split_text(unit, maximum, minimum) | |
| if count_speech_units(unit) > maximum | |
| else [unit] | |
| ) | |
| for piece in pieces: | |
| piece_units = count_speech_units(piece) | |
| if piece_units < minimum and chunks: | |
| merged = f"{chunks[-1]}{piece}" | |
| if count_speech_units(merged) <= maximum: | |
| chunks[-1] = merged | |
| continue | |
| chunks.append(piece) | |
| if len(chunks) > 1 and count_speech_units(chunks[0]) < minimum: | |
| merged = f"{chunks[0]}{chunks[1]}" | |
| if count_speech_units(merged) <= maximum: | |
| chunks[:2] = [merged] | |
| if total_units >= minimum and any( | |
| not minimum <= count_speech_units(chunk) <= maximum | |
| for chunk in chunks | |
| ): | |
| if all( | |
| count_speech_units(chunk) <= maximum | |
| and _ends_at_strong_boundary(chunk) | |
| for chunk in chunks | |
| ): | |
| return chunks | |
| # Repartition the offending neighborhood without silently publishing a | |
| # sub-minimum fragment. The fallback itself remains semantic-first. | |
| return _hard_split_text(text, maximum, minimum) | |
| return chunks or [text] | |
| def split_quality_text_for_tts( | |
| text: str, | |
| *, | |
| long_max_units: int = 80, | |
| min_chunk_units: int = 12, | |
| medium_max_units: int = 48, | |
| medium_chunk_units: int = 24, | |
| ) -> list[str]: | |
| """Use smaller semantic chunks only for bounded punctuated medium text. | |
| This keeps short single-clause requests and established long-form planning | |
| unchanged. Medium text is repartitioned only when it has an internal | |
| punctuation boundary, so the smaller limit cannot introduce an arbitrary | |
| pause into an otherwise continuous phrase. | |
| """ | |
| normalized = normalize_tts_text(text) | |
| if not normalized: | |
| return [] | |
| long_maximum = max(1, int(long_max_units)) | |
| minimum = max(1, min(int(min_chunk_units), long_maximum)) | |
| medium_maximum = max(minimum, min(int(medium_max_units), long_maximum)) | |
| medium_chunk_maximum = max( | |
| minimum, | |
| min(int(medium_chunk_units), medium_maximum), | |
| ) | |
| total_units = count_speech_units(normalized) | |
| has_internal_boundary = any( | |
| character in _QUALITY_MEDIUM_BREAKS | |
| for character in normalized[:-1] | |
| ) | |
| if not ( | |
| medium_chunk_maximum < total_units <= medium_maximum | |
| and has_internal_boundary | |
| ): | |
| return split_text_for_tts( | |
| normalized, | |
| max_chars=long_maximum, | |
| min_chunk_chars=minimum, | |
| ) | |
| # A smaller medium-text target is useful only when the source already has | |
| # a legal semantic seam. Calling the generic hard splitter with the | |
| # smaller bound can cut a structured number (for example, between | |
| # ``十七`` and ``點二五``) merely to satisfy the size target. Build a | |
| # bounded partition from real punctuation offsets instead and fall back to | |
| # the established long bound when no all-minimum partition exists. | |
| boundaries: set[int] = {len(normalized)} | |
| for index, character in enumerate(normalized): | |
| if character not in _QUALITY_MEDIUM_BREAKS: | |
| continue | |
| end = index + 1 | |
| if character in _SEMANTIC_STRONG_BREAKS: | |
| while ( | |
| end < len(normalized) | |
| and normalized[end] in _SEMANTIC_TRAILING_CLOSERS | |
| ): | |
| end += 1 | |
| boundaries.add(end) | |
| ordered = (0, *sorted(boundaries)) | |
| # ``best[i]`` stores (chunk_count, target-distance, raw cut offsets) for | |
| # the best punctuation-only partition beginning at ``ordered[i]``. | |
| best: list[tuple[int, int, tuple[int, ...]] | None] = [None] * len(ordered) | |
| best[-1] = (0, 0, ()) | |
| for start_index in range(len(ordered) - 2, -1, -1): | |
| start = ordered[start_index] | |
| for end_index in range(start_index + 1, len(ordered)): | |
| end = ordered[end_index] | |
| chunk = normalized[start:end].strip() | |
| units = count_speech_units(chunk) | |
| if units > medium_chunk_maximum: | |
| break | |
| if units < minimum: | |
| continue | |
| remainder = best[end_index] | |
| if remainder is None: | |
| continue | |
| candidate = ( | |
| 1 + remainder[0], | |
| abs(medium_chunk_maximum - units) + remainder[1], | |
| (end,) + remainder[2], | |
| ) | |
| if best[start_index] is None or candidate[:2] < best[start_index][:2]: | |
| best[start_index] = candidate | |
| plan = best[0] | |
| if plan is None or len(plan[2]) < 2: | |
| return split_text_for_tts( | |
| normalized, | |
| max_chars=long_maximum, | |
| min_chunk_chars=minimum, | |
| ) | |
| chunks: list[str] = [] | |
| start = 0 | |
| for end in plan[2]: | |
| chunks.append(normalized[start:end].strip()) | |
| start = end | |
| return chunks | |
| class NetworkFragmentProof: | |
| """Range-bound local slice of one planner-proven network identifier. | |
| ``chunk_start``/``chunk_end`` are offsets in the exact generation chunk. | |
| ``parent_start``/``parent_end`` are offsets in ``full_spoken_proof``. The | |
| two slices must be byte-for-byte equal after NFC normalization. Keeping | |
| both coordinate systems prevents a repeated label elsewhere in the chunk | |
| from borrowing a parent URL/email proof. | |
| """ | |
| span_index: int | |
| chunk_start: int | |
| chunk_end: int | |
| parent_start: int | |
| parent_end: int | |
| full_spoken_proof: str | |
| raw_identifier: str = "" | |
| url_rendering_proof: NetworkUrlRenderingProof | None = None | |
| identifier_kind: str = "" | |
| class GenerationChunkSpec: | |
| """One generation-only slice with immutable full-identifier provenance.""" | |
| text: str | |
| source_start: int | |
| source_end: int | |
| network_span_indices: tuple[int, ...] = () | |
| network_component_indices: tuple[tuple[int, int], ...] = () | |
| network_full_spoken_proofs: tuple[str, ...] = () | |
| network_fragment_proofs: tuple[NetworkFragmentProof, ...] = () | |
| boundary_after: str = "none" | |
| def network_conditioned(self) -> bool: | |
| return bool(self.network_span_indices) | |
| _EMAIL_DOMAIN_MAIL_TOKEN_RE = re.compile( | |
| r"(?<![A-Za-z])mail(?![A-Za-z])", | |
| flags=re.IGNORECASE, | |
| ) | |
| def email_domain_mail_generation_variant( | |
| text: str, | |
| fragment_proofs: Sequence[NetworkFragmentProof], | |
| ) -> str: | |
| """Spell a range-proven email-domain ``mail`` token for generation only. | |
| The returned text is an acoustic candidate target, not a new semantic | |
| target. Replacements are permitted only inside a planner-authenticated | |
| email-domain fragment whose parent proof exactly equals the canonical | |
| renderer image of its raw address. Callers must continue to verify audio | |
| against the unmodified canonical text. | |
| """ | |
| source = str(text) | |
| replacements: set[tuple[int, int]] = set() | |
| for proof in tuple(fragment_proofs): | |
| if not isinstance(proof, NetworkFragmentProof): | |
| raise ValueError("email generation variant requires fragment proofs") | |
| if not ( | |
| 0 <= proof.chunk_start <= proof.chunk_end <= len(source) | |
| and 0 | |
| <= proof.parent_start | |
| <= proof.parent_end | |
| <= len(proof.full_spoken_proof) | |
| ): | |
| raise ValueError("email generation fragment proof is out of range") | |
| chunk_fragment = source[proof.chunk_start : proof.chunk_end] | |
| parent_fragment = proof.full_spoken_proof[ | |
| proof.parent_start : proof.parent_end | |
| ] | |
| if chunk_fragment != parent_fragment: | |
| raise ValueError("email generation fragment proof does not bind to text") | |
| if proof.identifier_kind != "email": | |
| continue | |
| canonical_email = proof.full_spoken_proof | |
| if _spoken_email_subspans(canonical_email) != (canonical_email,): | |
| raise ValueError("email generation parent proof is not canonical") | |
| separator = canonical_email.find(" 小老鼠 ") | |
| if separator < 0: | |
| raise ValueError("email generation parent proof lacks a separator") | |
| domain_start = separator + len(" 小老鼠 ") | |
| for match in _EMAIL_DOMAIN_MAIL_TOKEN_RE.finditer(parent_fragment): | |
| parent_match_start = proof.parent_start + match.start() | |
| if parent_match_start < domain_start: | |
| continue | |
| replacements.add( | |
| ( | |
| proof.chunk_start + match.start(), | |
| proof.chunk_start + match.end(), | |
| ) | |
| ) | |
| output = source | |
| for start, end in sorted(replacements, reverse=True): | |
| output = f"{output[:start]}M A I L{output[end:]}" | |
| return output | |
| class _NetworkGenerationSpan: | |
| start: int | |
| end: int | |
| spoken_proof: str | |
| kind: str | |
| component_ranges: tuple[tuple[int, int], ...] | |
| mandatory_cut_offsets: tuple[int, ...] | |
| raw_identifier: str = "" | |
| url_rendering_proof: NetworkUrlRenderingProof | None = None | |
| def _raw_network_generation_matches( | |
| text: str, | |
| ) -> tuple[ | |
| tuple[ | |
| int, | |
| int, | |
| str, | |
| str, | |
| str, | |
| NetworkUrlRenderingProof | None, | |
| ], | |
| ..., | |
| ]: | |
| """Return non-overlapping raw identifiers and their exact spoken proofs.""" | |
| raw = unicodedata.normalize("NFC", str(text or "")) | |
| matches: list[ | |
| tuple[ | |
| int, | |
| int, | |
| str, | |
| str, | |
| str, | |
| NetworkUrlRenderingProof | None, | |
| ] | |
| ] = [] | |
| for match in _SPOKEN_URL_RE.finditer(raw): | |
| value = match.group(0) | |
| if not value.isascii(): | |
| raise ValueError("network generation does not support non-ASCII IRI") | |
| rendering_proof = _naturalized_url_rendering_proof(value) | |
| matches.append( | |
| ( | |
| match.start(), | |
| match.end(), | |
| _zh_url(value), | |
| "url", | |
| value if rendering_proof is not None else "", | |
| rendering_proof, | |
| ) | |
| ) | |
| for match in _SPOKEN_EMAIL_RE.finditer(raw): | |
| value = match.group(0) | |
| if not value.isascii(): | |
| raise ValueError("network generation does not support non-ASCII email") | |
| matches.append( | |
| (match.start(), match.end(), _zh_email(value), "email", "", None) | |
| ) | |
| selected: list[ | |
| tuple[ | |
| int, | |
| int, | |
| str, | |
| str, | |
| str, | |
| NetworkUrlRenderingProof | None, | |
| ] | |
| ] = [] | |
| for start, end, spoken, kind, raw_identifier, rendering_proof in sorted( | |
| matches, | |
| key=lambda item: (item[0], -item[1]), | |
| ): | |
| if any( | |
| start < kept_end and end > kept_start | |
| for kept_start, kept_end, *_ in selected | |
| ): | |
| continue | |
| selected.append( | |
| ( | |
| start, | |
| end, | |
| normalize_tts_text(spoken), | |
| kind, | |
| raw_identifier, | |
| rendering_proof, | |
| ) | |
| ) | |
| return tuple(sorted(selected)) | |
| def _network_component_ranges( | |
| spoken: str, | |
| *, | |
| absolute_start: int, | |
| hard_max_units: int, | |
| url_rendering_proof: NetworkUrlRenderingProof | None = None, | |
| ) -> tuple[tuple[int, int], ...]: | |
| """Split one proven identifier only at audible grammar delimiters.""" | |
| if url_rendering_proof is not None: | |
| if ( | |
| inverse_network_url_rendering(url_rendering_proof) | |
| != url_rendering_proof.raw_identifier | |
| or url_rendering_proof.spoken_text != spoken | |
| ): | |
| raise ValueError("naturalized URL component proof is inconsistent") | |
| if count_speech_units(spoken) > hard_max_units: | |
| raise ValueError( | |
| "an indivisible naturalized URL exceeds the generation limit" | |
| ) | |
| return ((absolute_start, absolute_start + len(spoken)),) | |
| token_matches = tuple(re.finditer(r"\S+", spoken)) | |
| tokens = tuple( | |
| _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(match.group(0), match.group(0)) | |
| for match in token_matches | |
| ) | |
| cuts: set[int] = {0, len(spoken)} | |
| scheme_slashes: set[int] = set() | |
| for index in range(2, len(tokens)): | |
| if tokens[index - 2 : index + 1] == ("冒號", "斜線", "斜線"): | |
| scheme_slashes.update({index - 1, index}) | |
| cuts.add(token_matches[index].end()) | |
| for index, (token, match) in enumerate(zip(tokens, token_matches, strict=True)): | |
| if token in _NETWORK_GENERATION_COMPONENT_AFTER: | |
| cuts.add(match.end()) | |
| if ( | |
| token in _NETWORK_GENERATION_COMPONENT_BEFORE | |
| and index not in scheme_slashes | |
| and not (token == "冒號" and index + 2 < len(tokens) and tokens[index + 1 : index + 3] == ("斜線", "斜線")) | |
| ): | |
| cuts.add(match.start()) | |
| ordered = sorted(cuts) | |
| ranges: list[tuple[int, int]] = [] | |
| for left, right in zip(ordered, ordered[1:]): | |
| component = normalize_tts_text(spoken[left:right]) | |
| if not component: | |
| continue | |
| if count_speech_units(component) > hard_max_units: | |
| raise ValueError("an indivisible network component exceeds the generation limit") | |
| ranges.append((absolute_start + left, absolute_start + right)) | |
| if not ranges: | |
| raise ValueError("network generation proof has no components") | |
| if ranges[0][0] != absolute_start or ranges[-1][1] != absolute_start + len(spoken): | |
| raise ValueError("network generation components do not cover their proof") | |
| return tuple(ranges) | |
| def _network_mandatory_cut_offsets( | |
| spoken: str, | |
| *, | |
| kind: str, | |
| absolute_start: int, | |
| url_rendering_proof: NetworkUrlRenderingProof | None = None, | |
| ) -> tuple[int, ...]: | |
| """Return grammar boundaries that no generation chunk may cross. | |
| Hybrid lexical labels are substantially easier for the model than spelling | |
| every letter, but a whole email/URL still asks one trajectory to preserve a | |
| long opaque identifier. Split email local/domain parts and URL | |
| scheme/remainder parts independently so coverage selection can combine | |
| exact fragments without relaxing whole-output verification. | |
| """ | |
| if url_rendering_proof is not None: | |
| if kind != "url" or url_rendering_proof.spoken_text != spoken: | |
| raise ValueError("naturalized URL mandatory-cut proof is inconsistent") | |
| return () | |
| token_matches = tuple(re.finditer(r"\S+", spoken)) | |
| tokens = tuple( | |
| _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(match.group(0), match.group(0)) | |
| for match in token_matches | |
| ) | |
| offsets: list[int] = [] | |
| if kind == "email": | |
| separators = [ | |
| match.start() | |
| for token, match in zip(tokens, token_matches, strict=True) | |
| if token == "小老鼠" | |
| ] | |
| if len(separators) != 1: | |
| raise ValueError("email generation proof lacks one separator") | |
| offsets.append(separators[0]) | |
| elif kind == "url": | |
| scheme_ends = [ | |
| token_matches[index].end() | |
| for index in range(2, len(tokens)) | |
| if tokens[index - 2 : index + 1] == ("冒號", "斜線", "斜線") | |
| ] | |
| if len(scheme_ends) > 1: | |
| raise ValueError("URL generation proof has ambiguous scheme separators") | |
| offsets.extend(scheme_ends) | |
| else: | |
| raise ValueError("network generation proof has an invalid kind") | |
| return tuple(absolute_start + offset for offset in offsets) | |
| def _network_generation_spans( | |
| raw_text: str, | |
| normalized_text: str, | |
| *, | |
| hard_max_units: int, | |
| ) -> tuple[_NetworkGenerationSpan, ...]: | |
| """Bind raw ASCII grammar to exact offsets in the normalized model text.""" | |
| normalized = normalize_tts_text(normalized_text) | |
| expected = normalize_spoken_forms(raw_text) | |
| if normalized != expected: | |
| raise ValueError("normalized text does not match the raw network request") | |
| raw = unicodedata.normalize("NFC", str(raw_text or "")) | |
| raw_matches = _raw_network_generation_matches(raw) | |
| spans: list[_NetworkGenerationSpan] = [] | |
| previous_raw_end: int | None = None | |
| previous_kind: str | None = None | |
| cursor = 0 | |
| for ( | |
| raw_start, | |
| raw_end, | |
| spoken, | |
| kind, | |
| raw_identifier, | |
| url_rendering_proof, | |
| ) in raw_matches: | |
| # Derive the offset from the exact raw prefix rather than searching for | |
| # a possibly duplicated spoken phrase. The frontend inserts a comma | |
| # on each unpunctuated side of a network identifier; a preceding | |
| # identifier at the end of the prefix needs its deferred suffix comma | |
| # accounted for as well. | |
| normalized_prefix = normalize_spoken_forms(raw[:raw_start]) | |
| boundary_count = 0 | |
| preceding = raw[:raw_start].rstrip()[-1:] | |
| if preceding and preceding not in _NETWORK_BOUNDARY_PUNCTUATION: | |
| boundary_count += 1 | |
| if ( | |
| previous_raw_end is not None | |
| and previous_kind == kind | |
| and not raw[previous_raw_end:raw_start].strip() | |
| ): | |
| following = raw[previous_raw_end:].lstrip()[:1] | |
| if following and following not in _NETWORK_BOUNDARY_PUNCTUATION: | |
| boundary_count += 1 | |
| start = len(normalized_prefix) + boundary_count | |
| if start < cursor or normalized[start : start + len(spoken)] != spoken: | |
| raise ValueError( | |
| "network spoken proof offset does not match normalized text" | |
| ) | |
| end = start + len(spoken) | |
| components = _network_component_ranges( | |
| spoken, | |
| absolute_start=start, | |
| hard_max_units=hard_max_units, | |
| url_rendering_proof=url_rendering_proof, | |
| ) | |
| mandatory_cuts = _network_mandatory_cut_offsets( | |
| spoken, | |
| kind=kind, | |
| absolute_start=start, | |
| url_rendering_proof=url_rendering_proof, | |
| ) | |
| component_boundaries = { | |
| boundary | |
| for component in components | |
| for boundary in component | |
| } | |
| if any( | |
| cut <= start or cut >= end or cut not in component_boundaries | |
| for cut in mandatory_cuts | |
| ): | |
| raise ValueError("mandatory network cut lacks component provenance") | |
| spans.append( | |
| _NetworkGenerationSpan( | |
| start=start, | |
| end=end, | |
| spoken_proof=spoken, | |
| kind=kind, | |
| component_ranges=components, | |
| mandatory_cut_offsets=mandatory_cuts, | |
| raw_identifier=raw_identifier, | |
| url_rendering_proof=url_rendering_proof, | |
| ) | |
| ) | |
| cursor = end | |
| previous_raw_end = raw_end | |
| previous_kind = kind | |
| return tuple(spans) | |
| def _network_generation_group_plan( | |
| text: str, | |
| *, | |
| group_start: int, | |
| group_end: int, | |
| original_boundaries: Sequence[int], | |
| spans: Sequence[_NetworkGenerationSpan], | |
| minimum: int, | |
| network_minimum: int, | |
| target: int, | |
| network_maximum: int, | |
| ordinary_maximum: int, | |
| split_atomic_email_at_separator: bool, | |
| ) -> tuple[tuple[int, int], ...]: | |
| """Deterministically pack one network-bearing sentence near the target.""" | |
| group_spans = tuple( | |
| span for span in spans if span.start < group_end and span.end > group_start | |
| ) | |
| if not group_spans: | |
| return tuple( | |
| zip( | |
| (group_start, *original_boundaries), | |
| (*original_boundaries, group_end), | |
| strict=True, | |
| ) | |
| ) | |
| preferred_cuts: set[int] = { | |
| group_start, | |
| group_end, | |
| *original_boundaries, | |
| } | |
| component_cuts = { | |
| offset | |
| for span in group_spans | |
| for component in span.component_ranges | |
| for offset in component | |
| } | |
| atomic_maximum = min(network_maximum, _NETWORK_ATOMIC_MAX_UNITS) | |
| atomic_spans = tuple( | |
| span | |
| for span in group_spans | |
| if count_speech_units(span.spoken_proof) <= atomic_maximum | |
| ) | |
| splittable_spans = tuple( | |
| span | |
| for span in group_spans | |
| if count_speech_units(span.spoken_proof) > atomic_maximum | |
| ) | |
| splittable_component_cuts = { | |
| offset | |
| for span in splittable_spans | |
| for component in span.component_ranges | |
| for offset in component | |
| } | |
| atomic_email_separator_cuts: set[int] = set() | |
| if split_atomic_email_at_separator and len(group_spans) == 1: | |
| span = group_spans[0] | |
| if ( | |
| span in atomic_spans | |
| and span.kind == "email" | |
| and len(span.mandatory_cut_offsets) == 1 | |
| ): | |
| separator = span.mandatory_cut_offsets[0] | |
| left_units = count_speech_units(text[group_start:separator]) | |
| right_units = count_speech_units(text[separator:group_end]) | |
| prose_units = count_speech_units(text[group_start:span.start]) | |
| domain_units = count_speech_units(text[separator:span.end]) | |
| if ( | |
| separator in component_cuts | |
| and prose_units >= _NETWORK_CONTEXTUAL_EMAIL_MIN_PROSE_UNITS | |
| and domain_units >= network_minimum | |
| and network_minimum <= left_units <= network_maximum | |
| and network_minimum <= right_units <= network_maximum | |
| ): | |
| atomic_email_separator_cuts.add(separator) | |
| legal_network_component_cuts = ( | |
| splittable_component_cuts | atomic_email_separator_cuts | |
| ) | |
| preferred_cuts.update(legal_network_component_cuts) | |
| mandatory_cuts = { | |
| offset | |
| for span in splittable_spans | |
| for offset in span.mandatory_cut_offsets | |
| } | atomic_email_separator_cuts | |
| if not mandatory_cuts.issubset(component_cuts): | |
| raise ValueError("mandatory network cuts lack component provenance") | |
| atomic_url_precomma_cuts = { | |
| span.end | |
| for span in atomic_spans | |
| if ( | |
| span.kind == "url" | |
| and span.end < group_end | |
| and text[span.end] in {"，", ","} | |
| ) | |
| } | |
| preferred_cuts.difference_update(atomic_url_precomma_cuts) | |
| network_interiors = { | |
| offset | |
| for span in group_spans | |
| for offset in range(span.start + 1, span.end) | |
| } | |
| for index in range(group_start, group_end): | |
| end = index + 1 | |
| if text[index] in _STRUCTURED_CLAUSE_BREAKS and end not in network_interiors: | |
| preferred_cuts.add(end) | |
| # Every source offset outside an identifier remains an emergency prose | |
| # split. Inside an identifier, only exact component boundaries are legal. | |
| cuts = { | |
| offset | |
| for offset in range(group_start, group_end + 1) | |
| if ( | |
| offset not in atomic_url_precomma_cuts | |
| and ( | |
| offset not in network_interiors | |
| or offset in legal_network_component_cuts | |
| ) | |
| ) | |
| } | |
| ordered = sorted(offset for offset in cuts if group_start <= offset <= group_end) | |
| network_span_starts = {span.start for span in group_spans} | |
| def solve( | |
| active_mandatory_cuts: set[int], | |
| ) -> tuple[int, int, int, int, int, tuple[int, ...]] | None: | |
| best: dict[ | |
| int, | |
| tuple[int, int, int, int, int, tuple[int, ...]], | |
| ] = { | |
| group_end: (0, 0, 0, 0, 0, ()) | |
| } | |
| for start in reversed(ordered[:-1]): | |
| selected: ( | |
| tuple[int, int, int, int, int, tuple[int, ...]] | None | |
| ) = None | |
| for end in ordered: | |
| if end <= start: | |
| continue | |
| if any(start < cut < end for cut in active_mandatory_cuts): | |
| continue | |
| chunk = text[start:end] | |
| units = count_speech_units(chunk) | |
| network_conditioned = any( | |
| span.start < end and span.end > start for span in group_spans | |
| ) | |
| maximum = ( | |
| network_maximum if network_conditioned else ordinary_maximum | |
| ) | |
| effective_minimum = ( | |
| network_minimum if network_conditioned else minimum | |
| ) | |
| if units > maximum: | |
| continue | |
| if units < effective_minimum and not ( | |
| start == group_start and end == group_end | |
| ): | |
| continue | |
| remainder = best.get(end) | |
| if remainder is None: | |
| continue | |
| candidate = ( | |
| 1 + remainder[0], | |
| int(end != group_end and end not in preferred_cuts) | |
| + remainder[1], | |
| int(start in network_span_starts) + remainder[2], | |
| max(units, remainder[3]), | |
| abs(target - units) + remainder[4], | |
| (end,) + remainder[5], | |
| ) | |
| if selected is None or candidate < selected: | |
| selected = candidate | |
| if selected is not None: | |
| best[start] = selected | |
| return best.get(group_start) | |
| plan = solve(mandatory_cuts) | |
| if plan is None: | |
| # A tiny common identifier such as ``https://a.tw`` cannot place both | |
| # sides of the preferred scheme cut above the network minimum. Retry | |
| # only after proving which identifier-local side is intrinsically short; | |
| # all other mandatory cuts and every grammar component boundary remain. | |
| short_identifier_cuts = { | |
| cut | |
| for span in group_spans | |
| for cut in span.mandatory_cut_offsets | |
| if count_speech_units(text[span.start:cut]) < network_minimum | |
| or count_speech_units(text[cut:span.end]) < network_minimum | |
| } | |
| if short_identifier_cuts: | |
| plan = solve(mandatory_cuts - short_identifier_cuts) | |
| if plan is None: | |
| raise ValueError("network-bearing sentence cannot satisfy generation limits") | |
| output: list[tuple[int, int]] = [] | |
| start = group_start | |
| for end in plan[5]: | |
| output.append((start, end)) | |
| start = end | |
| return tuple(output) | |
| def _semantic_sentence_ranges(text: str) -> tuple[tuple[int, int], ...]: | |
| """Return exact, contiguous strong-sentence source ranges.""" | |
| if not text: | |
| return () | |
| ranges: list[tuple[int, int]] = [] | |
| start = 0 | |
| index = 0 | |
| while index < len(text): | |
| if text[index] not in _SEMANTIC_STRONG_BREAKS: | |
| index += 1 | |
| continue | |
| end = index + 1 | |
| while end < len(text) and text[end] in _SEMANTIC_TRAILING_CLOSERS: | |
| end += 1 | |
| while end < len(text) and text[end].isspace(): | |
| end += 1 | |
| if count_speech_units(text[start:end]) > 0: | |
| ranges.append((start, end)) | |
| start = end | |
| index = end | |
| if start < len(text) and count_speech_units(text[start:]) > 0: | |
| ranges.append((start, len(text))) | |
| if not ranges or ranges[0][0] != 0 or ranges[-1][1] != len(text): | |
| raise ValueError("semantic source ranges do not cover normalized text") | |
| if any( | |
| left_end != right_start | |
| for (_, left_end), (right_start, _) in zip(ranges, ranges[1:]) | |
| ): | |
| raise ValueError("semantic source ranges are not contiguous") | |
| return tuple(ranges) | |
| def _ordinary_generation_ranges( | |
| text: str, | |
| *, | |
| start: int, | |
| end: int, | |
| minimum: int, | |
| maximum: int, | |
| ) -> tuple[tuple[int, int], ...]: | |
| """Hard-split one exact non-network source range without re-normalizing it.""" | |
| source = text[start:end] | |
| semantic_chunks = split_text_for_tts( | |
| source, | |
| max_chars=maximum, | |
| min_chunk_chars=minimum, | |
| ) | |
| if semantic_chunks and "".join(semantic_chunks) == source: | |
| ranges: list[tuple[int, int]] = [] | |
| cursor = start | |
| for chunk in semantic_chunks: | |
| next_cursor = cursor + len(chunk) | |
| ranges.append((cursor, next_cursor)) | |
| cursor = next_cursor | |
| if cursor == end: | |
| return tuple(ranges) | |
| if count_speech_units(source) <= maximum: | |
| return ((start, end),) | |
| output: list[tuple[int, int]] = [] | |
| cursor = start | |
| while count_speech_units(text[cursor:end]) > maximum: | |
| local = text[cursor:end] | |
| forbidden = _protected_split_offsets(local, maximum) | |
| candidates: list[tuple[int, int, int]] = [] | |
| for local_offset in range(1, len(local)): | |
| if local_offset in forbidden: | |
| continue | |
| head_units = count_speech_units(local[:local_offset]) | |
| tail_units = count_speech_units(local[local_offset:]) | |
| if not minimum <= head_units <= maximum: | |
| continue | |
| if 0 < tail_units < minimum: | |
| continue | |
| semantic = int( | |
| local[local_offset - 1] in _SEMANTIC_SOFT_BREAKS | |
| or local[local_offset].isspace() | |
| ) | |
| candidates.append((semantic, head_units, -local_offset)) | |
| if not candidates: | |
| raise ValueError("ordinary generation range cannot satisfy chunk limits") | |
| _, _, negative_offset = max(candidates) | |
| next_cursor = cursor - negative_offset | |
| output.append((cursor, next_cursor)) | |
| cursor = next_cursor | |
| output.append((cursor, end)) | |
| return tuple(output) | |
| def plan_generation_chunks( | |
| raw_text: str, | |
| normalized_text: str, | |
| *, | |
| min_units: int = 12, | |
| network_min_units: int = 8, | |
| target_units: int = 32, | |
| network_max_units: int = 36, | |
| ordinary_max_units: int = 80, | |
| split_atomic_email_at_separator: bool = False, | |
| ) -> tuple[GenerationChunkSpec, ...]: | |
| """Plan model chunks while retaining exact whole-network verification proof. | |
| Ordinary semantic chunking is unchanged. Only sentences that contain a raw | |
| ASCII URL/email are repacked at proven audible component delimiters. When | |
| ``split_atomic_email_at_separator`` is enabled, an otherwise bounded email | |
| is split only at its range-proven ``@``/``小老鼠`` grammar boundary when | |
| both generation rows can satisfy the network minimum. The caller must | |
| continue to verify the untouched ``normalized_text`` as one whole target | |
| before returning audio. | |
| """ | |
| if network_identifier_has_ambiguous_iri(raw_text): | |
| raise ValueError("network generation does not support non-ASCII IRI") | |
| minimum = max(1, int(min_units)) | |
| network_minimum = max(1, int(network_min_units)) | |
| target = max(minimum, int(target_units)) | |
| network_maximum = max(minimum, int(network_max_units)) | |
| ordinary_maximum = max(minimum, int(ordinary_max_units)) | |
| if not isinstance(split_atomic_email_at_separator, bool): | |
| raise ValueError("atomic email split policy must be boolean") | |
| if not ( | |
| network_minimum | |
| <= minimum | |
| <= target | |
| <= network_maximum | |
| <= ordinary_maximum | |
| ): | |
| raise ValueError("generation chunk limits are inconsistent") | |
| normalized = normalize_tts_text(normalized_text) | |
| if not contains_network_identifier(raw_text): | |
| ordinary = split_text_for_tts( | |
| normalized, | |
| max_chars=ordinary_maximum, | |
| min_chunk_chars=minimum, | |
| ) | |
| specs: list[GenerationChunkSpec] = [] | |
| cursor = 0 | |
| for index, chunk in enumerate(ordinary): | |
| start = normalized.find(chunk, cursor) | |
| if start < 0: | |
| raise ValueError("ordinary chunk is absent from normalized text") | |
| end = start + len(chunk) | |
| specs.append( | |
| GenerationChunkSpec( | |
| text=chunk, | |
| source_start=start, | |
| source_end=end, | |
| boundary_after=("semantic" if index + 1 < len(ordinary) else "none"), | |
| ) | |
| ) | |
| cursor = end | |
| return tuple(specs) | |
| spans = _network_generation_spans( | |
| raw_text, | |
| normalized, | |
| hard_max_units=network_maximum, | |
| ) | |
| if not spans: | |
| raise ValueError("network request has no bound spoken proof") | |
| planned_ranges: list[tuple[int, int]] = [] | |
| for sentence_start, sentence_end in _semantic_sentence_ranges(normalized): | |
| sentence_has_network = any( | |
| span.start < sentence_end and span.end > sentence_start | |
| for span in spans | |
| ) | |
| if sentence_has_network: | |
| planned_ranges.extend( | |
| _network_generation_group_plan( | |
| normalized, | |
| group_start=sentence_start, | |
| group_end=sentence_end, | |
| original_boundaries=(), | |
| spans=spans, | |
| minimum=minimum, | |
| network_minimum=network_minimum, | |
| target=target, | |
| network_maximum=network_maximum, | |
| ordinary_maximum=ordinary_maximum, | |
| split_atomic_email_at_separator=( | |
| split_atomic_email_at_separator | |
| ), | |
| ) | |
| ) | |
| else: | |
| planned_ranges.extend( | |
| _ordinary_generation_ranges( | |
| normalized, | |
| start=sentence_start, | |
| end=sentence_end, | |
| minimum=minimum, | |
| maximum=ordinary_maximum, | |
| ) | |
| ) | |
| if ( | |
| not planned_ranges | |
| or planned_ranges[0][0] != 0 | |
| or planned_ranges[-1][1] != len(normalized) | |
| or any( | |
| left_end != right_start | |
| for (_, left_end), (right_start, _) in zip( | |
| planned_ranges, | |
| planned_ranges[1:], | |
| ) | |
| ) | |
| ): | |
| raise ValueError("generation source ranges do not exactly cover the target") | |
| specs = [] | |
| internal_boundaries = { | |
| right | |
| for span in spans | |
| for _, right in span.component_ranges[:-1] | |
| } | |
| for start, end in planned_ranges: | |
| chunk = normalized[start:end] | |
| if not chunk or count_speech_units(chunk) <= 0: | |
| raise ValueError("generation planner produced an empty chunk") | |
| overlapping = tuple( | |
| span_index | |
| for span_index, span in enumerate(spans) | |
| if span.start < end and span.end > start | |
| ) | |
| component_indices = tuple( | |
| (span_index, component_index) | |
| for span_index in overlapping | |
| for component_index, (left, right) in enumerate( | |
| spans[span_index].component_ranges | |
| ) | |
| if left < end and right > start | |
| ) | |
| fragment_proofs = tuple( | |
| NetworkFragmentProof( | |
| span_index=span_index, | |
| chunk_start=max(start, spans[span_index].start) - start, | |
| chunk_end=min(end, spans[span_index].end) - start, | |
| parent_start=max(start, spans[span_index].start) | |
| - spans[span_index].start, | |
| parent_end=min(end, spans[span_index].end) | |
| - spans[span_index].start, | |
| full_spoken_proof=spans[span_index].spoken_proof, | |
| raw_identifier=spans[span_index].raw_identifier, | |
| url_rendering_proof=spans[span_index].url_rendering_proof, | |
| identifier_kind=spans[span_index].kind, | |
| ) | |
| for span_index in overlapping | |
| ) | |
| specs.append( | |
| GenerationChunkSpec( | |
| text=chunk, | |
| source_start=start, | |
| source_end=end, | |
| network_span_indices=overlapping, | |
| network_component_indices=component_indices, | |
| network_full_spoken_proofs=tuple( | |
| spans[span_index].spoken_proof for span_index in overlapping | |
| ), | |
| network_fragment_proofs=fragment_proofs, | |
| boundary_after=( | |
| "network_internal" | |
| if end in internal_boundaries | |
| else ("semantic" if end < len(normalized) else "none") | |
| ), | |
| ) | |
| ) | |
| if "".join(spec.text for spec in specs) != normalized: | |
| raise ValueError("generation chunks do not reconstruct the normalized target") | |
| if any( | |
| count_speech_units(spec.text) | |
| > (network_maximum if spec.network_conditioned else ordinary_maximum) | |
| for spec in specs | |
| ): | |
| raise ValueError("generation chunk exceeds its provenance-specific limit") | |
| for spec in specs: | |
| proofs = spec.network_fragment_proofs | |
| if spec.network_conditioned: | |
| if not ( | |
| len(proofs) | |
| == len(spec.network_span_indices) | |
| == len(spec.network_full_spoken_proofs) | |
| ): | |
| raise ValueError("network fragment provenance is incomplete") | |
| elif proofs or spec.network_full_spoken_proofs: | |
| raise ValueError("ordinary chunk carries network fragment provenance") | |
| previous_chunk_end = 0 | |
| for span_index, full_proof, proof in zip( | |
| spec.network_span_indices, | |
| spec.network_full_spoken_proofs, | |
| proofs, | |
| strict=True, | |
| ): | |
| if ( | |
| proof.span_index != span_index | |
| or proof.full_spoken_proof != full_proof | |
| or not 0 <= proof.chunk_start < proof.chunk_end <= len(spec.text) | |
| or not 0 <= proof.parent_start < proof.parent_end <= len(full_proof) | |
| or proof.chunk_start < previous_chunk_end | |
| or spec.text[proof.chunk_start : proof.chunk_end] | |
| != full_proof[proof.parent_start : proof.parent_end] | |
| ): | |
| raise ValueError("network fragment provenance is inconsistent") | |
| spoken_rendering = _naturalized_url_proof_from_spoken(full_proof) | |
| if spoken_rendering is not None: | |
| fresh_rendering = _naturalized_url_rendering_proof( | |
| proof.raw_identifier | |
| ) | |
| if ( | |
| not proof.raw_identifier | |
| or fresh_rendering is None | |
| or spoken_rendering != fresh_rendering | |
| or proof.url_rendering_proof != fresh_rendering | |
| or inverse_network_url_rendering(proof.url_rendering_proof) | |
| != proof.raw_identifier | |
| or proof.url_rendering_proof.spoken_text != full_proof | |
| or proof.parent_start != 0 | |
| or proof.parent_end != len(full_proof) | |
| ): | |
| raise ValueError( | |
| "naturalized URL fragment provenance is inconsistent" | |
| ) | |
| elif proof.raw_identifier or proof.url_rendering_proof is not None: | |
| raise ValueError( | |
| "non-naturalized network proof carries URL rendering provenance" | |
| ) | |
| previous_chunk_end = proof.chunk_end | |
| # The local fragments are not merely hints: taken across the plan they must | |
| # form an exact, non-overlapping partition of every parent identifier. | |
| for span_index, span in enumerate(spans): | |
| partition = sorted( | |
| ( | |
| proof.parent_start, | |
| proof.parent_end, | |
| proof.full_spoken_proof, | |
| ) | |
| for spec in specs | |
| for proof in spec.network_fragment_proofs | |
| if proof.span_index == span_index | |
| ) | |
| if ( | |
| not partition | |
| or partition[0][0] != 0 | |
| or partition[-1][1] != len(span.spoken_proof) | |
| or any( | |
| left_end != right_start | |
| for (_, left_end, _), (right_start, _, _) in zip( | |
| partition, | |
| partition[1:], | |
| ) | |
| ) | |
| or any(full_proof != span.spoken_proof for _, _, full_proof in partition) | |
| ): | |
| raise ValueError("network fragment provenance does not partition its parent") | |
| return tuple(specs) | |
| def coalesce_text_chunks( | |
| chunks: Sequence[str], | |
| *, | |
| max_chunks: int, | |
| max_units: int, | |
| ) -> list[str]: | |
| """Merge the smallest adjacent pairs until a runtime chunk cap is met.""" | |
| if isinstance(max_chunks, bool) or isinstance(max_units, bool): | |
| raise ValueError("chunk limits must be positive integers") | |
| limit = int(max_chunks) | |
| maximum = int(max_units) | |
| if limit <= 0 or maximum <= 0: | |
| raise ValueError("chunk limits must be positive integers") | |
| output = [normalize_tts_text(chunk) for chunk in chunks] | |
| if not output or any(not chunk for chunk in output): | |
| raise ValueError("chunks must be non-empty") | |
| while len(output) > limit: | |
| candidates: list[tuple[int, int]] = [] | |
| for index in range(len(output) - 1): | |
| merged = f"{output[index]}{output[index + 1]}" | |
| units = count_speech_units(merged) | |
| if units <= maximum: | |
| candidates.append((units, index)) | |
| if not candidates: | |
| raise ValueError("text cannot fit within the generated-chunk budget") | |
| _, index = min(candidates) | |
| output[index : index + 2] = [f"{output[index]}{output[index + 1]}"] | |
| return output | |
| def split_leading_clause( | |
| text: str, | |
| *, | |
| search_chars: int = 40, | |
| min_chunk_chars: int = 12, | |
| ) -> list[str]: | |
| """Split the onset only at a real punctuation boundary. | |
| A hard onset split gives the stop head a text fragment with no endpoint | |
| cue. That can turn the artificial chunk tail into extra speech. Leave the | |
| utterance intact when no suitable punctuation exists. | |
| """ | |
| normalized = normalize_tts_text(text) | |
| if not normalized: | |
| return [] | |
| minimum = max(1, int(min_chunk_chars)) | |
| upper = min(max(0, int(search_chars)), len(normalized) - minimum) | |
| if upper < minimum: | |
| return [normalized] | |
| for index, char in enumerate(normalized[:upper], start=1): | |
| if index >= minimum and char in "，,、；;：:。！？!?": | |
| return [normalized[:index], normalized[index:]] | |
| return [normalized] | |
| def punctuation_pause_seconds(text: str, fallback: float = 0.25) -> float: | |
| stripped = text.rstrip() | |
| if not stripped: | |
| return max(0.0, float(fallback)) | |
| if stripped[-1] in "，,、": | |
| return 0.15 | |
| if stripped[-1] in "；;：:": | |
| return 0.23 | |
| if stripped[-1] in "。！？.!?": | |
| return 0.35 | |
| return max(0.0, float(fallback)) | |
| def _peak_rms(audio: np.ndarray) -> tuple[float, float]: | |
| audio = np.asarray(audio, dtype=np.float32).reshape(-1) | |
| peak = float(np.max(np.abs(audio))) if audio.size else 0.0 | |
| rms = float(np.sqrt(np.mean(np.square(audio, dtype=np.float64)))) if audio.size else 0.0 | |
| return peak, rms | |
| def encode_pcm16(audio: np.ndarray) -> np.ndarray: | |
| """Encode finite mono audio with the standard signed PCM16 scale.""" | |
| values = np.asarray(audio) | |
| if values.ndim != 1 or values.size <= 0 or not np.issubdtype( | |
| values.dtype, | |
| np.floating, | |
| ): | |
| raise ValueError("audio must be a non-empty mono floating-point array") | |
| finite = values.astype(np.float64, copy=False) | |
| if not np.isfinite(finite).all(): | |
| raise ValueError("audio must contain only finite samples") | |
| if float(np.max(np.abs(finite))) > 1.0: | |
| raise ValueError("audio samples must be bounded to [-1, 1]") | |
| scaled = np.rint(finite * 32768.0) | |
| return np.clip(scaled, -32768.0, 32767.0).astype(np.int16) | |
| def decode_pcm16(pcm: np.ndarray) -> np.ndarray: | |
| """Decode one mono PCM16 buffer exactly as a WAV reader does.""" | |
| values = np.asarray(pcm) | |
| if values.ndim != 1 or values.size <= 0 or values.dtype != np.int16: | |
| raise ValueError("pcm must be a non-empty mono int16 array") | |
| return values.astype(np.float32) / np.float32(32768.0) | |
| def pcm16_verification_waveform(audio: np.ndarray) -> np.ndarray: | |
| """Return the decoder-equivalent waveform for the exact public PCM codes.""" | |
| return decode_pcm16(encode_pcm16(audio)) | |
| def pcm16_audio_output( | |
| sample_rate: int, | |
| audio: np.ndarray, | |
| ) -> tuple[int, np.ndarray]: | |
| """Serialize verified mono float audio as PCM16 without peak normalization. | |
| Gradio peak-normalizes floating-point ndarray outputs before writing them. | |
| Returning explicit PCM16 preserves the level already selected by the | |
| production loudness/peak limiter, including its intended noise headroom. | |
| Callers should verify ``pcm16_verification_waveform(audio)``; encoding that | |
| canonical waveform here is sample-for-sample idempotent. | |
| """ | |
| if isinstance(sample_rate, (bool, np.bool_)): | |
| raise ValueError("sample_rate must be a positive integer") | |
| try: | |
| rate_hz = operator.index(sample_rate) | |
| except (TypeError, ValueError, OverflowError) as error: | |
| raise ValueError("sample_rate must be a positive integer") from error | |
| if rate_hz <= 0: | |
| raise ValueError("sample_rate must be a positive integer") | |
| return int(rate_hz), encode_pcm16(audio) | |
| def match_chunk_rms(reference: np.ndarray, chunk: np.ndarray, max_adjust_db: float = 4.0) -> np.ndarray: | |
| output = np.asarray(chunk, dtype=np.float32).reshape(-1).copy() | |
| reference_peak, reference_rms = _peak_rms(reference) | |
| chunk_peak, chunk_rms = _peak_rms(output) | |
| del reference_peak | |
| if output.size == 0 or reference_rms <= 0.0 or chunk_rms <= 0.0: | |
| return output | |
| bound = 10.0 ** (max(0.0, float(max_adjust_db)) / 20.0) | |
| gain = float(np.clip(reference_rms / chunk_rms, 1.0 / bound, bound)) | |
| if chunk_peak > 0.0: | |
| gain = min(gain, 0.95 / chunk_peak) | |
| output *= gain | |
| return output | |
| def fade_internal_edges(chunks: list[np.ndarray], sample_rate: int, fade_ms: float = 80.0) -> list[np.ndarray]: | |
| outputs = [np.asarray(chunk, dtype=np.float32).reshape(-1).copy() for chunk in chunks] | |
| requested = max(0, int(round(float(fade_ms) * sample_rate / 1000.0))) | |
| if requested <= 0 or len(outputs) < 2: | |
| return outputs | |
| for index, output in enumerate(outputs): | |
| count = min(requested, output.size) | |
| if index > 0: | |
| output[:count] *= np.linspace(0.0, 1.0, count, endpoint=True, dtype=np.float32) | |
| if index + 1 < len(outputs): | |
| output[-count:] *= np.linspace(1.0, 0.0, count, endpoint=True, dtype=np.float32) | |
| return outputs | |
| def fade_variable_internal_edges( | |
| chunks: Sequence[np.ndarray], | |
| sample_rate: int, | |
| fade_ms_by_boundary: Sequence[float], | |
| ) -> list[np.ndarray]: | |
| """Apply independently bounded fades at each adjacent chunk boundary.""" | |
| outputs = [np.asarray(chunk, dtype=np.float32).reshape(-1).copy() for chunk in chunks] | |
| if len(fade_ms_by_boundary) != max(0, len(outputs) - 1): | |
| raise ValueError("fade boundary count must equal chunk count minus one") | |
| if sample_rate <= 0: | |
| raise ValueError("sample_rate must be positive") | |
| if any(output.size <= 0 for output in outputs): | |
| raise ValueError("audio chunks must be non-empty") | |
| widths: list[int] = [] | |
| for index, raw_fade_ms in enumerate(fade_ms_by_boundary): | |
| try: | |
| fade_ms = float(raw_fade_ms) | |
| except (TypeError, ValueError, OverflowError) as error: | |
| raise ValueError("fade duration must be finite and non-negative") from error | |
| if not math.isfinite(fade_ms) or fade_ms < 0.0: | |
| raise ValueError("fade duration must be finite and non-negative") | |
| requested = int(round(fade_ms * sample_rate / 1000.0)) | |
| widths.append( | |
| min(requested, outputs[index].size, outputs[index + 1].size) | |
| ) | |
| if any( | |
| widths[index - 1] + widths[index] > outputs[index].size | |
| for index in range(1, len(outputs) - 1) | |
| ): | |
| raise ValueError("adjacent fades overlap inside an audio chunk") | |
| for index, count in enumerate(widths): | |
| if count <= 0: | |
| continue | |
| outputs[index][-count:] *= np.linspace( | |
| 1.0, | |
| 0.0, | |
| count, | |
| endpoint=True, | |
| dtype=np.float32, | |
| ) | |
| outputs[index + 1][:count] *= np.linspace( | |
| 0.0, | |
| 1.0, | |
| count, | |
| endpoint=True, | |
| dtype=np.float32, | |
| ) | |
| return outputs | |
| def join_audio_chunks( | |
| chunks: list[np.ndarray], | |
| pauses: list[int], | |
| crossfade_samples: int = 0, | |
| *, | |
| pre_faded_edges: bool = False, | |
| ) -> np.ndarray: | |
| if not chunks: | |
| return np.zeros(0, dtype=np.float32) | |
| output = np.asarray(chunks[0], dtype=np.float32).copy() | |
| for index, chunk in enumerate(chunks[1:]): | |
| next_chunk = np.asarray(chunk, dtype=np.float32).copy() | |
| pause = max(0, int(pauses[index] if index < len(pauses) else 0)) | |
| crossfade = min(max(0, int(crossfade_samples)), output.size, next_chunk.size) | |
| if pause > 0: | |
| if crossfade > 0 and not pre_faded_edges: | |
| output[-crossfade:] *= np.linspace(1.0, 0.0, crossfade, dtype=np.float32) | |
| next_chunk[:crossfade] *= np.linspace(0.0, 1.0, crossfade, dtype=np.float32) | |
| output = np.concatenate((output, np.zeros(pause, dtype=np.float32), next_chunk)) | |
| elif crossfade > 0: | |
| if pre_faded_edges: | |
| overlap = output[-crossfade:] + next_chunk[:crossfade] | |
| else: | |
| fade_out = np.linspace(1.0, 0.0, crossfade, endpoint=False, dtype=np.float32) | |
| overlap = output[-crossfade:] * fade_out + next_chunk[:crossfade] * (1.0 - fade_out) | |
| output = np.concatenate((output[:-crossfade], overlap, next_chunk[crossfade:])) | |
| else: | |
| output = np.concatenate((output, next_chunk)) | |
| return output.astype(np.float32, copy=False) | |
| def join_audio_chunks_variable( | |
| chunks: Sequence[np.ndarray], | |
| pauses: Sequence[int], | |
| crossfade_samples_by_boundary: Sequence[int], | |
| *, | |
| pre_faded_edges: bool = False, | |
| ) -> np.ndarray: | |
| """Join chunks with one validated crossfade width per boundary.""" | |
| if not chunks: | |
| if pauses or crossfade_samples_by_boundary: | |
| raise ValueError("empty chunks cannot carry boundary metadata") | |
| return np.zeros(0, dtype=np.float32) | |
| boundary_count = len(chunks) - 1 | |
| if len(pauses) != boundary_count or len(crossfade_samples_by_boundary) != boundary_count: | |
| raise ValueError("pause/crossfade boundary counts must equal chunk count minus one") | |
| prepared = [ | |
| np.asarray(chunk, dtype=np.float32).reshape(-1).copy() | |
| for chunk in chunks | |
| ] | |
| if any(chunk.size <= 0 for chunk in prepared): | |
| raise ValueError("audio chunks must be non-empty") | |
| validated_pauses: list[int] = [] | |
| requested_crossfades: list[int] = [] | |
| for raw_pause, raw_crossfade in zip( | |
| pauses, | |
| crossfade_samples_by_boundary, | |
| strict=True, | |
| ): | |
| if isinstance(raw_pause, (bool, np.bool_)) or isinstance( | |
| raw_crossfade, | |
| (bool, np.bool_), | |
| ): | |
| raise ValueError("pause and crossfade widths must be non-negative integers") | |
| try: | |
| pause = operator.index(raw_pause) | |
| crossfade = operator.index(raw_crossfade) | |
| except (TypeError, ValueError, OverflowError) as error: | |
| raise ValueError( | |
| "pause and crossfade widths must be non-negative integers" | |
| ) from error | |
| if pause < 0 or crossfade < 0: | |
| raise ValueError("pause and crossfade widths must be non-negative integers") | |
| validated_pauses.append(int(pause)) | |
| requested_crossfades.append(int(crossfade)) | |
| widths = [ | |
| min(requested, prepared[index].size, prepared[index + 1].size) | |
| for index, requested in enumerate(requested_crossfades) | |
| ] | |
| if any( | |
| widths[index - 1] + widths[index] > prepared[index].size | |
| for index in range(1, len(prepared) - 1) | |
| ): | |
| raise ValueError("adjacent crossfades overlap inside an audio chunk") | |
| output = prepared[0] | |
| for index, next_chunk in enumerate(prepared[1:]): | |
| pause = validated_pauses[index] | |
| crossfade = widths[index] | |
| if pause > 0: | |
| if crossfade > 0 and not pre_faded_edges: | |
| output[-crossfade:] *= np.linspace( | |
| 1.0, | |
| 0.0, | |
| crossfade, | |
| dtype=np.float32, | |
| ) | |
| next_chunk[:crossfade] *= np.linspace( | |
| 0.0, | |
| 1.0, | |
| crossfade, | |
| dtype=np.float32, | |
| ) | |
| output = np.concatenate( | |
| (output, np.zeros(pause, dtype=np.float32), next_chunk) | |
| ) | |
| elif crossfade > 0: | |
| if pre_faded_edges: | |
| overlap = output[-crossfade:] + next_chunk[:crossfade] | |
| else: | |
| fade_out = np.linspace( | |
| 1.0, | |
| 0.0, | |
| crossfade, | |
| endpoint=False, | |
| dtype=np.float32, | |
| ) | |
| overlap = ( | |
| output[-crossfade:] * fade_out | |
| + next_chunk[:crossfade] * (1.0 - fade_out) | |
| ) | |
| output = np.concatenate( | |
| (output[:-crossfade], overlap, next_chunk[crossfade:]) | |
| ) | |
| else: | |
| output = np.concatenate((output, next_chunk)) | |
| return output.astype(np.float32, copy=False) | |
| def apply_loudness_floor( | |
| audio: np.ndarray, | |
| min_rms: float = 0.07, | |
| peak_limit: float = 0.95, | |
| max_gain: float = 3.0, | |
| ) -> np.ndarray: | |
| output = np.asarray(audio, dtype=np.float32).reshape(-1).copy() | |
| peak, rms = _peak_rms(output) | |
| if min_rms > 0.0 and 0.0 < rms < min_rms: | |
| gain = min(float(max_gain), float(min_rms) / rms) | |
| output *= gain | |
| peak *= gain | |
| if peak_limit > 0.0 and peak > peak_limit: | |
| output *= float(peak_limit) / peak | |
| return output | |
| def _comparison_text( | |
| text: str, | |
| *, | |
| locale: str, | |
| target_text: str | None = None, | |
| preserve_network_boundaries: bool = False, | |
| ) -> str: | |
| spoken = ( | |
| normalize_asr_spoken_forms(text, target_text, locale=locale) | |
| if target_text is not None | |
| else normalize_spoken_forms(text, locale=locale) | |
| ) | |
| proof_text = target_text if target_text is not None else text | |
| proof_network_spans = network_protected_spoken_spans(proof_text) | |
| if proof_network_spans: | |
| # The two commas emitted by the naturalized URL renderer are prosodic, | |
| # not lexical URL atoms. Remove only those two expected cue separators | |
| # before boundary provenance is encoded, so Whisper may omit them while | |
| # an extra punctuation mark inside the host/path remains a visible edit. | |
| if preserve_network_boundaries and any( | |
| _naturalized_url_proof_from_spoken(span) is not None | |
| for span in proof_network_spans | |
| ): | |
| spoken = _strip_naturalized_url_prosodic_commas(spoken) | |
| def network_symbol_reading(character: str) -> str: | |
| compatibility_character = unicodedata.normalize("NFKC", character) | |
| reading = ( | |
| _ZH_NETWORK_SYMBOL_READINGS.get(character) | |
| or ( | |
| _ZH_NETWORK_SYMBOL_READINGS.get(compatibility_character) | |
| if len(compatibility_character) == 1 | |
| else None | |
| ) | |
| ) | |
| return reading or f"符號{_zh_digit_sequence(str(ord(character)))}" | |
| def network_digit_key(character: str) -> str | None: | |
| if character in _ZH_DIGITS: | |
| return str(_ZH_DIGITS.index(character)) | |
| try: | |
| return str(unicodedata.decimal(character)) | |
| except (TypeError, ValueError): | |
| return None | |
| def network_boundary_key(value: str) -> str: | |
| canonical = normalize_tts_eval_text(value).casefold() | |
| for reading, letter in sorted( | |
| _ZH_NETWORK_READING_TO_LETTER.items(), | |
| key=lambda item: len(item[0]), | |
| reverse=True, | |
| ): | |
| canonical = canonical.replace(reading, letter) | |
| canonical = _T2S_CONVERTER.convert(canonical) | |
| output: list[str] = [] | |
| for character in canonical: | |
| digit_key = network_digit_key(character) | |
| if digit_key is not None: | |
| output.append(digit_key) | |
| elif character.isalnum(): | |
| output.append(character) | |
| return "".join(output) | |
| proof_network_keys = tuple( | |
| network_boundary_key(span) for span in proof_network_spans | |
| ) | |
| def replace_bounded_symbol_run(match: re.Match[str]) -> str: | |
| left = spoken[: match.start()].rstrip()[-1:] | |
| right = spoken[match.end() :].lstrip()[:1] | |
| symbol_run = "".join( | |
| character | |
| for character in match.group(1) | |
| if not character.isspace() | |
| ) | |
| # The protected-range alignment canonicalizes every delimiter to | |
| # one opaque sentinel later. Keep it symbolic here even when an | |
| # adjacent ASR insertion makes both sides look identifier-like. | |
| if preserve_network_boundaries and symbol_run and all( | |
| character in _NETWORK_BOUNDARY_PUNCTUATION | |
| for character in symbol_run | |
| ): | |
| return match.group(0) | |
| # A frontend delimiter remains outside a protected span even when | |
| # ASR inserts an ordinary letter immediately beside it. Preserve | |
| # that provenance when either side still matches a target network | |
| # span. If the extra letter is between the delimiter and the | |
| # scheme/path, neither side matches and the insertion stays inside | |
| # the protected range below. | |
| if symbol_run and all( | |
| character in _NETWORK_BOUNDARY_PUNCTUATION | |
| for character in symbol_run | |
| ): | |
| left_key = network_boundary_key(spoken[: match.start()]) | |
| right_key = network_boundary_key(spoken[match.end() :]) | |
| if any( | |
| (left_key.endswith(key) or right_key.startswith(key)) | |
| for key in proof_network_keys | |
| if key | |
| ): | |
| return match.group(0) | |
| def is_identifier_side(character: str) -> bool: | |
| if not character: | |
| return False | |
| if character.isascii() and character.isalnum(): | |
| return True | |
| if character.isdecimal() or character in _ZH_DIGITS: | |
| return True | |
| # Preserve non-Han IRI labels while leaving punctuation at a | |
| # Chinese-prose/network boundary as punctuation. The latter | |
| # is inserted deliberately by the frontend and must not turn | |
| # into an audible network token during comparison. | |
| unicode_name = unicodedata.name(character, "") | |
| return character.isalnum() and "CJK UNIFIED IDEOGRAPH" not in unicode_name | |
| if not (is_identifier_side(left) and is_identifier_side(right)): | |
| return match.group(0) | |
| readings = [ | |
| network_symbol_reading(character) | |
| for character in match.group(1) | |
| if not character.isspace() | |
| ] | |
| return f" {' '.join(readings)} " | |
| # Mixed Whisper renderings may put spaces around literal separators, | |
| # use Chinese digits, or insert a symbol not present in the target. | |
| # Canonicalize every punctuation run bounded by Unicode alphanumerics; | |
| # this makes equivalent separators match while keeping extra symbols | |
| # as visible alignment insertions instead of dropping them later. | |
| spoken = re.sub( | |
| r"(?<=[^\W_])\s*((?:[^\w\s]|_)+)\s*(?=[^\W_])", | |
| replace_bounded_symbol_run, | |
| spoken, | |
| ) | |
| output: list[str] = [] | |
| for character in spoken: | |
| digit_key = network_digit_key(character) | |
| if digit_key is not None: | |
| output.append(digit_key) | |
| continue | |
| if ( | |
| character.isspace() | |
| or character.isalnum() | |
| or character in _NETWORK_BOUNDARY_PUNCTUATION | |
| ): | |
| output.append(character) | |
| else: | |
| output.append(f" {network_symbol_reading(character)} ") | |
| spoken = "".join(output) | |
| for reading, letter in sorted( | |
| _ZH_NETWORK_READING_TO_LETTER.items(), | |
| key=lambda item: len(item[0]), | |
| reverse=True, | |
| ): | |
| spoken = spoken.replace(reading, f" {letter} ") | |
| normalized = normalize_tts_eval_text(spoken).casefold() | |
| locale_key = str(locale or "").replace("_", "-").lower() | |
| if locale_key in {"zh", "zh-tw", "zh-hant", "zh-cn", "zh-hans"}: | |
| normalized = _T2S_CONVERTER.convert(normalized) | |
| if not preserve_network_boundaries: | |
| return "".join(char for char in normalized if char.isalnum()) | |
| output: list[str] = [] | |
| for character in normalized: | |
| if character.isalnum(): | |
| output.append(character) | |
| elif character in _NETWORK_BOUNDARY_PUNCTUATION and ( | |
| not output or output[-1] != _NETWORK_ALIGNMENT_BOUNDARY_SENTINEL | |
| ): | |
| output.append(_NETWORK_ALIGNMENT_BOUNDARY_SENTINEL) | |
| return "".join(output) | |
| def mandarin_acoustic_units(normalized_text: str) -> tuple[str, ...]: | |
| """Map normalized Mandarin/ASCII text to deterministic tone-aware units. | |
| Whisper may render an acoustically identical Mandarin syllable with a | |
| different Han character. Endpoint validation should not reject that | |
| spelling choice, but it must still distinguish lexical tone (``藝`` yi4 | |
| from ``一`` yi1) and non-homophones. ASCII identifiers remain character | |
| units so email/URL omissions cannot disappear through phonetic scoring. | |
| """ | |
| readings = lazy_pinyin( | |
| str(normalized_text or ""), | |
| style=Style.TONE3, | |
| errors=lambda fragment: list(fragment), | |
| strict=True, | |
| v_to_u=False, | |
| neutral_tone_with_five=True, | |
| tone_sandhi=False, | |
| ) | |
| return tuple( | |
| reading.casefold() | |
| for reading in readings | |
| if reading and any(character.isalnum() for character in reading) | |
| ) | |
| def _levenshtein_alignment(source: str, hypothesis: str) -> list[tuple[str, int, int]]: | |
| """Return deterministic edit operations as ``(op, source_index, hyp_index)``.""" | |
| rows = len(source) + 1 | |
| columns = len(hypothesis) + 1 | |
| distance = [[0] * columns for _ in range(rows)] | |
| for source_index in range(rows): | |
| distance[source_index][0] = source_index | |
| for hyp_index in range(columns): | |
| distance[0][hyp_index] = hyp_index | |
| for source_index in range(1, rows): | |
| for hyp_index in range(1, columns): | |
| substitution = distance[source_index - 1][hyp_index - 1] + ( | |
| source[source_index - 1] != hypothesis[hyp_index - 1] | |
| ) | |
| deletion = distance[source_index - 1][hyp_index] + 1 | |
| insertion = distance[source_index][hyp_index - 1] + 1 | |
| distance[source_index][hyp_index] = min(substitution, deletion, insertion) | |
| operations: list[tuple[str, int, int]] = [] | |
| source_index = len(source) | |
| hyp_index = len(hypothesis) | |
| while source_index or hyp_index: | |
| if source_index and hyp_index: | |
| cost = source[source_index - 1] != hypothesis[hyp_index - 1] | |
| if distance[source_index][hyp_index] == distance[source_index - 1][hyp_index - 1] + cost: | |
| operations.append( | |
| ( | |
| "substitute" if cost else "equal", | |
| source_index - 1, | |
| hyp_index - 1, | |
| ) | |
| ) | |
| source_index -= 1 | |
| hyp_index -= 1 | |
| continue | |
| if source_index and distance[source_index][hyp_index] == distance[source_index - 1][hyp_index] + 1: | |
| operations.append(("delete", source_index - 1, hyp_index)) | |
| source_index -= 1 | |
| continue | |
| operations.append(("insert", source_index, hyp_index - 1)) | |
| hyp_index -= 1 | |
| operations.reverse() | |
| return operations | |
| def _protected_ranges_are_exact_in_all_optimal_alignments( | |
| source: str, | |
| hypothesis: str, | |
| protected_ranges: Sequence[tuple[int, int]], | |
| ) -> bool: | |
| """Require protected source ranges to survive every optimal edit path. | |
| A deterministic Levenshtein backtrace is insufficient when ordinary text | |
| duplicates a protected fragment: one optimal path can match the protected | |
| occurrence while another deletes it and matches the ordinary occurrence. | |
| Prefix/suffix distances let us inspect every edge that belongs to at least | |
| one globally optimal path. Protected characters may only traverse equal | |
| diagonal edges, and insertions may not be anchored inside the closed range | |
| (including immediately before its first or after its last character). | |
| """ | |
| ranges = tuple(protected_ranges) | |
| if not ranges: | |
| return True | |
| if any( | |
| isinstance(start, bool) | |
| or isinstance(end, bool) | |
| or not isinstance(start, int) | |
| or not isinstance(end, int) | |
| or not 0 <= start < end <= len(source) | |
| for start, end in ranges | |
| ): | |
| raise ValueError("protected alignment range is invalid") | |
| source_protected = [False] * len(source) | |
| insertion_protected = [False] * (len(source) + 1) | |
| for start, end in ranges: | |
| for source_index in range(start, end): | |
| source_protected[source_index] = True | |
| for source_offset in range(start, end + 1): | |
| insertion_protected[source_offset] = True | |
| source_length = len(source) | |
| hypothesis_length = len(hypothesis) | |
| forward = [ | |
| [0] * (hypothesis_length + 1) for _ in range(source_length + 1) | |
| ] | |
| for source_index in range(source_length + 1): | |
| forward[source_index][0] = source_index | |
| for hypothesis_index in range(hypothesis_length + 1): | |
| forward[0][hypothesis_index] = hypothesis_index | |
| for source_index in range(1, source_length + 1): | |
| for hypothesis_index in range(1, hypothesis_length + 1): | |
| forward[source_index][hypothesis_index] = min( | |
| forward[source_index - 1][hypothesis_index] + 1, | |
| forward[source_index][hypothesis_index - 1] + 1, | |
| forward[source_index - 1][hypothesis_index - 1] | |
| + ( | |
| source[source_index - 1] | |
| != hypothesis[hypothesis_index - 1] | |
| ), | |
| ) | |
| backward = [ | |
| [0] * (hypothesis_length + 1) for _ in range(source_length + 1) | |
| ] | |
| for source_index in range(source_length + 1): | |
| backward[source_index][hypothesis_length] = source_length - source_index | |
| for hypothesis_index in range(hypothesis_length + 1): | |
| backward[source_length][hypothesis_index] = ( | |
| hypothesis_length - hypothesis_index | |
| ) | |
| for source_index in range(source_length - 1, -1, -1): | |
| for hypothesis_index in range(hypothesis_length - 1, -1, -1): | |
| backward[source_index][hypothesis_index] = min( | |
| backward[source_index + 1][hypothesis_index] + 1, | |
| backward[source_index][hypothesis_index + 1] + 1, | |
| backward[source_index + 1][hypothesis_index + 1] | |
| + (source[source_index] != hypothesis[hypothesis_index]), | |
| ) | |
| optimal_distance = forward[source_length][hypothesis_length] | |
| for source_index in range(source_length + 1): | |
| for hypothesis_index in range(hypothesis_length + 1): | |
| prefix_distance = forward[source_index][hypothesis_index] | |
| if ( | |
| source_index < source_length | |
| and source_protected[source_index] | |
| and prefix_distance | |
| + 1 | |
| + backward[source_index + 1][hypothesis_index] | |
| == optimal_distance | |
| ): | |
| return False | |
| if ( | |
| source_index < source_length | |
| and hypothesis_index < hypothesis_length | |
| and source_protected[source_index] | |
| and source[source_index] != hypothesis[hypothesis_index] | |
| and prefix_distance | |
| + 1 | |
| + backward[source_index + 1][hypothesis_index + 1] | |
| == optimal_distance | |
| ): | |
| return False | |
| if ( | |
| hypothesis_index < hypothesis_length | |
| and insertion_protected[source_index] | |
| and prefix_distance | |
| + 1 | |
| + backward[source_index][hypothesis_index + 1] | |
| == optimal_distance | |
| ): | |
| return False | |
| return True | |
| class NetworkFragmentAsrEvidence: | |
| """Result of exact, range-bound local network-fragment verification.""" | |
| transcript_text: str | |
| protected_range_count: int | |
| passed: bool | |
| _NETWORK_FRAGMENT_IDENTIFIER_CHARACTERS = ( | |
| "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789" | |
| + "".join(_ZH_NETWORK_SYMBOL_READINGS) | |
| ) | |
| _NETWORK_FRAGMENT_READING_TO_ASCII = { | |
| reading: symbol for symbol, reading in _ZH_NETWORK_SYMBOL_READINGS.items() | |
| } | |
| def _network_fragment_token_ascii(token: str) -> str | None: | |
| """Invert one typed frontend network atom without prose heuristics.""" | |
| canonical = _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token) | |
| letter = _network_letter_token(canonical) | |
| if letter is not None: | |
| return letter | |
| if canonical.isascii() and canonical.isalnum(): | |
| # A hybrid frontend may intentionally retain a planner-bound opaque | |
| # label (``tour``/``islandmuseum``) while spelling only its separators | |
| # and short suffix. It is still an exact typed atom, never prose: the | |
| # range object below binds its precise occurrence to the parent URL. | |
| return canonical.casefold() | |
| if canonical and all(character in _ZH_DIGITS for character in canonical): | |
| return "".join(str(_ZH_DIGITS.index(character)) for character in canonical) | |
| return _NETWORK_FRAGMENT_READING_TO_ASCII.get(canonical) | |
| def _network_fragment_ascii(spoken_fragment: str) -> str | None: | |
| """Return the exact ASCII atom sequence for one planner-bound fragment.""" | |
| output: list[str] = [] | |
| for token in normalize_tts_text(spoken_fragment).split(): | |
| value = _network_fragment_token_ascii(token) | |
| if value is None: | |
| return None | |
| output.append(value) | |
| rendered = "".join(output) | |
| return rendered if rendered and any(character.isalnum() for character in rendered) else None | |
| def _network_fragment_alignment_text(value: str) -> str: | |
| """Composable strict key used only by the exact local fragment gate. | |
| Punctuation and symbols must remain visible here. Otherwise an unmatched | |
| mutation inside a protected identifier (``islandmu/seum`` or an emoji, | |
| for example) could disappear before the all-optimal alignment check. | |
| """ | |
| normalized = _T2S_CONVERTER.convert( | |
| unicodedata.normalize("NFC", str(value or "")).casefold() | |
| ) | |
| return "".join(character for character in normalized if not character.isspace()) | |
| def _network_fragment_mixed_render_pattern(spoken_fragment: str) -> str | None: | |
| """Build an exact ordered lattice for one planner-bound fragment. | |
| Whisper commonly mixes the frontend's spoken symbol readings with literal | |
| ASCII in one identifier, for example ``小老鼠islandmuseum.tw``. The | |
| fragment proof already fixes the exact parent slice, so each frontend atom | |
| may safely use either its own spoken spelling or its own exact ASCII value. | |
| No atom may be dropped, reordered, substituted, or inferred from prose. | |
| """ | |
| atom_patterns: list[str] = [] | |
| for token in normalize_tts_text(spoken_fragment).split(): | |
| canonical = _SIMPLIFIED_NETWORK_SYMBOL_READINGS.get(token, token) | |
| ascii_atom = _network_fragment_token_ascii(canonical) | |
| if ascii_atom is None: | |
| return None | |
| spoken_variants = {token, canonical} | |
| spoken_variants.update( | |
| alias | |
| for alias, canonical_reading in _SIMPLIFIED_NETWORK_SYMBOL_READINGS.items() | |
| if canonical_reading == canonical | |
| ) | |
| if canonical == "小老鼠": | |
| # Pinned Whisper decoders can romanize the audibly correct email | |
| # separator. These aliases remain one atom in this exact, | |
| # planner-range-bound lattice; the surrounding local part, domain, | |
| # TLD, ordering, and conservative identifier boundaries are still | |
| # mandatory and cannot be inferred from prose. | |
| spoken_variants.update({"xiaolaoshu", "xiaolaoshui"}) | |
| spoken_variants.add(ascii_atom) | |
| variants = sorted( | |
| { | |
| r"\s*".join(re.escape(character) for character in value) | |
| for value in spoken_variants | |
| if value | |
| }, | |
| key=lambda value: (-len(value), value), | |
| ) | |
| atom_patterns.append(f"(?:{'|'.join(variants)})") | |
| return r"\s*".join(atom_patterns) if atom_patterns else None | |
| def _naturalized_url_mixed_render_pattern( | |
| proof: NetworkUrlRenderingProof, | |
| ) -> str: | |
| """Return the exact ASR lattice proven by one natural URL rendering. | |
| Whisper commonly preserves the renderer's lexical cues while formatting | |
| host separators as literal dots and collapsing spelled ASCII atoms, for | |
| example ``安全網址是coastwatch.example.tw路徑是tide網址輸入完畢``. | |
| Every accepted alternative below is bound to one atom in the reversible | |
| raw-to-spoken proof. No atom may be omitted, reordered, substituted, or | |
| inferred from surrounding prose. The fully literal raw URL remains a | |
| separate case-sensitive alternative because URL paths are case-sensitive. | |
| """ | |
| if not isinstance(proof, NetworkUrlRenderingProof): | |
| raise ValueError("naturalized URL mixed pattern requires a rendering proof") | |
| inverse_network_url_rendering(proof) | |
| def spoken_pattern(value: str) -> str: | |
| variants = { | |
| unicodedata.normalize("NFC", value), | |
| _T2S_CONVERTER.convert(unicodedata.normalize("NFC", value)), | |
| } | |
| rendered: list[str] = [] | |
| for variant in sorted(variants, key=lambda item: (-len(item), item)): | |
| pieces: list[str] = [] | |
| whitespace_pending = False | |
| for character in variant: | |
| if character.isspace(): | |
| whitespace_pending = True | |
| continue | |
| if whitespace_pending: | |
| pieces.append(r"\s*") | |
| whitespace_pending = False | |
| if character in {"，", ","}: | |
| # These two renderer-authored commas are prosodic. Their | |
| # owning atom remains mandatory even when ASR omits only | |
| # the pause mark itself. | |
| pieces.append(r"\s*[，,]?\s*") | |
| else: | |
| pieces.append(re.escape(character)) | |
| if whitespace_pending: | |
| pieces.append(r"\s*") | |
| rendered.append("".join(pieces)) | |
| return f"(?:{'|'.join(dict.fromkeys(rendered))})" | |
| atom_patterns: list[str] = [] | |
| for atom in proof.atoms: | |
| spoken = spoken_pattern(atom.spoken_text) | |
| if atom.component == "terminator": | |
| # The raw terminator range is empty, so only its complete spoken | |
| # cue proves semantic completion of a mixed rendering. | |
| atom_patterns.append(spoken) | |
| continue | |
| raw = re.escape(atom.raw_text) | |
| atom_patterns.append(f"(?:{spoken}|{raw})") | |
| mixed = r"\s*".join(atom_patterns) | |
| exact_raw = f"(?-i:{re.escape(proof.raw_identifier)})" | |
| return f"(?:{exact_raw}|{mixed})" | |
| def canonicalize_asr_network_fragments( | |
| text: str, | |
| target_text: str, | |
| fragment_proofs: Sequence[NetworkFragmentProof], | |
| ) -> NetworkFragmentAsrEvidence: | |
| """Canonicalize and exact-gate local ASCII URL/email fragments. | |
| This helper is deliberately separate from :func:`compare_asr_text`. It is | |
| valid only for generation chunks carrying the range objects emitted by | |
| :func:`plan_generation_chunks`; whole/joined/final verification must not | |
| call it. Each literal replacement must equal the *entire* chunk-local | |
| parent slice. Substrings of a full URL cannot borrow its proof. | |
| """ | |
| raw_text = unicodedata.normalize("NFC", str(text or "")) | |
| target = unicodedata.normalize("NFC", str(target_text or "")) | |
| if isinstance(fragment_proofs, (str, bytes)): | |
| raise ValueError("network fragment proofs must be range objects") | |
| try: | |
| proofs = tuple(fragment_proofs) | |
| except TypeError as error: | |
| raise ValueError("network fragment proofs must be iterable") from error | |
| if not proofs: | |
| return NetworkFragmentAsrEvidence(raw_text, 0, True) | |
| if not target: | |
| raise ValueError("network fragment target must be non-empty") | |
| previous_chunk_end = 0 | |
| typed_fragments: dict[str, tuple[str, str]] = {} | |
| has_naturalized_url_proof = False | |
| for proof in proofs: | |
| if not isinstance(proof, NetworkFragmentProof): | |
| raise ValueError("network fragment proof has an invalid type") | |
| integer_fields = ( | |
| proof.span_index, | |
| proof.chunk_start, | |
| proof.chunk_end, | |
| proof.parent_start, | |
| proof.parent_end, | |
| ) | |
| if any(isinstance(value, bool) or not isinstance(value, int) for value in integer_fields): | |
| raise ValueError("network fragment proof offsets must be integers") | |
| full_proof = unicodedata.normalize("NFC", str(proof.full_spoken_proof or "")) | |
| if ( | |
| proof.span_index < 0 | |
| or not 0 <= proof.chunk_start < proof.chunk_end <= len(target) | |
| or not 0 <= proof.parent_start < proof.parent_end <= len(full_proof) | |
| or proof.chunk_start < previous_chunk_end | |
| or target[proof.chunk_start : proof.chunk_end] | |
| != full_proof[proof.parent_start : proof.parent_end] | |
| or full_proof not in network_protected_spoken_spans(full_proof) | |
| ): | |
| raise ValueError("network fragment proof does not bind its target range") | |
| previous_chunk_end = proof.chunk_end | |
| fragment = target[proof.chunk_start : proof.chunk_end] | |
| spoken_rendering = _naturalized_url_proof_from_spoken(full_proof) | |
| if spoken_rendering is not None: | |
| has_naturalized_url_proof = True | |
| fresh_rendering = _naturalized_url_rendering_proof( | |
| proof.raw_identifier | |
| ) | |
| if ( | |
| not proof.raw_identifier | |
| or fresh_rendering is None | |
| or spoken_rendering != fresh_rendering | |
| or proof.url_rendering_proof != fresh_rendering | |
| or inverse_network_url_rendering(proof.url_rendering_proof) | |
| != proof.raw_identifier | |
| or proof.url_rendering_proof.spoken_text != full_proof | |
| or proof.parent_start != 0 | |
| or proof.parent_end != len(full_proof) | |
| or fragment != full_proof | |
| ): | |
| raise ValueError( | |
| "naturalized URL fragment proof is inconsistent" | |
| ) | |
| ascii_fragment = proof.raw_identifier | |
| mixed_pattern = _naturalized_url_mixed_render_pattern( | |
| fresh_rendering | |
| ) | |
| elif proof.raw_identifier or proof.url_rendering_proof is not None: | |
| raise ValueError( | |
| "non-naturalized network proof carries URL rendering provenance" | |
| ) | |
| else: | |
| ascii_fragment = _network_fragment_ascii(fragment) | |
| mixed_pattern = _network_fragment_mixed_render_pattern(fragment) | |
| if ascii_fragment is None or mixed_pattern is None: | |
| # Unmapped atoms remain eligible only in their canonical spoken | |
| # form. A literal approximation is never inferred. | |
| continue | |
| old_fragment = typed_fragments.get(ascii_fragment.casefold()) | |
| if old_fragment is not None and old_fragment[0] != fragment: | |
| raise ValueError("network fragment ASCII proof is ambiguous") | |
| typed_fragments[ascii_fragment.casefold()] = (fragment, mixed_pattern) | |
| # Protect longer exact fragments first, normalize surrounding transcript | |
| # punctuation, then restore their planner-proven spoken forms. A mixed | |
| # render is eligible only at a conservative prose boundary. In particular, | |
| # an unknown adjacent symbol (backslash, angle bracket, emoji, and so on) | |
| # must remain visible rather than letting a global replacement erase its | |
| # provenance before the exact range alignment. | |
| canonicalized = raw_text | |
| unsafe_invisible_character = any( | |
| unicodedata.category(character).startswith("C") | |
| and not character.isspace() | |
| for character in raw_text | |
| ) | |
| placeholder_prefix = "藍鵲片段保護佔位符" | |
| while placeholder_prefix in raw_text or placeholder_prefix in target: | |
| placeholder_prefix += "號" | |
| fragment_placeholders: list[tuple[str, str]] = [] | |
| unsafe_typed_literal = False | |
| ascii_sentence_boundaries = frozenset(",.;:!?\"'()[]{}") | |
| def literal_boundary_kind(value: str, offset: int, direction: int) -> str: | |
| while 0 <= offset < len(value) and value[offset].isspace(): | |
| offset += direction | |
| if offset < 0 or offset >= len(value): | |
| return "safe" | |
| character = value[offset] | |
| if character.isascii() and character.isalnum(): | |
| return "identifier" | |
| if ( | |
| character.isspace() | |
| or _is_cjk(character) | |
| or character in _NETWORK_NONASCII_BOUNDARY_PUNCTUATION | |
| or character in ascii_sentence_boundaries | |
| ): | |
| return "safe" | |
| return "unsafe" | |
| for ascii_fragment, (spoken_fragment, mixed_pattern) in sorted( | |
| typed_fragments.items(), | |
| key=lambda item: (-len(item[0]), item[0]), | |
| ): | |
| pattern = re.compile( | |
| mixed_pattern, | |
| flags=re.IGNORECASE | re.ASCII, | |
| ) | |
| def protect_literal(match: re.Match[str]) -> str: | |
| nonlocal unsafe_typed_literal | |
| boundary_kinds = ( | |
| literal_boundary_kind(match.string, match.start() - 1, -1), | |
| literal_boundary_kind(match.string, match.end(), 1), | |
| ) | |
| if "identifier" in boundary_kinds: | |
| return match.group(0) | |
| if "unsafe" in boundary_kinds: | |
| unsafe_typed_literal = True | |
| return match.group(0) | |
| placeholder = ( | |
| f"{placeholder_prefix}" | |
| f"{_zh_integer(len(fragment_placeholders) + 1)}結束" | |
| ) | |
| fragment_placeholders.append((placeholder, spoken_fragment)) | |
| return placeholder | |
| canonicalized = pattern.sub(protect_literal, canonicalized) | |
| canonicalized = normalize_tts_text(canonicalized) | |
| for placeholder, spoken_fragment in fragment_placeholders: | |
| canonicalized = canonicalized.replace(placeholder, spoken_fragment) | |
| def alignment_key(value: str) -> str: | |
| source = ( | |
| _strip_naturalized_url_prosodic_commas(value) | |
| if has_naturalized_url_proof | |
| else value | |
| ) | |
| return _network_fragment_alignment_text(source) | |
| target_key = alignment_key(target) | |
| transcript_key = alignment_key(canonicalized) | |
| protected_ranges: list[tuple[int, int]] = [] | |
| rebuilt_target_key: list[str] = [] | |
| cursor = 0 | |
| key_cursor = 0 | |
| for proof in proofs: | |
| prefix_key = alignment_key(target[cursor : proof.chunk_start]) | |
| fragment_key = alignment_key( | |
| target[proof.chunk_start : proof.chunk_end] | |
| ) | |
| rebuilt_target_key.extend((prefix_key, fragment_key)) | |
| key_cursor += len(prefix_key) | |
| if not fragment_key: | |
| raise ValueError("network fragment proof has no comparable content") | |
| protected_ranges.append((key_cursor, key_cursor + len(fragment_key))) | |
| key_cursor += len(fragment_key) | |
| cursor = proof.chunk_end | |
| rebuilt_target_key.append(alignment_key(target[cursor:])) | |
| if "".join(rebuilt_target_key) != target_key: | |
| raise ValueError("network fragment comparison ranges are not composable") | |
| passed = ( | |
| not unsafe_typed_literal | |
| and not unsafe_invisible_character | |
| and _protected_ranges_are_exact_in_all_optimal_alignments( | |
| target_key, | |
| transcript_key, | |
| protected_ranges, | |
| ) | |
| ) | |
| return NetworkFragmentAsrEvidence( | |
| transcript_text=canonicalized, | |
| protected_range_count=len(protected_ranges), | |
| passed=bool(passed), | |
| ) | |
| class AsrComparison: | |
| """Orthographic whole-text and acoustic endpoint evidence for one candidate.""" | |
| target_text: str | |
| transcript_text: str | |
| edit_distance: int | |
| cer: float | |
| prefix_cer: float | |
| suffix_cer: float | |
| prefix_deletions: int | |
| suffix_deletions: int | |
| extra_tail: str | |
| extra_tail_units: int | |
| network_protected_spans: int | |
| network_protected_spans_passed: bool | |
| passed: bool | |
| def _network_protected_alignment_evidence( | |
| raw_target: str, | |
| raw_transcript: str, | |
| *, | |
| locale: str, | |
| ) -> tuple[int, bool]: | |
| """Require exact edits inside every normalized URL/email source range. | |
| The ordinary CER representation intentionally removes punctuation. That | |
| makes an insertion immediately outside a URL indistinguishable from an | |
| insertion at the first/last URL character. For this protected-range gate | |
| only, retain frontend punctuation as alignment provenance. An insertion | |
| before the leading delimiter or after the trailing delimiter remains | |
| ordinary prose, while an insertion between a delimiter and the protected | |
| span is still rejected as URL/email content. | |
| """ | |
| normalized_frontend_target = normalize_spoken_forms(raw_target, locale=locale) | |
| spoken_spans = network_protected_spoken_spans(normalized_frontend_target) | |
| if not spoken_spans: | |
| return 0, True | |
| if network_identifier_has_ambiguous_iri(raw_target): | |
| return len(spoken_spans), False | |
| target_alignment_text = _comparison_text( | |
| normalized_frontend_target, | |
| locale=locale, | |
| target_text=normalized_frontend_target, | |
| preserve_network_boundaries=True, | |
| ) | |
| transcript_alignment_text = _comparison_text( | |
| raw_transcript, | |
| locale=locale, | |
| target_text=normalized_frontend_target, | |
| preserve_network_boundaries=True, | |
| ) | |
| operations = _levenshtein_alignment( | |
| target_alignment_text, | |
| transcript_alignment_text, | |
| ) | |
| ranges: list[tuple[int, int]] = [] | |
| cursor = 0 | |
| for spoken in spoken_spans: | |
| protected_text = _comparison_text( | |
| spoken, | |
| locale=locale, | |
| target_text=normalized_frontend_target, | |
| preserve_network_boundaries=True, | |
| ) | |
| if not protected_text: | |
| return len(spoken_spans), False | |
| start = target_alignment_text.find(protected_text, cursor) | |
| if start < 0: | |
| return len(spoken_spans), False | |
| end = start + len(protected_text) | |
| ranges.append((start, end)) | |
| cursor = end | |
| equal_source_indices = { | |
| source_index | |
| for operation, source_index, _ in operations | |
| if operation == "equal" | |
| } | |
| insertions = [ | |
| source_index | |
| for operation, source_index, _ in operations | |
| if operation == "insert" | |
| ] | |
| boundary_indices = { | |
| boundary_index | |
| for start, end in ranges | |
| for boundary_index in (start - 1, end) | |
| if 0 <= boundary_index < len(target_alignment_text) | |
| and target_alignment_text[boundary_index] | |
| == _NETWORK_ALIGNMENT_BOUNDARY_SENTINEL | |
| } | |
| ambiguous_boundary_substitution = any( | |
| operation == "substitute" and source_index in boundary_indices | |
| for operation, source_index, _ in operations | |
| ) | |
| deleted_boundary_indices = { | |
| source_index | |
| for operation, source_index, _ in operations | |
| if operation == "delete" and source_index in boundary_indices | |
| } | |
| ambiguous_deleted_boundary_insertion = any( | |
| insertion_index in {boundary_index, boundary_index + 1} | |
| for boundary_index in deleted_boundary_indices | |
| for insertion_index in insertions | |
| ) | |
| passed = all( | |
| all(source_index in equal_source_indices for source_index in range(start, end)) | |
| and not any(start <= source_index <= end for source_index in insertions) | |
| for start, end in ranges | |
| ) and not ( | |
| ambiguous_boundary_substitution | |
| or ambiguous_deleted_boundary_insertion | |
| ) | |
| return len(ranges), bool(passed) | |
| def compare_asr_text( | |
| target: str, | |
| transcript: str, | |
| *, | |
| locale: str = "zh-TW", | |
| prefix_units: int = 6, | |
| suffix_units: int = 6, | |
| max_cer: float = 0.10, | |
| max_prefix_cer: float = 0.0, | |
| max_suffix_cer: float = 0.0, | |
| max_extra_tail_units: int = 0, | |
| ) -> AsrComparison: | |
| """Compare an ASR transcript with strict onset, completion, and tail gates. | |
| Punctuation and whitespace are ignored, while written dates/numbers are | |
| normalized before alignment. Empty targets, unsupported locales, invalid | |
| limits, and non-finite limits all fail closed via ``passed=False``. | |
| """ | |
| try: | |
| locale_key = str(locale or "").replace("_", "-").lower() | |
| ambiguous_frequency_target = bool( | |
| locale_key in {"zh", "zh-tw", "zh-hant", "zh-cn", "zh-hans"} | |
| and _has_ambiguous_canonical_frequency(str(target or "")) | |
| ) | |
| frequency_contradiction = bool( | |
| locale_key in {"zh", "zh-tw", "zh-hant", "zh-cn", "zh-hans"} | |
| and _frequency_measurements_contradict( | |
| str(transcript or ""), | |
| str(target or ""), | |
| ) | |
| ) | |
| target_text = _comparison_text( | |
| target, | |
| locale=locale, | |
| target_text=target, | |
| ) | |
| transcript_text = _comparison_text( | |
| transcript, | |
| locale=locale, | |
| target_text=target, | |
| ) | |
| locale_key = str(locale or "").replace("_", "-").lower() | |
| if locale_key in {"zh", "zh-tw", "zh-hant", "zh-cn", "zh-hans"}: | |
| target_endpoint_units = mandarin_acoustic_units(target_text) | |
| transcript_endpoint_units = mandarin_acoustic_units(transcript_text) | |
| else: | |
| target_endpoint_units = tuple(target_text) | |
| transcript_endpoint_units = tuple(transcript_text) | |
| except (TypeError, ValueError): | |
| ambiguous_frequency_target = True | |
| frequency_contradiction = True | |
| target_text = "" | |
| transcript_text = "" | |
| target_endpoint_units = () | |
| transcript_endpoint_units = () | |
| try: | |
| prefix_count = int(prefix_units) | |
| suffix_count = int(suffix_units) | |
| except (TypeError, ValueError, OverflowError): | |
| prefix_count = 0 | |
| suffix_count = 0 | |
| prefix_limit = min(max(0, prefix_count), len(target_endpoint_units)) | |
| suffix_start = max(0, len(target_endpoint_units) - max(0, suffix_count)) | |
| orthographic_operations = _levenshtein_alignment(target_text, transcript_text) | |
| try: | |
| network_protected_spans, network_protected_spans_passed = ( | |
| _network_protected_alignment_evidence( | |
| str(target or ""), | |
| str(transcript or ""), | |
| locale=locale, | |
| ) | |
| ) | |
| except (TypeError, ValueError): | |
| network_protected_spans = 0 | |
| network_protected_spans_passed = False | |
| edit_distance = sum( | |
| operation != "equal" for operation, _, _ in orthographic_operations | |
| ) | |
| cer = edit_distance / len(target_text) if target_text else math.inf | |
| orthographic_tail = "".join( | |
| transcript_text[hypothesis_index] | |
| for operation, source_index, hypothesis_index in orthographic_operations | |
| if operation == "insert" | |
| and source_index >= len(target_text) | |
| and 0 <= hypothesis_index < len(transcript_text) | |
| ) | |
| operations = _levenshtein_alignment( | |
| target_endpoint_units, | |
| transcript_endpoint_units, | |
| ) | |
| prefix_errors = 0 | |
| suffix_errors = 0 | |
| prefix_deletions = 0 | |
| suffix_deletions = 0 | |
| tail_characters: list[str] = [] | |
| for operation, source_index, hyp_index in operations: | |
| if operation == "equal": | |
| continue | |
| if operation == "insert": | |
| if source_index < prefix_limit: | |
| prefix_errors += 1 | |
| if source_index >= suffix_start: | |
| suffix_errors += 1 | |
| if ( | |
| source_index >= len(target_endpoint_units) | |
| and 0 <= hyp_index < len(transcript_endpoint_units) | |
| ): | |
| tail_characters.append(transcript_endpoint_units[hyp_index]) | |
| continue | |
| if source_index < prefix_limit: | |
| prefix_errors += 1 | |
| prefix_deletions += operation == "delete" | |
| if source_index >= suffix_start: | |
| suffix_errors += 1 | |
| suffix_deletions += operation == "delete" | |
| prefix_cer = prefix_errors / prefix_limit if prefix_limit else math.inf | |
| suffix_length = len(target_endpoint_units) - suffix_start | |
| suffix_cer = suffix_errors / suffix_length if suffix_length else math.inf | |
| extra_tail = orthographic_tail | |
| limits: list[float] = [] | |
| for value in (max_cer, max_prefix_cer, max_suffix_cer, max_extra_tail_units): | |
| try: | |
| limits.append(float(value)) | |
| except (TypeError, ValueError, OverflowError): | |
| limits.append(math.nan) | |
| valid_limits = all(math.isfinite(value) and value >= 0.0 for value in limits) | |
| passed = bool( | |
| target_text | |
| and transcript_text | |
| and not ambiguous_frequency_target | |
| and not frequency_contradiction | |
| and network_protected_spans_passed | |
| and prefix_limit > 0 | |
| and suffix_length > 0 | |
| and valid_limits | |
| and cer <= limits[0] | |
| and prefix_cer <= limits[1] | |
| and suffix_cer <= limits[2] | |
| and len(tail_characters) <= limits[3] | |
| ) | |
| return AsrComparison( | |
| target_text=target_text, | |
| transcript_text=transcript_text, | |
| edit_distance=edit_distance, | |
| cer=cer, | |
| prefix_cer=prefix_cer, | |
| suffix_cer=suffix_cer, | |
| prefix_deletions=prefix_deletions, | |
| suffix_deletions=suffix_deletions, | |
| extra_tail=extra_tail, | |
| extra_tail_units=len(tail_characters), | |
| network_protected_spans=network_protected_spans, | |
| network_protected_spans_passed=network_protected_spans_passed, | |
| passed=passed, | |
| ) | |
| def _finite_number(value, *, minimum: float | None = None, maximum: float | None = None) -> float | None: | |
| if isinstance(value, (bool, np.bool_)): | |
| return None | |
| try: | |
| number = float(value) | |
| except (TypeError, ValueError, OverflowError): | |
| return None | |
| if not math.isfinite(number): | |
| return None | |
| if minimum is not None and number < minimum: | |
| return None | |
| if maximum is not None and number > maximum: | |
| return None | |
| return number | |
| def candidate_local_score( | |
| *, | |
| cer: float, | |
| speaker_similarity: float, | |
| boundary_speaker_drop: float, | |
| prefix_cer: float = 0.0, | |
| suffix_cer: float = 0.0, | |
| pace_penalty: float = 0.0, | |
| style_penalty: float = 0.0, | |
| asr_passed: bool = True, | |
| truncated: bool = False, | |
| extra_tail: bool = False, | |
| speaker_weight: float = 0.05, | |
| boundary_weight: float = 0.10, | |
| prefix_weight: float = 0.50, | |
| suffix_weight: float = 0.50, | |
| pace_weight: float = 1.0, | |
| style_weight: float = 1.0, | |
| max_cer: float | None = None, | |
| min_speaker_similarity: float | None = None, | |
| max_boundary_speaker_drop: float | None = None, | |
| ) -> float: | |
| """Return a finite local candidate cost or ``inf`` for unsafe evidence.""" | |
| if asr_passed is not True or truncated is not False or extra_tail is not False: | |
| return math.inf | |
| metric_values = [ | |
| _finite_number(cer, minimum=0.0), | |
| _finite_number(speaker_similarity, minimum=-1.0, maximum=1.0), | |
| _finite_number(boundary_speaker_drop, minimum=0.0), | |
| _finite_number(prefix_cer, minimum=0.0), | |
| _finite_number(suffix_cer, minimum=0.0), | |
| _finite_number(pace_penalty, minimum=0.0), | |
| _finite_number(style_penalty, minimum=0.0), | |
| ] | |
| weights = [ | |
| _finite_number(speaker_weight, minimum=0.0), | |
| _finite_number(boundary_weight, minimum=0.0), | |
| _finite_number(prefix_weight, minimum=0.0), | |
| _finite_number(suffix_weight, minimum=0.0), | |
| _finite_number(pace_weight, minimum=0.0), | |
| _finite_number(style_weight, minimum=0.0), | |
| ] | |
| if any(value is None for value in metric_values + weights): | |
| return math.inf | |
| candidate_cer, similarity, boundary_drop, onset_cer, ending_cer, pace, style = metric_values | |
| gate_max_cer = None if max_cer is None else _finite_number(max_cer, minimum=0.0) | |
| gate_min_similarity = ( | |
| None | |
| if min_speaker_similarity is None | |
| else _finite_number(min_speaker_similarity, minimum=-1.0, maximum=1.0) | |
| ) | |
| gate_max_boundary = ( | |
| None | |
| if max_boundary_speaker_drop is None | |
| else _finite_number(max_boundary_speaker_drop, minimum=0.0) | |
| ) | |
| if ( | |
| max_cer is not None and gate_max_cer is None | |
| or min_speaker_similarity is not None and gate_min_similarity is None | |
| or max_boundary_speaker_drop is not None and gate_max_boundary is None | |
| ): | |
| return math.inf | |
| if gate_max_cer is not None and candidate_cer > gate_max_cer: | |
| return math.inf | |
| if gate_min_similarity is not None and similarity < gate_min_similarity: | |
| return math.inf | |
| if gate_max_boundary is not None and boundary_drop > gate_max_boundary: | |
| return math.inf | |
| speaker_w, boundary_w, prefix_w, suffix_w, pace_w, style_w = weights | |
| score = ( | |
| candidate_cer | |
| + speaker_w * (1.0 - similarity) | |
| + boundary_w * boundary_drop | |
| + prefix_w * onset_cer | |
| + suffix_w * ending_cer | |
| + pace_w * pace | |
| + style_w * style | |
| ) | |
| return score if math.isfinite(score) and score >= 0.0 else math.inf | |
| def candidate_transition_score( | |
| *, | |
| speaker_similarity: float, | |
| f0_delta: float, | |
| rms_delta: float, | |
| speaker_weight: float = 1.0, | |
| f0_weight: float = 1.0, | |
| rms_weight: float = 1.0, | |
| ) -> float: | |
| """Score continuity between adjacent candidates with finite-only inputs.""" | |
| values = [ | |
| _finite_number(speaker_similarity, minimum=-1.0, maximum=1.0), | |
| _finite_number(f0_delta, minimum=0.0), | |
| _finite_number(rms_delta, minimum=0.0), | |
| _finite_number(speaker_weight, minimum=0.0), | |
| _finite_number(f0_weight, minimum=0.0), | |
| _finite_number(rms_weight, minimum=0.0), | |
| ] | |
| if any(value is None for value in values): | |
| return math.inf | |
| similarity, pitch_delta, loudness_delta, speaker_w, pitch_w, loudness_w = values | |
| score = ( | |
| speaker_w * (1.0 - similarity) | |
| + pitch_w * pitch_delta | |
| + loudness_w * loudness_delta | |
| ) | |
| return score if math.isfinite(score) and score >= 0.0 else math.inf | |
| class CandidateSequenceSelection: | |
| candidate_indices: tuple[int, ...] | |
| total_score: float | |
| def select_candidate_sequence( | |
| local_scores: Sequence[Sequence[float]], | |
| transition_scores: Sequence[Sequence[Sequence[float]]] = (), | |
| ) -> CandidateSequenceSelection | None: | |
| """Select the minimum-cost candidate path in ``O(N K^2)``. | |
| Non-finite/negative scores remove only their candidate or edge. Malformed | |
| matrices and graphs with no complete finite path return ``None`` so callers | |
| cannot accidentally fall back to an unverified candidate. | |
| """ | |
| try: | |
| local_rows = [list(row) for row in local_scores] | |
| except TypeError: | |
| return None | |
| if not local_rows or any(not row for row in local_rows): | |
| return None | |
| safe_local = [ | |
| [ | |
| score if score is not None else math.inf | |
| for score in (_finite_number(value, minimum=0.0) for value in row) | |
| ] | |
| for row in local_rows | |
| ] | |
| if len(safe_local) == 1: | |
| try: | |
| if len(transition_scores) != 0: | |
| return None | |
| except TypeError: | |
| return None | |
| best_index = min(range(len(safe_local[0])), key=safe_local[0].__getitem__) | |
| best_score = safe_local[0][best_index] | |
| if not math.isfinite(best_score): | |
| return None | |
| return CandidateSequenceSelection((best_index,), best_score) | |
| try: | |
| transitions = [[list(row) for row in matrix] for matrix in transition_scores] | |
| except TypeError: | |
| return None | |
| if len(transitions) != len(safe_local) - 1: | |
| return None | |
| safe_transitions: list[list[list[float]]] = [] | |
| for index, matrix in enumerate(transitions): | |
| previous_count = len(safe_local[index]) | |
| current_count = len(safe_local[index + 1]) | |
| if len(matrix) != previous_count or any(len(row) != current_count for row in matrix): | |
| return None | |
| safe_transitions.append( | |
| [ | |
| [ | |
| score if score is not None else math.inf | |
| for score in (_finite_number(value, minimum=0.0) for value in row) | |
| ] | |
| for row in matrix | |
| ] | |
| ) | |
| previous_costs = safe_local[0] | |
| backpointers: list[list[int]] = [] | |
| for step in range(1, len(safe_local)): | |
| current_costs = [math.inf] * len(safe_local[step]) | |
| current_backpointers = [-1] * len(safe_local[step]) | |
| for current_index, local_score in enumerate(safe_local[step]): | |
| if not math.isfinite(local_score): | |
| continue | |
| for previous_index, previous_score in enumerate(previous_costs): | |
| edge_score = safe_transitions[step - 1][previous_index][current_index] | |
| if not math.isfinite(previous_score) or not math.isfinite(edge_score): | |
| continue | |
| total = previous_score + edge_score + local_score | |
| if math.isfinite(total) and total < current_costs[current_index]: | |
| current_costs[current_index] = total | |
| current_backpointers[current_index] = previous_index | |
| previous_costs = current_costs | |
| backpointers.append(current_backpointers) | |
| final_index = min(range(len(previous_costs)), key=previous_costs.__getitem__) | |
| total_score = previous_costs[final_index] | |
| if not math.isfinite(total_score): | |
| return None | |
| indices = [final_index] | |
| for pointers in reversed(backpointers): | |
| final_index = pointers[final_index] | |
| if final_index < 0: | |
| return None | |
| indices.append(final_index) | |
| indices.reverse() | |
| return CandidateSequenceSelection(tuple(indices), total_score) | |
| def extract_windowed_speaker_embedding( | |
| wav_path: str, | |
| encoder, | |
| *, | |
| device: str = "cpu", | |
| min_duration_seconds: float = 3.0, | |
| window_seconds: float = 3.0, | |
| hop_seconds: float = 1.5, | |
| max_windows: int = 12, | |
| full_clip_max_seconds: float = 12.0, | |
| ) -> torch.Tensor: | |
| """Extract one denoised ECAPA embedding from overlapping reference windows.""" | |
| import librosa | |
| waveform, _ = librosa.load(wav_path, sr=16000, mono=True) | |
| waveform = np.asarray(waveform, dtype=np.float32) | |
| duration = waveform.size / 16000.0 | |
| if duration < min_duration_seconds: | |
| raise ValueError( | |
| f"reference audio must be at least {min_duration_seconds:.1f} seconds; got {duration:.2f}" | |
| ) | |
| segments: list[np.ndarray] = [] | |
| if duration <= full_clip_max_seconds: | |
| segments.append(waveform) | |
| window = max(1, int(round(window_seconds * 16000))) | |
| hop = max(1, int(round(hop_seconds * 16000))) | |
| starts = list(range(0, max(0, waveform.size - window) + 1, hop)) | |
| if max_windows > 0 and len(starts) > max_windows: | |
| indices = np.linspace(0, len(starts) - 1, max_windows).round().astype(int) | |
| starts = [starts[index] for index in dict.fromkeys(indices.tolist())] | |
| segments.extend(waveform[start : start + window] for start in starts) | |
| embeddings: list[torch.Tensor] = [] | |
| for segment in segments: | |
| tensor = torch.from_numpy(np.ascontiguousarray(segment)).float().unsqueeze(0).to(device) | |
| embedding = encoder.encode_batch(tensor).reshape(-1) | |
| embeddings.append(torch.nn.functional.normalize(embedding, dim=0).cpu()) | |
| if not embeddings: | |
| raise ValueError("reference audio did not contain a usable speech window") | |
| return torch.nn.functional.normalize(torch.stack(embeddings).mean(dim=0), dim=0) | |