archive/scripts/audit_STATSLING4.py

# -*- coding: utf-8 -*-
"""Audit 4: precision des SEULES regles JOIN qui declenchent ; restauration d'accents ;
point final restreint au lingala ; traits d'union / chiffres."""
import csv, json, os, re, unicodedata
from collections import Counter

OUT = []
def P(*a):
    s = " ".join(str(x) for x in a); OUT.append(s); print(s)

sub = {}
with open("/root/sub_LMA06.csv", encoding="utf-8") as f:
    for row in csv.DictReader(f): sub[row["ID"]] = row["Target"]
lang = json.load(open("/root/test_lang.json", encoding="utf-8"))
hyp = {"lin": [], "sna": []}
for k, v in sub.items():
    if lang.get(k) in hyp: hyp[lang[k]].append(v)
train, val = {"lin": [], "sna": []}, {"lin": [], "sna": []}
for L in ("lin", "sna"):
    for sp, st in (("train", train), ("validation", val)):
        with open("/scratch/prep/manifests/waxal_%s_%s.jsonl" % (L, sp), encoding="utf-8") as f:
            for line in f:
                if line.strip(): st[L].append(json.loads(line)["text"])
tot_w = sum(len(t.split()) for L in hyp for t in hyp[L])
tot_c = sum(len(t) for L in hyp for t in hyp[L])
W_ERR, C_ERR = 0.5/tot_w, 0.5/tot_c
PUNCT = ".,;:!?\"()«»"
def toks(t): return [x for x in (w.strip(PUNCT) for w in t.split()) if x]

# ===== 1. JOIN : precision des seules regles declenchees =====
P("=== 1. JOIN : precision des SEULES regles qui declenchent sur le test (anti biais de selection) ===")
for L in ("lin", "sna"):
    uni = Counter(); bi = Counter()
    for t in train[L]:
        ws=[w.lower() for w in toks(t)]; uni.update(ws); bi.update(zip(ws, ws[1:]))
    uni_v = Counter(); bi_v = Counter()
    for t in val[L]:
        ws=[w.lower() for w in toks(t)]; uni_v.update(ws); bi_v.update(zip(ws, ws[1:]))
    hyp_bi = Counter()
    for t in hyp[L]:
        ws=[w.lower() for w in toks(t)]; hyp_bi.update(zip(ws, ws[1:]))
    for mr in (0.10, 0.25, 0.50):
        rules = {}
        for j, cj in uni.items():
            if cj < 10: continue
            for i in range(2, len(j)-1):
                a, b = j[:i], j[i:]
                if uni.get(a,0) < 20 or uni.get(b,0) < 20: continue
                cb = bi.get((a,b),0)
                if cb <= mr*cj: rules[(a,b)] = (cj, cb)
        fired = {p: c for p, c in hyp_bi.items() if p in rules}
        n = sum(fired.values())
        # precision restreinte aux regles declenchees, mesuree en held-out VAL
        tp = sum(uni_v.get(a+b,0) for (a,b) in fired)
        fp = sum(bi_v.get((a,b),0) for (a,b) in fired)
        prec_v = tp/max(tp+fp,1)
        # precision sur le TRAIN (grand echantillon) pour les memes regles
        tpt = sum(uni.get(a+b,0) for (a,b) in fired)
        fpt = sum(bi.get((a,b),0) for (a,b) in fired)
        prec_t = tpt/max(tpt+fpt,1)
        g = n*2*(2*prec_v-1)
        P("%s | maxRatio=%.2f : %d regles declenchent, %d occurrences" % (L, mr, len(fired), n))
        P("%s |   precision VAL (regles declenchees seulement) = %.2f%% (%d/%d)" % (L, 100*prec_v, tp, tp+fp))
        P("%s |   precision TRAIN (memes regles)              = %.2f%% (%d/%d)" % (L, 100*prec_t, tpt, tpt+fpt))
        P("%s |   GAIN = %.1f err-mot = %+.5f score" % (L, g, g*W_ERR))
        det = sorted(fired.items(), key=lambda x: -x[1])
        P("%s |   detail: %s" % (L, ", ".join("%s+%s x%d [val %d/%d]" % (a,b,c,uni_v.get(a+b,0),bi_v.get((a,b),0)) for (a,b),c in det[:25])))

# ===== 2. ACCENTS =====
P("")
P("=== 2. RESTAURATION D'ACCENTS (mot nu de hyp dont la forme accentuee domine le train) ===")
def deacc(s):
    return "".join(c for c in unicodedata.normalize("NFD", s) if unicodedata.category(c) != "Mn")
for L in ("lin", "sna"):
    uni = Counter()
    for t in train[L]+val[L]: uni.update(w.lower() for w in toks(t))
    groups = {}
    for w, c in uni.items():
        groups.setdefault(deacc(w), Counter())[w] = c
    opp = Counter(); n = 0; risky = 0
    for t in hyp[L]:
        for w in toks(t):
            wl = w.lower()
            if deacc(wl) != wl: continue          # deja accentue
            g = groups.get(wl)
            if not g or len(g) < 2: continue
            bare = g.get(wl, 0)
            acc_best, acc_c = max(((k, v) for k, v in g.items() if k != wl), key=lambda x: x[1])
            if acc_c >= 3 and acc_c > 3*bare:
                opp[(wl, acc_best, bare, acc_c)] += 1; n += 1
            elif acc_c >= 3: risky += 1
    P("%s | %d occurrences ou la forme ACCENTUEE domine (>3x) la forme nue dans le train" % (L, n))
    P("%s | %d occurrences ambigues (ecartees)" % (L, risky))
    for (wl, acc, b, c), k in opp.most_common(20):
        P("      %-18s -> %-18s x%d   [train nu=%d, accentue=%d]" % (wl, acc, k, b, c))
    # precision = part accentuee dans le train pour ces paires
    if n:
        tp = sum(c for (_,_,b,c) in opp); fp = sum(b for (_,_,b,c) in opp)
        p = tp/max(tp+fp,1)
        g = n*(2*p-1)
        P("%s | precision attendue=%.1f%% -> %.1f err-mot + %.1f err-car = %+.5f score" % (L, 100*p, g, g, g*W_ERR+g*C_ERR))

# ===== 3. POINT FINAL restreint au lingala (devhard) =====
P("")
P("=== 3. POINT FINAL : devhard restreint par langue ===")
allh = json.load(open("/root/devhard_allhyps.json", encoding="utf-8"))
refs = {}
for fn in os.listdir("/root/devhard"):
    with open("/root/devhard/"+fn, encoding="utf-8") as f:
        for line in f:
            if line.strip():
                d = json.loads(line); refs[d["id"]] = d.get("text","")
for mdl in ("joint_cont_best", "sna_ps_best"):
    hyps = allh[mdl]
    for pref, name in (("lin_", "lingala"), ("sna_", "shona")):
        pairs = [(h.strip().endswith("."), refs[i].strip().endswith("."))
                 for i, h in hyps.items() if i.startswith(pref) and i in refs and refs[i].strip()]
        if len(pairs) < 30: continue
        n = len(pairs)
        ph = sum(1 for a,b in pairs if a)/n; pr = sum(1 for a,b in pairs if b)/n
        acc = sum(1 for a,b in pairs if a==b)/n
        P("%-16s %-8s n=%4d | hyp '.'=%.3f ref '.'=%.3f | exact=%.3f  TOUJOURS='.'->%.3f  gain=%+.3f"
          % (mdl, name, n, ph, pr, acc, pr, pr-acc))
        # extrapolation sur 446 clips
        d = (pr-acc)*446
        P("%-16s %-8s   -> %+.1f clips sur 446 = %+.5f score" % (mdl, name, d, d*(W_ERR+C_ERR)))

# ===== 4. TRAITS D'UNION / CHIFFRES / APOSTROPHES =====
P("")
P("=== 4. TRAITS D'UNION, CHIFFRES, APOSTROPHES : opportunites concretes ===")
for L in ("lin", "sna"):
    uni = Counter()
    for t in train[L]+val[L]: uni.update(w.lower() for w in toks(t))
    hyp_bi = Counter(); hyp_uni = Counter()
    for t in hyp[L]:
        ws = [w.lower() for w in toks(t)]; hyp_uni.update(ws); hyp_bi.update(zip(ws, ws[1:]))
    # bigram (a,b) dont "a-b" est frequent dans le train
    n = 0; det = Counter()
    for (a, b), c in hyp_bi.items():
        h = a + "-" + b
        if uni.get(h, 0) >= 3: det[(a, b, uni[h])] += c; n += c
    P("%s | traits d'union: %d occurrences ou 'a-b' existe dans le train (>=3)" % (L, n))
    for (a,b,c), k in det.most_common(10): P("      %s-%s x%d [train %d]" % (a,b,k,c))
    # apostrophes internes
    n2 = 0; det2 = Counter()
    for w, c in hyp_uni.items():
        for i in range(1, len(w)):
            cand = w[:i] + "'" + w[i:]
            if uni.get(cand, 0) >= 3 and uni.get(w, 0) < uni[cand]:
                det2[(w, cand, uni.get(w,0), uni[cand])] += c; n2 += c; break
    P("%s | apostrophes internes: %d occurrences ou la forme avec ' domine" % (L, n2))
    for (w,cand,a,b), k in det2.most_common(10): P("      %s -> %s x%d [nu=%d avec'=%d]" % (w,cand,k,a,b))
    P("%s | GAIN combine max = %.1f err-mot = %+.5f score" % (L, (n+n2)*1.0, (n+n2)*(W_ERR+C_ERR)))

with open("/root/audit_STATSLING4_out.txt","w",encoding="utf-8") as f: f.write("\n".join(OUT))
print("\n[OK] /root/audit_STATSLING4_out.txt")