import re
import stdnum.ch.esr as esr # type: ignore[import-untyped]
from collections import defaultdict
from functools import cached_property
from datetime import date
from decimal import Decimal
from lxml import etree
from onegov.activity.collections import InvoiceCollection
from onegov.activity.models import Invoice
from onegov.activity.models import InvoiceItem
from onegov.activity.models import InvoiceReference
from onegov.activity.models.invoice_reference import FeriennetSchema
from onegov.user import User
from sqlalchemy import func
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterator
from sqlalchemy.orm import Query, Session
from uuid import UUID
[docs]
DOCUMENT_NS_EX = re.compile(r'.*<Document [^>]+>(.*)')
[docs]
def normalize_xml(xml: str) -> str:
# let's not bother with namespaces at all
return DOCUMENT_NS_EX.sub(r'<Document>\1', xml)
[docs]
class Transaction:
def __init__(self, **kwargs: object) -> None:
for key, value in kwargs.items():
setattr(self, key, value)
[docs]
self.username = '' # Has to be str for comparison used in sort
[docs]
def __repr__(self) -> str:
return repr(self.__dict__)
if TYPE_CHECKING:
# let mypy know that this can have arbitrary attributes
# FIXME: We should just specify all the attributes we put into
# the Transaction...
[docs]
def __getattr__(self, name: str) -> Any: ...
@cached_property
[docs]
def references(self) -> set[str]:
if self.invoice_schema == 'feriennet-v1' or not self.reference:
return set(self.extract_references())
# if possible, don't rely on manual extraction of the reference number.
return {self.reference}
@property
[docs]
def order(self) -> tuple[int, int] | tuple[int, int, str]:
state = self.state
if self.valuta_date:
date = -int(self.valuta_date.strftime('%Y%M%d'))
else:
date = -1
if state == 'success':
return 0, date, self.username
if state == 'paid':
return 1, date, self.username
if state == 'duplicate':
return 2, date, self.username
if state == 'warning':
return 3, date, self.username
return 4, date
@property
[docs]
def state(self) -> str:
if self.paid:
return 'paid'
if self.duplicate:
return 'duplicate'
if self.confidence == 1:
return 'success'
if self.confidence == 0.5:
return 'warning'
return 'unknown'
[docs]
def transaction_entries(root: 'etree._Element') -> 'Iterator[etree._Element]':
""" Yields the transaction entries from the given Camt.053 or Camt.054
xml. This works because for our purposes the entries of those two formats
are identical.
"""
# camt.053
for entry in root.xpath('/Document/BkToCstmrStmt/Stmt/Ntry'):
yield entry
# camt.054
for entry in root.xpath('/Document/BkToCstmrDbtCdtNtfctn/Ntfctn/Ntry'):
yield entry
[docs]
def get_esr(booking_text: str) -> str | None:
"""
Extracts the QR-bill reference number from the given text.
QR-bill reference numbers are usually 26 or 27 characters long but can
be of any length.
The 27-character version includes a check digit at the
end. The 26-character version doesn't include the check digit.
For any other length we don't know if the check digit is included or not.
For example:
input: 'Gutschrift QRR: 27 99029 05678 18860 27295 37059'
output: '269902905678188602729537059'
:returns: The extracted reference number or None if no reference number.
If the extracted reference number is only 26 characters long, the check
digit is appended to the end.
"""
def format_esr(esr_ref: str) -> str:
"""
For DB comparison we need the ESR number with leading zeros but no
spaces.
"""
return esr.compact(esr_ref).zfill(27)
# Pattern for any length ESR numbers
pattern = r'(\d\s*){1,27}'
match = re.search(pattern, booking_text)
if match:
esr_ref = re.sub(r'\s', '', match.group(0))
if esr.is_valid(esr_ref):
return format_esr(esr_ref)
try:
esr.validate(esr_ref)
except esr.InvalidChecksum:
esr_ref = esr_ref[:26]
check = esr.calc_check_digit(esr_ref)
return format_esr(esr_ref + check)
# If no match found, return None
return None
[docs]
def match_iso_20022_to_usernames(
xml: str,
session: 'Session',
period_id: 'UUID',
schema: str,
currency: str = 'CHF'
) -> 'Iterator[Transaction]':
""" Takes an ISO20022 camt.053 file and matches it with the invoice
items in the database.
Raises an error if the given xml cannot be processed.
:return: An iterator of transactions found in the xml file, together with
the matching username and a confidence attribute indicating how
certain the match is (1.0 indicating a sure match, 0.5 a possible match
and 0.0 a non-match).
"""
def items(period_id: 'UUID | None' = None) -> 'Query[InvoiceItem]':
invoices = InvoiceCollection(session, period_id=period_id)
return invoices.query_items().outerjoin(Invoice).outerjoin(User)
# Get all known transaction ids to check what was already paid
q1 = items().with_entities(InvoiceItem.tid, User.username)
q1 = q1.group_by(InvoiceItem.tid, User.username)
q1 = q1.filter(
InvoiceItem.paid == True,
InvoiceItem.source == 'xml'
)
paid_transaction_ids = {i.tid: i.username for i in q1}
# Get a list of reference/username pairs as fallback
q2 = session.query(InvoiceReference).outerjoin(Invoice).outerjoin(User)
username_by_ref = dict(q2.with_entities(
InvoiceReference.reference,
User.username
))
# Get the items matching the given period
q3 = items(period_id=period_id).outerjoin(InvoiceReference).with_entities(
User.username,
func.sum(InvoiceItem.amount).label('amount'),
InvoiceReference.reference,
)
q3 = q3.group_by(User.username, InvoiceReference.reference)
q3 = q3.filter(InvoiceItem.paid == False)
q3 = q3.order_by(User.username)
# Hash the invoices by reference (duplicates possible)
by_ref = defaultdict(list)
# Hash the invoices by amount (duplicates probable)
by_amount = defaultdict(list)
last_username = None
for record in q3:
by_ref[record.reference].append(record)
if last_username != record.username:
by_amount[record.amount].append(record)
last_username = record.username
# go through the transactions, comparing amount and code for a match
transactions = tuple(extract_transactions(xml, schema))
# mark duplicate transactions
seen: dict[str, Transaction] = {}
for t in transactions:
for ref in t.references:
if ref in seen:
t.duplicate = seen[ref].duplicate = True
seen[ref] = t
for t in transactions:
# debit transactions are completely irrelevant for us
if not t.credit:
continue
if t.currency != currency:
yield t
continue
if t.tid and t.tid in paid_transaction_ids:
# what if tid is None?
# it so happened that the records uploaded were marked as paid
# for all records with tid = None since it was in paid_transactions
t.paid = True
t.username = paid_transaction_ids[t.tid]
t.confidence = 1
yield t
continue
if t.credit and t.currency == currency:
amnt_usernames = {i.username for i in by_amount[t.amount]}
ref_usernames = {
i.username for ref in t.references for i in by_ref[ref]}
combined = amnt_usernames & ref_usernames
if len(combined) == 1:
t.username = next(u for u in combined)
t.confidence = 1.0
elif len(ref_usernames) == 1:
t.username = next(u for u in ref_usernames)
t.confidence = 0.5
elif len(amnt_usernames) == 1:
t.username = next(u for u in amnt_usernames)
t.confidence = 0.5
if not t.confidence:
for ref in t.references:
if ref in username_by_ref:
t.username = username_by_ref[ref]
t.confidence = 0.5
yield t