1#!/usr/bin/env python3
2"""Generic Gmail mailbox tool: search, preview, and batch-archive mail.
3 
4Uses existing `gog` OAuth secrets, fetches one mailbox snapshot, filters locally,
5archives selected messages with Gmail batchModify, retries on rate limits, and
6prints a JSON summary with before/selected/archived/after counts.
7 
8Examples:
9  python3 scripts/gmail_mail_tool.py --account you@example.com --from-email notifications@github.com
10  python3 scripts/gmail_mail_tool.py --account you@example.com --query 'from:[email protected] newer_than:30d'
11  python3 scripts/gmail_mail_tool.py --account you@example.com --from-email notifications@github.com --apply
12"""
13 
14from __future__ import annotations
15 
16import argparse
17import json
18import os
19import random
20import socket
21import subprocess
22import sys
23import tempfile
24import time
25import urllib.error
26import urllib.parse
27import urllib.request
28from dataclasses import dataclass
29from datetime import datetime, timezone
30from email.utils import parseaddr, parsedate_to_datetime
31from pathlib import Path
32from typing import Any
33 
34DEFAULT_ACCOUNT = os.environ.get("GMAIL_TOOL_ACCOUNT", os.environ.get("GOG_ACCOUNT", "")).strip()
35HOME = Path.home()
36GOG_CREDENTIALS = Path(
37    os.environ.get(
38        "GMAIL_TOOL_GOG_CREDENTIALS",
39        str(HOME / ".config/gogcli/credentials.json"),
40    )
41).expanduser()
42GOG_KEYRING_PASSWORD_FILE = Path(
43    os.environ.get(
44        "GMAIL_TOOL_GOG_KEYRING_PASSWORD_FILE",
45        str(HOME / ".config/gogcli/.keyring_password"),
46    )
47).expanduser()
48GMAIL_API_ROOT = "https://gmail.googleapis.com/gmail/v1"
49DEFAULT_LOG_FILE = Path(
50    os.environ.get(
51        "GMAIL_TOOL_LOG_FILE",
52        str(Path.cwd() / "gmail_mail_tool.last.log"),
53    )
54).expanduser()
55 
56LOG_FILE_PATH: Path | None = None
57 
58 
59class GmailApiError(RuntimeError):
60    pass
61 
62 
63@dataclass
64class MessageMeta:
65    message_id: str
66    thread_id: str
67    internal_date_ms: int
68    labels: list[str]
69    from_header: str
70    from_email: str
71    subject: str
72    snippet: str
73    list_id: str
74    to_header: str
75    cc_header: str
76 
77 
78def log_line(message: str) -> None:
79    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
80    line = f"[{timestamp}] {message}"
81    print(line, file=sys.stderr, flush=True)
82    if LOG_FILE_PATH is not None:
83        with LOG_FILE_PATH.open("a", encoding="utf-8") as handle:
84            handle.write(line + "\n")
85 
86 
87def configure_logging(log_file: str) -> Path:
88    path = Path(log_file).expanduser() if log_file else DEFAULT_LOG_FILE
89    path.parent.mkdir(parents=True, exist_ok=True)
90    path.write_text("", encoding="utf-8")
91    global LOG_FILE_PATH
92    LOG_FILE_PATH = path
93    return path
94 
95 
96def parse_args() -> argparse.Namespace:
97    parser = argparse.ArgumentParser(
98        description="Generic Gmail mailbox tool for local filtering and batch archive."
99    )
100    parser.add_argument("--account", default=DEFAULT_ACCOUNT, help="gog account email")
101    parser.add_argument(
102        "--query",
103        default="",
104        help="Optional Gmail search query; still fetched as one snapshot then filtered locally",
105    )
106    parser.add_argument(
107        "--all-mail",
108        action="store_true",
109        help="Search all mail instead of default INBOX scope",
110    )
111    parser.add_argument(
112        "--label",
113        action="append",
114        default=[],
115        help="Additional Gmail label ids/names to require in list request (repeatable)",
116    )
117    parser.add_argument(
118        "--from-email",
119        action="append",
120        default=[],
121        help="Exact sender email match (repeatable; OR within the same flag)",
122    )
123    parser.add_argument(
124        "--from-contains",
125        action="append",
126        default=[],
127        help="Substring match against raw From header (repeatable)",
128    )
129    parser.add_argument(
130        "--subject-contains",
131        action="append",
132        default=[],
133        help="Substring match against subject (repeatable)",
134    )
135    parser.add_argument(
136        "--snippet-contains",
137        action="append",
138        default=[],
139        help="Substring match against Gmail snippet (repeatable)",
140    )
141    parser.add_argument(
142        "--list-id-contains",
143        action="append",
144        default=[],
145        help="Substring match against List-Id header (repeatable)",
146    )
147    parser.add_argument(
148        "--to-contains",
149        action="append",
150        default=[],
151        help="Substring match against To header (repeatable)",
152    )
153    parser.add_argument(
154        "--cc-contains",
155        action="append",
156        default=[],
157        help="Substring match against Cc header (repeatable)",
158    )
159    parser.add_argument(
160        "--has-label",
161        action="append",
162        default=[],
163        help="Keep only messages that already have all these labels (repeatable)",
164    )
165    parser.add_argument(
166        "--missing-label",
167        action="append",
168        default=[],
169        help="Keep only messages that do not have any of these labels (repeatable)",
170    )
171    parser.add_argument(
172        "--older-than-days",
173        type=int,
174        default=0,
175        help="Keep only messages older than N full days",
176    )
177    parser.add_argument(
178        "--newer-than-days",
179        type=int,
180        default=0,
181        help="Keep only messages newer than N full days",
182    )
183    parser.add_argument(
184        "--sample-limit",
185        type=int,
186        default=25,
187        help="How many selected message samples to print in the summary",
188    )
189    parser.add_argument(
190        "--page-size",
191        type=int,
192        default=500,
193        help="Gmail list page size (max 500)",
194    )
195    parser.add_argument(
196        "--max-pages",
197        type=int,
198        default=0,
199        help="Optional mailbox page cap; 0 means fetch all pages",
200    )
201    parser.add_argument(
202        "--max-messages",
203        type=int,
204        default=0,
205        help="Optional message cap after listing; 0 means no cap",
206    )
207    parser.add_argument(
208        "--chunk-size",
209        type=int,
210        default=500,
211        help="batchModify chunk size (Gmail allows up to 1000)",
212    )
213    parser.add_argument(
214        "--metadata-delay-ms",
215        type=int,
216        default=0,
217        help="Optional pause between metadata GETs; 0 disables pacing",
218    )
219    parser.add_argument(
220        "--min-backoff-seconds",
221        type=int,
222        default=60,
223        help="Fallback backoff minimum when Retry-After is absent",
224    )
225    parser.add_argument(
226        "--max-backoff-seconds",
227        type=int,
228        default=90,
229        help="Fallback backoff maximum when Retry-After is absent",
230    )
231    parser.add_argument(
232        "--http-timeout-seconds",
233        type=int,
234        default=25,
235        help="Per-request HTTP timeout; stuck calls fail fast instead of hanging indefinitely",
236    )
237    parser.add_argument(
238        "--progress-every",
239        type=int,
240        default=25,
241        help="Progress log cadence for metadata fetch/archive phases",
242    )
243    parser.add_argument(
244        "--apply",
245        action="store_true",
246        help="Actually archive selected messages (remove INBOX label)",
247    )
248    parser.add_argument(
249        "--allow-empty-filter",
250        action="store_true",
251        help="Allow running without query/local filters (dangerous; preview only unless --apply)",
252    )
253    parser.add_argument(
254        "--max-apply-messages",
255        type=int,
256        default=200,
257        help="Safety cap: refuse --apply when selected messages exceed this count",
258    )
259    parser.add_argument(
260        "--max-apply-ratio",
261        type=float,
262        default=0.35,
263        help="Safety cap: refuse --apply when selected/scanned ratio exceeds this fraction",
264    )
265    parser.add_argument(
266        "--force-unsafe-apply",
267        action="store_true",
268        help="Bypass safety caps for large/query-only apply runs",
269    )
270    parser.add_argument(
271        "--out",
272        default="",
273        help="Optional path to write the JSON summary",
274    )
275    parser.add_argument(
276        "--log-file",
277        default=str(DEFAULT_LOG_FILE),
278        help="Path to runtime log file on disk",
279    )
280    return parser.parse_args()
281 
282 
283def load_keyring_password() -> str:
284    env_value = os.environ.get("GOG_KEYRING_PASSWORD", "")
285    if env_value:
286        return env_value
287    if not GOG_KEYRING_PASSWORD_FILE.exists():
288        raise GmailApiError(
289            "GOG keyring password file not found. Set GOG_KEYRING_PASSWORD or GMAIL_TOOL_GOG_KEYRING_PASSWORD_FILE."
290        )
291    return GOG_KEYRING_PASSWORD_FILE.read_text(encoding="utf-8").strip()
292 
293 
294def refresh_access_token(account: str, *, http_timeout_seconds: int) -> str:
295    started = time.monotonic()
296    log_line(f"TOKEN refresh start account={account}")
297    keyring_password = load_keyring_password()
298    with tempfile.NamedTemporaryFile("w+", suffix=".json", delete=True) as tmp:
299        env = dict(os.environ)
300        env["GOG_KEYRING_PASSWORD"] = keyring_password
301        subprocess.run(
302            [
303                "gog",
304                "auth",
305                "tokens",
306                "export",
307                account,
308                "--out",
309                tmp.name,
310                "--overwrite",
311            ],
312            check=True,
313            env=env,
314            stdout=subprocess.DEVNULL,
315            stderr=subprocess.DEVNULL,
316        )
317        tmp.seek(0)
318        token_data = json.load(tmp)
319 
320    if not GOG_CREDENTIALS.exists():
321        raise GmailApiError(
322            "gog credentials.json not found. Set GMAIL_TOOL_GOG_CREDENTIALS or run gog auth credentials first."
323        )
324    credentials = json.loads(GOG_CREDENTIALS.read_text(encoding="utf-8"))
325    payload = urllib.parse.urlencode(
326        {
327            "client_id": credentials["client_id"],
328            "client_secret": credentials["client_secret"],
329            "refresh_token": token_data["refresh_token"],
330            "grant_type": "refresh_token",
331        }
332    ).encode("utf-8")
333    request = urllib.request.Request(
334        "https://oauth2.googleapis.com/token",
335        data=payload,
336        method="POST",
337    )
338    with urllib.request.urlopen(request, timeout=http_timeout_seconds) as response:
339        data = json.loads(response.read().decode("utf-8"))
340    log_line(f"TOKEN refresh ok elapsed={time.monotonic() - started:.2f}s")
341    return data["access_token"]
342 
343 
344def parse_retry_after(value: str | None) -> float | None:
345    if not value:
346        return None
347    value = value.strip()
348    if not value:
349        return None
350    if value.isdigit():
351        return float(value)
352    try:
353        retry_at = parsedate_to_datetime(value)
354    except (TypeError, ValueError, IndexError):
355        return None
356    return max(0.0, retry_at.timestamp() - time.time())
357 
358 
359def extract_api_reason(raw: str) -> str:
360    try:
361        payload = json.loads(raw)
362    except json.JSONDecodeError:
363        return ""
364    error = payload.get("error", {})
365    errors = error.get("errors") or []
366    if errors and isinstance(errors[0], dict):
367        return str(errors[0].get("reason") or "")
368    return str(error.get("status") or "")
369 
370 
371def request_json(
372    access_token: str,
373    method: str,
374    path: str,
375    *,
376    params: dict[str, Any] | None = None,
377    body: dict[str, Any] | None = None,
378    max_attempts: int = 8,
379    min_backoff_seconds: int = 60,
380    max_backoff_seconds: int = 90,
381    http_timeout_seconds: int = 25,
382) -> Any:
383    url = f"{GMAIL_API_ROOT}{path}"
384    if params:
385        query = urllib.parse.urlencode(params, doseq=True)
386        url = f"{url}?{query}"
387 
388    headers = {"Authorization": f"Bearer {access_token}"}
389    data = None
390    if body is not None:
391        data = json.dumps(body).encode("utf-8")
392        headers["Content-Type"] = "application/json"
393 
394    for attempt in range(1, max_attempts + 1):
395        started = time.monotonic()
396        log_line(f"HTTP start method={method} path={path} attempt={attempt}/{max_attempts}")
397        request = urllib.request.Request(url, data=data, headers=headers, method=method)
398        try:
399            with urllib.request.urlopen(request, timeout=http_timeout_seconds) as response:
400                raw = response.read()
401                log_line(
402                    f"HTTP ok method={method} path={path} status={getattr(response, 'status', 'unknown')} elapsed={time.monotonic() - started:.2f}s"
403                )
404                if not raw:
405                    return None
406                return json.loads(raw.decode("utf-8"))
407        except urllib.error.HTTPError as error:
408            raw = error.read().decode("utf-8", errors="replace")
409            reason = extract_api_reason(raw)
410            retry_after = parse_retry_after(error.headers.get("Retry-After"))
411            log_line(
412                f"HTTP error method={method} path={path} code={error.code} reason={reason or 'unknown'} elapsed={time.monotonic() - started:.2f}s"
413            )
414            should_retry = (
415                (error.code == 403 and reason == "rateLimitExceeded")
416                or error.code in {429, 500, 502, 503, 504}
417            )
418            if should_retry and attempt < max_attempts:
419                delay = retry_after
420                if delay is None:
421                    delay = random.uniform(min_backoff_seconds, max_backoff_seconds)
422                print(
423                    f"[retry] {method} {path} failed with {error.code}/{reason or 'unknown'}; "
424                    f"sleeping {delay:.1f}s before retry {attempt + 1}/{max_attempts}",
425                    file=sys.stderr,
426                )
427                time.sleep(delay)
428                continue
429            raise GmailApiError(
430                f"{method} {path} failed with HTTP {error.code}: {raw}"
431            ) from error
432        except (urllib.error.URLError, TimeoutError, socket.timeout) as error:
433            reason = getattr(error, "reason", str(error))
434            log_line(
435                f"HTTP network-error method={method} path={path} reason={reason} elapsed={time.monotonic() - started:.2f}s"
436            )
437            if attempt < max_attempts:
438                delay = random.uniform(5, 10)
439                print(
440                    f"[retry] {method} {path} network error {reason}; "
441                    f"sleeping {delay:.1f}s before retry {attempt + 1}/{max_attempts}",
442                    file=sys.stderr,
443                )
444                time.sleep(delay)
445                continue
446            raise GmailApiError(f"{method} {path} failed: {error}") from error
447 
448    raise GmailApiError(f"{method} {path} exceeded retry budget")
449 
450 
451def normalize_many(values: list[str]) -> list[str]:
452    return [value.strip().lower() for value in values if value and value.strip()]
453 
454 
455def any_contains(haystack: str, needles: list[str]) -> bool:
456    if not needles:
457        return True
458    haystack = haystack.lower()
459    return any(needle in haystack for needle in needles)
460 
461 
462def compute_age_days(internal_date_ms: int) -> float:
463    if not internal_date_ms:
464        return 0.0
465    now = datetime.now(timezone.utc)
466    created = datetime.fromtimestamp(internal_date_ms / 1000, tz=timezone.utc)
467    return max(0.0, (now - created).total_seconds() / 86400)
468 
469 
470def list_message_ids(
471    access_token: str,
472    *,
473    label_ids: list[str],
474    query: str,
475    page_size: int,
476    max_pages: int,
477    max_messages: int,
478    min_backoff_seconds: int,
479    max_backoff_seconds: int,
480    http_timeout_seconds: int,
481) -> list[str]:
482    page_token = ""
483    pages_fetched = 0
484    message_ids: list[str] = []
485    log_line(
486        f"LIST start labels={label_ids or ['ALL']} query={query!r} page_size={page_size} max_pages={max_pages or 'all'} max_messages={max_messages or 'all'}"
487    )
488    while True:
489        params: dict[str, Any] = {
490            "userId": "me",
491            "maxResults": min(max(page_size, 1), 500),
492            "fields": "messages/id,nextPageToken,resultSizeEstimate",
493        }
494        if label_ids:
495            params["labelIds"] = label_ids
496        if query:
497            params["q"] = query
498        if page_token