code/error_mine.py

#!/usr/bin/env python3
"""Fouille d'erreurs sur les hypotheses champion (banc 4235 clips, refs connues).
Chaque piste n'est retenue que si son gain NET sur la validation complete est positif :
 A. normalisation apostrophes/typographie   B. paires de substitution frequentes
 C. majuscules systematiques (noms propres) D. chiffres vs lettres  E. hygiene CSV."""
import csv
import json
import unicodedata
from collections import Counter

import jiwer

MAP = json.load(open("/scratch/rehearsal/mapping.json", encoding="utf-8"))
HYPS = {r["ID"]: r["Target"] for r in csv.DictReader(open("/root/subs/sub_rehearsal_final.csv", encoding="utf-8"))}
LANGS = ["lin", "lug", "sna"]


def norm(t):
    return " ".join(str(t).replace("|", " ").split())


def pairs_for(lang):
    rs, hs = [], []
    for anon, info in MAP.items():
        if info["true_lang"] == lang and info["text"].strip():
            rs.append(norm(info["text"]))
            hs.append(norm(HYPS.get(anon, "")))
    return rs, hs


def sc(rs, hs):
    return 0.5 * jiwer.wer(rs, hs) + 0.5 * jiwer.cer(rs, hs)


def apply_map(hs, table):
    out = []
    for h in hs:
        out.append(" ".join(table.get(w, w) for w in h.split()))
    return out


print("=== E. hygiene CSV ===", flush=True)
for name in ("sub_public.csv", "sub_phase2.csv", "sub_rehearsal_final.csv"):
    rows = list(csv.DictReader(open(f"/root/subs/{name}", encoding="utf-8")))
    bad = sum(1 for r in rows if "�" in r["Target"] or "Ã" in r["Target"])
    empt = sum(1 for r in rows if not r["Target"].strip())
    lens = sorted(len(r["Target"].split()) for r in rows)
    print(f"{name}: {len(rows)} lignes | mojibake {bad} | vides {empt} | mots p50/p95 {lens[len(lens)//2]}/{lens[int(len(lens)*0.95)]}", flush=True)

CHAR_VARIANTS = [("’", "'"), ("‘", "'"), ("`", "'"), ("“", '"'), ("”", '"')]
for lang in LANGS:
    rs, hs = pairs_for(lang)
    base = sc(rs, hs)
    print(f"\n===== {lang} | baseline banc {base:.4f} ({len(rs)} clips) =====", flush=True)

    print("--- A. typographie ---", flush=True)
    rjoin, hjoin = " ".join(rs), " ".join(hs)
    for ch, repl in CHAR_VARIANTS:
        cr, chh = rjoin.count(ch), hjoin.count(ch)
        if cr or chh:
            print(f"  car {ch!r}: refs {cr} vs hyps {chh}", flush=True)
    hs2 = [h.translate(str.maketrans({"’": "'", "‘": "'", "`": "'"})) for h in hs]
    s2 = sc(rs, hs2)
    if s2 < base:
        print(f"  normalisation apostrophes: {base:.4f} -> {s2:.4f}  <== MIEUX", flush=True)
    else:
        print(f"  normalisation apostrophes: {s2 - base:+.5f} (neutre/negatif)", flush=True)

    print("--- B. paires de substitution ---", flush=True)
    sub = Counter()
    for r, h in zip(rs, hs):
        rw, hw = r.split(), h.split()
        ops = jiwer.process_words([r], [h])
        for al in ops.alignments[0]:
            if al.type == "substitute":
                for i, j in zip(range(al.ref_start_idx, al.ref_end_idx), range(al.hyp_start_idx, al.hyp_end_idx)):
                    sub[(hw[j], rw[i])] += 1
    cands = [(h, r, n) for (h, r), n in sub.most_common(60) if n >= 8 and h.lower() != r.lower()]
    print(f"  {len(cands)} paires candidates (>=8 occ.)", flush=True)
    kept = {}
    cur = list(hs)
    curs = base
    for h, r, n in cands:
        trial = apply_map(cur, {h: r})
        st = sc(rs, trial)
        if st < curs - 1e-6:
            kept[h] = r
            cur, curs = trial, st
    print(f"  paires retenues: {len(kept)} | banc {base:.4f} -> {curs:.4f} ({curs - base:+.4f})", flush=True)
    if kept:
        print("  ex:", dict(list(kept.items())[:8]), flush=True)

    print("--- C. majuscules systematiques ---", flush=True)
    wc, wcap = Counter(), Counter()
    for l in open(f"/scratch/prep/manifests/waxal_{lang}_train.jsonl", encoding="utf-8"):
        for w in norm(json.loads(l)["text"]).split():
            core = w.strip(".,;:!?\"'")
            if len(core) >= 3:
                wc[core.lower()] += 1
                if core[:1].isupper():
                    wcap[core.lower()] += 1
    always_cap = {w for w in wc if wc[w] >= 5 and wcap[w] / wc[w] >= 0.95}
    table = {}
    for h in hs:
        for w in h.split():
            core = w.strip(".,;:!?\"'")
            if core.lower() in always_cap and core[:1].islower():
                table[w] = w.replace(core, core[:1].upper() + core[1:], 1)
    hs3 = apply_map(hs, table)
    s3 = sc(rs, hs3)
    print(f"  {len(always_cap)} mots toujours-capitalises train | {len(table)} formes corrigees | {base:.4f} -> {s3:.4f} ({s3 - base:+.4f})", flush=True)

    print("--- D. chiffres ---", flush=True)
    rdig = sum(any(c.isdigit() for c in r) for r in rs)
    hdig = sum(any(c.isdigit() for c in h) for h in hs)
    print(f"  refs avec chiffres: {rdig}/{len(rs)} | hyps avec chiffres: {hdig}", flush=True)

print("\nMINE_DONE", flush=True)