Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
29 changes: 22 additions & 7 deletions pycamt/parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,10 @@
from lxml import etree as ET


class Camt053ParseError(Exception):
pass


class Camt053Parser:
"""
A parser class for camt.053 XML files, designed to be flexible and extensible for different CAMT.053 versions.
Expand Down Expand Up @@ -85,6 +89,22 @@ def _detect_version(self):
return version
return "unknown"

def _find_statements_or_reports(self):
"""
Finds the 'Stmt' or 'Rpt' elements in the XML content.

Returns
-------
elements : list[Element]
The detected CAMT.053 'Stmt' or 'Rpt' elements.
"""
for xmlpath in (".//Stmt", ".//Rpt"):
stmts = self.tree.findall(xmlpath, self.namespaces)
if len(stmts) != 0:
return stmts

raise Camt053ParseError("Neither 'Stmt' nor 'Rpt' element found")

def get_group_header(self):
"""
Extracts the group header information from the CAMT.053 file.
Expand Down Expand Up @@ -127,9 +147,8 @@ def get_transactions(self):
A list of dictionaries, each representing a transaction with its associated data.
"""
transactions = []
statements = self.tree.findall(".//Stmt", self.namespaces)

for statement in statements:
for statement in self._find_statements_or_reports():
entries = statement.findall(".//Ntry", self.namespaces)
for entry in entries:
transactions.extend(self._extract_transaction(entry, statement))
Expand Down Expand Up @@ -345,12 +364,8 @@ def get_statement_info(self):
- Currency: Account currency (if available)
"""
statements = []
stmts = self.tree.findall(".//Stmt", self.namespaces)
if len(stmts) == 0:
# Maybe we have a Rpt file
stmts = self.tree.findall(".//Rpt", self.namespaces)

for stmt in stmts:
for stmt in self._find_statements_or_reports():
# Extract IBAN
iban = stmt.find(".//Acct//Id//IBAN", self.namespaces)
iban_text = iban.text if iban is not None else None
Expand Down
167 changes: 114 additions & 53 deletions tests/test_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,60 +3,121 @@
from pycamt.parser import Camt053Parser


XML_DATA_STMT = """
<Document xmlns="urn:iso:std:iso:20022:tech:xsd:camt.053.001.02">
<BkToCstmrStmt>
<GrpHdr>
<MsgId>ABC123</MsgId>
<CreDtTm>2020-06-23T18:56:25.64Z</CreDtTm>
</GrpHdr>
<Stmt>
<Acct>
<Id>
<IBAN>GB33BUKB20201555555555</IBAN>
</Id>
</Acct>
<Bal>
<Tp>
<CdOrPrtry>
<Cd>OPBD</Cd>
</CdOrPrtry>
</Tp>
<Dt>
<Dt>2025-07-31</Dt>
</Dt>
<Amt Ccy="EUR">1000.00</Amt>
</Bal>
<Ntry>
<Amt Ccy="EUR">500.00</Amt>
<CdtDbtInd>CRDT</CdtDbtInd>
<BookgDt>
<Dt>2020-06-23</Dt>
</BookgDt>
<ValDt>
<Dt>2020-06-23</Dt>
</ValDt>
<AcctSvcrRef>123</AcctSvcrRef>
<NtryDtls>
<TxDtls>
<Refs>
<EndToEndId>ENDTOENDID123</EndToEndId>
</Refs>
<AmtDtls>
<TxAmt>
<Amt Ccy="EUR">500.00</Amt>
</TxAmt>
</AmtDtls>
</TxDtls>
</NtryDtls>
</Ntry>
</Stmt>
</BkToCstmrStmt>
</Document>
"""

XML_DATA_RPT = """
<Document xmlns="urn:iso:std:iso:20022:tech:xsd:camt.052.001.08" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="urn:iso:std:iso:20022:tech:xsd:camt.052.001.08 camt.052.001.08.xsd">
<BkToCstmrAcctRpt>
<GrpHdr>
<MsgId>ABC123</MsgId>
<CreDtTm>2020-06-23T18:56:25.64Z</CreDtTm>
</GrpHdr>
<Rpt>
<Acct>
<Id>
<IBAN>GB33BUKB20201555555555</IBAN>
</Id>
</Acct>
<Bal>
<Tp>
<CdOrPrtry>
<Cd>OPBD</Cd>
</CdOrPrtry>
</Tp>
<Dt>
<Dt>2025-07-31</Dt>
</Dt>
<Amt Ccy="EUR">1000.00</Amt>
</Bal>
<Ntry>
<Amt Ccy="EUR">500.00</Amt>
<CdtDbtInd>CRDT</CdtDbtInd>
<Sts>
<Cd>BOOK</Cd>
</Sts>
<BookgDt>
<Dt>2020-06-23</Dt>
</BookgDt>
<ValDt>
<Dt>2020-06-23</Dt>
</ValDt>
<AcctSvcrRef>123</AcctSvcrRef>
<NtryDtls>
<TxDtls>
<Refs>
<EndToEndId>ENDTOENDID123</EndToEndId>
</Refs>
<Amt Ccy="EUR">500.00</Amt>
</TxDtls>
</NtryDtls>
</Ntry>
</Rpt>
</BkToCstmrAcctRpt>
</Document>
"""

def pytest_generate_tests(metafunc):
if "parser" in metafunc.fixturenames:
metafunc.parametrize("parser", ["parser_stmt", "parser_rpt"], indirect=True)


@pytest.fixture
def parser():
xml_data = """
<Document xmlns="urn:iso:std:iso:20022:tech:xsd:camt.053.001.02">
<BkToCstmrStmt>
<GrpHdr>
<MsgId>ABC123</MsgId>
<CreDtTm>2020-06-23T18:56:25.64Z</CreDtTm>
</GrpHdr>
<Stmt>
<Acct>
<Id>
<IBAN>GB33BUKB20201555555555</IBAN>
</Id>
</Acct>
<Bal>
<Tp>
<CdOrPrtry>
<Cd>OPBD</Cd>
</CdOrPrtry>
</Tp>
<Dt>
<Dt>2025-07-31</Dt>
</Dt>
<Amt Ccy="EUR">1000.00</Amt>
</Bal>
<Ntry>
<Amt Ccy="EUR">500.00</Amt>
<CdtDbtInd>CRDT</CdtDbtInd>
<BookgDt>
<Dt>2020-06-23</Dt>
</BookgDt>
<ValDt>
<Dt>2020-06-23</Dt>
</ValDt>
<AcctSvcrRef>123</AcctSvcrRef>
<NtryDtls>
<TxDtls>
<Refs>
<EndToEndId>ENDTOENDID123</EndToEndId>
</Refs>
<AmtDtls>
<TxAmt>
<Amt Ccy="EUR">500.00</Amt>
</TxAmt>
</AmtDtls>
</TxDtls>
</NtryDtls>
</Ntry>
</Stmt>
</BkToCstmrStmt>
</Document>
"""
return Camt053Parser(xml_data)
def parser(request):
if request.param == "parser_stmt":
return Camt053Parser(XML_DATA_STMT)
if request.param == "parser_rpt":
return Camt053Parser(XML_DATA_RPT)
raise ValueError("invalid internal test config")


class TestCamt053Parser:
Expand Down
Loading