#!/usr/bin/env python3 # # Copyright (C) 2026 Danilo M. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """ mailctl - a deliberately narrow CLI wrapper around notmuch, meant for agent use (Claude Code, opencode, etc.) to search and organize mail. Design goals: - No SMTP, no send, no reply, no compose. Not "discouraged", not present in the code at all. An agent can't do what the tool has no code path for. - Read operations (search/show/tags/count/senders/subjects) run freely, --account is optional there, defaulting to a global search across all mailboxes. - Mutating operations (tag) are scoped to a single account by default. Touching more than one account requires the explicit --all-accounts flag, there's no accidental global mutation. - Tag mutations default to dry-run: they print what WOULD change and require --apply to actually touch the index. - Anything that looks destructive (removing 'inbox', adding 'deleted' or 'trash'/'spam') requires --apply AND --confirm-destructive, and gets logged regardless of account scope. - Every applied mutation is appended to ~/.local/state/mailctl/audit.log with a timestamp, so there's a plain-text trail of what an agent changed. - The account map (maildir names and real From addresses) is NOT in this file. It loads from ~/.config/mailctl/accounts.json, override with MAILCTL_CONFIG. The config is validated against its schema and against the actual maildirs on disk at import time; any problem is a hard exit before argparse runs, so a typo can't reach a query or a draft. This is a starting skeleton, not a finished tool. In particular: - The audit log doesn't distinguish "run by agent" vs "run by you", worth adding an --actor tag if that distinction matters to you. """ import argparse import json import os import re import socket import subprocess import sys import time from collections import Counter from datetime import datetime, timezone from email.message import EmailMessage from email.utils import make_msgid, formatdate from pathlib import Path AUDIT_LOG = Path.home() / ".local" / "state" / "mailctl" / "audit.log" MAIL_ROOT = Path(os.environ.get("MAILCTL_MAIL_ROOT", Path.home() / "Mail")) CONFIG_PATH = Path( os.environ.get("MAILCTL_CONFIG", Path.home() / ".config" / "mailctl" / "accounts.json") ) # The account map is real addresses and maildir names, so it lives outside the # repo. Same closed-set guarantee as when it was hardcoded: it is validated # against the schema AND against the disk at import time, and mailctl refuses # to run at all if anything is off. A typo can't silently produce a path # filter that matches nothing, or a draft sent from the wrong identity. ACCOUNT_KEY_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$") ADDRESS_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") ACCOUNT_FIELDS = {"maildir", "address", "drafts"} def config_error(problem, *hints): """Refuse to run. Config problems are typos in the user's own file, so the message names the file, the problem, and how to fix it.""" print(f"mailctl: bad config at {CONFIG_PATH}\n {problem}", file=sys.stderr) for h in hints: print(f" {h}", file=sys.stderr) sys.exit(2) def validate_accounts(raw, mail_root): """Check a parsed config against the schema and the filesystem. Returns (accounts, drafts_subdir) in the same shape the rest of the tool used when these were module constants. Calls config_error (exits) on the first problem found. """ if not isinstance(raw, dict): config_error(f"top level must be a JSON object, got {type(raw).__name__}", 'example: {"accounts": {"work": {...}}}') accounts_raw = raw.get("accounts") if accounts_raw is None: config_error('missing top-level "accounts" key', f"found instead: {', '.join(sorted(raw)) or '(empty file)'}") if not isinstance(accounts_raw, dict) or not accounts_raw: config_error('"accounts" must be a non-empty JSON object') accounts, drafts = {}, {} for key, spec in accounts_raw.items(): where = f'account "{key}"' if not ACCOUNT_KEY_RE.match(key): config_error(f"{where}: invalid name", "names are lowercase letters, digits, . _ - and must " "not start with a separator") if not isinstance(spec, dict): config_error(f"{where}: must be an object, got {type(spec).__name__}") unknown = set(spec) - ACCOUNT_FIELDS if unknown: config_error(f"{where}: unknown field(s) {', '.join(sorted(unknown))}", f"valid fields: {', '.join(sorted(ACCOUNT_FIELDS))}") for required in ("maildir", "address"): if required not in spec: config_error(f"{where}: missing required field \"{required}\"") maildir = spec["maildir"] if not isinstance(maildir, str) or not maildir: config_error(f"{where}: \"maildir\" must be a non-empty string") if maildir.startswith("/") or ".." in Path(maildir).parts: config_error(f"{where}: \"maildir\" must be a plain subdirectory " f"of {mail_root}, got {maildir!r}") if not (mail_root / maildir).is_dir(): config_error(f"{where}: maildir {mail_root / maildir} does not exist", "check the spelling against what mbsync actually synced") address = spec["address"] if not isinstance(address, str) or not ADDRESS_RE.match(address): config_error(f"{where}: \"address\" is not a valid email address: " f"{address!r}") drafts_dir = spec.get("drafts") if drafts_dir is not None: if not isinstance(drafts_dir, str) or not drafts_dir: config_error(f"{where}: \"drafts\" must be a non-empty string " "or null (null = account has no synced Drafts)") if not (mail_root / maildir / drafts_dir).is_dir(): config_error( f"{where}: Drafts maildir " f"{mail_root / maildir / drafts_dir} does not exist", "it must match the mbsync Patterns for this account " '(e.g. "[Gmail]/Bozze" vs "Drafts"), or set it to null') accounts[key] = (maildir, address) drafts[key] = drafts_dir dupes = Counter(a for a, _ in accounts.values()) for maildir, n in dupes.items(): if n > 1: config_error(f"maildir {maildir!r} is used by {n} accounts", "each account needs its own maildir, or scoping " "silently matches the wrong mail") return accounts, drafts def load_accounts(path=None, mail_root=None): path = Path(path) if path else CONFIG_PATH mail_root = mail_root or MAIL_ROOT try: text = path.read_text() except FileNotFoundError: print(f"mailctl: no config at {path}\n" " Create it with one entry per account:\n" ' {"accounts": {"work": {"maildir": "work-mbsync-dir",\n' ' "address": "you@example.org",\n' ' "drafts": "Drafts"}}}\n' ' "drafts" may be null if that account has no synced Drafts ' "folder.\n" " Override the location with MAILCTL_CONFIG=/path/to/file.", file=sys.stderr) sys.exit(2) except OSError as e: config_error(f"cannot read: {e}") try: raw = json.loads(text) except json.JSONDecodeError as e: config_error(f"not valid JSON: {e}") return validate_accounts(raw, mail_root) ACCOUNTS, DRAFTS_SUBDIR = load_accounts() DESTRUCTIVE_TAGS = {"deleted", "trash", "spam"} PROTECTED_REMOVALS = {"inbox"} DEFAULT_MAX_MESSAGES = 5000 # abort a tag --apply touching more than this unless --max-messages raises it # Function words filtered out of `subjects` term counts. Italian and English # only, because that's what this mail is; not a general-purpose stopword list. # ponytail: a hand-rolled set, no NLP dependency for what is word counting. # Add words here when a useless term keeps topping a real query. STOPWORDS = { # Italian "di", "il", "la", "le", "lo", "gli", "un", "una", "uno", "del", "della", "dei", "delle", "dello", "al", "alla", "ai", "alle", "allo", "da", "dal", "dalla", "in", "nel", "nella", "con", "su", "sul", "sulla", "per", "tra", "fra", "che", "chi", "cui", "non", "come", "piu", "più", "anche", "sono", "sei", "hai", "ha", "ho", "essere", "questo", "questa", "questi", "queste", "quello", "quella", "tuo", "tua", "tuoi", "tue", "mio", "mia", "ti", "si", "ci", "ne", "se", "ma", "così", "cosa", "tutto", "tutti", "già", "ora", "oggi", "solo", "ed", "od", "sta", "fa", "qui", "te", "lì", "là", "dove", "quando", "molto", "ancora", "sempre", "poi", # English (words under 2 chars are dropped by length, not listed here) "the", "an", "of", "to", "on", "at", "for", "and", "or", "but", "is", "are", "was", "were", "be", "been", "your", "you", "my", "it", "its", "this", "that", "these", "those", "with", "from", "by", "as", "we", "our", "has", "have", "had", "will", "can", "not", "new", "now", "all", "more", "re", "fwd", } def run_notmuch(args, capture=True): cmd = ["notmuch"] + args result = subprocess.run(cmd, capture_output=capture, text=True) if result.returncode != 0: print(f"notmuch error: {result.stderr.strip()}", file=sys.stderr) sys.exit(result.returncode) return result.stdout if capture else None def scoped_query(query, account): """Wrap a user query with a path: filter for the given account key. Returns the query unchanged if account is None (global scope).""" if account is None: return query if account not in ACCOUNTS: print(f"Unknown account '{account}'. Valid accounts: " f"{', '.join(ACCOUNTS)}", file=sys.stderr) sys.exit(1) subdir, _address = ACCOUNTS[account] return f'path:"{subdir}/**" and ({query})' def account_choices_help(): return "one of: " + ", ".join(ACCOUNTS) def cmd_search(args): q = scoped_query(args.query, args.account) out = run_notmuch(["search", "--format=json", "--output=summary", q]) results = json.loads(out) if args.json: print(json.dumps(results, indent=2)) return for r in results: date = r.get("date_relative", "") frm = r.get("authors", "") subj = r.get("subject", "(no subject)") tags = ",".join(r.get("tags", [])) print(f"{date:>12} {frm:<30.30} {subj:<60.60} [{tags}]") scope = args.account or "all accounts" print(f"\n{len(results)} thread(s) [scope: {scope}]", file=sys.stderr) def cmd_show(args): out = run_notmuch(["show", "--format=json", args.query]) print(out) def cmd_tags(args): out = run_notmuch(["search", "--output=tags", "*"]) print(out.strip()) def cmd_count(args): q = scoped_query(args.query, args.account) out = run_notmuch(["count", q]) scope = args.account or "all accounts" print(f"{out.strip()} [scope: {scope}]") def cmd_senders(args): q = scoped_query(args.query, args.account) out = run_notmuch(["address", "--output=sender", "--output=count", "--format=json", q]) # notmuch dedupes on name-addr, so one address shows up once per display # name it ever used. Merge on the address, keeping the longest name seen. merged = {} for a in json.loads(out): e = merged.setdefault(a["address"], {"address": a["address"], "name": "", "count": 0}) e["count"] += a["count"] if len(a["name"]) > len(e["name"]): e["name"] = a["name"] results = sorted(merged.values(), key=lambda a: a["count"], reverse=True) if args.top: results = results[:args.top] if args.json: print(json.dumps(results, indent=2)) return for a in results: print(f"{a['count']:>7} {a['address']:<45.45} {a['name']:.35}") scope = args.account or "all accounts" print(f"\n{len(results)} sender(s) [scope: {scope}]", file=sys.stderr) def subject_terms(subjects): """Count word frequency across subject lines, ignoring stopwords. Returns a Counter. Mail here is mixed Italian/English and heavy on emoji and marketing punctuation, so tokens are lowercased word characters only (emoji and '...' fall out), and single characters plus pure digits are dropped as noise. """ counts = Counter() for subj in subjects: seen = set() for word in re.findall(r"\w+", subj.lower()): if len(word) < 2 or word.isdigit() or word in STOPWORDS: continue seen.add(word) # count each term once per subject, so one shouty repeated word in a # single subject can't outrank a term used across many messages counts.update(seen) return counts def cmd_subjects(args): q = scoped_query(args.query, args.account) out = run_notmuch(["search", "--format=json", "--output=summary", q]) results = json.loads(out) subjects = [r.get("subject") or "" for r in results] terms = subject_terms(subjects).most_common(args.top) if args.json: print(json.dumps([{"term": t, "threads": n} for t, n in terms], indent=2)) return total = len(subjects) for term, n in terms: pct = 100 * n / total if total else 0 print(f"{n:>7} {pct:>5.1f}% {term}") scope = args.account or "all accounts" print(f"\n{len(terms)} term(s) across {total} thread(s) " f"[scope: {scope}]", file=sys.stderr) def cmd_tag(args): adds = args.add or [] removes = args.remove or [] if not adds and not removes: print("Nothing to do: specify --add and/or --remove", file=sys.stderr) sys.exit(1) # --- account scoping gate, applies before anything else --- if not args.account and not args.all_accounts: print("Refusing: 'tag' needs either --account NAME (recommended) " "or --all-accounts (explicit, for a deliberate cross-account " f"change). {account_choices_help()}", file=sys.stderr) sys.exit(1) if args.account and args.account not in ACCOUNTS: print(f"Unknown account '{args.account}'. {account_choices_help()}", file=sys.stderr) sys.exit(1) q = scoped_query(args.query, args.account) # None if --all-accounts is_destructive = bool( set(adds) & DESTRUCTIVE_TAGS or set(removes) & PROTECTED_REMOVALS ) count = run_notmuch(["count", q]).strip() tag_expr = [f"+{t}" for t in adds] + [f"-{t}" for t in removes] scope = args.account or "ALL ACCOUNTS" print(f"Scope: {scope}") print(f"Query: {args.query}") print(f"Effective query: {q}") print(f"Matches: {count} message(s)") print(f"Change: {' '.join(tag_expr)}") if is_destructive: print("\n[!] This includes a destructive tag change " f"({DESTRUCTIVE_TAGS | PROTECTED_REMOVALS} related).") n = int(count) if n > args.max_messages: print(f"\n[cap] {n} matches exceeds --max-messages " f"({args.max_messages}).") if not args.apply: print("\nDry run only. Re-run with --apply to actually change tags.") return if n > args.max_messages: print(f"\nRefusing to apply: {n} messages exceeds the " f"--max-messages cap ({args.max_messages}). Re-run with a " "higher --max-messages if this is intended.", file=sys.stderr) sys.exit(1) if is_destructive and not args.confirm_destructive: print("\nRefusing to apply: destructive change needs " "--apply AND --confirm-destructive.", file=sys.stderr) sys.exit(1) run_notmuch(["tag"] + tag_expr + ["--", q], capture=False) log_mutation(scope, args.query, tag_expr, int(count)) print(f"\nApplied. {count} message(s) affected.") print("Note: if synchronize_flags is on, this will also update Maildir " "flags and may propagate to the IMAP server on next mbsync run.") def cmd_draft(args): if args.account not in ACCOUNTS: print(f"Unknown account '{args.account}'. {account_choices_help()}", file=sys.stderr) sys.exit(1) subdir, from_addr = ACCOUNTS[args.account] drafts_subdir = DRAFTS_SUBDIR.get(args.account) if drafts_subdir is None: print(f"Account '{args.account}' has no synced Drafts folder " "(not in its mbsync Patterns). Pick a different account " "or add Drafts to that account's sync config first.", file=sys.stderr) sys.exit(1) drafts_path = MAIL_ROOT / subdir / drafts_subdir if not drafts_path.is_dir(): print(f"Expected Drafts maildir not found at {drafts_path}. " "Has this account been synced yet?", file=sys.stderr) sys.exit(1) body = args.body if args.body_file: body = Path(args.body_file).read_text() if body is None: body = sys.stdin.read() msg = EmailMessage() msg["From"] = from_addr msg["To"] = args.to if args.cc: msg["Cc"] = args.cc msg["Subject"] = args.subject msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid() msg.set_content(body) # Maildir atomic write: build the full file in tmp/, then rename # (not copy) into new/. Any reader (neomutt, notmuch, mbsync) only # ever sees either "not there yet" or "fully written", never a # partial file. unique = f"{int(time.time())}.M{os.getpid()}P{id(msg) % 100000}.{socket.gethostname()}" tmp_path = drafts_path / "tmp" / unique new_path = drafts_path / "new" / unique tmp_path.write_bytes(msg.as_bytes()) os.rename(tmp_path, new_path) print(f"Draft written: {new_path}") print(f"From: {from_addr}") print(f"To: {args.to}") print(f"Subject: {args.subject}") print("\nThis is a LOCAL draft only. It will appear on the server " "(and other devices) after the next mbsync run. Nothing has " "been sent, review and send it yourself in neomutt.") def log_mutation(scope, query, tag_expr, count): AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).isoformat(timespec="seconds") with open(AUDIT_LOG, "a") as f: f.write(f"{ts}\tscope={scope}\tquery={query!r}\t" f"change={' '.join(tag_expr)}\tcount={count}\n") def build_parser(): p = argparse.ArgumentParser( prog="mailctl", description="Read-heavy, agent-safe wrapper around notmuch. " "No send/reply/compose capability exists in this tool.", ) sub = p.add_subparsers(dest="command", required=True) sp = sub.add_parser("search", help="search mail, notmuch query syntax") sp.add_argument("query") sp.add_argument("--account", choices=list(ACCOUNTS), help="scope to one account, default is global") sp.add_argument("--json", action="store_true") sp.set_defaults(func=cmd_search) sp = sub.add_parser("show", help="show a full thread/message") sp.add_argument("query", help="e.g. thread:0000... or id:...") sp.set_defaults(func=cmd_show) sp = sub.add_parser("tags", help="list all tags currently in use") sp.set_defaults(func=cmd_tags) sp = sub.add_parser("count", help="count messages matching a query") sp.add_argument("query") sp.add_argument("--account", choices=list(ACCOUNTS), help="scope to one account, default is global") sp.set_defaults(func=cmd_count) sp = sub.add_parser("senders", help="rank senders by message count") sp.add_argument("query") sp.add_argument("--account", choices=list(ACCOUNTS), help="scope to one account, default is global") sp.add_argument("--top", type=int, metavar="N", help="show only the top N senders") sp.add_argument("--json", action="store_true") sp.set_defaults(func=cmd_senders) sp = sub.add_parser("subjects", help="rank subject terms by how many " "threads use them") sp.add_argument("query") sp.add_argument("--account", choices=list(ACCOUNTS), help="scope to one account, default is global") sp.add_argument("--top", type=int, default=25, metavar="N", help="show only the top N terms (default 25)") sp.add_argument("--json", action="store_true") sp.set_defaults(func=cmd_subjects) sp = sub.add_parser("tag", help="add/remove tags (dry-run unless --apply)") sp.add_argument("query") sp.add_argument("--account", choices=list(ACCOUNTS), help="required unless --all-accounts is given") sp.add_argument("--all-accounts", action="store_true", help="explicit opt-in to a cross-account mutation") sp.add_argument("--add", action="append", metavar="TAG") sp.add_argument("--remove", action="append", metavar="TAG") sp.add_argument("--apply", action="store_true", help="actually apply the change, default is dry-run") sp.add_argument("--confirm-destructive", action="store_true", help="required in addition to --apply for " "deleted/trash/spam or removing 'inbox'") sp.add_argument("--max-messages", type=int, default=DEFAULT_MAX_MESSAGES, metavar="N", help=f"abort --apply if the match count exceeds N " f"(default {DEFAULT_MAX_MESSAGES})") sp.set_defaults(func=cmd_tag) sp = sub.add_parser("draft", help="write a draft to the account's Drafts " "folder for later review/sending in " "neomutt. Never sends anything.") sp.add_argument("--account", required=True, choices=list(ACCOUNTS), help="which identity/mailbox to draft into") sp.add_argument("--to", required=True) sp.add_argument("--cc") sp.add_argument("--subject", required=True) sp.add_argument("--body", help="draft body text") sp.add_argument("--body-file", help="read body from a file instead") sp.set_defaults(func=cmd_draft) return p def main(): parser = build_parser() args = parser.parse_args() args.func(args) if __name__ == "__main__": main()