From 1cae63b671ada9c9cd3da091126e6df4f6e426c3 Mon Sep 17 00:00:00 2001 From: tamakyi Date: Mon, 7 Sep 2026 19:11:14 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E9=80=81=E4=B8=AD=E5=B7=A1=E6=A3=80?= =?UTF-8?q?=E9=99=90=E6=B5=81=E9=80=80=E4=BF=A1=EF=BC=9A=E5=8F=91=E7=8E=B0?= =?UTF-8?q?=E5=8D=B3=E4=B8=AD=E6=AD=A2=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景：限流退信是异步投到发件箱的——SMTP 当场返回 250，几十秒后收件箱 才收到「您的账号外发频率超过邮件系统限制」。只等发完再核查，等发现时 被限流的那一批早就全废了，白耗额度。 改动： - 把 IMAP 扫描抽成 _scan_imap_bounces()，供退信核查与发送中巡检共用 （新增 not_before 参数：只看指定时刻之后到达的退信） - 发送过程中每 --watch-every 封（默认 10）巡检一次收件箱，命中限流类 退信立即中止；只看本次运行开始后的新退信，历史退信不误触发 - SMTP 当场返回限流（如 450 MI:CEL）也立即中止；硬退信则照旧记入 无效地址清单并继续发下一封 - 中止后把限流地址从断点清单剔除，等限制恢复重跑即自动补发 - IMAP 连续两轮不可用则自动关闭本次巡检并提示，不影响发送 - 新增 --no-watch-bounces / --watch-every，报告新增 aborted 等字段 - 顺带：分类依据里的状态码改为只显示匹配到的码（原来是整串原文） --- README.md | 26 +++++ broadcast.py | 299 +++++++++++++++++++++++++++++++++++++-------------- 2 files changed, 247 insertions(+), 78 deletions(-) diff --git a/README.md b/README.md index dc736fc..bd7a304 100644 --- a/README.md +++ b/README.md @@ -180,6 +180,32 @@ conf/app.ini 路径的解析优先级： - --since 2026-09-07 可显式指定起点日期 - 断点清单为空时回退为 --since-days N（默认 3 天） +发送中限流巡检（边发边看，及时止损） +---------------------------------- +限流退信是**异步**投到发件箱的——SMTP 当场返回 250，几十秒后收件箱才收到 +「您的账号外发频率超过邮件系统限制」。如果只等发完再核查，等发现时限流那一批 +早就全废了。所以正式群发时会**边发边巡检收件箱**： + + - 每 `--watch-every` 封（默认 10）登录一次 IMAP 查新退信 + - 只看**本次运行开始之后**到达的退信，历史退信不会误触发 + - 一旦出现限流类退信 → 立即停止发送 + - SMTP 当场就返回限流（如 `450 MI:CEL 发送频率超限`）→ 同样立即停止 + - 善后：把限流退信的地址从断点清单剔除，等限制恢复后重跑自动补发 + - 巡检到的「地址不存在」类退信照旧写入无效地址清单 + + python3 broadcast.py --send --yes # 默认已开启巡检 + python3 broadcast.py --send --yes --watch-every 5 # 每 5 封查一次（更及时，IMAP 登录更频繁） + python3 broadcast.py --send --yes --no-watch-bounces # 关闭巡检（--to 单封测试本就关闭） + +中止时的输出示例： + + [中止发送] 收件箱出现 1 条限流退信（如 a@b.com：关键词「您的账号外发频率超过邮件系统限制」） + [善后] 已将 1 个限流退信地址从断点清单剔除，等限制恢复后重跑同一条命令即可自动补发。 + 发送中止：已成功 30 封，失败 0 封，未发送 30 封。 + +注意：巡检依赖 IMAP 可用。若连续两轮连不上 IMAP，会自动关闭本次巡检并提示 +（不影响发送），此时请发完手动跑一次 `--check-bounces`。 + 无效地址清单 broadcast_invalid.json ----------------------------------- 硬退信（地址不存在等）的地址会写进这里，之后**每次运行都直接跳过**， diff --git a/broadcast.py b/broadcast.py index c42e8ad..7927ca4 100644 --- a/broadcast.py +++ b/broadcast.py @@ -32,6 +32,9 @@ TamaBox 站内信群发工具（独立项目，零依赖，不依赖、不修改 - SMTP 拒收捕获：send_message 返回的 refused 计入失败 - 退信分类核查：只有「限流/临时性」退信才剔除补发；「地址不存在」这类 永久失败不补发，并记入无效地址清单（broadcast_invalid.json）永久跳过 + - 发送中限流巡检：每 --watch-every 封查一次收件箱，一旦出现 + 「外发频率超过邮件系统限制」类退信（或 SMTP 当场报限流）立即中止发送， + 并把这批限流地址从断点清单剔除，等限制恢复后重跑自动补发 用法示例： # 1) 演练：列出收件人 + 统计（不发信） @@ -73,7 +76,7 @@ import time from email.header import Header, decode_header from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText -from email.utils import formataddr, make_msgid +from email.utils import formataddr, make_msgid, parsedate_to_datetime from urllib.parse import quote_plus SCRIPT_VERSION = "2026-09-07.bounceclass.v3" @@ -871,10 +874,9 @@ def classify_bounce(status="", diagnostic="", text=""): """ for src in (status, diagnostic): c = _smtp_code_class(src or "") - if c == "5": - return CAT_HARD, f"DSN 状态码 {(src or '').strip()[:40]}" - if c == "4": - return CAT_RATE, f"DSN 状态码 {(src or '').strip()[:40]}" + if c in ("5", "4"): + m = _SMTP_ENHANCED_RE.search(src or "") or _SMTP_CODE_RE.search(src or "") + return (CAT_HARD if c == "5" else CAT_RATE), f"SMTP 状态码 {m.group(0)}" body = "\n".join(x for x in (text, diagnostic) if x) low = body.lower() @@ -1061,6 +1063,23 @@ def reset_invalid(path): return False +def watch_bounces(cfg, args, not_before, known_ids, verbose=True): + """发送过程中的轻量巡检：只看 not_before 之后新到的退信。 + + 不写任何文件、不动断点清单（善后交给 --check-bounces）。 + 返回 (bounced, ok)：ok=False 表示 IMAP 不可用，调用方应降级处理。 + """ + try: + res = _scan_imap_bounces(cfg, args, not_before, skip_ids=known_ids, + not_before=not_before, verbose=False) + except Exception as e: # noqa: BLE001 + if verbose: + print(f" [退信巡检] 本轮跳过：IMAP 检查失败（{e}）") + return {}, False + known_ids.update(res["touched_ids"]) + return res["bounced"], True + + def resolve_scan_start(args): """退信扫描起点：--since 显式日期 > auto 时取断点清单最早一条发送记录的 时间（精确到该时刻，不加余量）> 无记录时回退 --since-days。 @@ -1082,37 +1101,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 +1222,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() - except Exception: # noqa: BLE001 - pass + res = _scan_imap_bounces(cfg, args, start_dt, skip_ids=seen, start_desc=start_desc) + except Exception as e: # noqa: BLE001 + sys.exit(f"[FATAL] IMAP 扫描失败（请确认邮箱已开启 IMAP，密码用邮箱登录密码）: {e}") + + bounced = res["bounced"] + newly_seen, scanned = res["new_ids"], res["scanned"] + bounce_cnt = len(newly_seen) # ---- 按类别分组处理：只有「可重试」的才剔除补发 ---- groups = {CAT_RATE: [], CAT_HARD: [], CAT_UNKNOWN: []} @@ -1548,6 +1609,13 @@ def main(): help="本次不使用无效地址清单：既不跳过已知无效地址，也不写入新的硬退信") parser.add_argument("--bounce-report", default=DEFAULT_BOUNCE_REPORT, metavar="FILE", help="退信核查明细报告路径（JSON）；默认脚本同目录 broadcast_bounce_report.json") + parser.add_argument("--watch-bounces", dest="watch_bounces", action="store_true", default=True, + help="发送过程中巡检收件箱：每 --watch-every 封查一次，" + "一旦出现「外发频率超过邮件系统限制」等限流退信立即中止发送（默认开启）") + parser.add_argument("--no-watch-bounces", dest="watch_bounces", action="store_false", + help="关闭发送中巡检（--to 单封测试时本就关闭）") + parser.add_argument("--watch-every", type=int, default=10, metavar="N", + help="发送中每发 N 封巡检一次退信（默认 10；调小会更频繁登录 IMAP）") parser.add_argument("--imap-host", default="", help="IMAP 服务器地址（默认由 SMTP 域名推导：smtp.xxx → imap.xxx）") parser.add_argument("--imap-port", type=int, default=993, @@ -1825,6 +1893,9 @@ def main(): "smtp_rejected": 0, "skipped_already_sent": skipped, "skipped_invalid": invalid_skipped, + "aborted": None, + "abort_rate_bounces": [], + "watch_bounces": bool(getattr(args, "watch_bounces", True)) and not args.to, "failed_list": [], "missing_placeholders": [], "test_mode": bool(args.to), @@ -1882,11 +1953,21 @@ def main(): sent, failed = 0, [] smtp_rejected = 0 + attempted = 0 total = len(recipients) all_missing = set() + + # 发送中巡检：限流退信是异步投到发件箱的，边发边看才能及时止损 + watch_active = bool(getattr(args, "watch_bounces", True)) and not args.to + watch_start = datetime.now() + watch_every = max(1, getattr(args, "watch_every", 10) or 10) + known_ids, watch_fails = set(), 0 + watch_rate, watch_hard, abort_reason = {}, {}, "" + print(f"\n[发送中] 目标 {total} 封，每封间隔 {args.delay}s" + (f"，每组间隔 {args.group_pause}s" if args.group_pause > 0 else "") - + (f"，每 {args.pause_every} 封暂停 {args.pause_for}s" if args.pause_every > 0 else "") + " ...") + + (f"，每 {args.pause_every} 封暂停 {args.pause_for}s" if args.pause_every > 0 else "") + + (f"，每 {watch_every} 封巡检一次退信" if watch_active else "") + " ...") if args.single: # 单文件模式：不按语言分组，所有人同一模板 @@ -1932,10 +2013,12 @@ def main(): msg, missing = build_message(cfg, r["email"], ctx, subject, html, plain) if missing: all_missing.update(missing) + smtp_err = "" try: refused = server.send_message(msg) if refused: refused_desc = "; ".join(f"{a} {c}:{err}" for a, (c, err) in refused.items()) + smtp_err = refused_desc failed.append((r["email"], f"SMTP 拒收 {refused_desc}")) smtp_rejected += 1 print(f" [SMTP 拒收] {r['email']}: {refused_desc}") @@ -1946,8 +2029,49 @@ def main(): if sent % 25 == 0 or sent == total: print(f" 进度 {sent}/{total} 已成功 {sent}") except Exception as e: # noqa: BLE001 - failed.append((r["email"], str(e))) + smtp_err = str(e) + failed.append((r["email"], smtp_err)) print(f" [失败] {r['email']}: {e}") + + attempted += 1 + # SMTP 当场就返回限流信号 → 继续发只会更糟，立即中止 + if smtp_err: + cat, why = classify_bounce("", smtp_err, "") + if cat == CAT_HARD and not args.no_invalid_list: + record_invalid(args.invalid_file, {r["email"]: { + "reason": why, "detail": smtp_err[:160], "subject": subject, + "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}}) + elif cat == CAT_RATE and not abort_reason: + abort_reason = f"SMTP 返回疑似限流（{why}）：{smtp_err[:100]}" + + # 周期巡检收件箱：异步退信才是限流最常见的表现 + if watch_active and not abort_reason and attempted % watch_every == 0: + nb, ok = watch_bounces(cfg, args, watch_start, known_ids) + if not ok: + watch_fails += 1 + if watch_fails >= 2: + print(" [退信巡检] IMAP 连续不可用，本次已关闭巡检" + "（不影响发送；发完请手动跑一次 --check-bounces）") + watch_active = False + else: + watch_fails = 0 + for em, info in nb.items(): + if info["category"] == CAT_RATE: + watch_rate[em] = info + elif info["category"] == CAT_HARD: + watch_hard[em] = info + if watch_hard: + if not args.no_invalid_list: + record_invalid(args.invalid_file, watch_hard) + watch_hard = {} + if watch_rate: + abort_reason = ( + f"收件箱出现 {len(watch_rate)} 条限流退信" + f"（如 {sorted(watch_rate)[0]}：" + f"{watch_rate[sorted(watch_rate)[0]]['reason']}）") + + if abort_reason: + break if sent < total: time.sleep(max(0.0, args.delay)) if args.pause_every > 0 and sent % args.pause_every == 0: @@ -1956,6 +2080,8 @@ def main(): if args.group_pause > 0 and gi < len(group_order): print(f" [组间间隔] 下一组前暂停 {args.group_pause}s ...") time.sleep(max(0.0, args.group_pause)) + if abort_reason: + break finally: try: server.quit() @@ -1965,12 +2091,29 @@ def main(): report["sent"] = sent report["failed"] = len(failed) report["smtp_rejected"] = smtp_rejected + report["aborted"] = abort_reason or None + report["abort_rate_bounces"] = sorted(watch_rate) + report["watch_bounces"] = watch_active report["failed_list"] = [{"email": e, "error": err} for e, err in failed] report["missing_placeholders"] = sorted(all_missing) write_report_now() print("\n" + "=" * 56) - 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: