fetch_phonikud.py

"""Fetch Phonikud/phonikud-data knesset_nikud_v6 (5M lines, text<TAB>phonemes)
to the rababa-datasets volume.

Source: https://huggingface.co/datasets/Phonikud/phonikud-data
Labels were machine-generated by Dicta's diacritizer (~1k high-frequency
words manually corrected per their changelog). We keep it as a WEAK
pretraining corpus candidate only — never as a gold fine-tune stage
(no-teacher-poison rule). The .7z original is kept alongside the
extracted .txt for provenance.

Usage:
    modal run --detach fetch_phonikud.py
"""

from __future__ import annotations

from pathlib import Path

import modal

datasets_volume = modal.Volume.from_name("rababa-datasets", create_if_missing=True)

URL = "https://huggingface.co/datasets/Phonikud/phonikud-data/resolve/main/knesset_nikud_v6.txt.7z"
DEST = Path("/datasets/hebrew-phonikud")

image = modal.Image.debian_slim(python_version="3.11").pip_install(
    "requests", "py7zr>=0.21"
)

app = modal.App("rababa-fetch-phonikud", image=image)


@app.function(timeout=45 * 60, volumes={"/datasets": datasets_volume})
def fetch() -> dict:
    import py7zr
    import requests

    DEST.mkdir(parents=True, exist_ok=True)
    archive = DEST / "knesset_nikud_v6.txt.7z"
    if not archive.exists():
        print(f"[fetch] {URL}", flush=True)
        with requests.get(URL, stream=True, timeout=120) as r:
            r.raise_for_status()
            with open(archive, "wb") as f:
                for chunk in r.iter_content(chunk_size=1 << 20):
                    f.write(chunk)
        print(f"[fetch] {archive.stat().st_size} bytes", flush=True)

    txt = DEST / "knesset_nikud_v6.txt"
    if not txt.exists():
        print("[extract] ...", flush=True)
        with py7zr.SevenZipFile(archive, mode="r") as z:
            z.extractall(DEST)

    n = 0
    sample: list[str] = []
    with open(txt, encoding="utf-8") as f:
        for line in f:
            n += 1
            if n <= 3:
                sample.append(line.rstrip("\n"))
    print(f"[inspect] {n} lines, {txt.stat().st_size} bytes", flush=True)
    for s in sample:
        print(f"[inspect] {s[:200]}", flush=True)

    datasets_volume.commit()
    return {"lines": n, "bytes": txt.stat().st_size, "sample": sample}


@app.local_entrypoint()
def main():
    print(fetch.remote())