Merge main with timestamp and event mapping fixes

Signed-off-by: LauraGPT <[email protected]>

funasr/utils/postprocess_hotwords.py

@@ -19,6 +19,9 @@ HotwordInput = Union[str, Sequence[str], Mapping[str, str], None]
 
 _EXPLICIT_SEPARATORS = ("=>", "->", "→")
 _TOKEN_PATTERN = re.compile(r"[\u4e00-\u9fff]|[a-zA-Z]+|[0-9]+")
+_HAN_PATTERN = re.compile(r"[\u4e00-\u9fff]")
+_LATIN_PATTERN = re.compile(r"[a-zA-Z]")
+_ASCII_ALNUM = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
 
 _LAZY_PINYIN = None
 _PINYIN_STYLE = None
@@ -76,7 +79,10 @@ def _require_rapidfuzz():
 
 def _to_pinyin_key(text: str) -> str:
     lazy_pinyin, style = _require_pypinyin()
-    return "".join(lazy_pinyin(text, style=style.NORMAL, errors="ignore")).lower()
+    # Keep non-Chinese characters in the key. Dropping them makes a window with
+    # extra punctuation, digits or Latin letters score 1.0 against the target,
+    # and gives every Latin-only target and window the same empty key.
+    return "".join(lazy_pinyin(text, style=style.NORMAL, errors="default")).lower()
 
 
 def _parse_line(line: str) -> Tuple[Optional[str], Optional[str], bool]:
@@ -230,14 +236,21 @@ class PostprocessHotwordMatcher:
                 seen.add(target_s)
                 self.fuzzy_targets.append(target_s)
 
-        self._length_buckets: Dict[int, List[Tuple[str, str]]] = {}
+        self._length_buckets: Dict[int, List[Tuple[str, str, bool, bool]]] = {}
         self._fuzz = None
         if self.fuzzy_targets and self.enable_fuzzy:
             self._fuzz = _require_rapidfuzz()
             _require_pypinyin()
             for target in self.fuzzy_targets:
                 bucket = self._length_buckets.setdefault(len(target), [])
-                bucket.append((target, _to_pinyin_key(target)))
+                bucket.append(
+                    (
+                        target,
+                        _to_pinyin_key(target),
+                        bool(_HAN_PATTERN.search(target)),
+                        bool(_LATIN_PATTERN.search(target)),
+                    )
+                )
 
     def apply_text(self, text: str) -> Tuple[str, List[HotwordMatch]]:
         if not text:
@@ -330,12 +343,27 @@ class PostprocessHotwordMatcher:
                 segment = text[start:end]
                 if not segment or not _TOKEN_PATTERN.search(segment):
                     continue
+                # Do not split a Latin word or a number: with target "Win11",
+                # the window "Win1" inside "Win10" would become "Win110".
+                if start > 0 and text[start - 1] in _ASCII_ALNUM and text[start] in _ASCII_ALNUM:
+                    continue
+                if end < text_len and text[end - 1] in _ASCII_ALNUM and text[end] in _ASCII_ALNUM:
+                    continue
                 segment_py = _to_pinyin_key(segment)
+                segment_han = bool(_HAN_PATTERN.search(segment))
+                segment_latin = bool(_LATIN_PATTERN.search(segment))
 
                 for length in bucket_keys:
-                    for target, target_py in self._length_buckets[length]:
+                    for target, target_py, target_han, target_latin in self._length_buckets[length]:
                         if segment == target:
                             continue
+                        # Lower-cased Latin letters can spell pinyin, so keep
+                        # scripts apart unless the target mixes them: "AI" plus
+                        # a character read "li" must not become a Chinese
+                        # target read "ali", and a Latin target must not
+                        # replace a Chinese character read "ai".
+                        if (segment_latin and not target_latin) or (segment_han and not target_han):
+                            continue
                         score = self._fuzz.ratio(segment_py, target_py) / 100.0
                         if score >= self.threshold:
                             candidates.append(
@@ -351,6 +379,25 @@ class PostprocessHotwordMatcher:
         if not candidates:
             return text, []
 
+        # A target already spelled correctly must not be rewritten into itself
+        # by a neighbouring window that only resembles it. Only candidates for
+        # the same target are dropped, so a short target found inside a longer
+        # one does not block the longer one.
+        exact_spans: Dict[str, List[Tuple[int, int]]] = {}
+        for target in {c.replacement for c in candidates}:
+            spans = exact_spans.setdefault(target, [])
+            idx = text.find(target)
+            while idx >= 0:
+                spans.append((idx, idx + len(target)))
+                idx = text.find(target, idx + 1)
+        candidates = [
+            c
+            for c in candidates
+            if not any(c.start < end and start < c.end for start, end in exact_spans[c.replacement])
+        ]
+        if not candidates:
+            return text, []
+
         selected = _select_non_overlapping(candidates)
         updated = text
         applied: List[HotwordMatch] = []

funasr/utils/postprocess_utils.py

@@ -353,7 +353,7 @@ event_dict = {
     "<|Cry|>": "😭",
     "<|Sneeze|>": "🤧",
     "<|Breath|>": "",
-    "<|Cough|>": "🤧",
+    "<|Cough|>": "😷",
 }
 
 lang_dict = {

funasr/utils/timestamp_tools.py

@@ -189,11 +189,10 @@ def timestamp_sentence(
 
         punc_id = int(punc_id) if punc_id is not None else 1
         sentence_end = timestamp[1] if timestamp is not None else sentence_end
-        sentence_text_seg = (
-            sentence_text_seg[:-1] if sentence_text_seg and sentence_text_seg[-1] == " " else sentence_text_seg
-        )
         if punc_id > 1:
             sentence_text += punc_list[punc_id - 2]
+            if sentence_text_seg.endswith(" "):
+                sentence_text_seg = sentence_text_seg[:-1]
             if return_raw_text:
                 res.append(
                     {

runtime/python/libtorch/funasr_torch/utils/postprocess_utils.py

@@ -262,7 +262,7 @@ event_dict = {
     "<|Cry|>": "😭",
     "<|Sneeze|>": "🤧",
     "<|Breath|>": "",
-    "<|Cough|>": "🤧",
+    "<|Cough|>": "😷",
 }
 
 lang_dict = {

runtime/python/onnxruntime/funasr_onnx/utils/postprocess_utils.py

@@ -316,7 +316,7 @@ event_dict = {
     "<|Cry|>": "😭",
     "<|Sneeze|>": "🤧",
     "<|Breath|>": "",
-    "<|Cough|>": "🤧",
+    "<|Cough|>": "😷",
 }
 
 lang_dict = {

tests/test_postprocess_hotwords.py

@@ -79,6 +79,65 @@ class TestPostprocessHotwordMatcher(unittest.TestCase):
         self.assertTrue(matches)
         self.assertEqual(matches[0].replacement, "科大讯飞")
 
+    def test_fuzzy_leaves_surrounding_text_alone(self):
+        try:
+            import pypinyin  # noqa: F401
+            import rapidfuzz  # noqa: F401
+        except ImportError:
+            self.skipTest("pypinyin and rapidfuzz are required for fuzzy tests")
+
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["科大讯飞"])
+        # Already correct: nothing may change.
+        for text in ["我在科大讯飞工作", "去科大讯飞", "欢迎来到科大讯飞。", "今天科大讯飞，发布了新模型。"]:
+            self.assertEqual(matcher.apply_text(text), (text, []))
+        # Misrecognized: only the hotword changes, adjacent characters stay.
+        self.assertEqual(matcher.apply_text("欢迎来到科大迅飞。")[0], "欢迎来到科大讯飞。")
+        self.assertEqual(matcher.apply_text("科大迅飞2024年报")[0], "科大讯飞2024年报")
+
+    def test_fuzzy_short_target_does_not_block_longer_target(self):
+        try:
+            import pypinyin  # noqa: F401
+            import rapidfuzz  # noqa: F401
+        except ImportError:
+            self.skipTest("pypinyin and rapidfuzz are required for fuzzy tests")
+
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["讯飞", "科大讯飞"])
+        self.assertEqual(matcher.apply_text("可达讯飞发布")[0], "科大讯飞发布")
+
+    def test_fuzzy_latin_target_does_not_match_unrelated_text(self):
+        try:
+            import pypinyin  # noqa: F401
+            import rapidfuzz  # noqa: F401
+        except ImportError:
+            self.skipTest("pypinyin and rapidfuzz are required for fuzzy tests")
+
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["FunASR"])
+        text = "please open the model page today"
+        self.assertEqual(matcher.apply_text(text), (text, []))
+        # Windows must not split a Latin word or a number.
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["AI"])
+        self.assertEqual(matcher.apply_text("he said it"), ("he said it", []))
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["Win11"])
+        self.assertEqual(matcher.apply_text("Win10系统"), ("Win10系统", []))
+
+    def test_fuzzy_keeps_latin_and_chinese_apart(self):
+        try:
+            import pypinyin  # noqa: F401
+            import rapidfuzz  # noqa: F401
+        except ImportError:
+            self.skipTest("pypinyin and rapidfuzz are required for fuzzy tests")
+
+        # "AI" lower-cased is the pinyin "ai"; with the next character it would
+        # otherwise score 0.8571 against "ali".
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["阿里"])
+        self.assertEqual(matcher.apply_text("在AI里面"), ("在AI里面", []))
+        # A Latin target must not replace a Chinese character read "ai".
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["AI"])
+        self.assertEqual(matcher.apply_text("我爱你"), ("我爱你", []))
+        # A target that mixes scripts still corrects its Chinese homophone.
+        matcher = PostprocessHotwordMatcher(fuzzy_targets=["AI助手"])
+        self.assertEqual(matcher.apply_text("打开爱助手")[0], "打开AI助手")
+
     def test_missing_fuzzy_dependency_raises(self):
         with mock.patch.object(
             postprocess_hotwords,

tests/test_rich_transcription_postprocess.py

@@ -0,0 +1,34 @@
+"""Regression tests for SenseVoice rich transcription postprocessing."""
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+RUNTIME_PYTHON = REPO_ROOT / "runtime" / "python"
+POSTPROCESS_UTILS = [
+    REPO_ROOT / "funasr" / "utils" / "postprocess_utils.py",
+    RUNTIME_PYTHON / "onnxruntime" / "funasr_onnx" / "utils" / "postprocess_utils.py",
+    RUNTIME_PYTHON / "libtorch" / "funasr_torch" / "utils" / "postprocess_utils.py",
+]
+
+
+def _load(path):
+    spec = importlib.util.spec_from_file_location(f"_postprocess_utils_{path.parts[-3]}", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.mark.parametrize("path", POSTPROCESS_UTILS, ids=lambda path: path.parts[-3])
+@pytest.mark.parametrize(
+    "event, emoji",
+    [("<|Cough|>", "😷"), ("<|Sneeze|>", "🤧")],
+)
+def test_cough_and_sneeze_events_render_distinct_emojis(path, event, emoji):
+    module = _load(path)
+
+    text = module.rich_transcription_postprocess(f"<|en|><|NEUTRAL|>{event}<|woitn|>hello")
+
+    assert text == f"{emoji}hello"

tests/test_timestamp_tools.py

@@ -53,3 +53,14 @@ def test_timestamp_sentence_en_missing_trailing_timestamp_matches_zh_fallback():
 
     assert en_result[1]["start"] is None
     assert en_result[1]["start"] == zh_result[1]["start"]
+
+
+def test_timestamp_sentence_raw_text_keeps_token_separators():
+    result = timestamp_sentence(
+        punc_id_list=[1, 2, 1, 1, 3],
+        timestamp_postprocessed=[[0, 100], [100, 200], [200, 300], [300, 400], [400, 500]],
+        text_postprocessed="我 用 iphone pro max",
+        return_raw_text=True,
+    )
+
+    assert [sentence["raw_text"] for sentence in result] == ["我 用", "iphone pro max"]