合并信封批量发送 --batch-size：同内容多收件人一封多投，452 自动拆批

- 渲染后内容 sha256 分桶，仅完全相同的收件人合并信封；含 {{name}} 等
  个人化占位符时自动回退逐封，不会错合
- 一个信封 = 1×MAIL FROM + N×RCPT TO + 1×DATA：发信次数从「人数」降到
  「信封数」，显著降低触发「外发频率超过限制」类限流的概率
- To: 头只显示站点名+发件邮箱，真实收件人走 RCPT TO 信封互相不可见
- 452/too many recipients 特判：不算限流不中止，batch_size 砍半重试
- refused dict 逐人处理：5xx 记无效清单、限流码仍触发立即中止
- SmtpSession.send_to 统一重连重试；断点续发按人记录；巡检按收件人计数
- 顺带修复：交互模式保存 params.ini 时会抹掉手工加的 smtp_idle_reconnect
- 默认 0（逐封）不变，--batch-size N 或 params.ini [send] batch_size 显式开启

README.md

@@ -41,6 +41,20 @@ README — TamaBox 站内信群发工具（mail-broadcast）
   to          =                           ; 可选：默认测试收件邮箱
   delay       = 1.0
   group_pause = 3.0
+  smtp_idle_reconnect = 30                  ; SMTP 空闲超时秒数，超时就重连
+  batch_size  = 0                         ; 合并信封：内容相同的收件人每 N 人一封（0=逐封）
+
+  [bounce]                                ; 退信核查 / 发送中限流巡检（见下节）
+  watch_bounces  = 1                      ; 发送中巡检限流退信，发现即中止
+  watch_every    = 10                     ; 每发 N 封巡检一次
+  prune_unknown  = 0                      ; 未分类退信是否也按可重试补发
+  no_invalid_list= 0                      ; 停用无效地址清单
+  invalid_file   =                        ; 留空用默认 broadcast_invalid.json
+  bounce_report  =                        ; 留空用默认
+  imap_host      =                        ; 留空按 SMTP 域名推导
+  imap_port      = 993
+  since          = auto                   ; auto = 从断点清单最早记录开始扫
+  since_days     = 3
 
 conf/app.ini 路径的解析优先级：
   命令行 -c > params.ini [path] config > 环境变量 TAMABOX_CONFIG_PATH > ./conf/app.ini
@@ -48,6 +62,11 @@ conf/app.ini 路径的解析优先级：
 安全边界：params.ini 不提供 --send / --yes——群发只能在命令行显式指定
 （或走交互模式在会话中确认），防止改配置文件时误发全量邮件。
 
+优先级：**命令行参数 > params.ini > 内置默认**。所有 [bounce] 项都有同名命令行
+参数可临时覆盖（如 `--watch-every 5`、`--no-watch-bounces`）。布尔值写
+`1/0`、`yes/no`、`true/false`、`on/off` 均可，留空或写错会忽略并回退默认
+（控制台会给出警告）。
+
 交互模式（默认）
 --------
 直接运行 `python3 broadcast.py` 即进入交互模式（无需加任何参数）；
@@ -180,6 +199,42 @@ conf/app.ini 路径的解析优先级：
   - --since 2026-09-07 可显式指定起点日期
   - 断点清单为空时回退为 --since-days N（默认 3 天）
 
+发送中限流巡检（边发边看，及时止损）
+----------------------------------
+限流退信是**异步**投到发件箱的——SMTP 当场返回 250，几十秒后收件箱才收到
+「您的账号外发频率超过邮件系统限制」。如果只等发完再核查，等发现时限流那一批
+早就全废了。所以正式群发时会**边发边巡检收件箱**：
+
+  - 每 `--watch-every` 封（默认 10）登录一次 IMAP 查新退信
+  - 只看**本次运行开始之后**到达的退信，历史退信不会误触发
+  - 一旦出现限流类退信 → 立即停止发送
+  - SMTP 当场就返回限流（如 `450 MI:CEL 发送频率超限`）→ 同样立即停止
+  - 善后：把限流退信的地址从断点清单剔除，等限制恢复后重跑自动补发
+  - 巡检到的「地址不存在」类退信照旧写入无效地址清单
+
+这两个开关**优先写在 params.ini 的 [bounce] 小节**（长期生效），命令行只用来
+临时覆盖：
+
+  # params.ini
+  [bounce]
+  watch_bounces = 1      ; 0 关闭巡检
+  watch_every   = 10     ; 每 10 封查一次
+
+  # 命令行临时覆盖（优先级更高，只影响这一次）
+  python3 broadcast.py --send --yes --watch-every 5
+  python3 broadcast.py --send --yes --no-watch-bounces
+
+（--to 单封测试时本就不巡检）
+
+中止时的输出示例：
+
+  [中止发送] 收件箱出现 1 条限流退信（如 a@b.com：关键词「您的账号外发频率超过邮件系统限制」）
+  [善后] 已将 1 个限流退信地址从断点清单剔除，等限制恢复后重跑同一条命令即可自动补发。
+  发送中止：已成功 30 封，失败 0 封，未发送 30 封。
+
+注意：巡检依赖 IMAP 可用。若连续两轮连不上 IMAP，会自动关闭本次巡检并提示
+（不影响发送），此时请发完手动跑一次 `--check-bounces`。
+
 无效地址清单 broadcast_invalid.json
 -----------------------------------
 硬退信（地址不存在等）的地址会写进这里，之后**每次运行都直接跳过**，
@@ -204,6 +259,10 @@ conf/app.ini 路径的解析优先级：
   - 发送前打印数据库统计与语言分布，人工核对
   - 每封间隔 --delay 秒（默认 1.0）；--pause-every/--pause-for 防限流
   - 按语言分群发送，组间 --group-pause 秒（默认 3.0）
+  - SMTP 断线自动重连：服务器会掐掉空闲连接，--delay 调大（如 60s/封）时
+    必现「Server not connected / please run connect() first」，整批失败。
+    连接空闲超过 `smtp_idle_reconnect` 秒（默认 30，0 关闭）就主动重连；
+    仍遇到断线则立即重试最多 3 次（间隔 2s/4s），失败才会记为该收件人失败
   - SMTP 拒收（refused）计入失败并写入报告
   - From/Subject 头自动做 RFC2047 编码（中文显示名不会被 QQ 邮箱 550 拒收）
   - 自动定位 users 表所在 schema（避免 psql 命中别的同名空表）

broadcast.py

@@ -32,6 +32,12 @@ TamaBox 站内信群发工具（独立项目，零依赖，不依赖、不修改
   - SMTP 拒收捕获：send_message 返回的 refused 计入失败
   - 退信分类核查：只有「限流/临时性」退信才剔除补发；「地址不存在」这类
     永久失败不补发，并记入无效地址清单（broadcast_invalid.json）永久跳过
+  - 发送中限流巡检：每 N 封查一次收件箱，一旦出现
+    「外发频率超过邮件系统限制」类退信（或 SMTP 当场报限流）立即中止发送，
+    并把这批限流地址从断点清单剔除，等限制恢复后重跑自动补发
+
+参数优先级：命令行 > params.ini > 内置默认。退信/巡检相关开关（[bounce] 小节）
+与站点、限速等一样都能在 params.ini 里长期配置，命令行只在需要临时覆盖时用。
 
 用法示例：
   # 1) 演练：列出收件人 + 统计（不发信）
@@ -56,6 +62,7 @@ TamaBox 站内信群发工具（独立项目，零依赖，不依赖、不修改
 import argparse
 import configparser
 import email.utils
+import hashlib
 import imaplib
 import json
 import os
@@ -73,10 +80,10 @@ 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"
+SCRIPT_VERSION = "2026-09-07.batchenv.v4"
 
 DEFAULT_STATE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "broadcast_state.json")
 DEFAULT_REPORT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "broadcast_report.json")
@@ -593,24 +600,43 @@ def resolve_lang(lang_field, default_lang, available):
     return lang
 
 
-def build_message(cfg, to_addr, ctx, subject, html, plain):
+def render_content(subject, html, plain, ctx):
-    subject, miss1 = render_strict(subject, ctx)
+    """渲染 subject/html/plain 三段；返回 (渲染结果三元组, 缺失占位符列表)。
-    html, miss2 = render_strict(html, ctx)
-    plain, miss3 = render_strict(plain, ctx)
 
+    拆出来是为了批量发送：先渲染、按内容指纹分桶，内容完全相同的收件人
+    才共用一个 SMTP 信封；逐封路径与批量路径共用同一份渲染结果。
+    """
+    subject, m1 = render_strict(subject, ctx)
+    html, m2 = render_strict(html, ctx)
+    plain, m3 = render_strict(plain, ctx)
+    return subject, html, plain, sorted(set(m1 + m2 + m3))
+
+
+def build_message_from(cfg, to_header, subject, html, plain):
+    """用已渲染好的内容构造 MIME 邮件（不再做占位符替换）。
+
+    to_header 只影响收件人看到的「收件人」显示，与实际投递无关：
+    批量信封的真实收件人走 SMTP 信封（RCPT TO），互相不可见。
+    """
     msg = MIMEMultipart("alternative")
     from_user = cfg.get("app_title") or ""
     from_addr = cfg["mail_account"]
     # RFC5322/RFC2047：非 ASCII 显示名必须先编码再放进 From，
     # 否则 QQ 邮箱等会以 "From header is missing or invalid" 550 拒收。
     msg["From"] = formataddr((str(Header(from_user, "utf-8")), from_addr))
-    msg["To"] = to_addr
+    msg["To"] = to_header
     msg["Subject"] = Header(subject, "utf-8")
     msg["Date"] = email.utils.formatdate(localtime=True)
     msg["Message-ID"] = make_msgid(domain=from_addr.split("@")[-1])
     msg.attach(MIMEText(plain, "plain", "utf-8"))
     msg.attach(MIMEText(html, "html", "utf-8"))
-    return msg, sorted(set(miss1 + miss2 + miss3))
+    return msg
+
+
+def build_message(cfg, to_addr, ctx, subject, html, plain):
+    subject, html, plain, missing = render_content(subject, html, plain, ctx)
+    msg = build_message_from(cfg, to_addr, subject, html, plain)
+    return msg, missing
 
 
 def smtp_login(cfg):
@@ -631,6 +657,118 @@ def smtp_login(cfg):
     return server
 
 
+# 断线重连需要重试的异常：连接被服务器掐掉 / 压根没连上 / 网络层故障。
+# 注意 SMTP 拒收（SMTPRecipientsRefused / SMTPDataError 等）不在其中——
+# 那是服务器明确回了话，重试同样会失败，还会重复投递。
+SMTP_TRANSIENT_ERRORS = (smtplib.SMTPServerDisconnected, smtplib.SMTPConnectError, OSError)
+
+
+class SmtpSession:
+    """维持一条 SMTP 连接，断线自动重连。
+
+    为什么必须有：服务器会掐掉空闲连接（QQ/163/阿里云常见 5~15 分钟，
+    有的更短）。群发一旦把 --delay 调大（比如 60s/封），两封之间连接空闲
+    整整一分钟，第一封就报「Server not connected」，之后每封都是
+    「please run connect() first」——脚本此前只在开头登录一次，从不重连。
+
+    策略：
+      - 发送前若距上次活动超过 idle_reconnect 秒，主动断开重连（省得每次
+        都先吃一发失败）
+      - 仍遇到断线类异常则立即重连重试，最多 max_retries 次（重试间隔
+        2s/4s，给服务器喘息）
+    """
+
+    def __init__(self, cfg, max_retries=3, idle_reconnect=30):
+        self.cfg = cfg
+        self.max_retries = max(1, max_retries)
+        self.idle_reconnect = max(0, idle_reconnect)
+        self.server = None
+        self.last_used = 0.0
+        self.reconnects = 0
+        self.server = self._connect()   # 启动时就连，账号/网络问题第一时间暴露
+
+    def _connect(self):
+        self.server = smtp_login(self.cfg)
+        self.last_used = time.time()
+        return self.server
+
+    def close(self):
+        if self.server is not None:
+            try:
+                self.server.quit()
+            except Exception:  # noqa: BLE001
+                pass
+            self.server = None
+
+    def send_to(self, msg, rcpt_addrs=None):
+        """带重连的发送；rcpt_addrs 给定时做批量信封（1×MAIL FROM + N×RCPT TO + 1×DATA）。
+
+        - rcpt_addrs=None：按邮件头解析收件人（等价旧的逐封 send_message）
+        - rcpt_addrs=[...]：信封收件人列表，To: 头仅作显示（互相不可见）
+        返回 refused dict {地址: (码, 错误)}。整信封全部被拒时
+        SMTPRecipientsRefused 也转为 dict 返回——那是服务器明确回话，
+        不属于连接故障，不重试。连接类异常仍走重连重试，耗尽后抛原始异常。
+        """
+        if (self.idle_reconnect and self.server is not None
+                and time.time() - self.last_used > self.idle_reconnect):
+            self.close()
+        last_err = None
+        for attempt in range(1, self.max_retries + 1):
+            try:
+                if self.server is None:
+                    self.reconnects += 1
+                    print(f"  [重连] SMTP 连接已断开，正在重新登录"
+                          f"（第 {self.reconnects} 次）...")
+                    self._connect()
+                if rcpt_addrs is None:
+                    refused = self.server.send_message(msg)
+                else:
+                    refused = self.server.send_message(
+                        msg, from_addr=self.cfg["mail_account"],
+                        to_addrs=list(rcpt_addrs))
+                self.last_used = time.time()
+                return refused
+            except smtplib.SMTPRecipientsRefused as e:
+                self.last_used = time.time()
+                return dict(e.recipients or {})
+            except SMTP_TRANSIENT_ERRORS as e:  # noqa: PERF203
+                last_err = e
+                self.close()
+                if attempt < self.max_retries:
+                    wait = 2 * attempt
+                    print(f"  [重试] SMTP 连接异常（{e}），{wait}s 后重试"
+                          f"（{attempt}/{self.max_retries - 1}）")
+                    time.sleep(wait)
+        raise last_err
+
+    def send_message(self, msg):
+        """逐封发送（兼容旧调用），委托给 send_to。"""
+        return self.send_to(msg, None)
+
+
+def _is_too_many_rcpts(err):
+    """判断拒收是否为「单封收件人数超限」类（452 / too many recipients）。
+
+    这类拒绝可以砍小批次重试解决，不能按限流中止、也不能按硬退信处理。
+    """
+    low = str(err).strip().lower()
+    if low.startswith("452"):
+        return True
+    if "too many recipients" in low:
+        return True
+    if "收件人" in low and ("上限" in low or "超" in low or "过多" in low):
+        return True
+    return False
+
+
+def _fmt_smtp_result(item):
+    """把 smtplib refused dict 的值 (code, msg_bytes) 规整成可读字符串。"""
+    code, err = item
+    if isinstance(err, bytes):
+        err = err.decode("utf-8", "replace").strip()
+    return f"{code} {err}".strip()
+
+
 # ---------------------------------------------------------------------------
 # 断点续发状态（JSONL 追加写，崩溃安全）
 # ---------------------------------------------------------------------------
@@ -871,10 +1009,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 +1198,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,37 +1236,118 @@ def resolve_scan_start(args):
             f"断点清单为空，回退扫描近 {args.since_days} 天")
 
 
+def _parse_mail_date(value):
+    """解析邮件 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
+
+
+def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None,
+                       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:
+        raise RuntimeError("无法推导 IMAP 服务器地址，请用 --imap-host 指定")
+    account, password = cfg["mail_account"], cfg["mail_password"]
+    since = start_dt.strftime("%d-%b-%Y")
+
+    if verbose:
+        print(f"  [IMAP] {imap_host}:{args.imap_port}  账号 {account}")
+        print(f"  [扫描起点] {start_desc or start_dt.strftime('%Y-%m-%d %H:%M')}"
+              f" → 自 {since} 起的邮件")
+
+    conn = imaplib.IMAP4_SSL(imap_host, args.imap_port)
+    try:
+        conn.login(account, password)
+        conn.select("INBOX", readonly=True)
+        typ, data = conn.search(None, f'(SINCE "{since}")')
+        if typ != "OK":
+            raise RuntimeError("IMAP search 失败")
+        ids = data[0].split()
+        if verbose:
+            print(f"  [IMAP] 扫描范围内共 {len(ids)} 封待扫描")
+
+        skip_ids = skip_ids or set()
+        bounced, new_ids, touched, scanned = {}, [], [], 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)])")
+            if typ != "OK" or not hdata or hdata[0] is None:
+                continue
+            try:
+                head = email.message_from_bytes(hdata[0][1])
+            except Exception:  # noqa: BLE001
+                continue
+            msg_id = (head.get("Message-ID") or "").strip() or f"uid:{uid}"
+            if msg_id in skip_ids:
+                continue
+            scanned += 1
+            if not is_bounce_message(head):
+                continue
+            typ, fdata = conn.fetch(mid, "(RFC822)")
+            if typ != "OK" or not fdata or fdata[0] is None:
+                continue
+            try:
+                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)
+            new_ids.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 verbose:
+                desc = "  ".join(
+                    f"{r['email']}[{CAT_LABEL.get(r['category'], r['category'])}]"
+                    for r in rcpts) if rcpts else "(未解析出收件人)"
+                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。"""
-    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 指定")
-    account = cfg["mail_account"]
-    password = cfg["mail_password"]
     start_dt, start_desc = resolve_scan_start(args)
-    since = start_dt.strftime("%d-%b-%Y")
-
-    print(f"  [IMAP] {imap_host}:{args.imap_port}  账号 {account}")
-    print(f"  [扫描起点] {start_desc} → 自 {since} 起的邮件")
-    try:
-        conn = imaplib.IMAP4_SSL(imap_host, args.imap_port)
-        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 失败")
-    ids = data[0].split()
-    print(f"  [IMAP] 扫描范围内共 {len(ids)} 封待扫描")
 
     seen = set()
     if os.path.isfile(DEFAULT_BOUNCE_SEEN):
@@ -1122,53 +1357,14 @@ def check_bounces(cfg, args):
         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)])")
-        if typ != "OK" or not hdata or hdata[0] is None:
-            continue
-        try:
-            head = email.message_from_bytes(hdata[0][1])
-        except Exception:  # noqa: BLE001
-            continue
-        msg_id = (head.get("Message-ID") or "").strip() or f"uid:{uid}"
-        if msg_id in seen:
-            continue
-        scanned += 1
-        if not is_bounce_message(head):
-            continue
-        typ, fdata = conn.fetch(mid, "(RFC822)")
-        if typ != "OK" or not fdata or fdata[0] is None:
-            continue
-        try:
-            full = email.message_from_bytes(fdata[0][1])
-        except Exception:  # noqa: BLE001
-            continue
-        subject = _decode_hdr(full.get("Subject", "")) or "(无主题)"
-        date = _decode_hdr(full.get("Date", ""))
-        rcpts = extract_bounced_recipients(full, account)
-        bounce_cnt += 1
-        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:
-            desc = "  ".join(
-                f"{r['email']}[{CAT_LABEL.get(r['category'], r['category'])}]" for r in rcpts
-            )
-        else:
-            desc = "(未解析出收件人)"
-        print(f"  [退信 {i}/{len(ids)}] {subject[:40]}  →  {desc}")
-
     try:
-        conn.logout()
+        res = _scan_imap_bounces(cfg, args, start_dt, skip_ids=seen, start_desc=start_desc)
-    except Exception:  # noqa: BLE001
+    except Exception as e:  # noqa: BLE001
-        pass
+        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: []}
@@ -1289,25 +1485,84 @@ def load_params(path):
         sys.exit(f"[FATAL] 解析参数文件失败: {e}")
 
     data = {"config": "", "site_url": "", "box_prefix": "", "vars": {}, "to": "",
-            "limit": None, "delay": None, "group_pause": None}
+            "limit": None, "delay": None, "group_pause": None, "smtp_idle_reconnect": None,
+            "batch_size": None,
+            "watch_bounces": None, "watch_every": None, "prune_unknown": None,
+            "no_invalid_list": None, "invalid_file": "", "bounce_report": "",
+            "imap_host": "", "imap_port": None, "since": "", "since_days": None}
     data["config"] = cp.get("path", "config", fallback="").strip()
     data["site_url"] = cp.get("site", "site_url", fallback="").strip().rstrip("/")
     data["box_prefix"] = cp.get("site", "box_prefix", fallback="").strip()
     if cp.has_section("vars"):
         data["vars"] = {k: v.strip() for k, v in cp.items("vars")}
     data["to"] = cp.get("send", "to", fallback="").strip()
-    for key, cast in (("limit", int), ("delay", float), ("group_pause", float)):
+    for key, cast in (("limit", int), ("delay", float), ("group_pause", float),
+                      ("smtp_idle_reconnect", float), ("batch_size", int)):
         raw = cp.get("send", key, fallback="").strip()
         if raw:
             try:
                 data[key] = cast(raw)
             except ValueError:
                 print(f"[警告] params.ini [send] {key}={raw!r} 不是合法数字，已忽略")
+
+    # [bounce] 退信核查 / 发送中巡检的开关
+    for key in ("watch_bounces", "prune_unknown", "no_invalid_list"):
+        val = _parse_bool(cp.get("bounce", key, fallback="").strip(), key)
+        if val is not None:
+            data[key] = val
+    for key in ("invalid_file", "bounce_report", "imap_host", "since"):
+        data[key] = cp.get("bounce", key, fallback="").strip()
+    for key, cast in (("watch_every", int), ("imap_port", int), ("since_days", int)):
+        raw = cp.get("bounce", key, fallback="").strip()
+        if raw:
+            try:
+                data[key] = cast(raw)
+            except ValueError:
+                print(f"[警告] params.ini [bounce] {key}={raw!r} 不是合法整数，已忽略")
     return data
 
 
-def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None):
+def _parse_bool(raw, key=""):
-    """把参数写回 params.ini（覆盖写；vars 逐行 key = value）。"""
+    """解析 1/0、yes/no、true/false、on/off；空串返回 None（表示未配置）。"""
+    low = (raw or "").strip().lower()
+    if not low:
+        return None
+    if low in ("1", "yes", "y", "true", "on", "开", "启用"):
+        return True
+    if low in ("0", "no", "n", "false", "off", "关", "禁用"):
+        return False
+    print(f"[警告] params.ini [bounce] {key}={raw!r} 不是合法布尔值"
+          f"（用 1/0 或 yes/no），已忽略")
+    return None
+
+
+def _preserve_extra_sections(path, managed=("path", "site", "vars", "send", "bounce")):
+    """读取已有 params.ini 里非托管的小节，原样保留（避免保存时丢掉手写的配置）。"""
+    if not path or not os.path.isfile(path):
+        return []
+    cp = configparser.ConfigParser(comment_prefixes=(";", "#"),
+                                   inline_comment_prefixes=(";", "#"),
+                                   interpolation=None, strict=False)
+    cp.optionxform = str
+    try:
+        cp.read(path, encoding="utf-8")
+    except Exception:  # noqa: BLE001
+        return []
+    out = []
+    for sec in cp.sections():
+        if sec.strip().lower() in managed:
+            continue
+        out.append("")
+        out.append(f"[{sec}]")
+        for k, v in cp.items(sec):
+            out.append(f"{k} = {v}")
+    return out
+
+
+def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None,
+                bounce=None):
+    """把参数写回 params.ini（覆盖写；vars 逐行 key = value）。
+    bounce：当前生效的退信/巡检设置，回写进 [bounce] 小节以便下次直接沿用。"""
     lines = ["; broadcast.py 参数文件：每次运行自动读取；命令行参数优先级更高",
              "; 注意：本文件不控制 --send/--yes，正式发送仍需命令行显式指定", ""]
     lines.append("[path]")
@@ -1327,6 +1582,29 @@ def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None
     lines.append("; to = 可选：默认测试收件邮箱（正式群发用命令行 --send，不加 --to）")
     lines.append("delay = 1.0")
     lines.append("group_pause = 3.0")
+    lines.append("")
+    lines.append("; 退信核查 / 发送中限流巡检（均可用同名命令行参数临时覆盖）")
+    lines.append("[bounce]")
+    b = bounce or {}
+    yesno = lambda v: "1" if v else "0"  # noqa: E731
+    lines.append("; 发送中巡检收件箱，出现「外发频率超过邮件系统限制」类退信立即中止发送")
+    lines.append(f"watch_bounces = {yesno(b.get('watch_bounces', True))}")
+    lines.append("; 每发 N 封巡检一次（调小更及时，但 IMAP 登录更频繁）")
+    lines.append(f"watch_every = {b.get('watch_every', 10)}")
+    lines.append("; 退信核查时，未分类的退信也按可重试一并剔除补发")
+    lines.append(f"prune_unknown = {yesno(b.get('prune_unknown', False))}")
+    lines.append("; 停用无效地址清单（既不跳过已知无效地址，也不写入新的硬退信）")
+    lines.append(f"no_invalid_list = {yesno(b.get('no_invalid_list', False))}")
+    lines.append("; 留空则用脚本同目录的默认文件名")
+    lines.append(f"invalid_file = {b.get('invalid_file', '') or ''}")
+    lines.append(f"bounce_report = {b.get('bounce_report', '') or ''}")
+    lines.append("; 留空则按 SMTP 域名推导（smtp.xxx → imap.xxx）")
+    lines.append(f"imap_host = {b.get('imap_host', '') or ''}")
+    lines.append(f"imap_port = {b.get('imap_port', 993)}")
+    lines.append("; since = auto 表示从断点清单最早一条记录的时间开始扫描")
+    lines.append(f"since = {b.get('since', 'auto')}")
+    lines.append(f"since_days = {b.get('since_days', 3)}")
+    lines.extend(_preserve_extra_sections(path))
     with open(path, "w", encoding="utf-8") as f:
         f.write("\n".join(lines) + "\n")
 
@@ -1366,6 +1644,9 @@ def run_interactive(cfg, templates, extra_vars, args, config_path):
     print(f"  box_prefix = {box_prefix}")
     for k in sorted(extra_vars):
         print(f"  {{{{{k}}}}} = {extra_vars[k]}")
+    print(f"  发送中巡检退信 = {'开启' if args.watch_bounces else '关闭'}"
+          f"（每 {args.watch_every} 封一次）"
+          f"   来源：params.ini [bounce] / 命令行，改值请编辑 params.ini")
 
     # 站点地址：只有完全没值时才问（值链已在 main 里兜底过 external_url）
     if not site_url:
@@ -1457,12 +1738,27 @@ def run_interactive(cfg, templates, extra_vars, args, config_path):
     except EOFError:
         save_ans = ""
     if save_ans in ("y", "yes"):
-        save_params(DEFAULT_PARAMS, config_path, site_url, box_prefix, extra_vars)
+        save_params(DEFAULT_PARAMS, config_path, site_url, box_prefix, extra_vars,
+                    batch_size=getattr(args, "batch_size", 0) or 0,
+                    smtp_idle_reconnect=getattr(args, "smtp_idle_reconnect", 30.0) or 30.0,
+                    bounce={
+                        "watch_bounces": args.watch_bounces,
+                        "watch_every": args.watch_every,
+                        "prune_unknown": args.prune_unknown,
+                        "no_invalid_list": args.no_invalid_list,
+                        "invalid_file": args.invalid_file,
+                        "bounce_report": args.bounce_report,
+                        "imap_host": args.imap_host,
+                        "imap_port": args.imap_port,
+                        "since": args.since,
+                        "since_days": args.since_days,
+                    })
         print(f"[交互模式] 已保存到 {DEFAULT_PARAMS}")
     return site_url, box_prefix, extra_vars
 
 
-def main():
+def build_arg_parser():
+    """构造命令行解析器（单独抽出来，便于测试与复用）。"""
     parser = argparse.ArgumentParser(
         description="TamaBox 站内信群发工具（读取 conf/app.ini，模板在 templates/ 目录，按用户语言发送）",
         formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -1502,6 +1798,16 @@ def main():
                         help="--pause-every 触发时的长暂停秒数，默认 15.0")
     parser.add_argument("--group-pause", type=float, default=3.0,
                         help="不同语言分组之间的额外暂停秒数，默认 3.0（0 关闭）")
+    parser.add_argument("--smtp-idle-reconnect", type=float, default=None, metavar="SEC",
+                        help="SMTP 连接空闲超过该秒数就主动重连再发（默认 30）。"
+                             "--delay 调大时（如 60s/封）服务器会掐掉空闲连接，"
+                             "不重连的话每封都会先吃一发 Server not connected")
+    parser.add_argument("--batch-size", type=int, default=None, metavar="N",
+                        help="合并信封批量发送（0=逐封，默认）：渲染内容完全相同的收件人"
+                             "每 N 人共用一个 SMTP 信封，发送次数从「人数」降到「信封数」，"
+                             "可显著减少触发「外发频率超过限制」类限流（服务商按收件人计数"
+                             "则无效）。单封收件人有上限（常见 50~100），被 452 拒收时脚本"
+                             "自动砍半拆批重试，不会中止")
     parser.add_argument("--db-driver", default="auto",
                         choices=["auto", "psql", "mysql", "psycopg2", "pymysql"],
                         help="DB 驱动：auto（默认，优先系统 psql/mysql 客户端）/ psycopg2 / pymysql")
@@ -1536,27 +1842,34 @@ def main():
                         help="配合 --check-bounces：把「可重试」退信（限流/临时性）从断点续发清单"
                              "剔除后结束（仅剔除，本次不发送）；再加 --send 则剔除后立即补发。"
                              "「地址不存在」等永久失败不会被剔除（不补发）")
-    parser.add_argument("--prune-unknown", action="store_true",
+    parser.add_argument("--prune-unknown", action="store_true", default=None,
                         help="配合 --check-bounces：无法分类的退信也按「可重试」处理，一并剔除补发"
-                             "（默认不剔除，避免给死信地址反复重发）")
+                             "（默认不剔除，避免给死信地址反复重发；也可在 params.ini [bounce] 设置）")
-    parser.add_argument("--invalid-file", default=DEFAULT_INVALID, metavar="FILE",
+    parser.add_argument("--invalid-file", default=None, metavar="FILE",
                         help=f"无效地址清单（硬退信地址）路径；默认脚本同目录 broadcast_invalid.json。"
                              f"清单里的地址每次运行都会跳过，即使 --reset-state")
     parser.add_argument("--reset-invalid", action="store_true",
                         help="清空无效地址清单（慎用：这些地址多半仍不可达）")
-    parser.add_argument("--no-invalid-list", action="store_true",
+    parser.add_argument("--no-invalid-list", action="store_true", default=None,
                         help="本次不使用无效地址清单：既不跳过已知无效地址，也不写入新的硬退信")
-    parser.add_argument("--bounce-report", default=DEFAULT_BOUNCE_REPORT, metavar="FILE",
+    parser.add_argument("--bounce-report", default=None, metavar="FILE",
                         help="退信核查明细报告路径（JSON）；默认脚本同目录 broadcast_bounce_report.json")
-    parser.add_argument("--imap-host", default="",
+    parser.add_argument("--watch-bounces", dest="watch_bounces", action="store_true", default=None,
+                        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=None, metavar="N",
+                        help="发送中每发 N 封巡检一次退信（默认 10；调小会更频繁登录 IMAP）")
+    parser.add_argument("--imap-host", default=None,
                         help="IMAP 服务器地址（默认由 SMTP 域名推导：smtp.xxx → imap.xxx）")
-    parser.add_argument("--imap-port", type=int, default=993,
+    parser.add_argument("--imap-port", type=int, default=None,
                         help="IMAP 端口（默认 993，SSL）")
-    parser.add_argument("--since", default="auto", metavar="auto|YYYY-MM-DD",
+    parser.add_argument("--since", default=None, metavar="auto|YYYY-MM-DD",
                         help="退信扫描起点：auto（默认）= 从群发记录（断点清单）最早一条的"
                              "时间开始，只核查本次群发相关的退信；"
                              "也可显式指定日期如 2026-09-07")
-    parser.add_argument("--since-days", type=int, default=3,
+    parser.add_argument("--since-days", type=int, default=None,
                         help="断点清单为空时的回退扫描范围（天），默认 3 天")
     parser.add_argument("--subject", default=None,
                         help="全局覆盖邮件主题（忽略各语言 subject.txt；支持占位符）")
@@ -1564,8 +1877,12 @@ def main():
                         help="全局覆盖 HTML 正文模板文件（忽略按语言模板；调试用）")
     parser.add_argument("--plain", metavar="FILE",
                         help="全局覆盖纯文本正文模板文件（调试用）")
+    return parser
+
+
+def main():
     global args
-    args = parser.parse_args()
+    args = build_arg_parser().parse_args()
 
     # 默认交互模式：没给任何运行方式相关参数（--send/--to/--yes/--dry-run）就自动进入交互；
     # 显式 --no-interactive 强制关闭（脚本化/定时任务用）
@@ -1617,6 +1934,33 @@ def main():
         args.delay = params["delay"]
     if args.group_pause == 3.0 and params.get("group_pause") is not None:
         args.group_pause = params["group_pause"]
+    if args.smtp_idle_reconnect is None:
+        args.smtp_idle_reconnect = (30.0 if params.get("smtp_idle_reconnect") is None
+                                    else params["smtp_idle_reconnect"])
+    if args.batch_size is None:
+        args.batch_size = params.get("batch_size") or 0
+
+    # 退信核查 / 发送中巡检：命令行没给（None）时才用 params.ini [bounce]
+    if args.watch_bounces is None:
+        args.watch_bounces = True if params.get("watch_bounces") is None else params["watch_bounces"]
+    if args.watch_every is None:
+        args.watch_every = params.get("watch_every") or 10
+    if args.prune_unknown is None:
+        args.prune_unknown = bool(params.get("prune_unknown"))
+    if args.no_invalid_list is None:
+        args.no_invalid_list = bool(params.get("no_invalid_list"))
+    if args.invalid_file is None:
+        args.invalid_file = params.get("invalid_file") or DEFAULT_INVALID
+    if args.bounce_report is None:
+        args.bounce_report = params.get("bounce_report") or DEFAULT_BOUNCE_REPORT
+    if args.imap_host is None:
+        args.imap_host = params.get("imap_host") or ""
+    if args.imap_port is None:
+        args.imap_port = params.get("imap_port") or 993
+    if args.since is None:
+        args.since = params.get("since") or "auto"
+    if args.since_days is None:
+        args.since_days = params.get("since_days") or 3
     extra_vars = {**params.get("vars", {}), **parse_var(args.var)}  # CLI --var 覆盖文件
 
     # ---- conf/app.ini 路径解析：CLI > params.ini > 环境变量 > ./conf/app.ini ----
@@ -1825,6 +2169,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 +2229,63 @@ def main():
 
     sent, failed = 0, []
     smtp_rejected = 0
+    attempted = 0
     total = len(recipients)
     all_missing = set()
+    batch_size = max(0, int(getattr(args, "batch_size", 0) or 0))
+    envelopes = 0          # SMTP 信封数（批量模式下 < 发送人数）
+    watch_mark = 0
+
+    # 发送中巡检：限流退信是异步投到发件箱的，边发边看才能及时止损
+    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 = {}, {}, ""
+
+    def note_smtp_err(email, err_str, subject_label):
+        """SMTP 拒收按退信分类处理：硬退信记无效清单；返回限流中止理由（否则空串）。"""
+        cat, why = classify_bounce("", err_str, "")
+        if cat == CAT_HARD and not args.no_invalid_list:
+            record_invalid(args.invalid_file, {email: {
+                "reason": why, "detail": err_str[:160], "subject": subject_label,
+                "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}})
+        if cat == CAT_RATE:
+            return f"SMTP 返回疑似限流（{why}）：{err_str[:100]}"
+        return ""
+
+    def run_patrol():
+        """周期巡检收件箱：命中限流退信则给出中止理由，连续两轮失败则关巡检。"""
+        nonlocal watch_fails, watch_active, watch_hard, abort_reason
+        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
+            return
+        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 and not abort_reason:
+            abort_reason = (
+                f"收件箱出现 {len(watch_rate)} 条限流退信"
+                f"（如 {sorted(watch_rate)[0]}："
+                f"{watch_rate[sorted(watch_rate)[0]]['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"，合并信封每 {batch_size} 人/封" if batch_size > 0 else "")
+          + (f"，每 {watch_every} 封巡检一次退信" if watch_active else "") + " ...")
 
     if args.single:
         # 单文件模式：不按语言分组，所有人同一模板
@@ -1900,7 +2299,7 @@ def main():
         group_order = [lg for lg in lang_list if lg in groups] + [lg for lg in groups if lg not in lang_list]
 
     try:
-        server = smtp_login(cfg)
+        session = SmtpSession(cfg, idle_reconnect=getattr(args, "smtp_idle_reconnect", 30))
     except Exception as e:  # noqa: BLE001
         sys.exit(f"[FATAL] SMTP 登录失败: {e}")
 
@@ -1914,6 +2313,15 @@ def main():
                 label = LANG_LABEL.get(lg, lg)
                 tpl = templates[lg]
             print(f"\n[语言分组 {gi}/{len(group_order)}] {lg} ({label}) : 本组 {len(grp)} 封")
+
+            subject_raw = args.subject if args.subject else tpl["subject"]
+            html_raw = override_html if override_html is not None else tpl["html"]
+            plain_raw = override_plain if override_plain is not None else tpl["plain"]
+
+            # 渲染并按内容指纹分桶：只有渲染结果完全相同的收件人才共用信封。
+            # 模板含 {{name}}/{{box_link}} 等个人化占位符时结果逐人不同，
+            # 自然落回逐人一封，不会错合。
+            buckets, bucket_map = [], {}
             for r in grp:
                 box_link = f"{site_url}{args.box_prefix}{r['domain']}" if r["domain"] else site_url
                 ctx = {
@@ -1926,51 +2334,147 @@ def main():
                     "year": datetime.now().strftime("%Y"),
                     **extra_vars,
                 }
-                subject = args.subject if args.subject else tpl["subject"]
+                s, h, p, missing = render_content(subject_raw, html_raw, plain_raw, ctx)
-                html = override_html if override_html is not None else tpl["html"]
-                plain = override_plain if override_plain is not None else tpl["plain"]
-                msg, missing = build_message(cfg, r["email"], ctx, subject, html, plain)
                 if missing:
                     all_missing.update(missing)
-                try:
+                digest = hashlib.sha256(
-                    refused = server.send_message(msg)
+                    "\x00".join((s, h, p)).encode("utf-8", "replace")).hexdigest()
-                    if refused:
+                bkt = bucket_map.get(digest)
-                        refused_desc = "; ".join(f"{a} {c}:{err}" for a, (c, err) in refused.items())
+                if bkt is None:
-                        failed.append((r["email"], f"SMTP 拒收 {refused_desc}"))
+                    bkt = {"content": (s, h, p), "queue": []}
-                        smtp_rejected += 1
+                    bucket_map[digest] = bkt
-                        print(f"  [SMTP 拒收] {r['email']}: {refused_desc}")
+                    buckets.append(bkt)
+                bkt["queue"].append(r)
+            del bucket_map
+
+            for bkt in buckets:
+                subj_r, html_r, plain_r = bkt["content"]
+                queue = bkt["queue"]
+                while queue and not abort_reason:
+                    take = batch_size if batch_size > 0 else 1
+                    batch, queue = queue[:take], queue[take:]
+                    rcpts = [r["email"] for r in batch]
+
+                    if len(batch) == 1:
+                        to_header = batch[0]["email"]
                     else:
+                        # 批量信封：To: 头只作显示（站点名+发件邮箱），真实收件人
+                        # 走 RCPT TO 信封，互相不可见
+                        to_header = formataddr((str(Header(cfg.get("app_title") or "", "utf-8")),
+                                                cfg["mail_account"]))
+                    msg = build_message_from(cfg, to_header, subj_r, html_r, plain_r)
+
+                    envelopes += 1
+                    try:
+                        refused = session.send_to(msg, rcpts if len(batch) > 1 else None)
+                    except Exception as e:  # noqa: BLE001
+                        # 整信封异常（重连耗尽 / DATA 被拒等）：多收件人时无法
+                        # 区分到人，只记失败；若像限流（DATA 4xx/552）则立即中止。
+                        for r in batch:
+                            failed.append((r["email"], str(e)))
+                        attempted += len(batch)
+                        print(f"  [失败] 本信封 {len(batch)} 人: {e}")
+                        if len(batch) == 1:
+                            ab = note_smtp_err(batch[0]["email"], str(e), subj_r)
+                        else:
+                            cat, why = classify_bounce("", str(e), "")
+                            ab = f"SMTP 返回疑似限流（{why}）：{str(e)[:100]}" if cat == CAT_RATE else ""
+                        if ab and not abort_reason:
+                            abort_reason = ab
+                        continue
+
+                    err_map = {a.lower(): _fmt_smtp_result(v)
+                               for a, v in (refused or {}).items()}
+                    ok_n = 0
+                    for r in batch:
+                        if r["email"].lower() in err_map:
+                            continue
+                        ok_n += 1
                         sent += 1
                         if not args.to:
-                            record_sent(args.state, {"email": r["email"], "lang": lg, "ts": time.time()})
+                            record_sent(args.state, {"email": r["email"], "lang": lg,
-                        if sent % 25 == 0 or sent == total:
+                                                     "ts": time.time()})
-                            print(f"  进度 {sent}/{total}  已成功 {sent}")
+                    attempted += len(batch)
-                except Exception as e:  # noqa: BLE001
+
-                    failed.append((r["email"], str(e)))
+                    # 「单封收件人数超限」的拒绝：砍半批次重试，被拒的人排回队首；
-                    print(f"  [失败] {r['email']}: {e}")
+                    # 未被拒的收件人已随本次 DATA 投出，照常入账
-                if sent < total:
+                    oversized = [r for r in batch
-                    time.sleep(max(0.0, args.delay))
+                                 if _is_too_many_rcpts(err_map.get(r["email"].lower(), ""))]
-                    if args.pause_every > 0 and sent % args.pause_every == 0:
+                    if oversized:
-                        print(f"  [间隔] 已发 {sent} 封，暂停 {args.pause_for}s 防限流 ...")
+                        batch_size = max(1, batch_size // 2)
-                        time.sleep(max(0.0, args.pause_for))
+                        queue[:0] = oversized
+                        over_set = {r["email"].lower() for r in oversized}
+                        print(f"  [拆批] {len(oversized)} 人被拒（单封收件人数超限），"
+                              f"batch_size 降为 {batch_size} 后重试")
+                    else:
+                        over_set = set()
+
+                    for r in batch:
+                        em_low = r["email"].lower()
+                        e = err_map.get(em_low)
+                        if e is None or em_low in over_set:
+                            continue
+                        failed.append((r["email"], f"SMTP 拒收 {e}"))
+                        smtp_rejected += 1
+                        print(f"  [SMTP 拒收] {r['email']}: {e}")
+                        ab = note_smtp_err(r["email"], e, subj_r)
+                        if ab and not abort_reason:
+                            abort_reason = ab
+
+                    if len(batch) > 1:
+                        print(f"  [信封] 一封投递 {len(batch)} 人："
+                              f"接收 {ok_n}，拒收 {len(err_map) - len(over_set)}")
+                    if sent % 25 == 0 or sent == total:
+                        print(f"  进度 {sent}/{total}  已成功 {sent}")
+
+                    # 周期巡检收件箱：按收件人计数，异步退信才是限流最常见表现
+                    if watch_active and not abort_reason and attempted >= watch_mark + watch_every:
+                        watch_mark = attempted
+                        run_patrol()
+
+                    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:
+                            print(f"  [间隔] 已发 {sent} 封，暂停 {args.pause_for}s 防限流 ...")
+                            time.sleep(max(0.0, args.pause_for))
             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:
+        session.close()
-            server.quit()
-        except Exception:  # noqa: BLE001
-            pass
 
     report["sent"] = sent
     report["failed"] = len(failed)
     report["smtp_rejected"] = smtp_rejected
+    report["batch_size"] = batch_size
+    report["envelopes"] = envelopes
+    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)
-    print(f"发送完成：成功 {sent} 封，失败 {len(failed)} 封（其中 SMTP 拒收 {smtp_rejected} 封）。")
+    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))}")
     if not args.to and args.send:

params.ini

@@ -18,3 +18,44 @@ old_domain = box.tama.guru
 ; to = 可选：默认测试收件邮箱（正式群发用命令行 --send，不加 --to）
 delay = 1.0
 group_pause = 3.0
+; SMTP 连接空闲超过该秒数就主动重连再发。
+; delay 调大时（比如 60s/封）服务器会掐掉空闲连接，出现整批
+; 「please run connect() first」；保持默认 30 即可自动重连，设 0 关闭
+smtp_idle_reconnect = 30
+
+; 合并信封批量发送（0=逐封）。渲染内容完全相同的收件人每 N 人共用一个
+; SMTP 信封：发送次数从「人数」降到「信封数」（如 305 人、每 50 人一封
+; → 约 7 次发信），可显著降低触发「外发频率超过邮件系统限制」类限流的
+; 概率；若服务商按收件人计数则无缓解。
+; 注意：单封收件人有服务商上限（常见 50~100），超限被 452 拒收时脚本会
+; 自动砍半拆批重试，不会中止；delay 在批量模式下的语义是「每批之间」间隔。
+; 模板含 {{name}} 等个人化占位符时内容逐人不同，自动回退逐封，不会错合。
+batch_size = 0
+
+; 退信核查 / 发送中限流巡检开关
+; 这些都能在 params.ini 里长期配置，命令行同名参数可临时覆盖
+[bounce]
+; 发送中巡检收件箱：出现「外发频率超过邮件系统限制」类限流退信立即中止发送
+; 中止后会把限流地址从断点清单剔除，等限制恢复重跑即自动补发
+watch_bounces = 1
+
+; 每发 N 封巡检一次退信。调小更及时，但 IMAP 登录更频繁（不建议 < 5）
+watch_every = 10
+
+; 退信核查时，分类不明的退信是否也按「可重试」一并剔除补发（默认 0，保守不补发）
+prune_unknown = 0
+
+; 停用无效地址清单：既不跳过已知无效地址，也不写入新的硬退信
+no_invalid_list = 0
+
+; 留空则用脚本同目录的默认文件名
+invalid_file =
+bounce_report =
+
+; IMAP（退信核查要用发件邮箱的 IMAP；留空则按 SMTP 域名推导 smtp.xxx → imap.xxx）
+imap_host =
+imap_port = 993
+
+; since = auto 表示从断点清单最早一条发送记录的时间开始扫描，只查本次群发的退信
+since = auto
+since_days = 3