#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ TamaBox 站内信群发工具（独立项目，零依赖，不依赖、不修改 TamaBox 源码） 功能： - 复用生产 conf/app.ini 的 [database] 与 [mail] 配置 - 查询 users 表中「未注销」(deleted_at IS NULL) 且已填写邮箱的用户 - 按用户语言（users.language：zh-CN/zh-TW/en/ja）选择 templates/ 下的对应模板发送 - 模板全部是独立文件（templates/<语言>.html / .subject.txt / .plain.txt）， 想改文案直接编辑对应 HTML，无需碰任何 Python 代码 占位符（模板中直接写，发送时逐用户替换）： {{name}} 用户名（空则用「用户」） {{domain}} 用户个性域名（可能为空） {{box_link}} 用户的提问箱链接：{site-url}{box-prefix}{domain}；无个性域名时退回站点首页 {{email}} 收件邮箱 {{site}} 站点地址（--site-url 参数，未传则回退 app.ini external_url） {{site_title}} 站点名（app.ini [app] title） 任意自定义：--var old_domain=box.tama.guru --var expire_date=2026-12-24 （模板里写 {{old_domain}}、{{expire_date}} 即可引用） 注意：占位符若无对应值，发送时会替换为空串并在控制台列出缺失键，便于发现模板笔误。 安全设计（面向「全体用户」群发）： - 默认 dry-run：只列出收件人，绝不真正发信 - 发送前打印数据库统计（总/已注销/未注销/有邮箱），人工输入 yes 确认（--yes 跳过） - --to 测试地址：仅发一封；命中数据库用户则套用其姓名/域名/语言 - --limit N 小批量试水；--delay 间隔；--pause-every/--pause-for 防限流 - 按语言分群发送，组间 --group-pause 秒 - 断点续发：状态文件记录已发邮箱，重跑自动跳过；--reset-state 清空重发 - 监控报告：每次运行写 JSON 报告（--report / --no-report） - SMTP 拒收捕获：send_message 返回的 refused 计入失败 - 退信分类核查：只有「限流/临时性」退信才剔除补发；「地址不存在」这类 永久失败不补发，并记入无效地址清单（broadcast_invalid.json）永久跳过 - 发送中限流巡检：每 N 封查一次收件箱，一旦出现 「外发频率超过邮件系统限制」类退信（或 SMTP 当场报限流）立即中止发送， 并把这批限流地址从断点清单剔除，等限制恢复后重跑自动补发 参数优先级：命令行 > params.ini > 内置默认。退信/巡检相关开关（[bounce] 小节） 与站点、限速等一样都能在 params.ini 里长期配置，命令行只在需要临时覆盖时用。 用法示例： # 1) 演练：列出收件人 + 统计（不发信） python3 broadcast.py -c /path/to/conf/app.ini --dry-run \ --site-url https://box.shiroko.one \ --var old_domain=box.tama.guru --var expire_date=2026-12-24 # 2) 发一封到测试地址（先验证模板渲染 + SMTP 链路） python3 broadcast.py -c /path/to/conf/app.ini --send --to you@example.com \ --site-url https://box.shiroko.one \ --var old_domain=box.tama.guru --var expire_date=2026-12-24 # 3) 正式全员群发（建议先 --limit 50 试水） python3 broadcast.py -c /path/to/conf/app.ini --send --yes \ --site-url https://box.shiroko.one \ --var old_domain=box.tama.guru --var expire_date=2026-12-24 依赖：仅 Python 3 标准库。数据库默认调用系统 psql/mysql 客户端（零 pip 依赖）， 客户端不存在时可回退 psycopg2/pymysql（需自行 pip install）。 """ import argparse import configparser import email.utils import hashlib import imaplib import json import os import re import shutil import subprocess import tempfile from datetime import datetime, timedelta from html import escape # noqa: F401 (模板由用户维护，脚本不做 HTML 转义，仅保留占位符替换) from html.parser import HTMLParser import smtplib import ssl import sys 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, parsedate_to_datetime from urllib.parse import quote_plus 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") DEFAULT_TEMPLATES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "templates") DEFAULT_PARAMS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "params.ini") DEFAULT_BOUNCE_SEEN = os.path.join(os.path.dirname(os.path.abspath(__file__)), "broadcast_bounces_seen.json") DEFAULT_INVALID = os.path.join(os.path.dirname(os.path.abspath(__file__)), "broadcast_invalid.json") DEFAULT_BOUNCE_REPORT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "broadcast_bounce_report.json") # 语言展示标签（未知语言直接显示语言码） LANG_LABEL = { "zh-CN": "简体中文", "zh-TW": "繁體中文", "en": "English", "ja": "日本語", } # 常见语言在分群里靠前的固定顺序（其余语言排在后面） LANG_ORDER = ["zh-CN", "zh-TW", "en", "ja"] # --------------------------------------------------------------------------- # 配置读取：兼容 ini 行内注释，并去包围引号 # --------------------------------------------------------------------------- def load_config(path): if not os.path.isfile(path): sys.exit(f"[FATAL] 配置文件不存在: {path}") cp = configparser.ConfigParser( comment_prefixes=(";", "#"), inline_comment_prefixes=(";",), interpolation=None, strict=False, # 允许重复键（取后者） ) cp.optionxform = str # 保留键大小写 try: cp.read(path, encoding="utf-8") except Exception as e: # noqa: BLE001 sys.exit(f"[FATAL] 解析配置文件失败: {e}") def get(section, key, default=""): try: v = cp.get(section, key) except (configparser.NoSectionError, configparser.NoOptionError): return default if v is None: return default v = v.strip() if len(v) >= 2 and v[0] == '"' and v[-1] == '"': v = v[1:-1] return v cfg = { "app_external_url": get("app", "external_url").rstrip("/"), "app_title": get("app", "title") or "TamaBox", "app_default_lang": (get("app", "default_lang") or "zh-CN").strip(), "db_type": get("database", "type") or "postgres", "db_user": get("database", "user"), "db_password": get("database", "password"), "db_host": get("database", "host") or "127.0.0.1", "db_port": int(get("database", "port") or 5432), "db_name": get("database", "name"), "db_schema": get("database", "schema"), "db_sslmode": get("database", "sslmode") or "disable", "mail_account": get("mail", "account"), "mail_password": get("mail", "password"), "mail_port": int(get("mail", "port") or 465), "mail_smtp": get("mail", "smtp"), "mail_skip_tls_verify": get("mail", "skip_tls_verify").lower() in ("1", "true", "yes"), } return cfg # --------------------------------------------------------------------------- # 数据库访问（零依赖优先：系统 psql/mysql 客户端；回退 psycopg2/pymysql） # --------------------------------------------------------------------------- def _build_pg_uri(cfg): user = quote_plus(cfg["db_user"]) pw = quote_plus(cfg["db_password"]) sslmode = cfg.get("db_sslmode") or "disable" return ( f"postgresql://{user}:{pw}@{cfg['db_host']}:{cfg['db_port']}" f"/{quote_plus(cfg['db_name'])}?sslmode={sslmode}" ) def _choose_driver(cfg, args): want = (getattr(args, "db_driver", "auto") or "auto").lower() db_type = cfg["db_type"] def cli_or_lib(cli_name, lib_import): if shutil.which(cli_name): return cli_name try: __import__(lib_import) return "lib" except ImportError: return cli_name # 后续执行时会给出清晰报错 if want == "psql": return "psql" if want == "mysql": return "mysql" if want in ("psycopg2", "pymysql", "lib"): return "lib" if db_type == "postgres": return cli_or_lib("psql", "psycopg2") return cli_or_lib("mysql", "pymysql") def _psql_query(cfg, sql): uri = _build_pg_uri(cfg) sep = "\x1f" # 单元分隔符，数据里几乎不可能出现 cmd = ["psql", uri, "-X", "-t", "-A", "-F", sep, "-P", "null=\\N", "-P", "footer=off", "-c", sql] try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=120) except FileNotFoundError: sys.exit("[FATAL] 未找到 psql 客户端：请安装 PostgreSQL 客户端（如 apt install postgresql-client），" "或用 --db-driver psycopg2 并 pip install psycopg2-binary") if r.returncode != 0: sys.exit(f"[FATAL] psql 执行失败: {r.stderr.strip()[:500]}") rows = [] for line in r.stdout.splitlines(): if not line: continue parts = [None if p == "\\N" else p for p in line.split(sep)] rows.append(parts) return rows def _mysql_query(cfg, sql): fd, cnf = tempfile.mkstemp(suffix=".cnf", prefix="tb_my_") try: with os.fdopen(fd, "w") as f: f.write("[client]\npassword=%s\n" % cfg["db_password"]) os.chmod(cnf, 0o600) cmd = ["mysql", f"-h{cfg['db_host']}", f"-P{cfg['db_port']}", f"-u{cfg['db_user']}", f"--defaults-extra-file={cnf}", "-D", cfg["db_name"], "-N", "-B", "--default-character-set=utf8mb4", "-e", sql] try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=120) except FileNotFoundError: sys.exit("[FATAL] 未找到 mysql 客户端：请安装 MySQL 客户端（如 apt install mysql-client），" "或用 --db-driver pymysql 并 pip install pymysql") if r.returncode != 0: sys.exit(f"[FATAL] mysql 执行失败: {r.stderr.strip()[:500]}") finally: try: os.remove(cnf) except OSError: pass rows = [] for line in r.stdout.splitlines(): if not line: continue parts = [None if p == "\\N" else p for p in line.split("\t")] rows.append(parts) return rows def _lib_query(cfg, sql): db_type = cfg["db_type"] if db_type == "postgres": try: import psycopg2 except ImportError: sys.exit("[FATAL] 需要 psycopg2 驱动：pip install psycopg2-binary") kwargs = dict(host=cfg["db_host"], port=cfg["db_port"], user=cfg["db_user"], password=cfg["db_password"], dbname=cfg["db_name"], sslmode=cfg.get("db_sslmode") or "disable", connect_timeout=10) if cfg["db_schema"]: kwargs["options"] = f"-c search_path={cfg['db_schema']}" try: conn = psycopg2.connect(**kwargs) except Exception as e: # noqa: BLE001 sys.exit(f"[FATAL] 连接 PostgreSQL 失败: {e}") else: try: import pymysql except ImportError: sys.exit("[FATAL] 需要 pymysql 驱动：pip install pymysql") try: conn = pymysql.connect(host=cfg["db_host"], port=cfg["db_port"], user=cfg["db_user"], password=cfg["db_password"], database=cfg["db_name"], connect_timeout=10, charset="utf8mb4") except Exception as e: # noqa: BLE001 sys.exit(f"[FATAL] 连接 MySQL 失败: {e}") try: with conn.cursor() as cur: cur.execute(sql) rows = cur.fetchall() finally: conn.close() return [list(r) for r in rows] def run_query(cfg, sql, args): drv = _choose_driver(cfg, args) if drv == "psql": return _psql_query(cfg, sql) if drv == "mysql": return _mysql_query(cfg, sql) return _lib_query(cfg, sql) # --------------------------------------------------------------------------- # users 表定位：同库不同 schema 可能有同名 users 表，不限定 schema 时 # psql 会按 search_path 命中错误的空表（表现为统计全 0、语言列探测失败）。 # 枚举所有 schema 的 users 表并计数，选有数据的那个（其次带 language 列）。 # --------------------------------------------------------------------------- def _qid(s): return '"' + s.replace('"', '""') + '"' def resolve_users_table(cfg, args): """返回 (限定表名, schema 名, 信息或 None)。非 postgres / 探测失败时退回裸 users。""" if cfg["db_type"] != "postgres": return "users", "", None sql = ( "SELECT table_schema FROM information_schema.tables " "WHERE table_name = 'users' " "AND table_schema NOT LIKE 'pg\\_%' ESCAPE '\\' " "AND table_schema <> 'information_schema' ORDER BY table_schema" ) try: rows = run_query(cfg, sql, args) except SystemExit: raise except Exception: # noqa: BLE001 return "users", "", None schemas = [str(r[0]) for r in rows if r and r[0]] if not schemas: return "users", "", None stats = [] for s in schemas: n = 0 try: cnt_rows = run_query(cfg, f"SELECT COUNT(*) FROM {_qid(s)}.users", args) if cnt_rows and cnt_rows[0] and cnt_rows[0][0]: n = int(cnt_rows[0][0]) except Exception: # noqa: BLE001 n = 0 lang_col = None try: lrows = run_query( cfg, "SELECT column_name FROM information_schema.columns " "WHERE table_name = 'users' AND table_schema = " "'" + s.replace("'", "''") + "' " "AND column_name IN ('language', 'lang') ORDER BY column_name DESC", args, ) if lrows and lrows[0] and lrows[0][0]: lang_col = str(lrows[0][0]) except Exception: # noqa: BLE001 pass stats.append((s, n, lang_col)) stats.sort(key=lambda st: (1 if st[1] > 0 else 0, 1 if st[2] else 0, st[1]), reverse=True) s, n, lang = stats[0] return f"{_qid(s)}.users", s, {"rows": n, "lang_col": lang, "all": stats} def detect_lang_column(cfg, args, tbl="users"): """返回 (sql 表达式, 列名或 None)。 显式 --lang-column 时直接采用；否则真实试跑 language → lang → NULL， NULL 兜底必然成功（全员回退默认语言），不会因列缺失 FATAL。 """ specified = getattr(args, "lang_column", "auto") if specified not in (None, "", "auto"): return specified, specified last_exc = None for cand in ("language", "lang", None): expr = cand if cand else "NULL" probe = ( f"SELECT name, domain, email, {expr} FROM {tbl} " "WHERE deleted_at IS NULL AND email IS NOT NULL AND email <> '' " "ORDER BY id LIMIT 1" ) try: run_query(cfg, probe, args) return expr, cand except (SystemExit, Exception) as e: # noqa: BLE001 last_exc = e continue sys.exit(f"[FATAL] 语言列探测全失败（含 NULL 兜底），疑似 users 表不存在或权限不足: {last_exc}") def fetch_recipients(cfg, args, lang_expr=None, tbl="users"): if lang_expr is None: lang_expr, _ = detect_lang_column(cfg, args, tbl) sql = ( f"SELECT name, domain, email, {lang_expr} FROM {tbl} " "WHERE deleted_at IS NULL AND email IS NOT NULL AND email <> '' " "ORDER BY id" ) try: rows = run_query(cfg, sql, args) except SystemExit: raise except Exception as e: # noqa: BLE001 sys.exit(f"[FATAL] 查询 users 失败: {e}") recipients = [] for row in rows: name, domain, email_addr, lang = (list(row) + [None, None, None, None])[:4] recipients.append( { "name": (name or "").strip(), "domain": (domain or "").strip(), "email": (email_addr or "").strip(), "language": (lang or "").strip(), } ) return recipients def fetch_counts(cfg, args, tbl="users"): if cfg["db_type"] == "postgres": sql = ( "SELECT COUNT(*), " "COUNT(*) FILTER (WHERE deleted_at IS NOT NULL), " "COUNT(*) FILTER (WHERE deleted_at IS NULL), " "COUNT(*) FILTER (WHERE deleted_at IS NULL AND email <> ''), " "COUNT(*) FILTER (WHERE deleted_at IS NULL AND (email IS NULL OR email = '')) " f"FROM {tbl}" ) else: sql = ( "SELECT COUNT(*), " "SUM(CASE WHEN deleted_at IS NOT NULL THEN 1 ELSE 0 END), " "SUM(CASE WHEN deleted_at IS NULL THEN 1 ELSE 0 END), " "SUM(CASE WHEN deleted_at IS NULL AND email <> '' THEN 1 ELSE 0 END), " "SUM(CASE WHEN deleted_at IS NULL AND (email IS NULL OR email = '') THEN 1 ELSE 0 END) " f"FROM {tbl}" ) try: rows = run_query(cfg, sql, args) except SystemExit: raise except Exception as e: # noqa: BLE001 sys.exit(f"[FATAL] 统计 users 失败: {e}") if not rows: sys.exit("[FATAL] 统计 users 返回空结果") vals = [int(v) if v is not None else 0 for v in rows[0]] total, deactivated, active, active_with_email, active_no_email = (vals + [0] * 5)[:5] return { "total": total, "deactivated": deactivated, "active": active, "active_with_email": active_with_email, "active_no_email": active_no_email, } # --------------------------------------------------------------------------- # 模板加载：templates/.html（必需）、.subject.txt（必需）、 # .plain.txt（可选，缺省由 HTML 自动抽取纯文本） # --------------------------------------------------------------------------- class _PlainExtractor(HTMLParser): """把 HTML 正文抽成可读纯文本（跳过 style/script，块级标签转换行）。""" _BLOCK = ("p", "div", "tr", "li", "h1", "h2", "h3", "h4", "table", "section") def __init__(self): super().__init__(convert_charrefs=True) self.parts = [] self._skip = 0 def handle_starttag(self, tag, attrs): if tag in ("style", "script"): self._skip += 1 elif tag == "br": self.parts.append("\n") elif tag in self._BLOCK: self.parts.append("\n") def handle_endtag(self, tag): if tag in ("style", "script"): if self._skip: self._skip -= 1 elif tag in self._BLOCK: self.parts.append("\n") def handle_data(self, data): if not self._skip: self.parts.append(data) def html_to_plain(html_text): p = _PlainExtractor() try: p.feed(html_text) p.close() except Exception: # noqa: BLE001 return html_text lines = [ln.strip() for ln in "".join(p.parts).splitlines()] out, blank = [], 0 for ln in lines: if ln: out.append(ln) blank = 0 else: blank += 1 if blank == 1: out.append("") return "\n".join(out).strip() + "\n" # 单文件通用模板的保留文件名（不作为语言参与多语言扫描） SINGLE_TPL_NAME = "single" RESERVED_TEMPLATES = {SINGLE_TPL_NAME} def load_templates(templates_dir): """扫描模板目录，返回 {lang: {subject, html, plain}}；同时返回发现顺序。 reserved（single.*）不作为语言，单独用 load_single_template 加载。""" if not os.path.isdir(templates_dir): sys.exit(f"[FATAL] 模板目录不存在: {templates_dir}") langs = [] for fn in sorted(os.listdir(templates_dir)): if fn.endswith(".html"): lang = fn[: -len(".html")] if lang and lang not in RESERVED_TEMPLATES and lang not in langs: langs.append(lang) if not langs: sys.exit(f"[FATAL] 模板目录里没有找到任何 <语言>.html 文件: {templates_dir}") templates = {} for lang in langs: html_path = os.path.join(templates_dir, lang + ".html") subj_path = os.path.join(templates_dir, lang + ".subject.txt") plain_path = os.path.join(templates_dir, lang + ".plain.txt") with open(html_path, "r", encoding="utf-8") as f: html = f.read() if os.path.isfile(subj_path): with open(subj_path, "r", encoding="utf-8") as f: subject = f.read().strip() else: print(f"[警告] 缺少 {lang}.subject.txt，回退使用通用主题") subject = "【{{site_title}}】站内通知" if os.path.isfile(plain_path): with open(plain_path, "r", encoding="utf-8") as f: plain = f.read() else: plain = html_to_plain(html) templates[lang] = {"subject": subject, "html": html, "plain": plain} return templates, langs def load_single_template(templates_dir): """加载单文件通用模板 templates/single.html（+ single.subject.txt）。 不存在返回 None；plain 缺省由 HTML 抽取。""" html_path = os.path.join(templates_dir, SINGLE_TPL_NAME + ".html") if not os.path.isfile(html_path): return None with open(html_path, "r", encoding="utf-8") as f: html = f.read() subj_path = os.path.join(templates_dir, SINGLE_TPL_NAME + ".subject.txt") if os.path.isfile(subj_path): with open(subj_path, "r", encoding="utf-8") as f: subject = f.read().strip() else: print(f"[提示] 缺少 {SINGLE_TPL_NAME}.subject.txt，主题回退为「【{{{{site_title}}}}】站内通知」") subject = "【{{site_title}}】站内通知" plain_path = os.path.join(templates_dir, SINGLE_TPL_NAME + ".plain.txt") if os.path.isfile(plain_path): with open(plain_path, "r", encoding="utf-8") as f: plain = f.read() else: plain = html_to_plain(html) return {"subject": subject, "html": html, "plain": plain} def render(template, ctx): """{{key}} 占位符替换；返回 (结果, 未提供的键列表)。""" missing = [] def repl(m): key = m.group(1).strip() if key in ctx: return ctx[key] missing.append(key) return "" # 缺失占位符替换为空串，避免邮件里出现 {{xxx}} out = _PLACEHOLDER_RE.sub(repl, template) return out, missing import re # noqa: E402 _PLACEHOLDER_RE = re.compile(r"\{\{\s*([A-Za-z0-9_\-]+)\s*\}\}") def render_strict(template, ctx): out, missing = render(template, ctx) return out, sorted(set(missing)) def resolve_lang(lang_field, default_lang, available): """解析收件人语言：空 -> zh-CN；未知码 -> default_lang -> zh-CN -> 任一可用。""" lang = (lang_field or "").strip() if not lang: lang = "zh-CN" if lang not in available: lang = (default_lang or "").strip() if lang not in available: lang = "zh-CN" if lang not in available: lang = available[0] return lang 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_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 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): port = cfg["mail_port"] context = ssl.create_default_context() if cfg["mail_skip_tls_verify"]: context.check_hostname = False context.verify_mode = ssl.CERT_NONE if port == 465: server = smtplib.SMTP_SSL(cfg["mail_smtp"], port, timeout=30, context=context) else: server = smtplib.SMTP(cfg["mail_smtp"], port, timeout=30) server.ehlo() server.starttls(context=context) server.ehlo() server.login(cfg["mail_account"], cfg["mail_password"]) 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_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() 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() 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() 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 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 追加写，崩溃安全） # --------------------------------------------------------------------------- def load_sent_state(state_path): sent = {} if not os.path.isfile(state_path): return sent try: with open(state_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: obj = json.loads(line) except Exception: # noqa: BLE001 continue em = (obj.get("email") or "").strip().lower() if em: sent[em] = obj except Exception as e: # noqa: BLE001 print(f"[警告] 读取断点续发清单失败，将视为空清单继续：{e}") return sent def record_sent(state_path, entry): try: with open(state_path, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") except Exception as e: # noqa: BLE001 print(f"[警告] 写入断点续发清单失败（不影响本次发送）：{e}") def reset_state(state_path): try: if os.path.isfile(state_path): os.remove(state_path) except Exception as e: # noqa: BLE001 print(f"[警告] 清除断点续发清单失败：{e}") # --------------------------------------------------------------------------- # 退信核查（IMAP）：SMTP 250 ≠ 真正送达。企业邮箱（如阿里云）对限流、不存在的 # 收件人常「先收下、后异步退信」，退信通知会投到发件账号的收件箱。这里用标准库 # imaplib 登录 IMAP 扫描退信，解析出真正投递失败的收件人。 # # 关键：退信不是一类，必须分类处理—— # rate（限流/临时性，如「您的账号外发频率超过邮件系统限制」） # → 稍后重发大概率成功：从断点续发清单剔除，下次重跑自动补发 # hard（永久失败，如「收件人地址不存在」/ DSN 5.x.x） # → 重发也发不出去，只会浪费额度、拖垮发信信誉： # 不从断点清单剔除（即不补发），并记入无效地址清单永久跳过 # unknown（判不出来）→ 默认不剔除（保守），可用 --prune-unknown 强制剔除 # --------------------------------------------------------------------------- BOUNCE_SUBJECT_KEYWORDS = ( "退信", "投递失败", "无法投递", "未能送达", "无法送达", "邮件被退回", "被退回", "退回通知", "投递状态", "发送失败", "发送不成功", "undeliver", "delivery status", "delivery failed", "delivery failure", "returned mail", "failure notice", "not delivered", "mail delivery subsystem", "mail could not be delivered", ) EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}") # 中文退信常见「标签：地址」写法 + RFC DSN 的 Final-Recipient 等 # （(?[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,})", r"(?:Final-Recipient|Original-Recipient|X-Failed-Recipients)" r"[^\r\n@;]*[;:]\s*(?:rfc822\s*;?\s*)?" r"(?[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,})", ) SYSTEM_LOCALPARTS = ("mailer-daemon", "postmaster", "mail-daemon", "noreply", "no-reply") # --- 退信分类关键词 --------------------------------------------------------- # 判定优先级：DSN 状态码（5.x.x/4.x.x，最权威） > 硬退信关键词 > 限流关键词 > 未知 CAT_RATE = "rate" # 限流/临时性失败 → 可重试，剔除后补发 CAT_HARD = "hard" # 地址不存在等永久失败 → 不补发，记入无效地址清单 CAT_UNKNOWN = "unknown" # 判不出来 → 默认不剔除（保守），--prune-unknown 可强制 CAT_LABEL = { CAT_RATE: "可重试 · 限流/临时性", CAT_HARD: "永久失败 · 地址无效", CAT_UNKNOWN: "未分类（保守：不补发）", } # 限流 / 临时性失败（可稍后重试） RATE_LIMIT_KEYWORDS = ( "您的账号外发频率超过邮件系统限制", # 阿里云企业邮箱限流退信原文 "外发频率超过", "发送频率超过", "超过邮件系统限制", "发送频率受限", "发送流量超限", "超出发送限额", "发送数量超限", "超出邮件系统限制", "系统繁忙", "稍后重试", "稍后重新发送", "请稍后再试", "请稍后重新发送", "rate limit", "too many", "throttl", "try again later", "try later", "too many recipients", "temporarily", "temporary failure", "temporary local", "greylist", "grey list", "deferred", "deferral", "connection timed out", "try again", "please retry", "later time", ) # 永久失败：收件人/邮箱根本不存在 HARD_BOUNCE_KEYWORDS = ( "地址不存在", "用户不存在", "无此用户", "账号不存在", "帐户不存在", "收件人不存在", "不存在该用户", "邮箱不存在", "没有这个邮箱", "没有此邮箱", "查无此用户", "无效地址", "无效的收件人", "无效的收件地址", "收件人地址错误", "该邮箱不存在", "未知的用户", "no such user", "no such recipient", "no such mailbox", "no such address", "user unknown", "unknown user", "unknown recipient", "unknown address", "recipient address rejected", "recipient not found", "recipient rejected", "mailbox unavailable", "mailbox not found", "mailbox does not exist", "address does not exist", "address not found", "address rejected", "invalid recipient", "invalid address", "invalid mailbox", "bad destination", "bad address syntax", "user not found", "does not exist", "550 5.1.1", "no mailbox here", "mailbox disabled", ) # SMTP 状态码识别：增强状态码 5.1.1 / 三位码 550 _SMTP_ENHANCED_RE = re.compile(r"\b([245])\.(\d{1,3})\.(\d{1,3})\b") _SMTP_CODE_RE = re.compile(r"(? limit: break else: chunks.append(_part_payload_text(msg)) return "\n".join(c for c in chunks if c)[:limit] # DSN 里表示「已送达」的 Action，遇到就跳过（其余 failed/delayed/expired 都算失败） DSN_OK_ACTIONS = ("delivered", "relayed", "expanded") def _dsn_recipient_details(msg): """标准 DSN（message/delivery-status）中每个收件人的失败详情。 返回 [{email, action, status, diagnostic}]，逐个带自己的 Status / Diagnostic-Code，因此一封退信里的多个收件人可以分别归类。 """ out = [] for part in msg.walk(): if part.get_content_type() != "message/delivery-status": continue blocks = part.get_payload() if not isinstance(blocks, list): continue for blk in blocks: try: final = str(blk.get("Final-Recipient") or blk.get("Original-Recipient") or "") action = str(blk.get("Action") or "failed").strip().lower() status = str(blk.get("Status") or "").strip() diagnostic = str(blk.get("Diagnostic-Code") or "").strip() except Exception: # noqa: BLE001 continue if action in DSN_OK_ACTIONS: continue m = EMAIL_RE.search(final) if not m: continue out.append({ "email": m.group(0).strip(".").lower(), "action": action, "status": status, "diagnostic": diagnostic, }) return out def _smtp_code_class(text): """从 DSN Status / Diagnostic-Code 取 SMTP 状态码首位：5=永久、4=临时、2=成功。""" if not text: return None m = _SMTP_ENHANCED_RE.search(text) # 增强状态码，如 5.1.1 / 4.4.7 if m: return m.group(1) m = _SMTP_CODE_RE.search(text) # 三位码，如 550 / 451 if m: return m.group(0)[0] return None def classify_bounce(status="", diagnostic="", text=""): """判定退信类别，返回 (类别, 判定依据)。 顺序：DSN 状态码（最权威）→ 正文/诊断信息里的硬退信关键词 → 限流关键词 → 正文里的 SMTP 状态码 → unknown。 硬退信优先于限流：宁可不补发一个可疑地址，也不要反复给死信浪费额度。 """ for src in (status, diagnostic): c = _smtp_code_class(src or "") 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() for kw in HARD_BOUNCE_KEYWORDS: if kw.lower() in low: return CAT_HARD, f"关键词「{kw}」" for kw in RATE_LIMIT_KEYWORDS: if kw.lower() in low: return CAT_RATE, f"关键词「{kw}」" c = _smtp_code_class(text or "") if c == "5": return CAT_HARD, "正文 SMTP 5xx" if c == "4": return CAT_RATE, "正文 SMTP 4xx" return CAT_UNKNOWN, "" def _own_ok(addr, own_addr): a = addr.strip(".").lower() if not a or a == own_addr.lower(): return False local = a.split("@", 1)[0].lower() return local not in SYSTEM_LOCALPARTS def is_bounce_message(msg): """判断一封邮件是否退信通知：主题关键词 或 multipart/report 送达报告。""" subject = _decode_hdr(msg.get("Subject", "")).lower() if any(k in subject for k in BOUNCE_SUBJECT_KEYWORDS): return True if msg.get_content_type() == "multipart/report": report_type = str(msg.get_param("Report-Type", "")).lower() if "delivery" in report_type or "status" in report_type: return True return False def extract_bounced_recipients(msg, own_addr): """从退信中解析投递失败的收件人，并逐个分类。 返回 [{email, category, reason, detail}]（已去重、剔除发件人与系统地址）。 """ if not is_bounce_message(msg): return [] found = [] def add(em, category, reason, detail=""): em = (em or "").strip().strip(".").lower() if not _own_ok(em, own_addr): return if any(x["email"] == em for x in found): return found.append({"email": em, "category": category, "reason": reason, "detail": (detail or "")[:160]}) # 1) 标准 DSN 最可靠：每个收件人带自己的 Status/Diagnostic-Code，可逐个分类 for d in _dsn_recipient_details(msg): cat, reason = classify_bounce(d["status"], d["diagnostic"], "") add(d["email"], cat, reason, d["status"] or d["diagnostic"]) # 2) 中文退信的「标签：地址」写法 / 3) 正文兜底：整封共用一个分类结果 if not found: text = _part_text(msg) cat, reason = classify_bounce("", "", text) for pat in LABELED_PATTERNS: for m in re.finditer(pat, text, re.IGNORECASE): add(m.group("e"), cat, reason, "") if not found: for e in EMAIL_RE.findall(text): add(e, cat, reason, "") return found def prune_state(state_path, emails): """从断点续发清单剔除指定邮箱（实际未送达，重跑会补发）。返回剔除条数。""" if not os.path.isfile(state_path): return 0 targets = {e.strip().lower() for e in emails} kept, removed = [], 0 with open(state_path, "r", encoding="utf-8") as f: for line in f: s = line.strip() if not s: continue try: obj = json.loads(s) em = (obj.get("email") or "").strip().lower() except Exception: # noqa: BLE001 kept.append(s) continue if em in targets: removed += 1 else: kept.append(s) if removed: tmp = state_path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: f.write("\n".join(kept) + ("\n" if kept else "")) os.replace(tmp, state_path) return removed # --------------------------------------------------------------------------- # 无效地址清单：硬退信（收件人不存在 / DSN 5.x.x）的地址记在这里。 # 这些地址重发也发不出去，所以： # - 不从断点续发清单剔除（即不会「补发」） # - 记入本清单，后续每次运行（即使 --reset-state）都会直接跳过 # 只有确实要恢复发送时，才手动删条目或 --reset-invalid 清空。 # --------------------------------------------------------------------------- def load_invalid(path): """读取无效地址清单，返回 {email: {"reason","detail","subject","date","ts"}}。""" if not os.path.isfile(path): return {} try: with open(path, "r", encoding="utf-8") as f: raw = json.load(f) except Exception as e: # noqa: BLE001 print(f"[警告] 读取无效地址清单失败，按空清单处理：{e}") return {} out = {} if isinstance(raw, dict): items = raw.items() elif isinstance(raw, list): items = [(it.get("email"), it) for it in raw if isinstance(it, dict)] else: return out for em, info in items: em = str(em or "").strip().lower() if not em: continue info = info if isinstance(info, dict) else {"reason": str(info)} out[em] = { "reason": str(info.get("reason") or ""), "detail": str(info.get("detail") or ""), "subject": str(info.get("subject") or ""), "date": str(info.get("date") or ""), "ts": str(info.get("ts") or ""), } return out def record_invalid(path, entries): """合并写入无效地址清单。entries: {email: {reason, detail, subject, date}}。 返回新增条数（已有条目只更新原因）。""" if not entries: return 0 data = load_invalid(path) now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") added = 0 for em, info in entries.items(): em = (em or "").strip().lower() if not em: continue if em not in data: added += 1 data[em] = { "reason": str(info.get("reason") or ""), "detail": str(info.get("detail") or ""), "subject": str(info.get("subject") or ""), "date": str(info.get("date") or ""), "ts": now, } try: tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump({k: data[k] for k in sorted(data)}, f, ensure_ascii=False, indent=2) os.replace(tmp, path) except Exception as e: # noqa: BLE001 print(f"[警告] 写入无效地址清单失败：{e}") return 0 return added def reset_invalid(path): """清空无效地址清单（慎用：这些地址多半仍是不可达的）。""" try: if os.path.isfile(path): os.remove(path) return True except Exception as e: # noqa: BLE001 print(f"[警告] 清除无效地址清单失败：{e}") 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。 返回 (datetime, 说明文字)。""" since_arg = (getattr(args, "since", "auto") or "auto").strip() if since_arg and since_arg.lower() != "auto": try: dt = datetime.strptime(since_arg, "%Y-%m-%d") return dt, f"--since 指定日期 {since_arg}" except ValueError: print(f"[警告] --since 格式应为 YYYY-MM-DD，已忽略：{since_arg}（改用 auto）") sent = load_sent_state(args.state) ts_list = [obj.get("ts") for obj in sent.values() if isinstance(obj, dict) and obj.get("ts")] if ts_list: start = datetime.fromtimestamp(min(ts_list)) return start, f"群发首条记录时间 {start.strftime('%Y-%m-%d %H:%M')}" return (datetime.now() - timedelta(days=args.since_days), 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 _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, resume=None): """连 IMAP 扫描退信并逐收件人分类（check_bounces 与发送中巡检共用）。 skip_ids 已处理过的 Message-ID（seen 文件 / 本次运行内已见过），跳过不重复报 not_before 只统计 Date 头晚于该时刻的退信（发送中巡检用：只看本次运行之后 新到的退信，避免拿历史退信误触发中止）；Date 解析不了则放行 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() 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) 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)} 封" 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 = int(mid) if uid <= last_uid: # 防御：UID n:* 在 n 超过邮箱最大 UID 时会返回最后一封 continue last_uid = uid ustr = str(uid) # 两段式：先取头部（省流量），命中退信特征再取全文 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:{ustr}" if msg_id in skip_ids: continue scanned += 1 if not is_bounce_message(head): continue typ, fdata = conn.uid("fetch", ustr, "(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, "last_uid": last_uid, "uidvalidity": uv} finally: try: conn.logout() except Exception: # noqa: BLE001 pass def check_bounces(cfg, args): """登录发件邮箱 IMAP，扫描退信并解析失败收件人。 返回 {email: {"subject","date","category","reason","detail"}}；已处理过的 邮件（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, 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, 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: []} for em, info in bounced.items(): groups.setdefault(info["category"], []).append(em) prune_cats = [CAT_RATE] + ([CAT_UNKNOWN] if getattr(args, "prune_unknown", False) else []) print("\n========================================================") print(f"退信核查完成：扫描 {scanned} 封未处理邮件，其中退信 {bounce_cnt} 封，" f"涉及 {len(bounced)} 个收件人。") for cat in (CAT_RATE, CAT_HARD, CAT_UNKNOWN): ems = sorted(groups.get(cat, [])) if not ems: continue print(f"\n [{CAT_LABEL[cat]}] {len(ems)} 个：") for em in ems: info = bounced[em] why = info["reason"] or "未识别到明确原因" meta = " · ".join(x for x in (info["date"], info["subject"][:32], why) if x) print(f" - {em} （{meta}）") if not bounced: print("未发现新的退信，此前发送均正常送达（至少未被退回）。") else: # 硬退信：不补发 + 记入无效地址清单永久跳过 hard = sorted(groups.get(CAT_HARD, [])) if hard and not getattr(args, "no_invalid_list", False): n = record_invalid( args.invalid_file, {em: bounced[em] for em in hard}, ) print(f"\n[无效地址] {len(hard)} 个永久失败地址已记入 {args.invalid_file}" f"（新增 {n} 条）；这些地址后续运行（含 --reset-state）会直接跳过，不再浪费额度。") print(" 确认地址已修正后，删掉该文件或 --reset-invalid 可恢复发送。") elif hard: print(f"\n[无效地址] {len(hard)} 个永久失败地址未写入清单（--no-invalid-list），" f"但它们不会被补发。") # 可重试：从断点清单剔除，下次/本次补发 retryable = [em for cat in prune_cats for em in groups.get(cat, [])] if args.prune_state: n = prune_state(args.state, retryable) print(f"[剔除] 已从断点续发清单剔除 {n} 条「可重试」记录：{args.state}") if getattr(args, "send", False): print(" 本次发送将继续进行，这些用户会被包含在内（剔除并补发）。") else: print(" 本次不发送；下次重跑（同参数）会自动给这些用户补发。") elif retryable: print(f"\n[提示] 有 {len(retryable)} 个「可重试」退信等待补发。可选：" f"--prune-state 仅剔除；--prune-state --send 剔除后立即补发。") unknown = groups.get(CAT_UNKNOWN, []) if unknown and not getattr(args, "prune_unknown", False): print(f"[提示] 有 {len(unknown)} 个退信未能分类，默认不补发（保守）。" f"确认是限流后可加 --prune-unknown 一并剔除补发。") # 退信核查明细报告（便于事后核对，不受 --no-report 影响） try: with open(args.bounce_report, "w", encoding="utf-8") as f: json.dump({ "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "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)}, "recipients": {em: bounced[em] for em in sorted(bounced)}, }, f, ensure_ascii=False, indent=2) print(f"\n[退信报告] 已写入：{args.bounce_report}") 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 # --------------------------------------------------------------------------- # 主流程 # --------------------------------------------------------------------------- def parse_var(items): extra = {} for it in items or []: if "=" not in it: sys.exit(f"[FATAL] --var 格式应为 key=value：{it}") k, v = it.split("=", 1) k = k.strip() if not k: sys.exit(f"[FATAL] --var 的键不能为空：{it}") extra[k] = v return extra # --------------------------------------------------------------------------- # 参数文件（params.ini）：运行时自动读取，CLI 参数优先级更高。 # 注意：出于安全考虑，参数文件只能提供内容性参数（站点地址/自定义占位符/ # 限速/测试邮箱等），绝不会提供 --send / --yes——正式发送永远需要 # 命令行显式指定并在终端确认。 # --------------------------------------------------------------------------- def load_params(path): """读取参数文件；文件不存在返回空 dict（不报错，属可选配置）。""" 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 as e: # noqa: BLE001 sys.exit(f"[FATAL] 解析参数文件失败: {e}") 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} 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), ("smtp_idle_reconnect", float), ("batch_size", int)): 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 _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]") lines.append("; conf/app.ini 路径（相对路径按运行脚本时的当前目录解析）") lines.append(f"config = {config}") lines.append("") lines.append("[site]") lines.append(f"site_url = {site_url}") lines.append(f"box_prefix = {box_prefix}") lines.append("") lines.append("; 自定义占位符：模板里用 {{键名}} 引用") lines.append("[vars]") for k in sorted(extra_vars): lines.append(f"{k} = {extra_vars[k]}") lines.append("") lines.append("[send]") 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") def collect_template_placeholders(templates, builtin_keys): """扫描全部模板里出现的 {{占位符}}，返回非内置的键集合（用于交互模式逐项询问）。""" keys = set() for t in templates.values(): for txt in (t["subject"], t["html"], t["plain"]): keys.update(m.group(1) for m in _PLACEHOLDER_RE.finditer(txt)) return keys - builtin_keys def ask(prompt, default=""): """终端问答；非交互环境（EOF）返回默认值。""" try: suffix = f" [{default}]" if default != "" else "" ans = input(f"{prompt}{suffix}: ").strip() except EOFError: print() return default return ans if ans else default def run_interactive(cfg, templates, extra_vars, args, config_path): """交互模式：只补全缺失项——已有值的参数（params.ini / 命令行 / app.ini） 直接沿用不再询问；然后选择模板模式与运行方式。 返回 (site_url, box_prefix, extra_vars)；args.send / args.to / args.yes / args.limit / args.single 会按用户选择就地写入 args。""" site_url = (args.site_url or "").strip().rstrip("/") box_prefix = (args.box_prefix or "/_/").strip() # ---- 摘要：已有值直接沿用；只有缺的才问 ---- print("\n[参数确认] 以下直接沿用 params.ini / 命令行 / app.ini 的值，不再逐项询问" "（要改请编辑 params.ini，或本次直接写死到模板 html 里）：") print(f" site_url = {site_url or '(空)'}") 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: site_url = ask("站点地址 site_url（如 https://box.shiroko.one）", "") # 自定义占位符：只询问模板用到但还没有值的 builtin = {"name", "domain", "email", "box_link", "site", "site_title", "year"} need = [k for k in sorted(collect_template_placeholders(templates, builtin)) if not str(extra_vars.get(k, "")).strip()] for key in need: extra_vars[key] = ask(f"占位符 {{{{{key}}}}}（模板用到但还没有值；直接回车 = 留空）", "") # ---- 运行方式：在这里选择，而不是靠命令行 --send/--to/--yes ---- print("\n请选择本次运行方式：") print(" 1) 演练 —— 只列收件人、统计与渲染预览，不发信") print(" 2) 测试 —— 只给一个邮箱发一封，验证模板与发信链路") print(" 3) 群发 —— 向全体未注销用户发送（发送前仍会打印统计确认）") print(" 4) 退信核查 —— 登录发件箱扫描退信通知，核对哪些用户实际没收到") while True: try: mode = input("运行方式 [1/2/3/4]（直接回车 = 1）: ").strip() except EOFError: mode = "" if mode in ("", "1", "2", "3", "4"): break print(" 请输入 1、2、3 或 4") if mode == "4": args.check_bounces = True print("\n退信核查方式（会先按原因分类：限流/临时性 → 可重试；地址不存在 → 永久失败）：") print(" 1) 仅核查 —— 只报告哪些用户没收到、各自什么原因，不改动任何清单（默认）") print(" 2) 仅剔除 —— 只把「限流等可重试」的退信从断点清单剔除后结束，本次不发送") print(" （「地址不存在」的不会剔除，另记入无效地址清单永久跳过）") print(" 3) 剔除并补发 —— 同上剔除后继续正常发送流程（发送前仍有统计与确认）") while True: try: sub = input("核查方式 [1/2/3]（直接回车 = 1）: ").strip() except EOFError: sub = "" if sub in ("", "1", "2", "3"): break print(" 请输入 1、2 或 3") if sub in ("2", "3"): args.prune_state = True if sub == "3": args.send = True if args.prune_state: try: unk = input("未分类的退信是否也按「可重试」一并剔除补发？(y/N): ").strip().lower() except EOFError: unk = "" args.prune_unknown = unk in ("y", "yes") return site_url, box_prefix, extra_vars # ---- 模板模式：多语言 / 单文件通用 ---- print("\n请选择模板模式：") print(" 1) 多语言 —— 按 users.language 使用各自语言的模板发送（默认）") print(f" 2) 单文件 —— 所有人发送 templates/{SINGLE_TPL_NAME}.html 通用模板（以后复用改这一个文件）") while True: try: tm = input("模板模式 [1/2]（直接回车 = 1）: ").strip() except EOFError: tm = "" if tm in ("", "1", "2"): break print(" 请输入 1 或 2") args.single = (tm == "2") if mode == "2": args.send = True args.to = ask("测试收件邮箱", args.to or "") if not args.to: print(" [提示] 未填测试邮箱，本次退回演练模式") args.send = False elif mode == "3": args.send = True lim = ask("最多发送 N 封（直接回车 = 不限制；建议先小量试水）", "") if lim.isdigit() and int(lim) > 0: args.limit = int(lim) try: skip = input("发送前是否保留最终确认（打印统计后输入 yes）？(Y/n): ").strip().lower() except EOFError: skip = "" args.yes = skip in ("n", "no") # n = 跳过最终确认，直接发 # 询问是否保存回 params.ini try: save_ans = input("把以上参数保存到 params.ini 供下次直接使用？(y/N): ").strip().lower() except EOFError: 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, "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 build_arg_parser(): """构造命令行解析器（单独抽出来，便于测试与复用）。""" parser = argparse.ArgumentParser( description="TamaBox 站内信群发工具（读取 conf/app.ini，模板在 templates/ 目录，按用户语言发送）", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="""模板占位符： {{name}} {{domain}} {{box_link}} {{email}} {{site}} {{site_title}} 自定义：--var key=value 后模板里即可用 {{key}} 个性域名链接规则：box_link = {site-url}{box-prefix}{domain}；无个性域名时为站点首页 """, ) parser.add_argument( "-c", "--config", default=None, help="conf/app.ini 路径（默认依次取：params.ini [path] config -> 环境变量 " "TAMABOX_CONFIG_PATH -> ./conf/app.ini；交互模式会先询问）", ) parser.add_argument("--templates-dir", default=DEFAULT_TEMPLATES_DIR, help="模板目录（默认脚本同目录 templates/），需含 <语言>.html / <语言>.subject.txt") parser.add_argument("--site-url", default="", help="站点地址（如 https://box.shiroko.one）。未传则回退 app.ini 的 external_url") parser.add_argument("--box-prefix", default="/_/", help="个性域名链接前缀，默认 /_/（box_link = site-url + box-prefix + domain）") parser.add_argument("--var", action="append", metavar="KEY=VALUE", help="自定义占位符（可多次），如 --var old_domain=box.tama.guru") parser.add_argument("--dry-run", action="store_true", help="只列出收件人 + 统计，不真正发信（默认行为；与 --send 互斥）") parser.add_argument("--send", action="store_true", help="真正发信（不指定则仅 dry-run）") parser.add_argument("--to", metavar="EMAIL", help="只发给该测试地址（验证模板与发信链路），覆盖收件人列表") parser.add_argument("--limit", type=int, default=0, help="最多发送 N 封（0 表示不限制）") parser.add_argument("--delay", type=float, default=1.0, help="每封之间的间隔秒数，默认 1.0") parser.add_argument("--pause-every", type=int, default=0, help="每发送 N 封后额外长暂停一次（0 表示不启用）") parser.add_argument("--pause-for", type=float, default=15.0, 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("--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") parser.add_argument("--lang-column", default="auto", help="users 表语言列名。auto=自动探测（language/lang），都不存在则全员回退默认语言") parser.add_argument("--state", default=DEFAULT_STATE, help="断点续发状态文件路径（JSONL）；默认脚本同目录 broadcast_state.json") parser.add_argument("--reset-state", action="store_true", help="发送前清空断点续发状态，从头全量发送（慎用）；仅在 --send 时生效") parser.add_argument("--report", default=DEFAULT_REPORT, help="监控报告文件路径（JSON）；默认脚本同目录 broadcast_report.json") parser.add_argument("--no-report", action="store_true", help="关闭监控报告写入") parser.add_argument("--yes", action="store_true", help="跳过发送前的人工确认（非交互/自动化用）") parser.add_argument("--params", default=None, metavar="FILE", help=f"参数文件路径（默认自动读取脚本同目录 params.ini，不存在则跳过；CLI 参数优先级更高）") parser.add_argument("--no-params", action="store_true", help="不读取 params.ini 参数文件") parser.add_argument("--interactive", "-i", action="store_true", default=None, help="交互模式（默认开启：不带任何运行参数运行脚本时自动进入）。" "运行后逐项询问参数，并直接选择运行方式（演练/测试一封/群发）") parser.add_argument("--no-interactive", action="store_true", help="关闭交互模式，按命令行/params.ini 参数直接执行（脚本化用）") parser.add_argument("--single", action="store_true", help="单文件模式：所有人不论语言都发送 templates/single.html 通用模板" "（交互模式里也可选择）") parser.add_argument("--check-bounces", action="store_true", help="退信核查：登录发件邮箱 IMAP 扫描退信通知，找出实际未送达的收件人" "（SMTP 250 不代表送达；限流等退信是异步投到发件箱的）") parser.add_argument("--prune-state", action="store_true", help="配合 --check-bounces：把「可重试」退信（限流/临时性）从断点续发清单" "剔除后结束（仅剔除，本次不发送）；再加 --send 则剔除后立即补发。" "「地址不存在」等永久失败不会被剔除（不补发）") 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") parser.add_argument("--reset-invalid", action="store_true", help="清空无效地址清单（慎用：这些地址多半仍不可达）") parser.add_argument("--no-invalid-list", action="store_true", default=None, help="本次不使用无效地址清单：既不跳过已知无效地址，也不写入新的硬退信") parser.add_argument("--bounce-report", default=None, metavar="FILE", help="退信核查明细报告路径（JSON）；默认脚本同目录 broadcast_bounce_report.json") 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=None, help="IMAP 端口（默认 993，SSL）") parser.add_argument("--since", default=None, metavar="auto|YYYY-MM-DD", help="退信扫描起点：auto（默认）= 从群发记录（断点清单）最早一条的" "时间开始，只核查本次群发相关的退信；" "也可显式指定日期如 2026-09-07") parser.add_argument("--since-days", type=int, default=None, help="断点清单为空时的回退扫描范围（天），默认 3 天") parser.add_argument("--subject", default=None, help="全局覆盖邮件主题（忽略各语言 subject.txt；支持占位符）") parser.add_argument("--html", metavar="FILE", help="全局覆盖 HTML 正文模板文件（忽略按语言模板；调试用）") parser.add_argument("--plain", metavar="FILE", help="全局覆盖纯文本正文模板文件（调试用）") return parser def main(): global args args = build_arg_parser().parse_args() # 默认交互模式：没给任何运行方式相关参数（--send/--to/--yes/--dry-run）就自动进入交互； # 显式 --no-interactive 强制关闭（脚本化/定时任务用） if args.no_interactive: args.interactive = False elif args.interactive is None: args.interactive = not (args.send or args.to or args.yes or args.dry_run or args.check_bounces) # ---- 无效地址清单维护（--reset-invalid）：手动恢复发送硬退信地址时用 ---- if args.reset_invalid: if reset_invalid(args.invalid_file): print(f"[无效地址清单] 已清空：{args.invalid_file}") else: print(f"[无效地址清单] 文件不存在，无需清空：{args.invalid_file}") # 载入模板 templates, lang_list = load_templates(args.templates_dir) single_tpl = load_single_template(args.templates_dir) if args.single and single_tpl is None: sys.exit(f"[FATAL] 指定了 --single，但模板目录里没有 {SINGLE_TPL_NAME}.html: {args.templates_dir}") override_html = override_plain = None if args.html: with open(args.html, "r", encoding="utf-8") as f: override_html = f.read() if args.plain: with open(args.plain, "r", encoding="utf-8") as f: override_plain = f.read() # ---- 参数文件：自动读取 params.ini（CLI 参数优先级更高）---- params = {} params_path = None if not args.no_params: params_path = args.params or DEFAULT_PARAMS params = load_params(params_path) if params: print(f"[参数文件] 已读取：{params_path}") elif args.params: print(f"[警告] 指定的参数文件不存在，已跳过：{params_path}") if not args.site_url and params.get("site_url"): args.site_url = params["site_url"] if params.get("box_prefix"): args.box_prefix = params["box_prefix"] # 文件值覆盖默认 /_/ if not args.to and params.get("to"): args.to = params["to"] if args.limit == 0 and params.get("limit") is not None: args.limit = params["limit"] if args.delay == 1.0 and params.get("delay") is not None: 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"]) 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: 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 ---- config_path = (args.config or params.get("config") or os.environ.get("TAMABOX_CONFIG_PATH") or "conf/app.ini") config_explicit = bool(args.config or params.get("config")) # ---- 交互模式：配置路径已明确指定（CLI/params.ini）则不问，仅默认猜测时补问 ---- if args.interactive: if config_explicit: print(f"[交互模式] 配置文件沿用：{config_path}") else: config_path = ask("配置文件路径（conf/app.ini）", config_path) cfg = load_config(config_path) site_url = (args.site_url or "").strip().rstrip("/") # site_url 兜底链：CLI/params.ini > app.ini external_url（交互与非交互一致， # 有值时交互模式不再询问 site_url） if not site_url: site_url = (cfg["app_external_url"] or "").strip().rstrip("/") if site_url: print(f"[提示] 未显式提供 site_url，回退使用 app.ini external_url = {site_url}") elif not args.interactive: print("[警告] site_url 为空：{{site}} 与无个性域名用户的 {{box_link}} 将为空串") if args.interactive: site_url, args.box_prefix, extra_vars = run_interactive( cfg, templates, extra_vars, args, config_path) # ---- 退信核查模式：不查数据库、不发信；扫描（含可选剔除）后结束， # 除非同时指定 --send（剔除并补发），此时继续走正常发送流程 ---- if args.check_bounces: print("\n[退信核查] SMTP 250 不代表真正送达（限流/无效地址常被异步退回发件箱）") bounced = check_bounces(cfg, args) if args.send: print("\n[退信核查] 处理完成，继续执行发送流程（被剔除的退信用户会包含在本次发送中）...") else: if bounced and not args.prune_state: print("\n[提示] --prune-state 仅剔除不发送；--prune-state --send 剔除后立即补发。") sys.exit(0) print("=" * 56) print(f"已解析配置（站内信群发工具 v{SCRIPT_VERSION}）") print("=" * 56) print(f" [配置] 文件 = {config_path}") print(f" [模板] 目录 = {args.templates_dir}") print(f" [模板] 语言 = {', '.join(lang_list)}（{', '.join(LANG_LABEL.get(l, l) for l in lang_list)}）") print(f" [模板] 模式 = {'单文件通用 single.html（不按语言）' if args.single else '多语言（按 users.language）'}") print(f" [站点] site_url = {site_url or '(空)'}") print(f" [站点] box_prefix = {args.box_prefix}") if extra_vars: for k in sorted(extra_vars): print(f" [自定义] {k:<12} = {extra_vars[k]}") print(f" [app] title = {cfg['app_title']}") print(f" [app] default_lang = {cfg['app_default_lang']}") print(f" [db] type = {cfg['db_type']}") print(f" [db] host:port = {cfg['db_host']}:{cfg['db_port']} db={cfg['db_name'] or '(空)'}") print(f" [mail] smtp:port = {cfg['mail_smtp']}:{cfg['mail_port']}") print(f" [mail] account = {cfg['mail_account'] or '(空)'}") print(f" [断点续发] state_file = {args.state}") print("=" * 56) if not cfg["mail_account"] or not cfg["mail_smtp"]: sys.exit("[FATAL] [mail] account/smtp 为空，无法发信，已中止。") drv = _choose_driver(cfg, args) print(f" [DB 驱动] {drv}") # 连接诊断：确认 psql 实际连到的库和用户 if drv == "psql": try: cur = run_query(cfg, "SELECT current_database(), current_user", args) if cur and cur[0]: print(f" [DB 连接] database={cur[0][0]} user={cur[0][1]}") except Exception: # noqa: BLE001 pass # 定位 users 表（避免 search_path 命中别的 schema 里的同名空表） users_table, schema_name, schema_info = resolve_users_table(cfg, args) if schema_name: others = "、".join( f"{s}({c}行)" for s, c, _ in schema_info["all"] if s != schema_name ) or "无" print( f" [users 表] {schema_name}.users" f"（{schema_info['rows']} 行，语言列={schema_info['lang_col'] or '无'}）；" f"其他同名表：{others}" ) if schema_info["rows"] == 0: print(" [警告] 所有 schema 的 users 表均为空，请确认库名/连接串是否指向生产库！") else: print(" [users 表] users（未做 schema 定位，按默认 search_path）") lang_expr, lang_name = detect_lang_column(cfg, args, users_table) print(f" [语言列] {lang_name if lang_name else '(无，全员回退默认语言 ' + str(cfg['app_default_lang']) + ')'}") recipients = fetch_recipients(cfg, args, lang_expr=lang_expr, tbl=users_table) counts = fetch_counts(cfg, args, tbl=users_table) # 跳过历史硬退信地址（收件人不存在 / DSN 5.x.x）——重发也发不出去， # 且不受 --reset-state 影响；只有 --no-invalid-list 或清空清单才会恢复。 invalid_skipped = 0 if not args.no_invalid_list: invalid_map = load_invalid(args.invalid_file) if invalid_map: before = len(recipients) recipients = [r for r in recipients if (r["email"] or "").strip().lower() not in invalid_map] invalid_skipped = before - len(recipients) print(f" [无效地址清单] 已载入 {len(invalid_map)} 条" f"（{args.invalid_file}），本次跳过 {invalid_skipped} 个地址") print("数据库用户统计：") print(f" 总用户数（含已注销） : {counts['total']}") print(f" 已注销用户数 : {counts['deactivated']}") print(f" 未注销用户数 : {counts['active']}") print(f" 未注销且有邮箱（应发送） : {counts['active_with_email']}") print(f" 未注销但无邮箱（跳过） : {counts['active_no_email']}") lang_dist = {} for r in recipients: lg = resolve_lang(r.get("language"), cfg["app_default_lang"], lang_list) lang_dist[lg] = lang_dist.get(lg, 0) + 1 dist_parts = [] for lg in [l for l in LANG_ORDER if l in lang_list]: if lang_dist.get(lg): dist_parts.append(f"{lg}({LANG_LABEL.get(lg, lg)}):{lang_dist[lg]}") for lg in lang_dist: if lg not in lang_list or lg not in LANG_ORDER: dist_parts.append(f"{lg}:{lang_dist[lg]}") print(f" 按语言分布（应发送） : {', '.join(dist_parts) if dist_parts else '-'}") if args.to: test_email = args.to.strip() match = None for r in recipients: if (r.get("email") or "").strip().lower() == test_email.lower(): match = r break if match: recipients = [match] resolved = f"{site_url}{args.box_prefix}{match['domain']}" if match["domain"] else (site_url or "(空)") print(f"[测试模式] 命中数据库用户：{match['name'] or '(无名)'} " f"(lang={match['language'] or '空'}, domain={match['domain'] or '空'})，" f"套用该用户数据发送，box_link={resolved}") else: recipients = [{"name": "测试", "domain": "", "email": test_email, "language": ""}] print(f"[测试模式] 测试邮箱 {test_email} 在数据库中未找到，按简体中文（zh-CN）发送") skipped = 0 had_state = False if not args.to: sent_set = load_sent_state(args.state) had_state = bool(sent_set) if args.reset_state: if args.send: reset_state(args.state) sent_set = {} print(" [断点续发] 已按 --reset-state 清空状态，将从头全量发送。") else: print(" [断点续发] --reset-state 仅在 --send 时生效，dry-run 未清空。") if sent_set: before = len(recipients) recipients = [ r for r in recipients if (r["email"] or "").strip().lower() not in sent_set ] skipped = before - len(recipients) if skipped: print(f" [断点续发] 已跳过 {skipped} 封（此前已发送，见状态文件）") # --limit 必须在断点续发过滤之后应用：先剔除已发，再取前 N 封， # 否则会永远截到名单头部那批已发用户，跳过后剩 0 封。 if args.limit and args.limit > 0: if args.to: print("[注意] --to 与 --limit 同时指定时，--limit 不生效（仅 1 个测试地址）。") else: recipients = recipients[: args.limit] print(f"[限制] 仅发送前 {len(recipients)} 封（已剔除此前已发送的 {skipped} 封）。") if not args.to and had_state: print(f" [断点续发] 剩余待发送：{len(recipients)} 封") print(f"\n本次将发送：{len(recipients)} 封邮件。") report = { "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "script_version": SCRIPT_VERSION, "mode": "send" if args.send else "dry-run", "template_mode": "single" if args.single else "multilang", "config": { "site_url": site_url, "box_prefix": args.box_prefix, "extra_vars": extra_vars, "app_title": cfg["app_title"], "mail_account": cfg["mail_account"] or "(空)", "mail_smtp": cfg["mail_smtp"] or "(空)", "db_type": cfg["db_type"], "templates_dir": args.templates_dir, }, "db_stats": counts, "lang_distribution": dict(lang_dist), "target": len(recipients), "sent": 0, "failed": 0, "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), "test_email": args.to or None, "state_file": args.state, "report_file": None if args.no_report else args.report, } def write_report_now(): if args.no_report: return try: with open(args.report, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) print(f"\n[监控报告] 已写入：{args.report}") except Exception as e: # noqa: BLE001 print(f"[警告] 写入监控报告失败：{e}") if not recipients: report["note"] = "没有可发送的收件人" write_report_now() print("没有可发送的收件人，结束。") return # 预览前 3 条（邮箱打码），并展示解析出的 box_link，便于发现模板/链接问题 print("预览收件人（前 3 条，邮箱打码）：") for r in recipients[:3]: e = r["email"] masked = (e[:2] + "***" + e[e.rfind("@"):]) if "@" in e else e link = f"{site_url}{args.box_prefix}{r['domain']}" if r["domain"] else (site_url or "(空)") print(f" - {r['name'] or '(无名)'} <{masked}> domain={r['domain'] or '-'} " f"lang={resolve_lang(r.get('language'), cfg['app_default_lang'], lang_list)} box_link={link}") if not args.send: write_report_now() print("\n[dry-run] 未指定 --send，仅演练，未发送任何邮件。") print("如需正式发送，请追加 --send（会先要求输入 yes 确认；可先 --to 测试 / --limit 小批量）。") return if not args.yes: try: ans = input( f"\n请确认向以上 {len(recipients)} 位用户发送邮件（输入 yes 并回车继续，其他任意输入取消）：" ).strip().lower() except EOFError: print("\n[中止] 非交互环境未提供确认，已取消发送。如需自动确认请加 --yes。") report["note"] = "非交互环境未确认，已取消" write_report_now() return if ans not in ("yes", "y"): print("[已取消] 未确认，未发送任何邮件。") report["note"] = "用户取消" write_report_now() return sent, failed = 0, [] smtp_rejected = 0 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 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 = {}, {}, "" 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: # 单文件模式：不按语言分组，所有人同一模板 groups = {"single": list(recipients)} group_order = ["single"] else: groups = {} for r in recipients: lg = resolve_lang(r.get("language"), cfg["app_default_lang"], lang_list) groups.setdefault(lg, []).append(r) group_order = [lg for lg in lang_list if lg in groups] + [lg for lg in groups if lg not in lang_list] try: session = SmtpSession(cfg, idle_reconnect=getattr(args, "smtp_idle_reconnect", 30)) except Exception as e: # noqa: BLE001 sys.exit(f"[FATAL] SMTP 登录失败: {e}") try: for gi, lg in enumerate(group_order, 1): grp = groups[lg] if args.single: label = "通用模板 single.html" tpl = single_tpl else: 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 = { "name": r["name"] or "用户", "domain": r["domain"] or "", "email": r["email"], "box_link": box_link, "site": site_url, "site_title": cfg["app_title"], "year": datetime.now().strftime("%Y"), **extra_vars, } s, h, p, missing = render_content(subject_raw, html_raw, plain_raw, ctx) if missing: all_missing.update(missing) 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()}) 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}") # 周期巡检收件箱：按收件人计数，异步退信才是限流最常见表现 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)) if abort_reason: break finally: session.close() 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 report["failed_list"] = [{"email": e, "error": err} for e, err in failed] report["missing_placeholders"] = sorted(all_missing) write_report_now() print("\n" + "=" * 56) 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: print(f"断点续发状态已写入：{args.state}（重跑脚本会自动跳过已发用户）") print("[提示] SMTP 250 不代表一定送达（限流/无效地址常被异步退回发件箱）。" "发送完过几分钟可运行「python3 broadcast.py --check-bounces --prune-state」" "核查退信：限流类会自动剔除补发，地址不存在类记入无效清单不再重发。") if failed: print("失败清单：") for e, err in failed: print(f" - {e}: {err}") print("=" * 56) if __name__ == "__main__": main()