Skip to content

CAMT: join split Ustrd lines with a space, find names without Pty wrapper #226

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bjorn wants to merge 1 commit into
base: master
Choose a base branch
from
+150 −11
Open

CAMT: join split Ustrd lines with a space, find names without Pty wrapper #226

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 49 additions & 11 deletions fints/camt_parser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2897,6 +2897,32 @@ def _modify_key(clean_mnems, translate=True):
return clean_mnems


def _join_repeated(existing, addition):
"""Join the text of a repeated element onto what came before.

Banks split remittance information into several Ustrd elements, one
field per line. ING's own CSV export renders those with a space in
between ("NR XXXX 1234 MUSTERSTADT DE KAUFUMSATZ 12.01 30.95 ..."),
which is what a reader expects, so a space is inserted unless one side
already ends or starts with whitespace, as with chunks split mid-field
that carry a trailing blank ("CRED: " + "DE00ZZZ123456789").
"""
existing = existing or ''
addition = addition or ''
if not existing or not addition or existing[-1].isspace() or addition[0].isspace():
return existing + addition
return existing + ' ' + addition


def _first(record, *keys):
"""The first of several keys that is present, or None."""
for key in keys:
value = record.get(key)
if value is not None:
return value
return None


def _parse_element(element, parent_name='', translate=True):
data_dict = {}
for child in element:
Expand All @@ -2907,7 +2933,7 @@ def _parse_element(element, parent_name='', translate=True):
child_name,
translate=translate))
elif child_name in data_dict:
data_dict[child_name] += child.text
data_dict[child_name] = _join_repeated(data_dict[child_name], child.text)
else:
data_dict[child_name] = child.text

Expand All @@ -2925,26 +2951,38 @@ def _add_backwards_compat_keys(record, currency):
record["applicant_iban"] = record.get(
"EntryDetails.TransactionDetails.RelatedParties.DebtorAccount.Identification.IBAN"
)
record["applicant_name"] = record.get(
"EntryDetails.TransactionDetails.RelatedParties.Debtor.Party.Name"
record["applicant_name"] = _first(
record,
"EntryDetails.TransactionDetails.RelatedParties.Debtor.Party.Name",
"EntryDetails.TransactionDetails.RelatedParties.Debtor.Name",
)
record["recipient_name"] = record.get(
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Party.Name"
record["recipient_name"] = _first(
record,
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Party.Name",
# camt.052.001.02 (ING) has no Pty wrapper around the name.
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Name",
)
record["status"] = "C"
else:
record["amount"] = Amount(-amt, currency)
record["applicant_creditor_id"] = record.get(
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Party.Identification.PrivateIdentification.Other.Identification"
record["applicant_creditor_id"] = _first(
record,
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Party.Identification.PrivateIdentification.Other.Identification",
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Identification.PrivateIdentification.Other.Identification",
)
record["applicant_iban"] = record.get(
"EntryDetails.TransactionDetails.RelatedParties.CreditorAccount.Identification.IBAN"
)
record["applicant_name"] = record.get(
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Party.Name"
record["applicant_name"] = _first(
record,
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Party.Name",
# camt.052.001.02 (ING) has no Pty wrapper around the name.
"EntryDetails.TransactionDetails.RelatedParties.Creditor.Name",
)
record["recipient_name"] = record.get(
"EntryDetails.TransactionDetails.RelatedParties.Debtor.Party.Name"
record["recipient_name"] = _first(
record,
"EntryDetails.TransactionDetails.RelatedParties.Debtor.Party.Name",
"EntryDetails.TransactionDetails.RelatedParties.Debtor.Name",
)
record["status"] = "D"

Expand Down
101 changes: 101 additions & 0 deletions tests/test_camt_parser_052_001_02.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
import datetime
from decimal import Decimal

from fints.camt_parser import _join_repeated, camt053_to_dict
from fints.models import Amount

# Modelled on what ING (camt.052.001.02) returns: no Pty wrapper around
# party names, and the remittance information split into one Ustrd per
# field rather than one per 35 characters.
data = b"""<?xml version="1.0" encoding="UTF-8"?>
<Document xmlns="urn:iso:std:iso:20022:tech:xsd:camt.052.001.02">
<BkToCstmrAcctRpt>
<GrpHdr>
<MsgId>1</MsgId>
<CreDtTm>2026-01-15T12:00:00+01:00</CreDtTm>
</GrpHdr>
<Rpt>
<Id>1</Id>
<CreDtTm>2026-01-15T12:00:00+01:00</CreDtTm>
<Acct>
<Id><IBAN>DE02120300000000202051</IBAN></Id>
<Ccy>EUR</Ccy>
</Acct>
<Ntry>
<Amt Ccy="EUR">30.95</Amt>
<CdtDbtInd>DBIT</CdtDbtInd>
<Sts>BOOK</Sts>
<BookgDt><Dt>2026-01-14</Dt></BookgDt>
<ValDt><Dt>2026-01-14</Dt></ValDt>
<NtryDtls>
<TxDtls>
<RltdPties>
<Cdtr>
<Nm>VISA SUPERMARKT MUSTERSTADT </Nm>
</Cdtr>
</RltdPties>
<RmtInf>
<Ustrd>NR XXXX 1234 MUSTERSTADT DE</Ustrd>
<Ustrd>KAUFUMSATZ</Ustrd>
<Ustrd>12.01 30.95</Ustrd>
<Ustrd>123456</Ustrd>
<Ustrd>ARN00000000000000000000000</Ustrd>
</RmtInf>
</TxDtls>
</NtryDtls>
<AddtlNtryInf>Lastschrifteinzug</AddtlNtryInf>
</Ntry>
<Ntry>
<Amt Ccy="EUR">22.00</Amt>
<CdtDbtInd>CRDT</CdtDbtInd>
<Sts>BOOK</Sts>
<BookgDt><Dt>2026-01-12</Dt></BookgDt>
<ValDt><Dt>2026-01-12</Dt></ValDt>
<NtryDtls>
<TxDtls>
<RltdPties>
<Dbtr>
<Nm>Erika Musterfrau</Nm>
</Dbtr>
<DbtrAcct>
<Id><IBAN>DE02500105170137075030</IBAN></Id>
</DbtrAcct>
</RltdPties>
<RmtInf>
<Ustrd>Taschengeld Januar</Ustrd>
</RmtInf>
</TxDtls>
</NtryDtls>
<AddtlNtryInf>Gutschrift</AddtlNtryInf>
</Ntry>
</Rpt>
</BkToCstmrAcctRpt>
</Document>
"""


def test_join_repeated():
# Fields on separate lines get a space between them.
assert _join_repeated("NR XXXX 1234 MUSTERSTADT DE", "KAUFUMSATZ") == "NR XXXX 1234 MUSTERSTADT DE KAUFUMSATZ"
# A chunk that already ends in whitespace is left alone.
assert _join_repeated("CRED: ", "DE00ZZZ123456789") == "CRED: DE00ZZZ123456789"
assert _join_repeated("CRED:", " DE00ZZZ123456789") == "CRED: DE00ZZZ123456789"
# Missing text does not raise and adds nothing.
assert _join_repeated("abc", None) == "abc"
assert _join_repeated(None, "abc") == "abc"


def test_parse_without_party_wrapper():
debit, credit = camt053_to_dict(data)

assert debit["amount"] == Amount(Decimal("-30.95"), "EUR")
assert debit["applicant_name"] == "VISA SUPERMARKT MUSTERSTADT "
assert debit["purpose"] == "NR XXXX 1234 MUSTERSTADT DE KAUFUMSATZ 12.01 30.95 123456 ARN00000000000000000000000"
assert debit["AdditionalEntryInformation"] == "Lastschrifteinzug"
assert debit["entry_date"] == datetime.date(2026, 1, 14)

assert credit["amount"] == Amount(Decimal("22.00"), "EUR")
assert credit["applicant_name"] == "Erika Musterfrau"
assert credit["applicant_iban"] == "DE02500105170137075030"
assert credit["purpose"] == "Taschengeld Januar"
assert credit["AdditionalEntryInformation"] == "Gutschrift"