Skript für das Anlegen der Beitragsrechnungen

triad/pyproject.toml

@@ -0,0 +1,14 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "triad"
+version = "0.0.1"
+authors = [
+    { name="Fritz Grimpen", email="fritz@ccchb.de" }
+]
+description = "A useful frontend for the Tryton API client"
+readme = "README.md"
+requires_python = ">=3.8"
+

triad/triad/__init__.py

@@ -0,0 +1 @@
+__all__ = ['client']

triad/triad/client.py

@@ -0,0 +1,85 @@
+import codecs
+import contextlib
+
+import xmlrpc
+import proteus
+
+class BaseClient(object):
+    def __init__(self, config, session=None):
+        self.config = config
+        self.session = session
+    
+    def get_model(self, model_name, cls_name=None):
+        if cls_name is None:
+            cls_name = model_name.split('.')[-1].replace('_', ' ').title().replace(' ', '')
+        model_cls = proteus.Model.get(model_name, config=self.config)
+        # model_cls.__name__ = cls_name
+        # model_cls.__qualname__ = self.__class__.__qualname__ + "." + cls_name
+        # model_cls.__module__ = self.__class__.__module__
+        return model_cls
+
+    @classmethod
+    def from_session(cls, session):
+        config = proteus.config.XmlrpcConfig(session.url, headers=session.auth_headers())
+        return cls(config, session=session)
+    
+    def __repr__(self):
+        return f'<BaseClient {self.config!r}>'
+
+class Session(object):
+    def __init__(self, url, username, token):
+        self.url = url
+        self.username = username
+        self.token = token
+    
+    def get_client(self, client_cls=BaseClient):
+        return client_cls.from_session(self)
+    
+    def auth_headers(self):
+        auth_token = ':'.join(str(field) for field in [self.username] + self.token)
+        auth_token = codecs.encode(auth_token.encode('utf-8'), 'base64').decode('utf-8').replace('\n', '')
+        return [('Authorization', "Session " + auth_token)]
+
+    @classmethod
+    @contextlib.contextmanager
+    def start(cls, url, username, password=None):
+        server = xmlrpc.client.ServerProxy(url)
+        token = server.common.db.login(username, {'password': password})
+        session = cls(url, username, token)
+        yield session
+        session.get_client().config.server.common.db.logout()
+
+class AccountingClient(BaseClient):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.Account = self.get_model('account.account', cls_name='Account')
+        self.AccountType = self.get_model('account.account.type', cls_name='AccountType')
+        self.Journal = self.get_model('account.journal', cls_name='Journal')
+        self.Move = self.get_model('account.move', cls_name='Move')
+        self.MoveLine = self.get_model('account.move.line', cls_name='MoveLine')
+    
+    def get_account_by_code(self, code):
+        try:
+            return self.Account.find([('code', '=', str(code))])[0]
+        except IndexError:
+            raise KeyError(code)
+
+class InvoicingClient(BaseClient):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.Invoice = self.get_model('account.invoice', cls_name='Invoice')
+        self.InvoiceLine = self.get_model('account.invoice.line', cls_name='InvoiceLine')
+        self.PaymentTerm = self.get_model('account.invoice.payment_term', cls_name='PaymentTerm')
+        self.PaymentTermLine = self.get_model('account.invoice.payment_term.line', cls_name='PaymentTermLine')
+
+class PartyClient(BaseClient):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.Party = self.get_model('party.party', cls_name='Party')
+        self.Category = self.get_model('party.category', cls_name='Category')
+    
+    def get_party_by_code(self, code):
+        try:
+            return self.Party.find([('code', '=', str(code))])[0]
+        except IndexError:
+            raise KeyError(code)

tryton-scripts/beitragsrechnungen.py

@@ -0,0 +1,128 @@
+#!/bin/env python3
+
+import argparse
+import csv
+import datetime
+import decimal
+import getpass
+import sys
+
+import triad.client
+
+argparser = argparse.ArgumentParser(description='Beitragsrechnungen für ein Jahr erzeugen')
+argparser.add_argument('-u', '--username', help='Tryton-Benutzername', default='fgrimpen')
+argparser.add_argument('-p', '--password', help='Tryton-Passwort')
+argparser.add_argument('-U', '--url', help='Tryton-URL', default='https://kasse.verein.ccchb.de/tryton/')
+argparser.add_argument('--reference-date', help='Referenzdatum für Zahlungsbedingung', type=datetime.date.fromisoformat)
+argparser.add_argument('--claims-account', help='Forderungskonto', default='12001')
+argparser.add_argument('--fee-account', help='Beitragskonto', default='40000')
+argparser.add_argument('--default-payment-term', default='1m')
+argparser.add_argument('--csv-format', type=lambda x: x.split(",") if isinstance(x, str) else list(x), default='party_id,monthly_fee,payment_term,yearly_fee', help='CSV format specification')
+argparser.add_argument('-n', '--dry-run', action='store_true')
+argparser.add_argument('year', help='Beitragsjahr', type=lambda s: datetime.date(int(s), 1, 1))
+argparser.add_argument('csvfile', nargs='?', type=argparse.FileType('r'), default='-')
+
+args = argparser.parse_args()
+
+if not args.password:
+    args.password = getpass.getpass(f"Password for {args.username}@{args.url}: ")
+
+def calculate_reference_date(date):
+    today = datetime.date.today()
+    if today >= date:
+        date = today
+    while date.day >= 10:
+        date += datetime.timedelta(days=1)
+    while date.day != 15:
+        date += datetime.timedelta(days=1)
+    return date
+
+if not args.reference_date:
+    args.reference_date = calculate_reference_date(args.year)
+
+csvreader = csv.DictReader(args.csvfile, fieldnames=args.csv_format)
+
+PAYMENT_TERMS_MAPPING = [
+    ({'1m'}, 'Beitrag monatlich/jährlich'),
+    ({'3m'}, 'Beitrag quartalsweise/jährlich'),
+    ({'6m'}, 'Beitrag halbjährlich/jährlich'),
+    ({'1y', '12m'}, 'Beitrag jährlich/jährlich')
+]
+
+def load_payment_terms(session, invoicing_client):
+    ret = {}
+    for keys, payment_term_name in PAYMENT_TERMS_MAPPING:
+        payment_term = invoicing_client.PaymentTerm.find([('name', '=', payment_term_name)])[0]
+        ret.update({key: payment_term for key in keys})
+    return ret
+
+
+def create_invoice(party, payment_term, monthly_fee, yearly_fee, claims_account, fee_account, invoicing_client, reference_date, year):
+    invoice = invoicing_client.Invoice()
+    invoice.type = 'out'
+    invoice.party = party
+    invoice.description = f"Beiträge {year.year}"
+    invoice.account = claims_account
+    invoice.payment_term = payment_term
+    invoice.payment_term_date = reference_date
+    if monthly_fee:
+        invoice_line = invoicing_client.InvoiceLine()
+        invoice_line.description = f"Monatsbeiträge Januar--Dezember {year.year}"
+        invoice_line.account = fee_account
+        invoice_line.quantity = decimal.Decimal(12) # Months
+        invoice_line.unit_price = decimal.Decimal(monthly_fee)
+        invoice.lines.append(invoice_line)
+    if yearly_fee:
+        invoice_line = invoicing_client.InvoiceLine()
+        invoice_line.description = f"Jahresbeitrag {year.year}"
+        invoice_line.account = fee_account
+        invoice_line.quantity = decimal.Decimal(1)
+        invoice_line.unit_price = decimal.Decimal(yearly_fee)
+        invoice.lines.append(invoice_line)
+    invoice.state = 'draft'
+    return invoice
+
+with triad.client.Session.start(args.url, args.username, args.password) as session:
+    party_client = triad.client.PartyClient.from_session(session)
+    invoicing_client = triad.client.InvoicingClient.from_session(session)
+    accounting_client = triad.client.AccountingClient.from_session(session)
+
+    claims_account = accounting_client.get_account_by_code(args.claims_account)
+    print(f"Claims account: {claims_account.code}  {claims_account.name}", file=sys.stderr)
+    fee_account = accounting_client.get_account_by_code(args.fee_account)
+    print(f'Fee account: {fee_account.code}  {fee_account.name}', file=sys.stderr)
+    payment_terms = load_payment_terms(invoicing_client, invoicing_client)
+    for key, payment_term in payment_terms.items():
+        print(f"Payment term {key}: {payment_term.name}", file=sys.stderr)
+    print(f'Fee year: {args.year}', file=sys.stderr)
+    print(f'Reference date: {args.reference_date}', file=sys.stderr)
+    for member in csvreader:
+        party = party_client.get_party_by_code(member['party_id'])
+        print(f"Party: {party.code}  {party.name}", file=sys.stderr)
+        try:
+            payment_term = payment_terms[member['payment_term']]
+        except KeyError:
+            payment_term = payment_terms[args.default_payment_term]
+        print(f"\tPayment term: {payment_term.name}", file=sys.stderr)
+        try:
+            monthly_fee = decimal.Decimal(member['monthly_fee'])
+        except:
+            print(f"\tInvalid monthly fee: {member['monthly_fee']}", file=sys.stderr)
+            monthly_fee = None
+        else:
+            print(f"\tMonthly fee: {monthly_fee}", file=sys.stderr)
+        try:
+            yearly_fee = decimal.Decimal(member['yearly_fee'])
+        except:
+            print(f'\tInvalid yearly fee: {member['yearly_fee']}', file=sys.stderr)
+            yearly_fee = None
+        else:
+            print(f'\tYearly fee: {yearly_fee}', file=sys.stderr)
+        invoice = create_invoice(party, payment_term, monthly_fee, yearly_fee, claims_account, fee_account, invoicing_client, args.reference_date, args.year)
+        if not args.dry_run:
+            invoice.save()
+        print(f'\tInvoice: {invoice.id}  {invoice.description}', file=sys.stderr)
+        for line in invoice.lines:
+            print(f'\tLine: {line.description}  {line.quantity} * {line.unit_price}', file=sys.stderr)
+        print(f'\tYearly fee: {invoice.total_amount}', file=sys.stderr)
+        print(f"{member['party_id']},{invoice.id},{invoice.total_amount},{payment_term.name}")