phase2_corrected/code/audit_conformity.py

#!/usr/bin/env python3
"""AUDIT GRATUIT (0 soumission) — le plus haut rendement du dossier.
Un clip pathologique coûte ~0.001 du score final ; 5 clips = toute notre avance (0.0056).
Le train WAXAL est LA spec orthographique du test (même pipeline d'annotation).
Compare la soumission championne aux 13960+13665 références du train :
 (1) clips pathologiques (vides, trop courts, boucles de répétition, longueur aberrante)
 (2) inventaire des codepoints : apostrophes U+0027 vs U+2019, NFC/NFD, ɛ/ɔ, espaces
 (3) distribution marginale : ponctuation terminale, casse initiale, espace avant ponctuation
"""
import csv, json, os, sys, unicodedata
from collections import Counter

SUB = os.environ.get("SUB", "/root/sub_SNARESC.csv")
LANGF = "/root/test_lang.json"


def refs_for(lang):
    out = []
    for l in open("/root/devhard/train_%s_min.jsonl" % lang, encoding="utf-8"):
        t = json.loads(l).get("text", "").strip()
        if t:
            out.append(t)
    return out


def main():
    sub = {r["ID"]: r["Target"] for r in csv.DictReader(open(SUB, encoding="utf-8"))}
    lang = json.load(open(LANGF))
    print("=== AUDIT %s (%d clips) ===" % (os.path.basename(SUB), len(sub)))

    # ---------- (1) clips pathologiques ----------
    wl = {k: len(v.split()) for k, v in sub.items()}
    med = sorted(wl.values())[len(wl) // 2]
    print("\n--- (1) CLIPS PATHOLOGIQUES (mediane %d mots) ---" % med)
    bad = []
    for k, v in sub.items():
        w = v.split()
        why = []
        if not v.strip():
            why.append("VIDE")
        elif len(v.strip()) < 3:
            why.append("TRES_COURT(%r)" % v)
        if len(w) > 3 * med:
            why.append("TROP_LONG(%d mots)" % len(w))
        if len(w) >= 6:
            top = Counter(w).most_common(1)[0]
            if top[1] / len(w) > 0.4:
                why.append("MOT_REPETE(%s x%d/%d)" % (top[0], top[1], len(w)))
            grams = [tuple(w[i:i + 3]) for i in range(len(w) - 2)]
            if grams:
                g = Counter(grams).most_common(1)[0]
                if g[1] >= 3:
                    why.append("3GRAM_REPETE(x%d)" % g[1])
        if why:
            bad.append((k, lang.get(k, "?"), "; ".join(why), v[:70]))
    print("clips suspects : %d  (cout potentiel ~%.4f)" % (len(bad), 0.001 * len(bad)))
    for b in bad[:15]:
        print("  %s [%s] %s\n     %r" % b)

    # ---------- (2) codepoints vs references ----------
    print("\n--- (2) CODEPOINTS : soumission vs references du train ---")
    for lg in ("lin", "sna"):
        pred = " ".join(v for k, v in sub.items() if lang.get(k) == lg)
        ref = " ".join(refs_for(lg))
        cp = lambda s: Counter(c for c in s if not c.isalnum() and c != " ")
        cpr, cpp = cp(ref), cp(pred)
        nr, np_ = max(len(ref), 1), max(len(pred), 1)
        print("\n  [%s] ponctuation (taux pour 1000 caracteres) :" % lg)
        keys = sorted(set(cpr) | set(cpp), key=lambda c: -(cpr.get(c, 0) + cpp.get(c, 0)))
        for c in keys[:12]:
            r = 1000.0 * cpr.get(c, 0) / nr
            p = 1000.0 * cpp.get(c, 0) / np_
            flag = ""
            if r > 0.5 and p < 0.1 * r:
                flag = "  <-- ABSENT de nos sorties !"
            if p > 0.5 and r < 0.1 * p:
                flag = "  <-- EN TROP dans nos sorties !"
            print("    %-8s ref %6.2f | nous %6.2f%s" % (repr(c), r, p, flag))
        # apostrophes
        for name, ch in (("ASCII '", "'"), ("typo U+2019", "’")):
            print("    apostrophe %-12s ref %5d | nous %5d" % (name, ref.count(ch), pred.count(ch)))
        # caracteres speciaux lingala
        for ch in ("ɛ", "ɔ"):
            if ref.count(ch) or pred.count(ch):
                print("    %r (%s) ref %d | nous %d" % (ch, unicodedata.name(ch, "?"), ref.count(ch), pred.count(ch)))
        # normalisation unicode
        nfc = sum(1 for k, v in sub.items() if lang.get(k) == lg and unicodedata.normalize("NFC", v) != v)
        print("    clips non-NFC : %d" % nfc)
        # caracteres jamais vus dans les refs
        setr = set(ref)
        unseen = Counter(c for c in pred if c not in setr and not c.isspace())
        if unseen:
            print("    CARACTERES ABSENTS DES REFS : %s" % dict(unseen.most_common(8)))

    # ---------- (3) distributions marginales ----------
    print("\n--- (3) DISTRIBUTIONS (ref train vs nos sorties) ---")
    for lg in ("lin", "sna"):
        ref = refs_for(lg)
        pred = [v for k, v in sub.items() if lang.get(k) == lg]
        def stats(xs):
            n = max(len(xs), 1)
            maj = sum(1 for t in xs if t[:1].isupper()) / n
            fin = sum(1 for t in xs if t.rstrip()[-1:] in ".!?") / n
            vir = sum(1 for t in xs if "," in t) / n
            esp = sum(1 for t in xs if " ." in t or " ," in t) / n
            return maj, fin, vir, esp
        a = stats(ref); b = stats(pred)
        print("  [%s]                      REF     NOUS" % lg)
        for i, nm in enumerate(("majuscule initiale", "ponctuation finale", "contient virgule", "espace avant ponct")):
            d = abs(a[i] - b[i])
            flag = "   <-- ECART" if d > 0.20 else ""
            print("    %-20s %6.1f%% %6.1f%%%s" % (nm, 100 * a[i], 100 * b[i], flag))
    print("\nAUDIT_DONE")


if __name__ == "__main__":
    main()