Mailman3 批量初始化用户与邮件列表

WEB 端批量注册用户

Mailman Web 默认使用 django-allauth 处理注册与邮箱验证。为避免每个用户自行注册并点击验证链接，最直接的做法是在 /opt/mailman/web/settings.py 中关闭邮箱验证。在 settings.py 中新增或修改以下配置：

# 关闭邮箱验证（最关键）
ACCOUNT_EMAIL_VERIFICATION = "none"

# 允许不验证也能登录（一般默认就是 True，但明确写上）
ACCOUNT_EMAIL_REQUIRED = True
ACCOUNT_AUTHENTICATION_METHOD = "username_email"  # 你的版本若提示 deprecated 可忽略或按新项改

# 可选：如果你不希望开放自助注册，直接关掉注册入口（强烈建议内网这样做）
ACCOUNT_ALLOW_REGISTRATION = False   # 若版本不支持就用 URL/模板层面禁用 /accounts/signup/

bash

在 /opt/mailman/web/scripts 目录下创建 create_web_users.py：

#!/opt/mailman/venv/bin/python
import os
import sys

# === Django bootstrap (must be BEFORE importing Django models) ===
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings")
sys.path.insert(0, "/opt/mailman/web")

import django
django.setup()

# === then import Django/allauth models ===
from django.contrib.auth import get_user_model
from django.db import transaction
from allauth.account.models import EmailAddress

PASSWORD = "111111"

emails = [
    "wanghh@example.com",
]

User = get_user_model()

created = 0
updated = 0

with transaction.atomic():
    for email in emails:
        email = email.lower().strip()
        username = email.split("@")[0]

        user, is_new = User.objects.get_or_create(
            username=username,
            defaults={"email": email, "is_active": True},
        )

        user.email = email
        user.is_active = True
        user.set_password(PASSWORD)
        user.save()

        ea, _ = EmailAddress.objects.get_or_create(
            user=user,
            email=email,
            defaults={"primary": True, "verified": True},
        )
        ea.primary = True
        ea.verified = True
        ea.save()

        if is_new:
            created += 1
        else:
            updated += 1

print(f"Web users done: created={created}, updated={updated}, total={len(emails)}")

python

随后使用虚拟环境中的 python 运行：

/opt/mailman/venv/bin/python create_web_users.py

bash

Mailman 3 创建 list 并批量添加订阅者

在 /opt/mailman/scripts 目录下创建 create_list.py：

#!/opt/mailman/venv/bin/python
# -*- coding: utf-8 -*-

"""
Create-or-update a Mailman3 list and batch-add members from member.txt.
Also ensures default owners are assigned.

Usage:
  /opt/mailman/venv/bin/python create_list.py dev
  /opt/mailman/venv/bin/python create_list.py dev --domain lists.example.com --members-file member.txt

Env (optional):
  MAILMAN_REST_URL   default: http://127.0.0.1:8001/3.1
  MAILMAN_REST_USER  default: restadmin
  MAILMAN_REST_PASS  default: restpass
"""

import argparse
import os
import sys
import time
from urllib.error import HTTPError

from mailmanclient import Client

DEFAULT_OWNERS = [
    "pengyq@example.com",
    "wangwy@example.com",
    "xiaorzh@example.com",
]


def read_emails(path: str) -> list[str]:
    emails: list[str] = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            s = line.strip()
            if not s or s.startswith("#"):
                continue
            emails.append(s)
    # 去重但保持顺序
    seen = set()
    out = []
    for e in emails:
        if e not in seen:
            seen.add(e)
            out.append(e)
    return out


def http_status(e: Exception) -> int | None:
    # urllib.error.HTTPError 有 code 字段
    return getattr(e, "code", None)


def get_or_create_domain(client: Client, domain: str):
    """
    Return domain proxy; create if not exists.
    """
    try:
        return client.get_domain(domain)
    except HTTPError as e:
        if http_status(e) == 404:
            try:
                client.create_domain(domain)
            except HTTPError as e2:
                # 可能域已存在：400 Duplicate email host / 409 Conflict
                if http_status(e2) not in (400, 409):
                    raise
            return client.get_domain(domain)
        raise


def robust_get_list(client: Client, list_id: str, retries: int = 3, sleep_s: float = 0.5):
    """
    After a create_list 500, the list may still have been created.
    Retry-get to confirm.
    """
    last = None
    for _ in range(retries):
        try:
            return client.get_list(list_id)
        except HTTPError as e:
            last = e
            if http_status(e) == 404:
                time.sleep(sleep_s)
                continue
            raise
    # retries exhausted
    if last:
        raise last
    raise RuntimeError("unexpected")


def get_or_create_list(client: Client, domain: str, local_part: str):
    """
    Return (mlist, created_bool)
    list_id in Mailman REST is typically "<local_part>.<domain>".
    """
    list_id = f"{local_part}.{domain}"

    # 先查是否存在
    try:
        mlist = client.get_list(list_id)
        return mlist, False
    except HTTPError as e:
        if http_status(e) != 404:
            raise

    # 不存在 -> 创建
    dom = get_or_create_domain(client, domain)

    try:
        mlist = dom.create_list(local_part)
        return mlist, True

    except HTTPError as e:
        # 关键：有时服务端返回 500，但 list 已写入 DB
        if http_status(e) == 500:
            # 再查一次确认是否已创建
            mlist = robust_get_list(client, list_id, retries=5, sleep_s=0.5)
            return mlist, True

        # 已存在/冲突等：再查一次兜底
        if http_status(e) in (400, 409):
            mlist = robust_get_list(client, list_id, retries=3, sleep_s=0.2)
            return mlist, False

        raise


def ensure_owners(mlist, owners: list[str]) -> tuple[int, int]:
    """
    Ensure owners are assigned. Returns (added, skipped).
    Uses role API mlist.add_owner(address).
    """
    added = 0
    skipped = 0
    for addr in owners:
        try:
            mlist.add_owner(addr)
            added += 1
        except HTTPError as e:
            if http_status(e) in (400, 409):
                skipped += 1
                continue
            raise
    return added, skipped


def add_members(mlist, members: list[str]) -> tuple[int, int]:
    """
    Subscribe members in batch. Returns (added, skipped).
    pre_verified/pre_confirmed/pre_approved 可以绕过每人单独确认/验证。
    """
    added = 0
    skipped = 0
    for addr in members:
        try:
            mlist.subscribe(
                addr,
                pre_verified=True,
                pre_confirmed=True,
                pre_approved=True,
            )
            added += 1
        except HTTPError as e:
            if http_status(e) in (400, 409):
                skipped += 1
                continue
            raise
    return added, skipped


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("listname", help="mailing list local-part (e.g. dev)")
    ap.add_argument("--domain", default="lists.example.com", help="mail host/domain (default: lists.example.com)")
    ap.add_argument("--members-file", default="member.txt", help="members file path (default: member.txt)")
    args = ap.parse_args()

    rest_url = os.getenv("MAILMAN_REST_URL", "http://127.0.0.1:8001/3.1")
    rest_user = os.getenv("MAILMAN_REST_USER", "restadmin")
    rest_pass = os.getenv("MAILMAN_REST_PASS", "restpass")

    client = Client(rest_url, rest_user, rest_pass)

    if not os.path.exists(args.members_file):
        print(f"[FATAL] members file not found: {args.members_file}", file=sys.stderr)
        sys.exit(2)

    members = read_emails(args.members_file)

    mlist, created = get_or_create_list(client, args.domain, args.listname)
    print(f"[OK] list: {mlist.fqdn_listname} (created={created})")

    a, s = ensure_owners(mlist, DEFAULT_OWNERS)
    print(f"[OK] owners ensured: added={a}, skipped(existing/dup)={s}")

    a, s = add_members(mlist, members)
    print(f"[OK] members subscribed: added={a}, skipped(existing/dup)={s}")

    print("[DONE]")


if __name__ == "__main__":
    main()

python

使用虚拟环境中的 python 执行脚本：

/opt/mailman/venv/bin/python create_list.py "dev"

bash