From 6dc33604d10c88dbcad0111365228c0a83b82fa1 Mon Sep 17 00:00:00 2001 From: tamakyi Date: Mon, 7 Sep 2026 21:25:48 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=88=E5=B9=B6=E4=BF=A1=E5=B0=81=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E5=8F=91=E9=80=81=20--batch-size=EF=BC=9A=E5=90=8C?= =?UTF-8?q?=E5=86=85=E5=AE=B9=E5=A4=9A=E6=94=B6=E4=BB=B6=E4=BA=BA=E4=B8=80?= =?UTF-8?q?=E5=B0=81=E5=A4=9A=E6=8A=95=EF=BC=8C452=20=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E6=8B=86=E6=89=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 渲染后内容 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 | 1 + broadcast.py | 307 +++++++++++++++++++++++++++++++++++++++------------ params.ini | 9 ++ 3 files changed, 244 insertions(+), 73 deletions(-) diff --git a/README.md b/README.md index c061172..0327ea5 100644 --- a/README.md +++ b/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 ; 发送中巡检限流退信，发现即中止 diff --git a/broadcast.py b/broadcast.py index 5c7c51d..2524f07 100644 --- a/broadcast.py +++ b/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): - subject, miss1 = render_strict(subject, ctx) - html, miss2 = render_strict(html, ctx) - plain, miss3 = render_strict(plain, ctx) +def render_content(subject, html, plain, ctx): + """渲染 subject/html/plain 三段；返回 (渲染结果三元组, 缺失占位符列表)。 + 拆出来是为了批量发送：先渲染、按内容指纹分桶，内容完全相同的收件人 + 才共用一个 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): - """带重连的 send_message；失败时抛最后一次的原始异常。""" + 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() @@ -693,9 +720,17 @@ class SmtpSession: print(f" [重连] SMTP 连接已断开，正在重新登录" f"（第 {self.reconnects} 次）...") self._connect() - refused = self.server.send_message(msg) + 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,76 +2334,111 @@ def main(): "year": datetime.now().strftime("%Y"), **extra_vars, } - subject = args.subject if args.subject else tpl["subject"] - 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) + s, h, p, missing = render_content(subject_raw, html_raw, plain_raw, ctx) if missing: all_missing.update(missing) - smtp_err = "" - try: - refused = session.send_message(msg) - if refused: - refused_desc = "; ".join(f"{a} {c}:{err}" for a, (c, err) in refused.items()) - smtp_err = refused_desc - failed.append((r["email"], f"SMTP 拒收 {refused_desc}")) - smtp_rejected += 1 - print(f" [SMTP 拒收] {r['email']}: {refused_desc}") + digest = hashlib.sha256( + "\x00".join((s, h, p)).encode("utf-8", "replace")).hexdigest() + bkt = bucket_map.get(digest) + if bkt is None: + bkt = {"content": (s, h, p), "queue": []} + bucket_map[digest] = bkt + 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()}) - 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}") + record_sent(args.state, {"email": r["email"], "lang": lg, + "ts": time.time()}) + attempted += len(batch) - attempted += 1 - # SMTP 当场就返回限流信号 → 继续发只会更糟，立即中止 - if smtp_err: - cat, why = classify_bounce("", smtp_err, "") - if cat == CAT_HARD and not args.no_invalid_list: - record_invalid(args.invalid_file, {r["email"]: { - "reason": why, "detail": smtp_err[:160], "subject": subject, - "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}}) - elif cat == CAT_RATE and not abort_reason: - abort_reason = f"SMTP 返回疑似限流（{why}）：{smtp_err[:100]}" - - # 周期巡检收件箱：异步退信才是限流最常见的表现 - if watch_active and not abort_reason and attempted % watch_every == 0: - nb, ok = watch_bounces(cfg, args, watch_start, known_ids) - if not ok: - watch_fails += 1 - if watch_fails >= 2: - print(" [退信巡检] IMAP 连续不可用，本次已关闭巡检" - "（不影响发送；发完请手动跑一次 --check-bounces）") - watch_active = False + # 「单封收件人数超限」的拒绝：砍半批次重试，被拒的人排回队首； + # 未被拒的收件人已随本次 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: - 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']}）") + over_set = set() - 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)) + 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)) @@ -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 diff --git a/params.ini b/params.ini index 71785a7..4ce3d74 100644 --- a/params.ini +++ b/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]