Deploy session-scoped glossary (GitHub 26806a5)

README.md

@@ -13,6 +13,8 @@ short_description: Dịch truyện Trung → Việt
 # HachimiMT — Dịch Trung Việt
 
 Space này là bản demo CPU của HachimiMT/MoxhiMT/HirashibaMT để dịch thử đoạn ngắn.
+Giao diện có glossary theo phiên để chuẩn hóa alias tên riêng/thuật ngữ và cho phép
+nhập/xuất TSV hoặc JSON.
 Vì chạy trên CPU dùng chung, nếu muốn dịch file dài hoặc nguyên bộ truyện, hãy dùng
 notebook GPU hoặc bản local:
 
@@ -21,11 +23,3 @@ notebook GPU hoặc bản local:
 - Bản local Windows/macOS/Linux: tải trong giao diện Space, chạy được offline sau khi tải model.
 
 Mã nguồn chính: https://github.com/ngocdang8311/qt2
-
-## Model revisions deployed
-
-- `DanVP/MoxhiMT-30`: tokenizer hotfix `v4.0.1` (`d0a8923e`).
-- `DanVP/MoxhiMT-60`: tokenizer hotfix (`3ae60c79`).
-- `DanVP/MoxhiMT-30-QT`: tokenizer hotfix (`4f144362`).
-
-Updated 2026-07-14. A cold Space deployment fetches these current Hub revisions.

README_hachimimt_colab.md

@@ -159,6 +159,8 @@ Repo mô hình trên Hugging Face:
 - 📄 **Dán văn bản hoặc tải file `.txt`** (tự nhận mã GB18030 / Big5 / UTF-8, hỗ
   trợ cả phồn thể lẫn giản thể).
 - ⚙️ **Chọn mô hình** và **beam** (đánh đổi tốc độ ↔ chất lượng).
+- 📚 **Glossary tên riêng/thuật ngữ**: nhập cặp Trung → Việt, alias tùy chọn, hỗ
+  trợ nhập/xuất TSV/JSON; cột loại entity không bắt buộc.
 - 🈶 **Chuẩn hóa xưng hô Hán-Việt** (tùy chọn nâng cao, thử nghiệm): chuyển 哥哥/姐姐…
   sang ca ca / tỷ tỷ… theo văn phong tiên hiệp, nhận diện bối cảnh cổ trang vs hiện
   đại. Mặc định tắt — bật khi cần.
@@ -173,11 +175,12 @@ Repo mô hình trên Hugging Face:
 
 - **Tên riêng hiếm có thể dịch lệch giữa các đoạn.** Mô hình nhỏ nên một số tên ít
   gặp có thể ra vài biến thể Hán-Việt khác nhau trong cùng truyện. Đây chính là chỗ
-  **từ điển pin tay của QuickTranslator vẫn nhỉnh hơn** — bạn cố định một tên là cố
-  định mãi. (HachimiMT không cho sửa từ điển.)
+  glossary hỗ trợ chuẩn hóa các alias đã biết. Bản glossary an toàn hiện không tự
+  chèn tên nếu mô hình bỏ hẳn entity.
 - **Cần GPU để nhanh.** CPU (bản máy / demo HF) chậm hơn nhiều lần; dịch nguyên bộ
   truyện trên CPU không thực tế.
-- **Mô hình cố định**, không tùy biến thuật ngữ / không thêm từ điển riêng.
+- **Glossary chưa phải constrained decoding**: nó chuẩn hóa alias sau dịch thay vì
+  ép model sinh một entity bằng placeholder.
 - Mô hình nhỏ → đôi khi vẫn có câu khó hiểu ở đoạn quá dài hoặc nội dung hiếm gặp.
 
 Nếu những điểm trên là then chốt với bạn (đọc offline tuyệt đối, kiểm soát chặt

hachimimt-local.zip

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:e8f2f549b3c44a83304fbe7b333d642558737c195f984d0cd9d14b734d3c8137
-size 128715
+oid sha256:7765446cc8ea375a105e4127d4e129dcf7745c31d7362ae9e399170b5b02a6c3
+size 135912

src/app.py

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
-import atexit
-import html
-import os
-import platform
+import atexit
+import hashlib
+import html
+import os
+import platform
 import tempfile
 import time
 import unicodedata
@@ -43,11 +44,19 @@ def _patch_windows_platform() -> None:
 
 _patch_windows_platform()
 
-import gradio as gr
-
-import hardware
-from hardware import detect_hardware_profile
-from progress_tracker import finish_progress, reset_progress, set_progress, snapshot
+import gradio as gr
+
+from glossary import (
+    GlossaryValidationError,
+    apply_glossary_rows,
+    compile_glossary,
+    glossary_table_rows,
+    read_glossary_file,
+    write_glossary_file,
+)
+import hardware
+from hardware import detect_hardware_profile
+from progress_tracker import finish_progress, reset_progress, set_progress, snapshot
 from text_preprocess import (
     NORMALIZE_AUTO,
     NORMALIZE_NONE,
@@ -79,9 +88,11 @@ def _env_float(name: str, default: float, *, min_value: float = 0.0, max_value:
 
 
 ROOT = Path(__file__).resolve().parent.parent
-EXPORTS_DIR = ROOT / "exports"
-PID_FILE = ROOT / ".hachimimt.pid"
-APP_PORT = 7860
+EXPORTS_DIR = ROOT / "exports"
+PID_FILE = ROOT / ".hachimimt.pid"
+APP_PORT = 7860
+EXPORT_FILE_PREFIX = "hachimi_export_"
+SPACE_EXPORT_MAX_AGE_SECONDS = 24 * 60 * 60
 # HF Space tự set SPACE_ID. Khi ở Space: CPU-only (ẩn nút cài GPU/torch), process
 # do HF quản (bỏ PID file), HF tự lo host/port (không ép 127.0.0.1). App chạy
 # local KHÔNG đổi gì.
@@ -1233,23 +1244,121 @@ def cap_input_for_space(source: str) -> tuple[str, str]:
     return capped, notice
 
 
-def _exception_message(exc: Exception) -> str:
-    return str(exc).strip() or exc.__class__.__name__
-
-
-def export_translation(full_text: str, filename_stem: str) -> str | None:
-    if not full_text.strip():
-        raise gr.Error("Chưa có bản dịch để xuất.")
-
-    safe_stem = "".join(c if c.isalnum() or c in "-_" else "_" for c in filename_stem) or "translation"
-    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-    out_path = EXPORTS_DIR / f"{safe_stem}_vi_{timestamp}.txt"
-    out_path.write_text(full_text, encoding="utf-8")
-    return str(out_path)
-
-
-def _format_duration(seconds: float) -> str:
-    """Định dạng thời gian gọn: '12,3 giây' hoặc '2 phút 5 giây'."""
+def _exception_message(exc: Exception) -> str:
+    return str(exc).strip() or exc.__class__.__name__
+
+
+def _request_session_token(request: gr.Request | None) -> str:
+    """Return a short, non-reversible scope token without exposing session_hash."""
+    session_hash = str(getattr(request, "session_hash", "") or "")
+    if not session_hash:
+        return "local"
+    return hashlib.sha256(session_hash.encode("utf-8")).hexdigest()[:12]
+
+
+def _unique_export_path(
+    kind: str,
+    suffix: str,
+    request: gr.Request | None = None,
+) -> Path:
+    safe_kind = "".join(c if c.isalnum() or c in "-_" else "_" for c in kind)
+    safe_kind = safe_kind.strip("_")[:80] or "file"
+    safe_suffix = suffix.lower().lstrip(".")
+    if safe_suffix not in {"txt", "tsv", "json"}:
+        raise ValueError(f"Unsupported export suffix: {suffix}")
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    session_token = _request_session_token(request)
+    nonce = uuid.uuid4().hex
+    return EXPORTS_DIR / (
+        f"{EXPORT_FILE_PREFIX}{safe_kind}_{timestamp}_{session_token}_{nonce}.{safe_suffix}"
+    )
+
+
+def _cleanup_old_space_exports(
+    *,
+    now: float | None = None,
+    max_age_seconds: int = SPACE_EXPORT_MAX_AGE_SECONDS,
+) -> None:
+    """Best-effort cleanup of this app's old ephemeral exports on HF Space only."""
+    if not IS_HF_SPACE:
+        return
+    cutoff = (time.time() if now is None else now) - max_age_seconds
+    try:
+        candidates = list(EXPORTS_DIR.iterdir())
+    except OSError:
+        return
+    for path in candidates:
+        if (
+            not path.is_file()
+            or not path.name.startswith(EXPORT_FILE_PREFIX)
+            or path.suffix.lower() not in {".txt", ".tsv", ".json"}
+        ):
+            continue
+        try:
+            if path.stat().st_mtime < cutoff:
+                path.unlink()
+        except OSError:
+            continue
+
+
+def export_translation(
+    full_text: str,
+    filename_stem: str,
+    request: gr.Request | None = None,
+) -> str | None:
+    if not full_text.strip():
+        raise gr.Error("Chưa có bản dịch để xuất.")
+
+    safe_stem = "".join(c if c.isalnum() or c in "-_" else "_" for c in filename_stem) or "translation"
+    _cleanup_old_space_exports()
+    out_path = _unique_export_path(f"{safe_stem}_vi", "txt", request)
+    out_path.write_text(full_text, encoding="utf-8")
+    return str(out_path)
+
+
+def export_translation_ui(
+    full_text: str,
+    request: gr.Request | None = None,
+) -> str | None:
+    return export_translation(full_text, "hachimimt", request)
+
+
+def import_glossary_ui(file_obj) -> tuple[list[list[object]], str]:
+    if file_obj is None:
+        raise gr.Error("Chọn file glossary .tsv hoặc .json.")
+    try:
+        rows = read_glossary_file(Path(file_obj))
+    except (OSError, GlossaryValidationError) as exc:
+        raise gr.Error(f"Không nạp được glossary: {_exception_message(exc)}") from exc
+    return rows, f"Đã nạp **{len(rows)}** mục từ `{Path(file_obj).name}`."
+
+
+def export_glossary_ui(
+    rows,
+    file_format: str,
+    request: gr.Request | None = None,
+) -> tuple[str, str]:
+    try:
+        table_rows = glossary_table_rows(rows)
+    except GlossaryValidationError as exc:
+        raise gr.Error(f"Glossary không hợp lệ: {_exception_message(exc)}") from exc
+    if not table_rows:
+        raise gr.Error("Glossary đang trống.")
+
+    file_format = (file_format or "tsv").strip().lower()
+    if file_format not in {"tsv", "json"}:
+        raise gr.Error("Định dạng glossary phải là TSV hoặc JSON.")
+    _cleanup_old_space_exports()
+    out_path = _unique_export_path("glossary", file_format, request)
+    try:
+        write_glossary_file(out_path, table_rows, file_format=file_format)
+    except (OSError, GlossaryValidationError) as exc:
+        raise gr.Error(f"Không xuất được glossary: {_exception_message(exc)}") from exc
+    return str(out_path), f"Đã xuất **{len(table_rows)}** mục glossary."
+
+
+def _format_duration(seconds: float) -> str:
+    """Định dạng thời gian gọn: '12,3 giây' hoặc '2 phút 5 giây'."""
     if seconds < 60:
         return f"{seconds:.1f} giây".replace(".", ",")
     minutes = int(seconds // 60)
@@ -1489,9 +1598,9 @@ def _apply_postprocess_rows(
     }
 
 
-def _translate_run(
-    source: str,
-    model_key: str,
+def _translate_run(
+    source: str,
+    model_key: str,
     backend: str,
     beam_size: float,
     chunk_mode: str,
@@ -1499,21 +1608,29 @@ def _translate_run(
     honorific_kinship: bool,
     honorific_pronouns: bool,
     pronoun_harmonizer_v9: bool,
-    auto_batch: bool,
-    manual_batch: float,
-    *,
-    filename_stem: str,
-    summary_prefix: str,
-    source_kind: str,
+    auto_batch: bool,
+    manual_batch: float,
+    *,
+    glossary_rows=None,
+    filename_stem: str,
+    summary_prefix: str,
+    source_kind: str,
     session: str | None = None,
 ) -> Iterator[tuple]:
     try:
         source, space_cap_notice = cap_input_for_space(source)
-        original_source = source
-        source = normalize_chinese_text(source, normalize_mode)
-        normalize_msg = normalization_message(original_source, source, normalize_mode)
-        honorific_kinship = bool(honorific_kinship)
-        honorific_pronouns = bool(honorific_pronouns)
+        original_source = source
+        source = normalize_chinese_text(source, normalize_mode)
+        normalize_msg = normalization_message(original_source, source, normalize_mode)
+        try:
+            glossary_entries = compile_glossary(
+                glossary_rows,
+                normalize_source=lambda term: normalize_chinese_text(term, normalize_mode),
+            )
+        except GlossaryValidationError as exc:
+            raise gr.Error(f"Glossary không hợp lệ: {_exception_message(exc)}") from exc
+        honorific_kinship = bool(honorific_kinship)
+        honorific_pronouns = bool(honorific_pronouns)
         honorific_on = honorific_kinship or honorific_pronouns
         pronoun_harmonizer_v9 = bool(pronoun_harmonizer_v9)
 
@@ -1573,13 +1690,26 @@ def _translate_run(
             source_text=source,
             chunk_mode=chunk_mode,
         )
-        genre_decision = postprocess_report.get("genre_decision")
-        honorific_changed = int(postprocess_report.get("honorific_changed") or 0)
-        honorific_pronouns_effective = bool(postprocess_report.get("honorific_pronouns_effective"))
-        pronoun_report = postprocess_report.get("pronoun_report") or {}
-        postprocess_warning = str(postprocess_report.get("warning") or "")
-
-        # KHÔNG auto-tạo file .txt mỗi lần dịch (tránh rác temp) — file chỉ sinh
+        genre_decision = postprocess_report.get("genre_decision")
+        honorific_changed = int(postprocess_report.get("honorific_changed") or 0)
+        honorific_pronouns_effective = bool(postprocess_report.get("honorific_pronouns_effective"))
+        pronoun_report = postprocess_report.get("pronoun_report") or {}
+        postprocess_warning = str(postprocess_report.get("warning") or "")
+        glossary_report = None
+        glossary_warning = ""
+        if glossary_entries:
+            before_glossary_rows = rows
+            before_glossary_text = full_text
+            try:
+                rows, glossary_report = apply_glossary_rows(rows, glossary_entries)
+                if glossary_report.changed_rows:
+                    full_text = rebuild_paragraph_layout(source, rows)
+            except Exception as exc:
+                rows = before_glossary_rows
+                full_text = before_glossary_text
+                glossary_warning = _exception_message(exc)
+
+        # KHÔNG auto-tạo file .txt mỗi lần dịch (tránh rác temp) — file chỉ sinh
         # khi user bấm nút "Xuất bản dịch .txt" (export_btn.click). Bản dịch đầy đủ
         # vẫn nằm trong ô + full_text_state để xuất khi cần.
         download_path = None
@@ -1616,8 +1746,21 @@ def _translate_run(
                 f"route `{pronoun_report.get('route', 'n/a')}`."
             )
         route_note = f" Route hậu kỳ `{genre_decision.route}`." if genre_decision and (honorific_on or pronoun_harmonizer_v9) else ""
-        warning_note = f" Hậu kỳ lỗi, đã giữ bản dịch thô: {postprocess_warning}." if postprocess_warning else ""
-        summary = f"{summary_prefix} **{chunk_count}** chunk · {time_note}. {normalize_msg}{fallback_note}{route_note}{honorific_note}{pronoun_note}{warning_note}{space_cap_notice}"
+        warning_note = f" Hậu kỳ lỗi, đã giữ bản dịch thô: {postprocess_warning}." if postprocess_warning else ""
+        glossary_note = ""
+        if glossary_report is not None:
+            glossary_note = (
+                f" Glossary: {glossary_report.entries} mục, "
+                f"{glossary_report.source_hits} hit nguồn; "
+                f"{glossary_report.replacements} thay thế, "
+                f"{glossary_report.satisfied} đã đúng, "
+                f"{glossary_report.unresolved} chưa khớp alias."
+            )
+        elif glossary_warning:
+            glossary_note = (
+                f" Glossary lỗi, đã giữ bản trước glossary: {glossary_warning}."
+            )
+        summary = f"{summary_prefix} **{chunk_count}** chunk · {time_note}. {normalize_msg}{fallback_note}{route_note}{honorific_note}{pronoun_note}{warning_note}{glossary_note}{space_cap_notice}"
         status = f"{status} · {time_note}"
         finish_progress(
             f"Hoàn tất — {chunk_count} chunk trong {duration} (100%)",
@@ -1645,7 +1788,7 @@ def _translate_run(
         raise
 
 
-def translate_text_ui(
+def translate_text_ui(
     source: str,
     model_key: str,
     backend: str,
@@ -1655,10 +1798,11 @@ def translate_text_ui(
     honorific_kinship: bool,
     honorific_pronouns: bool,
     pronoun_harmonizer_v9: bool,
-    auto_batch: bool,
-    manual_batch: float,
-    request: gr.Request | None = None,
-) -> Iterator[tuple]:
+    auto_batch: bool,
+    manual_batch: float,
+    glossary_rows=None,
+    request: gr.Request | None = None,
+) -> Iterator[tuple]:
     if not source.strip():
         raise gr.Error("Nhập văn bản tiếng Trung cần dịch.")
 
@@ -1672,16 +1816,17 @@ def translate_text_ui(
         honorific_kinship,
         honorific_pronouns,
         pronoun_harmonizer_v9,
-        auto_batch,
-        manual_batch,
-        filename_stem="hachimimt",
+        auto_batch,
+        manual_batch,
+        glossary_rows=glossary_rows,
+        filename_stem="hachimimt",
         summary_prefix="Đã dịch",
         source_kind="text",
         session=_session_key(request),
     )
 
 
-def translate_file_ui(
+def translate_file_ui(
     file_obj,
     model_key: str,
     backend: str,
@@ -1691,10 +1836,11 @@ def translate_file_ui(
     honorific_kinship: bool,
     honorific_pronouns: bool,
     pronoun_harmonizer_v9: bool,
-    auto_batch: bool,
-    manual_batch: float,
-    request: gr.Request | None = None,
-) -> Iterator[tuple]:
+    auto_batch: bool,
+    manual_batch: float,
+    glossary_rows=None,
+    request: gr.Request | None = None,
+) -> Iterator[tuple]:
     if file_obj is None:
         raise gr.Error("Chọn file .txt cần dịch.")
 
@@ -1716,9 +1862,10 @@ def translate_file_ui(
         honorific_kinship,
         honorific_pronouns,
         pronoun_harmonizer_v9,
-        auto_batch,
-        manual_batch,
-        filename_stem=path.stem,
+        auto_batch,
+        manual_batch,
+        glossary_rows=glossary_rows,
+        filename_stem=path.stem,
         summary_prefix=f"Đã dịch từ `{path.name}` —",
         source_kind="file",
         session=_session_key(request),
@@ -1926,8 +2073,8 @@ def build_ui() -> gr.Blocks:
                     elem_classes=["status-box"],
                 )
 
-        with gr.Accordion("🧪 Tuỳ chọn chuẩn hóa xưng hô (nâng cao · thử nghiệm)", open=False,
-                          elem_id="honorific-accordion"):
+        with gr.Accordion("🧪 Tuỳ chọn chuẩn hóa xưng hô (nâng cao · thử nghiệm)", open=False,
+                          elem_id="honorific-accordion"):
             with gr.Row():
                 honorific_kinship = gr.Checkbox(
                     value=False, label="Thân tộc (tỷ / muội / ca ca…)", scale=1,
@@ -1946,11 +2093,58 @@ def build_ui() -> gr.Blocks:
                 "khi route cấp chương là văn cổ trang · **Ổn định ngôi hiện đại** chỉ "
                 "rewrite khi route hiện đại, ví dụ th��y/em, mẹ/con, anh/em. Mixed/unknown "
                 "sẽ guard để tránh sửa quá tay.",
-                elem_classes=["honorific-hint"],
-            )
-
-        if IS_HF_SPACE:
-            with gr.Accordion("🚀 Dùng nhanh/mạnh hơn — Google Colab/Kaggle (GPU) hoặc cài máy (offline)", open=False):
+                elem_classes=["honorific-hint"],
+            )
+
+        with gr.Accordion("📚 Tên riêng & thuật ngữ (glossary)", open=False,
+                          elem_id="glossary-accordion"):
+            gr.Markdown(
+                "Glossary chỉ chuẩn hóa **alias tiếng Việt** khi dòng nguồn chứa đúng "
+                "`source_zh`; không tự chèn tên bị model bỏ sót. "
+                "`Loại` chỉ để quản lý và **không bắt buộc**. Nhiều alias ngăn bằng `|`.",
+                elem_classes=["honorific-hint"],
+            )
+            glossary_table = gr.Dataframe(
+                value=[["", "", "", "", True]],
+                headers=[
+                    "Nguồn Trung *",
+                    "Đích Việt *",
+                    "Loại (tùy chọn)",
+                    "Alias Việt (ngăn bằng |)",
+                    "Bật",
+                ],
+                datatype=["str", "str", "str", "str", "bool"],
+                type="array",
+                row_count=3,
+                column_count=5,
+                column_widths=["20%", "22%", "16%", "32%", "10%"],
+                max_height=360,
+                interactive=True,
+                show_row_numbers=True,
+                label="Glossary của phiên hiện tại",
+                elem_id="glossary-table",
+            )
+            with gr.Row():
+                glossary_file = gr.File(
+                    label="Nhập glossary (.tsv / .json)",
+                    file_types=[".tsv", ".json"],
+                    type="filepath",
+                    scale=3,
+                )
+                glossary_format = gr.Radio(
+                    [("TSV", "tsv"), ("JSON", "json")],
+                    value="tsv",
+                    label="Định dạng xuất",
+                    scale=1,
+                )
+            with gr.Row():
+                glossary_import_btn = gr.Button("Nạp vào bảng", variant="secondary")
+                glossary_export_btn = gr.Button("Xuất glossary", variant="secondary")
+            glossary_download = gr.File(label="Tải glossary")
+            glossary_status = gr.Markdown()
+
+        if IS_HF_SPACE:
+            with gr.Accordion("🚀 Dùng nhanh/mạnh hơn — Google Colab/Kaggle (GPU) hoặc cài máy (offline)", open=False):
                 gr.Markdown(
                     f"""Bản Space này chạy **CPU** nên hợp dịch thử. Ba cách dùng nhanh/mạnh hơn:
 
@@ -2021,9 +2215,21 @@ mục `hachimimt`, chạy `setup.bat` (cài thư viện + tải model mặc đ
             inputs=[backend_select, model_select],
             outputs=[engine_hint, model_badge],
         )
-        auto_batch.change(on_auto_batch_toggle, inputs=[auto_batch], outputs=[manual_batch])
-
-        gpu_install_btn.click(
+        auto_batch.change(on_auto_batch_toggle, inputs=[auto_batch], outputs=[manual_batch])
+        glossary_import_btn.click(
+            import_glossary_ui,
+            inputs=[glossary_file],
+            outputs=[glossary_table, glossary_status],
+            queue=False,
+        )
+        glossary_export_btn.click(
+            export_glossary_ui,
+            inputs=[glossary_table, glossary_format],
+            outputs=[glossary_download, glossary_status],
+            queue=False,
+        )
+
+        gpu_install_btn.click(
             lambda: gr.update(visible=True),
             outputs=[gpu_install_log],
         ).then(
@@ -2134,11 +2340,11 @@ mục `hachimimt`, chạy `setup.bat` (cài thư viện + tải model mặc đ
         ], feedback_context_state)
         translate_outputs = [progress_html, progress_active_state, *result_outputs]
 
-        translate_inputs = [
-            model_select, backend_select, beam_size, chunk_mode, normalize_mode,
-            honorific_kinship, honorific_pronouns, pronoun_harmonizer_v9,
-            auto_batch, manual_batch,
-        ]
+        translate_inputs = [
+            model_select, backend_select, beam_size, chunk_mode, normalize_mode,
+            honorific_kinship, honorific_pronouns, pronoun_harmonizer_v9,
+            auto_batch, manual_batch, glossary_table,
+        ]
 
         text_event = text_btn.click(
             prepare_text_progress_ui,
@@ -2176,10 +2382,10 @@ mục `hachimimt`, chạy `setup.bat` (cài thư viện + tải model mặc đ
             concurrency_id="translate",
         )
 
-        export_btn.click(
-            lambda text: export_translation(text, "hachimimt"),
-            inputs=[full_text_state], outputs=[download_file],
-        )
+        export_btn.click(
+            export_translation_ui,
+            inputs=[full_text_state], outputs=[download_file],
+        )
 
         progress_timer = gr.Timer(0.5, active=True)
         progress_timer.tick(

src/glossary.py

@@ -0,0 +1,351 @@
+"""User glossary parsing and safe post-translation canonicalization.
+
+The MVP intentionally does not inject placeholders into model input.  It only
+canonicalizes an explicitly listed Vietnamese alias when the corresponding
+Chinese term is present in the same source row.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import re
+import unicodedata
+from collections.abc import Callable, Iterable, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+GLOSSARY_HEADERS = ("source_zh", "target_vi", "type", "aliases_vi", "enabled")
+MAX_GLOSSARY_BYTES = 2_000_000
+MAX_GLOSSARY_ROWS = 5_000
+
+
+class GlossaryValidationError(ValueError):
+    """Raised when active glossary rows are ambiguous or incomplete."""
+
+
+@dataclass(frozen=True, slots=True)
+class GlossaryEntry:
+    source_zh: str
+    target_vi: str
+    entry_type: str = ""
+    aliases_vi: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class GlossaryReport:
+    entries: int
+    source_hits: int
+    replacements: int
+    satisfied: int
+    unresolved: int
+    changed_rows: int
+    unresolved_terms: tuple[str, ...] = ()
+
+
+def _clean_text(value: Any) -> str:
+    if value is None:
+        return ""
+    try:
+        if value != value:  # NaN from a pandas-backed Dataframe.
+            return ""
+    except (TypeError, ValueError):
+        pass
+    return unicodedata.normalize("NFC", str(value).strip())
+
+
+def _aliases_cell(value: Any) -> str:
+    if isinstance(value, (list, tuple, set)):
+        return "|".join(_clean_text(item) for item in value if _clean_text(item))
+    return _clean_text(value)
+
+
+def _parse_enabled(value: Any) -> bool:
+    if value is None or _clean_text(value) == "":
+        return True
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return bool(value)
+    normalized = _clean_text(value).lower()
+    if normalized in {"1", "true", "yes", "y", "on", "x", "✓", "bật"}:
+        return True
+    if normalized in {"0", "false", "no", "n", "off", "✗", "tắt"}:
+        return False
+    raise GlossaryValidationError(
+        f"Giá trị enabled không hợp lệ: {value!r}; dùng true/false hoặc 1/0."
+    )
+
+
+def glossary_table_rows(value: Any) -> list[list[Any]]:
+    """Coerce Gradio/pandas/JSON values to the five-column UI table shape."""
+    if value is None:
+        return []
+    if hasattr(value, "values") and hasattr(value.values, "tolist"):
+        value = value.values.tolist()
+    elif hasattr(value, "tolist") and not isinstance(value, (str, bytes, dict)):
+        value = value.tolist()
+    if isinstance(value, dict) and "data" in value:
+        value = value["data"]
+    if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
+        raise GlossaryValidationError("Glossary phải là một bảng hoặc danh sách các dòng.")
+
+    rows: list[list[Any]] = []
+    for row_index, raw_row in enumerate(value, start=1):
+        if isinstance(raw_row, dict):
+            row = [
+                raw_row.get("source_zh", raw_row.get("source", "")),
+                raw_row.get("target_vi", raw_row.get("target", "")),
+                raw_row.get("type", raw_row.get("entry_type", "")),
+                raw_row.get("aliases_vi", raw_row.get("aliases", "")),
+                raw_row.get("enabled", True),
+            ]
+        elif isinstance(raw_row, Sequence) and not isinstance(raw_row, (str, bytes)):
+            row = list(raw_row[: len(GLOSSARY_HEADERS)])
+            row.extend([""] * (len(GLOSSARY_HEADERS) - len(row)))
+            if len(raw_row) < len(GLOSSARY_HEADERS):
+                row[-1] = True
+        else:
+            raise GlossaryValidationError(f"Dòng glossary {row_index} không phải một hàng dữ liệu.")
+
+        source = _clean_text(row[0])
+        target = _clean_text(row[1])
+        entry_type = _clean_text(row[2])
+        aliases = _aliases_cell(row[3])
+        if not any((source, target, entry_type, aliases)):
+            continue
+        rows.append([source, target, entry_type, aliases, _parse_enabled(row[4])])
+        if len(rows) > MAX_GLOSSARY_ROWS:
+            raise GlossaryValidationError(
+                f"Glossary vượt giới hạn {MAX_GLOSSARY_ROWS:,} dòng."
+            )
+    return rows
+
+
+def _parse_aliases(value: str, target_vi: str) -> tuple[str, ...]:
+    aliases: list[str] = []
+    seen = {target_vi}
+    for raw_alias in value.split("|"):
+        alias = _clean_text(raw_alias)
+        if alias and alias not in seen:
+            aliases.append(alias)
+            seen.add(alias)
+    aliases.sort(key=lambda item: (-len(item), item))
+    return tuple(aliases)
+
+
+def compile_glossary(
+    value: Any,
+    *,
+    normalize_source: Callable[[str], str] | None = None,
+) -> list[GlossaryEntry]:
+    """Validate active rows, normalize keys, merge identical mappings."""
+    normalize_source = normalize_source or (lambda text: text)
+    merged: dict[str, GlossaryEntry] = {}
+    first_rows: dict[str, int] = {}
+
+    for row_index, row in enumerate(glossary_table_rows(value), start=1):
+        source_zh, target_vi, entry_type, aliases_cell, enabled = row
+        if not enabled:
+            continue
+        if not source_zh or not target_vi:
+            raise GlossaryValidationError(
+                f"Dòng glossary {row_index}: source_zh và target_vi là bắt buộc "
+                "khi mục đang bật."
+            )
+        normalized_source = _clean_text(normalize_source(source_zh))
+        if not normalized_source:
+            raise GlossaryValidationError(
+                f"Dòng glossary {row_index}: source_zh rỗng sau chuẩn hóa."
+            )
+        aliases = _parse_aliases(aliases_cell, target_vi)
+        existing = merged.get(normalized_source)
+        if existing is None:
+            merged[normalized_source] = GlossaryEntry(
+                source_zh=normalized_source,
+                target_vi=target_vi,
+                entry_type=entry_type,
+                aliases_vi=aliases,
+            )
+            first_rows[normalized_source] = row_index
+            continue
+        if existing.target_vi != target_vi:
+            raise GlossaryValidationError(
+                f"Xung đột source_zh {normalized_source!r}: dòng "
+                f"{first_rows[normalized_source]} → {existing.target_vi!r}, "
+                f"dòng {row_index} → {target_vi!r}."
+            )
+        combined_aliases = tuple(
+            sorted(
+                set(existing.aliases_vi).union(aliases),
+                key=lambda item: (-len(item), item),
+            )
+        )
+        merged[normalized_source] = GlossaryEntry(
+            source_zh=normalized_source,
+            target_vi=target_vi,
+            entry_type=existing.entry_type or entry_type,
+            aliases_vi=combined_aliases,
+        )
+
+    return sorted(
+        merged.values(),
+        key=lambda entry: (-len(entry.source_zh), entry.source_zh, entry.target_vi),
+    )
+
+
+def _source_matches(
+    source_text: str,
+    entries: Sequence[GlossaryEntry],
+) -> list[GlossaryEntry]:
+    candidates: list[tuple[int, int, int, GlossaryEntry]] = []
+    for entry_index, entry in enumerate(entries):
+        start = source_text.find(entry.source_zh)
+        while start >= 0:
+            end = start + len(entry.source_zh)
+            candidates.append((start, -len(entry.source_zh), entry_index, entry))
+            start = source_text.find(entry.source_zh, start + 1)
+    candidates.sort(key=lambda item: item[:3])
+
+    selected: list[GlossaryEntry] = []
+    occupied_until = -1
+    for start, negative_length, _entry_index, entry in candidates:
+        end = start - negative_length
+        if start < occupied_until:
+            continue
+        selected.append(entry)
+        occupied_until = end
+    return selected
+
+
+def _literal_pattern(value: str) -> re.Pattern[str]:
+    left = r"(?<!\w)" if value[0].isalnum() or value[0] == "_" else ""
+    right = r"(?!\w)" if value[-1].isalnum() or value[-1] == "_" else ""
+    return re.compile(f"{left}{re.escape(value)}{right}")
+
+
+def _replace_entry_aliases(text: str, entry: GlossaryEntry) -> tuple[str, int, int]:
+    marker_index = 0
+    marker = "\ue000HACHIMI_GLOSSARY\ue001"
+    while marker in text or marker in entry.target_vi or marker in entry.aliases_vi:
+        marker_index += 1
+        marker = f"\ue000HACHIMI_GLOSSARY_{marker_index}\ue001"
+
+    protected, canonical_count = _literal_pattern(entry.target_vi).subn(marker, text)
+    replacements = 0
+    for alias in entry.aliases_vi:
+        protected, count = _literal_pattern(alias).subn(marker, protected)
+        replacements += count
+    return protected.replace(marker, entry.target_vi), replacements, canonical_count
+
+
+def apply_glossary_rows(
+    rows: Iterable[tuple[int, str, str]],
+    entries: Sequence[GlossaryEntry],
+) -> tuple[list[tuple[int, str, str]], GlossaryReport]:
+    """Apply aliases only in rows whose source contains the mapped Chinese term."""
+    output_rows: list[tuple[int, str, str]] = []
+    source_hits = 0
+    replacements = 0
+    satisfied = 0
+    unresolved = 0
+    changed_rows = 0
+    unresolved_terms: set[str] = set()
+
+    for index, source_zh, translated_vi in rows:
+        matches = _source_matches(source_zh, entries)
+        source_hits += len(matches)
+        unique_matches = list(dict.fromkeys(matches))
+        fixed_vi = translated_vi
+        row_changed = False
+        for entry in unique_matches:
+            fixed_vi, replaced_count, canonical_count = _replace_entry_aliases(fixed_vi, entry)
+            if replaced_count:
+                replacements += replaced_count
+                row_changed = True
+            elif canonical_count:
+                satisfied += 1
+            else:
+                unresolved += 1
+                unresolved_terms.add(entry.source_zh)
+        if row_changed:
+            changed_rows += 1
+        output_rows.append((index, source_zh, fixed_vi))
+
+    report = GlossaryReport(
+        entries=len(entries),
+        source_hits=source_hits,
+        replacements=replacements,
+        satisfied=satisfied,
+        unresolved=unresolved,
+        changed_rows=changed_rows,
+        unresolved_terms=tuple(sorted(unresolved_terms)),
+    )
+    return output_rows, report
+
+
+def read_glossary_file(path: Path) -> list[list[Any]]:
+    path = Path(path)
+    if path.suffix.lower() not in {".tsv", ".json"}:
+        raise GlossaryValidationError("Chỉ hỗ trợ glossary .tsv hoặc .json.")
+    if path.stat().st_size > MAX_GLOSSARY_BYTES:
+        raise GlossaryValidationError(
+            f"File glossary vượt giới hạn {MAX_GLOSSARY_BYTES // 1_000_000} MB."
+        )
+
+    if path.suffix.lower() == ".json":
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8-sig"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise GlossaryValidationError(f"Không đọc được JSON glossary: {exc}") from exc
+        if isinstance(payload, dict) and "entries" in payload:
+            payload = payload["entries"]
+        return glossary_table_rows(payload)
+
+    try:
+        with path.open("r", encoding="utf-8-sig", newline="") as handle:
+            raw_rows = list(csv.reader(handle, delimiter="\t"))
+    except UnicodeDecodeError as exc:
+        raise GlossaryValidationError("TSV glossary phải dùng UTF-8.") from exc
+    if not raw_rows:
+        return []
+
+    normalized_header = [_clean_text(item).lower() for item in raw_rows[0]]
+    if {"source_zh", "target_vi"}.issubset(normalized_header):
+        positions = {name: normalized_header.index(name) for name in GLOSSARY_HEADERS if name in normalized_header}
+        data_rows = []
+        for raw_row in raw_rows[1:]:
+            data_rows.append(
+                [
+                    raw_row[positions[name]] if name in positions and positions[name] < len(raw_row)
+                    else (True if name == "enabled" else "")
+                    for name in GLOSSARY_HEADERS
+                ]
+            )
+    else:
+        data_rows = raw_rows
+    return glossary_table_rows(data_rows)
+
+
+def write_glossary_file(path: Path, value: Any, *, file_format: str = "tsv") -> Path:
+    rows = glossary_table_rows(value)
+    path = Path(path)
+    file_format = _clean_text(file_format).lower()
+    if file_format == "json":
+        payload = [
+            dict(zip(GLOSSARY_HEADERS, row, strict=True))
+            for row in rows
+        ]
+        path.write_text(
+            json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
+            encoding="utf-8",
+        )
+        return path
+    if file_format != "tsv":
+        raise GlossaryValidationError("Định dạng xuất glossary phải là tsv hoặc json.")
+    with path.open("w", encoding="utf-8", newline="") as handle:
+        writer = csv.writer(handle, delimiter="\t", lineterminator="\n")
+        writer.writerow(GLOSSARY_HEADERS)
+        writer.writerows(rows)
+    return path