127 lines
6.9 KiB
Python
127 lines
6.9 KiB
Python
"""AMEL1-R2 : rejoue la batterie cas-v2.jsonl sur un crochet ecriture-gardien donne.
|
|
|
|
Usage : python -I -X utf8 test_crochet.py <crochet.py> [cas.jsonl] [--tout | --resume]
|
|
sans option : lignes non conformes + bilan ; --tout : toutes les lignes ; --resume : bilan seul.
|
|
|
|
Pour chaque cas, la charge PreToolUse est simulee (session_id, transcript_path, cwd, hook_event_name, tool_name,
|
|
tool_input.command, agent_type) et passee sur l'entree standard de `python -I <crochet>`, comme dans settings.json.
|
|
Aucune commande du cas n'est executee : seul le crochet tourne, depuis le dossier temporaire du poste.
|
|
Les marques {{PROFIL}}, {{PROFIL/}}, {{COMPTE}} des cas sont remplacees par le profil du poste qui rejoue.
|
|
Code 0 = passe, 2 = refus, autre = erreur (jamais conforme).
|
|
"""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
ICI = os.path.dirname(os.path.abspath(__file__))
|
|
UP = os.environ.get("USERPROFILE", "")
|
|
COMPTE = os.path.basename(UP)
|
|
CWD_DEFAUT = r"D:\Game\LifePlay-master 5.8"
|
|
|
|
|
|
def restaurer(texte):
|
|
if not texte:
|
|
return texte
|
|
return texte.replace("{{PROFIL/}}", UP.replace("\\", "/")).replace("{{PROFIL}}", UP).replace("{{COMPTE}}", COMPTE)
|
|
|
|
|
|
def court(texte, n):
|
|
texte = texte.replace("\r", "").replace("\n", "\\n").replace("|", "\\|")
|
|
return texte if len(texte) <= n else texte[: n - 3] + "..."
|
|
|
|
|
|
def main():
|
|
args = [a for a in sys.argv[1:] if not a.startswith("--")]
|
|
tout, resume = "--tout" in sys.argv, "--resume" in sys.argv
|
|
if not args:
|
|
print("usage : python -I -X utf8 test_crochet.py <crochet.py> [cas.jsonl] [--tout | --resume]")
|
|
return 64
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
crochet = os.path.abspath(args[0])
|
|
fichier = args[1] if len(args) > 1 else os.path.join(ICI, "cas-v2.jsonl")
|
|
print("crochet : " + crochet.replace(UP, "%USERPROFILE%"))
|
|
print("batterie : " + os.path.abspath(fichier))
|
|
print("TEMP vu par le crochet : " + os.environ.get("TEMP", "?").replace(UP, "%USERPROFILE%"))
|
|
print("id | groupe | attendu | obtenu | conforme | ecriture (1re ligne)")
|
|
lignes = []
|
|
with open(fichier, encoding="utf-8") as f:
|
|
cas_liste = [json.loads(l) for l in f if l.strip() and not l.startswith("#")]
|
|
durees = []
|
|
for cas in cas_liste:
|
|
charge = {
|
|
"session_id": cas.get("session_id") or "00000000-0000-0000-0000-000000000000",
|
|
"transcript_path": "",
|
|
"cwd": restaurer(cas.get("cwd")) or CWD_DEFAUT,
|
|
"hook_event_name": "PreToolUse",
|
|
"tool_name": cas["tool_name"],
|
|
"tool_input": {"command": restaurer(cas["command"]), "description": "AMEL1-R2 " + cas["id"]},
|
|
}
|
|
if cas.get("agent_type"):
|
|
charge["agent_type"] = cas["agent_type"]
|
|
entree = bytes.fromhex(cas["charge_hex"]) if "charge_hex" in cas else json.dumps(charge, ensure_ascii=True).encode("ascii")
|
|
t0 = time.perf_counter()
|
|
p = subprocess.run([sys.executable, "-I", crochet], input=entree, capture_output=True, timeout=30,
|
|
cwd=tempfile.gettempdir())
|
|
durees.append(time.perf_counter() - t0)
|
|
code = p.returncode
|
|
obtenu = {0: "passe", 2: "refus"}.get(code, "erreur")
|
|
erreur = p.stderr.decode("utf-8", "replace")
|
|
ecriture = ""
|
|
for l in erreur.splitlines():
|
|
if l.strip().startswith("Ecriture:"):
|
|
ecriture = l.strip()[len("Ecriture:"):].strip()
|
|
break
|
|
if code not in (0, 2) or "Traceback" in erreur:
|
|
ecriture = "ERREUR " + court(erreur.strip(), 200)
|
|
entete = erreur.startswith("Refus du crochet ecriture-gardien.") and "hook error" not in erreur
|
|
ok = obtenu == cas["attendu"] and (not cas.get("message_requis") or entete)
|
|
if cas.get("message_requis") and obtenu == "refus" and not entete:
|
|
ecriture = "MESSAGE ABSENT OU NON CONFORME " + court(erreur.strip(), 120)
|
|
g = cas["groupe"].split("-")[0]
|
|
lignes.append(dict(id=cas["id"], g=g, groupe=cas["groupe"], attendu=cas["attendu"], obtenu=obtenu, ok=ok,
|
|
ecriture=ecriture.replace(UP, "%USERPROFILE%").replace(COMPTE, "<compte>"), code=code,
|
|
impossible="analyse impossible" in erreur, stricte="analyse stricte" in erreur))
|
|
if tout or (not ok and not resume):
|
|
print("{} | {} | {} | {} | {} | {}".format(cas["id"], cas["groupe"], cas["attendu"], obtenu,
|
|
"oui" if ok else "NON", court(lignes[-1]["ecriture"], 150)))
|
|
print("-" * 80)
|
|
print("BILAN")
|
|
for g, nom in (("a", "a : refus a tort de P3"), ("b", "b : refus utiles de P3"), ("c", "c : cas adverses"),
|
|
("t", "t : temoins"), ("r", "r : charges illisibles")):
|
|
sel = [x for x in lignes if x["g"] == g]
|
|
print(" {:<26} total {:>3} | refus {:>3} | passe {:>3} | erreur {:>2} | conformes {:>3} / {}".format(
|
|
nom, len(sel), sum(x["obtenu"] == "refus" for x in sel), sum(x["obtenu"] == "passe" for x in sel),
|
|
sum(x["obtenu"] == "erreur" for x in sel), sum(x["ok"] for x in sel), len(sel)))
|
|
a = [x for x in lignes if x["g"] == "a"]
|
|
b = [x for x in lignes if x["g"] == "b"]
|
|
c = [x for x in lignes if x["g"] == "c"]
|
|
ab_refus = [x for x in a + b if x["obtenu"] == "refus"]
|
|
a_refus = [x for x in a if x["obtenu"] == "refus"]
|
|
atort_v2 = [x for x in ab_refus if x["attendu"] == "passe"]
|
|
print("I2a (definition P3 : refus du groupe a / refus des groupes a+b) : {} / {} = {}".format(
|
|
len(a_refus), len(ab_refus), pct(len(a_refus), len(ab_refus))))
|
|
print("I2a (definition v2 : refus dont l'attendu est passe / refus des groupes a+b) : {} / {} = {}".format(
|
|
len(atort_v2), len(ab_refus), pct(len(atort_v2), len(ab_refus))))
|
|
print("I2b (adverses refuses / adverses) : {} / {} = {}".format(
|
|
sum(x["obtenu"] == "refus" for x in c), len(c), pct(sum(x["obtenu"] == "refus" for x in c), len(c))))
|
|
print("groupe a refuses : " + (" ".join(x["id"] for x in a_refus) or "aucun"))
|
|
print("groupe b acceptes : " + (" ".join(x["id"] for x in b if x["obtenu"] == "passe") or "aucun"))
|
|
print("groupe b attendus acceptes : " + (" ".join(x["id"] for x in b if x["attendu"] == "passe") or "aucun"))
|
|
print("adverses qui passent : " + (" ".join(x["id"] for x in c if x["obtenu"] != "refus") or "aucun"))
|
|
print("analyse impossible : {} ; analyse stricte : {} ; codes hors 0/2 : {}".format(
|
|
sum(x["impossible"] for x in lignes), sum(x["stricte"] for x in lignes), sum(x["code"] not in (0, 2) for x in lignes)))
|
|
print("duree par cas : moyenne {:.0f} ms, max {:.0f} ms".format(1000 * sum(durees) / max(1, len(durees)), 1000 * max(durees or [0])))
|
|
n_ok = sum(x["ok"] for x in lignes)
|
|
print("conformes : {} / {}".format(n_ok, len(lignes)))
|
|
return 0 if n_ok == len(lignes) else 1
|
|
|
|
|
|
def pct(a, b):
|
|
return "{:.1f} %".format(100.0 * a / b) if b else "n/a"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|