tinymind-native-8b-remote-handoff/bundle/data/thai_grounding_corpus.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
import csv
import hashlib
import json
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
THIRD_PARTY = ROOT / "third_party"


def _sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def _read_csv(path: Path) -> list[dict[str, str]]:
    if not path.exists():
        return []
    with path.open("r", encoding="utf-8-sig", newline="") as f:
        return [dict(row) for row in csv.DictReader(f)]


def _jsonl(path: Path, rows: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="\n") as f:
        for row in rows:
            f.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")


def _record(user: str, assistant: str, source: str, metadata: dict) -> dict:
    return {
        "messages": [
            {
                "role": "system",
                "content": "You are TinyMind Thai grounding specialist. Answer in natural Thai or bilingual Thai-English with exact provenance, no invented facts, and clear limitations.",
            },
            {"role": "user", "content": user},
            {"role": "assistant", "content": assistant},
        ],
        "source": source,
        "metadata": metadata,
    }


@dataclass(frozen=True)
class ThaiGroundingPolicy:
    max_ner_sentences: int = 4000
    skip_ner_sentences: int = 0
    max_code_chars: int = 24000


class ThaiGroundingCorpusBuilder:
    def __init__(self, third_party_root: str | Path = THIRD_PARTY, policy: ThaiGroundingPolicy | None = None):
        self.third_party_root = Path(third_party_root).resolve()
        self.policy = policy or ThaiGroundingPolicy()

    def build(self, out_dir: str | Path) -> dict:
        out = Path(out_dir)
        out.mkdir(parents=True, exist_ok=True)
        records: list[dict] = []
        records.extend(self._province_records())
        records.extend(self._synonym_records())
        records.extend(self._ner_records())
        records.extend(self._mt_opus_records())
        records.extend(self._code_records("nodejs_thailand_id_card_mqtt", "thai_id_card_mqtt_code"))
        records.extend(self._code_records("node_maxmind_db", "node_maxmind_db_code"))

        train_path = out / "thai_grounding_train.jsonl"
        eval_path = out / "thai_grounding_eval.jsonl"
        train_count = 0
        eval_count = 0
        with train_path.open("w", encoding="utf-8", newline="\n") as train_f, eval_path.open("w", encoding="utf-8", newline="\n") as eval_f:
            for idx, row in enumerate(records):
                target = eval_f if idx % 13 == 0 else train_f
                target.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")
                if idx % 13 == 0:
                    eval_count += 1
                else:
                    train_count += 1

        manifest = {
            "schema_version": "tinymind-thai-grounding-corpus-v1",
            "created_at": datetime.now(timezone.utc).isoformat(),
            "records_written": len(records),
            "train_records": train_count,
            "eval_records": eval_count,
            "train_path": str(train_path),
            "eval_path": str(eval_path),
            "source_counts": self._counts(records),
            "claim_gate": {
                "thai_grounding_corpus_ready": len(records) > 0,
                "private_identity_data_included": False,
                "reason": "Thai geographic, lexical, NER, translation-project, and code-learning records are converted from public repositories. No real private ID-card data is generated or embedded.",
            },
        }
        manifest_path = out / "thai_grounding_corpus_manifest.json"
        manifest["manifest_path"] = str(manifest_path)
        manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
        return manifest

    def _province_records(self) -> list[dict]:
        root = self.third_party_root / "thai_province_data" / "formats" / "csv"
        geos = {r["id"]: r for r in _read_csv(root / "geographies.csv")}
        provinces = _read_csv(root / "provinces.csv")
        districts = _read_csv(root / "districts.csv")
        subdistricts = _read_csv(root / "sub_districts.csv")
        province_by_id = {r["id"]: r for r in provinces}
        district_by_id = {r["id"]: r for r in districts}
        district_count: dict[str, int] = {}
        subdistrict_count: dict[str, int] = {}
        for d in districts:
            district_count[d.get("province_id", "")] = district_count.get(d.get("province_id", ""), 0) + 1
        for s in subdistricts:
            district_id = s.get("district_id", "")
            district = district_by_id.get(district_id, {})
            province_id = district.get("province_id", "")
            subdistrict_count[province_id] = subdistrict_count.get(province_id, 0) + 1
        records = []
        for p in provinces:
            geo = geos.get(p.get("geography_id", ""), {})
            records.append(
                _record(
                    f"อธิบายข้อมูลจังหวัด {p.get('name_th')} เป็นไทย-อังกฤษแบบตรวจสอบได้",
                    f"{p.get('name_th')} ({p.get('name_en')}) อยู่ในภูมิภาค {geo.get('name_th', 'ไม่ระบุ')} ({geo.get('name_en', 'n/a')}). "
                    f"ชุดข้อมูลนี้ระบุจำนวนอำเภอ/เขตประมาณ {district_count.get(p.get('id', ''), 0)} และตำบล/แขวงประมาณ {subdistrict_count.get(p.get('id', ''), 0)} รายการตามไฟล์ public thai-province-data.",
                    "thai_province_data",
                    {"province_id": p.get("id"), "source_file": "formats/csv/provinces.csv"},
                )
            )
        return records

    def _synonym_records(self) -> list[dict]:
        path = self.third_party_root / "thai_synonym" / "data.csv"
        rows = _read_csv(path)
        records = []
        for row in rows:
            synonyms = [s for s in row.get("synonym", "").split("|") if s]
            if not row.get("word") or not synonyms:
                continue
            records.append(
                _record(
                    f"คำว่า '{row['word']}' มีคำพ้องความหมายอะไรบ้าง และควรใช้อย่างไร?",
                    "คำพ้องที่พบใน thai-synonym: " + ", ".join(synonyms[:40]) + f". หมวด POS: {row.get('pos', 'n/a')}. ควรเลือกใช้ตามบริบทและระดับภาษา ไม่สรุปว่าทุกคำแทนกันได้ 100%.",
                    "thai_synonym",
                    {"word": row.get("word"), "pos": row.get("pos"), "source_sha256": _sha256_file(path)},
                )
            )
        return records

    def _ner_records(self) -> list[dict]:
        root = self.third_party_root / "thai_ner_data" / "data"
        files = sorted(root.glob("*.conll"))
        records = []
        seen = 0
        skipped = 0
        for path in files:
            tokens: list[tuple[str, str]] = []
            for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
                if not line.strip():
                    if tokens:
                        if skipped < self.policy.skip_ner_sentences:
                            skipped += 1
                            tokens = []
                            continue
                        records.append(self._ner_record(tokens, path))
                        seen += 1
                        tokens = []
                        if seen >= self.policy.max_ner_sentences:
                            return records
                    continue
                parts = line.split()
                if len(parts) >= 2:
                    tokens.append((parts[0], parts[-1]))
            if tokens and seen < self.policy.max_ner_sentences:
                if skipped < self.policy.skip_ner_sentences:
                    skipped += 1
                    continue
                records.append(self._ner_record(tokens, path))
                seen += 1
        return records

    def _ner_record(self, tokens: list[tuple[str, str]], path: Path) -> dict:
        text = "".join(tok for tok, _ in tokens)
        entities = [{"token": tok, "tag": tag} for tok, tag in tokens if tag != "O"][:64]
        return _record(
            "ทำ Thai NER จากประโยคนี้และอธิบาย entity tags:\n" + text[:1000],
            "ผล NER จาก public PyThaiNLP dataset:\n```json\n" + json.dumps(entities, ensure_ascii=False, indent=2) + "\n```",
            "thai_named_entity_recognition_data",
            {"source_file": path.name, "source_sha256": _sha256_file(path)},
        )

    def _mt_opus_records(self) -> list[dict]:
        root = self.third_party_root / "mt_opus"
        records = []
        for rel in ("README.md", "tokenize.py", "run_fairseq.sh", "script_fairseq_eval_for_n_epochs.sh"):
            path = root / rel
            if not path.exists():
                continue
            text = path.read_text(encoding="utf-8", errors="replace")[: self.policy.max_code_chars]
            records.append(
                _record(
                    f"สรุปบทเรียนจาก mt-opus/{rel} สำหรับงานแปลไทย-อังกฤษ",
                    "สาระจากไฟล์:\n```text\n" + text + "\n```\nใช้เป็นความรู้ด้าน pipeline แปลภาษา/ตัดคำ/ประเมินผล ไม่ใช่ผล benchmark ที่อ้างอันดับ.",
                    "vistec_mt_opus",
                    {"source_file": rel, "source_sha256": _sha256_file(path)},
                )
            )
        return records

    def _code_records(self, repo: str, source: str) -> list[dict]:
        root = self.third_party_root / repo
        if not root.exists():
            return []
        records = []
        allowed = {".md", ".js", ".ts", ".json", ".c", ".h", ".py"}
        for path in sorted(root.rglob("*")):
            rel_parts = {p.lower() for p in path.relative_to(root).parts}
            if rel_parts & {".git", "node_modules", "build", "dist", "coverage", ".vs", "debug"}:
                continue
            if not path.is_file() or path.suffix.lower() not in allowed:
                continue
            text = path.read_text(encoding="utf-8", errors="replace")[: self.policy.max_code_chars]
            if not text.strip():
                continue
            records.append(
                _record(
                    f"เรียนรู้โค้ด/เอกสารจาก {repo}: {path.relative_to(root).as_posix()} แบบปลอดภัย",
                    "สรุปเพื่อเรียนรู้สถาปัตยกรรมและ data tooling จาก public repo:\n```text\n" + text + "\n```",
                    source,
                    {"source_file": path.relative_to(root).as_posix(), "source_sha256": _sha256_file(path)},
                )
            )
        return records

    def _counts(self, records: list[dict]) -> dict[str, int]:
        counts: dict[str, int] = {}
        for row in records:
            counts[row["source"]] = counts.get(row["source"], 0) + 1
        return counts