发送中巡检限流退信：发现即中止发送

背景：限流退信是异步投到发件箱的——SMTP 当场返回 250，几十秒后收件箱
才收到「您的账号外发频率超过邮件系统限制」。只等发完再核查，等发现时
被限流的那一批早就全废了，白耗额度。

改动：
- 把 IMAP 扫描抽成 _scan_imap_bounces()，供退信核查与发送中巡检共用
  （新增 not_before 参数：只看指定时刻之后到达的退信）
- 发送过程中每 --watch-every 封（默认 10）巡检一次收件箱，命中限流类
  退信立即中止；只看本次运行开始后的新退信，历史退信不误触发
- SMTP 当场返回限流（如 450 MI:CEL）也立即中止；硬退信则照旧记入
  无效地址清单并继续发下一封
- 中止后把限流地址从断点清单剔除，等限制恢复重跑即自动补发
- IMAP 连续两轮不可用则自动关闭本次巡检并提示，不影响发送
- 新增 --no-watch-bounces / --watch-every，报告新增 aborted 等字段
- 顺带：分类依据里的状态码改为只显示匹配到的码（原来是整串原文）

README.md

@@ -180,6 +180,32 @@ conf/app.ini 路径的解析优先级：
   - --since 2026-09-07 可显式指定起点日期
   - 断点清单为空时回退为 --since-days N（默认 3 天）
 
+发送中限流巡检（边发边看，及时止损）
+----------------------------------
+限流退信是**异步**投到发件箱的——SMTP 当场返回 250，几十秒后收件箱才收到
+「您的账号外发频率超过邮件系统限制」。如果只等发完再核查，等发现时限流那一批
+早就全废了。所以正式群发时会**边发边巡检收件箱**：
+
+  - 每 `--watch-every` 封（默认 10）登录一次 IMAP 查新退信
+  - 只看**本次运行开始之后**到达的退信，历史退信不会误触发
+  - 一旦出现限流类退信 → 立即停止发送
+  - SMTP 当场就返回限流（如 `450 MI:CEL 发送频率超限`）→ 同样立即停止
+  - 善后：把限流退信的地址从断点清单剔除，等限制恢复后重跑自动补发
+  - 巡检到的「地址不存在」类退信照旧写入无效地址清单
+
+  python3 broadcast.py --send --yes                    # 默认已开启巡检
+  python3 broadcast.py --send --yes --watch-every 5    # 每 5 封查一次（更及时，IMAP 登录更频繁）
+  python3 broadcast.py --send --yes --no-watch-bounces # 关闭巡检（--to 单封测试本就关闭）
+
+中止时的输出示例：
+
+  [中止发送] 收件箱出现 1 条限流退信（如 a@b.com：关键词「您的账号外发频率超过邮件系统限制」）
+  [善后] 已将 1 个限流退信地址从断点清单剔除，等限制恢复后重跑同一条命令即可自动补发。
+  发送中止：已成功 30 封，失败 0 封，未发送 30 封。
+
+注意：巡检依赖 IMAP 可用。若连续两轮连不上 IMAP，会自动关闭本次巡检并提示
+（不影响发送），此时请发完手动跑一次 `--check-bounces`。
+
 无效地址清单 broadcast_invalid.json
 -----------------------------------
 硬退信（地址不存在等）的地址会写进这里，之后**每次运行都直接跳过**，

broadcast.py

@@ -32,6 +32,9 @@ TamaBox 站内信群发工具（独立项目，零依赖，不依赖、不修改
   - SMTP 拒收捕获：send_message 返回的 refused 计入失败
   - 退信分类核查：只有「限流/临时性」退信才剔除补发；「地址不存在」这类
     永久失败不补发，并记入无效地址清单（broadcast_invalid.json）永久跳过
+  - 发送中限流巡检：每 --watch-every 封查一次收件箱，一旦出现
+    「外发频率超过邮件系统限制」类退信（或 SMTP 当场报限流）立即中止发送，
+    并把这批限流地址从断点清单剔除，等限制恢复后重跑自动补发
 
 用法示例：
   # 1) 演练：列出收件人 + 统计（不发信）
@@ -73,7 +76,7 @@ import time
 from email.header import Header, decode_header
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
-from email.utils import formataddr, make_msgid
+from email.utils import formataddr, make_msgid, parsedate_to_datetime
 from urllib.parse import quote_plus
 
 SCRIPT_VERSION = "2026-09-07.bounceclass.v3"
@@ -871,10 +874,9 @@ def classify_bounce(status="", diagnostic="", text=""):
     """
     for src in (status, diagnostic):
         c = _smtp_code_class(src or "")
-        if c == "5":
+        if c in ("5", "4"):
-            return CAT_HARD, f"DSN 状态码 {(src or '').strip()[:40]}"
+            m = _SMTP_ENHANCED_RE.search(src or "") or _SMTP_CODE_RE.search(src or "")
-        if c == "4":
+            return (CAT_HARD if c == "5" else CAT_RATE), f"SMTP 状态码 {m.group(0)}"
-            return CAT_RATE, f"DSN 状态码 {(src or '').strip()[:40]}"
 
     body = "\n".join(x for x in (text, diagnostic) if x)
     low = body.lower()
@@ -1061,6 +1063,23 @@ def reset_invalid(path):
     return False
 
 
+def watch_bounces(cfg, args, not_before, known_ids, verbose=True):
+    """发送过程中的轻量巡检：只看 not_before 之后新到的退信。
+
+    不写任何文件、不动断点清单（善后交给 --check-bounces）。
+    返回 (bounced, ok)：ok=False 表示 IMAP 不可用，调用方应降级处理。
+    """
+    try:
+        res = _scan_imap_bounces(cfg, args, not_before, skip_ids=known_ids,
+                                 not_before=not_before, verbose=False)
+    except Exception as e:  # noqa: BLE001
+        if verbose:
+            print(f"  [退信巡检] 本轮跳过：IMAP 检查失败（{e}）")
+        return {}, False
+    known_ids.update(res["touched_ids"])
+    return res["bounced"], True
+
+
 def resolve_scan_start(args):
     """退信扫描起点：--since 显式日期 > auto 时取断点清单最早一条发送记录的
     时间（精确到该时刻，不加余量）> 无记录时回退 --since-days。
@@ -1082,52 +1101,64 @@ def resolve_scan_start(args):
             f"断点清单为空，回退扫描近 {args.since_days} 天")
 
 
-def check_bounces(cfg, args):
+def _parse_mail_date(value):
-    """登录发件邮箱 IMAP，扫描近 N 天的退信并解析失败收件人。
+    """解析邮件 Date 头为本地 naive datetime；解析不了返回 None。"""
+    try:
+        dt = parsedate_to_datetime(value)
+    except Exception:  # noqa: BLE001
+        return None
+    if dt is None:
+        return None
+    if dt.tzinfo is not None:
+        dt = dt.astimezone().replace(tzinfo=None)
+    return dt
 
-    返回 {email: {"subject","date","category","reason","detail"}}；已处理过的
+
-    邮件（Message-ID）记录在 broadcast_bounces_seen.json，重复核查不会重复报告。
+def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None,
-    category 见 CAT_RATE / CAT_HARD / CAT_UNKNOWN。"""
+                       start_desc="", verbose=True):
+    """连 IMAP 扫描退信并逐收件人分类（check_bounces 与发送中巡检共用）。
+
+    skip_ids   已处理过的 Message-ID（seen 文件 / 本次运行内已见过），跳过不重复报
+    not_before 只统计 Date 头晚于该时刻的退信（发送中巡检用：只看本次运行之后
+               新到的退信，避免拿历史退信误触发中止）；Date 解析不了则放行
+    返回 dict：bounced / new_ids / scanned / touched_ids
+      bounced     {email: {"subject","date","category","reason","detail"}}
+      new_ids     本次计入的退信 Message-ID（是否落盘由调用方决定）
+      touched_ids 本次扫到的所有退信 Message-ID（含被 not_before 过滤掉的），
+                  供巡检做运行期去重，避免每轮重复拉取同一封全文
+    """
     smtp_host = cfg["mail_smtp"] or ""
     imap_host = (args.imap_host or "").strip()
     if not imap_host:
         imap_host = smtp_host.replace("smtp.", "imap.", 1) if smtp_host else ""
     if not imap_host:
-        sys.exit("[FATAL] 无法推导 IMAP 服务器地址，请用 --imap-host 指定")
+        raise RuntimeError("无法推导 IMAP 服务器地址，请用 --imap-host 指定")
-    account = cfg["mail_account"]
+    account, password = cfg["mail_account"], cfg["mail_password"]
-    password = cfg["mail_password"]
-    start_dt, start_desc = resolve_scan_start(args)
     since = start_dt.strftime("%d-%b-%Y")
 
+    if verbose:
         print(f"  [IMAP] {imap_host}:{args.imap_port}  账号 {account}")
-    print(f"  [扫描起点] {start_desc} → 自 {since} 起的邮件")
+        print(f"  [扫描起点] {start_desc or start_dt.strftime('%Y-%m-%d %H:%M')}"
-    try:
+              f" → 自 {since} 起的邮件")
+
     conn = imaplib.IMAP4_SSL(imap_host, args.imap_port)
+    try:
         conn.login(account, password)
         conn.select("INBOX", readonly=True)
-    except Exception as e:  # noqa: BLE001
-        sys.exit(f"[FATAL] IMAP 登录失败（请确认邮箱已开启 IMAP，密码用邮箱登录密码）: {e}")
-
         typ, data = conn.search(None, f'(SINCE "{since}")')
         if typ != "OK":
-        sys.exit("[FATAL] IMAP search 失败")
+            raise RuntimeError("IMAP search 失败")
         ids = data[0].split()
+        if verbose:
             print(f"  [IMAP] 扫描范围内共 {len(ids)} 封待扫描")
 
-    seen = set()
+        skip_ids = skip_ids or set()
-    if os.path.isfile(DEFAULT_BOUNCE_SEEN):
+        bounced, new_ids, touched, scanned = {}, [], [], 0
-        try:
-            with open(DEFAULT_BOUNCE_SEEN, "r", encoding="utf-8") as f:
-                seen = set(json.load(f))
-        except Exception:  # noqa: BLE001
-            seen = set()
-
-    bounced = {}   # email -> {"subject","date","category","reason","detail"}
-    scanned, bounce_cnt, newly_seen = 0, 0, []
         for i, mid in enumerate(ids, 1):
             uid = mid.decode() if isinstance(mid, bytes) else str(mid)
             # 两段式：先取头部（省流量），命中退信特征再取全文
-        typ, hdata = conn.fetch(mid, "(BODY.PEEK[HEADER.FIELDS (SUBJECT MESSAGE-ID FROM CONTENT-TYPE)])")
+            typ, hdata = conn.fetch(
+                mid, "(BODY.PEEK[HEADER.FIELDS (SUBJECT MESSAGE-ID FROM CONTENT-TYPE)])")
             if typ != "OK" or not hdata or hdata[0] is None:
                 continue
             try:
@@ -1135,7 +1166,7 @@ def check_bounces(cfg, args):
             except Exception:  # noqa: BLE001
                 continue
             msg_id = (head.get("Message-ID") or "").strip() or f"uid:{uid}"
-        if msg_id in seen:
+            if msg_id in skip_ids:
                 continue
             scanned += 1
             if not is_bounce_message(head):
@@ -1147,29 +1178,59 @@ def check_bounces(cfg, args):
                 full = email.message_from_bytes(fdata[0][1])
             except Exception:  # noqa: BLE001
                 continue
+            touched.append(msg_id)
+            if not_before is not None:
+                arrived = _parse_mail_date(full.get("Date", ""))
+                if arrived is not None and arrived < not_before:
+                    continue
             subject = _decode_hdr(full.get("Subject", "")) or "(无主题)"
             date = _decode_hdr(full.get("Date", ""))
             rcpts = extract_bounced_recipients(full, account)
-        bounce_cnt += 1
+            new_ids.append(msg_id)
-        newly_seen.append(msg_id)
             for r in rcpts:
                 bounced.setdefault(r["email"], {
                     "subject": subject, "date": date,
                     "category": r["category"], "reason": r["reason"], "detail": r["detail"],
                 })
-        if rcpts:
+            if verbose:
                 desc = "  ".join(
-                f"{r['email']}[{CAT_LABEL.get(r['category'], r['category'])}]" for r in rcpts
+                    f"{r['email']}[{CAT_LABEL.get(r['category'], r['category'])}]"
-            )
+                    for r in rcpts) if rcpts else "(未解析出收件人)"
-        else:
-            desc = "(未解析出收件人)"
                 print(f"  [退信 {i}/{len(ids)}] {subject[:40]}  →  {desc}")
-
+        return {"bounced": bounced, "new_ids": new_ids,
+                "scanned": scanned, "touched_ids": touched}
+    finally:
         try:
             conn.logout()
         except Exception:  # noqa: BLE001
             pass
 
+
+def check_bounces(cfg, args):
+    """登录发件邮箱 IMAP，扫描近 N 天的退信并解析失败收件人。
+
+    返回 {email: {"subject","date","category","reason","detail"}}；已处理过的
+    邮件（Message-ID）记录在 broadcast_bounces_seen.json，重复核查不会重复报告。
+    category 见 CAT_RATE / CAT_HARD / CAT_UNKNOWN。"""
+    start_dt, start_desc = resolve_scan_start(args)
+
+    seen = set()
+    if os.path.isfile(DEFAULT_BOUNCE_SEEN):
+        try:
+            with open(DEFAULT_BOUNCE_SEEN, "r", encoding="utf-8") as f:
+                seen = set(json.load(f))
+        except Exception:  # noqa: BLE001
+            seen = set()
+
+    try:
+        res = _scan_imap_bounces(cfg, args, start_dt, skip_ids=seen, start_desc=start_desc)
+    except Exception as e:  # noqa: BLE001
+        sys.exit(f"[FATAL] IMAP 扫描失败（请确认邮箱已开启 IMAP，密码用邮箱登录密码）: {e}")
+
+    bounced = res["bounced"]
+    newly_seen, scanned = res["new_ids"], res["scanned"]
+    bounce_cnt = len(newly_seen)
+
     # ---- 按类别分组处理：只有「可重试」的才剔除补发 ----
     groups = {CAT_RATE: [], CAT_HARD: [], CAT_UNKNOWN: []}
     for em, info in bounced.items():
@@ -1548,6 +1609,13 @@ def main():
                         help="本次不使用无效地址清单：既不跳过已知无效地址，也不写入新的硬退信")
     parser.add_argument("--bounce-report", default=DEFAULT_BOUNCE_REPORT, metavar="FILE",
                         help="退信核查明细报告路径（JSON）；默认脚本同目录 broadcast_bounce_report.json")
+    parser.add_argument("--watch-bounces", dest="watch_bounces", action="store_true", default=True,
+                        help="发送过程中巡检收件箱：每 --watch-every 封查一次，"
+                             "一旦出现「外发频率超过邮件系统限制」等限流退信立即中止发送（默认开启）")
+    parser.add_argument("--no-watch-bounces", dest="watch_bounces", action="store_false",
+                        help="关闭发送中巡检（--to 单封测试时本就关闭）")
+    parser.add_argument("--watch-every", type=int, default=10, metavar="N",
+                        help="发送中每发 N 封巡检一次退信（默认 10；调小会更频繁登录 IMAP）")
     parser.add_argument("--imap-host", default="",
                         help="IMAP 服务器地址（默认由 SMTP 域名推导：smtp.xxx → imap.xxx）")
     parser.add_argument("--imap-port", type=int, default=993,
@@ -1825,6 +1893,9 @@ def main():
         "smtp_rejected": 0,
         "skipped_already_sent": skipped,
         "skipped_invalid": invalid_skipped,
+        "aborted": None,
+        "abort_rate_bounces": [],
+        "watch_bounces": bool(getattr(args, "watch_bounces", True)) and not args.to,
         "failed_list": [],
         "missing_placeholders": [],
         "test_mode": bool(args.to),
@@ -1882,11 +1953,21 @@ def main():
 
     sent, failed = 0, []
     smtp_rejected = 0
+    attempted = 0
     total = len(recipients)
     all_missing = set()
+
+    # 发送中巡检：限流退信是异步投到发件箱的，边发边看才能及时止损
+    watch_active = bool(getattr(args, "watch_bounces", True)) and not args.to
+    watch_start = datetime.now()
+    watch_every = max(1, getattr(args, "watch_every", 10) or 10)
+    known_ids, watch_fails = set(), 0
+    watch_rate, watch_hard, abort_reason = {}, {}, ""
+
     print(f"\n[发送中] 目标 {total} 封，每封间隔 {args.delay}s"
           + (f"，每组间隔 {args.group_pause}s" if args.group_pause > 0 else "")
-          + (f"，每 {args.pause_every} 封暂停 {args.pause_for}s" if args.pause_every > 0 else "") + " ...")
+          + (f"，每 {args.pause_every} 封暂停 {args.pause_for}s" if args.pause_every > 0 else "")
+          + (f"，每 {watch_every} 封巡检一次退信" if watch_active else "") + " ...")
 
     if args.single:
         # 单文件模式：不按语言分组，所有人同一模板
@@ -1932,10 +2013,12 @@ def main():
                 msg, missing = build_message(cfg, r["email"], ctx, subject, html, plain)
                 if missing:
                     all_missing.update(missing)
+                smtp_err = ""
                 try:
                     refused = server.send_message(msg)
                     if refused:
                         refused_desc = "; ".join(f"{a} {c}:{err}" for a, (c, err) in refused.items())
+                        smtp_err = refused_desc
                         failed.append((r["email"], f"SMTP 拒收 {refused_desc}"))
                         smtp_rejected += 1
                         print(f"  [SMTP 拒收] {r['email']}: {refused_desc}")
@@ -1946,8 +2029,49 @@ def main():
                         if sent % 25 == 0 or sent == total:
                             print(f"  进度 {sent}/{total}  已成功 {sent}")
                 except Exception as e:  # noqa: BLE001
-                    failed.append((r["email"], str(e)))
+                    smtp_err = str(e)
+                    failed.append((r["email"], smtp_err))
                     print(f"  [失败] {r['email']}: {e}")
+
+                attempted += 1
+                # SMTP 当场就返回限流信号 → 继续发只会更糟，立即中止
+                if smtp_err:
+                    cat, why = classify_bounce("", smtp_err, "")
+                    if cat == CAT_HARD and not args.no_invalid_list:
+                        record_invalid(args.invalid_file, {r["email"]: {
+                            "reason": why, "detail": smtp_err[:160], "subject": subject,
+                            "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}})
+                    elif cat == CAT_RATE and not abort_reason:
+                        abort_reason = f"SMTP 返回疑似限流（{why}）：{smtp_err[:100]}"
+
+                # 周期巡检收件箱：异步退信才是限流最常见的表现
+                if watch_active and not abort_reason and attempted % watch_every == 0:
+                    nb, ok = watch_bounces(cfg, args, watch_start, known_ids)
+                    if not ok:
+                        watch_fails += 1
+                        if watch_fails >= 2:
+                            print("  [退信巡检] IMAP 连续不可用，本次已关闭巡检"
+                                  "（不影响发送；发完请手动跑一次 --check-bounces）")
+                            watch_active = False
+                    else:
+                        watch_fails = 0
+                        for em, info in nb.items():
+                            if info["category"] == CAT_RATE:
+                                watch_rate[em] = info
+                            elif info["category"] == CAT_HARD:
+                                watch_hard[em] = info
+                        if watch_hard:
+                            if not args.no_invalid_list:
+                                record_invalid(args.invalid_file, watch_hard)
+                            watch_hard = {}
+                        if watch_rate:
+                            abort_reason = (
+                                f"收件箱出现 {len(watch_rate)} 条限流退信"
+                                f"（如 {sorted(watch_rate)[0]}："
+                                f"{watch_rate[sorted(watch_rate)[0]]['reason']}）")
+
+                if abort_reason:
+                    break
                 if sent < total:
                     time.sleep(max(0.0, args.delay))
                     if args.pause_every > 0 and sent % args.pause_every == 0:
@@ -1956,6 +2080,8 @@ def main():
             if args.group_pause > 0 and gi < len(group_order):
                 print(f"  [组间间隔] 下一组前暂停 {args.group_pause}s ...")
                 time.sleep(max(0.0, args.group_pause))
+            if abort_reason:
+                break
     finally:
         try:
             server.quit()
@@ -1965,11 +2091,28 @@ def main():
     report["sent"] = sent
     report["failed"] = len(failed)
     report["smtp_rejected"] = smtp_rejected
+    report["aborted"] = abort_reason or None
+    report["abort_rate_bounces"] = sorted(watch_rate)
+    report["watch_bounces"] = watch_active
     report["failed_list"] = [{"email": e, "error": err} for e, err in failed]
     report["missing_placeholders"] = sorted(all_missing)
     write_report_now()
 
     print("\n" + "=" * 56)
+    if abort_reason:
+        print("!" * 56)
+        print(f"[中止发送] {abort_reason}")
+        print("继续硬发只会持续被限流，且这批邮件多半根本进不了收件箱，故已停手。")
+        if watch_rate and not args.to:
+            n = prune_state(args.state, list(watch_rate))
+            print(f"[善后] 已将 {n} 个限流退信地址从断点清单剔除（{args.state}），"
+                  f"等限制恢复后重跑同一条命令即可自动补发。")
+        print("建议：等限流窗口过去（通常几十分钟到数小时）再重跑；"
+              "重跑前可先 --limit 20 小批量试探。")
+        print("!" * 56)
+        print(f"发送中止：已成功 {sent} 封，失败 {len(failed)} 封，"
+              f"未发送 {max(0, total - sent - len(failed))} 封。")
+    else:
         print(f"发送完成：成功 {sent} 封，失败 {len(failed)} 封（其中 SMTP 拒收 {smtp_rejected} 封）。")
     if all_missing:
         print(f"[注意] 模板中有未提供值的占位符（已替换为空串）：{', '.join(sorted(all_missing))}")