fix: align contiguous Chinese clipping with token timestamps

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):

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])