SMTP 断线自动重连：修掉大 delay 下整批 please run connect() first

现象：--delay 调到 60s/封、发 305 人时，第一封报 Server not connected，
之后每一封都是 please run connect() first，全军覆没。

根因：SMTP 服务器会掐掉空闲连接（QQ/163/阿里云常见 5~15 分钟，有的更短）。
脚本只在开头 smtp_login 一次，之后从不重连——连接一旦被服务器关掉，
同一个死掉的连接对象会被反复使用，除第一封外全是 connect() 报错。
delay 越大必然越早触发；以前默认 1s/封时几乎等不到超时，所以没暴露过。

改动：
- 新增 SmtpSession 包装连接：
  * 空闲超过 smtp_idle_reconnect 秒（默认 30，可 0 关闭）就主动断开重连，
    不必每封都先吃一发失败
  * 仍遇到断线类异常（ServerDisconnected/ConnectError/OSError）立即重连
    重试，最多 3 次（间隔 2s/4s）；耗尽后抛原始异常、该收件人如实计失败
  * 只重试连接类异常；SMTP 拒收（5xx/4xx 回话）不重试，避免重复投递
  * 「Server not connected」不会被误判为限流，不会误触发发送中止
- 新增 --smtp-idle-reconnect / params.ini [send] smtp_idle_reconnect
- 发送循环改用 SmtpSession，结束统一 close

README.md

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

broadcast.py

@@ -32,6 +32,12 @@ TamaBox 站内信群发工具（独立项目，零依赖，不依赖、不修改
   - SMTP 拒收捕获：send_message 返回的 refused 计入失败
   - 退信分类核查：只有「限流/临时性」退信才剔除补发；「地址不存在」这类
     永久失败不补发，并记入无效地址清单（broadcast_invalid.json）永久跳过
+  - 发送中限流巡检：每 N 封查一次收件箱，一旦出现
+    「外发频率超过邮件系统限制」类退信（或 SMTP 当场报限流）立即中止发送，
+    并把这批限流地址从断点清单剔除，等限制恢复后重跑自动补发
+
+参数优先级：命令行 > params.ini > 内置默认。退信/巡检相关开关（[bounce] 小节）
+与站点、限速等一样都能在 params.ini 里长期配置，命令行只在需要临时覆盖时用。
 
 用法示例：
   # 1) 演练：列出收件人 + 统计（不发信）
@@ -73,7 +79,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"
@@ -631,6 +637,76 @@ 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_message(self, msg):
+        """带重连的 send_message；失败时抛最后一次的原始异常。"""
+        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()
+                refused = self.server.send_message(msg)
+                self.last_used = time.time()
+                return refused
+            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
+
+
 # ---------------------------------------------------------------------------
 # 断点续发状态（JSONL 追加写，崩溃安全）
 # ---------------------------------------------------------------------------
@@ -871,10 +947,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 +1136,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 +1174,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 +1295,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: []}
@@ -1289,25 +1423,83 @@ 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,
+            "watch_bounces": None, "watch_every": None, "prune_unknown": None,
+            "no_invalid_list": None, "invalid_file": "", "bounce_report": "",
+            "imap_host": "", "imap_port": None, "since": "", "since_days": None}
     data["config"] = cp.get("path", "config", fallback="").strip()
     data["site_url"] = cp.get("site", "site_url", fallback="").strip().rstrip("/")
     data["box_prefix"] = cp.get("site", "box_prefix", fallback="").strip()
     if cp.has_section("vars"):
         data["vars"] = {k: v.strip() for k, v in cp.items("vars")}
     data["to"] = cp.get("send", "to", fallback="").strip()
-    for key, cast in (("limit", int), ("delay", float), ("group_pause", float)):
+    for key, cast in (("limit", int), ("delay", float), ("group_pause", float),
+                      ("smtp_idle_reconnect", float)):
         raw = cp.get("send", key, fallback="").strip()
         if raw:
             try:
                 data[key] = cast(raw)
             except ValueError:
                 print(f"[警告] params.ini [send] {key}={raw!r} 不是合法数字，已忽略")
+
+    # [bounce] 退信核查 / 发送中巡检的开关
+    for key in ("watch_bounces", "prune_unknown", "no_invalid_list"):
+        val = _parse_bool(cp.get("bounce", key, fallback="").strip(), key)
+        if val is not None:
+            data[key] = val
+    for key in ("invalid_file", "bounce_report", "imap_host", "since"):
+        data[key] = cp.get("bounce", key, fallback="").strip()
+    for key, cast in (("watch_every", int), ("imap_port", int), ("since_days", int)):
+        raw = cp.get("bounce", key, fallback="").strip()
+        if raw:
+            try:
+                data[key] = cast(raw)
+            except ValueError:
+                print(f"[警告] params.ini [bounce] {key}={raw!r} 不是合法整数，已忽略")
     return data
 
 
-def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None):
-    """把参数写回 params.ini（覆盖写；vars 逐行 key = value）。"""
+def _parse_bool(raw, key=""):
+    """解析 1/0、yes/no、true/false、on/off；空串返回 None（表示未配置）。"""
+    low = (raw or "").strip().lower()
+    if not low:
+        return None
+    if low in ("1", "yes", "y", "true", "on", "开", "启用"):
+        return True
+    if low in ("0", "no", "n", "false", "off", "关", "禁用"):
+        return False
+    print(f"[警告] params.ini [bounce] {key}={raw!r} 不是合法布尔值"
+          f"（用 1/0 或 yes/no），已忽略")
+    return None
+
+
+def _preserve_extra_sections(path, managed=("path", "site", "vars", "send", "bounce")):
+    """读取已有 params.ini 里非托管的小节，原样保留（避免保存时丢掉手写的配置）。"""
+    if not path or not os.path.isfile(path):
+        return []
+    cp = configparser.ConfigParser(comment_prefixes=(";", "#"),
+                                   inline_comment_prefixes=(";", "#"),
+                                   interpolation=None, strict=False)
+    cp.optionxform = str
+    try:
+        cp.read(path, encoding="utf-8")
+    except Exception:  # noqa: BLE001
+        return []
+    out = []
+    for sec in cp.sections():
+        if sec.strip().lower() in managed:
+            continue
+        out.append("")
+        out.append(f"[{sec}]")
+        for k, v in cp.items(sec):
+            out.append(f"{k} = {v}")
+    return out
+
+
+def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None,
+                bounce=None):
+    """把参数写回 params.ini（覆盖写；vars 逐行 key = value）。
+    bounce：当前生效的退信/巡检设置，回写进 [bounce] 小节以便下次直接沿用。"""
     lines = ["; broadcast.py 参数文件：每次运行自动读取；命令行参数优先级更高",
              "; 注意：本文件不控制 --send/--yes，正式发送仍需命令行显式指定", ""]
     lines.append("[path]")
@@ -1327,6 +1519,29 @@ def save_params(path, config, site_url, box_prefix, extra_vars, lang_labels=None
     lines.append("; to = 可选：默认测试收件邮箱（正式群发用命令行 --send，不加 --to）")
     lines.append("delay = 1.0")
     lines.append("group_pause = 3.0")
+    lines.append("")
+    lines.append("; 退信核查 / 发送中限流巡检（均可用同名命令行参数临时覆盖）")
+    lines.append("[bounce]")
+    b = bounce or {}
+    yesno = lambda v: "1" if v else "0"  # noqa: E731
+    lines.append("; 发送中巡检收件箱，出现「外发频率超过邮件系统限制」类退信立即中止发送")
+    lines.append(f"watch_bounces = {yesno(b.get('watch_bounces', True))}")
+    lines.append("; 每发 N 封巡检一次（调小更及时，但 IMAP 登录更频繁）")
+    lines.append(f"watch_every = {b.get('watch_every', 10)}")
+    lines.append("; 退信核查时，未分类的退信也按可重试一并剔除补发")
+    lines.append(f"prune_unknown = {yesno(b.get('prune_unknown', False))}")
+    lines.append("; 停用无效地址清单（既不跳过已知无效地址，也不写入新的硬退信）")
+    lines.append(f"no_invalid_list = {yesno(b.get('no_invalid_list', False))}")
+    lines.append("; 留空则用脚本同目录的默认文件名")
+    lines.append(f"invalid_file = {b.get('invalid_file', '') or ''}")
+    lines.append(f"bounce_report = {b.get('bounce_report', '') or ''}")
+    lines.append("; 留空则按 SMTP 域名推导（smtp.xxx → imap.xxx）")
+    lines.append(f"imap_host = {b.get('imap_host', '') or ''}")
+    lines.append(f"imap_port = {b.get('imap_port', 993)}")
+    lines.append("; since = auto 表示从断点清单最早一条记录的时间开始扫描")
+    lines.append(f"since = {b.get('since', 'auto')}")
+    lines.append(f"since_days = {b.get('since_days', 3)}")
+    lines.extend(_preserve_extra_sections(path))
     with open(path, "w", encoding="utf-8") as f:
         f.write("\n".join(lines) + "\n")
 
@@ -1366,6 +1581,9 @@ def run_interactive(cfg, templates, extra_vars, args, config_path):
     print(f"  box_prefix = {box_prefix}")
     for k in sorted(extra_vars):
         print(f"  {{{{{k}}}}} = {extra_vars[k]}")
+    print(f"  发送中巡检退信 = {'开启' if args.watch_bounces else '关闭'}"
+          f"（每 {args.watch_every} 封一次）"
+          f"   来源：params.ini [bounce] / 命令行，改值请编辑 params.ini")
 
     # 站点地址：只有完全没值时才问（值链已在 main 里兜底过 external_url）
     if not site_url:
@@ -1457,12 +1675,25 @@ def run_interactive(cfg, templates, extra_vars, args, config_path):
     except EOFError:
         save_ans = ""
     if save_ans in ("y", "yes"):
-        save_params(DEFAULT_PARAMS, config_path, site_url, box_prefix, extra_vars)
+        save_params(DEFAULT_PARAMS, config_path, site_url, box_prefix, extra_vars,
+                    bounce={
+                        "watch_bounces": args.watch_bounces,
+                        "watch_every": args.watch_every,
+                        "prune_unknown": args.prune_unknown,
+                        "no_invalid_list": args.no_invalid_list,
+                        "invalid_file": args.invalid_file,
+                        "bounce_report": args.bounce_report,
+                        "imap_host": args.imap_host,
+                        "imap_port": args.imap_port,
+                        "since": args.since,
+                        "since_days": args.since_days,
+                    })
         print(f"[交互模式] 已保存到 {DEFAULT_PARAMS}")
     return site_url, box_prefix, extra_vars
 
 
-def main():
+def build_arg_parser():
+    """构造命令行解析器（单独抽出来，便于测试与复用）。"""
     parser = argparse.ArgumentParser(
         description="TamaBox 站内信群发工具（读取 conf/app.ini，模板在 templates/ 目录，按用户语言发送）",
         formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -1502,6 +1733,10 @@ def main():
                         help="--pause-every 触发时的长暂停秒数，默认 15.0")
     parser.add_argument("--group-pause", type=float, default=3.0,
                         help="不同语言分组之间的额外暂停秒数，默认 3.0（0 关闭）")
+    parser.add_argument("--smtp-idle-reconnect", type=float, default=None, metavar="SEC",
+                        help="SMTP 连接空闲超过该秒数就主动重连再发（默认 30）。"
+                             "--delay 调大时（如 60s/封）服务器会掐掉空闲连接，"
+                             "不重连的话每封都会先吃一发 Server not connected")
     parser.add_argument("--db-driver", default="auto",
                         choices=["auto", "psql", "mysql", "psycopg2", "pymysql"],
                         help="DB 驱动：auto（默认，优先系统 psql/mysql 客户端）/ psycopg2 / pymysql")
@@ -1536,27 +1771,34 @@ def main():
                         help="配合 --check-bounces：把「可重试」退信（限流/临时性）从断点续发清单"
                              "剔除后结束（仅剔除，本次不发送）；再加 --send 则剔除后立即补发。"
                              "「地址不存在」等永久失败不会被剔除（不补发）")
-    parser.add_argument("--prune-unknown", action="store_true",
+    parser.add_argument("--prune-unknown", action="store_true", default=None,
                         help="配合 --check-bounces：无法分类的退信也按「可重试」处理，一并剔除补发"
-                             "（默认不剔除，避免给死信地址反复重发）")
-    parser.add_argument("--invalid-file", default=DEFAULT_INVALID, metavar="FILE",
+                             "（默认不剔除，避免给死信地址反复重发；也可在 params.ini [bounce] 设置）")
+    parser.add_argument("--invalid-file", default=None, metavar="FILE",
                         help=f"无效地址清单（硬退信地址）路径；默认脚本同目录 broadcast_invalid.json。"
                              f"清单里的地址每次运行都会跳过，即使 --reset-state")
     parser.add_argument("--reset-invalid", action="store_true",
                         help="清空无效地址清单（慎用：这些地址多半仍不可达）")
-    parser.add_argument("--no-invalid-list", action="store_true",
+    parser.add_argument("--no-invalid-list", action="store_true", default=None,
                         help="本次不使用无效地址清单：既不跳过已知无效地址，也不写入新的硬退信")
-    parser.add_argument("--bounce-report", default=DEFAULT_BOUNCE_REPORT, metavar="FILE",
+    parser.add_argument("--bounce-report", default=None, metavar="FILE",
                         help="退信核查明细报告路径（JSON）；默认脚本同目录 broadcast_bounce_report.json")
-    parser.add_argument("--imap-host", default="",
+    parser.add_argument("--watch-bounces", dest="watch_bounces", action="store_true", default=None,
+                        help="发送过程中巡检收件箱：每 --watch-every 封查一次，"
+                             "一旦出现「外发频率超过邮件系统限制」等限流退信立即中止发送（默认开启）")
+    parser.add_argument("--no-watch-bounces", dest="watch_bounces", action="store_false",
+                        help="关闭发送中巡检（--to 单封测试时本就关闭）")
+    parser.add_argument("--watch-every", type=int, default=None, metavar="N",
+                        help="发送中每发 N 封巡检一次退信（默认 10；调小会更频繁登录 IMAP）")
+    parser.add_argument("--imap-host", default=None,
                         help="IMAP 服务器地址（默认由 SMTP 域名推导：smtp.xxx → imap.xxx）")
-    parser.add_argument("--imap-port", type=int, default=993,
+    parser.add_argument("--imap-port", type=int, default=None,
                         help="IMAP 端口（默认 993，SSL）")
-    parser.add_argument("--since", default="auto", metavar="auto|YYYY-MM-DD",
+    parser.add_argument("--since", default=None, metavar="auto|YYYY-MM-DD",
                         help="退信扫描起点：auto（默认）= 从群发记录（断点清单）最早一条的"
                              "时间开始，只核查本次群发相关的退信；"
                              "也可显式指定日期如 2026-09-07")
-    parser.add_argument("--since-days", type=int, default=3,
+    parser.add_argument("--since-days", type=int, default=None,
                         help="断点清单为空时的回退扫描范围（天），默认 3 天")
     parser.add_argument("--subject", default=None,
                         help="全局覆盖邮件主题（忽略各语言 subject.txt；支持占位符）")
@@ -1564,8 +1806,12 @@ def main():
                         help="全局覆盖 HTML 正文模板文件（忽略按语言模板；调试用）")
     parser.add_argument("--plain", metavar="FILE",
                         help="全局覆盖纯文本正文模板文件（调试用）")
+    return parser
+
+
+def main():
     global args
-    args = parser.parse_args()
+    args = build_arg_parser().parse_args()
 
     # 默认交互模式：没给任何运行方式相关参数（--send/--to/--yes/--dry-run）就自动进入交互；
     # 显式 --no-interactive 强制关闭（脚本化/定时任务用）
@@ -1617,6 +1863,31 @@ 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"])
+
+    # 退信核查 / 发送中巡检：命令行没给（None）时才用 params.ini [bounce]
+    if args.watch_bounces is None:
+        args.watch_bounces = True if params.get("watch_bounces") is None else params["watch_bounces"]
+    if args.watch_every is None:
+        args.watch_every = params.get("watch_every") or 10
+    if args.prune_unknown is None:
+        args.prune_unknown = bool(params.get("prune_unknown"))
+    if args.no_invalid_list is None:
+        args.no_invalid_list = bool(params.get("no_invalid_list"))
+    if args.invalid_file is None:
+        args.invalid_file = params.get("invalid_file") or DEFAULT_INVALID
+    if args.bounce_report is None:
+        args.bounce_report = params.get("bounce_report") or DEFAULT_BOUNCE_REPORT
+    if args.imap_host is None:
+        args.imap_host = params.get("imap_host") or ""
+    if args.imap_port is None:
+        args.imap_port = params.get("imap_port") or 993
+    if args.since is None:
+        args.since = params.get("since") or "auto"
+    if args.since_days is None:
+        args.since_days = params.get("since_days") or 3
     extra_vars = {**params.get("vars", {}), **parse_var(args.var)}  # CLI --var 覆盖文件
 
     # ---- conf/app.ini 路径解析：CLI > params.ini > 环境变量 > ./conf/app.ini ----
@@ -1825,6 +2096,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 +2156,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:
         # 单文件模式：不按语言分组，所有人同一模板
@@ -1900,7 +2184,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}")
 
@@ -1932,10 +2216,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)
+                    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}")
@@ -1946,8 +2232,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,21 +2283,37 @@ 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()
-        except Exception:  # noqa: BLE001
-            pass
+        session.close()
 
     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:

params.ini

@@ -18,3 +18,35 @@ 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
+
+; 退信核查 / 发送中限流巡检开关
+; 这些都能在 params.ini 里长期配置，命令行同名参数可临时覆盖
+[bounce]
+; 发送中巡检收件箱：出现「外发频率超过邮件系统限制」类限流退信立即中止发送
+; 中止后会把限流地址从断点清单剔除，等限制恢复重跑即自动补发
+watch_bounces = 1
+
+; 每发 N 封巡检一次退信。调小更及时，但 IMAP 登录更频繁（不建议 < 5）
+watch_every = 10
+
+; 退信核查时，分类不明的退信是否也按「可重试」一并剔除补发（默认 0，保守不补发）
+prune_unknown = 0
+
+; 停用无效地址清单：既不跳过已知无效地址，也不写入新的硬退信
+no_invalid_list = 0
+
+; 留空则用脚本同目录的默认文件名
+invalid_file =
+bounce_report =
+
+; IMAP（退信核查要用发件邮箱的 IMAP；留空则按 SMTP 域名推导 smtp.xxx → imap.xxx）
+imap_host =
+imap_port = 993
+
+; since = auto 表示从断点清单最早一条发送记录的时间开始扫描，只查本次群发的退信
+since = auto
+since_days = 3