"""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)(?\"{re.escape(_NETWORK_NONASCII_BOUNDARY_PUNCTUATION)}]+" ) _SPOKEN_EMAIL_RE = re.compile( r"(?i)(?日[元圓圆])\s*" rf"(?P{_CURRENCY_NUMBER_PATTERN}|{_ASR_JPY_ZH_AMOUNT_PATTERN})" r"(?P元)?" 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) @dataclass(frozen=True) 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) @dataclass(frozen=True) 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 @dataclass(frozen=True) 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"(?安全網址是|網址是) " rf"(?P{_NATURALIZED_URL_SPOKEN_LABEL_PATTERN}" rf"(?: 點 {_NATURALIZED_URL_SPOKEN_LABEL_PATTERN})+)" r"(?:，路徑是 (?P[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)(? 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"|(? 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 @dataclass(frozen=True) 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 = "" @dataclass(frozen=True) 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" @property def network_conditioned(self) -> bool: return bool(self.network_span_indices) _EMAIL_DOMAIN_MAIL_TOKEN_RE = re.compile( r"(? 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 @dataclass(frozen=True) 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 @dataclass(frozen=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), ) @dataclass(frozen=True) 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 @dataclass(frozen=True) 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) @torch.no_grad() 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)