vereinsgwrschtl/tryton-scripts/mailings.py

#!/bin/env python3

import argparse
import csv
import datetime
import decimal
import email.message
import functools
import getpass
import operator
import pathlib
import subprocess
import tempfile
import tomllib

import babel.numbers
import dateutil.relativedelta
import jinja2

import triad.client

def strip_iban(iban):
    iban = [c for c in iban if not c.isspace()]
    return "".join(iban[0:3] + ["X" for c in iban[3:-3]] + iban[-3:])

def format_currency(amount, currency='EUR', locale='de_DE'):
    return babel.numbers.format_currency(amount, currency, locale=locale)

def format_quantity(amount, locale='de_DE'):
    return babel.numbers.format_decimal(amount, locale=locale)

class Env(object):
    jinja2_env: jinja2.Environment
    args: argparse.Namespace
    session: triad.client.Session

def deduce_payments(client, invoice):
    total_amount = invoice.total_amount
    remainder = total_amount
    for line in invoice.payment_term.lines:
        if line.type == 'fixed':
            amount = line.amount
        elif line.type == 'percent':
            amount = remainder * line.ratio
        elif line.type =='percent_on_total':
            amount = total_amount * line.ratio
        elif line.type == 'remainder':
            amount = remainder
        else:
            continue
        relativedeltas = (
            dateutil.relativedelta.relativedelta(
                day=relativedelta.day,
                month=int(relativedelta.month.index) if relativedelta.month else None,
                days=relativedelta.days,
                weeks=relativedelta.weeks,
                months=relativedelta.months,
                weekday=int(relativedelta.weekday.index) if relativedelta.weekday else None
            ) for relativedelta in line.relativedeltas
        )
        date = functools.reduce(operator.add, relativedeltas, invoice.payment_term_date)
        remainder -= amount
        yield (date, amount)

def fee_invoice(env: Env):
    csvreader = csv.DictReader(env.args.csvfile, fieldnames=env.args.csv_fields)
    csvwriter = csv.DictWriter(env.args.output, fieldnames=csvreader.fieldnames + (["message_id"] if 'message_id' not in csvreader.fieldnames else []))
    csvwriter.writeheader()
    invoicing_client = triad.client.InvoicingClient.from_session(env.session)
    template = env.jinja2_env.get_template("beitrag.eml")
    for member in csvreader:
        try:
            invoice = invoicing_client.Invoice.find([(env.args.lookup_key, "=", str(member['invoice_id'])), ("type", "=", "out")])[0]
        except IndexError:
            continue
        year = invoice.invoice_date or env.args.year
        months = [datetime.date(year.year, month, 1) for month in range(1, 13)]
        email_address = member.get('email') or invoice.party.email
        try:
            sepa_mandate = invoice.party.reception_direct_debits[0]
        except IndexError:
            sepa_mandate = None
        else:
            sepa_mandate = sepa_mandate.sepa_mandate
        if invoice.state in {"draft", "validated"} and sepa_mandate is not None:
            payments = list(deduce_payments(invoicing_client, invoice))
        else:
            payments = [(move.maturity_date, move.amount) for move in invoice.lines_to_pay if not move.reconciliation and move.payable_receivable_balance != 0]
        template_args = {
            'monthly_fee': decimal.Decimal(member.get('monthly_fee') or '0.0'),
            "yearly_fee": decimal.Decimal(member.get('yearly_fee') or '0.0'),
            'invoice': invoice,
            "payments": payments,
            'sepa_mandate': sepa_mandate,
            'payments': payments,
            'sender': env.args.email_sender,
            "year": year,
            "fee_months": [datetime.date(year.year, month, 1) for month in range(1, 13)]
        }
        member.setdefault('message_id', email.utils.make_msgid(domain='verein.ccchb.de'))
        msg = email.message.EmailMessage()
        msg["From"] = env.args.email_from
        msg["To"] = email_address
        msg["Date"] = env.args.email_date or datetime.datetime.now()
        msg["Subject"] = f"Beitragsrechnung {year.year} ({invoice.number})" if invoice.number else f"Vorläufige Beitragsrechnung {year.year}"
        msg["Message-Id"] = member['message_id']
        msg.set_content(template.render(**template_args))
        with (env.args.output_dir / member['message_id']).open("wb") as mailfile:
            mailfile.write(bytes(msg))
        csvwriter.writerow(member)

def sepa_mandate(env: Env):
    csvreader = csv.DictReader(env.args.csvfile, fieldnames=env.args.csv_fields)
    csvwriter = csv.DictWriter(env.args.output, fieldnames=csvreader.fieldnames + ["message_id"])
    csvwriter.writeheader()
    client = triad.client.BaseClient.from_session(env.session)
    template = env.jinja2_env.get_template("sepa_mandat.eml")
    for mandate in csvreader:
        email_address= mandate.get('email')
        if 'sepa_mandate_ref' in mandate:
            sepa_mandate = client.get_model('account.payment.sepa.mandate').find([('identification', '=', mandate['sepa_mandate_ref'])])
            party = sepa_mandate.party
        elif 'party_id' in mandate:
            party = client.get_model('party.party').find([('code', '=', mandate['party_id'])])[0]
            email_address = email_address or party.email
            sepa_mandate = party.reception_direct_debits[0].sepa_mandate
        else:
            continue
        template_args = {
            "party": party,
            "sepa_mandate": sepa_mandate
        }
        msg = email.message.EmailMessage()
        msg["From"] = env.args.email_from
        msg["To"] = email_address
        msg["Date"] = env.args.email_date or datetime.datetime.now()
        msg["Subject"] = f"SEPA-Lastschriftmandat MREF: {sepa_mandate.identification}"
        msg["Message-Id"] = email.utils.make_msgid(domain="verein.ccchb.de")
        mandate['message_id'] = msg["Message-Id"]
        msg.set_content(template.render(**template_args))
        if env.args.pdf:
            msg.add_attachment(sepa_mandate_build_pdf(env, sepa_mandate), 'application', 'pdf', filename='SEPA-Lastschriftmandat.pdf')
        with (env.args.output_dir / mandate['message_id']).open("wb") as mailfile:
            mailfile.write(bytes(msg))
        csvwriter.writerow(mandate)

def sepa_mandate_build_pdf(env, sepa_mandate):
    with tempfile.TemporaryDirectory() as tempdir:
        tempdir_path = pathlib.Path(tempdir)
        jobname = sepa_mandate.identification
        with (tempdir_path / (jobname + "-settings.tex")).open("w") as fh:
            fh.write(f'''
\\csdef{{sepa-m001}}{{{sepa_mandate.identification}}}
\\csdef{{sepa-m006}}{{Wiederkehrende Zahlung}}
\\csdef{{sepa-p001}}{{{sepa_mandate.party.name}}}
\\csdef{{sepa-d001}}{{{sepa_mandate.account_number.number}}}
\\csdef{{sepa-m008}}{{{sepa_mandate.signature_date}}}
\\csdef{{sepa-m009}}{{\\emph{{entfällt}}}}
\\csdef{{sepa-d002a}}{{}}
\\csdef{{sepa-d002b}}{{}}
\\csdef{{sepa-p005}}{{}}
\\csdef{{sepa-remarks}}{{Dieses Mandat ist eine Abschrift des dem Zahlungsempfänger vorliegenden, originalen SEPA-Lastschriftmandats.\\\\
Erstellt am {datetime.date.today()}}}
''')
        compile_latex(jobname, env.args.latex_template, tempdir)
        with (tempdir_path / (jobname + ".pdf")).open("rb") as fh:
            return fh.read()

def compile_latex(jobname, texfile, directory, debug=True):
    latexmk_args = ["latexmk", "-lualatex", str(texfile), f"-jobname={jobname}"]
    if not debug:
        latexmk_args.append("-silent")
    subprocess.run(latexmk_args, cwd=directory)

def main():
    parser = argparse.ArgumentParser()

    parser.add_argument("-U", "--uri", help="Tryton URI")
    parser.add_argument("-u", "--username", help="Tryton username")
    parser.add_argument("-p", "--password", help="Tryton password")

    parser.add_argument("-O", "--output-dir", help="email output directory", default=".", type=pathlib.Path)
    parser.add_argument("-T", "--template-dir", help="email template directory", default="mail_templates")

    parser.add_argument("-c", "--config", type=argparse.FileType("rb"))

    # Email arguments
    parser.add_argument('--email-sender', help='email signature sender')
    parser.add_argument("--email-from", help="email from header")
    parser.add_argument("--email-date", help="email date header")
    
    subparsers = parser.add_subparsers()

    fee_invoice_parser = subparsers.add_parser("fee-invoice")
    fee_invoice_parser.add_argument("-o", "--output", type=argparse.FileType("w"), default='-')
    fee_invoice_parser.add_argument("-F", "--csv-fields", type=lambda f: f.split(","), help="csv fields")
    fee_invoice_parser.add_argument("-y", "--year", type=lambda y: datetime.date(int(y), 1, 1))
    fee_invoice_parser.add_argument("--months", help="Beitragsmonate", type=lambda m: m.split(","), default="1,2,3,4,5,6,7,8,9,10,11,12")
    fee_invoice_parser.add_argument("-k", "--lookup-key", choices={'number', 'id'}, default='id', help='lookup key for invoice')
    fee_invoice_parser.add_argument("csvfile", type=argparse.FileType("r"), default="-", nargs="?")
    fee_invoice_parser.set_defaults(func=fee_invoice)

    sepa_mandate_parser = subparsers.add_parser('sepa-mandate')
    sepa_mandate_parser.add_argument('-o', '--output', type=argparse.FileType("w"), default='-')
    sepa_mandate_parser.add_argument('-F', '--csv-fields', type=lambda f: f.split(','), help='csv fields')
    sepa_mandate_parser.add_argument('--pdf', action='store_true', help='build pdf attachment', default=False)
    sepa_mandate_parser.add_argument('--no-pdf', action='store_false', dest='pdf', help='don\'t build pdf attachment')
    sepa_mandate_parser.add_argument('--latex-template', help='latex template in $TEXINPUTS')
    sepa_mandate_parser.add_argument('csvfile', type=argparse.FileType('r'), default='-', nargs='?')
    sepa_mandate_parser.set_defaults(func=sepa_mandate)

    args = parser.parse_args()

    env = Env()
    env.jinja2_env = jinja2.Environment(
        loader=jinja2.FileSystemLoader(args.template_dir)
    )
    env.jinja2_env.filters["strip_iban"] = strip_iban
    env.jinja2_env.filters["format_currency"] = format_currency
    env.jinja2_env.filters["format_quantity"] = format_quantity
    env.args = args

    if args.config:
        env.config = tomllib.load(args.config)
        args.config.close()
        if "tryton" in env.config:
            if "uri" in env.config["tryton"]:
                env.args.uri = env.config["tryton"]["uri"]
            if "username" in env.config["tryton"]:
                env.args.username = env.config["tryton"]["username"]
            if "password" in env.config["tryton"]:
                env.args.password = env.config["tryton"]["password"]
        if "email" in env.config:
            if "sender" in env.config["email"]:
                env.args.email_sender = env.config["email"]["sender"]
            if "from" in env.config["email"]:
                env.args.email_from = env.config["email"]["from"]

    if not args.uri:
        args.uri = input("URI for Tryton: ")

    if not args.username:
        args.username = input("Username for Tryton: ")

    if not args.password:
        args.password = getpass.getpass(f"Password for Tryton user `{args.username}': ")

    with triad.client.Session.start(args.uri, args.username, args.password) as session:
        env.session = session
        args.func(env)

main()