Postman Email
CommunityRead, search, send and manage email over IMAP/SMTP: inbox, replies, forwarding, flags, moving, deleting and attachments. Works with QQ, NetEase, Gmail, Outlook, Yahoo and custom servers.
PYLinTech v1.0.0
Install command
$ cow skill install postman Run in a terminal or send to CowAgent to install
Productivity
通过 IMAP/SMTP 协议读取、搜索、发送及管理电子邮件。适用于用户需要查看收件箱、查找或阅读特定邮件、撰写或发送邮件、回复或转发邮件、更改邮件的已读或星标状态、移动或删除邮件，以及保存附件等场景。支持 QQ 、网易、Gmail、Outlook、Yahoo、搜狐、新浪、阿里云以及任何自定义 IMAP/SMTP 服务器，并能正确读取和发送 HTML 邮件。仅提供标准库，无依赖项。
# Python
__pycache__/
*.py[cod]
*$py.class
# 虚拟环境（本项目零依赖，但本地调试时容易生成）
.venv/
venv/
# 编辑器与操作系统
.DS_Store
Thumbs.db
.idea/
.vscode/
# 凭据——本仓库没有 CI，一旦提交上去就是公开事故
#
# 工具本身**没有**默认配置文件路径：mail.json 只是文档里的建议名，
# 真实文件名由使用者决定。因此这里覆盖的是常见命名，不是某一个固定名。
# 推荐做法是把配置放在仓库之外（`--config ~/mail.json`），但仍在此拦截。
mail.json
mail*.json
*-mail.json
*.local.json
# 环境变量
.env
.env.*
!.env.example
# download 子命令的 --dir 默认是当前目录；用仓库根目录跑容易把附件落在树里
/downloads/
MIT License
Copyright (c) 2026 PYLinTech/Postman-Skill contributors
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
# PYLinTech/Postman-Skill
命令行 IMAP/SMTP 邮件客户端，**只用 Python 标准库**——没有 `pip install`，没有虚拟环境，没有构建步骤，Python 3.8+ 直接能跑。
它同时是两种东西：用 `SKILL.md` 注册进 agent 之后是一个邮件 Skill，脱离 agent 也能当普通命令行工具用。
- skill 标识 `postman`（`SKILL.md` 的 `name`），入口 `scripts/mail.py`，12 个子命令
- 仓库地址与项目全名以 **`PYLinTech/Postman-Skill`** 为准，本文档一律使用全名
> **为什么仓库叫 `PYLinTech/Postman-Skill` 而 skill 叫 `postman`**：仓库名要带
> 归属前缀以在 GitHub 上不撞名，而 skill 标识必须匹配 `^[a-z0-9-]+$`——斜杠不合法。
> 两者指向同一个东西，文档里出现的 `PYLinTech/Postman-Skill`、`postman` 和
> `scripts/mail.py` 分别对应「项目」「skill 标识」「命令入口」三个不同层面。
## 它做什么
- **收信搜索**：原生 IMAP 检索条件，或 Gmail 的 `X-GM-RAW` 语法；按未读、星标、带附件、日期区间过滤，`--limit` / `--offset` 分页。全程用 `BODY.PEEK`，**搜索不会把邮件标成已读**。
- **读信**：RFC 2047 标题解码（中文主题直接可读，不再是 `=?utf-8?B?...?=`）；没有 `text/plain` 部件时自动把 HTML 转成可读文本（这是很多同类客户端直接返回空的地方）；附件列出文件名与大小。正文默认截断到 20000 字符，截断时显式告知。
- **发信**：纯文本 / HTML / 两者并存（`multipart/alternative`，收件端表现最稳）；支持附件；Bcc 不出现在可见头里；`--in-reply-to` 串联会话；`--dry-run` 只组装不投递。
- **整理**：存草稿、改标记、移动文件夹、删除（有 Trash 就移进去，没有才彻底删）、下载附件。
- **多账号**：一个配置文件放多个账户，`--account <名称>` 切换。
支持 QQ、网易（163 / 126 / yeah）、Gmail、Outlook、Yahoo、搜狐、新浪、阿里邮箱，以及任何自定义 IMAP/SMTP 服务——服务器地址从邮箱域名自动推导，不需要手工填。
## 目录
```
PYLinTech/Postman-Skill/
├── SKILL.md # Agent Skill 定义：描述何时该用它
├── README.md
├── LICENSE
├── resources/
│ └── config.example.json # 凭据配置模板
└── scripts/
├── mail.py # CLI 入口，12 个子命令
├── selftest.py # 自检：74 项离线检查 + 可选联网探测
└── mailkit/ # 实现，扁平子模块，无顶层再导出
├── __init__.py # 仅一行 docstring，存在理由见下
├── config.py # 凭据解析、服务商推导、传输方式选择
├── mime.py # MIME 组装与解析、RFC 2047、HTML 转文本
├── imap_client.py # IMAP 会话、文件夹特殊用途、UID 操作
├── smtp_client.py # SMTP 投递
└── commands.py # 子命令实现
```
> `mailkit/__init__.py` 只有一行 docstring，不是残留：它让这个包成为**常规包**。
> 没有它就会退化成命名空间包（PEP 420），而命名空间包在 `sys.path` 上遇到同名
> 常规包时会被后者覆盖——即使本目录排在前面。一个碰巧叫 `mailkit` 的第三方包
> 就足以让整个 skill 起不来。
## 安装
把仓库拷进你的 agent skills 目录，没有其它步骤。
```bash
git clone [email protected]:PYLinTech/Postman-Skill.git
```
## 配置凭据
**几乎所有服务商都不接受账号密码，必须用「应用专用密码」**，且要先在账号设置里开启 IMAP/SMTP。QQ 和网易尤其如此——直接填普通密码一定失败。
| 来源 | 方式 |
|---|---|
| 配置文件 | `--config PATH`，或 `MAIL_CONFIG`；模板见 `resources/config.example.json` |
| 环境变量 | `MAIL_ADDRESS`、`MAIL_PASSWORD`，可选 `MAIL_PROVIDER`、`MAIL_SENDER_NAME` |
解析优先级：命令行参数 > 配置文件 > 环境变量。**工具没有默认配置文件路径**——文件名由你决定。
多账户用 `accounts` 映射，配 `--account` 切换：
```json
{
"accounts": {
"work": { "address": "[email protected]", "password": "...",
"imap_host": "imap.company.com", "smtp_host": "smtp.company.com" },
"personal": { "address": "[email protected]", "password": "..." }
}
}
```
```bash
cp resources/config.example.json ~/mail.json
chmod 600 ~/mail.json # 里面是明文密码
python scripts/mail.py --config ~/mail.json --account work check
```
`--account` 指定了不存在的账户会**直接报错并列出可用名称**，不会静默回落到别的账户——那会导致连错邮箱。
## 用法
```bash
M=./scripts/mail.py
python $M check # 验证凭据、列文件夹、报未读数
python $M selftest # 验证这份代码本身，不需要网络
python $M selftest --live # 追加只读 IMAP 探测
python $M search --unread --limit 10
python $M read --uid 12345
python $M send --to [email protected] --subject 'Hi' --text 'Hello'
python $M send --to [email protected] --subject 'Q3' --html-file report.html --attach q3.pdf
python $M download --uid 12345 --dir ~/Downloads
```
长 HTML 用 `json` 子命令传，避免 shell 引号地狱；任何子命令都可加 `--json` 取结构化输出，也都有 `--help` 可以查完整参数：
```bash
python $M json '{"command":"send","to":["[email protected]"],"html":"<p>hi</p>"}'
```
### 命令表
| 命令 | 说明 |
|---|---|
| `check` | 验证凭据、列文件夹、报未读数 |
| `folders` | 列出文件夹及其特殊用途标记 |
| `search` | 检索邮件，**不改动已读状态** |
| `read` | 读完整正文、HTML、附件列表 |
| `send` | 经 SMTP 投递 |
| `draft` | 存入草稿箱 |
| `flag` | 改标记 |
| `move` | 移动邮件 |
| `delete` | 删除（有 Trash 就先进 Trash） |
| `download` | 保存附件 |
| `selftest` | 验证本构建，`--live` 追加只读探测 |
| `json` | 用一个 JSON 对象执行命令 |
全局选项 `--config` / `--account` / `--json` 对所有子命令生效。常用参数：
| 子命令 | 参数 |
|---|---|
| `search` | `-f/--folder`、`-q/--query`（默认 `ALL`）、`--gmail-raw`、`--since` / `--before`（YYYY-MM-DD）、`--unread`、`--flagged`、`--has-attachment`、`-n/--limit`（默认 20）、`--offset` |
| `read` | `--uid`、`--latest N`、`-f/--folder`、`--mark-read`、`--html`、`--max-chars N`（默认 20000，`0` 关闭截断） |
| `send` / `draft` | `--to`（可重复或逗号分隔）、`--cc`、`--bcc`、`--subject`、`--text` / `--text-file`、`--html` / `--html-file`、`--attach`（可重复）、`--reply-to`、`--in-reply-to`、`--reference`；`send` 另有 `--dry-run`，`draft` 的 `-f/--folder` 默认 `Drafts` |
| `flag` | `--uid`（可重复）、`-f/--folder`、`--flags`（可重复）、`--mode {add,remove}` |
| `move` | `--uid`（可重复）、`-f/--folder`（源）、`--to-folder` |
| `delete` | `--uid`（可重复）、`-f/--folder`、`--permanent` |
| `download` | `--uid`、`-f/--folder`、`--dir`（默认当前目录）、`--name`（可重复） |
文件夹接受别名（`inbox`、`sent`、`drafts`、`trash`、`junk`、`archive`）和本地化名称（如 `已发送`）。`--dir` 默认是当前目录——在仓库根目录跑 `download` 会把附件落在工作树里，建议显式给一个仓库外的目录。
## 几个设计取舍
- **全程用 UID，不用序号**。别的客户端动一下邮箱，序号就漂了；UID 不会。`search` 返回 UID，后续所有命令都接 UID。
- **搜索用 `BODY.PEEK`**。`search` 的本意是「找出有哪些」，顺手改已读状态是副作用而非结果。
- **已读状态默认不动**。列信和读信都不改标记，除非显式 `--mark-read`。
- **文件夹名按账户自身的特殊用途标记解析**。`inbox` / `sent` / `drafts` / `trash` / `junk` / `archive` 落到对应的 `\Sent`、`\Drafts` 上，所以德语或中文界面的邮箱不用配置也能用。
- **原始 HTML 默认不返回**。里面的 CSS、内联 base64 图片对读取内容是纯噪声，而抽取出的文本已经覆盖了信息；要原始部分加 `--html`。
- **正文体积设上限**。不设限时，一封营销邮件的 JSON 输出能放大到正文体积的两倍多，直接灌爆上下文。
- **密码永不回显**，`check` 里显示为 `***`。
- **零第三方依赖**。整个仓库只用标准库，代价是要自己处理 MIME 编码、HTML 转文本和服务商差异，收益是拷进去就能跑，不会因为依赖冲突和宿主的其它工具链互相污染。
## 验证一份构建
```bash
python scripts/selftest.py # 74 项离线检查，不需要网络和凭据
python scripts/selftest.py --live # 追加只读 IMAP 探测（connect / SELECT / LIST / SEARCH）
```
离线部分覆盖：RFC 2047 标题解码、HTML 转文本、MIME 组装与往返、服务器解析、标记语义、正文截断、多账户选择、文件名安全与重名避让。
命令行为异常时先跑它——能把「这份代码坏了」和「服务器拒绝了我们」分开。
## 已知边界
- 仅 IMAP，不支持 POP3。
- 发信只做 `multipart/alternative` 与 `text/*`，不支持日历邀请，也不支持把邮件作为 `message/rfc822` 附件再发。
- `move` 用 COPY + EXPUNGE，**EXPUNGE 不是 UID 范围的**：该文件夹里其它已标记删除的邮件会一起被清掉。这是 IMAP 的固有行为，不是 bug。
- 权限模型不是沙箱。运行它的进程能碰到的东西，它就能碰到。
- 正文截断默认 20000 字符；需要全文请显式 `--max-chars 0`。
- 声明支持 Python 3.8+，但开发与验证均在 3.9 及以上进行，3.8 未实机跑过。
## 协议
MIT，见 [LICENSE](LICENSE)。
---
name: postman
description: Read, search, send and manage email over IMAP/SMTP. Use when the user wants to check their inbox, find or read specific messages, draft or send mail, reply or forward, change read or starred state, move or delete messages, or save attachments. Supports QQ, NetEase, Gmail, Outlook, Yahoo, Sohu, Sina, Aliyun and any custom IMAP/SMTP server, and reads and sends HTML mail correctly. Standard library only, no dependencies.
license: MIT
allowed-tools:
- Bash
- Read
- Write
metadata:
version: "1.0.0"
python: ">=3.8"
---
# Email
IMAP/SMTP from the command line. Standard library only — nothing to install.
## Setup
Credentials are required. Run `check` first; if it reports a configuration
error, ask the user for an address and an **app-specific password** (most
providers reject the normal account password for IMAP/SMTP).
| Source | How |
| --- | --- |
| File | `--config mail.json`; see `resources/config.example.json`. Multi-account files use an `accounts` map plus `--account name` |
| Env | `MAIL_ADDRESS`, `MAIL_PASSWORD`, optionally `MAIL_PROVIDER`, `MAIL_SENDER_NAME`, `MAIL_CONFIG` |
If neither is set, ask the user — never guess. The password is never printed
back in any output.
## Running
Set the path once, then use `$MAIL`:
```bash
MAIL=/absolute/path/to/this/skill/scripts/mail.py
python "$MAIL" check
```
Do not `cd` into the skill directory; it may be reset between runs. Output is
human-readable; add `--json` for structured output, or pass long HTML via
`python "$MAIL" json '{"command":"send","to":["[email protected]"],"html":"<p>hi</p>"}'`
when shell quoting gets awkward. Every command takes `--help`. Allow a bash
timeout of at least 120 seconds (TLS handshakes are slow on some providers) and
run these synchronously.
## Commands
| Command | Purpose |
| --- | --- |
| `check` | Verify credentials, list folders, unread count |
| `selftest` | Verify this build; `--live` also probes the mailbox |
| `folders` | List folders with special-use flags |
| `search` | Find messages — never changes the read flag |
| `read` | Full bodies, HTML, attachment listings |
| `send` / `draft` | Send over SMTP, or save to Drafts |
| `flag` | read/unread, star, and other flags |
| `move` / `delete` | Reorganise, with Trash fallback |
| `download` | Save attachments |
### Search and read
```bash
python "$MAIL" search --unread --limit 10
python "$MAIL" search --query 'FROM "[email protected]" SINCE 2026-09-01'
python "$MAIL" search --gmail-raw 'has:attachment larger:5M' # Gmail only
python "$MAIL" read --uid 12345
python "$MAIL" read --latest 3
python "$MAIL" download --uid 12345 --dir ~/Downloads
```
`--query` takes raw IMAP criteria; `--gmail-raw` takes Gmail's native syntax.
Common options: `-f/--folder` applies to every folder-scoped command
(`check`, `search`, `read`, `flag`, `move`, `delete`, `download`), defaulting to
`INBOX`; `draft` uses the same flag but defaults to `Drafts`. `search` also
takes `--unread`, `--flagged`, `--has-attachment`, `--since` / `--before`
(YYYY-MM-DD), `--limit` and `--offset` for paging; `download` takes `--name` to
save one attachment by name. Folders accept aliases (`inbox`, `sent`, `drafts`,
`trash`, `junk`, `archive`) and localised names such as `已发送`.
Search returns **UIDs**, stable even when other clients move mail — use them for
every follow-up. `read` reports `text_source`: `plain`, `html` (converted to
readable text), or `empty`. Reading never alters state unless `--mark-read` is
given. Bodies are capped at 20 000 characters (`--max-chars`, `0` disables) and
the output says when it truncated; pass `--html` only if you need the raw part,
since the extracted text already covers its content.
### Send
```bash
python "$MAIL" send --to [email protected] --subject 'Hi' --text 'Hello'
python "$MAIL" send --to [email protected] --cc [email protected] --subject 'Q3' \
--html-file report.html --attach ./q3.pdf
python "$MAIL" send --to [email protected] --subject 'Re: x' --text 'y' \
--in-reply-to '<[email protected]>'
```
Both `--text` and `--html` produce `multipart/alternative` — what recipients'
clients expect. Prefer `--html-file` / `--text-file` for long bodies. `--bcc`
recipients stay out of the visible To and Cc headers.
**Always show the user the resolved recipients, subject and body and get an
explicit go-ahead before sending.** It cannot be undone. Use `--dry-run` to
validate and preview without delivering.
### Organise
```bash
python "$MAIL" draft --to [email protected] --subject 'Idea' --text '...'
python "$MAIL" flag --uid 12345 --flags unread # clears the Seen flag
python "$MAIL" flag --uid 12345 --flags star --mode remove
python "$MAIL" move --uid 12345 --to-folder archive
python "$MAIL" delete --uid 12345 # Trash when present
```
Flag names: `read`, `unread`, `star`, `answered`, `deleted`, `draft`. `read`
and `unread` name a desired state and invert the Seen flag; `--mode remove`
flips them. `--permanent` on `delete` is irreversible — confirm first and prefer
the Trash default.
## Providers
Servers derive from the address domain; `provider` in the config file or
`MAIL_PROVIDER` overrides that.
| Domain | IMAP | SMTP |
| --- | --- | --- |
| `qq.com`, `foxmail.com` | `imap.qq.com:993` | `smtp.qq.com:587` |
| `163.com` / `126.com` / `yeah.net` | each its own host | each its own host |
| `gmail.com` | `imap.gmail.com:993` | `smtp.gmail.com:587` |
| `outlook.com`, `hotmail.com`, `live.*` | `outlook.office365.com:993` | `smtp.office365.com:587` |
| `yahoo.*`, `ymail.com` | `imap.mail.yahoo.com:993` | `smtp.mail.yahoo.com:587` |
| `sohu.com` | `imap.sohu.com:993` | `smtp.sohu.com:465` |
| `sina.*` | `imap.sina.com:993` | `smtp.sina.com:465` |
| `aliyun.com` | `imap.aliyun.com:993` | `smtp.aliyun.com:465` |
Anything else: set `imap_host` / `imap_port` / `smtp_host` / `smtp_port`.
Transport follows the port — 465 implicit TLS, 587 STARTTLS, otherwise plain —
and `smtp_security` pins it explicitly. NetEase and QQ require an RFC 2971
client ID before login; handled automatically.
Every provider needs an **app-specific password** with IMAP/SMTP enabled. A
login failure almost always means one of those is missing.
## Troubleshooting
| Message | Cause |
| --- | --- |
| `no mailbox address configured` | Credentials absent — ask the user |
| `IMAP login failed` / `SMTP authentication failed` | Wrong password, or an app password is required |
| `folder 'x' not found` | Run `folders`; localised names resolve automatically |
| TLS certificate verification failed | Corporate interception; `verify_ssl: false` only if the user accepts the risk |
## Notes
- `python "$MAIL" selftest` runs 74 offline checks — header decoding, HTML
conversion, MIME assembly, server resolution, flag planning, filename safety —
with no network or credentials. `--live` adds read-only IMAP probes against
the configured account. Run `selftest` first when something behaves oddly: it
separates "this build is broken" from "the server refused us".
- The password is never echoed; `check` reports `***`. Prefer `MAIL_PASSWORD`
over a command argument, which can land in shell history.
- `search` and `read` leave read state alone unless `--mark-read` is passed.
- `move` uses COPY + EXPUNGE, which is not UID-scoped: anything else already
flagged deleted in that folder goes too. Standard IMAP behaviour.
{
"_comment": [
"Copy to a location outside the skill directory, fill in the values, and",
"protect it: chmod 600 mail.json",
"Pass it with: mail.py --config /path/to/mail.json <command>",
"or set MAIL_CONFIG=/path/to/mail.json in the environment."
],
"address": "[email protected]",
"password": "your-app-specific-password",
"sender_name": "Your Name",
"_optional_fields": {
"provider": "Leave as auto to derive the server from the address domain. Set explicitly (qq, 163, 126, yeah, gmail, outlook, hotmail, live, yahoo, foxmail, sohu, sina, aliyun, tom) only when the address domain is unusual.",
"imap_host": "Override only for a custom or corporate server.",
"imap_port": 993,
"smtp_host": "Override only for a custom or corporate server.",
"smtp_port": 587,
"smtp_security": "Leave empty to derive from the port: 465 = ssl, 587 = starttls, anything else = plain. Set to ssl / starttls / none to pin it.",
"timeout": 30,
"verify_ssl": true,
"send_as": "Set when the From address differs from the login address.",
"client_name": "Name sent in the RFC 2971 IMAP ID handshake. Optional; defaults to python-mailkit. Some providers (NetEase, QQ) reject login without an ID."
},
"accounts": {
"work": {
"address": "[email protected]",
"password": "work-app-password",
"imap_host": "imap.company.com",
"imap_port": 993,
"smtp_host": "smtp.company.com",
"smtp_port": 587
},
"personal": {
"address": "[email protected]",
"password": "personal-app-password"
}
},
"_how_to_select_an_account": "mail.py --config mail.json --account work check"
}
#!/usr/bin/env python3
"""email — a general-purpose IMAP/SMTP command line client.
Standard library only. No third-party dependency.
mail.py check
mail.py search --unread --limit 10
mail.py read --uid 12345
mail.py send --to [email protected] --subject "Hi" --html '<p>Hello</p>'
mail.py json '{"command": "search", "query": "UNSEEN"}'
``json`` is a generic escape hatch for callers that prefer to pass a single
object (shell quoting is awkward otherwise); it is not required.
Output is human-readable by default; pass ``--json`` for structured output.
Run ``mail.py <command> --help`` for per-command options.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from typing import Any, Dict, List, Optional
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mailkit import commands # noqa: E402
from mailkit.config import ConfigError # noqa: E402
from mailkit.mime import MailError # noqa: E402
# Cap on the characters a single message body may contribute. Newsletters and
# long threads otherwise flood whatever agent is reading the result.
DEFAULT_MAX_BODY_CHARS = 20000
EXIT_OK = 0
EXIT_USAGE = 2
EXIT_CONFIG = 3
EXIT_MAIL = 4
EXIT_INTERNAL = 5
# --------------------------------------------------------------------------
# Argument plumbing
# --------------------------------------------------------------------------
def _split_list(values: Optional[List[str]]) -> List[str]:
"""Flatten repeatable flags and comma/semicolon separated strings."""
out: List[str] = []
for value in values or []:
for part in str(value).replace(";", ",").split(","):
part = part.strip()
if part:
out.append(part)
return out
def _read_text(inline: Optional[str], path: Optional[str], what: str) -> str:
if inline and path:
raise MailError(f"give either --{what} or --{what}-file, not both")
if path:
try:
with open(os.path.expanduser(path), "r", encoding="utf-8") as handle:
return handle.read()
except OSError as exc:
raise MailError(f"cannot read --{what}-file {path}: {exc}") from exc
return inline or ""
def _base_args(namespace: argparse.Namespace) -> Dict[str, Any]:
return {
"config_path": namespace.config,
"account": namespace.account,
}
# --------------------------------------------------------------------------
# Per-command builders
# --------------------------------------------------------------------------
def _cmd_check(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
if ns.folder:
args["folder"] = ns.folder
return commands.cmd_check(args)
def _cmd_folders(ns: argparse.Namespace) -> Dict[str, Any]:
return commands.cmd_folders(_base_args(ns))
def _cmd_search(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
args.update({
"folder": ns.folder,
"query": ns.query,
"since": ns.since,
"before": ns.before,
"unread_only": ns.unread,
"flagged_only": ns.flagged,
"has_attachment": ns.has_attachment,
"gmail_raw": ns.gmail_raw,
"limit": ns.limit,
"offset": ns.offset,
})
return commands.cmd_search(args)
def _cmd_read(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
args.update({
"uid": _split_list([ns.uid]),
"folder": ns.folder,
"include_html": ns.html,
"mark_read": ns.mark_read,
"max_body_chars": ns.max_chars,
})
if ns.latest:
args.pop("uid")
args["query"] = "ALL"
args["limit"] = ns.latest
return commands.cmd_read(args)
def _message_args(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
args.update({
"to": _split_list(ns.to),
"cc": _split_list(ns.cc),
"bcc": _split_list(ns.bcc),
"subject": ns.subject or "",
"text": _read_text(ns.text, ns.text_file, "text"),
"html": _read_text(ns.html, ns.html_file, "html"),
"attachments": ns.attach or [],
"attachment_base": os.getcwd(),
"reply_to": ns.reply_to or "",
"in_reply_to": ns.in_reply_to or "",
"references": _split_list(ns.reference),
})
if not args["text"] and not args["html"]:
raise MailError("a body is required: --text/--text-file or --html/--html-file")
return args
def _cmd_send(ns: argparse.Namespace) -> Dict[str, Any]:
args = _message_args(ns)
if ns.dry_run:
args["dry_run"] = True
return commands.cmd_send(args)
def _cmd_draft(ns: argparse.Namespace) -> Dict[str, Any]:
args = _message_args(ns)
args["folder"] = ns.folder
return commands.cmd_draft(args)
def _cmd_flag(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
args.update({
"uid": _split_list([ns.uid]),
"folder": ns.folder,
"flags": _split_list([ns.flags]),
"mode": ns.mode,
})
return commands.cmd_flag(args)
def _cmd_move(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
args.update({
"uid": _split_list([ns.uid]),
"folder": ns.folder,
"to_folder": ns.to_folder,
})
return commands.cmd_move(args)
def _cmd_delete(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
args.update({
"uid": _split_list([ns.uid]),
"folder": ns.folder,
"permanent": ns.permanent,
})
return commands.cmd_delete(args)
def _cmd_download(ns: argparse.Namespace) -> Dict[str, Any]:
args = _base_args(ns)
args.update({
"uid": ns.uid,
"folder": ns.folder,
"dir": ns.dir,
"names": _split_list([ns.name]),
})
return commands.cmd_download(args)
# --------------------------------------------------------------------------
# Rendering
# --------------------------------------------------------------------------
def _render_check(data: Dict[str, Any]) -> str:
account = data.get("account", {})
return "\n".join([
f"account {account.get('address', '?')}",
f"provider {account.get('provider') or '(auto)'}",
f"imap {account.get('imap_host')}:{account.get('imap_port')}",
f"smtp {account.get('smtp_host')}:{account.get('smtp_port')}",
f"folder {data.get('folder')} unread: {data.get('unread')}",
f"folders {len(data.get('folders', []))}",
])
def _render_folders(data: Dict[str, Any]) -> str:
return "\n".join(
f" {f['name']}" + (f" [{f['special']}]" if f.get("special") else "")
for f in data.get("folders", [])
) or "(no folders)"
def _render_search(data: Dict[str, Any]) -> str:
lines = [f"{data['folder']} matched={data['matched']} shown={data['returned']}"
f" unread={data['unread_in_folder']}"]
for m in data.get("messages", []):
lines.append(f" uid={m.get('uid', '?'):>8} {m.get('date', ''):<32} {m.get('subject', '')}")
lines.append(f" from: {m.get('from', '')}")
if data.get("has_more"):
remaining = data["matched"] - data["offset"] - data["returned"]
lines.append(f" ... {remaining} more; raise --limit or set --offset")
return "\n".join(lines)
def _render_read(data: Dict[str, Any]) -> str:
blocks = []
for m in data.get("messages", []):
head = [
f"uid {m.get('uid')}",
f"from {m.get('from')}",
f"to {', '.join(m.get('to', []))}",
]
if m.get("cc"):
head.append(f"cc {', '.join(m['cc'])}")
head += [
f"date {m.get('date')}",
f"subject {m.get('subject')}",
f"body {m.get('text_source')} ({m.get('size', 0)} bytes)",
]
if m.get("truncated"):
head.append(f"truncated showing {len(m.get('text', ''))} of "
f"{m.get('text_chars', '?')} characters "
"(raise --max-chars or 0 for all)")
if m.get("attachments"):
head.append("attachments " + ", ".join(
f"{a['filename']} ({a['size']}B, {a['content_type']})" for a in m["attachments"]))
blocks.append("\n".join(head) + "\n\n" + m.get("text", ""))
return ("\n\n" + "-" * 60 + "\n\n").join(blocks) or "(no messages)"
def _render_send(data: Dict[str, Any]) -> str:
summary = data.get("summary", {})
return (f"sent from {data.get('from', '?')}\n"
f" via {data.get('server', '?')} ({data.get('transport', '?')})\n"
f" to {', '.join(summary.get('to', []))}\n"
f" subj {summary.get('subject', '')}")
def _render_draft(data: Dict[str, Any]) -> str:
summary = data.get("summary", {})
return (f"draft saved in {data.get('draft_saved_in', '?')}\n"
f" to {', '.join(summary.get('to', []))}\n"
f" subj {summary.get('subject', '')}")
def _render_dry_run(data: Dict[str, Any]) -> str:
return "DRY RUN — nothing sent\n" + json.dumps(
data.get("would_send", {}), ensure_ascii=False, indent=2)
def _render_flag(data: Dict[str, Any]) -> str:
parts = []
if data.get("added_count"):
parts.append(f"added {data['added_count']} ({', '.join(data.get('added') or [])})")
if data.get("cleared_count"):
parts.append(f"cleared {data['cleared_count']} ({', '.join(data.get('cleared') or [])})")
done = "; ".join(parts) or "nothing to change"
return f"{done} on {len(data.get('uids', []))} message(s) in {data.get('folder')}"
def _render_delete(data: Dict[str, Any]) -> str:
where = f"moved to {data['moved_to']}" if data.get("moved_to") else "expunged"
note = f" ({data['note']})" if data.get("note") else ""
return f"deleted {data.get('deleted', 0)} message(s) — {where}{note}"
def _render_download(data: Dict[str, Any]) -> str:
return "\n".join(f" saved {s['path']} ({s['size']}B)" for s in data.get("saved", []))
def _render_selftest(data: Dict[str, Any]) -> str:
lines = [data.get("summary", "")]
for section in ("offline", "live"):
info = data.get(section) or {}
for name in info.get("failures", []):
lines.append(f" FAIL {section}: {name}")
return "\n".join(lines)
_RENDERERS = {
"check": _render_check,
"folders": _render_folders,
"search": _render_search,
"read": _render_read,
"send": _render_send,
"draft": _render_draft,
"flag": _render_flag,
"delete": _render_delete,
"download": _render_download,
"selftest": _render_selftest,
}
def _render(command: str, data: Dict[str, Any]) -> str:
"""Human-readable rendering. The JSON output stays the source of truth."""
if data.get("dry_run"):
return _render_dry_run(data)
renderer = _RENDERERS.get(command)
if renderer is None:
return json.dumps(data, ensure_ascii=False, indent=2, default=str)
return renderer(data)
# --------------------------------------------------------------------------
# Parser
# --------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="mail.py",
description="General-purpose IMAP/SMTP client (stdlib only).",
)
parser.add_argument("--config", metavar="PATH", help="credential JSON file")
parser.add_argument("--account", metavar="NAME", help="named account inside that file")
parser.add_argument("--json", action="store_true", help="emit JSON instead of text")
sub = parser.add_subparsers(dest="command", metavar="<command>")
def add(name: str, handler: Any, help_text: str) -> argparse.ArgumentParser:
child = sub.add_parser(name, help=help_text, description=help_text)
child.set_defaults(handler=handler)
return child
p = add("check", _cmd_check, "verify credentials and show the mailbox")
p.add_argument("--folder", help="folder to select (default INBOX)")
add("folders", _cmd_folders, "list mail folders")
p = add("search", _cmd_search, "search messages; never marks mail as read")
p.add_argument("--folder", "-f", default="INBOX", help="folder (default INBOX)")
p.add_argument("--query", "-q", default="ALL", help="IMAP search criteria (default ALL)")
p.add_argument("--gmail-raw", help="Gmail X-GM-RAW query, used verbatim when supported")
p.add_argument("--since", help="SINCE date, YYYY-MM-DD")
p.add_argument("--before", help="BEFORE date, YYYY-MM-DD")
p.add_argument("--unread", action="store_true", help="unread only")
p.add_argument("--flagged", action="store_true", help="starred only")
p.add_argument("--has-attachment", action="store_true", help="multipart mail only")
p.add_argument("--limit", "-n", type=int, default=20, help="max results (default 20)")
p.add_argument("--offset", type=int, default=0, help="skip N newest first")
p = add("read", _cmd_read, "read full message bodies")
p.add_argument("--uid", help="message UID; omit with --latest")
p.add_argument("--latest", type=int, metavar="N", help="read the N newest instead of --uid")
p.add_argument("--folder", "-f", default="INBOX")
p.add_argument("--mark-read", action="store_true", help="set the Seen flag (off by default)")
p.add_argument("--html", action="store_true",
help="also return the raw HTML part (off by default; it is "
"markup, CSS and inline images the text already covers)")
p.add_argument("--max-chars", type=int, default=DEFAULT_MAX_BODY_CHARS,
metavar="N", help="truncate the body at N characters "
f"(default {DEFAULT_MAX_BODY_CHARS}; 0 disables)")
def add_message(name: str, handler: Any, help_text: str) -> None:
p = add(name, handler, help_text)
p.add_argument("--to", action="append", help="recipient (repeatable, or comma separated)")
p.add_argument("--cc", action="append", help="Cc recipient")
p.add_argument("--bcc", action="append", help="Bcc recipient (never shown in To/Cc)")
p.add_argument("--subject", help="subject line")
p.add_argument("--text", help="plain-text body")
p.add_argument("--text-file", help="read the plain-text body from a file")
p.add_argument("--html", help="HTML body")
p.add_argument("--html-file", help="read the HTML body from a file")
p.add_argument("--attach", action="append", help="attachment path (repeatable)")
p.add_argument("--reply-to", help="Reply-To header")
p.add_argument("--in-reply-to", help="Message-ID being replied to (threads the reply)")
p.add_argument("--reference", action="append", help="extra References entry")
add_message("send", _cmd_send, "send a message over SMTP")
sub.choices["send"].add_argument("--dry-run", action="store_true",
help="build and validate the message without sending")
add_message("draft", _cmd_draft, "save a message to the Drafts folder")
sub.choices["draft"].add_argument("--folder", "-f", default="Drafts",
help="target draft folder (default Drafts)")
p = add("flag", _cmd_flag, "change flags on messages")
p.add_argument("--uid", required=True, help="message UID (repeatable)")
p.add_argument("--folder", "-f", default="INBOX")
p.add_argument("--flags", required=True,
help="read/unread, star, answered, deleted, draft (repeatable)")
p.add_argument("--mode", choices=("add", "remove"), default="add",
help="add (default) or remove")
p = add("move", _cmd_move, "move messages to another folder")
p.add_argument("--uid", required=True, help="message UID (repeatable)")
p.add_argument("--folder", "-f", default="INBOX", help="source folder")
p.add_argument("--to-folder", required=True, help="destination folder")
p = add("delete", _cmd_delete, "delete messages (Trash when present)")
p.add_argument("--uid", required=True, help="message UID (repeatable)")
p.add_argument("--folder", "-f", default="INBOX")
p.add_argument("--permanent", action="store_true",
help="expunge instead of moving to Trash; cannot be undone")
p = add("download", _cmd_download, "save attachments from one message")
p.add_argument("--uid", required=True)
p.add_argument("--folder", "-f", default="INBOX")
p.add_argument("--dir", default=".", help="destination directory (default .)")
p.add_argument("--name", action="append", help="only save this filename (repeatable)")
p = add("selftest", _cmd_selftest, "verify this build; --live also connects to the mailbox")
p.add_argument("--live", action="store_true",
help="also run read-only IMAP probes against the configured mailbox")
p = add("json", None, "run a command from one JSON object")
p.add_argument("payload", help='e.g. \'{"command":"search","query":"UNSEEN"}\'')
p.set_defaults(handler=None)
return parser
# --------------------------------------------------------------------------
# Entry point
# --------------------------------------------------------------------------
def _cmd_selftest(ns: argparse.Namespace) -> Dict[str, Any]:
import selftest # local module, avoids a package-level import cycle
return selftest.run(live=ns.live)
def _run_json_command(ns: argparse.Namespace) -> Dict[str, Any]:
"""Dispatch the ``json`` escape hatch onto the normal command table."""
try:
payload = json.loads(ns.payload)
except json.JSONDecodeError as exc:
raise MailError(f"not valid JSON: {exc}") from exc
if not isinstance(payload, dict):
raise MailError("payload must be a JSON object")
name = str(payload.pop("command", "")).strip().lower()
handler = commands.COMMANDS.get(name)
if handler is None:
raise MailError(
f"unknown command {name!r}; available: {', '.join(sorted(commands.COMMANDS))}"
)
if ns.config and "config_path" not in payload:
payload["config_path"] = ns.config
return handler(payload)
def main(argv: Optional[List[str]] = None) -> int:
parser = build_parser()
ns = parser.parse_args(argv if argv is not None else sys.argv[1:])
if not getattr(ns, "command", None):
parser.print_help()
return EXIT_USAGE
try:
if ns.command == "json":
result = _run_json_command(ns)
else:
result = ns.handler(ns)
except ConfigError as exc:
return _emit(ns, {"ok": False, "error": str(exc), "type": "config"},
"set MAIL_ADDRESS and MAIL_PASSWORD, or pass --config", EXIT_CONFIG)
except MailError as exc:
return _emit(ns, {"ok": False, "error": str(exc), "type": "mail"}, "", EXIT_MAIL)
except KeyboardInterrupt:
return _emit(ns, {"ok": False, "error": "interrupted", "type": "mail"}, "", 130)
except Exception as exc: # noqa: BLE001 - report unexpected failures as data
return _emit(ns, {"ok": False, "error": f"{type(exc).__name__}: {exc}",
"type": "internal"}, "", EXIT_INTERNAL)
return _emit(ns, result, "", EXIT_OK)
def _emit(ns: argparse.Namespace, data: Dict[str, Any], hint: str, code: int) -> int:
payload = dict(data)
if hint and not payload.get("ok", False):
payload["hint"] = hint
if getattr(ns, "json", False):
print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
else:
command = getattr(ns, "command", "")
if payload.get("ok") is False:
print(f"error: {payload.get('error')}", file=sys.stderr)
if hint:
print(f"hint: {hint}", file=sys.stderr)
else:
print(_render(command, payload))
return code
if __name__ == "__main__":
sys.exit(main())
"""PYLinTech/Postman-Skill 的内部实现包。
这一行 docstring 是文件存在的唯一理由：它让 mailkit 成为一个**常规包**。
没有 __init__.py 时它会退化成命名空间包（PEP 420），而命名空间包在
sys.path 上遇到同名常规包时会被后者覆盖——即使本目录排在前面。
这会让一个碰巧叫 mailkit 的第三方包让整个 skill 起不来。
保留此文件，不要往里加再导出：调用方一律使用子模块路径
（from mailkit import commands / from mailkit.mime import ...）。
"""
"""Command implementations.
Every function takes the parsed argument dict and returns a JSON-serialisable
result. The password never appears in a return value or an error message.
"""
from __future__ import annotations
import contextlib
import email
import re
from pathlib import Path
from typing import Any, Dict, List
from .config import load_config, redact
from .imap_client import ImapSession
from .mime import (
MailError,
build_message,
decode_header_value,
safe_filename,
summarize_text,
unique_path,
)
from .smtp_client import send_message as smtp_send
_SEEN = "\\Seen"
# read/unread name a desired *state*, so they invert \Seen rather than toggling
# it: asking for "unread" must clear the flag, whatever --mode says.
_STATE_FLAGS = {"read": (_SEEN, True), "seen": (_SEEN, True), "unread": (_SEEN, False)}
_FLAG_ALIASES = {
"star": "\\Flagged",
"starred": "\\Flagged",
"flagged": "\\Flagged",
"answered": "\\Answered",
"deleted": "\\Deleted",
"draft": "\\Draft",
"forwarded": "$Forwarded",
"recent": "\\Recent",
}
def _as_list(value: Any) -> List[str]:
if value in (None, ""):
return []
if isinstance(value, list):
return [str(v).strip() for v in value if str(v).strip()]
return [part.strip() for part in re.split(r"[;,]", str(value)) if part.strip()]
def _dedupe(items: List[str]) -> List[str]:
seen, out = set(), []
for item in items:
if item not in seen:
seen.add(item)
out.append(item)
return out
def _plan_flag_changes(flags: List[str], mode: str) -> tuple:
"""Return ``(add, remove)`` IMAP flags for a request.
Later names win: asking for ``read unread`` leaves the message unread,
and a flag never lands on both lists.
"""
add: List[str] = []
remove: List[str] = []
for raw in flags:
key = str(raw).strip().lower().lstrip("-")
if key in _STATE_FLAGS:
flag, want = _STATE_FLAGS[key]
if mode == "remove":
want = not want
else:
flag = _FLAG_ALIASES.get(key, str(raw).strip())
want = mode != "remove"
if want:
remove = [f for f in remove if f != flag]
add.append(flag)
else:
add = [f for f in add if f != flag]
remove.append(flag)
return _dedupe(add), _dedupe(remove)
def _resolve_uids(args: Dict[str, Any], session: ImapSession) -> List[str]:
uids = [str(u).strip() for u in _as_list(args.get("uid")) if str(u).strip()]
if uids:
return uids
# Falling back to a query keeps ad-hoc usage possible without a prior search.
if args.get("query") or args.get("since"):
found = session.search(
str(args.get("query") or "ALL"),
since=str(args.get("since") or ""),
limit=int(args.get("limit") or 1),
)
if found:
return found
raise MailError("no uid given and no query to fall back on")
# --------------------------------------------------------------------------
# Commands
# --------------------------------------------------------------------------
@contextlib.contextmanager
def _mailbox(args: Dict[str, Any], *, select: str = ""):
"""Yield ``(config, session)`` with a live connection and *select* applied.
Centralises the resolve-credentials / connect / select / close sequence so
every command shares one session lifecycle and one error surface.
"""
config = load_config(args.get("config_path", ""), args)
session = ImapSession(config)
try:
session.connect()
if select:
session.select(select)
yield config, session
finally:
session.close()
def _target_folder(args: Dict[str, Any]) -> str:
return str(args.get("folder") or "INBOX")
def cmd_check(args: Dict[str, Any]) -> Dict[str, Any]:
with _mailbox(args, select=_target_folder(args)) as (config, session):
folders = session.list_folders()
return {
"ok": True,
"account": redact(config),
"folder": session.current_folder,
"unread": session.count_unseen(),
"folders": [f["name"] for f in folders],
"gmail_raw": session.supports_gmail_raw(),
"quota": session.quota_summary(),
}
def cmd_folders(args: Dict[str, Any]) -> Dict[str, Any]:
with _mailbox(args) as (_config, session):
return {"ok": True, "folders": session.list_folders()}
def cmd_search(args: Dict[str, Any]) -> Dict[str, Any]:
limit = max(1, int(args.get("limit") or 20))
offset = max(0, int(args.get("offset") or 0))
with _mailbox(args, select=_target_folder(args)) as (_config, session):
folder = session.current_folder
uids = session.search(
str(args.get("query") or "ALL"),
since=str(args.get("since") or ""),
before=str(args.get("before") or ""),
unread_only=bool(args.get("unread_only")),
flagged_only=bool(args.get("flagged_only")),
has_attachment=bool(args.get("has_attachment")),
gmail_raw=str(args.get("gmail_raw") or ""),
)
total = len(uids)
# Newest first for display; the caller still addresses UIDs.
window = list(reversed(uids))[offset:offset + limit]
messages = session.fetch_headers(window)
messages.reverse()
unread = session.count_unseen()
return {
"ok": True,
"folder": folder,
"matched": total,
"returned": len(messages),
"offset": offset,
"has_more": offset + len(messages) < total,
"unread_in_folder": unread,
"messages": messages,
}
def cmd_read(args: Dict[str, Any]) -> Dict[str, Any]:
mark_read = bool(args.get("mark_read"))
with _mailbox(args, select=_target_folder(args)) as (_config, session):
folder = session.current_folder
uids = _resolve_uids(args, session)
messages = session.fetch_messages(
uids,
include_html=bool(args.get("include_html", False)),
max_body_chars=int(args.get("max_body_chars") or 0),
)
if mark_read and messages:
session.set_flags(uids, ["\\Seen"], mode="add")
for item in messages:
# A short preview keeps list output useful when many mails are read.
item["preview"] = summarize_text(item.get("text", ""), 160)
return {
"ok": True,
"folder": folder,
"count": len(messages),
"marked_read": mark_read,
"messages": messages,
}
def _build_from_args(args: Dict[str, Any], config: Dict[str, Any]) -> tuple:
"""Assemble a message from a command payload.
Shared by ``send`` and ``draft`` so both accept the same headers; a draft
just skips delivery.
"""
to = _as_list(args.get("to"))
return build_message(
sender=str(config.get("send_as") or config["address"]),
sender_name=str(args.get("sender_name") or config.get("sender_name") or ""),
to=to or ["undisclosed-recipients:;"],
subject=str(args.get("subject") or ""),
text=str(args.get("text") or ""),
html=str(args.get("html") or ""),
cc=_as_list(args.get("cc")),
bcc=_as_list(args.get("bcc")),
reply_to=str(args.get("reply_to") or ""),
in_reply_to=str(args.get("in_reply_to") or ""),
references=_as_list(args.get("references")),
attachments=_as_list(args.get("attachments")),
attachment_paths=Path(args["attachment_base"]).expanduser() if args.get("attachment_base") else None,
extra_headers=args.get("headers") if isinstance(args.get("headers"), dict) else None,
)
def cmd_send(args: Dict[str, Any]) -> Dict[str, Any]:
config = load_config(args.get("config_path", ""), args)
if not _as_list(args.get("to")):
raise MailError("'to' is required (string, list, or ',' separated)")
message, summary = _build_from_args(args, config)
if args.get("dry_run"):
return {"ok": True, "dry_run": True, "would_send": summary,
"size": len(message.as_bytes())}
result = smtp_send(config, message)
result["summary"] = summary
return result
def cmd_draft(args: Dict[str, Any]) -> Dict[str, Any]:
config = load_config(args.get("config_path", ""), args)
message, summary = _build_from_args(args, config)
# Marks the message as an unsent draft for clients that look for it.
message["X-Unsent"] = "1"
with ImapSession(config) as session:
folder = session.append(str(args.get("folder") or "Drafts"), message)
return {"ok": True, "draft_saved_in": folder, "summary": summary}
def cmd_flag(args: Dict[str, Any]) -> Dict[str, Any]:
mode = str(args.get("mode") or "add").lower()
if mode not in {"add", "remove"}:
raise MailError("mode must be add or remove")
requested = _as_list(args.get("flags"))
add, remove = _plan_flag_changes(requested, mode)
if not add and not remove:
raise MailError("no usable flag names given")
with _mailbox(args, select=_target_folder(args)) as (_config, session):
folder = session.current_folder
uids = _resolve_uids(args, session)
counts: Dict[str, int] = {}
if add:
counts["added_count"] = session.set_flags(uids, add, mode="add")["updated"]
if remove:
counts["cleared_count"] = session.set_flags(uids, remove, mode="remove")["updated"]
# `added`/`cleared` are the requested IMAP flags; the counts are separate
# keys so one name never carries two different types.
return {
"ok": True, "folder": folder, "uids": uids, "requested": requested,
"mode": mode, "added": add, "cleared": remove, "updated": len(uids),
**counts,
}
def cmd_move(args: Dict[str, Any]) -> Dict[str, Any]:
destination = str(args.get("to_folder") or args.get("destination") or "")
if not destination:
raise MailError("'to_folder' is required")
with _mailbox(args, select=_target_folder(args)) as (_config, session):
folder = session.current_folder
uids = _resolve_uids(args, session)
target = session.move(uids, destination)
return {"ok": True, "from": folder, "to": target, "moved": len(uids), "uids": uids}
def cmd_delete(args: Dict[str, Any]) -> Dict[str, Any]:
permanent = bool(args.get("permanent"))
with _mailbox(args, select=_target_folder(args)) as (_config, session):
folder = session.current_folder
uids = _resolve_uids(args, session)
target = ""
if not permanent:
try:
target = session.move(uids, "Trash")
except MailError:
target = "" # No Trash folder on this account; fall through.
if target:
result = {"deleted": len(uids), "expunged": False, "moved_to": target}
else:
session.set_flags(uids, ["\\Deleted"], mode="add")
session.expunge()
result = {"deleted": len(uids), "expunged": True}
if not permanent:
result["note"] = "no Trash folder on this account; deleted permanently"
result.update({"ok": True, "from": folder, "uids": uids})
return result
def cmd_download(args: Dict[str, Any]) -> Dict[str, Any]:
uid = str(args.get("uid") or "").strip()
if not uid:
raise MailError("'uid' is required")
target_dir = Path(str(args.get("dir") or ".")).expanduser()
target_dir.mkdir(parents=True, exist_ok=True)
with _mailbox(args, select=_target_folder(args)) as (_config, session):
folder = session.current_folder
raw = session.fetch_raw(uid)
msg = email.message_from_bytes(raw)
wanted = {n.strip() for n in _as_list(args.get("names"))}
saved: List[Dict[str, Any]] = []
for part in msg.walk():
filename = decode_header_value(part.get_filename() or "")
if not filename or (wanted and filename not in wanted):
continue
payload = part.get_payload(decode=True) or b""
path = unique_path(target_dir, safe_filename(filename))
path.write_bytes(payload)
saved.append({"filename": path.name, "source_name": filename,
"path": str(path), "size": len(payload),
"content_type": part.get_content_type()})
if not saved:
raise MailError("no matching attachments on that message")
return {"ok": True, "folder": folder, "uid": uid, "saved": saved}
COMMANDS = {
"check": cmd_check,
"folders": cmd_folders,
"search": cmd_search,
"read": cmd_read,
"send": cmd_send,
"draft": cmd_draft,
"flag": cmd_flag,
"move": cmd_move,
"delete": cmd_delete,
"download": cmd_download,
}
"""Server presets and credential resolution for the email skill.
Pure stdlib. No network access lives here so the resolution logic stays
unit-testable without a mailbox.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
# (imap_host, imap_port, smtp_host, smtp_port)
MailServers = Tuple[str, int, str, int]
# Named providers. Ports follow each vendor's documented configuration:
# 993 = IMAP over implicit TLS
# 587 = SMTP submission over STARTTLS
# 465 = SMTP over implicit TLS
# 25 = plain SMTP (last resort)
_PROVIDER_PRESETS: Dict[str, MailServers] = {
"qq": ("imap.qq.com", 993, "smtp.qq.com", 587),
"gmail": ("imap.gmail.com", 993, "smtp.gmail.com", 587),
"163": ("imap.163.com", 993, "smtp.163.com", 587),
"126": ("imap.126.com", 993, "smtp.126.com", 587),
"yeah": ("imap.yeah.net", 993, "smtp.yeah.net", 587),
"outlook": ("outlook.office365.com", 993, "smtp.office365.com", 587),
"hotmail": ("outlook.office365.com", 993, "smtp.office365.com", 587),
"live": ("outlook.office365.com", 993, "smtp.office365.com", 587),
"yahoo": ("imap.mail.yahoo.com", 993, "smtp.mail.yahoo.com", 587),
"foxmail": ("imap.qq.com", 993, "smtp.qq.com", 587),
"sohu": ("imap.sohu.com", 993, "smtp.sohu.com", 465),
"sina": ("imap.sina.com", 993, "smtp.sina.com", 465),
"aliyun": ("imap.aliyun.com", 993, "smtp.aliyun.com", 465),
"tom": ("imap.tom.com", 993, "smtp.tom.com", 465),
}
# Domains that map onto a preset without the user naming a provider.
_DOMAIN_TO_PROVIDER: Dict[str, str] = {
"qq.com": "qq",
"foxmail.com": "foxmail",
"gmail.com": "gmail",
"googlemail.com": "gmail",
"163.com": "163",
"126.com": "126",
"yeah.net": "yeah",
"yahoo.com": "yahoo",
"yahoo.com.cn": "yahoo",
"ymail.com": "yahoo",
"outlook.com": "outlook",
"hotmail.com": "hotmail",
"live.com": "live",
"live.cn": "live",
"msn.com": "live",
"sohu.com": "sohu",
"sina.com": "sina",
"sina.cn": "sina",
"aliyun.com": "aliyun",
"tom.com": "tom",
}
DEFAULT_TIMEOUT = 30
DEFAULT_IMAP_HOST = "imap.qq.com"
DEFAULT_IMAP_PORT = 993
DEFAULT_SMTP_HOST = "smtp.qq.com"
DEFAULT_SMTP_PORT = 587
ENV_ADDRESS = "MAIL_ADDRESS"
ENV_PASSWORD = "MAIL_PASSWORD"
ENV_PROVIDER = "MAIL_PROVIDER"
ENV_CONFIG = "MAIL_CONFIG"
ENV_NAME = "MAIL_SENDER_NAME"
# RFC 2971 client identification. Generic and overridable via `client_name` so
# the skill carries no vendor branding. NetEase and QQ refuse LOGIN without it,
# so it is sent on every connect; servers that do not implement it simply reject
# the command, which callers treat as harmless.
IMAP_ID_NAME = "python-mailkit"
IMAP_ID_VERSION = "1.0.0"
def build_imap_id(name: str = IMAP_ID_NAME) -> str:
"""Return the argument string for the IMAP ``ID`` command."""
return (f'("name" "{name}" "version" "{IMAP_ID_VERSION}" '
f'"vendor" "{name}" "support-email" "noreply@localhost")')
class ConfigError(Exception):
"""Raised when credentials or server settings cannot be resolved."""
def _as_int(value: Any, default: int) -> int:
try:
parsed = int(str(value).strip())
except (TypeError, ValueError):
return default
return parsed if parsed > 0 else default
def email_domain(address: str) -> str:
text = str(address or "").strip().lower()
return text.rsplit("@", 1)[-1] if "@" in text else ""
def smtp_security_for_port(port: int, override: Optional[str] = None) -> str:
"""Map an SMTP port onto a transport mode.
465 is implicit TLS, 587 is STARTTLS, everything else is treated as plain.
An explicit ``smtp_security`` always wins so odd corporate ports work.
"""
if override:
mode = str(override).strip().lower()
if mode in {"ssl", "starttls", "none"}:
return mode
if port == 465:
return "ssl"
if port == 587:
return "starttls"
return "none"
def detect_provider(address: str, explicit: str = "") -> str:
"""Resolve a provider key from an explicit choice, then the email domain."""
name = str(explicit or "").strip().lower()
if name and name != "auto":
return name
return _DOMAIN_TO_PROVIDER.get(email_domain(address), "")
def resolve_servers(
address: str,
provider: str = "",
overrides: Optional[Dict[str, Any]] = None,
) -> Tuple[MailServers, str]:
"""Return ``((imap_host, imap_port, smtp_host, smtp_port), provider_key)``.
Precedence: explicit per-host overrides, then a named provider preset, then
the email domain, then the QQ defaults.
"""
overrides = overrides or {}
key = detect_provider(address, provider)
if key and key not in _PROVIDER_PRESETS:
raise ConfigError(
f"unknown mail provider {key!r}; use one of {sorted(_PROVIDER_PRESETS)} "
"or 'auto'/'custom'"
)
preset = _PROVIDER_PRESETS.get(key)
if preset is None:
preset = (DEFAULT_IMAP_HOST, DEFAULT_IMAP_PORT, DEFAULT_SMTP_HOST, DEFAULT_SMTP_PORT)
imap_host = str(overrides.get("imap_host") or "").strip() or preset[0]
imap_port = _as_int(overrides.get("imap_port"), preset[1])
smtp_host = str(overrides.get("smtp_host") or "").strip() or preset[2]
smtp_port = _as_int(overrides.get("smtp_port"), preset[3])
return (imap_host, imap_port, smtp_host, smtp_port), key
def _read_json(path: Path) -> Dict[str, Any]:
try:
data = json.loads(path.read_text(encoding="utf-8-sig"))
except FileNotFoundError as exc:
raise ConfigError(f"config file not found: {path}") from exc
except (json.JSONDecodeError, OSError) as exc:
raise ConfigError(f"cannot read config file {path}: {exc}") from exc
if not isinstance(data, dict):
raise ConfigError(f"config file {path} must contain a JSON object")
return data
def _pick(source: Dict[str, Any], key: str, env_name: str, default: str = "") -> str:
value = source.get(key)
if value not in (None, ""):
return str(value).strip()
return str(os.environ.get(env_name, default) or "").strip()
def load_config(
config_path: str = "",
overrides: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Resolve the effective mailbox settings.
Precedence for every field: explicit override, then a JSON config file
(argument > ``MAIL_CONFIG`` env), then environment variables.
The password is required; everything else falls back to a preset. The
returned dict is safe to serialise for diagnostics *except* ``password``,
which :func:`redact` strips.
"""
overrides = dict(overrides or {})
path = str(overrides.get("config_path") or config_path or os.environ.get(ENV_CONFIG, "")).strip()
file_data: Dict[str, Any] = {}
if path:
file_data = _read_json(Path(path).expanduser())
# A file may hold several accounts under an "accounts" map; --account
# picks one. An unknown name must fail loudly — falling through to the
# file's top-level fields would quietly address the wrong mailbox.
account = str(overrides.get("account") or "").strip()
if account:
accounts = file_data.get("accounts")
if not isinstance(accounts, dict):
raise ConfigError(
f"--account {account!r} was given but {path} has no "
'"accounts" map'
)
selected = accounts.get(account)
if not isinstance(selected, dict):
raise ConfigError(
f"account {account!r} not found in {path}; available: "
+ (", ".join(sorted(accounts)) or "(none)")
)
file_data = {k: v for k, v in file_data.items() if k != "accounts"}
file_data.update(selected)
def field(key: str, env_name: str) -> str:
return _pick(file_data, key, env_name) or _pick(overrides, key, "", "")
address = field("address", ENV_ADDRESS)
password = field("password", ENV_PASSWORD)
if not address:
raise ConfigError(
"no mailbox address configured; set MAIL_ADDRESS, pass config_path, "
"or copy resources/config.example.json"
)
if not password:
raise ConfigError(
"no mailbox password configured; set MAIL_PASSWORD or put it in the config file"
)
if "@" not in address:
raise ConfigError(f"invalid mailbox address: {address!r}")
provider = field("provider", ENV_PROVIDER)
servers, key = resolve_servers(address, provider, overrides=file_data)
imap_host = str(file_data.get("imap_host") or "").strip() or servers[0]
imap_port = _as_int(file_data.get("imap_port"), servers[1])
smtp_host = str(file_data.get("smtp_host") or "").strip() or servers[2]
smtp_port = _as_int(file_data.get("smtp_port"), servers[3])
# An explicit provider preset is only a default; a host the user pinned
# always wins, even when it is a NetEase host the preset would not choose.
imap_host = str(overrides.get("imap_host") or "").strip() or imap_host
smtp_host = str(overrides.get("smtp_host") or "").strip() or smtp_host
return {
"address": address,
"password": password,
"sender_name": field("sender_name", ENV_NAME),
"client_name": str(file_data.get("client_name") or "").strip() or IMAP_ID_NAME,
"provider": key,
"imap_host": imap_host,
"imap_port": imap_port,
"smtp_host": smtp_host,
"smtp_port": smtp_port,
"smtp_security": str(file_data.get("smtp_security") or "").strip().lower(),
"timeout": _as_int(file_data.get("timeout"), DEFAULT_TIMEOUT),
"verify_ssl": bool(file_data.get("verify_ssl", True)),
"send_as": str(file_data.get("send_as") or "").strip() or address,
}
def redact(config: Dict[str, Any]) -> Dict[str, Any]:
"""Copy of *config* with the password replaced by a fixed placeholder."""
safe = dict(config)
safe["password"] = "***" if config.get("password") else ""
return safe
"""IMAP session management.
Uses UID commands exclusively: sequence numbers shift whenever another client
touches the mailbox, while a UID is stable for the life of the message. Search
uses ``BODY.PEEK`` so listing mail never flips the read flag as a side effect.
"""
from __future__ import annotations
import imaplib
import re
import ssl
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
from .config import IMAP_ID_NAME, ConfigError, build_imap_id
from .mime import MailError, parse_headers_only
# RFC 2971 client identification; imaplib does not register the command.
imaplib.Commands.setdefault("ID", ("AUTH", "NONAUTH", "SELECTED"))
_DEFAULT_FOLDER = "INBOX"
_FOLDER_ALIASES = {
"inbox": "INBOX",
"sent": "Sent",
"sent items": "Sent",
"draft": "Drafts",
"drafts": "Drafts",
"trash": "Trash",
"deleted": "Trash",
"junk": "Junk",
"spam": "Junk",
"archive": "Archive",
"all": "",
}
_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
_DATE_RE = re.compile(r"^(\d{4})-(\d{1,2})-(\d{1,2})$")
def _quote_folder(name: str) -> str:
"""Quote a folder name for IMAP, escaping backslashes and quotes."""
return '"' + str(name).replace("\\", "\\\\").replace('"', '\\"') + '"'
def _check(typ: str, data: Any, action: str) -> Any:
if typ != "OK":
detail = ""
if isinstance(data, list) and data:
raw = data[0]
detail = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else str(raw)
raise MailError(f"{action} failed ({typ}): {detail}".strip())
return data
def _normalise_date(value: str) -> str:
"""Convert ``YYYY-MM-DD`` to IMAP's ``01-Jan-2024``; pass others through."""
match = _DATE_RE.match(str(value or "").strip())
if not match:
return str(value or "").strip()
year, month, day = match.groups()
if not 1 <= int(month) <= 12:
raise MailError(f"invalid month in date: {value!r}")
return f"{int(day):02d}-{_MONTHS[int(month) - 1]}-{year}"
def normalise_folder(name: str) -> str:
"""Map a friendly alias onto a real folder name; ``''`` means all folders."""
raw = str(name or "").strip()
if not raw:
return _DEFAULT_FOLDER
return _FOLDER_ALIASES.get(raw.lower(), raw)
def _special_key(name: str) -> str:
"""Normalise for special-use matching: ``Drafts`` and ``\\Draft`` agree."""
key = str(name or "").strip().lstrip("\\").lower()
return key[:-1] if key.endswith("s") else key
class ImapSession:
"""A logged-in IMAP session. Use as a context manager."""
def __init__(self, config: Dict[str, Any]) -> None:
self._config = config
self._imap: Optional[imaplib.IMAP4_SSL] = None
self._folder: str = _DEFAULT_FOLDER
# -- lifecycle -------------------------------------------------------
def __enter__(self) -> "ImapSession":
self.connect()
return self
def __exit__(self, *_exc: Any) -> None:
self.close()
def connect(self) -> "ImapSession":
host = self._config["imap_host"]
port = int(self._config["imap_port"])
timeout = int(self._config.get("timeout", 30))
context = ssl.create_default_context()
if not self._config.get("verify_ssl", True):
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
try:
self._imap = imaplib.IMAP4_SSL(host, port, ssl_context=context, timeout=timeout)
except ssl.SSLCertVerificationError as exc:
raise ConfigError(
f"TLS certificate verification failed for {host}:{exc}. "
"Use a valid certificate, or set verify_ssl=false in the config "
"if you accept the risk."
) from exc
except OSError as exc:
raise MailError(f"cannot connect to {host}:{port}: {exc}") from exc
self._send_id()
try:
self._imap.login(self._config["address"], self._config["password"])
except imaplib.IMAP4.error as exc:
raise MailError(
f"IMAP login failed for {self._config['address']} at {host}:{port}. "
"Check the address and password — most providers require an "
"app-specific password rather than the account password. "
f"({exc})"
) from exc
return self
def _send_id(self) -> None:
"""Send the RFC 2971 client ID; NetEase and QQ reject LOGIN without it."""
assert self._imap is not None
args = build_imap_id(str(self._config.get("client_name") or IMAP_ID_NAME))
try:
typ, _ = self._imap._simple_command("ID", args) # noqa: SLF001
if typ != "OK":
raise MailError(f"IMAP ID failed: {typ}")
except imaplib.IMAP4.error:
# Servers that do not advertise ID reject the command; that is fine.
pass
def close(self) -> None:
if self._imap is not None:
try:
self._imap.close()
except Exception:
pass
try:
self._imap.logout()
except Exception:
pass
self._imap = None
@property
def imap(self) -> imaplib.IMAP4_SSL:
if self._imap is None:
raise MailError("IMAP session is not connected")
return self._imap
# -- capabilities ----------------------------------------------------
def capabilities(self) -> Tuple[str, ...]:
return tuple(str(c).upper() for c in (self.imap.capabilities or ()))
def supports_gmail_raw(self) -> bool:
return "X-GM-RAW" in self.capabilities()
def _is_gmail(self) -> bool:
return "imap.gmail.com" in str(self._config.get("imap_host", "")).lower()
# -- folders ---------------------------------------------------------
def list_folders(self) -> List[Dict[str, Any]]:
typ, data = self.imap.list()
_check(typ, data, "LIST")
out: List[Dict[str, Any]] = []
for raw in data or []:
line = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else str(raw)
match = re.match(r'\((?P<flags>[^)]*)\)\s+"?(?P<delim>[^" ]+)\s+(?P<name>.+?)"?$', line)
if not match:
continue
name = match.group("name").strip().strip('"')
flags = match.group("flags").lower().split()
out.append({
"name": name,
"special": next((f.lstrip("\\") for f in flags if f.startswith("\\")), ""),
"delimiter": match.group("delim"),
})
return out
def find_folder(self, wanted: str) -> str:
"""Resolve an alias to a real folder, tolerating localised names."""
target = normalise_folder(wanted)
if target == _DEFAULT_FOLDER:
return target
folders = {f["name"]: f for f in self.list_folders()}
if target in folders:
return target
# Localised mailboxes (Sent Items, 已发送, Gesendet, ...) differ per UI
# language; fall back to the standard special-use flag, ignoring a
# trailing plural on either side so Drafts matches \Drafts.
wanted_key = _special_key(target)
for name, meta in folders.items():
if meta["special"] and _special_key(meta["special"]) == wanted_key:
return name
lowered = {name.lower(): name for name in folders}
if target.lower() in lowered:
return lowered[target.lower()]
raise MailError(
f"folder {target!r} not found; available: " + ", ".join(sorted(folders)[:20])
)
def select(self, folder: str = "") -> str:
name = self.find_folder(folder or _DEFAULT_FOLDER)
typ, data = self.imap.select(_quote_folder(name), readonly=False)
_check(typ, data, f"SELECT {name}")
self._folder = name
return name
@property
def current_folder(self) -> str:
return self._folder
# -- searching -------------------------------------------------------
def search(
self,
query: str = "ALL",
*,
since: str = "",
before: str = "",
unread_only: bool = False,
flagged_only: bool = False,
has_attachment: bool = False,
gmail_raw: str = "",
limit: int = 0,
) -> List[str]:
"""Return UIDs, newest last. ``gmail_raw`` uses Gmail's native syntax."""
criteria: List[str] = []
if gmail_raw and (self.supports_gmail_raw() or self._is_gmail()):
typ, data = self.imap.uid("search", None, "X-GM-RAW", gmail_raw)
_check(typ, data, "UID SEARCH (X-GM-RAW)")
uids = (data[0] or b"").split()
else:
if query:
criteria.append(str(query).strip() or "ALL")
if since:
criteria.append(f'"SINCE" "{_normalise_date(since)}"')
if before:
criteria.append(f'"BEFORE" "{_normalise_date(before)}"')
if unread_only:
criteria.append("UNSEEN")
if flagged_only:
criteria.append("FLAGGED")
if has_attachment:
criteria.append('HEADER Content-Type "multipart/mixed"')
typ, data = self.imap.uid("search", *criteria)
_check(typ, data, "UID SEARCH")
uids = (data[0] or b"").split()
out = [u.decode() for u in uids if u]
return out[-limit:] if limit and limit > 0 else out
def count_unseen(self) -> int:
try:
typ, data = self.imap.status(_quote_folder(self._folder), "(UNSEEN)")
_check(typ, data, "STATUS")
raw = data[0].decode() if isinstance(data[0], bytes) else str(data[0])
match = re.search(r"UNSEEN\s+(\d+)", raw)
return int(match.group(1)) if match else 0
except Exception:
return 0
# -- fetching --------------------------------------------------------
def fetch_headers(self, uids: Sequence[str]) -> List[Dict[str, Any]]:
if not uids:
return []
typ, data = self.imap.uid(
"fetch", ",".join(uids), "(BODY.PEEK[HEADER.FIELDS (FROM TO CC SUBJECT DATE MESSAGE-ID)])"
)
_check(typ, data, "UID FETCH headers")
return self._collect(data, headers_only=True)
def fetch_messages(
self,
uids: Sequence[str],
*,
include_html: bool = False,
max_body_chars: int = 0,
) -> List[Dict[str, Any]]:
if not uids:
return []
typ, data = self.imap.uid("fetch", ",".join(uids), "(BODY.PEEK[])")
_check(typ, data, "UID FETCH body")
from .mime import parse_message # local import keeps module import cheap
out: List[Dict[str, Any]] = []
for uid, raw in self._raw_chunks(data):
item = parse_message(raw, include_html=include_html,
max_body_chars=max_body_chars)
item["uid"] = uid
out.append(item)
return out
def fetch_raw(self, uid: str) -> bytes:
"""Fetch one message's full RFC 822 bytes without marking it read."""
typ, data = self.imap.uid("fetch", str(uid), "(BODY.PEEK[])")
_check(typ, data, f"UID FETCH {uid}")
for _found_uid, raw in self._raw_chunks(data):
return raw
raise MailError(f"message uid {uid} not found or has no body")
def _raw_chunks(self, data: Any) -> Iterable[Tuple[str, bytes]]:
for chunk in data or []:
if not isinstance(chunk, tuple) or len(chunk) < 2:
continue
meta, payload = chunk[0], chunk[1]
uid = ""
if isinstance(meta, bytes):
head = meta.decode("utf-8", "replace")
match = re.search(r"UID\s+(\d+)", head)
if match:
uid = match.group(1)
if isinstance(payload, bytes) and payload:
yield uid, payload
def _collect(self, data: Any, *, headers_only: bool) -> List[Dict[str, Any]]:
out: List[Dict[str, Any]] = []
for uid, raw in self._raw_chunks(data):
item = parse_headers_only(raw) if headers_only else {}
item["uid"] = uid
out.append(item)
return out
# -- flags -----------------------------------------------------------
def set_flags(self, uids: Sequence[str], flags: Sequence[str], *, mode: str = "set") -> Dict[str, Any]:
if not uids:
return {"updated": 0, "requested": list(flags), "mode": mode}
typed = tuple(flags)
if mode == "add":
typ, data = self.imap.uid("store", ",".join(uids), "+FLAGS.SILENT", typed)
elif mode == "remove":
typ, data = self.imap.uid("store", ",".join(uids), "-FLAGS.SILENT", typed)
else:
typ, data = self.imap.uid("store", ",".join(uids), "FLAGS.SILENT", typed)
_check(typ, data, "UID STORE")
return {"updated": len(uids), "requested": list(flags), "mode": mode}
def copy(self, uids: Sequence[str], destination: str) -> str:
target = self.find_folder(destination)
typ, data = self.imap.uid("copy", ",".join(uids), _quote_folder(target))
_check(typ, data, f"UID COPY to {target}")
return target
def move(self, uids: Sequence[str], destination: str) -> str:
"""Move via COPY + ``\\Deleted`` + EXPUNGE, which every server supports.
EXPUNGE is not UID-scoped, so anything else already flagged ``\\Deleted``
in this folder goes too. That matches every other IMAP client.
"""
target = self.copy(uids, destination)
self.set_flags(uids, ["\\Deleted"], mode="add")
self.expunge()
return target
def expunge(self) -> None:
typ, data = self.imap.expunge()
_check(typ, data, "EXPUNGE")
def append(self, folder: str, message: Any) -> str:
target = self.find_folder(folder)
raw = message.as_bytes()
typ, data = self.imap.append(_quote_folder(target), None, None, raw)
_check(typ, data, f"APPEND to {target}")
return target
def quota_summary(self) -> Dict[str, Any]:
"""Best-effort mailbox usage; not all servers implement QUOTA."""
try:
typ, data = self.imap.getquota(_quote_folder(""))
if typ != "OK" or not data or not isinstance(data[0], bytes):
return {}
parts = data[0].decode("utf-8", "replace").split()
out: Dict[str, Any] = {}
for index in range(0, len(parts) - 1, 2):
out[parts[index]] = int(parts[index + 1])
return out
except Exception:
return {}
"""MIME parsing and building, including HTML mail support.
Everything here is pure-Python and offline so it can be unit-tested without a
mailbox. The HTML-to-text converter is the piece that matters most: a plain
``text/plain`` part is preferred when the sender supplied one, and HTML is only
converted as a fallback, so rich newsletters degrade to readable text instead of
returning nothing.
"""
from __future__ import annotations
import email
import email.header
import email.utils
import mimetypes
import re
from email.header import decode_header
from email.message import EmailMessage, Message
from html.parser import HTMLParser
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
# Elements whose entire subtree never contributes readable text.
_SKIP_CONTENT = {"script", "style", "head", "noscript", "template", "svg", "canvas"}
# Elements that force a line break when they open or close.
_BLOCK_TAGS = {
"address", "article", "aside", "blockquote", "center", "dd", "div", "dl", "dt",
"fieldset", "figcaption", "figure", "footer", "form", "h1", "h2", "h3", "h4",
"h5", "h6", "header", "hr", "li", "main", "nav", "ol", "p", "pre", "section",
"table", "tbody", "tfoot", "thead", "tr", "ul",
}
_UNSAFE_FILENAME = re.compile(r'[\x00-\x1f/\\:*?"<>|]+')
class MailError(Exception):
"""Raised for user-facing failures (bad folder, missing UID, auth, ...)."""
# --------------------------------------------------------------------------
# Headers
# --------------------------------------------------------------------------
def decode_header_value(raw: Any) -> str:
"""Decode an RFC 2047 encoded header into text.
Chinese subjects arrive as ``=?utf-8?B?...?=``; without this the caller sees
raw base64. Falls back to the original string when a part cannot be decoded.
"""
if raw is None:
return ""
text = raw if isinstance(raw, str) else str(raw)
if not text:
return ""
try:
parts = decode_header(text)
except Exception:
return text
out: List[str] = []
for value, charset in parts:
if isinstance(value, bytes):
out.append(_decode_bytes(value, charset))
else:
out.append(value)
return "".join(out).strip()
def _decode_bytes(value: bytes, charset: Optional[str]) -> str:
"""Decode one RFC 2047 chunk, trying the declared charset first.
``errors="replace"`` never raises, so the candidate list is walked with a
strict decode and only the final attempt is allowed to substitute. Without
that, a mislabelled charset silently yields mojibake.
"""
candidates = [charset] if charset else []
candidates += ["utf-8", "gb18030"]
for encoding in candidates:
if not encoding:
continue
try:
return value.decode(encoding)
except (LookupError, UnicodeDecodeError):
continue
return value.decode("utf-8", errors="replace")
def format_address(value: str) -> str:
"""Render an address as ``Display Name <addr@host>`` for use in a header.
``email.utils.formataddr`` re-encodes a non-ASCII display name, which is
correct for a header we are composing but wrong when we are reading one.
Use :func:`display_address` for the parsing side.
"""
name, addr = email.utils.parseaddr(str(value or ""))
name = decode_header_value(name)
if not addr:
return name or str(value or "")
return email.utils.formataddr((name, addr)) if name else addr
def display_address(value: Any) -> str:
"""Decode a header address for human/model consumption, without re-encoding."""
name, addr = email.utils.parseaddr(decode_header_value(str(value or "")))
name = name.strip()
if name and addr:
return f"{name} <{addr}>"
return addr or name or str(value or "").strip()
def parse_address_list(value: Any) -> List[Tuple[str, str]]:
"""Return ``[(display_name, addr_spec), ...]`` for a header field."""
if value is None:
return []
items = value if isinstance(value, list) else str(value).split(";")
out: List[Tuple[str, str]] = []
for item in items:
name, addr = email.utils.parseaddr(decode_header_value(item).strip())
if addr:
out.append((name, addr))
return out
# --------------------------------------------------------------------------
# HTML -> text
# --------------------------------------------------------------------------
class _HTMLToText(HTMLParser):
"""Flatten HTML into readable plain text.
Handles the shapes that actually show up in mail: block elements become
line breaks, ``<br>`` is a hard break, list items get a bullet, table rows
become one line per row, and links keep their href when the anchor text
would otherwise lose the destination.
"""
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.parts: List[str] = []
self._skip_depth = 0
self._href: Optional[str] = None
self._anchor_text: List[str] = []
# -- helpers ---------------------------------------------------------
def _break(self) -> None:
if self.parts and not self.parts[-1].endswith("\n"):
self.parts.append("\n")
def _newline(self) -> None:
if self.parts and self.parts[-1] not in ("\n", "\n\n"):
self.parts.append("\n\n")
# -- parser hooks ----------------------------------------------------
def handle_starttag(self, tag: str, attrs: List[Tuple[str, Optional[str]]]) -> None:
tag = tag.lower()
if tag in _SKIP_CONTENT:
self._skip_depth += 1
return
if self._skip_depth:
return
if tag == "br":
self.parts.append("\n")
elif tag == "li":
self._newline()
self.parts.append("- ")
elif tag == "hr":
self._newline()
self.parts.append("---")
self._newline()
elif tag == "td" or tag == "th":
self.parts.append("\t")
elif tag in _BLOCK_TAGS:
self._newline()
elif tag == "a":
href = dict(attrs).get("href")
self._href = href.strip() if isinstance(href, str) else None
self._anchor_text = []
def handle_startendtag(self, tag: str, attrs: List[Tuple[str, Optional[str]]]) -> None:
if tag.lower() == "br" and not self._skip_depth:
self.parts.append("\n")
def handle_endtag(self, tag: str) -> None:
tag = tag.lower()
if tag in _SKIP_CONTENT:
if self._skip_depth:
self._skip_depth -= 1
return
if self._skip_depth:
return
if tag == "a" and self._href:
# Only surface the URL when the visible text is not already the URL.
label = "".join(self._anchor_text).strip()
if self._href not in label and not self._href.lower().startswith("javascript:"):
self.parts.append(f" <{self._href}>")
self._href = None
self._anchor_text = []
elif tag == "li":
self._break()
elif tag in ("td", "th"):
self.parts.append(" ")
elif tag == "tr" or tag in _BLOCK_TAGS:
self._newline()
def handle_data(self, data: str) -> None:
if self._skip_depth or not data:
return
if not data.strip():
if self.parts and not self.parts[-1].endswith("\n"):
# Preserve a single separating space between inline runs.
self.parts.append(" ")
return
self.parts.append(data)
if self._href is not None:
self._anchor_text.append(data)
def result(self) -> str:
text = "".join(self.parts).replace("\r\n", "\n").replace("\r", "\n")
# Collapse runs of spaces, and runs of tabs, but keep tabs as column
# separators so table rows stay readable.
lines = [re.sub(r" {2,}", " ", re.sub(r"\t{2,}", "\t", line)).strip(" \t") for line in text.split("\n")]
out: List[str] = []
blank = False
for line in lines:
if not line:
if out and not blank:
out.append("")
blank = True
continue
blank = False
out.append(line)
return "\n".join(out).strip()
def html_to_text(html: str) -> str:
"""Convert an HTML document or fragment to readable plain text."""
if not html or not str(html).strip():
return ""
parser = _HTMLToText()
try:
parser.feed(str(html))
parser.close()
except Exception:
# A malformed document should still yield whatever was parsed.
pass
text = parser.result()
if text:
return text
# Last resort: strip tags outright so a broken document is not lost.
stripped = re.sub(r"<[^>]+>", " ", str(html))
stripped = re.sub(r"\s+", " ", stripped)
return stripped.strip()
def summarize_text(html: str, limit: int = 120) -> str:
"""Derive a one-line summary from HTML, for list previews."""
text = html_to_text(html)
text = " ".join(text.split())
return text[:limit]
# --------------------------------------------------------------------------
# Parsing
# --------------------------------------------------------------------------
def _part_text(part: Message) -> str:
payload = part.get_payload(decode=True)
if payload is None:
return ""
return _decode_bytes(payload, part.get_content_charset())
def safe_filename(name: str, fallback: str = "attachment") -> str:
"""Strip path separators and control characters from an attachment name."""
cleaned = _UNSAFE_FILENAME.sub("_", str(name or "").strip())
cleaned = cleaned.strip(". ") or fallback
return cleaned[:120]
def unique_path(directory: Path, filename: str) -> Path:
"""Resolve *filename* inside *directory*, avoiding collisions on disk.
Two parts in one message can share a name; silently overwriting the first
would lose data, so ``report.pdf`` becomes ``report(1).pdf`` and onward.
"""
candidate = directory / filename
if not candidate.exists():
return candidate
stem, dot, suffix = filename.rpartition(".")
if not dot:
stem, suffix = filename, ""
for index in range(1, 1000):
attempt = f"{stem}({index}){'.' + suffix if suffix else ''}"
candidate = directory / attempt
if not candidate.exists():
return candidate
raise MailError(f"too many files named {filename!r} in {directory}")
def parse_message(
raw: bytes,
*,
include_html: bool = False,
max_body_chars: int = 0,
) -> Dict[str, Any]:
"""Parse a raw RFC 822 message into a JSON-serialisable dict.
``text`` is the readable body: the sender's ``text/plain`` part when
present, otherwise the HTML part converted. ``text_source`` records which
path was taken so the caller can tell an HTML-only mail from an empty one.
``include_html`` is off by default: the extracted text is what a reader
wants, while the raw part mostly carries markup, CSS and inline base64
images that would multiply the context an agent has to read.
``max_body_chars`` truncates the body when positive and records the
original length, so a caller can tell a short mail from a cut one.
"""
msg = email.message_from_bytes(raw)
plain_parts: List[str] = []
html_parts: List[str] = []
attachments: List[Dict[str, Any]] = []
related: List[Dict[str, Any]] = []
in_related = False
for part in msg.walk():
if part.get_content_maintype() == "multipart":
subtype = (part.get_content_subtype() or "").lower()
if subtype == "related":
in_related = True
continue
disposition = (part.get_content_disposition() or "").lower()
ctype = part.get_content_type()
filename = part.get_filename()
if filename:
filename = decode_header_value(filename)
if disposition == "attachment" or (filename and ctype not in ("text/plain", "text/html")):
payload = part.get_payload(decode=True) or b""
attachments.append({
"filename": safe_filename(filename or "", f"part-{len(attachments) + 1}"),
"content_type": ctype,
"size": len(payload),
})
continue
if ctype == "text/plain":
body = _part_text(part)
if body and not in_related:
plain_parts.append(body)
elif ctype == "text/html":
body = _part_text(part)
if body:
html_parts.append(body)
elif ctype.startswith("image/") and in_related:
related.append({
"content_type": ctype,
"content_id": (part.get("Content-ID") or "").strip("<>"),
"size": len(part.get_payload(decode=True) or b""),
})
html = "\n".join(html_parts).strip()
plain = "\n\n".join(p for p in plain_parts if p.strip()).strip()
if plain:
text, source = plain, "plain"
elif html:
text, source = html_to_text(html), "html"
else:
text, source = "", "empty"
body_bytes = len(raw)
original_chars = len(text)
truncated = False
if max_body_chars and original_chars > max_body_chars:
text = text[:max_body_chars]
truncated = True
result: Dict[str, Any] = {
"message_id": decode_header_value(msg.get("Message-ID", "")),
"in_reply_to": decode_header_value(msg.get("In-Reply-To", "")),
"references": decode_header_value(msg.get("References", "")),
"subject": decode_header_value(msg.get("Subject", "")),
"from": display_address(msg.get("From", "")),
"to": [display_address(a) for _, a in parse_address_list(msg.get("To", ""))],
"cc": [display_address(a) for _, a in parse_address_list(msg.get("Cc", ""))],
"date": decode_header_value(msg.get("Date", "")),
"text": text,
"text_source": source,
"text_chars": original_chars,
"truncated": truncated,
"attachments": attachments,
"size": body_bytes,
}
if include_html and html:
result["html"] = html
if related:
result["inline_images"] = related
return result
def parse_headers_only(raw: bytes) -> Dict[str, str]:
"""Header-only parse, used by list/search to avoid fetching bodies."""
msg = email.message_from_bytes(raw)
return {
"message_id": decode_header_value(msg.get("Message-ID", "")),
"subject": decode_header_value(msg.get("Subject", "")),
"from": display_address(msg.get("From", "")),
"date": decode_header_value(msg.get("Date", "")),
}
# --------------------------------------------------------------------------
# Building
# --------------------------------------------------------------------------
def _guess_type(path: Path) -> Tuple[str, str]:
guessed, _ = mimetypes.guess_type(str(path))
maintype, _, subtype = (guessed or "application/octet-stream").partition("/")
return maintype or "application", subtype or "octet-stream"
def _attach_files(message: EmailMessage, paths: Iterable[str], base_dir: Optional[Path]) -> List[str]:
attached: List[str] = []
for raw_path in paths:
candidate = Path(str(raw_path)).expanduser()
if not candidate.is_absolute() and base_dir is not None:
candidate = (base_dir / candidate).resolve()
if not candidate.is_file():
raise MailError(f"attachment not found: {raw_path}")
maintype, subtype = _guess_type(candidate)
data = candidate.read_bytes()
message.add_attachment(
data,
maintype=maintype,
subtype=subtype,
filename=candidate.name,
)
attached.append(candidate.name)
return attached
def build_message(
*,
sender: str,
sender_name: str = "",
to: Iterable[str],
subject: str = "",
text: str = "",
html: str = "",
cc: Iterable[str] = (),
bcc: Iterable[str] = (),
reply_to: str = "",
in_reply_to: str = "",
references: Iterable[str] = (),
attachments: Iterable[str] = (),
attachment_paths: Optional[Path] = None,
date: Optional[str] = None,
extra_headers: Optional[Dict[str, str]] = None,
) -> Tuple[EmailMessage, Dict[str, Any]]:
"""Build a MIME message and return ``(message, summary)``.
Body assembly follows RFC 2046: ``multipart/alternative`` when both text
and HTML are supplied, plain ``text/html`` for HTML-only, bare
``text/plain`` otherwise. Attachments wrap that in ``multipart/mixed``.
"""
to_list = [a for a in (str(x).strip() for x in to) if a]
cc_list = [a for a in (str(x).strip() for x in cc) if a]
bcc_list = [a for a in (str(x).strip() for x in bcc) if a]
attach_list = [a for a in (str(x).strip() for x in attachments) if a]
if not to_list:
raise MailError("at least one recipient is required")
if not text and not html:
raise MailError("either text or html body is required")
message = EmailMessage()
message["From"] = format_address(f"{sender_name} <{sender}>" if sender_name else sender)
message["To"] = ", ".join(format_address(a) for a in to_list)
if cc_list:
message["Cc"] = ", ".join(format_address(a) for a in cc_list)
if bcc_list:
message["Bcc"] = ", ".join(format_address(a) for a in bcc_list)
message["Subject"] = subject or ""
message["Date"] = date or email.utils.formatdate(localtime=True)
message["Message-ID"] = email.utils.make_msgid(domain=sender.rsplit("@", 1)[-1] or None)
if reply_to:
message["Reply-To"] = reply_to
ref_list = [r for r in (str(x).strip() for x in references) if r]
if in_reply_to:
message["In-Reply-To"] = in_reply_to
if not ref_list:
ref_list = [in_reply_to]
if ref_list:
message["References"] = " ".join(ref_list)
for key, value in (extra_headers or {}).items():
if value not in (None, "") and key not in message:
message[key] = str(value)
if text and html:
message.set_content(text, subtype="plain", charset="utf-8")
message.add_alternative(html, subtype="html", charset="utf-8")
elif html:
message.set_content(html, subtype="html", charset="utf-8")
else:
message.set_content(text, subtype="plain", charset="utf-8")
names: List[str] = []
if attach_list:
names = _attach_files(message, attach_list, attachment_paths)
summary = {
"message_id": message["Message-ID"],
"to": to_list,
"cc": cc_list,
"bcc": bcc_list,
"subject": subject or "",
"date": message["Date"],
"format": "multipart/alternative" if (text and html) else ("text/html" if html else "text/plain"),
"attachments": names,
}
return message, summary
"""SMTP delivery.
Transport mode follows the port (465 implicit TLS, 587 STARTTLS, otherwise
plain) and can be pinned per account. The password is only ever passed to
``login``; it is never logged or included in error messages.
"""
from __future__ import annotations
import smtplib
import ssl
from typing import Any, Dict, Optional
from .config import smtp_security_for_port
from .mime import MailError
def _ssl_context(verify: bool) -> ssl.SSLContext:
context = ssl.create_default_context()
if not verify:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return context
def _describe(host: str, port: int, mode: str) -> str:
return f"{host}:{port} ({mode})"
def send_message(
config: Dict[str, Any],
message: Any,
*,
recipients: Optional[list] = None,
) -> Dict[str, Any]:
"""Deliver *message*; returns a delivery summary.
``recipients`` defaults to every envelope recipient in the message (To, Cc
and Bcc) so Bcc is not silently dropped.
"""
host = str(config["smtp_host"])
port = int(config["smtp_port"])
mode = smtp_security_for_port(port, config.get("smtp_security"))
timeout = int(config.get("timeout", 30))
verify = bool(config.get("verify_ssl", True))
address = str(config["address"])
password = str(config.get("password", ""))
context = _ssl_context(verify)
targets = recipients if recipients else _envelope_recipients(message)
if not targets:
raise MailError("no recipients to deliver to")
smtp: Optional[smtplib.SMTP] = None
try:
if mode == "ssl":
smtp = smtplib.SMTP_SSL(host, port, timeout=timeout, context=context)
else:
smtp = smtplib.SMTP(host, port, timeout=timeout)
if mode == "starttls":
smtp.ehlo()
smtp.starttls(context=context)
smtp.ehlo()
# A fresh EHLO after STARTTLS is mandatory; some servers only advertise
# AUTH afterwards.
smtp.ehlo()
try:
smtp.login(address, password)
except smtplib.SMTPAuthenticationError as exc:
raise MailError(
f"SMTP authentication failed for {address} via {_describe(host, port, mode)}. "
"Most providers require an app-specific password instead of the "
f"account password, and some require SMTP authorization to be "
f"enabled first. ({exc.smtp_code} {exc.smtp_error!r})"
) from exc
except smtplib.SMTPException as exc:
raise MailError(
f"SMTP login failed for {address} via {_describe(host, port, mode)}: {exc}"
) from exc
refused = smtp.send_message(message, from_addr=str(config.get("send_as") or address),
to_addrs=targets)
if refused:
raise MailError(f"recipients refused by the server: {sorted(refused)}")
except ssl.SSLCertVerificationError as exc:
raise MailError(
f"TLS certificate verification failed for {host}:{port}. Use a valid "
"certificate, or set verify_ssl=false if you accept the risk."
) from exc
except smtplib.SMTPResponseException as exc:
raise MailError(
f"server rejected the message ({exc.smtp_code} {exc.smtp_error!r}) "
f"via {_describe(host, port, mode)}"
) from exc
except (smtplib.SMTPException, OSError) as exc:
if isinstance(exc, MailError):
raise
raise MailError(f"cannot send via {_describe(host, port, mode)}: {exc}") from exc
finally:
if smtp is not None:
try:
smtp.quit()
except Exception:
try:
smtp.close()
except Exception:
pass
return {
"sent": True,
"from": address,
"recipients": list(targets),
"transport": mode,
"server": f"{host}:{port}",
}
def _envelope_recipients(message: Any) -> list:
import email.utils
out = []
for header in ("To", "Cc", "Bcc"):
for _name, addr in email.utils.getaddresses(message.get_all(header, [])):
if addr:
out.append(addr)
seen, unique = set(), []
for addr in out:
if addr.lower() not in seen:
seen.add(addr.lower())
unique.append(addr)
return unique
#!/usr/bin/env python3
"""Self-check for the email skill.
Ships with the skill so the offline behaviour is reproducible by whoever
installs it, without a separate test tree:
python mail.py selftest # offline: parsing, MIME, servers, flags
python mail.py selftest --live # additionally connects to the mailbox
The offline half needs no network and no credentials. The live half reads the
same credentials as every other command and only performs read-only IMAP work,
so it is safe to run against a real account.
"""
from __future__ import annotations
import base64
import json
import os
import sys
import tempfile
from pathlib import Path
from typing import Any, Callable, Dict, List, Tuple
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mailkit.commands import _plan_flag_changes, _target_folder # noqa: E402
from mailkit.config import ( # noqa: E402
build_imap_id,
detect_provider,
resolve_servers,
smtp_security_for_port,
)
from mailkit.imap_client import _special_key, normalise_folder # noqa: E402
from mailkit.mime import ( # noqa: E402
build_message,
html_to_text,
parse_message,
safe_filename,
summarize_text,
unique_path,
)
def _enc(text: str, charset: str = "utf-8") -> str:
return f"=?{charset}?B?{base64.b64encode(text.encode(charset)).decode()}?="
def _checks() -> List[Tuple[str, Callable[[], Any], Any]]:
"""(name, thunk, expected) triples. Each thunk must be side-effect free."""
cases: List[Tuple[str, Callable[[], Any], Any]] = []
def add(name: str, thunk: Callable[[], Any], expected: Any) -> None:
cases.append((name, thunk, expected))
# -- header decoding ------------------------------------------------
add("header/utf8", lambda: _dec(_enc("订单 通知")), "订单 通知")
add("header/adjacent", lambda: _dec(_enc("订单") + " " + _enc("通知")), "订单通知")
add("header/mislabelled", lambda: _dec(_enc("中文", "gb2312")), "中文")
add("header/plain", lambda: _dec("plain subject"), "plain subject")
add("header/empty", lambda: _dec(None), "")
# -- HTML to text ---------------------------------------------------
add("html/drops-script", lambda: "alert" in html_to_text("<script>alert(1)</script><p>ok</p>"), False)
add("html/drops-style", lambda: "color" in html_to_text("<style>p{color:red}</style><p>ok</p>"), False)
add("html/entities", lambda: "A&B" in html_to_text("<p>A&B</p>"), True)
add("html/table-tabs", lambda: "\t" in html_to_text("<table><tr><td>a</td><td>b</td></tr></table>"), True)
add("html/list-marker", lambda: "- a" in html_to_text("<ul><li>a</li></ul>"), True)
add("html/link-href", lambda: "https://x.test/1" in html_to_text('<a href="https://x.test/1">go</a>'), True)
add("html/link-no-dup", lambda: html_to_text('<a href="https://x.test/1">https://x.test/1</a>').count("https://x.test/1"), 1)
add("html/chinese", lambda: html_to_text("<p>您的订单已发货</p>"), "您的订单已发货")
add("html/empty", lambda: html_to_text(" "), "")
add("html/preview-one-line", lambda: "\n" in summarize_text("<p>a</p><p>b</p>"), False)
# -- message parsing ------------------------------------------------
def html_only() -> Dict[str, Any]:
raw = ("From: [email protected]\r\nSubject: s\r\nMIME-Version: 1.0\r\n"
"Content-Type: text/html; charset=utf-8\r\n\r\n"
"<p>请查收附件报价单</p>").encode()
return parse_message(raw)
add("parse/html-fallback", lambda: html_only()["text_source"], "html")
add("parse/html-text", lambda: html_only()["text"], "请查收附件报价单")
add("parse/subject-decoded", lambda: html_only()["subject"], "s")
add("parse/no-html-flag", lambda: "html" in parse_message(
"From: a@b\r\nMIME-Version: 1.0\r\nContent-Type: text/html; charset=utf-8\r\n\r\n<p>x</p>".encode(),
include_html=False), False)
def multipart() -> Dict[str, Any]:
raw = ("From: [email protected]\r\nTo: [email protected]\r\nMIME-Version: 1.0\r\n"
"Content-Type: multipart/mixed; boundary=\"B\"\r\n\r\n"
"--B\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n见附件\r\n"
"--B\r\nContent-Type: application/pdf\r\n"
"Content-Disposition: attachment; filename=\"report.pdf\"\r\n"
"Content-Transfer-Encoding: base64\r\n\r\naGVsbG8=\r\n--B--\r\n").encode()
return parse_message(raw)
add("parse/attachment-name", lambda: multipart()["attachments"][0]["filename"], "report.pdf")
add("parse/attachment-size", lambda: multipart()["attachments"][0]["size"], 5)
add("parse/plain-preferred", lambda: multipart()["text_source"], "plain")
# -- message building -----------------------------------------------
def built() -> Tuple[Any, Dict[str, Any]]:
return build_message(sender="[email protected]", to=["[email protected]"], subject="会议纪要",
text="正文", html="<p>正文</p>", bcc=["[email protected]"])
add("build/subject-roundtrip", lambda: parse_message(built()[0].as_bytes())["subject"], "会议纪要")
add("build/alternative", lambda: built()[1]["format"], "multipart/alternative")
add("build/html-only", lambda: build_message(sender="[email protected]", to=["[email protected]"],
subject="s", html="<p>x</p>")[1]["format"], "text/html")
add("build/bcc-hidden", lambda: "[email protected]" in (
built()[0]["To"] + (built()[0].get("Cc") or "")), False)
add("build/needs-recipient", lambda: _raises(lambda: build_message(
sender="[email protected]", to=[], subject="s", text="t")), True)
add("build/needs-body", lambda: _raises(lambda: build_message(
sender="[email protected]", to=["[email protected]"], subject="s")), True)
add("build/missing-attachment", lambda: _raises(lambda: build_message(
sender="[email protected]", to=["[email protected]"], subject="s", text="t",
attachments=["/nonexistent/x.txt"])), True)
# -- flag planning --------------------------------------------------
add("flag/unread-clears", lambda: _plan_flag_changes(["unread"], "add"), ([], ["\\Seen"]))
add("flag/read-sets", lambda: _plan_flag_changes(["read"], "add"), (["\\Seen"], []))
add("flag/star-add", lambda: _plan_flag_changes(["star"], "add"), (["\\Flagged"], []))
add("flag/star-remove", lambda: _plan_flag_changes(["star"], "remove"), ([], ["\\Flagged"]))
add("flag/remove-inverts-state", lambda: _plan_flag_changes(["read"], "remove"), ([], ["\\Seen"]))
add("flag/last-wins", lambda: _plan_flag_changes(["read", "unread"], "add"), ([], ["\\Seen"]))
add("flag/dedupes", lambda: _plan_flag_changes(["star", "starred", "flagged"], "add"), (["\\Flagged"], []))
# -- servers --------------------------------------------------------
add("server/163", lambda: resolve_servers("[email protected]")[0][0], "imap.163.com")
add("server/126-distinct", lambda: resolve_servers("[email protected]")[0][0] != resolve_servers("[email protected]")[0][0], True)
add("server/yeah", lambda: resolve_servers("[email protected]")[0][0], "imap.yeah.net")
add("server/qq", lambda: resolve_servers("[email protected]")[1], "foxmail")
add("server/gmail", lambda: resolve_servers("[email protected]")[0][0], "imap.gmail.com")
add("server/provider-override", lambda: resolve_servers("[email protected]", "qq")[1], "qq")
add("server/host-override", lambda: resolve_servers("[email protected]", overrides={"imap_host": "imap.corp.test"})[0][0], "imap.corp.test")
add("server/detect", lambda: detect_provider("[email protected]"), "outlook")
add("server/unknown-provider", lambda: _raises(lambda: resolve_servers("[email protected]", "nope")), True)
add("server/ssl-465", lambda: smtp_security_for_port(465), "ssl")
add("server/starttls-587", lambda: smtp_security_for_port(587), "starttls")
add("server/plain-25", lambda: smtp_security_for_port(25), "none")
# -- folders --------------------------------------------------------
add("folder/drafts-plural", lambda: _special_key("Drafts"), _special_key("\\Draft"))
add("folder/sent", lambda: _special_key("Sent"), _special_key("\\Sent"))
add("folder/alias-inbox", lambda: normalise_folder("INBOX"), "INBOX")
add("folder/alias-sent", lambda: normalise_folder("sent"), "Sent")
add("folder/alias-trash", lambda: normalise_folder("deleted"), "Trash")
add("folder/target-default", lambda: _target_folder({}), "INBOX")
# -- IMAP ID --------------------------------------------------------
add("id/default", lambda: build_imap_id().count('"python-mailkit"'), 2)
add("id/custom", lambda: build_imap_id("acme").count('"acme"'), 2)
# -- body size guard ------------------------------------------------
def long_message(limit: int, include_html: bool = False) -> Dict[str, Any]:
body = "<div>" + ("内容" * 8000) + "</div>"
raw = ("From: [email protected]\r\nSubject: 长\r\nMIME-Version: 1.0\r\n"
"Content-Type: text/html; charset=utf-8\r\n\r\n" + body).encode()
return parse_message(raw, include_html=include_html, max_body_chars=limit)
add("body/defaults-to-no-html", lambda: "html" in long_message(0), False)
add("body/html-opt-in", lambda: "html" in long_message(0, True), True)
add("body/truncates", lambda: long_message(500)["truncated"], True)
add("body/keeps-length", lambda: len(long_message(500)["text"]), 500)
add("body/reports-original", lambda: long_message(500)["text_chars"] > 500, True)
add("body/zero-disables", lambda: long_message(0)["truncated"], False)
add("body/under-limit-not-truncated", lambda: long_message(999999)["truncated"], False)
# -- multi-account selection ----------------------------------------
_ACCOUNTS = {"accounts": {"work": {"address": "[email protected]", "password": "p"},
"home": {"address": "[email protected]", "password": "p"}}}
def selected(data: Any, name: str) -> Any:
"""Load *name* from a config file, keeping the temp dir alive."""
from mailkit.config import ConfigError, load_config
with tempfile.TemporaryDirectory() as tmp:
path = os.path.join(tmp, "mail.json")
with open(path, "w", encoding="utf-8") as handle:
json.dump(data, handle)
try:
return load_config(path, {"account": name})
except ConfigError as exc:
return f"ERR:{exc}"
add("account/picks-work", lambda: selected(_ACCOUNTS, "work")["imap_host"], "imap.sohu.com")
add("account/picks-home", lambda: selected(_ACCOUNTS, "home")["imap_host"], "imap.sina.com")
add("account/unknown-fails", lambda: str(selected(_ACCOUNTS, "ghost")).startswith("ERR:"), True)
add("account/error-lists-available",
lambda: "available: home, work" in str(selected(_ACCOUNTS, "ghost")), True)
add("account/missing-map-fails",
lambda: str(selected({"address": "[email protected]", "password": "p"}, "work")).startswith("ERR:"),
True)
# -- filenames ------------------------------------------------------
add("file/strips-traversal", lambda: "/" in safe_filename("../../etc/passwd"), False)
add("file/strips-backslash", lambda: "\\" in safe_filename(r"C:\x.exe"), False)
add("file/keeps-unicode", lambda: safe_filename("报表.pdf"), "报表.pdf")
add("file/blank-fallback", lambda: safe_filename(" "), "attachment")
add("file/collision", lambda: _collide("a.pdf"), "a(1).pdf")
add("file/collision-no-ext", lambda: _collide("README"), "README(1)")
return cases
def _dec(value: Any) -> str:
from mailkit.mime import decode_header_value
return decode_header_value(value)
def _raises(thunk: Callable[[], Any]) -> bool:
try:
thunk()
except Exception:
return True
return False
def _collide(name: str) -> str:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / name).write_text("x")
return unique_path(root, name).name
def run_offline() -> Tuple[int, int, List[str]]:
failures: List[str] = []
cases = _checks()
for name, thunk, expected in cases:
try:
got = thunk()
except Exception as exc: # noqa: BLE001 - report as a check failure
failures.append(f"{name}: raised {type(exc).__name__}: {exc}")
continue
if got != expected:
failures.append(f"{name}: got {got!r}, want {expected!r}")
return len(cases), len(failures), failures
def run_live() -> Tuple[int, int, List[str]]:
"""Read-only connectivity probes against the configured mailbox."""
results: List[Tuple[str, Callable[[], Any], Any]] = [
("imap-connect", _live_connect, True),
("select-inbox", _live_select, "INBOX"),
("list-folders", lambda: bool(_folders()), True),
("search", lambda: isinstance(_search(), list), True),
]
failures: List[str] = []
try:
for name, thunk, expected in results:
try:
got = thunk()
except Exception as exc: # noqa: BLE001 - report as a check failure
failures.append(f"{name}: {type(exc).__name__}: {exc}")
continue
if got != expected:
failures.append(f"{name}: got {got!r}, want {expected!r}")
finally:
if _SESSION.get("session") is not None:
_SESSION["session"].close()
return len(results), len(failures), failures
_SESSION: Dict[str, Any] = {}
def _live_connect() -> bool:
from mailkit.config import load_config
from mailkit.imap_client import ImapSession
config = load_config("", {})
_SESSION["config"] = {
"address": config["address"],
"imap_host": config["imap_host"],
"imap_port": config["imap_port"],
}
_SESSION["session"] = ImapSession(config)
_SESSION["session"].connect()
return True
def _live_select() -> Any:
return _SESSION["session"].select("INBOX")
def _folders() -> Any:
return _SESSION["session"].list_folders()
def _search() -> Any:
return _SESSION["session"].search("ALL", limit=1)
def run(live: bool = False) -> Dict[str, Any]:
"""Run the checks and return a JSON-serialisable report.
This is the single entry point; :func:`main` only adds printing.
"""
payload: Dict[str, Any] = {"ok": True, "mode": "offline+live" if live else "offline"}
total, failed, failures = run_offline()
payload["offline"] = {"total": total, "failed": failed, "failures": failures}
if live:
ltotal, lfailed, lfailures = run_live()
payload["live"] = {"total": ltotal, "failed": lfailed, "failures": lfailures}
passed = total - failed
summary = f"{passed}/{total} offline checks passed"
if live:
info = payload["live"]
summary += f", {info['total'] - info['failed']}/{info['total']} live"
payload["summary"] = summary
payload["ok"] = failed + payload.get("live", {}).get("failed", 0) == 0
return payload
def main(argv: List[str]) -> int:
report = run(live="--live" in argv)
for section in ("offline", "live"):
for name in report.get(section, {}).get("failures", []):
print(f" FAIL {section}: {name}", file=sys.stderr)
if "--json" in argv:
print(json.dumps(report, ensure_ascii=False, indent=2))
else:
print(report["summary"])
if "--live" in argv and report.get("live", {}).get("failures"):
print("\nlive failures (need a working account):", file=sys.stderr)
for name in report["live"]["failures"]:
print(f" {name}", file=sys.stderr)
return 0 if report["ok"] else 1
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))