Created
November 14, 2022 13:57
-
-
Save alexander-matz/1cacf8697fc629dcb48d951c92a751ff to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from typing import Sequence, List, NamedTuple, Optional | |
from pprint import pprint | |
from os import getenv | |
import json | |
import random | |
import requests | |
import sys | |
class Person(NamedTuple): | |
name: str | |
email: str | |
class Message(NamedTuple): | |
giver: Person | |
subject: str | |
message: str | |
class Pair(NamedTuple): | |
receiver: Person | |
giver: Person | |
def get_env_config(name: str, default: Optional[str] = None): | |
value = getenv(name) | |
if value is not None: | |
return value | |
if default is not None: | |
return default | |
raise ValueError("Environment variable config {} must be set".format(name)) | |
def load_people(filename: str) -> List[Person]: | |
people = [] | |
for line in open(filename): | |
line = line.strip() | |
if len(line) == 0: | |
continue | |
name, mail = line.split(':', 2) | |
if len(name.strip()) == 0: | |
raise ValueError("No name provided in line '{}'. Line format: '<name>: <mail>'".format(line)) | |
if mail is None or len(mail.strip()) == 0: | |
raise ValueError("No email provided in line '{}'. Line format: '<name>: <mail>'".format(line)) | |
people.append(Person( | |
name = name.strip(), | |
email = mail.strip(), | |
)) | |
return people | |
def send_mails(messages: Sequence[Message], dry: bool = False): | |
endpoint = "https://api.mailjet.com/v3.1/send" | |
account = get_env_config("MAILJET_ACCOUNT") | |
key = get_env_config("MAILJET_KEY") | |
if account is None or key is None: | |
raise ValueError("Can't send mails without MAILJET_ACCOUNT or MAILJET_KEY being set") | |
mailjet_messages = [ | |
{ | |
"From": { | |
"Name": get_env_config("SENDER_NAME", "n/a"), | |
"Email": get_env_config("SENDER_MAIL", "n@a"), | |
}, | |
"To": [ | |
{ | |
"Name": message.giver.name, | |
"Email": message.giver.email, | |
} | |
], | |
"Subject": message.subject, | |
"TextPart": message.message, | |
"CustomID": get_env_config("MAILJET_CUSTOM_ID", "SecretSanta") | |
} for message in messages | |
] | |
if dry: | |
pprint(mailjet_messages) | |
return | |
response = requests.post( | |
endpoint, | |
auth = (account, key), | |
json = {"Messages": mailjet_messages} | |
) | |
# Does not print message text, so no pair leakage | |
pprint(json.loads(response.text)) | |
def generate_pairs(people: Sequence[Person]) -> List[Pair]: | |
generate_candidate = lambda: list(zip(people, random.sample(people, len(people)))) | |
while True: | |
candidate = generate_candidate() | |
has_self_ref = any([pair[0] == pair[1] for pair in candidate]) | |
if has_self_ref: | |
continue | |
return [Pair(receiver = pair[0], giver = pair[1]) for pair in candidate] | |
def generate_messages(pairs: Sequence[Pair]) -> List[Message]: | |
template = """Servus {giver}! | |
Dein Wichtel ist {receiver}, viel Spaß! | |
Liebe Grüße, | |
das Wichtel-Komitee""" | |
messages = [ | |
Message( | |
giver = pair.giver, | |
subject = "Dein Wichtel", | |
message = template.format( | |
giver = pair.giver.name, | |
receiver = pair.receiver.name, | |
) | |
) for pair in pairs | |
] | |
return messages | |
def main(args): | |
if len(args) == 1: | |
print('usage: MAILJET_ACCOUNT=... MAILJET_KEY=... santa.py people-file') | |
return 1 | |
people_file = args[1] | |
people = load_people(people_file) | |
pairs = generate_pairs(people) | |
messages = generate_messages(pairs) | |
send_mails(messages) | |
return 0 | |
def test(): | |
test_people = [ | |
Person(name = "a", email = "amail"), | |
Person(name = "b", email = "bmail"), | |
Person(name = "c", email = "cmail"), | |
Person(name = "d", email = "dmail"), | |
Person(name = "e", email = "email"), | |
Person(name = "f", email = "fmail"), | |
] | |
pprint(generate_messages(generate_pairs(test_people))) | |
sys.exit(main(sys.argv)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment