feat(chunk): paragraph mode giu xuong dong - line_restore.py

src/line_restore.py

@@ -0,0 +1,229 @@
+"""Khôi phục xuống dòng cho chế độ dịch "Theo đoạn".
+
+Khi gom nhiều dòng gốc thành 1 chunk để model có thêm ngữ cảnh, Marian/CT2
+nuốt ký tự xuống dòng → output thành một khối liền. Module này phân bổ lại bản
+dịch về ĐÚNG số dòng gốc theo tỉ lệ (số câu / số ký tự), giữ nguyên bố cục dòng.
+
+Thuật toán port từ Space đã chạy thật `DanVP/MoxhiMT-30-demo` (chế độ "Per chunk"):
+chain heading-aware → sentence-proportional → char-proportional (fallback), mỗi
+bước trả về đúng `len(source_lines)` dòng để ghép 1:1 với dòng gốc.
+"""
+
+from __future__ import annotations
+
+import re
+
+# Câu tiếng Việt: nuốt tới dấu kết câu (kèm dấu đóng ngoặc kép) hoặc hết chuỗi.
+VI_SENTENCE_RE = re.compile(r".+?(?:[.!?…]+[”’\"']*|$)", re.S)
+# Ký tự được coi là ranh giới ngắt hợp lệ khi canh theo ký tự.
+VI_BREAK_CHARS = set(".!?…。！？;；,:，、”’\"'")
+# Dòng tiêu đề chương: "第N章/节/回/卷/部/篇".
+HEADING_RE = re.compile(r"^第[0-9零〇一二三四五六七八九十百千万两]+[章节回卷部篇]")
+
+
+def is_heading_line(line: str) -> bool:
+    """Dòng tiêu đề chương ngắn (giữ riêng 1 dòng, không gộp vào đoạn)."""
+    stripped = (line or "").strip()
+    if not stripped or "\n" in stripped:
+        return False
+    if len(stripped) > 24:
+        return False
+    if re.search(r"[。！？!?；;，,：:\"“”]", stripped):
+        return False
+    return bool(HEADING_RE.match(stripped))
+
+
+def split_vi_sentences(text: str) -> list[str]:
+    """Tách bản dịch tiếng Việt thành danh sách câu (gọn khoảng trắng)."""
+    units = [
+        re.sub(r"\s+", " ", m.group(0)).strip()
+        for m in VI_SENTENCE_RE.finditer(text or "")
+    ]
+    return [u for u in units if u]
+
+
+def restore_line_breaks_by_sentences(
+    source_text: str, target_text: str
+) -> tuple[str | None, str]:
+    """Phân bổ câu bản dịch về các dòng gốc theo tỉ lệ độ dài ký tự nguồn.
+
+    Trả (None, lý do) khi số câu đích < số dòng gốc (không đủ để chia) → caller
+    rơi sang canh-theo-ký-tự.
+    """
+    source_lines = source_text.splitlines()
+    nonblank = [line.strip() for line in source_lines if line.strip()]
+    if len(nonblank) <= 1:
+        return target_text.strip(), "single-line"
+
+    target_units = split_vi_sentences(target_text)
+    if len(target_units) < len(nonblank):
+        return None, "too-few-target-sentences"
+
+    total_source_chars = max(sum(len(line) for line in nonblank), 1)
+    total_target_units = len(target_units)
+    remaining_lines = len(nonblank)
+    source_seen = 0
+    target_cursor = 0
+    restored: list[str] = []
+
+    for line in source_lines:
+        stripped = line.strip()
+        if not stripped:
+            restored.append("")
+            continue
+
+        source_seen += len(stripped)
+        remaining_lines -= 1
+        ideal_end = round(source_seen / total_source_chars * total_target_units)
+        min_end = target_cursor + 1
+        max_end = total_target_units - remaining_lines
+        end = min(max(ideal_end, min_end), max_end)
+        restored.append(" ".join(target_units[target_cursor:end]).strip())
+        target_cursor = end
+
+    if target_cursor < total_target_units:
+        for i in range(len(restored) - 1, -1, -1):
+            if restored[i].strip():
+                restored[i] = (
+                    restored[i] + " " + " ".join(target_units[target_cursor:])
+                ).strip()
+                break
+    return "\n".join(restored).strip(), "sentence-proportional"
+
+
+def find_nearest_break(text: str, desired: int, min_pos: int, max_pos: int) -> int:
+    """Tìm vị trí ngắt gần `desired` nhất rơi vào ranh giới câu/khoảng trắng."""
+    desired = min(max(desired, min_pos), max_pos)
+    window_start = max(min_pos, desired - 80)
+    window_end = min(max_pos, desired + 80)
+    best_idx, best_score = desired, float("inf")
+
+    for idx in range(window_start, window_end + 1):
+        prev = text[idx - 1] if idx > 0 else ""
+        cur = text[idx] if idx < len(text) else ""
+        if prev in VI_BREAK_CHARS or cur.isspace():
+            score = abs(idx - desired)
+            if prev in ".!?…。！？":
+                score -= 8
+            if score < best_score:
+                best_idx, best_score = idx, score
+    return best_idx
+
+
+def restore_line_breaks_by_chars(
+    source_text: str, target_text: str
+) -> tuple[str, str]:
+    """Fallback: chia bản dịch theo tỉ lệ ký tự, ngắt tại ranh giới gần nhất."""
+    source_lines = source_text.splitlines()
+    nonblank = [line.strip() for line in source_lines if line.strip()]
+    if len(nonblank) <= 1:
+        return target_text.strip(), "single-line"
+
+    compact_target = re.sub(r"\s+", " ", target_text or "").strip()
+    if not compact_target:
+        return "", "char-proportional"
+    total_source_chars = max(sum(len(line) for line in nonblank), 1)
+    target_len = len(compact_target)
+    source_seen = 0
+    last_pos = 0
+    parts: list[str] = []
+
+    for line in nonblank[:-1]:
+        source_seen += len(line)
+        desired = round(source_seen / total_source_chars * target_len)
+        remaining_parts = len(nonblank) - len(parts) - 1
+        min_pos = min(last_pos + 1, target_len)
+        max_pos = min(target_len, max(min_pos, target_len - remaining_parts))
+        pos = find_nearest_break(compact_target, desired, min_pos, max_pos)
+        parts.append(compact_target[last_pos:pos].strip())
+        last_pos = pos
+    parts.append(compact_target[last_pos:].strip())
+
+    restored: list[str] = []
+    part_cursor = 0
+    for line in source_lines:
+        if line.strip():
+            restored.append(parts[part_cursor] if part_cursor < len(parts) else "")
+            part_cursor += 1
+        else:
+            restored.append("")
+    return "\n".join(restored).strip(), "char-proportional"
+
+
+def assemble_paragraph_output(
+    source_text: str,
+    chunks: list[str],
+    translations: list[str],
+) -> tuple[list[tuple[int, str, str]], str]:
+    """Ghép kết quả chế độ "Theo đoạn" rồi khôi phục bố cục dòng.
+
+    Trả `(rows, full_text)` trong đó `full_text` có ĐÚNG số dòng (kể cả dòng
+    trống) như văn bản nguồn, và `rows` ghép 1:1 từng dòng-gốc-không-rỗng với
+    dòng-dịch tương ứng (để bảng đối chiếu khớp với bản tải về).
+
+    Quy trình khớp Space `translate_per_chunk`: nối các chunk dịch (tiêu đề chèn
+    \\n để tách riêng), gọn khoảng trắng, rồi `restore_line_breaks`.
+    """
+    parts: list[str] = []
+    for source_chunk, translated_chunk in zip(chunks, translations):
+        translated_chunk = (translated_chunk or "").strip()
+        if not translated_chunk:
+            continue
+        if is_heading_line(source_chunk):
+            parts.append(f"\n{translated_chunk}\n")
+        else:
+            parts.append(translated_chunk)
+    raw_output = " ".join(parts)
+    raw_output = re.sub(r"[ \t]+", " ", raw_output)
+    raw_output = re.sub(r" *\n *", "\n", raw_output).strip()
+
+    full_text, _mode = restore_line_breaks(source_text, raw_output)
+
+    # rows: ghép từng dòng gốc không rỗng với dòng dịch cùng vị trí.
+    source_nonblank = [line.strip() for line in source_text.splitlines() if line.strip()]
+    target_nonblank = [line.strip() for line in full_text.splitlines() if line.strip()]
+    rows: list[tuple[int, str, str]] = []
+    for index, source_line in enumerate(source_nonblank, start=1):
+        translated = target_nonblank[index - 1] if index - 1 < len(target_nonblank) else ""
+        rows.append((index, source_line, translated))
+    return rows, full_text
+
+
+def restore_line_breaks(source_text: str, target_text: str) -> tuple[str, str]:
+    """Khôi phục bố cục dòng của bản dịch khớp với văn bản nguồn.
+
+    Orchestrator: nếu dòng đầu là tiêu đề chương thì tách riêng nó rồi xử lý phần
+    thân; còn lại thử canh-theo-câu trước, không được thì canh-theo-ký-tự. Luôn
+    trả về chuỗi có đúng bố cục dòng (kể cả dòng trống) của nguồn.
+    """
+    source_lines = source_text.splitlines()
+    first_nonblank = next(
+        (i for i, line in enumerate(source_lines) if line.strip()), None
+    )
+    target_lines = [line.strip() for line in (target_text or "").splitlines()]
+    target_lines = [line for line in target_lines if line]
+
+    if (
+        first_nonblank is not None
+        and len(target_lines) > 1
+        and is_heading_line(source_lines[first_nonblank])
+    ):
+        body_start = first_nonblank + 1
+        blank_after_heading = 0
+        while body_start < len(source_lines) and not source_lines[body_start].strip():
+            blank_after_heading += 1
+            body_start += 1
+        body_source = "\n".join(source_lines[body_start:])
+        body_target = "\n".join(target_lines[1:])
+        restored_body, body_mode = restore_line_breaks(body_source, body_target)
+        restored = [""] * first_nonblank
+        restored.append(target_lines[0])
+        restored.extend([""] * blank_after_heading)
+        if restored_body:
+            restored.append(restored_body)
+        return "\n".join(restored).strip(), f"heading+{body_mode}"
+
+    restored, mode = restore_line_breaks_by_sentences(source_text, target_text)
+    if restored is not None:
+        return restored, mode
+    return restore_line_breaks_by_chars(source_text, target_text)