tts_robust_normalizer_single_script.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
TTS 输入鲁棒性正则化器（非语义 TN）

目标
----
1. 只做“鲁棒性清洗”，不做数字/单位/日期/金额等语义展开。
2. 优先保护高风险 token，避免把 `.map`、`app.js.map`、`v2.3.1`、URL、Email、@mention、#hashtag 清坏。
3. `[]` / `{}` / `【】` / `〖〗` / `『』` / `「」` 统一转成双引号包裹内容。
4. 对结构性符号做“替换而非删除”：
   - `【】 / 〖〗 / 『』 / 「」` 转成双引号包裹内容。
   - `《》` 只在“独立标题/栏目名”场景拆开；嵌入式标题保持不变。
   - `—— / -- / ——...` 转成句边界。
5. 对社交平台常见噪声做弱归一化：
   - `...... / ……` -> `。`
   - `？？？！！！` -> `？！`
   - `！！！` -> `！`
6. 空格按脚本类型处理：
   - 西文片段内部：连续空格压缩为 1 个。
   - 汉字 / 日文假名片段内部：删除空格。
   - 汉字 / 日文假名 与“拉丁字母类 token / 受保护 token”相邻：保留或补 1 个空格。
   - 汉字 / 日文假名 与纯数字相邻：不强行补空格。
   - 普通文本中的 `_` 转成空格；已被保护的 URL / Email / mention / hashtag / 文件名中的 `_` 保留。
7. 轻量处理 Markdown 与换行：
   - `[text](url)` -> `text url`
   - 去掉标题 `#`、引用 `>`、列表前缀
   - 换行转句边界 `。`
8. 与 WeTextProcessing 联动时，中文分支会在进入 WeText 前额外处理单个 ASCII 连字符 `-`，
   以避免被中文 TN 误读成“减”。这部分逻辑实现于 `text_normalization_pipeline.py`，约定如下：
   - `数字-数字` 保留：`10-3`、`2024-05-01`
   - 明显负号保留：`-2`、`为-2`、`为 -2`、`x=-2`、`(-2)`、`计算-2的绝对值`
   - `中文-中文` 转成停顿边界：`请求接入-身份判定` -> `请求接入，身份判定`
   - 其余非数字连字符转成空格：`A-B`、`A-100`、`GPU-A100`、`TTS-demo`、
     `中文-ABC`、`ABC-中文`
   - 连字符两侧是否带空格不影响判断：`A-B` / `A - B`、`为-2` / `为 -2`
   - 当前仅对中文 WeText 分支生效；若文本被判定为英文分支，则仍按英文 WeText 原生行为处理，
     例如 `is -2` 当前会归一化为 `is negative two`

非目标
------
1. 不决定“应该怎么读”。
2. 不在本脚本内直接实现 WeText 的语义 TN，相关联动仅在共享 pipeline 中做最小预处理。
3. 不做 HTML/SSML/语义标签解释。
"""

from __future__ import annotations

import re
import unicodedata


# ---------------------------
# 基础常量与正则
# ---------------------------

# 不依赖空格分词的脚本：汉字 + 日文假名
_CJK_CHARS = r"\u3400-\u4dbf\u4e00-\u9fff\u3040-\u30ff"
_CJK = f"[{_CJK_CHARS}]"

# 保护占位符
_PROT = r"___PROT\d+___"

# 需要保护的高风险 token
_URL_RE = re.compile(r"https?://[^\s\u3000，。！？；、）】》〉」』]+")
_EMAIL_RE = re.compile(r"(?<![\w.+-])[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}(?![\w.-])")
_MENTION_RE = re.compile(r"(?<![A-Za-z0-9_])@[A-Za-z0-9_]{1,32}")
_REDDIT_RE = re.compile(r"(?<![A-Za-z0-9_])(?:u|r)/[A-Za-z0-9_]+")
_HASHTAG_RE = re.compile(r"(?<![A-Za-z0-9_])#(?!\s)[^\s#]+")

# `.map` / `.env` / `.gitignore`
_DOT_TOKEN_RE = re.compile(r"(?<![A-Za-z0-9_])\.(?=[A-Za-z0-9._-]*[A-Za-z0-9])[A-Za-z0-9._-]+")

# `app.js.map` / `index.d.ts` / `v2.3.1` / `foo/bar-baz.py` 等
# 这里不把“仅含下划线的 snake_case 普通文本”视为高风险 token，
# 以便后续 `_` -> 空格 规则能生效；真正的文件名/路径通常仍会因为含 `.` `/` `-` `:` `+` 被保护。
_FILELIKE_RE = re.compile(
    r"(?<![A-Za-z0-9_])"
    r"(?=[A-Za-z0-9._/+:-]*[A-Za-z])"
    r"(?=[A-Za-z0-9._/+:-]*[./+:-])"
    r"[A-Za-z0-9][A-Za-z0-9._/+:-]*"
    r"(?![A-Za-z0-9_])"
)

# 参与“中英混排边界补空格”的 token：必须至少含 1 个拉丁字母，或本身就是受保护 token
_LATINISH = rf"(?:{_PROT}|(?=[A-Za-z0-9._/+:-]*[A-Za-z])[A-Za-z0-9][A-Za-z0-9._/+:-]*)"

# 零宽字符
_ZERO_WIDTH_RE = re.compile(r"[\u200b-\u200d\ufeff]")
_TRAILING_CLOSERS = set('"\')]}）】》〉」』”’')


# ---------------------------
# 主函数
# ---------------------------

def normalize_tts_text(text: str) -> str:
    """对 TTS 输入做鲁棒性正则化。"""
    text = _base_cleanup(text)
    text = _normalize_markdown_and_lines(text)
    text = _normalize_flow_arrows(text)
    text, protected = _protect_spans(text)
    text = _normalize_visible_underscores(text)

    text = _normalize_spaces(text)
    text = _normalize_structural_punctuation(text)
    text = _normalize_repeated_punctuation(text)
    text = _normalize_spaces(text)

    text = _restore_spans(text, protected)
    text = text.strip()
    return _ensure_terminal_punctuation_by_line(text)


# ---------------------------
# 具体规则
# ---------------------------

def _base_cleanup(text: str) -> str:
    text = text.replace("\r\n", "\n").replace("\r", "\n").replace("\u3000", " ")
    text = _ZERO_WIDTH_RE.sub("", text)

    cleaned = []
    for ch in text:
        cat = unicodedata.category(ch)
        if ch in "\n\t " or not cat.startswith("C"):
            cleaned.append(ch)
    return "".join(cleaned)


def _normalize_markdown_and_lines(text: str) -> str:
    # Markdown 链接：[text](url) -> text url
    text = re.sub(r"\[([^\[\]]+?)\]\((https?://[^)\s]+)\)", r"\1 \2", text)

    lines = []
    for raw in text.splitlines():
        line = raw.strip()
        if not line:
            continue

        line = re.sub(r"^#{1,6}\s+", "", line)   # 标题
        line = re.sub(r"^>\s+", "", line)        # 引用
        line = re.sub(r"^[-*+]\s+", "", line)    # 无序列表
        line = re.sub(r"^\d+[.)]\s+", "", line)  # 有序列表
        lines.append(line)

    if not lines:
        return ""

    merged: list[str] = [lines[0]]
    for line in lines[1:]:
        previous = merged[-1]
        merged[-1] = _ensure_terminal_punctuation(previous)
        merged.append(line)
    return "".join(merged)


def _protect_spans(text: str) -> tuple[str, list[str]]:
    protected: list[str] = []

    def repl(match: re.Match[str]) -> str:
        idx = len(protected)
        protected.append(match.group(0))
        return f"___PROT{idx}___"

    for pattern in (
        _URL_RE,
        _EMAIL_RE,
        _MENTION_RE,
        _REDDIT_RE,
        _HASHTAG_RE,
        _DOT_TOKEN_RE,
        _FILELIKE_RE,
    ):
        text = pattern.sub(repl, text)

    return text, protected


def _restore_spans(text: str, protected: list[str]) -> str:
    for idx, original in enumerate(protected):
        text = text.replace(f"___PROT{idx}___", original)
    return text


def _normalize_visible_underscores(text: str) -> str:
    parts = re.split(rf"({_PROT})", text)
    return "".join(
        part if re.fullmatch(_PROT, part) else part.replace("_", " ")
        for part in parts
    )


def _normalize_flow_arrows(text: str) -> str:
    return re.sub(
        r"\s*(?:<[-=]+>|[-=]+>|<[-=]+|[→←↔⇒⇐⇔⟶⟵⟷⟹⟸⟺↦↤↪↩])\s*",
        "，",
        text,
    )


def _normalize_spaces(text: str) -> str:
    # 统一空白
    text = re.sub(r"[ \t\r\f\v]+", " ", text)

    # 汉字 / 日文片段内部：删除空格
    text = re.sub(rf"({_CJK})\s+(?={_CJK})", r"\1", text)

    # 汉字 / 日文 与纯数字之间：删除空格（不强行保留）
    text = re.sub(rf"({_CJK})\s+(?=\d)", r"\1", text)
    text = re.sub(rf"(\d)\s+(?={_CJK})", r"\1", text)

    # 汉字 / 日文 与拉丁字母类 token / protected token 相邻：保留或补 1 个空格
    text = re.sub(rf"({_CJK})(?=({_LATINISH}))", r"\1 ", text)
    text = re.sub(rf"(({_LATINISH}))(?={_CJK})", r"\1 ", text)

    # 再压一遍连续空格
    text = re.sub(r" {2,}", " ", text)

    # 中文标点前后不保留空格
    text = re.sub(r"\s+([，。！？；：、”’」』】）》])", r"\1", text)
    text = re.sub(r"([（【「『《“‘])\s+", r"\1", text)
    text = re.sub(r"([，。！？；：、])\s*", r"\1", text)

    # ASCII 标点前不留空格；后面的英文空格不强改
    text = re.sub(r"\s+([,.;!?])", r"\1", text)

    return re.sub(r" {2,}", " ", text).strip()


def _normalize_structural_punctuation(text: str) -> str:
    # 各类结构性括号：统一转成双引号包裹内容
    text = re.sub(r"\[\s*([^\[\]]+?)\s*\]", r'"\1"', text)
    text = re.sub(r"\{\s*([^{}]+?)\s*\}", r'"\1"', text)
    text = re.sub(r"[【〖『「]\s*([^】〗』」]+?)\s*[】〗』」]", r'"\1"', text)

    # 《》只处理独立标题，不处理嵌入式标题
    # 例：重磅。《新品发布》——现在开始！ -> 重磅。新品发布。现在开始！
    text = re.sub(
        r"(^|[。！？!?；;]\s*)《([^》]+)》(?=\s*(?:___PROT\d+___|[—–―-]{2,}|$|[。！？!?；;，,]))",
        r"\1\2",
        text,
    )

    # 流程 / 映射箭头：转成中文逗号，避免把 `->` 之类原样喂给 TTS。
    text = _normalize_flow_arrows(text)

    # 长破折号 / 多连字符：转句边界
    text = re.sub(r"\s*(?:—|–|―|-){2,}\s*", "。", text)

    return text


def _normalize_repeated_punctuation(text: str) -> str:
    # 省略号 / 连续句点
    text = re.sub(r"(?:\.{3,}|…{2,}|……+)", "。", text)

    # 同类重复标点
    text = re.sub(r"[。．]{2,}", "。", text)
    text = re.sub(r"[，,]{2,}", "，", text)
    text = re.sub(r"[!！]{2,}", "！", text)
    text = re.sub(r"[?？]{2,}", "？", text)

    # 混合问叹号：收敛到 ？！
    def _mixed_qe(match: re.Match[str]) -> str:
        s = match.group(0)
        has_q = any(ch in s for ch in "?？")
        has_e = any(ch in s for ch in "!！")
        if has_q and has_e:
            return "？！"
        return "？" if has_q else "！"

    text = re.sub(r"[!?！？]{2,}", _mixed_qe, text)
    return text


def _ensure_terminal_punctuation(text: str) -> str:
    if not text:
        return text

    index = len(text) - 1
    while index >= 0 and text[index].isspace():
        index -= 1
    while index >= 0 and text[index] in _TRAILING_CLOSERS:
        index -= 1

    if index >= 0 and unicodedata.category(text[index]).startswith("P"):
        return text
    return text + "。"


def _ensure_terminal_punctuation_by_line(text: str) -> str:
    if not text:
        return text
    lines = text.split("\n")
    normalized_lines = [_ensure_terminal_punctuation(line.strip()) if line.strip() else "" for line in lines]
    return "\n".join(normalized_lines).strip()


# ---------------------------
# 测试
# ---------------------------

TEST_CASES = [
    # 1) .map / dot-leading token / 文件名 / 版本号
    (
        "dot_map_sentence",
        "2026 年 3 月 31 日，安全研究员 Chaofan Shou (@Fried_rice) 发现 Anthropic 的 npm 包中暴露了 .map 文件，",
        "2026年3月31日，安全研究员 Chaofan Shou (@Fried_rice) 发现 Anthropic 的 npm 包中暴露了 .map 文件，",
    ),
    ("dot_tokens", "别把 .env、.npmrc、.gitignore 提交上去。", "别把 .env、.npmrc、.gitignore 提交上去。"),
    ("file_names", "请检查 bundle.min.js、package.json 和 processing_moss_tts.py。", "请检查 bundle.min.js、package.json 和 processing_moss_tts.py。"),
    ("index_d_ts", "index.d.ts 里也有同样的问题。", "index.d.ts 里也有同样的问题。"),
    ("version_build", "Bug 的讨论可以精确到 v2.3.1 (Build 15)。", "Bug 的讨论可以精确到 v2.3.1 (Build 15)。"),
    ("version_rc", "3.0.0-rc.1 还不能上生产。", "3.0.0-rc.1 还不能上生产。"),
    ("jar_name", "fabric-api-0.91.3+1.20.2.jar 需要单独下载。", "fabric-api-0.91.3+1.20.2.jar 需要单独下载。"),

    # 2) URL / Email / mention / hashtag / Reddit
    ("url", "仓库地址是 https://github.com/instructkr/claude-code", "仓库地址是 https://github.com/instructkr/claude-code。"),
    ("email", "联系邮箱：ops+tts@example.ai", "联系邮箱：ops+tts@example.ai。"),
    ("mention", "@Fried_rice 说这是 source map 暴露。", "@Fried_rice 说这是 source map 暴露。"),
    ("reddit", "去 r/singularity 看讨论。", "去 r/singularity 看讨论。"),
    ("hashtag_chain", "#张雪峰#张雪峰[话题]#张雪峰事件", "#张雪峰#张雪峰[话题]#张雪峰事件。"),
    ("mention_hashtag_boundary", "关注@biscuit0228_并转发#thetime_tbs", "关注 @biscuit0228_ 并转发 #thetime_tbs。"),

    # 3) bracket / 控制 token：统一转成双引号
    ("speaker_bracket", "[S1]你好。[S2]收到。", '"S1"你好。"S2"收到。'),
    ("event_bracket", "请模仿 {whisper} 的语气说“别出声”。", '请模仿 "whisper" 的语气说“别出声”。'),
    ("order_bracket", "订单号：[AB-1234-XYZ]", '订单号："AB-1234-XYZ"。'),

    # 4) 结构性符号：转成双引号或句边界，而不是直接删除
    ("struct_headline", "〖重磅〗《新品发布》——现在开始！", '"重磅"《新品发布》。现在开始！'),
    ("struct_notice", "【公告】今天 20:00 维护——预计 30 分钟。", '"公告"今天20:00维护。预计30分钟。'),
    ("struct_quote_chain", "『特别提醒』「不要外传」", '"特别提醒""不要外传"。'),
    ("struct_embedded_quote", "他说【重要通知】明天发布。", '他说"重要通知"明天发布。'),
    ("flow_arrow_chain", "请求接入 -> 身份与策略判定 -> 域服务处理", "请求接入，身份与策略判定，域服务处理。"),
    ("flow_arrow_no_space", "A->B", "A，B。"),
    ("flow_arrow_unicode", "配置中心→推理编排→运行时执行", "配置中心，推理编排，运行时执行。"),

    # 5) 嵌入式标题：保留
    ("embedded_title", "我喜欢《哈姆雷特》这本书。", "我喜欢《哈姆雷特》这本书。"),

    # 6) 重复标点 / 社交噪声
    ("noise_qe", "真的假的？？？！！！", "真的假的？！"),
    ("noise_ellipsis", "这个包把 app.js.map 也发上去了......太离谱了！！！", "这个包把 app.js.map 也发上去了。太离谱了！"),
    ("noise_ellipsis_cn", "【系统提示】请模仿{sad}低沉语气，说“今天下雨了……”", '"系统提示"请模仿"sad"低沉语气，说“今天下雨了。”'),

    # 7) 空格规则：英文压缩、中文删除、中英混排保留边界
    ("english_spaces", "This   is   a   test.", "This is a test."),
    ("chinese_spaces", "这 是　一 段  含有多种空白的文本。", "这是一段含有多种空白的文本。"),
    ("mixed_spaces_1", "这是Anthropic的npm包", "这是 Anthropic 的 npm 包。"),
    ("mixed_spaces_2", "今天update到v2.3.1了", "今天 update 到 v2.3.1 了。"),
    ("mixed_spaces_3", "处理app.js.map文件", "处理 app.js.map 文件。"),
    ("underscore_plain_1", "foo_bar", "foo bar。"),
    ("underscore_plain_2", "中文_ABC", "中文 ABC。"),
    ("underscore_protected_mention", "关注@foo_bar", "关注 @foo_bar。"),

    # 8) Markdown / 列表 / 换行
    ("markdown_link", "详情见 [release note](https://github.com/example/release)", "详情见 release note https://github.com/example/release。"),
    ("markdown_heading", "# I made a free open source app to help with markdown files", "I made a free open source app to help with markdown files。"),
    ("list_lines", "- 修复 .map 泄露\n- 发布 v2.3.1", "修复 .map 泄露。发布 v2.3.1。"),
    ("numbered_lines", "1. 安装依赖\n2. 运行测试\n3. 发布 v2.3.1", "安装依赖。运行测试。发布 v2.3.1。"),
    ("newlines", "第一行\n第二行\n第三行", "第一行。第二行。第三行。"),

    # 9) 句末补标点
    ("terminal_punct_plain", "今天发布", "今天发布。"),
    ("terminal_punct_quoted", '他说"你好"', '他说"你好"。'),
    ("terminal_punct_existing", "今天发布。", "今天发布。"),
    ("terminal_punct_newlines", "第一行\n第二行。", "第一行。第二行。"),
    ("terminal_punct_blank_lines", "第一行\n\n第二行", "第一行。第二行。"),

    # 10) 零宽字符 / 幂等性
    ("zero_width_url", "详见 https://x.com/\u200bSafety", "详见 https://x.com/Safety。"),
]


def run_tests(verbose: bool = True) -> None:
    failed = []

    for name, text, expected in TEST_CASES:
        actual = normalize_tts_text(text)
        if actual != expected:
            failed.append((name, text, expected, actual))
            continue

        # 幂等性：第二次归一化不应继续改动结果
        second = normalize_tts_text(actual)
        if second != actual:
            failed.append((name + "_idempotence", actual, actual, second))

    if failed:
        lines = ["\nTEST FAILED:\n"]
        for name, text, expected, actual in failed:
            lines.append(f"[{name}]")
            lines.append(f"input   : {text}")
            lines.append(f"expected: {expected}")
            lines.append(f"actual  : {actual}")
            lines.append("")
        raise AssertionError("\n".join(lines))

    if verbose:
        print(f"All {len(TEST_CASES)} tests passed.")


if __name__ == "__main__":
    run_tests()