mirror of
https://github.com/modelscope/FunASR.git
synced 2026-10-12 02:19:18 +00:00
The fuzzy key was built with lazy_pinyin(errors="ignore"), which drops every non-Chinese character. A window holding the hotword plus an adjacent full stop, comma, digit or Latin letter got the same key as the target and scored 1.0, and the tie-break on length then picked it, so the replacement deleted that character. A Latin-only target such as "FunASR" had an empty key and matched every Latin window of similar length, overwriting the transcript. Exact occurrences of a target were skipped but not reserved, so an overlapping near match could still rewrite correct text and duplicate the first character of the hotword. Keep non-Chinese characters in the key, and drop a fuzzy candidate when it overlaps an exact occurrence of the target it would write. Other targets can still replace across that span, so a short target found inside a longer one does not block the longer one. Lower-cased Latin letters can spell pinyin, so a window with Latin letters is only compared with targets that contain Latin letters, and a window with Chinese characters only with targets that contain Chinese characters. Targets that mix both scripts are compared with any window. A window also may not start or end inside a run of ASCII letters or digits, so a Latin target cannot rewrite part of a word or a number (with target "Win11", "Win10" no longer becomes "Win110").
223 lines
9.6 KiB
Python
223 lines
9.6 KiB
Python
import importlib.util
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
_MODULE_PATH = Path(__file__).resolve().parents[1] / "funasr" / "utils" / "postprocess_hotwords.py"
|
|
_SPEC = importlib.util.spec_from_file_location("postprocess_hotwords", _MODULE_PATH)
|
|
postprocess_hotwords = importlib.util.module_from_spec(_SPEC)
|
|
sys.modules["postprocess_hotwords"] = postprocess_hotwords
|
|
assert _SPEC.loader is not None
|
|
_SPEC.loader.exec_module(postprocess_hotwords)
|
|
|
|
HotwordMatch = postprocess_hotwords.HotwordMatch
|
|
PostprocessHotwordMatcher = postprocess_hotwords.PostprocessHotwordMatcher
|
|
apply_postprocess_hotwords_to_results = postprocess_hotwords.apply_postprocess_hotwords_to_results
|
|
build_postprocess_hotword_matcher = postprocess_hotwords.build_postprocess_hotword_matcher
|
|
parse_hotword_file = postprocess_hotwords.parse_hotword_file
|
|
parse_postprocess_hotwords = postprocess_hotwords.parse_postprocess_hotwords
|
|
|
|
|
|
class TestPostprocessHotwordParsing(unittest.TestCase):
|
|
def test_parse_list_and_dict(self):
|
|
explicit, fuzzy = parse_postprocess_hotwords(["科大讯飞", "东方财富"])
|
|
self.assertEqual(explicit, {})
|
|
self.assertEqual(fuzzy, ["科大讯飞", "东方财富"])
|
|
|
|
explicit, fuzzy = parse_postprocess_hotwords({"科大迅飞": "科大讯飞", "东方财富": "东方财富"})
|
|
self.assertEqual(explicit, {"科大迅飞": "科大讯飞"})
|
|
self.assertEqual(fuzzy, ["东方财富"])
|
|
|
|
def test_parse_inline_mapping(self):
|
|
explicit, fuzzy = parse_postprocess_hotwords(["撒贝你=>撒贝宁", "康辉"])
|
|
self.assertEqual(explicit, {"撒贝你": "撒贝宁"})
|
|
self.assertEqual(fuzzy, ["康辉"])
|
|
|
|
def test_parse_hotword_file(self):
|
|
with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False) as f:
|
|
f.write("# comment\n")
|
|
f.write("科大讯飞\n")
|
|
f.write("科大迅飞=>科大讯飞\n")
|
|
path = f.name
|
|
|
|
try:
|
|
explicit, fuzzy = parse_hotword_file(path)
|
|
self.assertEqual(explicit, {"科大迅飞": "科大讯飞"})
|
|
self.assertEqual(fuzzy, ["科大讯飞"])
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
class TestPostprocessHotwordMatcher(unittest.TestCase):
|
|
def test_explicit_replace(self):
|
|
matcher = PostprocessHotwordMatcher(
|
|
explicit_map={"撒贝你": "撒贝宁"},
|
|
enable_fuzzy=False,
|
|
)
|
|
text, matches = matcher.apply_text("我非常喜欢撒贝你说的新闻")
|
|
self.assertEqual(text, "我非常喜欢撒贝宁说的新闻")
|
|
self.assertEqual(len(matches), 1)
|
|
self.assertEqual(matches[0].replacement, "撒贝宁")
|
|
self.assertEqual(matches[0].score, 1.0)
|
|
|
|
def test_fuzzy_replace_with_optional_deps(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=["科大讯飞"],
|
|
threshold=0.75,
|
|
)
|
|
text, matches = matcher.apply_text("科大迅飞的语音识别很强")
|
|
self.assertIn("科大讯飞", text)
|
|
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,
|
|
"_require_pypinyin",
|
|
side_effect=ImportError("missing pypinyin"),
|
|
):
|
|
with self.assertRaises(ImportError):
|
|
PostprocessHotwordMatcher(
|
|
fuzzy_targets=["科大讯飞"],
|
|
threshold=0.85,
|
|
)
|
|
|
|
def test_invalid_threshold_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
PostprocessHotwordMatcher(explicit_map={"a": "b"}, threshold=1.5)
|
|
|
|
def test_sentence_info_and_timestamp_preserved(self):
|
|
matcher = PostprocessHotwordMatcher(
|
|
explicit_map={"撒贝你": "撒贝宁"},
|
|
enable_fuzzy=False,
|
|
)
|
|
result = {
|
|
"text": "撒贝你主持节目",
|
|
"timestamp": [[0, 100], [100, 200]],
|
|
"sentence_info": [
|
|
{"text": "撒贝你主持", "sentence": "撒贝你主持", "start": 0, "end": 1000},
|
|
{"text": "节目", "sentence": "节目", "start": 1000, "end": 1500},
|
|
],
|
|
}
|
|
matcher.apply_result(result, return_matches=True)
|
|
self.assertEqual(result["text"], "撒贝宁主持节目")
|
|
self.assertEqual(result["sentence_info"][0]["text"], "撒贝宁主持")
|
|
self.assertEqual(result["sentence_info"][0]["sentence"], "撒贝宁主持")
|
|
self.assertEqual(result["timestamp"], [[0, 100], [100, 200]])
|
|
self.assertEqual(result["postprocess_hotword_matches"][0]["replacement"], "撒贝宁")
|
|
|
|
def test_matcher_reused_for_multiple_results(self):
|
|
build_calls = {"count": 0}
|
|
original_build = build_postprocess_hotword_matcher
|
|
|
|
def counting_build(*args, **kwargs):
|
|
build_calls["count"] += 1
|
|
return original_build(*args, **kwargs)
|
|
|
|
results = [
|
|
{"text": "撒贝你主持", "timestamp": [1]},
|
|
{"text": "康灰播报", "timestamp": [2]},
|
|
]
|
|
cfg = {
|
|
"postprocess_hotwords": {"撒贝你": "撒贝宁", "康灰": "康辉"},
|
|
"return_postprocess_hotword_matches": True,
|
|
}
|
|
with mock.patch.object(
|
|
postprocess_hotwords,
|
|
"build_postprocess_hotword_matcher",
|
|
side_effect=counting_build,
|
|
):
|
|
updated = apply_postprocess_hotwords_to_results(results, cfg)
|
|
|
|
self.assertEqual(build_calls["count"], 1)
|
|
self.assertEqual(updated[0]["text"], "撒贝宁主持")
|
|
self.assertEqual(updated[1]["text"], "康辉播报")
|
|
self.assertEqual(len(updated[0]["postprocess_hotword_matches"]), 1)
|
|
|
|
def test_noop_when_not_configured(self):
|
|
results = [{"text": "不变", "timestamp": [1]}]
|
|
updated = apply_postprocess_hotwords_to_results(results, {})
|
|
self.assertIs(updated, results)
|
|
self.assertEqual(updated[0]["text"], "不变")
|
|
|
|
|
|
class TestHotwordMatch(unittest.TestCase):
|
|
def test_as_dict(self):
|
|
match = HotwordMatch("a", "b", 0.9, 1, 2)
|
|
self.assertEqual(
|
|
match.as_dict(),
|
|
{"original": "a", "replacement": "b", "score": 0.9, "start": 1, "end": 2},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|