295 lines
13 KiB
Python
295 lines
13 KiB
Python
#!/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 newsletter(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()
|
|
template = env.jinja2_env.get_template(env.args.template)
|
|
for member in csvreader:
|
|
email_address = member.get('email')
|
|
if not email_address:
|
|
continue
|
|
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"] = env.args.subject
|
|
msg["Message-Id"] = member.get("message_id") or email.utils.make_msgid(domain="verein.ccchb.de")
|
|
member['message_id'] = msg["Message-Id"]
|
|
template_args = {
|
|
"name": member.get("name"),
|
|
"email": email_address,
|
|
"direct_debit": member.get("sepa_mandate"),
|
|
"fee_old": decimal.Decimal(member.get("monthly_fee") or '0.0'),
|
|
"member": member,
|
|
}
|
|
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_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)
|
|
fee_invoice_parser.set_defaults(requires_tryton=True)
|
|
|
|
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)
|
|
sepa_mandate_parser.set_defaults(requires_tryton=True)
|
|
|
|
newsletter_parser = subparsers.add_parser('newsletter')
|
|
newsletter_parser.add_argument('-o', '--output', type=argparse.FileType("w"), default='-')
|
|
newsletter_parser.add_argument('-F', '--csv_fields', type=lambda f: f.split(','), help='csv fields')
|
|
newsletter_parser.add_argument("-s", "--subject", default="")
|
|
newsletter_parser.add_argument('template')
|
|
newsletter_parser.add_argument('csvfile', type=argparse.FileType('r'), default='-', nargs='?')
|
|
newsletter_parser.set_defaults(func=newsletter)
|
|
newsletter_parser.set_defaults(requires_tryton=False)
|
|
|
|
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 and args.requires_tryton:
|
|
args.uri = input("URI for Tryton: ")
|
|
|
|
if not args.username and args.requires_tryton:
|
|
args.username = input("Username for Tryton: ")
|
|
|
|
if not args.password and args.requires_tryton:
|
|
args.password = getpass.getpass(f"Password for Tryton user `{args.username}': ")
|
|
|
|
if args.requires_tryton:
|
|
with triad.client.Session.start(args.uri, args.username, args.password) as session:
|
|
env.session = session
|
|
args.func(env)
|
|
else:
|
|
args.func(env)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|