Thierry Tallagrand
f121f0a196
- Fiche agent maison (AGENTS.md) : prevision de nuit relayee par script, import
EDF, regle JEEP midi, saisie vehicule/niveau/depart, concierge qui relaie
« ✅ Noté » ou « Rien n'est noté », groupe A deplace (28 649 -> 25 831 car.).
- Nouvelles references : reference-concierge.md, reference-workspace.md.
- Scripts : rapport-solaire.js, prevision-nuit.js, airtria-ha-vehicule.js,
concierge-live.js (answer ne sort plus jamais vide, mode CONCIERGE_SIMULATION).
- Courrier Orange / AIRTRIA / Gmail : audits, regles, menage, recents.
- Bancs EVCS tests-pv (G5 accepte « anomal »).
- .gitignore : liste blanche etendue aux nouveaux scripts et bancs.
88 lines
5.1 KiB
Python
88 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
# orange-audit.py — Phase 1 du ménage Orange : AUDIT EN LECTURE SEULE de la boîte
|
|
# thierry.tallagrand@wanadoo.fr par IMAP. Équivalent de gmail_audit.py (juin 2026).
|
|
#
|
|
# Ne modifie RIEN : sélection de la boîte en lecture seule, en-têtes lus avec
|
|
# BODY.PEEK (ne marque pas comme lu). Sorties dans le workspace :
|
|
# orange-audit.md synthèse par expéditeur / domaine / année, base de orange-rules.md
|
|
# orange-audit-raw.json une ligne par message (uid, from, date, lu, désabonnement)
|
|
#
|
|
# Le secret vient de /home/node/.openclaw/.env (ORANGE_IMAP_PASSWORD), jamais en clair ici.
|
|
# Usage (dans le conteneur) : python3 /home/node/.openclaw/workspace/orange-audit.py [dossier=INBOX]
|
|
import imaplib, re, json, sys, time, email, email.utils
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
|
|
USER = "thierry.tallagrand@wanadoo.fr"
|
|
HOST, PORT = "imap.orange.fr", 993
|
|
WS = "/home/node/.openclaw/workspace"
|
|
FOLDER = sys.argv[1] if len(sys.argv) > 1 else "INBOX"
|
|
CHUNK = 1000
|
|
|
|
env = dict(l.strip().split("=", 1) for l in open("/home/node/.openclaw/.env") if "=" in l and not l.startswith("#"))
|
|
PWD = env["ORANGE_IMAP_PASSWORD"].strip().strip('"').strip("'")
|
|
|
|
t0 = time.time()
|
|
m = imaplib.IMAP4_SSL(HOST, PORT)
|
|
m.login(USER, PWD)
|
|
typ, n = m.select(f'"{FOLDER}"', readonly=True)
|
|
total = int(n[0])
|
|
typ, data = m.uid("search", None, "ALL")
|
|
uids = data[0].split()
|
|
print(f"{FOLDER} : {total} messages, {len(uids)} UID", flush=True)
|
|
|
|
RX_META = re.compile(rb'UID (\d+).*?FLAGS \((.*?)\).*?INTERNALDATE "([^"]+)"', re.S)
|
|
rows = []
|
|
for i in range(0, len(uids), CHUNK):
|
|
chunk = uids[i:i + CHUNK]
|
|
rng = f"{chunk[0].decode()}:{chunk[-1].decode()}"
|
|
for attempt in range(3):
|
|
try:
|
|
typ, resp = m.uid("fetch", rng, "(UID FLAGS INTERNALDATE BODY.PEEK[HEADER.FIELDS (FROM DATE LIST-UNSUBSCRIBE)])")
|
|
break
|
|
except imaplib.IMAP4.abort:
|
|
time.sleep(5); m = imaplib.IMAP4_SSL(HOST, PORT); m.login(USER, PWD); m.select(f'"{FOLDER}"', readonly=True)
|
|
for part in resp:
|
|
if not isinstance(part, tuple): continue
|
|
meta, hdr = part
|
|
mm = RX_META.search(meta)
|
|
if not mm: continue
|
|
uid, flags, idate = int(mm.group(1)), mm.group(2).decode(errors="replace"), mm.group(3).decode()
|
|
msg = email.message_from_bytes(hdr)
|
|
name, addr = email.utils.parseaddr(str(email.header.make_header(email.header.decode_header(msg.get("From", "")))) if msg.get("From") else "")
|
|
try: d = email.utils.parsedate_to_datetime(idate.replace('"', ''))
|
|
except Exception: d = None
|
|
rows.append({"uid": uid, "from": addr.lower(), "name": name[:60], "date": d.strftime("%Y-%m-%d") if d else idate[:11],
|
|
"seen": "\\Seen" in flags, "unsub": bool(msg.get("List-Unsubscribe"))})
|
|
print(f" {min(i + CHUNK, len(uids))}/{len(uids)} ({time.time() - t0:.0f}s)", flush=True)
|
|
m.logout()
|
|
|
|
json.dump(rows, open(f"{WS}/orange-audit-raw.json", "w"), ensure_ascii=False)
|
|
|
|
# --- agrégation ---
|
|
by_addr = defaultdict(lambda: {"n": 0, "unread": 0, "unsub": 0, "first": "9999", "last": "0000", "name": ""})
|
|
by_dom = defaultdict(lambda: {"n": 0, "unread": 0})
|
|
by_year = defaultdict(lambda: {"n": 0, "unread": 0})
|
|
for r in rows:
|
|
a = by_addr[r["from"] or "(vide)"]; a["n"] += 1; a["unread"] += (not r["seen"]); a["unsub"] += r["unsub"]
|
|
a["first"] = min(a["first"], r["date"]); a["last"] = max(a["last"], r["date"]); a["name"] = a["name"] or r["name"]
|
|
dom = r["from"].split("@")[-1] if "@" in r["from"] else "(vide)"
|
|
by_dom[dom]["n"] += 1; by_dom[dom]["unread"] += (not r["seen"])
|
|
by_year[r["date"][:4]]["n"] += 1; by_year[r["date"][:4]]["unread"] += (not r["seen"])
|
|
|
|
unread_total = sum(1 for r in rows if not r["seen"])
|
|
L = [f"# Audit Orange — {FOLDER} — {datetime.now().strftime('%Y-%m-%d %H:%M')}", "",
|
|
f"_Lecture seule, {len(rows)} messages lus en {time.time() - t0:.0f} s. Source : orange-audit-raw.json._", "",
|
|
f"- Total : **{len(rows)}** messages, dont **{unread_total}** non lus ({100 * unread_total // max(1, len(rows))} %)",
|
|
f"- Expéditeurs distincts : {len(by_addr)} ; domaines distincts : {len(by_dom)}",
|
|
f"- Messages avec en-tête de désabonnement (newsletters) : {sum(1 for r in rows if r['unsub'])}", "",
|
|
"## Par année", "", "| Année | Messages | Non lus |", "|---|---:|---:|"]
|
|
for y in sorted(by_year): L.append(f"| {y} | {by_year[y]['n']} | {by_year[y]['unread']} |")
|
|
L += ["", "## Top 40 domaines", "", "| Domaine | Messages | Non lus |", "|---|---:|---:|"]
|
|
for dom, v in sorted(by_dom.items(), key=lambda kv: -kv[1]["n"])[:40]: L.append(f"| {dom} | {v['n']} | {v['unread']} |")
|
|
L += ["", "## Top 150 expéditeurs", "", "| Expéditeur | Nom | Messages | Non lus | Newsletter | Premier | Dernier |", "|---|---|---:|---:|---:|---|---|"]
|
|
for addr, v in sorted(by_addr.items(), key=lambda kv: -kv[1]["n"])[:150]:
|
|
L.append(f"| {addr} | {v['name'].replace('|', '/')} | {v['n']} | {v['unread']} | {v['unsub']} | {v['first']} | {v['last']} |")
|
|
open(f"{WS}/orange-audit.md", "w").write("\n".join(L) + "\n")
|
|
print(f"écrit : {WS}/orange-audit.md et orange-audit-raw.json — {len(rows)} messages, {len(by_addr)} expéditeurs")
|