发信账号来源开关：app.ini / 程序自带 SMTP

- params.ini 新增 [mail]：source=app（默认，用 conf/app.ini [mail]）或
  own（用 account/password/smtp/port/skip_tls_verify 这套）；--mail-source
  app|own 可临时覆盖（命令行 > params.ini）
- 切换后发信与退信核查（IMAP，含主机推导）都走这套账号；切换到 own 但
  配置缺项时直接 FATAL，不用半套配置
- password 支持 env:变量名：params.ini 会被 git 跟踪，明文密码时打印警告
- 交互模式：[mail] 配齐时多问一次来源；保存时 [mail] 小节原样回写
- 修复：save_params 缺少 batch_size/smtp_idle_reconnect 形参，但调用点已在
  传 → 交互模式选「保存到 params.ini」会 TypeError 崩溃（上一并行编辑漏改）
- 补回上一轮被漏掉的 README「合并信封批量发送」章节与 452 安全机制条目

README.md

@@ -41,6 +41,8 @@ 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                      ; 发送中巡检限流退信，发现即中止
@@ -247,9 +249,61 @@ conf/app.ini 路径的解析优先级：
   - 用的是 conf/app.ini [mail] 的账号与密码（需邮箱已开启 IMAP，密码为邮箱登录密码）
   - 交互模式的运行方式「4) 退信核查」里同样三选一：仅核查 / 仅剔除 / 剔除并补发，
     选择「仅剔除 / 剔除并补发」时会再问一次是否连未分类的一起补发
-  - 已核查过的退信（按 Message-ID）记录在 broadcast_bounces_seen.json，
+  - 已核查过的退信（按 Message-ID）与**上次扫描位置**（INBOX UID 游标 +
-    重复核查不会重复报告/重复剔除
+    UIDVALIDITY）都记录在 broadcast_bounces_seen.json：
-  - 每次核查另写一份明细报告 broadcast_bounce_report.json（含每个收件人的类别与判定依据）
+    默认从上次位置续扫（UID n+1:*），只拉新邮件的头部，重复核查不会
+    重复报告/重复剔除，也不会随邮箱邮件增多越扫越慢
+  - 以下情况自动回退为按日期窗口从头扫（扫完位置照常更新）：
+    首次核查（无位置记录）、邮箱 UIDVALIDITY 变化（换号/重建邮箱）、
+    显式指定 --since、加 --full-scan
+  - 发送中巡检不推进扫描位置（它只看本次运行之后新到的退信，
+    范围内的历史邮件不标记「已扫」），位置只由 --check-bounces 推进
+  - 每次核查另写一份明细报告 broadcast_bounce_report.json（含每个收件人的
+    类别与判定依据，以及 scan_mode / scan_cursor）
+
+发信账号来源：app.ini / 程序自带 SMTP
+------------------------------------
+  默认用 conf/app.ini [mail] 的 SMTP（站内信跟随站点自己的邮箱）。切到程序
+  自带的一套后，发信与退信核查（IMAP）都走这套账号：
+
+    [mail]
+    source = app          ; app（默认）= conf/app.ini [mail]；own = 下面这套
+    account = bot@other.com
+    password = env:MY_SMTP_PASSWORD     ; 支持 env:变量名，推荐
+    smtp = smtp.other.com
+    port = 465
+    skip_tls_verify = 0
+
+  临时切换（优先级高于 params.ini）：`--mail-source own` / `--mail-source app`
+  交互模式里若 [mail] 配齐了，会多问一次选哪个来源。
+
+  安全提醒：
+  - params.ini 会被 git 跟踪，**不要把密码明文写进去**；用 `env:变量名`
+    从环境变量读取，或把 params.ini 加入 .gitignore
+  - 明文写死时脚本会打印警告，但不会阻止运行
+  - 切到 own 但 [mail] 缺 smtp/account/password 时直接 FATAL，不会用半套配置
+
+合并信封批量发送（可选）
+------------------------
+  默认逐人一封（一个 SMTP 信封 = 1 个 RCPT TO）。开启 batch_size 后，
+  「渲染后内容完全相同」的收件人合并发送：一个信封 = 1×MAIL FROM +
+  N×RCPT TO + 1×DATA，发送次数从「人数」降到「信封数」（如 305 人、
+  每 50 人一封 → 约 7 次发信）。
+
+  开启方式（0=关闭）：
+    broadcast.py --send --single --batch-size 25
+    或 params.ini [send] batch_size = 25
+
+  与限流的关系（要点）：
+  - 服务商按「发信次数」计频率 → 合并后成倍降低触发概率（主要收益）
+  - 服务商按「单位时间收件人总数」计数 → 无缓解，仍靠 delay/pause 控制节奏
+  - 单封收件人数上限常见 50~100：超限的 RCPT 会被 452 拒收，脚本自动把
+    batch_size 砍半、被拒者重试，不会中止也不会误判为限流
+  - 信封收件人对其他收件人不可见；批量信封的 To: 头显示为「站点名+发件邮箱」
+
+  适用的前提是内容逐字相同：模板含 {{name}}/{{box_link}} 等个人化占位符时，
+  渲染结果逐人不同，会自动落回逐封，不会错合。断点续发按人记录（DATA 被
+  服务器接收即整批入账，被拒的除外）；发送中限流巡检照常按收件人数计数。
 
 安全机制
 --------
@@ -257,7 +311,12 @@ 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）计入失败并写入报告
+  - 批量信封遇 452「收件人数超限」自动砍半拆批重试；限流类拒收仍立即中止
   - From/Subject 头自动做 RFC2047 编码（中文显示名不会被 QQ 邮箱 550 拒收）
   - 自动定位 users 表所在 schema（避免 psql 命中别的同名空表）
   - 语言列自动试跑探测（language → lang → NULL 兜底），老库没有该列也能发

broadcast.py

@@ -62,6 +62,7 @@ TamaBox 站内信群发工具（独立项目，零依赖，不依赖、不修改
 import argparse
 import configparser
 import email.utils
+import hashlib
 import imaplib
 import json
 import os
@@ -82,7 +83,7 @@ from email.mime.text import MIMEText
 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-08.mailsource.v5"
 
 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")
@@ -599,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):
@@ -637,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 追加写，崩溃安全）
 # ---------------------------------------------------------------------------
@@ -1117,18 +1249,59 @@ def _parse_mail_date(value):
     return dt
 
 
+def _load_bounce_seen():
+    """读取已核查记录文件（DEFAULT_BOUNCE_SEEN）。
+
+    旧格式：Message-ID 列表；新格式：{"seen": [...], "cursor": {...}}。
+    返回 (seen 集合, cursor dict 或 None)。
+    cursor = {"uidvalidity": int, "last_uid": int}，即上次扫描在 INBOX 里
+    扫到的位置；UIDVALIDITY 对不上（邮箱重建/换号）时游标作废回退全扫。
+    """
+    if not os.path.isfile(DEFAULT_BOUNCE_SEEN):
+        return set(), None
+    try:
+        with open(DEFAULT_BOUNCE_SEEN, "r", encoding="utf-8") as f:
+            data = json.load(f)
+    except Exception:  # noqa: BLE001
+        return set(), None
+    if isinstance(data, dict):
+        seen = set(data.get("seen") or [])
+        cur = data.get("cursor")
+        if (isinstance(cur, dict)
+                and isinstance(cur.get("uidvalidity"), int)
+                and isinstance(cur.get("last_uid"), int)):
+            return seen, cur
+        return seen, None
+    return set(data or []), None
+
+
+def _save_bounce_seen(seen, cursor=None):
+    try:
+        with open(DEFAULT_BOUNCE_SEEN, "w", encoding="utf-8") as f:
+            json.dump({"seen": sorted(seen), "cursor": cursor}, f, ensure_ascii=False)
+    except Exception as e:  # noqa: BLE001
+        print(f"[警告] 记录已核查邮件/扫描位置失败（不影响本次结果）：{e}")
+
+
 def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None,
-                       start_desc="", verbose=True):
+                       start_desc="", verbose=True, resume=None):
     """连 IMAP 扫描退信并逐收件人分类（check_bounces 与发送中巡检共用）。
 
     skip_ids   已处理过的 Message-ID（seen 文件 / 本次运行内已见过），跳过不重复报
     not_before 只统计 Date 头晚于该时刻的退信（发送中巡检用：只看本次运行之后
                新到的退信，避免拿历史退信误触发中止）；Date 解析不了则放行
-    返回 dict：bounced / new_ids / scanned / touched_ids
+    resume     上次扫描位置 {"uidvalidity": int, "last_uid": int}；UIDVALIDITY
+               一致时从 last_uid 之后续扫（UID 检索），否则回退按日期窗口全扫
+               （全新 / 邮箱 UID 代际变更 / 显式 --since / --full-scan）。
+               只由 check_bounces 传入推进——发送中巡检的 not_before 会过滤掉
+               范围内的历史邮件，若推进游标会把没处理过的邮件标记成「已扫」。
+    返回 dict：bounced / new_ids / scanned / touched_ids / last_uid / uidvalidity
       bounced     {email: {"subject","date","category","reason","detail"}}
       new_ids     本次计入的退信 Message-ID（是否落盘由调用方决定）
       touched_ids 本次扫到的所有退信 Message-ID（含被 not_before 过滤掉的），
                   供巡检做运行期去重，避免每轮重复拉取同一封全文
+      last_uid    本次检查到的最大 UID（供调用方更新扫描位置）
+      uidvalidity 当前邮箱的 UIDVALIDITY（读不到为 None，调用方据此不落游标）
     """
     smtp_host = cfg["mail_smtp"] or ""
     imap_host = (args.imap_host or "").strip()
@@ -1148,33 +1321,58 @@ def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None,
     try:
         conn.login(account, password)
         conn.select("INBOX", readonly=True)
-        typ, data = conn.search(None, f'(SINCE "{since}")')
+        try:
+            uv = int(conn.response("UIDVALIDITY")[1][0])
+        except Exception:  # noqa: BLE001
+            uv = None
+        resume_uid = None
+        if resume:
+            if uv is not None and uv == resume.get("uidvalidity"):
+                resume_uid = int(resume["last_uid"])
+                if verbose:
+                    print(f"  [扫描位置] 续扫：从 UID {resume_uid} 之后开始"
+                          f"（uidvalidity {uv}）")
+            elif verbose:
+                print("  [扫描位置] 邮箱 UIDVALIDITY 已变化（或无法读取），"
+                      "本次回退按日期窗口全量扫描")
+        if resume_uid is not None:
+            typ, data = conn.uid("SEARCH", None, f"UID {resume_uid + 1}:*")
+        else:
+            typ, data = conn.uid("SEARCH", None, f'(SINCE "{since}")')
         if typ != "OK":
             raise RuntimeError("IMAP search 失败")
         ids = data[0].split()
         if verbose:
-            print(f"  [IMAP] 扫描范围内共 {len(ids)} 封待扫描")
+            print(f"  [IMAP] 扫描范围内共 {len(ids)} 封"
+                  f"（{'续扫' if resume_uid is not None else '日期窗口'}）")
 
         skip_ids = skip_ids or set()
         bounced, new_ids, touched, scanned = {}, [], [], 0
+        last_uid = resume_uid or 0
         for i, mid in enumerate(ids, 1):
-            uid = mid.decode() if isinstance(mid, bytes) else str(mid)
+            uid = int(mid)
+            if uid <= last_uid:
+                # 防御：UID n:* 在 n 超过邮箱最大 UID 时会返回最后一封
+                continue
+            last_uid = uid
+            ustr = str(uid)
             # 两段式：先取头部（省流量），命中退信特征再取全文
-            typ, hdata = conn.fetch(
+            typ, hdata = conn.uid(
-                mid, "(BODY.PEEK[HEADER.FIELDS (SUBJECT MESSAGE-ID FROM CONTENT-TYPE)])")
+                "fetch", ustr,
+                "(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}"
+            msg_id = (head.get("Message-ID") or "").strip() or f"uid:{ustr}"
             if msg_id in skip_ids:
                 continue
             scanned += 1
             if not is_bounce_message(head):
                 continue
-            typ, fdata = conn.fetch(mid, "(RFC822)")
+            typ, fdata = conn.uid("fetch", ustr, "(RFC822)")
             if typ != "OK" or not fdata or fdata[0] is None:
                 continue
             try:
@@ -1201,7 +1399,8 @@ def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None,
                     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}
+                "scanned": scanned, "touched_ids": touched,
+                "last_uid": last_uid, "uidvalidity": uv}
     finally:
         try:
             conn.logout()
@@ -1210,29 +1409,43 @@ def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None,
 
 
 def check_bounces(cfg, args):
-    """登录发件邮箱 IMAP，扫描近 N 天的退信并解析失败收件人。
+    """登录发件邮箱 IMAP，扫描退信并解析失败收件人。
 
     返回 {email: {"subject","date","category","reason","detail"}}；已处理过的
-    邮件（Message-ID）记录在 broadcast_bounces_seen.json，重复核查不会重复报告。
+    邮件（Message-ID）与上次扫描位置（UID 游标）都记录在
+    broadcast_bounces_seen.json：默认从上次位置续扫，只查新到的邮件，
+    游标不存在/作废、显式 --since 或 --full-scan 时才按日期窗口从头扫。
     category 见 CAT_RATE / CAT_HARD / CAT_UNKNOWN。"""
     start_dt, start_desc = resolve_scan_start(args)
 
-    seen = set()
+    seen, cursor = _load_bounce_seen()
-    if os.path.isfile(DEFAULT_BOUNCE_SEEN):
+    since_explicit = ((getattr(args, "since", "auto") or "auto").strip().lower()
-        try:
+                      not in ("", "auto"))
-            with open(DEFAULT_BOUNCE_SEEN, "r", encoding="utf-8") as f:
+    full_scan = bool(getattr(args, "full_scan", False))
-                seen = set(json.load(f))
+    resume = None
-        except Exception:  # noqa: BLE001
+    if full_scan:
-            seen = set()
+        print("[扫描位置] --full-scan：忽略上次扫描位置，从头全量扫描")
+    elif since_explicit:
+        print("[扫描位置] 显式 --since 指定窗口，本次忽略上次扫描位置")
+    elif cursor is None:
+        print("[扫描位置] 全新扫描（尚无上次位置记录），按日期窗口全扫")
+    else:
+        resume = cursor
+        print(f"[扫描位置] 从上次扫到的 UID {cursor['last_uid']} 之后续扫"
+              f"（uidvalidity {cursor['uidvalidity']}）")
 
     try:
-        res = _scan_imap_bounces(cfg, args, start_dt, skip_ids=seen, start_desc=start_desc)
+        res = _scan_imap_bounces(cfg, args, start_dt, skip_ids=seen,
+                                 start_desc=start_desc, resume=resume)
     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)
+    scan_mode = "resume" if resume is not None else "full"
+    scan_cursor = ({"uidvalidity": res["uidvalidity"], "last_uid": res["last_uid"]}
+                   if res.get("uidvalidity") is not None and res.get("last_uid") else None)
 
     # ---- 按类别分组处理：只有「可重试」的才剔除补发 ----
     groups = {CAT_RATE: [], CAT_HARD: [], CAT_UNKNOWN: []}
@@ -1296,6 +1509,8 @@ def check_bounces(cfg, args):
                 "script_version": SCRIPT_VERSION,
                 "scan_from": start_dt.strftime("%Y-%m-%d %H:%M:%S"),
                 "scan_from_desc": start_desc,
+                "scan_mode": scan_mode,
+                "scan_cursor": scan_cursor,
                 "scanned": scanned,
                 "bounce_messages": bounce_cnt,
                 "counts": {c: len(groups.get(c, [])) for c in (CAT_RATE, CAT_HARD, CAT_UNKNOWN)},
@@ -1305,12 +1520,10 @@ def check_bounces(cfg, args):
     except Exception as e:  # noqa: BLE001
         print(f"[警告] 写入退信报告失败（不影响本次结果）：{e}")
 
-    if newly_seen:
+    # 落盘：已核查 Message-ID + 新的扫描位置
-        try:
+    # （本次读不到 uidvalidity 时沿用旧游标：只可能往回多扫几封头部，不会漏）
-            with open(DEFAULT_BOUNCE_SEEN, "w", encoding="utf-8") as f:
+    if newly_seen or scan_cursor or cursor:
-                json.dump(sorted(set(newly_seen) | seen), f, ensure_ascii=False)
+        _save_bounce_seen(set(newly_seen) | seen, scan_cursor or cursor)
-        except Exception as e:  # noqa: BLE001
-            print(f"[警告] 记录已核查邮件失败（不影响本次结果）：{e}")
     return bounced
 
 
@@ -1353,7 +1566,8 @@ 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, "mail": {},
             "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}
@@ -1363,7 +1577,8 @@ def load_params(path):
     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:
@@ -1385,9 +1600,77 @@ def load_params(path):
                 data[key] = cast(raw)
             except ValueError:
                 print(f"[警告] params.ini [bounce] {key}={raw!r} 不是合法整数，已忽略")
+
+    # [mail] 程序自带 SMTP（source=own 时取代 conf/app.ini 的 [mail]）
+    if cp.has_section("mail"):
+        m = {}
+        m["source"] = cp.get("mail", "source", fallback="").strip().lower() or "app"
+        for key in ("account", "password", "smtp"):
+            m[key] = cp.get("mail", key, fallback="").strip()
+        m["skip_tls_verify"] = bool(_parse_bool(
+            cp.get("mail", "skip_tls_verify", fallback="").strip(), "skip_tls_verify"))
+        raw = cp.get("mail", "port", fallback="").strip()
+        try:
+            m["port"] = int(raw) if raw else 465
+        except ValueError:
+            print(f"[警告] params.ini [mail] port={raw!r} 不是合法整数，已忽略")
+            m["port"] = 465
+        data["mail"] = m
     return data
 
 
+def resolve_mail_password(raw):
+    """SMTP 密码：支持 env:变量名 从环境变量取值，避免明文写进配置文件。"""
+    raw = (raw or "").strip()
+    if raw.lower().startswith("env:"):
+        name = raw[4:].strip()
+        if not name:
+            return ""
+        return (os.environ.get(name) or "").strip()
+    return raw
+
+
+def apply_mail_source(cfg, args, params, verbose=True):
+    """发信账号来源切换：app（conf/app.ini [mail]，默认）/ own（params.ini [mail]）。
+
+    就地覆盖 cfg 的 mail_* 键，后续 SMTP 发信与 IMAP 退信核查都跟着走这套。
+    优先级：命令行 --mail-source > params.ini [mail] source > app。
+    返回实际生效的来源。
+    """
+    m = (params or {}).get("mail") or {}
+    src = (getattr(args, "mail_source", None) or "").strip().lower()
+    if src not in ("app", "own"):
+        src = (m.get("source") or "app").strip().lower()
+    if src != "own":
+        if verbose:
+            print(f"[发信账号] 来源 = conf/app.ini [mail]"
+                  f"（{cfg['mail_smtp']} / {cfg['mail_account']}）")
+        return "app"
+
+    smtp = (m.get("smtp") or "").strip()
+    account = (m.get("account") or "").strip()
+    raw_pw = (m.get("password") or "").strip()
+    password = resolve_mail_password(raw_pw)
+    missing = [k for k, v in (("smtp", smtp), ("account", account), ("password", password))
+               if not v]
+    if missing:
+        sys.exit(f"[FATAL] params.ini [mail] 缺少或不完整：{', '.join(missing)}。"
+                 f"补全后重试，或改用 --mail-source app（用 conf/app.ini 的 SMTP）")
+    if raw_pw and not raw_pw.lower().startswith("env:"):
+        print("[警告] [mail] password 以明文写在 params.ini（该文件通常会被 git 跟踪），"
+              "建议改成 env:变量名 从环境变量读取，或把 params.ini 加入 .gitignore")
+
+    cfg["mail_smtp"] = smtp
+    cfg["mail_account"] = account
+    cfg["mail_password"] = password
+    cfg["mail_port"] = int(m.get("port") or 465)
+    cfg["mail_skip_tls_verify"] = bool(m.get("skip_tls_verify"))
+    if verbose:
+        print(f"[发信账号] 来源 = 程序自带 params.ini [mail]"
+              f"（{smtp}:{cfg['mail_port']} / {account}）")
+    return "own"
+
+
 def _parse_bool(raw, key=""):
     """解析 1/0、yes/no、true/false、on/off；空串返回 None（表示未配置）。"""
     low = (raw or "").strip().lower()
@@ -1402,7 +1685,27 @@ def _parse_bool(raw, key=""):
     return None
 
 
-def _preserve_extra_sections(path, managed=("path", "site", "vars", "send", "bounce")):
+def _read_mail_section(path):
+    """读取已有 params.ini 的 [mail] 小节（保存时回写，避免被覆盖丢弃）。"""
+    out = {}
+    if not path or not os.path.isfile(path):
+        return out
+    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
+    if not cp.has_section("mail"):
+        return out
+    for k in ("source", "account", "password", "smtp", "port", "skip_tls_verify"):
+        out[k] = cp.get("mail", k, fallback="").strip()
+    return {k: v for k, v in out.items() if v}
+
+
+def _preserve_extra_sections(path, managed=("path", "site", "vars", "send", "bounce", "mail")):
     """读取已有 params.ini 里非托管的小节，原样保留（避免保存时丢掉手写的配置）。"""
     if not path or not os.path.isfile(path):
         return []
@@ -1426,9 +1729,11 @@ def _preserve_extra_sections(path, managed=("path", "site", "vars", "send", "bou
 
 
 def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None,
-                bounce=None):
+                bounce=None, batch_size=0, smtp_idle_reconnect=30.0, mail=None):
     """把参数写回 params.ini（覆盖写；vars 逐行 key = value）。
-    bounce：当前生效的退信/巡检设置，回写进 [bounce] 小节以便下次直接沿用。"""
+    bounce：当前生效的退信/巡检设置，回写进 [bounce] 小节以便下次直接沿用。
+    batch_size / smtp_idle_reconnect / mail：同属 [send]/[mail] 小节，必须回写，
+    否则覆盖写会把手工加的键抹掉（mail 的小节值从原文件读回，只覆盖传入项）。"""
     lines = ["; broadcast.py 参数文件：每次运行自动读取；命令行参数优先级更高",
              "; 注意：本文件不控制 --send/--yes，正式发送仍需命令行显式指定", ""]
     lines.append("[path]")
@@ -1448,6 +1753,10 @@ 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("; SMTP 连接空闲超过该秒数就主动重连再发（delay 调大时必开，0 关闭）")
+    lines.append(f"smtp_idle_reconnect = {float(smtp_idle_reconnect):g}")
+    lines.append("; 合并信封批量发送：渲染内容完全相同的收件人每 N 人共用一个信封（0=逐封）")
+    lines.append(f"batch_size = {int(batch_size)}")
     lines.append("")
     lines.append("; 退信核查 / 发送中限流巡检（均可用同名命令行参数临时覆盖）")
     lines.append("[bounce]")
@@ -1470,6 +1779,24 @@ def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None
     lines.append("; since = auto 表示从断点清单最早一条记录的时间开始扫描")
     lines.append(f"since = {b.get('since', 'auto')}")
     lines.append(f"since_days = {b.get('since_days', 3)}")
+    lines.append("")
+    # [mail]：值从原文件读回（含 env: 形式的密码），再用传入的覆盖项更新
+    mv = {"source": "app", "account": "", "password": "", "smtp": "", "port": "465",
+          "skip_tls_verify": "0"}
+    mv.update(_read_mail_section(path))
+    for k, v in (mail or {}).items():
+        if v not in (None, ""):
+            mv[k] = str(v)
+    lines.append("; 发信账号来源：app = 用 conf/app.ini [mail] 的 SMTP（默认）；")
+    lines.append(";                  own = 用下面这套程序自带 SMTP（--mail-source 可临时覆盖）")
+    lines.append("[mail]")
+    lines.append(f"source = {mv['source'] or 'app'}")
+    lines.append(f"account = {mv['account']}")
+    lines.append("; password 支持 env:变量名（推荐），避免明文写进会被 git 跟踪的文件")
+    lines.append(f"password = {mv['password']}")
+    lines.append(f"smtp = {mv['smtp']}")
+    lines.append(f"port = {mv['port'] or 465}")
+    lines.append(f"skip_tls_verify = {mv['skip_tls_verify'] or 0}")
     lines.extend(_preserve_extra_sections(path))
     with open(path, "w", encoding="utf-8") as f:
         f.write("\n".join(lines) + "\n")
@@ -1514,6 +1841,15 @@ def run_interactive(cfg, templates, extra_vars, args, config_path):
           f"（每 {args.watch_every} 封一次）"
           f"   来源：params.ini [bounce] / 命令行，改值请编辑 params.ini")
 
+    # 发信账号来源：只有 params.ini [mail] 配齐了才提供切换
+    if getattr(args, "_own_mail_ready", False):
+        cur = (getattr(args, "mail_source", None) or "").strip().lower()
+        default_src = cur if cur in ("app", "own") else "app"
+        src = ask("发信账号来源（app = conf/app.ini 的 SMTP；own = 程序自带 params.ini [mail]）",
+                  default_src).strip().lower()
+        args.mail_source = "own" if src == "own" else "app"
+        print(f"  发信账号 = {'程序自带 params.ini [mail]' if args.mail_source == 'own' else 'conf/app.ini [mail]'}")
+
     # 站点地址：只有完全没值时才问（值链已在 main 里兜底过 external_url）
     if not site_url:
         site_url = ask("站点地址 site_url（如 https://box.shiroko.one）", "")
@@ -1605,6 +1941,9 @@ def run_interactive(cfg, templates, extra_vars, args, config_path):
         save_ans = ""
     if save_ans in ("y", "yes"):
         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,
+                    mail={"source": getattr(args, "mail_source", None) or "app"},
                     bounce={
                         "watch_bounces": args.watch_bounces,
                         "watch_every": args.watch_every,
@@ -1662,6 +2001,16 @@ def build_arg_parser():
                         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")
@@ -1681,6 +2030,12 @@ def build_arg_parser():
                         help=f"参数文件路径（默认自动读取脚本同目录 params.ini，不存在则跳过；CLI 参数优先级更高）")
     parser.add_argument("--no-params", action="store_true",
                         help="不读取 params.ini 参数文件")
+    parser.add_argument("--mail-source", default=None, choices=["app", "own"],
+                        help="发信账号来源：app = conf/app.ini [mail]（默认）；"
+                             "own = 程序自带 SMTP（读 params.ini [mail] 的"
+                             " account/password/smtp/port，password 支持 env:变量名）。"
+                             "命令行优先于 params.ini [mail] source。切换后发信与"
+                             "退信核查（IMAP）都走这套账号")
     parser.add_argument("--interactive", "-i", action="store_true", default=None,
                         help="交互模式（默认开启：不带任何运行参数运行脚本时自动进入）。"
                              "运行后逐项询问参数，并直接选择运行方式（演练/测试一封/群发）")
@@ -1699,6 +2054,9 @@ def build_arg_parser():
     parser.add_argument("--prune-unknown", action="store_true", default=None,
                         help="配合 --check-bounces：无法分类的退信也按「可重试」处理，一并剔除补发"
                              "（默认不剔除，避免给死信地址反复重发；也可在 params.ini [bounce] 设置）")
+    parser.add_argument("--full-scan", action="store_true",
+                        help="退信核查忽略上次扫描位置，按日期窗口从头全量扫描"
+                             "（默认记住上次扫到的位置只查新邮件；显式 --since 也会忽略位置）")
     parser.add_argument("--invalid-file", default=None, metavar="FILE",
                         help=f"无效地址清单（硬退信地址）路径；默认脚本同目录 broadcast_invalid.json。"
                              f"清单里的地址每次运行都会跳过，即使 --reset-state")
@@ -1788,6 +2146,11 @@ 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:
@@ -1836,10 +2199,20 @@ def main():
         elif not args.interactive:
             print("[警告] site_url 为空：{{site}} 与无个性域名用户的 {{box_link}} 将为空串")
 
+    # 程序自带 SMTP 是否已配齐（交互模式据此提供来源切换）
+    _m = params.get("mail") or {}
+    args._own_mail_ready = bool((_m.get("smtp") or "").strip()
+                                and (_m.get("account") or "").strip()
+                                and (_m.get("password") or "").strip())
+
     if args.interactive:
         site_url, args.box_prefix, extra_vars = run_interactive(
             cfg, templates, extra_vars, args, config_path)
 
+    # ---- 发信账号来源切换（app.ini / 程序自带 params.ini [mail]）----
+    # 必须在交互模式之后（交互里可能改选）、退信核查与发信之前（两者都用 cfg）
+    mail_source = apply_mail_source(cfg, args, params)
+
     # ---- 退信核查模式：不查数据库、不发信；扫描（含可选剔除）后结束，
     #      除非同时指定 --send（剔除并补发），此时继续走正常发送流程 ----
     if args.check_bounces:
@@ -1870,6 +2243,8 @@ def main():
     print(f"  [db]     host:port    = {cfg['db_host']}:{cfg['db_port']} db={cfg['db_name'] or '(空)'}")
     print(f"  [mail]   smtp:port    = {cfg['mail_smtp']}:{cfg['mail_port']}")
     print(f"  [mail]   account      = {cfg['mail_account'] or '(空)'}")
+    print(f"  [mail]   来源         = "
+          f"{'程序自带 params.ini [mail]' if mail_source == 'own' else 'conf/app.ini [mail]'}")
     print(f"  [断点续发] state_file = {args.state}")
     print("=" * 56)
 
@@ -2081,6 +2456,9 @@ def main():
     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
@@ -2089,9 +2467,48 @@ def main():
     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"，合并信封每 {batch_size} 人/封" if batch_size > 0 else "")
           + (f"，每 {watch_every} 封巡检一次退信" if watch_active else "") + " ...")
 
     if args.single:
@@ -2106,7 +2523,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}")
 
@@ -2120,6 +2537,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 = {
@@ -2132,90 +2558,124 @@ 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)
-                smtp_err = ""
+                digest = hashlib.sha256(
-                try:
+                    "\x00".join((s, h, p)).encode("utf-8", "replace")).hexdigest()
-                    refused = server.send_message(msg)
+                bkt = bucket_map.get(digest)
-                    if refused:
+                if bkt is None:
-                        refused_desc = "; ".join(f"{a} {c}:{err}" for a, (c, err) in refused.items())
+                    bkt = {"content": (s, h, p), "queue": []}
-                        smtp_err = refused_desc
+                    bucket_map[digest] = bkt
-                        failed.append((r["email"], f"SMTP 拒收 {refused_desc}"))
+                    buckets.append(bkt)
-                        smtp_rejected += 1
+                bkt["queue"].append(r)
-                        print(f"  [SMTP 拒收] {r['email']}: {refused_desc}")
+            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
-                    smtp_err = str(e)
-                    failed.append((r["email"], smtp_err))
-                    print(f"  [失败] {r['email']}: {e}")
 
-                attempted += 1
+                    # 「单封收件人数超限」的拒绝：砍半批次重试，被拒的人排回队首；
-                # SMTP 当场就返回限流信号 → 继续发只会更糟，立即中止
+                    # 未被拒的收件人已随本次 DATA 投出，照常入账
-                if smtp_err:
+                    oversized = [r for r in batch
-                    cat, why = classify_bounce("", smtp_err, "")
+                                 if _is_too_many_rcpts(err_map.get(r["email"].lower(), ""))]
-                    if cat == CAT_HARD and not args.no_invalid_list:
+                    if oversized:
-                        record_invalid(args.invalid_file, {r["email"]: {
+                        batch_size = max(1, batch_size // 2)
-                            "reason": why, "detail": smtp_err[:160], "subject": subject,
+                        queue[:0] = oversized
-                            "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}})
+                        over_set = {r["email"].lower() for r in oversized}
-                    elif cat == CAT_RATE and not abort_reason:
+                        print(f"  [拆批] {len(oversized)} 人被拒（单封收件人数超限），"
-                        abort_reason = f"SMTP 返回疑似限流（{why}）：{smtp_err[:100]}"
+                              f"batch_size 降为 {batch_size} 后重试")
-
-                # 周期巡检收件箱：异步退信才是限流最常见的表现
-                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
+                        over_set = set()
-                        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:
+                    for r in batch:
-                    break
+                        em_low = r["email"].lower()
-                if sent < total:
+                        e = err_map.get(em_low)
-                    time.sleep(max(0.0, args.delay))
+                        if e is None or em_low in over_set:
-                    if args.pause_every > 0 and sent % args.pause_every == 0:
+                            continue
-                        print(f"  [间隔] 已发 {sent} 封，暂停 {args.pause_for}s 防限流 ...")
+                        failed.append((r["email"], f"SMTP 拒收 {e}"))
-                        time.sleep(max(0.0, args.pause_for))
+                        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

params.ini

@@ -18,6 +18,31 @@ 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
+
+; 发信账号来源：app = 用 conf/app.ini [mail] 的 SMTP（默认，站内信跟随站点邮箱）
+;                 own = 用下面这套程序自带 SMTP（命令行 --mail-source own 可临时覆盖）
+; 切到 own 后，发信与退信核查（IMAP）都走这套账号
+[mail]
+source = app
+account =
+; password 支持 env:变量名（推荐），避免明文写进会被 git 跟踪的文件
+password =
+smtp =
+port = 465
+skip_tls_verify = 0
 
 ; 退信核查 / 发送中限流巡检开关
 ; 这些都能在 params.ini 里长期配置，命令行同名参数可临时覆盖