deploy standalone ASR pipeline on tools-pve

2026-09-18 02:32:14 -07:00
parent f30f72b295
commit b23f00f41a
38 changed files with 182773 additions and 0 deletions
@@ -0,0 +1,170 @@
+#!/usr/bin/env python3
+"""Conservative Chinese-number classification and normalization."""
+from __future__ import annotations
+
+import re
+from dataclasses import asdict, dataclass
+
+DIGIT = {
+    "零": 0, "〇": 0, "一": 1, "幺": 1, "二": 2, "两": 2, "三": 3,
+    "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9,
+}
+SMALL = {"十": 10, "百": 100, "千": 1000}
+BIG = {"万": 10_000, "亿": 100_000_000}
+CN = "零〇一二两三四五六七八九十百千万亿幺"
+CN_RE = rf"[{CN}]+"
+PURE_DIGITS = set(DIGIT)
+
+
+@dataclass
+class Decision:
+    raw: str
+    normalized: str
+    kind: str
+    confidence: float
+    start: int
+    end: int
+    context: str
+
+    def as_dict(self) -> dict:
+        return asdict(self)
+
+
+def chinese_to_int(raw: str) -> int | None:
+    raw = raw.replace("两", "二")
+    if not raw or any(c not in DIGIT and c not in SMALL and c not in BIG for c in raw):
+        return None
+    if set(raw) <= PURE_DIGITS:
+        return int("".join(str(DIGIT[c]) for c in raw))
+    # "两万五" is conventionally ambiguous between 20,005 and 25,000.
+    if any(c in BIG for c in raw) and raw[-1] in DIGIT:
+        return None
+    total = section = number = 0
+    for c in raw:
+        if c in DIGIT:
+            number = DIGIT[c]
+        elif c in SMALL:
+            section += (number or 1) * SMALL[c]
+            number = 0
+        else:
+            section += number
+            total += (section or 1) * BIG[c]
+            section = number = 0
+    return total + section + number
+
+
+def _ctx(text: str, start: int, end: int, width: int = 18) -> str:
+    return text[max(0, start - width):min(len(text), end + width)]
+
+
+def _add(found: list[Decision], text: str, start: int, end: int,
+         normalized: str, kind: str, confidence: float):
+    raw = text[start:end]
+    if raw == normalized:
+        return
+    found.append(Decision(raw, normalized, kind, confidence, start, end,
+                          _ctx(text, start, end)))
+
+
+def _overlaps(a: Decision, accepted: list[Decision]) -> bool:
+    return any(a.start < b.end and b.start < a.end for b in accepted)
+
+
+def _money_text(raw: str, currency: str) -> str | None:
+    if raw and raw[-1] in BIG:
+        base = chinese_to_int(raw[:-1])
+        return f"{base}{raw[-1]}{currency}" if base is not None else None
+    value = chinese_to_int(raw)
+    return f"{value}{currency}" if value is not None else None
+
+
+def normalize(text: str) -> tuple[str, list[dict]]:
+    """Normalize only values whose semantic type is visible in nearby syntax."""
+    found: list[Decision] = []
+
+    # Percentages.
+    for m in re.finditer(rf"百分之(?P<n>{CN_RE})", text):
+        value = chinese_to_int(m.group("n"))
+        if value is not None:
+            _add(found, text, m.start(), m.end(), f"{value}%", "percentage", 0.99)
+
+    # Full dates and standalone day-of-month references.
+    for m in re.finditer(
+            rf"(?:(?P<y>{CN_RE})年)?(?P<mo>{CN_RE})月(?P<d>{CN_RE})(?P<u>日|号)", text):
+        y = chinese_to_int(m.group("y")) if m.group("y") else None
+        mo, d = chinese_to_int(m.group("mo")), chinese_to_int(m.group("d"))
+        if mo is not None and d is not None and 1 <= mo <= 12 and 1 <= d <= 31:
+            norm = f"{y}年" if y is not None else ""
+            norm += f"{mo}月{d}{m.group('u')}"
+            _add(found, text, m.start(), m.end(), norm, "date", 0.98)
+    for m in re.finditer(rf"(?P<n>{CN_RE})(?P<u>日|号)", text):
+        value = chinese_to_int(m.group("n"))
+        if value is not None and 1 <= value <= 31:
+            _add(found, text, m.start(), m.end(), f"{value}{m.group('u')}",
+                 "day", 0.93)
+
+    # Money: preserve the stated currency and avoid silently inventing one.
+    for m in re.finditer(
+            rf"(?P<n>{CN_RE})(?P<u>元|块钱|人民币|美元|美金|港币)", text):
+        norm = _money_text(m.group("n"), m.group("u"))
+        if norm is not None:
+            _add(found, text, m.start(), m.end(), norm,
+                 "money", 0.98)
+    for m in re.finditer(rf"(?P<n>[一二两三四五六七八九十百千万]+)个(?P<u>亿|万)", text):
+        value = chinese_to_int(m.group("n"))
+        if value is not None and re.search(r"投|融|估值|收入|销售|利润|资金|金额",
+                                           _ctx(text, m.start(), m.end())):
+            _add(found, text, m.start(), m.end(), f"{value}{m.group('u')}元",
+                 "money", 0.90)
+    for m in re.finditer(
+            rf"(?P<n>[一二两三四五六七八九十百千]+)(?P<u>万|亿)", text):
+        context = _ctx(text, m.start(), m.end())
+        value = chinese_to_int(m.group("n"))
+        if value is not None and re.search(
+                r"投|融|估值|收入|销售|利润|资金|金额|打款|退出|回购|报价", context):
+            _add(found, text, m.start(), m.end(),
+                 f"{value}{m.group('u')}元", "money", 0.91)
+
+    # Quantities with an explicit unit.
+    units = "个月|小时|分钟|公斤|毫克|毫升|人|家|份|批|次|台|套|件|天|年|吨|克|升|例|名"
+    for m in re.finditer(rf"(?P<n>{CN_RE})(?P<u>{units})", text):
+        value = chinese_to_int(m.group("n"))
+        if value is not None:
+            _add(found, text, m.start(), m.end(), f"{value}{m.group('u')}",
+                 "quantity", 0.96)
+
+    # Codes are spoken digit-by-digit. 幺 is a particularly strong signal.
+    for m in re.finditer(
+            rf"(?<![{CN}])(?P<n>[零〇一二两三四五六七八九幺]{{2,}})(?![{CN}])",
+            text):
+        raw = m.group("n")
+        context = _ctx(text, m.start(), m.end())
+        code_context = ("幺" in raw or
+                        re.search(r"项目|编号|代号|批号|工单|报告|BF|DF|CF|MFG",
+                                  context, re.I))
+        if code_context:
+            value = "".join(str(DIGIT[c]) for c in raw)
+            _add(found, text, m.start(), m.end(), value, "code", 0.97)
+
+    # Highest-confidence, longest non-overlapping decisions win.
+    accepted: list[Decision] = []
+    for d in sorted(found, key=lambda x: (-x.confidence, -(x.end - x.start), x.start)):
+        if not _overlaps(d, accepted):
+            accepted.append(d)
+
+    out = text
+    for d in sorted(accepted, key=lambda x: x.start, reverse=True):
+        out = out[:d.start] + d.normalized + out[d.end:]
+
+    # Preserve unresolved numeric speech as an explicit structured observation.
+    uncertain: list[Decision] = []
+    occupied = accepted[:]
+    for m in re.finditer(CN_RE, text):
+        if len(m.group()) < 2:
+            continue
+        d = Decision(m.group(), m.group(), "unknown", 0.30, m.start(), m.end(),
+                     _ctx(text, m.start(), m.end()))
+        if not _overlaps(d, occupied):
+            uncertain.append(d)
+    all_decisions = sorted(accepted + uncertain, key=lambda x: x.start)
+    return out, [d.as_dict() for d in all_decisions]