Move list management functions to a class to reuse shared arguments
This commit is contained in:
parent
c49f3c7635
commit
9e3a042a58
122
mailman_sync.py
122
mailman_sync.py
@ -5,6 +5,7 @@ Update Mailman 2 lists via a json API of the form {"LIST": ["ADDRESS", ...]}
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import argparse
|
import argparse
|
||||||
|
from dataclasses import dataclass
|
||||||
import os
|
import os
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
import secrets
|
import secrets
|
||||||
@ -18,28 +19,33 @@ PASSWORD_CHARS = string.ascii_letters + string.digits + string.punctuation
|
|||||||
PASSWORD_LEN = 18
|
PASSWORD_LEN = 18
|
||||||
|
|
||||||
|
|
||||||
def newlist(mailman_bin: Path, listname: str, urlhost: str, admin: str):
|
@dataclass
|
||||||
password = "".join(secrets.choice(PASSWORD_CHARS) for i in range(PASSWORD_LEN))
|
class ListManager:
|
||||||
|
mailman_bin: Path
|
||||||
|
list_name: str
|
||||||
|
dry_run: bool
|
||||||
|
|
||||||
output = subprocess.run(
|
def newlist(self, urlhost: str, admin: str):
|
||||||
[
|
password = "".join(secrets.choice(PASSWORD_CHARS) for i in range(PASSWORD_LEN))
|
||||||
mailman_bin / "newlist",
|
|
||||||
"--quiet",
|
|
||||||
f"--urlhost={urlhost}",
|
|
||||||
listname,
|
|
||||||
admin,
|
|
||||||
password,
|
|
||||||
],
|
|
||||||
encoding="ascii",
|
|
||||||
capture_output=True,
|
|
||||||
check=True,
|
|
||||||
)
|
|
||||||
for line in output.stdout.splitlines():
|
|
||||||
print(f"[Creating {listname}] {line}")
|
|
||||||
|
|
||||||
|
output = subprocess.run(
|
||||||
|
[
|
||||||
|
self.mailman_bin / "newlist",
|
||||||
|
"--quiet",
|
||||||
|
f"--urlhost={urlhost}",
|
||||||
|
self.list_name,
|
||||||
|
admin,
|
||||||
|
password,
|
||||||
|
],
|
||||||
|
encoding="ascii",
|
||||||
|
capture_output=True,
|
||||||
|
check=True,
|
||||||
|
)
|
||||||
|
for line in output.stdout.splitlines():
|
||||||
|
print(f"[Creating {self.list_name}] {line}")
|
||||||
|
|
||||||
def config_list(mailman_bin: Path, mailing_list: str, dry_run: bool):
|
def config_list(self):
|
||||||
config_changes = """
|
config_changes = """
|
||||||
# TODO: set real_name, moderator, subject prefix, and reply-to address
|
# TODO: set real_name, moderator, subject prefix, and reply-to address
|
||||||
from_is_list = 1
|
from_is_list = 1
|
||||||
anonymous_list = 1
|
anonymous_list = 1
|
||||||
@ -60,54 +66,51 @@ generic_nonmember_action = 3 # discard non-member emails
|
|||||||
forward_auto_discards = 0 # don't notify admin about discards
|
forward_auto_discards = 0 # don't notify admin about discards
|
||||||
"""
|
"""
|
||||||
|
|
||||||
with tempfile.NamedTemporaryFile("w", suffix=".py") as config_file:
|
with tempfile.NamedTemporaryFile("w", suffix=".py") as config_file:
|
||||||
config_file.write(config_changes)
|
config_file.write(config_changes)
|
||||||
config_file.flush()
|
config_file.flush()
|
||||||
|
|
||||||
|
command = [
|
||||||
|
self.mailman_bin / "config_list",
|
||||||
|
"--inputfile",
|
||||||
|
config_file.name,
|
||||||
|
self.list_name,
|
||||||
|
]
|
||||||
|
if self.dry_run:
|
||||||
|
command.append("--checkonly")
|
||||||
|
|
||||||
|
output = subprocess.run(
|
||||||
|
command,
|
||||||
|
encoding="ascii",
|
||||||
|
capture_output=True,
|
||||||
|
check=True,
|
||||||
|
)
|
||||||
|
for line in output.stdout.splitlines():
|
||||||
|
print(f"[Configuring {self.list_name}] {line}")
|
||||||
|
|
||||||
|
def sync_members(self, members: list[str]):
|
||||||
command = [
|
command = [
|
||||||
mailman_bin / "config_list",
|
self.mailman_bin / "sync_members",
|
||||||
"--inputfile",
|
"--welcome-msg=no",
|
||||||
config_file.name,
|
"--goodbye-msg=no",
|
||||||
mailing_list,
|
"--notifyadmin=no",
|
||||||
|
"--file",
|
||||||
|
"-",
|
||||||
|
self.list_name,
|
||||||
]
|
]
|
||||||
if dry_run:
|
if self.dry_run:
|
||||||
command.append("--checkonly")
|
command.append("--no-change")
|
||||||
|
|
||||||
|
members_data = "\n".join(members)
|
||||||
output = subprocess.run(
|
output = subprocess.run(
|
||||||
command,
|
command,
|
||||||
|
input=members_data,
|
||||||
encoding="ascii",
|
encoding="ascii",
|
||||||
capture_output=True,
|
capture_output=True,
|
||||||
check=True,
|
check=True,
|
||||||
)
|
)
|
||||||
for line in output.stdout.splitlines():
|
for line in output.stdout.splitlines():
|
||||||
print(f"[Configuring {mailing_list}] {line}")
|
print(f"[Syncing {self.list_name}] {line}")
|
||||||
|
|
||||||
|
|
||||||
def sync_members(
|
|
||||||
mailman_bin: Path, mailing_list: str, members: list[str], dry_run: bool
|
|
||||||
):
|
|
||||||
command = [
|
|
||||||
mailman_bin / "sync_members",
|
|
||||||
"--welcome-msg=no",
|
|
||||||
"--goodbye-msg=no",
|
|
||||||
"--notifyadmin=no",
|
|
||||||
"--file",
|
|
||||||
"-",
|
|
||||||
mailing_list,
|
|
||||||
]
|
|
||||||
if dry_run:
|
|
||||||
command.append("--no-change")
|
|
||||||
|
|
||||||
members_data = "\n".join(members)
|
|
||||||
output = subprocess.run(
|
|
||||||
command,
|
|
||||||
input=members_data,
|
|
||||||
encoding="ascii",
|
|
||||||
capture_output=True,
|
|
||||||
check=True,
|
|
||||||
)
|
|
||||||
for line in output.stdout.splitlines():
|
|
||||||
print(f"[Syncing {mailing_list}] {line}")
|
|
||||||
|
|
||||||
|
|
||||||
def main(
|
def main(
|
||||||
@ -133,16 +136,17 @@ def main(
|
|||||||
certification_lists = r.json()
|
certification_lists = r.json()
|
||||||
for name, members in certification_lists.items():
|
for name, members in certification_lists.items():
|
||||||
list_name = name + list_suffix
|
list_name = name + list_suffix
|
||||||
|
list_manager = ListManager(mailman_bin, list_name, dry_run)
|
||||||
if list_name not in existing_lists:
|
if list_name not in existing_lists:
|
||||||
if dry_run:
|
if dry_run:
|
||||||
print(f"Skipping non-existing list {list_name} in dry run mode")
|
print(f"Skipping non-existing list {list_name} in dry run mode")
|
||||||
continue
|
continue
|
||||||
else:
|
else:
|
||||||
newlist(mailman_bin, list_name, urlhost, admin)
|
list_manager.newlist(urlhost, admin)
|
||||||
|
|
||||||
print(f"Configuring/syncing {list_name}...")
|
print(f"Configuring/syncing {list_name}...")
|
||||||
config_list(mailman_bin, list_name, dry_run)
|
list_manager.config_list()
|
||||||
sync_members(mailman_bin, list_name, members, dry_run)
|
list_manager.sync_members(members)
|
||||||
|
|
||||||
|
|
||||||
def parse_arguments():
|
def parse_arguments():
|
||||||
|
Loading…
Reference in New Issue
Block a user