fix(transcriber): resolve codex findings on CJK bounds, duration limits, and timing overlaps

Author: Mapleeeeeeeeeee

src/bilingualsub/core/transcriber.py

@@ -110,56 +110,96 @@ def _transcribe_single(
         raise TranscriptionError(f"Failed to parse transcription result: {e}") from e
 
 
-def _is_short_text(text: str, min_words: int, min_cjk_chars: int) -> bool:
-    """Check if the text segment is too short."""
-    has_cjk = any(
-        "\u3400" <= c <= "\u4dbf"
-        or "\u4e00" <= c <= "\u9fff"
-        or "\uf900" <= c <= "\ufaff"
+def _has_cjk(text: str) -> bool:
+    """Check if the text contains CJK characters (Chinese, Japanese, or Korean)."""
+    return any(
+        "\u4e00" <= c <= "\u9fff"  # CJK Unified Ideographs
+        or "\u3400" <= c <= "\u4dbf"  # Extension A
+        or "\uf900" <= c <= "\ufaff"  # Compatibility Ideographs
+        or "\u3040" <= c <= "\u309f"  # Hiragana
+        or "\u30a0" <= c <= "\u30ff"  # Katakana
+        or "\uac00" <= c <= "\ud7af"  # Hangul Syllables
+        or "\u1100" <= c <= "\u11ff"  # Hangul Jamo
+        or "\u3000" <= c <= "\u303f"  # CJK Symbols and Punctuation (e.g. 。、)
         for c in text
     )
-    if has_cjk:
+
+
+def _is_short_text(text: str, min_words: int, min_cjk_chars: int) -> bool:
+    """Check if the text segment is too short."""
+    if _has_cjk(text):
         return len(text) < min_cjk_chars
     return len(text.split()) < min_words
 
 
 def _split_long_part_by_length(
     part: str,
+    part_duration: float,
+    max_duration_sec: float,
     max_chars: int,
     min_words: int,
     min_cjk_chars: int,
 ) -> list[str]:
-    """Force-split a long text part into length-restricted chunks."""
-    has_cjk = any(
-        "\u3400" <= c <= "\u4dbf"
-        or "\u4e00" <= c <= "\u9fff"
-        or "\uf900" <= c <= "\ufaff"
-        for c in part
-    )
+    """Force-split a long text part into length/duration-restricted chunks."""
+    has_cjk = _has_cjk(part)
+
     if has_cjk:
-        chunk_size = max_chars
+        if part_duration > 0:
+            chars_per_sec = len(part) / part_duration
+            max_len_by_dur = max(1, int(chars_per_sec * max_duration_sec))
+            chunk_size = min(max_chars, max_len_by_dur)
+        else:
+            chunk_size = max_chars
+
         chunks = [part[i : i + chunk_size] for i in range(0, len(part), chunk_size)]
         if len(chunks) > 1 and _is_short_text(chunks[-1], min_words, min_cjk_chars):
             merged_len = len(chunks[-2]) + len(chunks[-1])
-            if merged_len <= max_chars:
+            merged_duration = part_duration * (merged_len / len(part))
+            if merged_len <= max_chars and merged_duration <= max_duration_sec:
                 chunks[-2] = chunks[-2] + chunks[-1]
                 chunks.pop()
         return chunks
 
     words = part.split()
-    chunk_size = max(1, max_chars // 6)
+    if not words:
+        return []
+
+    if part_duration > 0:
+        words_per_sec = len(words) / part_duration
+        max_words_by_dur = max(1, int(words_per_sec * max_duration_sec))
+    else:
+        max_words_by_dur = len(words)
+
     word_chunks = []
-    for i in range(0, len(words), chunk_size):
-        chunk = " ".join(words[i : i + chunk_size])
-        if chunk:
-            word_chunks.append(chunk)
+    current_chunk: list[str] = []
+
+    for word in words:
+        prospective_len = len(" ".join([*current_chunk, word]))
+        if current_chunk and (
+            prospective_len > max_chars or len(current_chunk) >= max_words_by_dur
+        ):
+            word_chunks.append(" ".join(current_chunk))
+            current_chunk = [word]
+        else:
+            current_chunk.append(word)
+
+    if current_chunk:
+        word_chunks.append(" ".join(current_chunk))
+
     if len(word_chunks) > 1 and _is_short_text(
         word_chunks[-1], min_words, min_cjk_chars
     ):
-        merged_len = len(word_chunks[-2]) + len(word_chunks[-1]) + 1
-        if merged_len <= max_chars:
-            word_chunks[-2] = f"{word_chunks[-2]} {word_chunks[-1]}"
+        merged_text = f"{word_chunks[-2]} {word_chunks[-1]}"
+        merged_words = merged_text.split()
+        merged_duration = part_duration * (len(merged_text) / len(part))
+        if (
+            len(merged_text) <= max_chars
+            and len(merged_words) <= max_words_by_dur
+            and merged_duration <= max_duration_sec
+        ):
+            word_chunks[-2] = merged_text
             word_chunks.pop()
+
     return word_chunks
 
 
@@ -193,7 +233,9 @@ def _split_long_entries(
             continue
 
         # Split by punctuation first (sentences and clauses)
-        raw_parts = re.split(r"(?<=[.?!,;\uff0c\uff1b])\s*", entry.text)
+        raw_parts = re.split(
+            r"(?<=[.?!,;\uff0c\uff1b\u3002\uff01\uff1f\u3001])\s*", entry.text
+        )
         parts = [p.strip() for p in raw_parts if p.strip()]
 
         # Merge adjacent parts that are too short,
@@ -212,12 +254,7 @@ def _split_long_entries(
                 or _is_short_text(merged_parts[-1], min_words, min_cjk_chars)
             ) and (part_duration + last_duration <= max_duration_sec):
                 last_part = merged_parts[-1]
-                has_cjk = any(
-                    "\u3400" <= c <= "\u4dbf"
-                    or "\u4e00" <= c <= "\u9fff"
-                    or "\uf900" <= c <= "\ufaff"
-                    for c in last_part + part
-                )
+                has_cjk = _has_cjk(last_part + part)
                 if has_cjk:
                     merged_parts[-1] = f"{last_part}{part}"
                 else:
@@ -231,7 +268,12 @@ def _split_long_entries(
             part_duration = duration * (len(part) / len(entry.text))
             if part_duration > max_duration_sec or len(part) > max_chars:
                 chunks = _split_long_part_by_length(
-                    part, max_chars, min_words, min_cjk_chars
+                    part,
+                    part_duration,
+                    max_duration_sec,
+                    max_chars,
+                    min_words,
+                    min_cjk_chars,
                 )
                 refined_parts.extend(chunks)
             else:
@@ -251,12 +293,12 @@ def _split_long_entries(
             continue
 
         current_time = entry.start
-        for part in refined_parts:
+        for idx, part in enumerate(refined_parts):
             part_ratio = len(part) / total_len
             part_dur = timedelta(seconds=duration * part_ratio)
             part_end = current_time + part_dur
 
-            if part == refined_parts[-1]:
+            if idx == len(refined_parts) - 1:
                 part_end = entry.end
 
             if part_end > current_time:

tests/unit/core/test_transcriber.py

@@ -620,11 +620,8 @@ def test_splits_long_entry_by_word_if_no_punctuation(self):
         )
         res = _split_long_entries([entry], max_duration_sec=6.0, max_chars=80)
         assert len(res) == 2
-        assert (
-            res[0].text
-            == "word1 word2 word3 word4 word5 word6 word7 word8 word9 word10 word11 word12 word13"
-        )
-        assert res[1].text == "word14 word15"
+        assert res[0].text == "word1 word2 word3 word4 word5 word6 word7 word8 word9"
+        assert res[1].text == "word10 word11 word12 word13 word14 word15"
 
     def test_splits_long_entry_by_chars_for_cjk(self):
         # Duration 10 seconds, CJK text with no punctuation
@@ -636,8 +633,8 @@ def test_splits_long_entry_by_chars_for_cjk(self):
         )
         res = _split_long_entries([entry], max_duration_sec=6.0, max_chars=15)
         assert len(res) == 2
-        assert res[0].text == "一二三四五六七八九十一二三四五"
-        assert res[1].text == "六七八九十"
+        assert res[0].text == "一二三四五六七八九十一二"
+        assert res[1].text == "三四五六七八九十"
 
     def test_merges_short_clauses_if_within_limits(self):
         # Entry with duration 8 seconds, text has 3 clauses split by commas