199 lines
8.2 KiB
Python
199 lines
8.2 KiB
Python
from zeep.client import Client as ZeepClient, Settings as ZeepSettings
|
|
from zeep.exceptions import Fault
|
|
|
|
from .lists import MailingList, MLUser
|
|
from .constants import SUBTOPICS, TOPICS, TEMPLATES
|
|
|
|
|
|
class Client:
|
|
def __init__(self, sympa_url: str):
|
|
self.sympa_url = sympa_url
|
|
self.zeep = ZeepClient(sympa_url + "/wsdl", settings=ZeepSettings(strict=False))
|
|
self.cookie = None
|
|
|
|
def login(self, email: str, password: str) -> None:
|
|
"""
|
|
Login into the API. Set a cookie for future connexions.
|
|
"""
|
|
result = self.zeep.service.login(email, password)
|
|
element = result._raw_elements[0]
|
|
self.cookie = element.text
|
|
self.zeep.settings.extra_http_headers = [("Cookie", f"sympa_session={element.text}")]
|
|
if self.check_cookie() != email:
|
|
# FIXME Better exception
|
|
raise Exception("Unknown error: given cookie is invalid")
|
|
self.email = email
|
|
print("Successfully authenticated!")
|
|
|
|
def logout(self):
|
|
"""
|
|
Clear cookie
|
|
"""
|
|
self.cookie = None
|
|
self.email = None
|
|
self.zeep.settings.extra_http_headers = []
|
|
|
|
def check_cookie(self) -> str:
|
|
"""
|
|
From the current cookie, retrieve the email address.
|
|
"""
|
|
result = self.zeep.service.checkCookie()
|
|
element = result._raw_elements[0]
|
|
return element.text
|
|
|
|
def is_subscriber(self, email: str, mailing_list: str, function: str = "subscriber") -> bool:
|
|
"""
|
|
Check if the given `email` is a member of type `function` in the `mailing_list`.
|
|
The function parameter is one between subscriber, editor or owner.
|
|
"""
|
|
if function not in ["subscriber", "editor", "owner"]:
|
|
raise ValueError("function of a mailing list member must be subscriber, editor or owner.")
|
|
result = self.zeep.service.amI(mailing_list, function, email)
|
|
element = result._raw_elements[0]
|
|
return element.text == "true"
|
|
|
|
def get_subscribers(self, mailing_list: str, emails_only: bool = True) -> list:
|
|
"""
|
|
Get the list of all subscribers of a list, including the administrators and the editors.
|
|
If emails_only == True, retrieve the list of email addresses only.
|
|
Else, retrieve MLUser object, with the name of the user and the role.
|
|
"""
|
|
if not emails_only:
|
|
users = list()
|
|
elements = self.zeep.service.fullReview(mailing_list)
|
|
for element in elements:
|
|
children = element.getchildren()
|
|
kwargs = dict(mailing_list=mailing_list)
|
|
for child in children:
|
|
tag = child.tag
|
|
if "gecos" in tag:
|
|
kwargs["name"] = child.text
|
|
elif "email" in tag:
|
|
kwargs["email"] = child.text
|
|
elif "isSubscriber" in tag:
|
|
kwargs["subscriber"] = child.text == "true"
|
|
elif "isEditor" in tag:
|
|
kwargs["editor"] = child.text == "true"
|
|
elif "isOwner" in tag:
|
|
kwargs["owner"] = child.text == "true"
|
|
else:
|
|
print("Unknown child tag:", tag)
|
|
user = MLUser(**kwargs)
|
|
users.append(user)
|
|
return users
|
|
return self.zeep.service.review(mailing_list)
|
|
|
|
def lists(self, topic: str, subtopic: str) -> list:
|
|
"""
|
|
Get all the (visible) lists that match the given topic and the given subtopic.
|
|
See TOPICS and SUBTOPICS for valid topics and subtopics.
|
|
"""
|
|
if topic not in TOPICS:
|
|
raise ValueError(f"'{topic}' is not a valid topic.")
|
|
if subtopic and f"{topic}/{subtopic}" not in SUBTOPICS:
|
|
raise ValueError(f"'{topic}/{subtopic}' is not a valid subtopic.")
|
|
result = self.zeep.service.lists(topic, subtopic)._value_1
|
|
if result is None:
|
|
return list()
|
|
lists = list()
|
|
for list_info in result:
|
|
split = list_info.split(';')
|
|
kwargs = dict()
|
|
for data in split:
|
|
key, value = data.split("=", 2)
|
|
if key == "listAddress":
|
|
key = "list_address"
|
|
kwargs[key] = value
|
|
ml = MailingList(**kwargs)
|
|
lists.append(ml)
|
|
return lists
|
|
|
|
def all_lists(self) -> list:
|
|
"""
|
|
Retrieve all lists.
|
|
"""
|
|
elem = self.zeep.service.complexLists()._raw_elements[0]
|
|
lists = list()
|
|
for list_info in elem.getchildren():
|
|
kwargs = dict()
|
|
for child in list_info.getchildren():
|
|
if "listAddress" in child.tag:
|
|
key = "list_address"
|
|
elif "subject" in child.tag:
|
|
key = "subject"
|
|
elif "homepage" in child.tag:
|
|
key = "homepage"
|
|
else:
|
|
raise ValueError(f"Tag {child.tag} is unknown")
|
|
kwargs[key] = child.text
|
|
ml = MailingList(**kwargs)
|
|
lists.append(ml)
|
|
return lists
|
|
|
|
def create_list(self, list_name: str, subject: str, template: str, description: str, topic: str,
|
|
use_custom_template: bool = False, raise_error: bool = True) -> bool:
|
|
"""
|
|
Create a new mailing-list.
|
|
"""
|
|
if topic not in TOPICS and topic not in SUBTOPICS:
|
|
raise ValueError(f"Topic '{topic}' does not exist.")
|
|
if template not in TEMPLATES and not use_custom_template:
|
|
raise ValueError(f"Template '{template}' does not exist.")
|
|
try:
|
|
result = self.zeep.service.createList(list_name, subject, template, description, topic)
|
|
return result._raw_elements[0].text == "true"
|
|
except Fault as e:
|
|
if raise_error:
|
|
raise Fault(f"An unknown error occured while creating the list {list_name}. "
|
|
f"Maybe the list already exists?", e)
|
|
else:
|
|
return False
|
|
|
|
def delete_list(self, list_name: str, raise_error: bool = False) -> bool:
|
|
"""
|
|
Close a mailing list.
|
|
Warning: the list is not deleted in order to keep the history. Please use the web interface to fully
|
|
delete the list.
|
|
Well, the main reason is that the API does not provide a delete method.
|
|
"""
|
|
try:
|
|
result = self.zeep.service.closeList(list_name)
|
|
return result._raw_elements[0].text == "true"
|
|
except Fault as e:
|
|
if raise_error:
|
|
raise Fault(f"An unknown error occured while deleting the list {list_name}. "
|
|
f"Maybe the list did not exist?", e)
|
|
else:
|
|
return False
|
|
|
|
def subscribe(self, email: str, list_address: str, quiet: bool, name: str = "", raise_error: bool = False) -> bool:
|
|
"""
|
|
Subscribe the user with the given email to the given mailing list.
|
|
If the quiet mode is enabled, the user won't receive a notification that they got subscribed.
|
|
"""
|
|
try:
|
|
result = self.zeep.service.add(list_address, email, name, quiet)
|
|
return result._raw_elements[0].text == "true"
|
|
except Fault as e:
|
|
if raise_error:
|
|
raise Fault(f"An unknown error occured while subscribing to the list {list_address}. "
|
|
f"Maybe the user is already a member?", e)
|
|
else:
|
|
return False
|
|
|
|
def unsubscribe(self, email: str, list_address: str, quiet: bool, raise_error: bool = False) -> bool:
|
|
"""
|
|
Subscribe the user with the given email to the given mailing list.
|
|
If the quiet mode is enabled, the user won't receive a notification that they got subscribed.
|
|
"""
|
|
try:
|
|
# del is a reserved keyword
|
|
result = getattr(self.zeep.service, "del")(list_address, email, quiet)
|
|
return result._raw_elements[0].text == "true"
|
|
except Fault as e:
|
|
if raise_error:
|
|
raise Fault(f"An unknown error occured while subscribing to the list {list_address}. "
|
|
f"Maybe the user is already a member?", e)
|
|
else:
|
|
return False
|