247 lines
8.2 KiB
Python
Executable File
247 lines
8.2 KiB
Python
Executable File
#!/usr/bin/python3
|
|
import mailbox
|
|
import email.header
|
|
import email.utils
|
|
import yaml
|
|
import datetime
|
|
import sys, os
|
|
import argparse
|
|
import quopri
|
|
import json
|
|
import subprocess
|
|
import re
|
|
from pprint import pprint
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
import pypandoc
|
|
import jinja2
|
|
|
|
import util
|
|
|
|
class Top:
|
|
def __init__(self, title=None, sender=None, body=None, protostub=None, message=None):
|
|
if message:
|
|
subject = message["Subject"]
|
|
needs_stripping = subject[:6] == "[top] "
|
|
self.title = util.deEmojify(util.decode_header(subject[6:] if needs_stripping else subject))
|
|
real_name, address = email.utils.parseaddr(message["From"])
|
|
real_name = util.decode_header(real_name)
|
|
self.sender = real_name or address
|
|
payload = get_body_text(message)
|
|
self.body = str(payload.rpartition("\n--")[0] if "\n--" in payload else payload)
|
|
elif title:
|
|
self.title = title
|
|
self.sender = sender
|
|
self.body = body
|
|
self.protostub = protostub
|
|
else:
|
|
raise ValueError("One of title or message is needed")
|
|
|
|
if self.body is None:
|
|
self.body = ""
|
|
|
|
def __repr__(self):
|
|
return "<TOP "+self.title+">"
|
|
|
|
def wiki2latex(intext):
|
|
intext = intext.replace(":\n", ":\n\n")
|
|
return pypandoc.convert_text(intext, 'latex', format='md')
|
|
|
|
def j2replace(intext):
|
|
return j2env.from_string(intext).render(context)
|
|
|
|
def date(indate):
|
|
return indate.strftime("%d.%m.%Y")
|
|
|
|
def time(intime):
|
|
return intime.strftime("%H:%M")
|
|
|
|
def weekday(indate):
|
|
return util.WEEKDAYS[indate.weekday()]
|
|
|
|
def prototop(top):
|
|
result = ""
|
|
|
|
if "protostub" in dir(top) and top.protostub:
|
|
result = j2env.from_string(top.protostub).render(context, top=top)
|
|
elif top.body:
|
|
result = j2env.from_string(top.body).render(context)
|
|
|
|
for search,replace in config["protoreplace"].items():
|
|
result = result.replace(search, replace)
|
|
|
|
return result
|
|
|
|
|
|
def conf2top(top):
|
|
sender = None
|
|
body = None
|
|
protostub = None
|
|
try:
|
|
sender = top["sender"]
|
|
except KeyError:
|
|
pass
|
|
try:
|
|
body = top["body"]
|
|
except KeyError:
|
|
pass
|
|
try:
|
|
protostub = top["protostub"]
|
|
except KeyError:
|
|
pass
|
|
|
|
if "file" in top:
|
|
try:
|
|
body = open(top["file"]).read()
|
|
except OSError as e:
|
|
print("Warning: Error opening", top["file"], file=sys.stderr)
|
|
|
|
if "command" in top:
|
|
try:
|
|
body = subprocess.run(top["command"], shell=True, text=True, capture_output=True, check=True).stdout
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Warning: Command for '{top['title']}' returned non-zero exit code, output not used")
|
|
|
|
if "proto_command" in top:
|
|
try:
|
|
protostub = subprocess.run(top["proto_command"], shell=True, text=True, capture_output=True, check=True).stdout
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Warning: Protocommand for '{top['title']}' returned non-zero exit code, output not used")
|
|
|
|
return Top(top["title"], sender, body, protostub)
|
|
|
|
|
|
def main(rawargs=None):
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--config", "-c", default=util.CONFIG_FILE)
|
|
mode = parser.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--invite", action="store_true")
|
|
mode.add_argument("--mm-invite", action="store_true")
|
|
mode.add_argument("--presentation", action="store_true")
|
|
mode.add_argument("--protocol", action="store_true")
|
|
parser.add_argument("--debug", action="store_true", help=argparse.SUPPRESS)
|
|
parser.add_argument("--print-config", action="store_true", help=argparse.SUPPRESS)
|
|
parser.add_argument("--write-mbox", action="store_true")
|
|
parser.add_argument("--send-mm", action="store_true")
|
|
parser.add_argument("--send-mail", action="store_true")
|
|
parser.add_argument("--save", action="store_true")
|
|
parser.add_argument("--time")
|
|
parser.add_argument("--date")
|
|
args = parser.parse_args(rawargs)
|
|
|
|
config = util.get_config(util.get_normalized_config_path(args.config))
|
|
|
|
if args.print_config:
|
|
pprint(config)
|
|
sys.exit(0)
|
|
|
|
if args.invite:
|
|
template_file = config["invite_template_file"]
|
|
elif args.presentation:
|
|
template_file = config["presentation_template_file"]
|
|
elif args.mm_invite:
|
|
template_file = config["mminvite_template_file"]
|
|
elif args.protocol:
|
|
template_file = config["protocol_template_file"]
|
|
else:
|
|
raise Exception("Should never happen")
|
|
|
|
global j2env
|
|
j2env = jinja2.Environment()
|
|
j2env.filters["wiki2latex"] = wiki2latex
|
|
j2env.filters["j2replace"] = j2replace
|
|
j2env.filters["date"] = date
|
|
j2env.filters["time"] = time
|
|
j2env.filters["weekday"] = weekday
|
|
j2env.filters["prototop"] = prototop
|
|
template = j2env.from_string(open(template_file).read())
|
|
|
|
mbox = mailbox.mbox(config["top_mbox_file"])
|
|
|
|
current_date = util.next_weekday(datetime.date.today(), config["default_weekday"])
|
|
if args.date:
|
|
current_date = datetime.date.fromisoformat(args.date)
|
|
#next_date = current_date + datetime.timedelta(days=7)
|
|
next_date = util.next_weekday(current_date, config["default_weekday"])
|
|
last_date = util.last_weekday(current_date, config["default_weekday"])
|
|
|
|
actual_time = datetime.time.fromisoformat(args.time or config["default_time"])
|
|
|
|
pre_tops = []
|
|
post_tops = []
|
|
|
|
for top in config["pre_tops"]:
|
|
pre_tops.append(conf2top(top))
|
|
|
|
for top in config["post_tops"]:
|
|
post_tops.append(conf2top(top))
|
|
|
|
email_tops = []
|
|
|
|
for message in mbox:
|
|
if args.debug:
|
|
print(message.get_payload())
|
|
top = Top(message=message)
|
|
email_tops.append(top)
|
|
|
|
to = pre_tops + email_tops + post_tops
|
|
|
|
global context
|
|
context = {"to": to,
|
|
"redeleitung": config["redeleitung"],
|
|
"protokoll": config["protokoll"],
|
|
"date": current_date,
|
|
"time": actual_time,
|
|
"place": config["place"],
|
|
"next_date": next_date,
|
|
"last_date": last_date,
|
|
"meeting_link": config["meeting_link"],
|
|
"email_tops": email_tops,
|
|
"WEEKDAYS": util.WEEKDAYS}
|
|
|
|
if args.debug:
|
|
for top in to:
|
|
pprint(top.__dict__)
|
|
pprint(context)
|
|
elif args.save:
|
|
if args.invite:
|
|
filename = Path(config["invite_save_path"]) / Path("invite_"+datetime.date.today().isoformat()+".txt")
|
|
elif args.presentation:
|
|
filename = Path(config["presentation_save_path"]) / Path("presentation_"+datetime.date.today().isoformat()+".tex")
|
|
elif args.protocol:
|
|
filename = Path(config["protocol_save_path"]) / Path(datetime.date.today().isoformat())
|
|
else:
|
|
raise Exception("Should never happen")
|
|
|
|
with open(filename, "w") as file:
|
|
file.write(template.render(context))
|
|
|
|
elif args.send_mail:
|
|
msg = email.message.EmailMessage()
|
|
msg.set_content(template.render(context))
|
|
msg["Subject"] = j2env.from_string(config["invite_subject"]).render(context)
|
|
msg["From"] = email.utils.formataddr((config["redeleitung"]["name"], config["redeleitung"]["email"]))
|
|
msg["To"] = config["invite_mail"]
|
|
subprocess.run([*config["sendmail"], config["invite_mail"]], input=str(msg), text=True, check=True)
|
|
|
|
elif args.write_mbox:
|
|
msg = email.message.EmailMessage()
|
|
msg.set_content(template.render(context))
|
|
msg["Subject"] = j2env.from_string(config["invite_subject"]).render(context)
|
|
msg["From"] = email.utils.formataddr((config["redeleitung"]["name"], config["redeleitung"]["email"]))
|
|
msg["To"] = config["invite_mail"]
|
|
mbox = mailbox.mbox(config["mbox_out"])
|
|
mbox.add(msg)
|
|
mbox.close()
|
|
print(mbox)
|
|
elif args.send_mm:
|
|
headers = {'Content-Type': 'application/json',}
|
|
values = json.dumps({ "text": template.render(context), "username": "test"})
|
|
response = requests.post(config["mm_url"], headers=headers, data=values)
|
|
else:
|
|
print(template.render(context))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|