File size: 4,835 Bytes
5840d20 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """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
|