#!/usr/bin/env python3 """gmail_check.py — checker MECCANICO "c'e' posta nuova rispetto all'ultimo controllo?" NESSUN LLM, nessun giudizio di importanza, nessuna classificazione. Puro confronto di timestamp via API Gmail REST. Pensato per essere invocato da un `servizio` (systemd su Nave, stesso pattern di servizio-inbox-watch / servizio-gate-watch) ogni ~20 minuti: SOLO se questo script stampa qualcosa il servizio fara' partire lo step successivo (giudizio LLM su sessione fresca) — che non e' parte di questo file. Standard di progetto (Dropbox/adrian/CLAUDE.md): Python 3.11, httpx come client HTTP, REST diretto senza librerie Google pesanti, PEP8 / max 100 char. ------------------------------------------------------------------------------------ USO python gmail_check.py --since 2026-09-09T08:00:00Z python gmail_check.py --since 2026-09-09T08:00:00Z --env /percorso/.env INPUT --since timestamp ISO 8601 (con 'Z' o offset). Solo i messaggi con data di ricezione STRETTAMENTE successiva a questo istante contano come "nuovi". --env percorso del file .env con le credenziali OAuth (default: .env accanto a questo script). Variabili richieste: GMAIL_CHECK_CLIENT_ID GMAIL_CHECK_CLIENT_SECRET GMAIL_CHECK_REFRESH_TOKEN Scope OAuth necessario: gmail.readonly (sola lettura). --query override della query server-side (default: "-category:promotions -category:social -category:forums newer_than:1d") --debug diagnostica su stderr (conteggi, tempi) — mai contenuti delle email. OUTPUT (stdout) - Se NON c'e' nulla di nuovo: nessun output, exit 0 (silenzioso, come slave_mailbox_read.sh — un poll che non deve fare rumore a vuoto). - Se c'e' qualcosa: un JSON array, un oggetto per messaggio (non per thread): [{"thread_id": "...", "message_id": "...", "from": "...", "subject": "...", "date_iso": "2026-09-09T08:12:34+00:00"}, ...] Solo header, nessuno snippet / corpo: la lettura piu' profonda, se servira', la fara' lo step LLM successivo, non questo script. EXIT CODE 0 esecuzione ok (sia "niente di nuovo" sia "trovati N messaggi") 2 errore d'uso (argomenti / .env mancante o incompleto) 3 errore API / rete / OAuth (il servizio chiamante deve distinguere "errore" da "niente di nuovo": qui l'output e' vuoto ma l'exit code NON e' 0) GOTCHA riusato dalla sentinella MCP precedente Un thread puo' contenere messaggi di date diverse. NON si deduplica per threadId e non ci si fida della data del thread: si filtra messaggio per messaggio confrontando `internalDate` (epoch ms, preciso) contro --since. """ from __future__ import annotations import argparse import json import sys import time from datetime import datetime, timezone from pathlib import Path import httpx TOKEN_URL = "https://oauth2.googleapis.com/token" GMAIL_BASE = "https://gmail.googleapis.com/gmail/v1/users/me" DEFAULT_QUERY = "-category:promotions -category:social -category:forums newer_than:1d" HTTP_TIMEOUT = 30.0 REQUIRED_ENV = ( "GMAIL_CHECK_CLIENT_ID", "GMAIL_CHECK_CLIENT_SECRET", "GMAIL_CHECK_REFRESH_TOKEN", ) def _log(debug: bool, msg: str) -> None: if debug: print(f"[gmail_check] {msg}", file=sys.stderr) def load_env(path: Path) -> dict[str, str]: """Parser .env minimale (KEY=VALUE, righe # ignorate, quote opzionali). Nessuna dipendenza esterna: il formato qui e' banale e controllato da noi.""" if not path.is_file(): raise FileNotFoundError(f".env non trovato: {path}") out: dict[str, str] = {} for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, val = line.partition("=") key = key.strip() val = val.strip().strip('"').strip("'") if key: out[key] = val return out def parse_since(value: str) -> int: """ISO 8601 -> epoch millisecondi UTC. Accetta il suffisso 'Z'.""" text = value.strip() if text.endswith("Z"): text = text[:-1] + "+00:00" dt = datetime.fromisoformat(text) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return int(dt.timestamp() * 1000) def get_access_token(client: httpx.Client, env: dict[str, str], debug: bool) -> str: """refresh_token -> access_token (flusso REST standard Google, nessuna libreria).""" resp = client.post( TOKEN_URL, data={ "client_id": env["GMAIL_CHECK_CLIENT_ID"], "client_secret": env["GMAIL_CHECK_CLIENT_SECRET"], "refresh_token": env["GMAIL_CHECK_REFRESH_TOKEN"], "grant_type": "refresh_token", }, ) if resp.status_code != 200: raise RuntimeError( f"refresh token fallito: HTTP {resp.status_code} {resp.text[:200]}" ) token = resp.json().get("access_token") if not token: raise RuntimeError("risposta OAuth priva di access_token") _log(debug, "access_token ottenuto") return token def list_message_ids( client: httpx.Client, token: str, query: str, debug: bool ) -> list[dict[str, str]]: """Tutti gli id/threadId che passano il filtro server-side (pagina tutte le pagine).""" headers = {"Authorization": f"Bearer {token}"} ids: list[dict[str, str]] = [] page_token: str | None = None while True: params = {"q": query, "maxResults": 100} if page_token: params["pageToken"] = page_token resp = client.get(f"{GMAIL_BASE}/messages", params=params, headers=headers) if resp.status_code != 200: raise RuntimeError( f"list messages fallito: HTTP {resp.status_code} {resp.text[:200]}" ) data = resp.json() ids.extend(data.get("messages", []) or []) page_token = data.get("nextPageToken") if not page_token: break _log(debug, f"{len(ids)} messaggi passano il filtro server-side") return ids def fetch_metadata( client: httpx.Client, token: str, message_id: str ) -> dict: """Metadati leggeri di un messaggio: internalDate + header From/Subject/Date.""" headers = {"Authorization": f"Bearer {token}"} params = [ ("format", "metadata"), ("metadataHeaders", "From"), ("metadataHeaders", "Subject"), ("metadataHeaders", "Date"), ] resp = client.get( f"{GMAIL_BASE}/messages/{message_id}", params=params, headers=headers ) if resp.status_code != 200: raise RuntimeError( f"get message {message_id} fallito: HTTP {resp.status_code} {resp.text[:200]}" ) return resp.json() def header_value(payload: dict, name: str) -> str: for h in payload.get("payload", {}).get("headers", []): if h.get("name", "").lower() == name.lower(): return h.get("value", "") return "" def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Checker meccanico: c'e' posta nuova dopo --since? (nessun LLM)" ) parser.add_argument("--since", required=True, help="timestamp ISO 8601 (es. 2026-09-09T08:00:00Z)") parser.add_argument( "--env", default=str(Path(__file__).resolve().parent / ".env"), help="percorso file .env con le credenziali OAuth", ) parser.add_argument("--query", default=DEFAULT_QUERY, help="override query Gmail server-side") parser.add_argument("--debug", action="store_true", help="diagnostica su stderr (mai contenuti)") args = parser.parse_args(argv) try: since_ms = parse_since(args.since) except ValueError as exc: print(f"--since non valido: {exc}", file=sys.stderr) return 2 try: env = load_env(Path(args.env)) except FileNotFoundError as exc: print(str(exc), file=sys.stderr) return 2 missing = [k for k in REQUIRED_ENV if not env.get(k)] if missing: print(f".env incompleto, mancano: {', '.join(missing)}", file=sys.stderr) return 2 started = time.monotonic() try: with httpx.Client(timeout=HTTP_TIMEOUT) as client: token = get_access_token(client, env, args.debug) candidates = list_message_ids(client, token, args.query, args.debug) fresh: list[dict[str, str]] = [] for item in candidates: msg_id = item.get("id") if not msg_id: continue meta = fetch_metadata(client, token, msg_id) try: internal_ms = int(meta.get("internalDate", "0")) except (TypeError, ValueError): internal_ms = 0 # gotcha: filtro messaggio-per-messaggio, mai per thread if internal_ms <= since_ms: continue date_iso = datetime.fromtimestamp( internal_ms / 1000, tz=timezone.utc ).isoformat() fresh.append( { "thread_id": meta.get("threadId", item.get("threadId", "")), "message_id": msg_id, "from": header_value(meta, "From"), "subject": header_value(meta, "Subject"), "date_iso": date_iso, } ) except (httpx.HTTPError, RuntimeError) as exc: print(f"errore API/rete/OAuth: {exc}", file=sys.stderr) return 3 _log( args.debug, f"{len(fresh)} nuovi dopo --since su {len(candidates)} candidati " f"in {time.monotonic() - started:.1f}s", ) if not fresh: return 0 # silenzioso: nessun output # ordine cronologico crescente per stabilita' dell'output fresh.sort(key=lambda m: m["date_iso"]) print(json.dumps(fresh, ensure_ascii=False)) return 0 if __name__ == "__main__": sys.exit(main())