src/knowledge_parsing/config.py

"""Satu-satunya tempat pengaturan pipeline.

⭐ Titik flip GPU ada di sini: ubah `backend` dari "pipeline" ke "vlm" dan tidak
ada berkas lain yang perlu disentuh.

    pipeline  - murni CPU. Terbukti jalan di laptop tanpa GPU (19,0 dtk/hal @ 20 hal).
                Dipakai sementara supaya pipeline bisa dibangun & diuji sekarang.
    vlm       - target resmi (akurasi 95,30). BUTUH GPU, VRAM >= 8 GB.
    hybrid    - butuh GPU juga. Di T4 kecepatannya setara pipeline-CPU, jadi
                GPU di jalur ini membeli mutu, bukan kecepatan.
"""

from __future__ import annotations

import hashlib
import json
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Literal

Backend = Literal["pipeline", "vlm", "hybrid"]

# Akar penyimpanan: root repo (src/knowledge_parsing/ -> ../../).
# Semua keluaran (cache/, runs/) adalah DATA, bukan kode — di-gitignore.
ROOT = Path(__file__).resolve().parent.parent.parent


@dataclass
class PipelineConfig:
    # --- yang paling sering diubah ---
    backend: Backend = "pipeline"

    # ⚠️ MinerU TIDAK punya kode bahasa Indonesia. Dokumen berbahasa Indonesia
    # memakai "ch", yang cakupannya "Chinese, English, Japanese, Chinese
    # Traditional, LATIN" — aksara Latin itu yang mencakup bahasa Indonesia.
    # Mengisi "id" akan gagal: ValueError: Language id not supported.
    lang: str = "ch"

    # --- lokasi ---
    input_dir: Path = ROOT / "data" / "knowledge_docs"
    cache_dir: Path = ROOT / "data" / "knowledge_cache"     # hasil parse, dikunci isi dokumen
    runs_dir: Path = ROOT / "data" / "knowledge_runs"       # hasil normalize + manifest per run

    # --- opsi MinerU ---
    formula_enable: bool = True
    table_enable: bool = True
    effort: str = "medium"               # hanya dipakai backend hybrid
    start_page: int = 0
    end_page: int | None = None

    # --- perilaku ---
    resume: bool = True                  # lewati dokumen yang sudah selesai
    write_debug_pdf: bool = True         # _layout.pdf & _span.pdf: cek mutu paling cepat

    # --- ambang pemeriksaan mutu (lihat checks.py) ---
    min_latex_len: int = 8               # latex lebih pendek dari ini dicurigai terpotong

    extra: dict[str, Any] = field(default_factory=dict)

    def lang_kanonik(self) -> str:
        """Kode bahasa yang benar-benar dipakai MinerU.

        `validate_public_ocr_lang` sekaligus memvalidasi dan mengkanonikkan:
        alias 'latin' / 'en' / 'japan' -> 'ch', dan kode tak dikenal ditolak.
        Ini fungsi yang sama yang memunculkan pesan "Language id not supported".

        Dipakai juga oleh sidik jari cache: tanpa dikanonikkan, `--lang latin`
        dan `--lang ch` menghasilkan kunci cache berbeda padahal hasil
        parsingnya identik — dokumen yang sama akan diparse dua kali.

        Catatan: hanya ImportError yang ditangkap (MinerU belum terpasang).
        ValueError sengaja DIBIARKAN naik — itu justru kode bahasa yang salah,
        dan menelannya berarti kesalahan baru ketahuan setelah model dimuat.
        """
        try:
            from mineru.utils.ocr_language import validate_public_ocr_lang
        except ImportError:
            return self.lang
        return validate_public_ocr_lang(self.lang)

    def periksa(self) -> None:
        """Gagal cepat kalau pengaturannya salah.

        Tanpa ini, kode bahasa yang keliru baru ketahuan setelah ~30 detik
        memuat model — dan pada batch besar, setelah dokumen ke sekian.
        """
        try:
            self.lang_kanonik()
        except ValueError as e:
            raise SystemExit(
                f"{e}\n"
                "  -> Dokumen berbahasa Indonesia memakai 'ch' "
                "(mencakup aksara Latin). MinerU tidak punya kode 'id'."
            ) from e

    def fingerprint(self) -> str:
        """Sidik jari pengaturan yang MEMPENGARUHI hasil parse.

        Dipakai sebagai bagian kunci cache. Opsi yang tidak mengubah hasil
        (resume, runs_dir, ambang pemeriksaan) sengaja tidak ikut, supaya
        mengubahnya tidak membatalkan cache yang masih sah.
        """
        bahan = {
            "backend": self.backend,
            "lang": self.lang_kanonik(),
            "formula_enable": self.formula_enable,
            "table_enable": self.table_enable,
            "effort": self.effort if self.backend == "hybrid" else None,
            "start_page": self.start_page,
            "end_page": self.end_page,
        }
        payload = json.dumps(bahan, sort_keys=True).encode()
        return hashlib.sha256(payload).hexdigest()[:12]

    def to_dict(self) -> dict[str, Any]:
        d = asdict(self)
        for k, v in d.items():
            if isinstance(v, Path):
                d[k] = str(v)
        d["fingerprint"] = self.fingerprint()
        return d