合并信封批量发送 --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

@@ -42,6 +42,7 @@ README — TamaBox 站内信群发工具（mail-broadcast）
   delay       = 1.0
   group_pause = 3.0
   smtp_idle_reconnect = 30                  ; SMTP 空闲超时秒数，超时就重连
+  batch_size  = 0                         ; 合并信封：内容相同的收件人每 N 人一封（0=逐封）
 
   [bounce]                                ; 退信核查 / 发送中限流巡检（见下节）
   watch_bounces  = 1                      ; 发送中巡检限流退信，发现即中止

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-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")
@@ -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):
@@ -680,8 +700,15 @@ class SmtpSession:
                 pass
             self.server = None
 
-    def send_message(self, msg):
+    def send_to(self, msg, rcpt_addrs=None):
-        """带重连的 send_message；失败时抛最后一次的原始异常。"""
+        """带重连的发送；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()
@@ -693,9 +720,17 @@ class SmtpSession:
                     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()
@@ -706,6 +741,33 @@ class SmtpSession:
                     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 追加写，崩溃安全）
@@ -1424,6 +1486,7 @@ def load_params(path):
 
     data = {"config": "", "site_url": "", "box_prefix": "", "vars": {}, "to": "",
             "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}
@@ -1434,7 +1497,7 @@ def load_params(path):
         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),
-                      ("smtp_idle_reconnect", float)):
+                      ("smtp_idle_reconnect", float), ("batch_size", int)):
         raw = cp.get("send", key, fallback="").strip()
         if raw:
             try:
@@ -1676,6 +1739,8 @@ 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,
                     bounce={
                         "watch_bounces": args.watch_bounces,
                         "watch_every": args.watch_every,
@@ -1737,6 +1802,12 @@ def build_arg_parser():
                         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")
@@ -1866,6 +1937,8 @@ def main():
     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:
@@ -2159,6 +2232,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
@@ -2167,9 +2243,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:
@@ -2198,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 = {
@@ -2210,68 +2334,103 @@ 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 = session.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,
+                                                     "ts": time.time()})
+                    attempted += len(batch)
+
+                    # 「单封收件人数超限」的拒绝：砍半批次重试，被拒的人排回队首；
+                    # 未被拒的收件人已随本次 DATA 投出，照常入账
+                    oversized = [r for r in batch
+                                 if _is_too_many_rcpts(err_map.get(r["email"].lower(), ""))]
+                    if oversized:
+                        batch_size = max(1, batch_size // 2)
+                        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}")
-                except Exception as e:  # noqa: BLE001
-                    smtp_err = str(e)
-                    failed.append((r["email"], smtp_err))
-                    print(f"  [失败] {r['email']}: {e}")
 
-                attempted += 1
+                    # 周期巡检收件箱：按收件人计数，异步退信才是限流最常见表现
-                # SMTP 当场就返回限流信号 → 继续发只会更糟，立即中止
+                    if watch_active and not abort_reason and attempted >= watch_mark + watch_every:
-                if smtp_err:
+                        watch_mark = attempted
-                    cat, why = classify_bounce("", smtp_err, "")
+                        run_patrol()
-                    if cat == CAT_HARD and not args.no_invalid_list:
-                        record_invalid(args.invalid_file, {r["email"]: {
-                            "reason": why, "detail": smtp_err[:160], "subject": subject,
-                            "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}})
-                    elif cat == CAT_RATE and not abort_reason:
-                        abort_reason = f"SMTP 返回疑似限流（{why}）：{smtp_err[:100]}"
-
-                # 周期巡检收件箱：异步退信才是限流最常见的表现
-                if watch_active and not abort_reason and attempted % watch_every == 0:
-                    nb, ok = watch_bounces(cfg, args, watch_start, known_ids)
-                    if not ok:
-                        watch_fails += 1
-                        if watch_fails >= 2:
-                            print("  [退信巡检] IMAP 连续不可用，本次已关闭巡检"
-                                  "（不影响发送；发完请手动跑一次 --check-bounces）")
-                            watch_active = False
-                    else:
-                        watch_fails = 0
-                        for em, info in nb.items():
-                            if info["category"] == CAT_RATE:
-                                watch_rate[em] = info
-                            elif info["category"] == CAT_HARD:
-                                watch_hard[em] = info
-                        if watch_hard:
-                            if not args.no_invalid_list:
-                                record_invalid(args.invalid_file, watch_hard)
-                            watch_hard = {}
-                        if watch_rate:
-                            abort_reason = (
-                                f"收件箱出现 {len(watch_rate)} 条限流退信"
-                                f"（如 {sorted(watch_rate)[0]}："
-                                f"{watch_rate[sorted(watch_rate)[0]]['reason']}）")
 
                     if abort_reason:
                         break
@@ -2291,6 +2450,8 @@ def main():
     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

@@ -23,6 +23,15 @@ group_pause = 3.0
 ; 「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]