scripts/m365_mail.py — Approval-Gated Mail

#!/usr/bin/env python3
# ============================================================
# File: m365_mail.py — Approval-Gated Mail 0.2.1
# Created by: Zero Width Lab
# Date: 2026-09-12
# Description: Read, search, save, reply to, and send Microsoft 365 and Gmail mail.
# By Zero Width Lab (zerowidthlab.com)
# ============================================================

from __future__ import annotations

import argparse
import hashlib
import html
import hmac
import json
import os
import sys
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from html.parser import HTMLParser
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence
from typing import Protocol

__author__ = "Zero Width Lab"

GRAPH_ROOT = "https://graph.microsoft.com/v1.0"
LOGIN_HOST = "login.microsoftonline.com"
GRAPH_HOST = "graph.microsoft.com"
DEFAULT_MAX_CHARS = 4000
MESSAGE_FIELDS = (
  "id,subject,from,toRecipients,ccRecipients,bccRecipients,receivedDateTime,"
  "sentDateTime,isRead,hasAttachments,conversationId,body"
)
SUMMARY_FIELDS = (
  "id,subject,from,receivedDateTime,sentDateTime,isRead,hasAttachments,"
  "conversationId,toRecipients,ccRecipients,bccRecipients"
)
SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"


class CliError(Exception):
  """An expected, user-facing command failure."""


class MailBackend(Protocol):
  def list_messages(self, mailbox: str, count: int, search: str | None = None, sender: str | None = None) -> list[dict[str, Any]]: ...
  def get_message(self, mailbox: str, message_id: str, include_body: bool = True) -> dict[str, Any]: ...
  def thread_messages(self, mailbox: str, thread_id: str) -> list[dict[str, Any]]: ...


class SkillArgumentParser(argparse.ArgumentParser):
  """Make all argument errors follow the command's exit-code contract."""

  def __init__(self, *args: Any, **kwargs: Any) -> None:
    kwargs.setdefault("epilog", SIGNATURE)
    super().__init__(*args, **kwargs)

  def error(self, message: str) -> None:
    raise CliError(message)


class ReadableHTMLParser(HTMLParser):
  """Small standard-library HTML to readable-text converter."""

  BLOCK_TAGS = {
    "address",
    "article",
    "aside",
    "blockquote",
    "br",
    "div",
    "dl",
    "dt",
    "dd",
    "footer",
    "h1",
    "h2",
    "h3",
    "h4",
    "h5",
    "h6",
    "header",
    "hr",
    "li",
    "main",
    "ol",
    "p",
    "pre",
    "section",
    "table",
    "tr",
    "ul",
  }
  HIDDEN_TAGS = {"script", "style", "template"}

  def __init__(self) -> None:
    super().__init__(convert_charrefs=True)
    self.parts: list[str] = []
    self.hidden_depth = 0

  def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
    del attrs
    lowered = tag.lower()
    if lowered in self.HIDDEN_TAGS:
      self.hidden_depth += 1
    elif not self.hidden_depth and lowered in self.BLOCK_TAGS:
      self.parts.append("\n")

  def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
    self.handle_starttag(tag, attrs)
    if tag.lower() in self.HIDDEN_TAGS:
      self.handle_endtag(tag)

  def handle_endtag(self, tag: str) -> None:
    lowered = tag.lower()
    if lowered in self.HIDDEN_TAGS:
      self.hidden_depth = max(0, self.hidden_depth - 1)
    elif not self.hidden_depth and lowered in self.BLOCK_TAGS:
      self.parts.append("\n")

  def handle_data(self, data: str) -> None:
    if not self.hidden_depth:
      self.parts.append(data)

  def text(self) -> str:
    lines: list[str] = []
    for raw_line in "".join(self.parts).replace("\r", "").split("\n"):
      cleaned = " ".join(raw_line.split())
      if cleaned:
        lines.append(cleaned)
      elif lines and lines[-1] != "":
        lines.append("")
    while lines and not lines[-1]:
      lines.pop()
    return "\n".join(lines)


def html_to_text(value: str) -> str:
  parser = ReadableHTMLParser()
  try:
    parser.feed(value)
    parser.close()
    return parser.text()
  except (AssertionError, ValueError):
    return " ".join(html.unescape(value).split())


def positive_int(value: str) -> int:
  try:
    parsed = int(value)
  except ValueError as exc:
    raise argparse.ArgumentTypeError(
      "must be an integer greater than zero"
    ) from exc
  if parsed < 1:
    raise argparse.ArgumentTypeError("must be an integer greater than zero")
  return parsed


def load_config(path_value: str | None) -> dict[str, Any]:
  if not path_value:
    return {}
  path = Path(path_value)
  try:
    content = path.read_text(encoding="utf-8")
  except OSError as exc:
    raise CliError(f"cannot read config file {path}: {exc}") from exc
  try:
    parsed = json.loads(content)
  except json.JSONDecodeError as exc:
    raise CliError(f"invalid JSON config file {path}: {exc}") from exc
  if not isinstance(parsed, dict):
    raise CliError("config file must contain a JSON object")
  return parsed


def config_value(config: Mapping[str, Any], name: str) -> str | None:
  value = os.environ.get(name)
  if value:
    return value
  candidates = (name, name.lower(), name.removeprefix("M365_").lower())
  for key in candidates:
    candidate = config.get(key)
    if isinstance(candidate, str) and candidate:
      return candidate
  return None


def quote_segment(value: str) -> str:
  return urllib.parse.quote(value, safe="")


def query_url(path: str, params: Mapping[str, Any] | None = None) -> str:
  url = f"{GRAPH_ROOT}{path}"
  if params:
    url = f"{url}?{urllib.parse.urlencode(params)}"
  return url


def quoted_kql_term(value: str) -> str:
  return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'


def checked_remote_url(url: str) -> None:
  parsed = urllib.parse.urlparse(url)
  if parsed.scheme != "https" or parsed.hostname not in {GRAPH_HOST, LOGIN_HOST}:
    raise CliError(f"refusing network access outside approved hosts: {url}")


class ApprovedRedirectHandler(urllib.request.HTTPRedirectHandler):
  """Allow redirects only when the destination remains on an approved host."""

  def redirect_request(
    self,
    req: urllib.request.Request,
    fp: Any,
    code: int,
    msg: str,
    headers: Any,
    newurl: str,
  ) -> urllib.request.Request | None:
    checked_remote_url(newurl)
    if (
      urllib.parse.urlparse(req.full_url).hostname
      != urllib.parse.urlparse(newurl).hostname
    ):
      raise CliError("refusing a cross-host HTTP redirect")
    return super().redirect_request(req, fp, code, msg, headers, newurl)


class GraphClient:
  def __init__(self, config: Mapping[str, Any]) -> None:
    self.config = config
    self.access_token: str | None = None
    self.fake_path = os.environ.get("M365_MAIL_FAKE_TRANSPORT")
    self.fake_responses: list[dict[str, Any]] = []
    self.opener = urllib.request.build_opener(ApprovedRedirectHandler())
    if self.fake_path:
      if os.environ.get("M365_MAIL_ALLOW_FAKE") != "1":
        raise CliError(
          "M365_MAIL_FAKE_TRANSPORT is set but M365_MAIL_ALLOW_FAKE=1 "
          "was not also set; refusing to run. This second gate "
          "prevents an inherited or accidentally-set environment "
          "variable from silently faking every Graph API call "
          "instead of making a real request."
        )
      print(
        "warning: M365_MAIL_FAKE_TRANSPORT is set -- every Graph API "
        f"call in this run is served from the fixture at "
        f"{self.fake_path} and no real network request will be made.",
        file=sys.stderr,
      )
      self.fake_responses = self._load_fake_responses(Path(self.fake_path))

  @staticmethod
  def _load_fake_responses(path: Path) -> list[dict[str, Any]]:
    try:
      content = path.read_text(encoding="utf-8")
    except OSError as exc:
      raise CliError(f"cannot read fake transport fixture {path}: {exc}") from exc
    try:
      parsed = json.loads(content)
    except json.JSONDecodeError as exc:
      raise CliError(f"invalid fake transport JSON {path}: {exc}") from exc
    if isinstance(parsed, dict):
      parsed = parsed.get("responses")
    if not isinstance(parsed, list) or not all(
      isinstance(item, dict) for item in parsed
    ):
      raise CliError(
        "fake transport fixture must be a response list or an object "
        "with a responses list"
      )
    return list(parsed)

  def _token(self) -> str:
    if self.access_token:
      return self.access_token
    tenant_id = config_value(self.config, "M365_TENANT_ID")
    client_id = config_value(self.config, "M365_CLIENT_ID")
    client_secret = config_value(self.config, "M365_CLIENT_SECRET")
    missing = [
      name
      for name, value in (
        ("M365_TENANT_ID", tenant_id),
        ("M365_CLIENT_ID", client_id),
        ("M365_CLIENT_SECRET", client_secret),
      )
      if not value
    ]
    if missing:
      raise CliError(f"missing required configuration: {', '.join(missing)}")
    token_url = f"https://{LOGIN_HOST}/{quote_segment(tenant_id)}/oauth2/v2.0/token"
    form = urllib.parse.urlencode(
      {
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "https://graph.microsoft.com/.default",
        "grant_type": "client_credentials",
      }
    ).encode("utf-8")
    response = self.request(
      "POST",
      token_url,
      raw_body=form,
      headers={"Content-Type": "application/x-www-form-urlencoded"},
      authenticated=False,
    )
    if not isinstance(response, dict) or not isinstance(
      response.get("access_token"), str
    ):
      raise CliError("token endpoint response did not contain an access token")
    self.access_token = response["access_token"]
    return self.access_token

  def request(
    self,
    method: str,
    url: str,
    *,
    json_body: Any = None,
    raw_body: bytes | None = None,
    headers: Mapping[str, str] | None = None,
    authenticated: bool = True,
  ) -> Any:
    checked_remote_url(url)
    method = method.upper()
    if self.fake_path:
      return self._fake_request(method, url, json_body)
    request_headers = {"Accept": "application/json"}
    if headers:
      request_headers.update(headers)
    if authenticated:
      request_headers["Authorization"] = f"Bearer {self._token()}"
    data = raw_body
    if json_body is not None:
      data = json.dumps(json_body, ensure_ascii=False).encode("utf-8")
      request_headers["Content-Type"] = "application/json"
    request = urllib.request.Request(
      url, data=data, headers=request_headers, method=method
    )
    try:
      with self.opener.open(request, timeout=30) as response:
        payload = response.read()
        status = response.status
    except urllib.error.HTTPError as exc:
      detail = exc.read().decode("utf-8", errors="replace").strip()
      suffix = f": {detail}" if detail else ""
      retry_after = exc.headers.get("Retry-After") if exc.headers else None
      if retry_after:
        suffix += f" (Retry-After: {retry_after})"
      raise CliError(f"HTTP {exc.code} from {url}{suffix}") from exc
    except urllib.error.URLError as exc:
      raise CliError(f"network request failed for {url}: {exc.reason}") from exc
    except TimeoutError as exc:
      raise CliError(f"network request timed out for {url}") from exc
    if status < 200 or status >= 300:
      raise CliError(f"HTTP {status} from {url}")
    if not payload:
      return None
    try:
      return json.loads(payload.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
      raise CliError(f"invalid JSON response from {url}: {exc}") from exc

  def _fake_request(self, method: str, url: str, json_body: Any) -> Any:
    parsed_url = urllib.parse.urlparse(url)
    for index, response in enumerate(self.fake_responses):
      expected_method = str(response.get("method", method)).upper()
      if expected_method != method:
        continue
      exact_url = response.get("url")
      contains = response.get("url_contains")
      expected_path = response.get("path")
      if exact_url is not None and exact_url != url:
        continue
      if contains is not None and str(contains) not in url:
        continue
      if expected_path is not None and str(expected_path) != parsed_url.path:
        continue
      expected_json = response.get("request_json")
      if "request_json" in response and expected_json != json_body:
        continue
      matched = self.fake_responses.pop(index)
      status = int(matched.get("status", 200))
      if status < 200 or status >= 300:
        detail_value = matched.get("body", matched.get("json", ""))
        detail = (
          json.dumps(detail_value, ensure_ascii=False)
          if isinstance(detail_value, (dict, list))
          else str(detail_value)
        )
        suffix = f": {detail}" if detail else ""
        fake_headers = matched.get("headers")
        retry_after = (
          fake_headers.get("Retry-After")
          if isinstance(fake_headers, dict)
          else None
        )
        if retry_after:
          suffix += f" (Retry-After: {retry_after})"
        raise CliError(f"HTTP {status} from {url}{suffix}")
      if "json" in matched:
        return matched["json"]
      body = matched.get("body")
      if body in (None, ""):
        return None
      if isinstance(body, (dict, list)):
        return body
      try:
        return json.loads(str(body))
      except json.JSONDecodeError as exc:
        raise CliError(
          f"invalid JSON response from fake transport: {exc}"
        ) from exc
    raise CliError(f"fake transport has no matching response for {method} {url}")

  def get_collection(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    max_records: int | None = None,
  ) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    next_url: str | None = url
    while next_url:
      response = self.request("GET", next_url, headers=headers)
      if not isinstance(response, dict) or not isinstance(
        response.get("value"), list
      ):
        raise CliError(f"Graph collection response is invalid for {next_url}")
      records.extend(item for item in response["value"] if isinstance(item, dict))
      if max_records is not None and len(records) >= max_records:
        return records[:max_records]
      candidate = response.get("@odata.nextLink")
      next_url = candidate if isinstance(candidate, str) and candidate else None
    return records


def mailbox_for(args: argparse.Namespace, config: Mapping[str, Any]) -> str:
  mailbox = getattr(args, "user", None) or config_value(config, "M365_DEFAULT_USER")
  if not mailbox:
    raise CliError("mailbox is required, pass --user or set M365_DEFAULT_USER")
  return mailbox


def sender_for(args: argparse.Namespace, config: Mapping[str, Any]) -> str:
  from_user = getattr(args, "from_user", None)
  user = getattr(args, "user", None)
  sender = from_user or user or config_value(config, "M365_DEFAULT_USER")
  if not sender:
    raise CliError("send requires --from, --user, or M365_DEFAULT_USER")
  if from_user and user and user.casefold() != from_user.casefold():
    raise CliError("--user and --from must identify the same sending mailbox")
  return sender


def email_address(value: Any) -> dict[str, str]:
  if not isinstance(value, dict):
    return {"name": "", "address": ""}
  nested = value.get("emailAddress")
  if not isinstance(nested, dict):
    return {"name": "", "address": ""}
  return {
    "name": str(nested.get("name") or ""),
    "address": str(nested.get("address") or ""),
  }


def addresses(values: Any) -> list[dict[str, str]]:
  if not isinstance(values, list):
    return []
  return [email_address(value) for value in values]


def display_address(value: Mapping[str, str]) -> str:
  name = display_text(value.get("name", ""))
  address = display_text(value.get("address", ""))
  if name and address:
    return f"{name} <{address}>"
  return address or name


def unicode_escape(character: str) -> str:
  code = ord(character)
  return f"\\u{code:04x}" if code <= 0xffff else f"\\U{code:08x}"


def invisible_character(character: str) -> bool:
  code = ord(character)
  return code in {0x034f, 0x115f, 0x1160, 0x3164, 0xffa0, 0x2800} or 0x180b <= code <= 0x180f or 0xfe00 <= code <= 0xfe0f or 0xe0100 <= code <= 0xe01ef


def display_text(value: Any) -> str:
  return "".join(unicode_escape(char) if not char.isprintable() or char == "\\" or invisible_character(char) else char for char in str(value))


def display_body(value: Any) -> str:
  rendered = str(value)
  return "".join(
    unicode_escape(char) if char not in "\n\t" and (
      unicodedata.category(char) in {"Cc", "Cf", "Co", "Cn", "Cs", "Zl", "Zp"}
      or (unicodedata.category(char) == "Zs" and char != " ")
      or invisible_character(char)
      or (char == "\\" and rendered[index + 1:index + 2] in {"u", "U"})
    ) else char for index, char in enumerate(rendered)
  )


def print_preview_body(value: Any) -> None:
  print("--- body start ---")
  for line in display_body(value).split("\n"):
    print("| " + line)
  print("--- body end ---")


def display_addresses(values: Sequence[Mapping[str, str]]) -> str:
  return ", ".join(display_address(value) for value in values)


def body_text(message: Mapping[str, Any]) -> str:
  body = message.get("body")
  if not isinstance(body, dict):
    return ""
  content = str(body.get("content") or "")
  content_type = str(body.get("contentType") or "").casefold()
  if content_type == "html":
    return html_to_text(content)
  return content.replace("\r\n", "\n").replace("\r", "\n")


def message_record(message: Mapping[str, Any], *, include_body: bool) -> dict[str, Any]:
  record: dict[str, Any] = {
    "id": str(message.get("id") or ""),
    "subject": str(message.get("subject") or ""),
    "from": email_address(message.get("from")),
    "to": addresses(message.get("toRecipients")),
    "cc": addresses(message.get("ccRecipients")),
    "bcc": addresses(message.get("bccRecipients")),
    "receivedDateTime": str(message.get("receivedDateTime") or ""),
    "sentDateTime": str(message.get("sentDateTime") or ""),
    "isRead": bool(message.get("isRead", False)),
    "hasAttachments": bool(message.get("hasAttachments", False)),
    "conversationId": str(message.get("conversationId") or ""),
  }
  if include_body:
    record["body"] = body_text(message)
  return record


def truncate_body(record: dict[str, Any], max_chars: int, full: bool) -> None:
  body = str(record.get("body") or "")
  record["bodyTruncated"] = False
  if not full and len(body) > max_chars:
    record["body"] = body[:max_chars]
    record["bodyTruncated"] = True


def print_json(value: Any) -> None:
  print(json.dumps(value, indent=2, ensure_ascii=False))


def print_summary(records: Sequence[Mapping[str, Any]]) -> None:
  if not records:
    print("No messages found.")
    return
  for index, record in enumerate(records):
    if index:
      print()
    print(f"ID: {display_text(record.get('id', ''))}")
    print(f"Subject: {display_text(record.get('subject', ''))}")
    print(f"From: {display_address(record.get('from', {}))}")
    print(f"Received: {display_text(record.get('receivedDateTime', ''))}")
    print(f"Read: {'yes' if record.get('isRead') else 'no'}")
    print(f"Attachments: {'yes' if record.get('hasAttachments') else 'no'}")
    print(f"Conversation ID: {display_text(record.get('conversationId', ''))}")


def print_message(record: Mapping[str, Any]) -> None:
  print(f"Subject: {display_text(record.get('subject', ''))}")
  print(f"From: {display_address(record.get('from', {}))}")
  print(f"To: {display_addresses(record.get('to', []))}")
  print(f"Cc: {display_addresses(record.get('cc', []))}")
  print(f"Date: {display_text(record.get('receivedDateTime') or record.get('sentDateTime', ''))}")
  print(f"Read: {'yes' if record.get('isRead') else 'no'}")
  print(f"Attachments: {'yes' if record.get('hasAttachments') else 'no'}")
  print(f"Conversation ID: {display_text(record.get('conversationId', ''))}")
  print(f"Message ID: {display_text(record.get('id', ''))}")
  print()
  print(display_body(record.get("body", "")))
  if record.get("bodyTruncated"):
    print("\n[body truncated]")


def list_messages(
  client: GraphClient,
  mailbox: str,
  *,
  count: int,
  search: str | None = None,
) -> list[dict[str, Any]]:
  params: dict[str, Any] = {"$select": SUMMARY_FIELDS, "$top": count}
  headers: dict[str, str] = {}
  if search is None:
    params["$orderby"] = "receivedDateTime desc"
  else:
    params["$search"] = quoted_kql_term(search)
    headers["ConsistencyLevel"] = "eventual"
  url = query_url(f"/users/{quote_segment(mailbox)}/messages", params)
  values = client.get_collection(url, headers=headers, max_records=count)
  return [message_record(value, include_body=False) for value in values[:count]]


def get_message(client: GraphClient, mailbox: str, message_id: str) -> dict[str, Any]:
  url = query_url(
    f"/users/{quote_segment(mailbox)}/messages/{quote_segment(message_id)}",
    {"$select": MESSAGE_FIELDS},
  )
  value = client.request("GET", url)
  if not isinstance(value, dict):
    raise CliError("Graph message response is invalid")
  return message_record(value, include_body=True)


def thread_messages(
  client: GraphClient, mailbox: str, conversation_id: str
) -> list[dict[str, Any]]:
  escaped_id = conversation_id.replace("'", "''")
  params = {
    "$filter": f"conversationId eq '{escaped_id}'",
    "$select": MESSAGE_FIELDS,
    "$top": 100,
  }
  url = query_url(f"/users/{quote_segment(mailbox)}/messages", params)
  values = client.get_collection(url)
  records = [message_record(value, include_body=True) for value in values]
  return sorted(
    records,
    key=lambda item: str(
      item.get("receivedDateTime") or item.get("sentDateTime") or ""
    ),
  )


def parse_recipients(
  values: Iterable[str] | None, label: str, *, required: bool = False
) -> list[str]:
  recipients: list[str] = []
  for value in values or []:
    recipients.extend(part.strip() for part in value.split(",") if part.strip())
  if required and not recipients:
    raise CliError(f"send requires at least one --{label} recipient")
  if any("\n" in item or "\r" in item for item in recipients):
    raise CliError(f"invalid newline in --{label} recipient")
  return recipients


def recipient_payload(values: Sequence[str]) -> list[dict[str, dict[str, str]]]:
  return [{"emailAddress": {"address": value}} for value in values]


def body_from_args(args: argparse.Namespace) -> str:
  inline = getattr(args, "body", None)
  body_file = getattr(args, "body_file", None)
  if (inline is None) == (body_file is None):
    raise CliError("send requires exactly one of --body or --body-file")
  if inline is not None:
    return inline
  path = Path(body_file)
  try:
    return path.read_text(encoding="utf-8")
  except OSError as exc:
    raise CliError(f"cannot read body file {path}: {exc}") from exc


def sent_items(client: GraphClient, mailbox: str, count: int) -> list[dict[str, Any]]:
  params = {
    "$select": SUMMARY_FIELDS,
    "$orderby": "sentDateTime desc",
    "$top": count,
  }
  url = query_url(
    f"/users/{quote_segment(mailbox)}/mailFolders/sentitems/messages", params
  )
  values = client.get_collection(url, max_records=count)
  return [message_record(value, include_body=False) for value in values[:count]]


def canonical_recipient_set(values: Sequence[Mapping[str, str]]) -> set[str]:
  return {
    str(value.get("address") or "").strip().casefold()
    for value in values
    if value.get("address")
  }


def match_sent_record(
  records: Sequence[dict[str, Any]],
  subject: str,
  to_values: Sequence[str] | None,
  cc_values: Sequence[str] | None,
  bcc_values: Sequence[str] | None,
  excluded_ids: set[str] | None,
) -> dict[str, Any] | None:
  expected = {
    "to": None if to_values is None else {value.casefold() for value in to_values},
    "cc": None if cc_values is None else {value.casefold() for value in cc_values},
    "bcc": None
    if bcc_values is None
    else {value.casefold() for value in bcc_values},
  }
  for record in records:
    if excluded_ids and str(record.get("id") or "") in excluded_ids:
      continue
    if str(record.get("subject") or "") != subject:
      continue
    if any(
      expected[label] is not None
      and canonical_recipient_set(record.get(label, [])) != expected[label]
      for label in ("to", "cc", "bcc")
    ):
      continue
    return record
  return None


def verify_sent_record(
  client: GraphClient,
  mailbox: str,
  subject: str,
  to_values: Sequence[str] | None,
  cc_values: Sequence[str] | None,
  bcc_values: Sequence[str] | None,
  count: int,
  *,
  retries: int,
  excluded_ids: set[str] | None = None,
) -> dict[str, Any]:
  for attempt in range(retries):
    record = match_sent_record(
      sent_items(client, mailbox, count),
      subject,
      to_values,
      cc_values,
      bcc_values,
      excluded_ids,
    )
    if record:
      return record
    if attempt + 1 < retries and not client.fake_path:
      time.sleep(attempt + 1)
  raise CliError("matching message was not found in Sent Items")


def build_send_request(
  mailbox: str,
  *,
  to_values: Sequence[str],
  cc_values: Sequence[str],
  bcc_values: Sequence[str],
  subject: str,
  body: str,
  reply_message_id: str | None,
) -> dict[str, Any]:
  message = {
    "subject": subject,
    "body": {"contentType": "Text", "content": body},
    "toRecipients": recipient_payload(to_values),
    "ccRecipients": recipient_payload(cc_values),
    "bccRecipients": recipient_payload(bcc_values),
  }
  if reply_message_id:
    url = query_url(
      f"/users/{quote_segment(mailbox)}/messages/"
      f"{quote_segment(reply_message_id)}/reply"
    )
    return {"method": "POST", "url": url, "jsonBody": {"message": message}}
  url = query_url(f"/users/{quote_segment(mailbox)}/sendMail")
  return {
    "method": "POST",
    "url": url,
    "jsonBody": {"message": message, "saveToSentItems": True},
  }


def approval_digest(request: Mapping[str, Any]) -> str:
  approved_content = {
    "method": request["method"],
    "url": request["url"],
    "jsonBody": request["jsonBody"],
  }
  if "accountAddress" in request:
    approved_content["accountAddress"] = request["accountAddress"]
  if "replyTarget" in request:
    approved_content["replyTarget"] = request["replyTarget"]
  canonical = json.dumps(
    approved_content, sort_keys=True, ensure_ascii=False, separators=(",", ":")
  )
  return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def send_message(client: GraphClient, request: Mapping[str, Any]) -> None:
  client.request(
    request["method"], request["url"], json_body=request["jsonBody"]
  )


def command_recent(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  records = client.list_messages("", args.count) if getattr(client, "is_gmail", False) else list_messages(client, mailbox_for(args, config), count=args.count)
  if getattr(client, "is_gmail", False) and client.skipped:
    print_json({"messages": records, "incomplete": True, "skipped": client.skipped}) if getattr(args, "json", False) else (print_summary(records), print("Incomplete; skipped: " + ", ".join(client.skipped)))
    return
  if getattr(args, "json", False):
    print_json(records)
  else:
    print_summary(records)


def command_search(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  records = client.list_messages("", args.count, search=args.query) if getattr(client, "is_gmail", False) else list_messages(client, mailbox_for(args, config), count=args.count, search=args.query)
  if getattr(client, "is_gmail", False) and client.skipped:
    print_json({"messages": records, "incomplete": True, "skipped": client.skipped}) if getattr(args, "json", False) else (print_summary(records), print("Incomplete; skipped: " + ", ".join(client.skipped)))
    return
  if getattr(args, "json", False):
    print_json(records)
  else:
    print_summary(records)


def command_from(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  records = client.list_messages("", args.count, sender=args.sender) if getattr(client, "is_gmail", False) else list_messages(client, mailbox_for(args, config), count=args.count, search=f"from:{args.sender}")
  if getattr(client, "is_gmail", False) and client.skipped:
    print_json({"messages": records, "incomplete": True, "skipped": client.skipped}) if getattr(args, "json", False) else (print_summary(records), print("Incomplete; skipped: " + ", ".join(client.skipped)))
    return
  if getattr(args, "json", False):
    print_json(records)
  else:
    print_summary(records)


def command_read(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  record = client.get_message("", args.message_id) if getattr(client, "is_gmail", False) else get_message(client, mailbox_for(args, config), args.message_id)
  truncate_body(
    record,
    getattr(args, "max_chars", DEFAULT_MAX_CHARS),
    getattr(args, "full", False),
  )
  if getattr(args, "json", False):
    print_json(record)
  else:
    print_message(record)


def command_thread(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  records = client.thread_messages("", args.conversation_id) if getattr(client, "is_gmail", False) else thread_messages(client, mailbox_for(args, config), args.conversation_id)
  for record in records:
    truncate_body(
      record,
      getattr(args, "max_chars", DEFAULT_MAX_CHARS),
      getattr(args, "full", False),
    )
  skipped = getattr(client, "skipped", [])
  if skipped and not getattr(args, "json", False):
    print("Incomplete; skipped: " + ", ".join(skipped))
  if getattr(args, "json", False):
    print_json({"messages": records, "incomplete": True, "skipped": skipped} if skipped else records)
    return
  if not records:
    print("No messages found.")
    return
  for index, record in enumerate(records):
    if index:
      print("\n" + "=" * 60 + "\n")
    print_message(record)


def command_attachments(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  if getattr(client, "is_gmail", False):
    records = client.attachments(args.message_id)
  else:
    mailbox = mailbox_for(args, config)
    url = query_url(
    f"/users/{quote_segment(mailbox)}/messages/"
    f"{quote_segment(args.message_id)}/attachments",
    {"$select": "id,name,size,contentType"},
  )
    values = client.get_collection(url)
    records = [
    {
      "id": str(value.get("id") or ""),
      "name": str(value.get("name") or ""),
      "size": int(value.get("size") or 0),
      "contentType": str(value.get("contentType") or ""),
    }
      for value in values
    ]
  if getattr(args, "json", False):
    print_json(records)
    return
  if not records:
    print("No attachments found.")
    return
  for index, record in enumerate(records):
    if index:
      print()
    print(f"ID: {display_text(record['id'])}")
    print(f"Name: {display_text(record['name'])}")
    print(f"Size: {record['size']}")
    print(f"Content type: {display_text(record['contentType'])}")


def markdown_message(record: Mapping[str, Any], max_chars: int) -> str:
  body = display_body(record.get("body") or "")
  lines = [
    f"# {display_text(record.get('subject', ''))}",
    "",
    f"- From: {display_address(record.get('from', {}))}",
    f"- To: {display_addresses(record.get('to', []))}",
    f"- Cc: {display_addresses(record.get('cc', []))}",
    f"- Date: {display_text(record.get('receivedDateTime') or record.get('sentDateTime', ''))}",
    f"- Read: {'yes' if record.get('isRead') else 'no'}",
    f"- Has attachments: {'yes' if record.get('hasAttachments') else 'no'}",
    f"- Conversation ID: {display_text(record.get('conversationId', ''))}",
    f"- Message ID: {display_text(record.get('id', ''))}",
    "",
    body,
    "",
  ]
  if record.get("bodyTruncated"):
    lines.extend(
      [
        f"Body truncated to {max_chars} characters; rerun with --full for the complete text.",
        "",
      ]
    )
  return "\n".join(lines)


def command_save(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  output_path = Path(args.output_path)
  if not output_path.parent.exists():
    raise CliError(f"output parent directory does not exist: {output_path.parent}")
  if os.path.lexists(output_path):
    raise CliError(f"output path already exists: {output_path}")
  record = client.get_message("", args.message_id) if getattr(client, "is_gmail", False) else get_message(client, mailbox_for(args, config), args.message_id)
  truncate_body(
    record,
    getattr(args, "max_chars", DEFAULT_MAX_CHARS),
    getattr(args, "full", False),
  )
  markdown_text = markdown_message(
    record, getattr(args, "max_chars", DEFAULT_MAX_CHARS)
  )
  try:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if os.name == "posix":
      flags |= os.O_NOFOLLOW
    file_descriptor = os.open(output_path, flags, 0o600)
    try:
      with os.fdopen(file_descriptor, "w", encoding="utf-8") as output_file:
        output_file.write(markdown_text)
    except BaseException:
      try:
        os.close(file_descriptor)
      except OSError:
        pass
      try:
        os.unlink(output_path)
      except OSError:
        pass
      raise
  except OSError as exc:
    raise CliError(f"cannot save message to {output_path}: {exc}") from exc
  result = {"saved": str(output_path), "messageId": record["id"]}
  if getattr(args, "json", False):
    print_json(result)
  else:
    print(f"Saved message {display_text(record['id'])} to {display_text(output_path)}")


def command_send(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  if getattr(client, "is_gmail", False):
    return command_gmail_write(args, client, config, "send")
  if getattr(args, "message_id", None):
    raise CliError("--message-id applies to the Gmail provider only")
  if args.in_reply_to:
    raise CliError(
      "--in-reply-to is unsupported; run thread CONVERSATION_ID, "
      "then pass --reply-to-message-id ID"
    )
  if args.approved:
    raise CliError(
      "--approved is obsolete; preview with --dry-run and pass --approve <digest>"
    )
  if not args.dry_run and not args.approve:
    raise CliError(
      "refusing to send without --approve <digest> from the exact dry run"
    )
  mailbox = sender_for(args, config)
  to_values = parse_recipients(args.to, "to", required=True)
  cc_values = parse_recipients(args.cc, "cc")
  bcc_values = parse_recipients(args.bcc, "bcc")
  body = body_from_args(args)
  request = build_send_request(
    mailbox,
    to_values=to_values,
    cc_values=cc_values,
    bcc_values=bcc_values,
    subject=args.subject,
    body=body,
    reply_message_id=args.reply_to_message_id,
  )
  digest = approval_digest(request)
  if args.dry_run:
    preview = {
      "dryRun": True,
      **request,
      "approvalDigest": digest,
    }
    if getattr(args, "json", False):
      print(json.dumps(preview, indent=2, ensure_ascii=True))
    else:
      print("Dry run only. No token or network request was made.")
      print(f"Method: {preview['method']}")
      print(f"URL: {preview['url']}")
      print("JSON body:")
      print(json.dumps(preview["jsonBody"], indent=2, ensure_ascii=True))
      print_preview_body(body)
      print(f"approval digest: {digest}")
    return
  if not hmac.compare_digest(args.approve.encode("utf-8"), digest.encode("ascii")):
    raise CliError(
      "missing or mismatched approval digest; rerun --dry-run and pass "
      "--approve <digest>"
    )
  baseline_ids = {
    str(record.get("id") or "")
    for record in sent_items(client, mailbox, 25)
    if record.get("id")
  }
  try:
    send_message(client, request)
  except CliError as exc:
    raise CliError(
      "send did not return confirmed success; delivery state may be "
      "uncertain, do not resend automatically, reconcile with verify-sent: "
      f"{exc}"
    ) from exc
  try:
    verified = verify_sent_record(
      client,
      mailbox,
      args.subject,
      to_values,
      cc_values,
      bcc_values,
      25,
      retries=3,
      excluded_ids=baseline_ids,
    )
  except CliError as exc:
    raise CliError(
      "message submission was accepted, but Sent Items verification failed; "
      "do not resend automatically, reconcile with verify-sent: "
      f"{exc}"
    ) from exc
  result = {"sent": True, "verifiedInSentItems": True, "message": verified}
  if getattr(args, "json", False):
    print_json(result)
  else:
    print("Sent Items verification succeeded.")
    print(f"Message ID: {display_text(verified.get('id', ''))}")
    print(f"Sent: {display_text(verified.get('sentDateTime', ''))}")
    print(f"To: {display_addresses(verified.get('to', []))}")
    print(f"Cc: {display_addresses(verified.get('cc', []))}")
    print(f"Bcc: {display_addresses(verified.get('bcc', []))}")


def command_gmail_write(args: argparse.Namespace, client: Any, config: dict[str, Any], kind: str) -> None:
  del config
  if getattr(args, "approved", False):
    raise CliError("--approved is obsolete; use --approve <digest>")
  if getattr(args, "in_reply_to", None):
    raise CliError("--in-reply-to is unsupported; use --reply-to-message-id")
  if not args.dry_run and not args.approve:
    raise CliError(f"refusing to {kind} without --approve <digest> from the exact dry run")
  if not args.dry_run and not args.message_id:
    raise CliError(f"{kind} requires --message-id from the exact dry run")
  sender = args.from_user or getattr(args, "user", None) or os.environ.get("GMAIL_DEFAULT_USER")
  if not sender:
    raise CliError("Gmail send and draft require --from or GMAIL_DEFAULT_USER")
  to_values = list(args.to or [])
  cc_values = list(args.cc or [])
  bcc_values = list(args.bcc or [])
  if not to_values:
    raise CliError("send requires at least one --to recipient")
  body = body_from_args(args)
  reply = client.reply_metadata(args.reply_to_message_id) if args.reply_to_message_id else None
  request = client.build_request(kind, sender, to_values, cc_values, bcc_values, args.subject, body, reply, args.message_id)
  request["accountAddress"] = client.account_address()
  from email import policy
  from email.parser import BytesParser
  from email.utils import getaddresses
  from gmail_backend import b64decode
  raw = request["jsonBody"]["message"]["raw"] if kind == "draft" else request["jsonBody"]["raw"]
  parsed = BytesParser(policy=policy.default).parsebytes(b64decode(raw))
  preview = {name: str(parsed.get(name, "")) for name in ("From", "To", "Cc", "Bcc", "Subject", "In-Reply-To", "References", "Message-ID")}
  preview["accountAddress"] = request["accountAddress"]
  preview["threadId"] = (reply or {}).get("thread_id", "")
  body_part = parsed.get_body(preferencelist=("plain",))
  preview["body"] = body_part.get_content().replace("\r\n", "\n").replace("\r", "\n").rstrip("\n") if body_part else ""
  digest = approval_digest(request)
  if args.dry_run:
    output = {"dryRun": True, **request, "preview": preview, "approvalDigest": digest}
    if getattr(args, "json", False):
      print_json(output)
    else:
      print("Dry run only. No write was made.")
      print(f"Method: {request['method']}")
      print(f"URL: {request['url']}")
      print("JSON body:")
      print(json.dumps(request["jsonBody"], indent=2, ensure_ascii=True))
      recipients = "; ".join(name.lower() + "=" + ", ".join(address for _, address in getaddresses(parsed.get_all(name, []))) for name in ("To", "Cc", "Bcc"))
      print("Recipients: " + display_text(recipients))
      for name, value in preview.items():
        if name == "body":
          print_preview_body(value)
        else:
          print(f"{name}: {display_text(value)}")
      print(f"approval digest: {digest}")
    return
  if not hmac.compare_digest(args.approve.encode("utf-8"), digest.encode("ascii")):
    raise CliError("missing or mismatched approval digest; rerun --dry-run and pass --approve <digest>; Message-ID: " + display_text(request["messageId"]))
  client.approved_account = request["accountAddress"]
  client.refuse_replay(request["messageId"])
  wire = {key: request[key] for key in ("method", "url", "jsonBody")}
  try:
    result = client.request("POST", wire["url"], json_body=wire["jsonBody"])
  except CliError as exc:
    raise CliError(f"{exc}; Message-ID: {display_text(request['messageId'])}") from exc
  except (Exception, KeyboardInterrupt) as exc:
    hint = " and check the Drafts folder" if kind == "draft" else ""
    message = "outcome unknown; run verify-sent with the Message-ID " + display_text(request["messageId"]) + hint
    if isinstance(exc, KeyboardInterrupt):
      print("error: " + message, file=sys.stderr)
      raise SystemExit(130) from exc
    raise CliError(message) from exc
  if kind == "draft":
    print_json({"draftCreated": True, "result": result, "messageId": request["messageId"]}) if getattr(args, "json", False) else print("Draft created. Message-ID: " + display_text(request["messageId"]))
    return
  msgid = request["messageId"]
  try:
    server_id = result.get("id") if isinstance(result, dict) else None
    if not isinstance(server_id, str) or not server_id:
      raise CliError("send response did not contain a message ID")
    verified = client.get_message("", server_id, include_body=False)
    if "SENT" not in verified.get("labelIds", []) or verified.get("rfc822msgid") != msgid:
      raise CliError("returned message lacks SENT label or matching RFC Message-ID")
  except Exception as exc:
    raise CliError("send outcome unknown; run verify-sent with the Message-ID " + display_text(msgid) + ": " + str(exc)) from exc
  print_json({"sent": True, "verifiedInSentItems": True, "messageId": msgid, "message": verified}) if getattr(args, "json", False) else print("SENT verification succeeded. Message-ID: " + display_text(msgid))


def command_draft(args: argparse.Namespace, client: Any, config: dict[str, Any]) -> None:
  if not getattr(client, "is_gmail", False):
    if getattr(args, "message_id", None):
      raise CliError("--message-id applies to the Gmail provider only")
    raise CliError("Gmail draft is available only with --provider gmail")
  command_gmail_write(args, client, config, "draft")


def command_auth(args: argparse.Namespace, client: Any, config: dict[str, Any]) -> None:
  del config
  if not getattr(client, "is_gmail", False):
    raise CliError("auth is available only with --provider gmail")
  if args.auth_command == "status":
    print_json(client.auth_status())
  elif args.auth_command == "login":
    client.auth_login()
    print("Gmail authorization stored locally.")
  else:
    remote_error = client.auth_revoke()
    if remote_error == "no local Gmail token":
      print(remote_error)
      return
    print("Gmail local token deleted.")
    if remote_error:
      raise CliError("remote revocation failed: " + display_text(remote_error))


def command_verify_sent(
  args: argparse.Namespace, client: GraphClient, config: dict[str, Any]
) -> Any:
  if getattr(client, "is_gmail", False):
    records = client.sent_items(args.count)
    requested_id = getattr(args, "rfc822msgid", None)
    if not requested_id:
      raise CliError("Gmail verify-sent requires --rfc822msgid")
    from email.utils import getaddresses
    to_values, cc_values, bcc_values = (None if values is None else [address.casefold() for _, address in getaddresses(values) if address] for values in (args.to, args.cc, args.bcc))
    found = match_sent_record([item for item in records if "SENT" in item.get("labelIds", []) and item.get("rfc822msgid") == requested_id], args.subject, to_values, cc_values, bcc_values, None)
    if not found:
      raise CliError("matching Message-ID was not found in SENT; outcome remains unknown")
    print_json(found) if getattr(args, "json", False) else print_message(found)
    return
  to_values = parse_recipients(args.to, "to") if args.to is not None else None
  cc_values = parse_recipients(args.cc, "cc") if args.cc is not None else None
  bcc_values = parse_recipients(args.bcc, "bcc") if args.bcc is not None else None
  record = verify_sent_record(
    client,
    mailbox_for(args, config),
    args.subject,
    to_values,
    cc_values,
    bcc_values,
    args.count,
    retries=1,
  )
  if getattr(args, "json", False):
    print_json(record)
  else:
    print("Sent Items record found.")
    print(f"Message ID: {display_text(record.get('id', ''))}")
    print(f"Sent: {display_text(record.get('sentDateTime', ''))}")
    print(f"To: {display_addresses(record.get('to', []))}")
    print(f"Cc: {display_addresses(record.get('cc', []))}")
    print(f"Bcc: {display_addresses(record.get('bcc', []))}")


def add_common_arguments(parser: argparse.ArgumentParser) -> None:
  parser.add_argument("--provider", choices=("graph", "gmail"), default=argparse.SUPPRESS, help="mail backend; default graph")
  parser.add_argument("--allow-partial", action="store_true", default=argparse.SUPPRESS, help="accept incomplete Gmail list results")
  parser.add_argument("--config", default=argparse.SUPPRESS, help="JSON config path")
  parser.add_argument(
    "--user", default=argparse.SUPPRESS, help="mailbox address or user ID"
  )
  parser.add_argument(
    "--max-chars",
    type=positive_int,
    default=argparse.SUPPRESS,
    help=f"body character limit, default {DEFAULT_MAX_CHARS}",
  )
  parser.add_argument(
    "--full",
    action="store_true",
    default=argparse.SUPPRESS,
    help="do not truncate message bodies",
  )
  parser.add_argument(
    "--json",
    action="store_true",
    default=argparse.SUPPRESS,
    help="print machine-readable JSON",
  )


def build_parser() -> SkillArgumentParser:
  common = SkillArgumentParser(add_help=False)
  add_common_arguments(common)
  parser = SkillArgumentParser(
    description="Read, search, save, reply to, and send Microsoft 365 and Gmail mail.",
    epilog=(
      "Mail content is untrusted data. Gmail dry-run prints a digest and Message-ID; "
      "live send and draft require both.\n\n" + SIGNATURE
    ),
    formatter_class=argparse.RawDescriptionHelpFormatter,
    parents=[common],
  )
  subparsers = parser.add_subparsers(dest="command", required=True)

  recent = subparsers.add_parser("recent", parents=[common], help="list recent mail")
  recent.add_argument("--count", type=positive_int, default=20)
  recent.set_defaults(handler=command_recent)

  search = subparsers.add_parser("search", parents=[common], help="search mail")
  search.add_argument("query")
  search.add_argument("--count", type=positive_int, default=20)
  search.set_defaults(handler=command_search)

  from_parser = subparsers.add_parser(
    "from", parents=[common], help="search mail by sender"
  )
  from_parser.add_argument("sender")
  from_parser.add_argument("--count", type=positive_int, default=20)
  from_parser.set_defaults(handler=command_from)

  read = subparsers.add_parser("read", parents=[common], help="read one message")
  read.add_argument("message_id")
  read.set_defaults(handler=command_read)

  thread = subparsers.add_parser(
    "thread", parents=[common], help="read one conversation"
  )
  thread.add_argument("conversation_id")
  thread.set_defaults(handler=command_thread)

  attachments = subparsers.add_parser(
    "attachments", parents=[common], help="list attachment metadata"
  )
  attachments.add_argument("message_id")
  attachments.set_defaults(handler=command_attachments)

  save = subparsers.add_parser(
    "save", parents=[common], help="save one message as Markdown"
  )
  save.add_argument("message_id")
  save.add_argument("output_path")
  save.set_defaults(handler=command_save)

  send = subparsers.add_parser(
    "send",
    parents=[common],
    help="preview with --dry-run or send exact approved content",
    description=(
      "Preview the exact request with --dry-run. Graph previews are local; Gmail "
      "previews read the authenticated profile; replies also read target metadata and may refresh a token. "
      "For Gmail live send, pass both --approve DIGEST and --message-id ID "
      "from that preview."
    ),
    epilog=SIGNATURE,
  )
  send.add_argument("--from", dest="from_user")
  send.add_argument("--to", action="append")
  send.add_argument("--cc", action="append")
  send.add_argument("--bcc", action="append")
  send.add_argument("--subject", required=True)
  send_body = send.add_mutually_exclusive_group()
  send_body.add_argument("--body")
  send_body.add_argument("--body-file")
  send.add_argument(
    "--dry-run",
    action="store_true",
    help="preview the exact request; Gmail replies fetch target metadata",
  )
  send.add_argument(
    "--approve",
    metavar="DIGEST",
    help="SHA-256 digest printed by the matching --dry-run",
  )
  send.add_argument("--message-id", help="Gmail Message-ID from the exact dry-run preview")
  send.add_argument(
    "--approved",
    action="store_true",
    help=argparse.SUPPRESS,
  )
  replies = send.add_mutually_exclusive_group()
  replies.add_argument("--in-reply-to", metavar="CONVERSATION_ID")
  replies.add_argument("--reply-to-message-id", metavar="MESSAGE_ID")
  send.set_defaults(handler=command_send)

  draft = subparsers.add_parser("draft", parents=[common], help="create an approved Gmail draft", epilog=SIGNATURE)
  draft.add_argument("--from", dest="from_user")
  draft.add_argument("--to", action="append")
  draft.add_argument("--cc", action="append")
  draft.add_argument("--bcc", action="append")
  draft.add_argument("--subject", required=True)
  draft_body = draft.add_mutually_exclusive_group()
  draft_body.add_argument("--body")
  draft_body.add_argument("--body-file")
  draft.add_argument("--dry-run", action="store_true", help="preview the exact request; replies fetch target metadata")
  draft.add_argument("--approve", metavar="DIGEST")
  draft.add_argument("--message-id", help="Gmail Message-ID from the exact dry-run preview")
  draft.add_argument("--reply-to-message-id", metavar="MESSAGE_ID")
  draft.set_defaults(handler=command_draft, approved=False, in_reply_to=None)

  auth = subparsers.add_parser("auth", parents=[common], help="manage Gmail authorization", epilog=SIGNATURE)
  auth.add_argument("auth_command", choices=("login", "status", "revoke"))
  auth.set_defaults(handler=command_auth)

  verify = subparsers.add_parser(
    "verify-sent", parents=[common], help="read back a Sent Items record"
  )
  verify.add_argument("--subject", required=True)
  verify.add_argument("--to", action="append")
  verify.add_argument("--cc", action="append")
  verify.add_argument("--bcc", action="append")
  verify.add_argument("--count", type=positive_int, default=25)
  verify.add_argument("--rfc822msgid")
  verify.set_defaults(handler=command_verify_sent)

  return parser


def main(argv: Sequence[str] | None = None) -> int:
  try:
    parser = build_parser()
    args = parser.parse_args(argv)
    config = load_config(getattr(args, "config", None))
    provider = getattr(args, "provider", os.environ.get("MAIL_PROVIDER", "graph"))
    if provider == "gmail":
      import gmail_backend
      client = gmail_backend.GmailBackend(sys.modules[__name__], config)
    elif provider == "graph":
      client = GraphClient(config)
    else:
      raise CliError("MAIL_PROVIDER must be graph or gmail")
    args.handler(args, client, config)
    return 2 if getattr(client, "skipped", []) and not getattr(args, "allow_partial", False) else 0
  except CliError as exc:
    print(f"error: {display_text(exc)}", file=sys.stderr)
    return 1
  except KeyboardInterrupt:
    print("error: interrupted", file=sys.stderr)
    return 1
  except Exception as exc:
    print(f"error: {display_text(exc)}", file=sys.stderr)
    return 1


if __name__ == "__main__":
  raise SystemExit(main())

License: MIT (read it).

Back to the listing