code/kod/faz1_run_all.py

"""
Faz 1 — TAM koşu orchestrator'ı (4 kaynak paralel, retry, ayrı log).

12B token mixture'a göre kaynak hedefleri, faz1_01'i paralel subprocess olarak çağırır.
Her kaynak kendi log'una (kod/data/shards/<src>.log) yazar; çökerse 3 kez retry
(stream baştan, idempotent overwrite). Dekontaminasyon faz1_01 default'unda açık.

Kullanım (arka planda): python kod/faz1_run_all.py
Tek kaynağı tekrar çalıştırmak: python kod/faz1_run_all.py --only tr_fineweb2_hq
"""
import os, sys, subprocess, threading, json, time, argparse

TOTAL = 12_000_000_000
MIX = {
    "en_fineweb_edu":   0.55,   # 6.60B
    "tr_fineweb2_hq":   0.22,   # 2.64B
    "code_codeparrot":  0.13,   # 1.56B
    "math_openwebmath": 0.10,   # 1.20B
}
OUT = "kod/data/shards"
MAX_RETRY = 3
_lock = threading.Lock()


def log(msg):
    with _lock:
        print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)


def run_source(src, target):
    logf = os.path.join(OUT, f"{src}.log")
    env = dict(os.environ, PYTHONUTF8="1", PYTHONIOENCODING="utf-8",
               HF_HUB_DISABLE_SYMLINKS_WARNING="1")
    cmd = [sys.executable, "kod/faz1_01_tokenize_shard.py", "--source", src,
           "--target_tokens", str(target), "--out_dir", OUT]
    for attempt in range(1, MAX_RETRY + 1):
        log(f"{src}: başlıyor (deneme {attempt}/{MAX_RETRY}, hedef {target/1e9:.2f}B)")
        with open(logf, "a", encoding="utf-8") as f:
            f.write(f"\n==== attempt {attempt} @ {time.ctime()} ====\n"); f.flush()
            rc = subprocess.call(cmd, stdout=f, stderr=subprocess.STDOUT, env=env)
        if rc == 0:
            log(f"{src}: ✓ TAMAM (deneme {attempt})"); return
        log(f"{src}: ✗ exit={rc}, 15s sonra retry")
        time.sleep(15)
    log(f"{src}: !!! {MAX_RETRY} deneme başarısız — log: {logf}")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--only", default=None, help="sadece tek kaynak")
    args = ap.parse_args()
    os.makedirs(OUT, exist_ok=True)
    sources = {args.only: MIX[args.only]} if args.only else MIX

    log(f"Faz 1 TAM koşu | {TOTAL/1e9:.0f}B token | {len(sources)} kaynak paralel")
    t0 = time.time()
    threads = []
    for src, frac in sources.items():
        th = threading.Thread(target=run_source, args=(src, int(TOTAL * frac)))
        th.start(); threads.append(th)
    for th in threads:
        th.join()

    log(f"=== TÜM KAYNAKLAR BİTTİ ({(time.time()-t0)/3600:.1f}h) ===")
    grand = 0
    for src in sources:
        mf = os.path.join(OUT, src, "manifest.json")
        if os.path.exists(mf):
            m = json.load(open(mf, encoding="utf-8"))
            grand += m.get("n_tokens", 0)
            log(f"  {src}: {m.get('n_tokens',0)/1e9:.2f}B tok | {len(m.get('shards',[]))} shard "
                f"| decontam-skip {m.get('n_decontam_skipped',0)} ({m.get('n_docs',0)} doc)")
        else:
            log(f"  {src}: MANIFEST YOK — başarısız")
    log(f"  >>> TOPLAM: {grand/1e9:.2f}B token | -> {OUT}/")
    log("Sıradaki: kod/faz1_04_push_hf.py ile HF dataset repo'ya push.")


if __name__ == "__main__":
    main()