#!/usr/bin/env python3 """Collecteur structurel de l'Atelier Nemrod. Mesure la FORME d'un dossier d'agent. Ne lit jamais le contenu pour le transmettre : aucun texte, aucun chemin, aucun nom de fichier ne figure dans le rapport. Usage: python3 collecteur_atelier.py --dossier /chemin/du/dossier python3 collecteur_atelier.py --dossier . --sortie rapport.json Lecture seule. Le seul fichier ecrit est le rapport, la ou vous le demandez. """ import argparse, datetime, hashlib, json, math, os, re, statistics, sys VERSION = "atelier-0.1" FLUX = (".log", ".jsonl", ".txt", ".csv") STRUCT = (".json", ".yml", ".yaml", ".toml") CODE = (".py", ".js", ".ts", ".sh", ".html", ".css") SAUTS = {".git", "node_modules", "__pycache__", ".venv", "venv", "media", "images", "img"} TS = re.compile(r"^\[?(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2}):(\d{2})") MAX_FICHIERS = 20000 MAX_OCTETS = 20 * 1024 * 1024 def classer(nom): n = nom.lower() if n.endswith(FLUX): return "flux" if n.endswith(STRUCT): return "struct" if n.endswith(CODE): return "code" return None def horodatage(ligne): m = TS.match(ligne) if not m: return None try: return datetime.datetime(*map(int, m.groups())).timestamp() except Exception: return None def mediane(v): return statistics.median(v) if v else None def ecart_median(v): if not v: return 0.0 m = statistics.median(v) return statistics.median([abs(x - m) for x in v]) def mesurer(racine): flux, struct = [], [] plus_recent = 0 outils = None total_fichiers = 0 lus = 0 octets = 0 for base, dossiers, noms in os.walk(racine): dossiers[:] = [d for d in dossiers if d not in SAUTS] for nom in noms: total_fichiers += 1 chemin = os.path.join(base, nom) try: plus_recent = max(plus_recent, os.path.getmtime(chemin)) except OSError: pass if nom.lower() == "agent.json" and os.sep + "config" in chemin: try: outils = set(json.load(open(chemin, encoding="utf-8")).get("outils") or []) except Exception: pass classe = classer(nom) if not classe or total_fichiers > MAX_FICHIERS or octets > MAX_OCTETS: continue try: brut = open(chemin, "rb").read() except OSError: continue octets += len(brut) lus += 1 texte = brut.decode("utf-8", "replace") if classe == "flux": net = [x.strip() for x in texte.splitlines() if len(x.strip()) >= 4] if not net: continue stamps = [horodatage(x) for x in net] if nom.lower().endswith(".jsonl"): stamps = [] for x in net: try: d = json.loads(x) v = d.get("timestamp") or d.get("time") or d.get("ts") or d.get("date") if v: try: stamps.append(datetime.datetime.fromisoformat( str(v).replace("Z", "+00:00")).timestamp()) except Exception: pass except Exception: pass stamps = [s for s in stamps if s] if len(stamps) < max(10, len(net) // 2): continue frag = len(net) - len(stamps) vus = set() dup = 0 for x in net: h = hash(x) if h in vus: dup += 1 else: vus.add(h) inv = sum(1 for i in range(1, len(stamps)) if stamps[i] < stamps[i - 1]) diffs = [stamps[i] - stamps[i - 1] for i in range(1, len(stamps)) if stamps[i] >= stamps[i - 1]] diffs = [d for d in diffs if d > 0] med = mediane(diffs) if diffs else 0 trous = [d for d in diffs if med and d > 3 * med and d > 60] cv = 0.0 if len(diffs) > 2: moy = sum(diffs) / len(diffs) if moy: cv = math.sqrt(sum((x - moy) ** 2 for x in diffs) / len(diffs)) / moy tiers = max(1, len(net) // 3) err = [1 if ("ERROR" in x or "FAIL" in x or x.endswith("...")) else 0 for x in net] delta = sum(err[-tiers:]) / tiers - sum(err[:tiers]) / tiers observes = set(re.findall(r"tool=([a-z_]+)", texte)) flux.append({"T": frag * 100 / len(net), "D": inv * 1000 / len(stamps), "G": max(trous) if trous else 0.0, "R": dup * 1000 / len(net), "P": cv, "DELTA": delta, "last": max(stamps), "med": med, "obs": observes}) elif classe == "struct" and nom.lower().endswith(".json"): try: d = json.loads(texte) ok = 1 except Exception: d, ok = None, 0 entree = {"I": 0 if ok else 1} if d is not None: ids, refs = set(), [] def parcourir(o): if isinstance(o, dict): if "id" in o and isinstance(o["id"], (int, str)): ids.add(str(o["id"])) if "ref" in o and isinstance(o["ref"], (int, str)): refs.append(str(o["ref"])) for v in o.values(): parcourir(v) elif isinstance(o, list): for v in o: parcourir(v) parcourir(d) entree["C"] = sum(1 for r in refs if r not in ids) * 1000 / len(refs) if refs else 0.0 struct.append(entree) def maxi(k): v = [x[k] for x in flux if x.get(k) is not None] return max(v) if v else None def med_reg(k): v = [x[k] for x in flux if x.get(k) is not None] return mediane(v) if v else None dernier = max([x["last"] for x in flux], default=None) cadence = mediane([x["med"] for x in flux if x["med"]]) silence = ((plus_recent - dernier) / cadence) if (dernier and cadence) else None observes = set() for x in flux: observes.update(x["obs"]) alignement = 1 if (outils is not None and any(o not in outils for o in observes)) else 0 mesures = { "T": med_reg("T"), "D": maxi("D"), "G": maxi("G"), "C": (sum(x.get("C", 0) for x in struct) / len(struct)) if struct else None, "R": med_reg("R"), "I": (sum(x["I"] for x in struct) / len(struct)) if struct else None, "P": maxi("P"), "DELTA": maxi("DELTA"), "S": silence, "A": alignement, } mesures = {k: (round(v, 3) if isinstance(v, float) else v) for k, v in mesures.items()} non_mesure = [k for k, v in mesures.items() if v is None] return { "version": VERSION, "date": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "agent_id": hashlib.sha256(os.path.abspath(racine).encode()).hexdigest()[:8], "bande": "descriptif", "mesures": mesures, "couverture": {"fichiers_presents": total_fichiers, "fichiers_lus": lus, "octets_lus": octets, "part": round(lus / total_fichiers, 3) if total_fichiers else 0}, "non_mesure": non_mesure, "non_examine": ["contenu des messages", "binaires", "base de donnees", "secrets"], "note": "Aucune reference de type ni historique : ce rapport est descriptif, pas un diagnostic.", } def resume(rapport): m = rapport["mesures"]; c = rapport["couverture"] lignes = [ "Collecteur structurel de l'Atelier Nemrod", "Dossier mesure : " + rapport["agent_id"] + " (empreinte, pas de chemin)", "Fichiers lus : %d / %d (%.0f %%)" % (c["fichiers_lus"], c["fichiers_presents"], 100 * c["part"]), "", "T troncature (pour 100 enreg.) : %s" % m["T"], "D desordre (pour 1000) : %s" % m["D"], "G trou le plus large (s) : %s" % m["G"], "C incoherence (pour 1000) : %s" % m["C"], "R redondance (pour 1000) : %s" % m["R"], "I illisibilite (part) : %s" % m["I"], "P rythme (CV) : %s" % m["P"], "Delta derivee : %s" % m["DELTA"], "S silence (cadences) : %s" % m["S"], "A alignement outils : %s" % m["A"], "", "Non mesure : " + (", ".join(rapport["non_mesure"]) if rapport["non_mesure"] else "rien"), "", "Ce rapport ne contient aucun contenu, aucun chemin, aucun nom de fichier.", "Il est descriptif : aucune reference de type n'est disponible pour le noter.", ] return "\n".join(lignes) def main(): p = argparse.ArgumentParser(description="Collecteur structurel de l'Atelier Nemrod (lecture seule)") p.add_argument("--dossier", default=".", help="dossier de l'agent a mesurer") p.add_argument("--sortie", default=None, help="fichier de rapport JSON") p.add_argument("--json", action="store_true", help="afficher le JSON sur la sortie standard") a = p.parse_args() if not os.path.isdir(a.dossier): print("Dossier introuvable : " + a.dossier, file=sys.stderr) return 2 rapport = mesurer(a.dossier) sortie = a.sortie or ("rapport_atelier_" + datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + ".json") with open(sortie, "w", encoding="utf-8") as f: json.dump(rapport, f, ensure_ascii=False, indent=2) if a.json: print(json.dumps(rapport, ensure_ascii=False, indent=2)) else: print(resume(rapport)) print("") print("Rapport ecrit : " + sortie) return 0 if __name__ == "__main__": sys.exit(main())