path: root/mailctl.py

#!/usr/bin/env python3
#
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# 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

import mailrules

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 cmd_rules_list(args):
    store = mailrules.load()
    for warning in store.warnings:
        print(f"warning: {warning}", file=sys.stderr)
    if store.missing:
        print(f"No rules file at {mailrules.default_path()}", file=sys.stderr)
        return
    for rule in mailrules.ordered(store.rules) if args.enabled_only \
            else store.rules:
        state = " " if rule.enabled else "-"
        tags = " ".join(mailrules.tag_arguments(rule))
        print(f"{state} {rule.stage:>4}  {rule.id:<28.28}  {tags}")
    total = len(store.rules)
    disabled = sum(1 for r in store.rules if not r.enabled)
    print(f"\n{total} rule(s), {disabled} disabled", file=sys.stderr)


def cmd_rules_show(args):
    store = mailrules.load()
    for rule in store.rules:
        if rule.id != args.id:
            continue
        print(f"id:      {rule.id}")
        print(f"stage:   {rule.stage}")
        print(f"enabled: {rule.enabled}")
        print(f"add:     {', '.join(rule.add) or '(none)'}")
        print(f"remove:  {', '.join(rule.remove) or '(none)'}")
        print(f"query:   {rule.query}")
        if rule.note:
            print(f"note:    {rule.note}")
        return
    print(f"No rule with id '{args.id}'", file=sys.stderr)
    sys.exit(1)


def cmd_rules_dry_run(args):
    """What each rule matches right now. Two numbers, because they answer
    different questions: the corpus count says whether a rule is still doing
    anything and whether it would bury real correspondence, the pending count
    says what the next sync would tag."""
    store = mailrules.load()
    rules = [r for r in store.rules if not args.id or r.id == args.id]
    if not rules:
        print(f"No rule with id '{args.id}'", file=sys.stderr)
        sys.exit(1)

    print(f"{'rule':<28}  {'corpus':>8}  {'pending':>8}")
    for rule in rules:
        corpus = run_notmuch(["count", mailrules.scoped_query(rule, None)])
        pending = run_notmuch(["count",
                               mailrules.scoped_query(rule, "tag:new")])
        state = "" if rule.enabled else "  (disabled)"
        print(f"{rule.id:<28.28}  {corpus.strip():>8}  "
              f"{pending.strip():>8}{state}")


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)

    sp = sub.add_parser("rules", help="inspect the shared tagging rules "
                                      "(read-only)")
    rules_sub = sp.add_subparsers(dest="rules_command", required=True)

    rp = rules_sub.add_parser("list", help="list every rule in stage order")
    rp.add_argument("--enabled-only", action="store_true",
                    help="only the rules the hook would run")
    rp.set_defaults(func=cmd_rules_list)

    rp = rules_sub.add_parser("show", help="show one rule in full")
    rp.add_argument("id")
    rp.set_defaults(func=cmd_rules_show)

    rp = rules_sub.add_parser("dry-run",
                              help="count what each rule matches, changing "
                                   "nothing")
    rp.add_argument("id", nargs="?", help="one rule, default is all")
    rp.set_defaults(func=cmd_rules_dry_run)

    return p


def main():
    parser = build_parser()
    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()