diff --git a/README.md b/README.md index 0327ea5..3243e5b 100644 --- a/README.md +++ b/README.md @@ -249,9 +249,17 @@ conf/app.ini 路径的解析优先级： - 用的是 conf/app.ini [mail] 的账号与密码（需邮箱已开启 IMAP，密码为邮箱登录密码） - 交互模式的运行方式「4) 退信核查」里同样三选一：仅核查 / 仅剔除 / 剔除并补发， 选择「仅剔除 / 剔除并补发」时会再问一次是否连未分类的一起补发 - - 已核查过的退信（按 Message-ID）记录在 broadcast_bounces_seen.json， - 重复核查不会重复报告/重复剔除 - - 每次核查另写一份明细报告 broadcast_bounce_report.json（含每个收件人的类别与判定依据） + - 已核查过的退信（按 Message-ID）与**上次扫描位置**（INBOX UID 游标 + + UIDVALIDITY）都记录在 broadcast_bounces_seen.json： + 默认从上次位置续扫（UID n+1:*），只拉新邮件的头部，重复核查不会 + 重复报告/重复剔除，也不会随邮箱邮件增多越扫越慢 + - 以下情况自动回退为按日期窗口从头扫（扫完位置照常更新）： + 首次核查（无位置记录）、邮箱 UIDVALIDITY 变化（换号/重建邮箱）、 + 显式指定 --since、加 --full-scan + - 发送中巡检不推进扫描位置（它只看本次运行之后新到的退信， + 范围内的历史邮件不标记「已扫」），位置只由 --check-bounces 推进 + - 每次核查另写一份明细报告 broadcast_bounce_report.json（含每个收件人的 + 类别与判定依据，以及 scan_mode / scan_cursor） 安全机制 -------- diff --git a/broadcast.py b/broadcast.py index 2524f07..a92a8c0 100644 --- a/broadcast.py +++ b/broadcast.py @@ -1249,18 +1249,59 @@ def _parse_mail_date(value): return dt +def _load_bounce_seen(): + """读取已核查记录文件（DEFAULT_BOUNCE_SEEN）。 + + 旧格式：Message-ID 列表；新格式：{"seen": [...], "cursor": {...}}。 + 返回 (seen 集合, cursor dict 或 None)。 + cursor = {"uidvalidity": int, "last_uid": int}，即上次扫描在 INBOX 里 + 扫到的位置；UIDVALIDITY 对不上（邮箱重建/换号）时游标作废回退全扫。 + """ + if not os.path.isfile(DEFAULT_BOUNCE_SEEN): + return set(), None + try: + with open(DEFAULT_BOUNCE_SEEN, "r", encoding="utf-8") as f: + data = json.load(f) + except Exception: # noqa: BLE001 + return set(), None + if isinstance(data, dict): + seen = set(data.get("seen") or []) + cur = data.get("cursor") + if (isinstance(cur, dict) + and isinstance(cur.get("uidvalidity"), int) + and isinstance(cur.get("last_uid"), int)): + return seen, cur + return seen, None + return set(data or []), None + + +def _save_bounce_seen(seen, cursor=None): + try: + with open(DEFAULT_BOUNCE_SEEN, "w", encoding="utf-8") as f: + json.dump({"seen": sorted(seen), "cursor": cursor}, f, ensure_ascii=False) + except Exception as e: # noqa: BLE001 + print(f"[警告] 记录已核查邮件/扫描位置失败（不影响本次结果）：{e}") + + def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None, - start_desc="", verbose=True): + start_desc="", verbose=True, resume=None): """连 IMAP 扫描退信并逐收件人分类（check_bounces 与发送中巡检共用）。 skip_ids 已处理过的 Message-ID（seen 文件 / 本次运行内已见过），跳过不重复报 not_before 只统计 Date 头晚于该时刻的退信（发送中巡检用：只看本次运行之后 新到的退信，避免拿历史退信误触发中止）；Date 解析不了则放行 - 返回 dict：bounced / new_ids / scanned / touched_ids + resume 上次扫描位置 {"uidvalidity": int, "last_uid": int}；UIDVALIDITY + 一致时从 last_uid 之后续扫（UID 检索），否则回退按日期窗口全扫 + （全新 / 邮箱 UID 代际变更 / 显式 --since / --full-scan）。 + 只由 check_bounces 传入推进——发送中巡检的 not_before 会过滤掉 + 范围内的历史邮件，若推进游标会把没处理过的邮件标记成「已扫」。 + 返回 dict：bounced / new_ids / scanned / touched_ids / last_uid / uidvalidity bounced {email: {"subject","date","category","reason","detail"}} new_ids 本次计入的退信 Message-ID（是否落盘由调用方决定） touched_ids 本次扫到的所有退信 Message-ID（含被 not_before 过滤掉的）， 供巡检做运行期去重，避免每轮重复拉取同一封全文 + last_uid 本次检查到的最大 UID（供调用方更新扫描位置） + uidvalidity 当前邮箱的 UIDVALIDITY（读不到为 None，调用方据此不落游标） """ smtp_host = cfg["mail_smtp"] or "" imap_host = (args.imap_host or "").strip() @@ -1280,33 +1321,58 @@ def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None, try: conn.login(account, password) conn.select("INBOX", readonly=True) - typ, data = conn.search(None, f'(SINCE "{since}")') + try: + uv = int(conn.response("UIDVALIDITY")[1][0]) + except Exception: # noqa: BLE001 + uv = None + resume_uid = None + if resume: + if uv is not None and uv == resume.get("uidvalidity"): + resume_uid = int(resume["last_uid"]) + if verbose: + print(f" [扫描位置] 续扫：从 UID {resume_uid} 之后开始" + f"（uidvalidity {uv}）") + elif verbose: + print(" [扫描位置] 邮箱 UIDVALIDITY 已变化（或无法读取），" + "本次回退按日期窗口全量扫描") + if resume_uid is not None: + typ, data = conn.uid("SEARCH", None, f"UID {resume_uid + 1}:*") + else: + typ, data = conn.uid("SEARCH", None, f'(SINCE "{since}")') if typ != "OK": raise RuntimeError("IMAP search 失败") ids = data[0].split() if verbose: - print(f" [IMAP] 扫描范围内共 {len(ids)} 封待扫描") + print(f" [IMAP] 扫描范围内共 {len(ids)} 封" + f"（{'续扫' if resume_uid is not None else '日期窗口'}）") skip_ids = skip_ids or set() bounced, new_ids, touched, scanned = {}, [], [], 0 + last_uid = resume_uid or 0 for i, mid in enumerate(ids, 1): - uid = mid.decode() if isinstance(mid, bytes) else str(mid) + uid = int(mid) + if uid <= last_uid: + # 防御：UID n:* 在 n 超过邮箱最大 UID 时会返回最后一封 + continue + last_uid = uid + ustr = str(uid) # 两段式：先取头部（省流量），命中退信特征再取全文 - typ, hdata = conn.fetch( - mid, "(BODY.PEEK[HEADER.FIELDS (SUBJECT MESSAGE-ID FROM CONTENT-TYPE)])") + typ, hdata = conn.uid( + "fetch", ustr, + "(BODY.PEEK[HEADER.FIELDS (SUBJECT MESSAGE-ID FROM CONTENT-TYPE)])") if typ != "OK" or not hdata or hdata[0] is None: continue try: head = email.message_from_bytes(hdata[0][1]) except Exception: # noqa: BLE001 continue - msg_id = (head.get("Message-ID") or "").strip() or f"uid:{uid}" + msg_id = (head.get("Message-ID") or "").strip() or f"uid:{ustr}" if msg_id in skip_ids: continue scanned += 1 if not is_bounce_message(head): continue - typ, fdata = conn.fetch(mid, "(RFC822)") + typ, fdata = conn.uid("fetch", ustr, "(RFC822)") if typ != "OK" or not fdata or fdata[0] is None: continue try: @@ -1333,7 +1399,8 @@ def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None, for r in rcpts) if rcpts else "(未解析出收件人)" print(f" [退信 {i}/{len(ids)}] {subject[:40]} → {desc}") return {"bounced": bounced, "new_ids": new_ids, - "scanned": scanned, "touched_ids": touched} + "scanned": scanned, "touched_ids": touched, + "last_uid": last_uid, "uidvalidity": uv} finally: try: conn.logout() @@ -1342,29 +1409,43 @@ def _scan_imap_bounces(cfg, args, start_dt, skip_ids=None, not_before=None, def check_bounces(cfg, args): - """登录发件邮箱 IMAP，扫描近 N 天的退信并解析失败收件人。 + """登录发件邮箱 IMAP，扫描退信并解析失败收件人。 返回 {email: {"subject","date","category","reason","detail"}}；已处理过的 - 邮件（Message-ID）记录在 broadcast_bounces_seen.json，重复核查不会重复报告。 + 邮件（Message-ID）与上次扫描位置（UID 游标）都记录在 + broadcast_bounces_seen.json：默认从上次位置续扫，只查新到的邮件， + 游标不存在/作废、显式 --since 或 --full-scan 时才按日期窗口从头扫。 category 见 CAT_RATE / CAT_HARD / CAT_UNKNOWN。""" start_dt, start_desc = resolve_scan_start(args) - seen = set() - if os.path.isfile(DEFAULT_BOUNCE_SEEN): - try: - with open(DEFAULT_BOUNCE_SEEN, "r", encoding="utf-8") as f: - seen = set(json.load(f)) - except Exception: # noqa: BLE001 - seen = set() + seen, cursor = _load_bounce_seen() + since_explicit = ((getattr(args, "since", "auto") or "auto").strip().lower() + not in ("", "auto")) + full_scan = bool(getattr(args, "full_scan", False)) + resume = None + if full_scan: + print("[扫描位置] --full-scan：忽略上次扫描位置，从头全量扫描") + elif since_explicit: + print("[扫描位置] 显式 --since 指定窗口，本次忽略上次扫描位置") + elif cursor is None: + print("[扫描位置] 全新扫描（尚无上次位置记录），按日期窗口全扫") + else: + resume = cursor + print(f"[扫描位置] 从上次扫到的 UID {cursor['last_uid']} 之后续扫" + f"（uidvalidity {cursor['uidvalidity']}）") try: - res = _scan_imap_bounces(cfg, args, start_dt, skip_ids=seen, start_desc=start_desc) + res = _scan_imap_bounces(cfg, args, start_dt, skip_ids=seen, + start_desc=start_desc, resume=resume) except Exception as e: # noqa: BLE001 sys.exit(f"[FATAL] IMAP 扫描失败（请确认邮箱已开启 IMAP，密码用邮箱登录密码）: {e}") bounced = res["bounced"] newly_seen, scanned = res["new_ids"], res["scanned"] bounce_cnt = len(newly_seen) + scan_mode = "resume" if resume is not None else "full" + scan_cursor = ({"uidvalidity": res["uidvalidity"], "last_uid": res["last_uid"]} + if res.get("uidvalidity") is not None and res.get("last_uid") else None) # ---- 按类别分组处理：只有「可重试」的才剔除补发 ---- groups = {CAT_RATE: [], CAT_HARD: [], CAT_UNKNOWN: []} @@ -1428,6 +1509,8 @@ def check_bounces(cfg, args): "script_version": SCRIPT_VERSION, "scan_from": start_dt.strftime("%Y-%m-%d %H:%M:%S"), "scan_from_desc": start_desc, + "scan_mode": scan_mode, + "scan_cursor": scan_cursor, "scanned": scanned, "bounce_messages": bounce_cnt, "counts": {c: len(groups.get(c, [])) for c in (CAT_RATE, CAT_HARD, CAT_UNKNOWN)}, @@ -1437,12 +1520,10 @@ def check_bounces(cfg, args): except Exception as e: # noqa: BLE001 print(f"[警告] 写入退信报告失败（不影响本次结果）：{e}") - if newly_seen: - try: - with open(DEFAULT_BOUNCE_SEEN, "w", encoding="utf-8") as f: - json.dump(sorted(set(newly_seen) | seen), f, ensure_ascii=False) - except Exception as e: # noqa: BLE001 - print(f"[警告] 记录已核查邮件失败（不影响本次结果）：{e}") + # 落盘：已核查 Message-ID + 新的扫描位置 + # （本次读不到 uidvalidity 时沿用旧游标：只可能往回多扫几封头部，不会漏） + if newly_seen or scan_cursor or cursor: + _save_bounce_seen(set(newly_seen) | seen, scan_cursor or cursor) return bounced @@ -1845,6 +1926,9 @@ def build_arg_parser(): parser.add_argument("--prune-unknown", action="store_true", default=None, help="配合 --check-bounces：无法分类的退信也按「可重试」处理，一并剔除补发" "（默认不剔除，避免给死信地址反复重发；也可在 params.ini [bounce] 设置）") + parser.add_argument("--full-scan", action="store_true", + help="退信核查忽略上次扫描位置，按日期窗口从头全量扫描" + "（默认记住上次扫到的位置只查新邮件；显式 --since 也会忽略位置）") parser.add_argument("--invalid-file", default=None, metavar="FILE", help=f"无效地址清单（硬退信地址）路径；默认脚本同目录 broadcast_invalid.json。" f"清单里的地址每次运行都会跳过，即使 --reset-state")