mailman-sync/mailman_sync.py

#!/usr/bin/env python3

"""
Update Mailman 2 lists via a json API of the form {"LIST": ["ADDRESS", ...]}
"""

import argparse
from pathlib import Path
import subprocess

import requests


def sync_members(mailman_bin: Path, mailing_list: str, members: list[str]):
    members_data = "\n".join(members)
    output = subprocess.run(
        [
            mailman_bin / "sync_members",
            "--no-change",
            "--welcome-msg=no",
            "--goodbye-msg=no",
            "--notifyadmin=no",
            "--file",
            "-",
            mailing_list,
        ],
        input=members_data,
        encoding="ascii",
        capture_output=True,
        check=True,
    )
    print(output.stdout)


def main(mailman_bin: Path, api: str, token: str, list_suffix: str):
    r = requests.get(api, headers={"Authorization": "Token " + token})
    if not r.ok:
        print(f"Failed to get mailing list data from api: {r.status_code} {r.text}")
        return

    existing_lists = subprocess.run(
        [mailman_bin / "list_lists", "-b"],
        encoding="ascii",
        capture_output=True,
        check=True,
    ).stdout.split("\n")
    certification_lists = r.json()
    for name, members in certification_lists.items():
        list_name = name + list_suffix
        if list_name in existing_lists:
            sync_members(mailman_bin, name, members)
        else:
            print(f"Skipping {list_name}, as it does not exist in Mailman")


if __name__ == "__main__":
    argp = argparse.ArgumentParser(description=__doc__)
    argp.add_argument(
        "--bin",
        default="/usr/local/mailman",
        type=Path,
        help="Path to Mailman site admin scripts (default %(default)s)",
    )
    argp.add_argument("--api", required=True, help="API endpoint to retrieve JSON from")
    argp.add_argument("--token", required=True, help="Authorization Token for API")
    argp.add_argument("--list-suffix", help="Suffix for mailing lists")
    args = argp.parse_args()

    try:
        main(args.bin, args.api, args.token, args.list_suffix)
    except subprocess.CalledProcessError as e:
        print(e.stderr)
        raise