diff --git a/funclip/utils/trans_utils.py b/funclip/utils/trans_utils.py index 3fd2dd4..0de49ae 100644 --- a/funclip/utils/trans_utils.py +++ b/funclip/utils/trans_utils.py @@ -24,35 +24,44 @@ def pre_proc(text): res += text[i]+' ' else: res += text[i] - if res[-1] == ' ': + if res.endswith(' '): res = res[:-1] return res + +def _matching_tokens(text): + """Normalize transcript text into the same token space used by timestamps.""" + return pre_proc(text).translate(ASCII_LOWER_TABLE).split() + + def proc(raw_text, timestamp, dest_text, lang='zh'): - # simple matching - ld = len(dest_text.split()) - normalized_raw_text = raw_text.translate(ASCII_LOWER_TABLE) - normalized_dest_text = dest_text.translate(ASCII_LOWER_TABLE) - if not normalized_dest_text or not timestamp: + # Match in token space so contiguous Chinese text stays aligned with + # token-level timestamps while preserving the existing ASCII case behavior. + raw_tokens = _matching_tokens(raw_text) + dest_tokens = dest_text.translate(ASCII_LOWER_TABLE).split() + if not dest_tokens or not timestamp: return [] - mi, ts = [], [] - offset = 0 - while True: - fi = normalized_raw_text.find( - normalized_dest_text, offset, len(normalized_raw_text) - ) - ti = raw_text[:fi].count(' ') - if fi == -1: - break - offset = fi + len(normalized_dest_text) - end_index = ti + ld - 1 - if ti >= len(timestamp) or end_index >= len(timestamp): - continue - mi.append(fi) - ts.append([timestamp[ti][0]*16, timestamp[end_index][1]*16]) + + if len(raw_tokens) > len(timestamp): + # Keep timestamp indexing safe if a model returns non-timestamped tokens. + raw_tokens = raw_tokens[:len(timestamp)] + + ts = [] + match_len = len(dest_tokens) + start = 0 + last_start = len(raw_tokens) - match_len + while start <= last_start: + end = start + match_len + if raw_tokens[start:end] == dest_tokens and end <= len(timestamp): + ts.append([timestamp[start][0] * 16, timestamp[end - 1][1] * 16]) + # Preserve the previous string-search policy: after a successful + # match, continue after the whole matched range so repeated output + # does not duplicate overlapping source audio. + start = end + else: + start += 1 return ts - def proc_spk(dest_spk, sd_sentences): ts = [] for d in sd_sentences: @@ -94,7 +103,7 @@ def load_state(output_dir): if os.path.exists(output_dir+'/sd_sentences'): with open(output_dir+'/sd_sentences') as fin: line = fin.read() - state['sd_sentences'] = eval(line) + state['sd_sentences'] = eval(line) return state def convert_pcm_to_float(data): diff --git a/tests/test_trans_utils_matching.py b/tests/test_trans_utils_matching.py new file mode 100644 index 0000000..c715f54 --- /dev/null +++ b/tests/test_trans_utils_matching.py @@ -0,0 +1,50 @@ +import sys +from pathlib import Path + +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "funclip")) + +from utils.trans_utils import pre_proc, proc +from videoclipper import VideoClipper + + +def test_proc_matches_contiguous_chinese_using_token_timestamps(): + raw = "嗯那么今天我们就简" + timestamps = [ + [230, 290], [590, 650], [710, 770], + [890, 950], [1010, 1070], [1250, 1310], + [1430, 1490], [1610, 1670], [1970, 2030], + ] + + assert proc(raw, timestamps, pre_proc(raw[:8])) == [[3680, 26720]] + assert proc(raw, timestamps, pre_proc("简")) == [[31520, 32480]] + + +def test_proc_preserves_ascii_case_insensitive_matching(): + timestamps = [[0, 100], [100, 200]] + assert proc("Hello WORLD", timestamps, "hello world") == [[0, 3200]] + + +def test_proc_keeps_repeated_matches_non_overlapping(): + timestamps = [[0, 100], [100, 200], [200, 300], [300, 400]] + expected = [[0, 3200], [3200, 6400]] + + assert proc("哈哈哈哈", timestamps, pre_proc("哈哈")) == expected + assert proc("哈 哈 哈 哈", timestamps, pre_proc("哈哈")) == expected + + +def test_clip_does_not_duplicate_audio_for_repeated_matches(): + timestamps = [[0, 100], [100, 200], [200, 300], [300, 400]] + state = { + "audio_input": (16000, np.arange(6400, dtype=np.float64)), + "recog_res_raw": "哈 哈 哈 哈", + "timestamp": timestamps, + "sentences": [], + } + clipper = VideoClipper(None) + + (_, audio), _, _ = clipper.clip("哈哈", 0, 0, state) + + assert len(audio) == 6400 + np.testing.assert_array_equal(audio, state["audio_input"][1])