mailman-sync/mailman_sync.py

#!/usr/bin/env python3

"""
Update Mailman 3 lists via a json API of the form:
{
  LIST: {
    "config"?: {
      "real_name": REAL_NAME,
      "subject_prefix": PREFIX,
      "reply_to_address": REPLY_TO_ADDRESS,
    }
    "moderators"?: [ADDRESS, ...],
    "members": [ADDRESS, ...]
  },
  ...
}
"""

import argparse
import os
import email.utils

import mailmanclient
import requests


def config_list(
    list: mailmanclient.MailingList,
    dry_run: bool,
    real_name: str,
    subject_prefix: str,
    reply_to_address: str,
):
    # TODO: some of this is announcement list specific, some not
    config_changes = {
        "display_name": real_name,
        "subject_prefix": subject_prefix,
        "reply_to_address": reply_to_address,
        "reply_goes_to_list": "explicit_header_only",
        "first_strip_reply_to": True,
        # quiet member management
        "send_welcome_message": False,
        "send_goodbye_message": False,
        "allow_list_posts": False,
        "anonymous_list": True,
        "archive_policy": "private",
        "member_roster_visibility": "moderators",
        "subscribe_policy": "confirm_then_moderate",
        "unsubscribe_policy": "confirm",
        "default_member_action": "hold",
        "default_nonmember_action": "discard",
        "advertised": False,
    }
    for setting, value in config_changes.items():
        old_value = list.settings.get(setting)
        if old_value != value:
            print(f"Changed {setting}: {old_value} -> {value}")

    if not dry_run:
        list.settings.update(config_changes)
        list.settings.save()


# TODO: this will never update someone's display name
def diff_roster(
    expected_members: list[str], existing_members: list[mailmanclient.Member]
) -> tuple[dict[str, str], list[mailmanclient.Member]]:
    expected_members_dict = dict(
        list(reversed(email.utils.parseaddr(member))) for member in expected_members
    )
    existing_members_dict = {member.address: member for member in existing_members}
    members_to_add = {
        k: expected_members_dict[k]
        for k in set(expected_members_dict) - set(existing_members_dict)
    }
    members_to_remove = [
        existing_members_dict[k]
        for k in set(existing_members_dict) - set(expected_members_dict)
    ]
    return members_to_add, members_to_remove


def sync_members(
    list: mailmanclient.MailingList, dry_run: bool, expected_members: list[str]
):
    members_to_add, members_to_remove = diff_roster(expected_members, list.members)

    for address, display_name in members_to_add.items():
        print(f"Adding '{display_name} <{address}>' to list {list}")
        if not dry_run:
            list.subscribe(
                address,
                display_name,
                pre_verified=True,
                pre_confirmed=True,
                pre_approved=True,
                send_welcome_message=False,
            )

    # TODO: could use `.mass_unsubscribe()` instead
    for member in members_to_remove:
        print(f"Removing {member} from list {list}")
        if not dry_run:
            list.unsubscribe(member.address, pre_approved=True, pre_confirmed=True)


def sync_moderators(
    list: mailmanclient.MailingList, dry_run: bool, expected_members: list[str]
):
    members_to_add, members_to_remove = diff_roster(expected_members, list.moderators)

    for address, display_name in members_to_add.items():
        print(f"Adding '{display_name} <{address}>' as moderator to list {list}")
        if not dry_run:
            list.subscribe(address, display_name)

    for member in members_to_remove:
        print(f"Removing {member} as moderator from list {list}")
        if not dry_run:
            list.remove_moderator(member.address)


def main(
    api: str,
    api_auth: str,
    mailman_client: mailmanclient.Client,
    dry_run: bool,
    mail_host: str,
):
    r = requests.get(api, headers={"Authorization": api_auth})
    if not r.ok:
        print(f"Failed to get mailing list data from api: {r.status_code} {r.text}")
        return
    expected_lists = r.json()

    domain = mailman_client.get_domain(mail_host)
    existing_lists = {list.list_name: list for list in domain.get_lists()}

    for name, props in expected_lists.items():
        if name in existing_lists:
            list = existing_lists[name]
        elif dry_run:
            print(f"Skipping non-existing list {name} in dry run mode")
            continue
        else:
            list = domain.create_list(name, props.get("style"))

        if "config" not in props:
            print(f"Not configuring {name}, as it has no config section")
        else:
            print(f"Configuring/syncing {name}...")
            config_list(list, dry_run, **props["config"])

        if "moderators" in props:
            sync_moderators(list, dry_run, props["moderators"])
        sync_members(list, dry_run, props["members"])


def parse_arguments():
    argp = argparse.ArgumentParser(description=__doc__)
    argp.add_argument("--api", required=True, help="API endpoint to retrieve JSON from")
    argp.add_argument("--mail-host", help="Base domain for all lists", required=True)
    argp.add_argument(
        "--mailman-url",
        help="base URL for for Mailman3 REST API",
        default="http://localhost:9001",
    )
    argp.add_argument(
        "--mailman-user", help="Username for Mailman3 REST API", default="restadmin"
    )
    argp.add_argument(
        "-n",
        "--dry-run",
        action="store_true",
        help="Don't make changes, just print what would happen",
    )
    return argp.parse_args()


if __name__ == "__main__":
    args = parse_arguments()

    for env_var in ("API_AUTH", "MAILMAN_PASS"):
        if env_var not in os.environ:
            print(f"Missing {env_var} environment variable")
            exit(-1)

    mailman_client = mailmanclient.Client(
        args.mailman_url + "/3.1",
        args.mailman_user,
        os.environ["MAILMAN_PASS"],
    )

    main(
        args.api,
        os.environ["API_AUTH"],
        mailman_client,
        args.dry_run,
        args.mail_host,
    )