tamakyi
6dc33604d1
- 渲染后内容 sha256 分桶，仅完全相同的收件人合并信封；含 {{name}} 等
个人化占位符时自动回退逐封，不会错合
- 一个信封 = 1×MAIL FROM + N×RCPT TO + 1×DATA：发信次数从「人数」降到
「信封数」，显著降低触发「外发频率超过限制」类限流的概率
- To: 头只显示站点名+发件邮箱，真实收件人走 RCPT TO 信封互相不可见
- 452/too many recipients 特判：不算限流不中止，batch_size 砍半重试
- refused dict 逐人处理：5xx 记无效清单、限流码仍触发立即中止
- SmtpSession.send_to 统一重连重试；断点续发按人记录；巡检按收件人计数
- 顺带修复：交互模式保存 params.ini 时会抹掉手工加的 smtp_idle_reconnect
- 默认 0（逐封）不变，--batch-size N 或 params.ini [send] batch_size 显式开启
2494 lines
113 KiB
Python
2494 lines
113 KiB
Python
#!/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 [email protected] \
|
||
--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/<lang>.html（必需）、<lang>.subject.txt（必需）、
|
||
# <lang>.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]) 防御：避免把乱码前缀吞进邮箱地址）
|
||
LABELED_PATTERNS = (
|
||
r"(?:收信地址|收件人地址|退信地址|收件人|无法送达的地址)[：:\s\r\n]*"
|
||
r"(?<![A-Za-z0-9])(?P<e>[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])(?P<e>[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"(?<![\d.])[245]\d{2}(?![\d.])")
|
||
|
||
|
||
def _decode_hdr(value):
|
||
"""解码 MIME 编码的邮件头（Subject 等）为可读文本。"""
|
||
if not value:
|
||
return ""
|
||
try:
|
||
return "".join(
|
||
frag.decode(cs or "utf-8", "replace") if isinstance(frag, bytes) else frag
|
||
for frag, cs in decode_header(str(value))
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
return str(value)
|
||
|
||
|
||
def _part_payload_text(part):
|
||
"""取单个 part 的文本。
|
||
|
||
两个坑都要避开：
|
||
- 直接 get_payload(decode=True) 遇到 str 形态的非 ASCII payload 会走
|
||
raw-unicode-escape，中文变成 \\uXXXX 字面串，关键词匹配全部失效；
|
||
- 反过来只用 get_payload() 又可能拿到 base64/QP 的原文（utf-8 正文默认
|
||
base64 传输编码），同样匹配不到。
|
||
所以：非 ASCII 的 str 直接用（已解码），其余交给 decode=True 解 base64/QP。
|
||
"""
|
||
try:
|
||
raw = part.get_payload()
|
||
if isinstance(raw, str) and not raw.isascii():
|
||
return raw
|
||
raw = part.get_payload(decode=True)
|
||
if isinstance(raw, bytes):
|
||
return raw.decode(part.get_content_charset() or "utf-8", "replace")
|
||
if isinstance(raw, str):
|
||
return raw
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _part_text(msg, limit=30000):
|
||
"""抽取邮件正文文本（plain / html / DSN 原始块），拼接后截断。"""
|
||
chunks = []
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
ct = part.get_content_type()
|
||
if ct in ("text/plain", "text/html"):
|
||
chunks.append(_part_payload_text(part))
|
||
elif ct == "message/delivery-status":
|
||
blocks = part.get_payload()
|
||
if isinstance(blocks, list):
|
||
for blk in blocks:
|
||
try:
|
||
for k, v in blk.items():
|
||
chunks.append(f"{k}: {v}")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
if sum(len(c) for c in chunks) > 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 _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。"""
|
||
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()
|
||
|
||
try:
|
||
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: []}
|
||
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,
|
||
"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}")
|
||
|
||
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}")
|
||
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("--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()
|