Coverage for mt940/processors.py: 0%
111 statements
« prev ^ index » next coverage.py v7.2.7, created at 2025-03-01 10:17 +0000
« prev ^ index » next coverage.py v7.2.7, created at 2025-03-01 10:17 +0000
1# encoding=utf-8
2import re
3import functools
4import calendar
5import collections
8def add_currency_pre_processor(currency, overwrite=True):
9 def _add_currency_pre_processor(transactions, tag, tag_dict, *args):
10 if 'currency' not in tag_dict or overwrite: # pragma: no branch
11 tag_dict['currency'] = currency
13 return tag_dict
15 return _add_currency_pre_processor
18def date_fixup_pre_processor(transactions, tag, tag_dict, *args):
19 """
20 Replace illegal February 29, 30 dates with the last day of February.
22 German banks use a variant of the 30/360 interest rate calculation,
23 where each month has always 30 days even February. Python's datetime
24 module won't accept such dates.
25 """
26 if tag_dict['month'] == '02':
27 year = int(tag_dict['year'], 10)
28 _, max_month_day = calendar.monthrange(year, 2)
29 if int(tag_dict['day'], 10) > max_month_day:
30 tag_dict['day'] = str(max_month_day)
32 return tag_dict
35def date_cleanup_post_processor(transactions, tag, tag_dict, result):
36 for k in ('day', 'month', 'year', 'entry_day', 'entry_month'):
37 result.pop(k, None)
39 return result
42def mBank_set_transaction_code(transactions, tag, tag_dict, *args):
43 """
44 mBank Collect uses transaction code 911 to distinguish icoming mass
45 payments transactions, adding transaction_code may be helpful in further
46 processing
47 """
48 tag_dict['transaction_code'] = int(
49 tag_dict[tag.slug].split(';')[0].split(' ', 1)[0]
50 )
52 return tag_dict
55iph_id_re = re.compile(r' ID IPH: X*(?P<iph_id>\d{0,14});')
58def mBank_set_iph_id(transactions, tag, tag_dict, *args):
59 """
60 mBank Collect uses ID IPH to distinguish between virtual accounts,
61 adding iph_id may be helpful in further processing
62 """
63 matches = iph_id_re.search(tag_dict[tag.slug])
65 if matches: # pragma no branch
66 tag_dict['iph_id'] = matches.groupdict()['iph_id']
68 return tag_dict
71tnr_re = re.compile(
72 r'TNR:[ \n](?P<tnr>\d+\.\d+)',
73 flags=re.MULTILINE | re.UNICODE
74)
77def mBank_set_tnr(transactions, tag, tag_dict, *args):
78 """
79 mBank Collect states TNR in transaction details as unique id for
80 transactions, that may be used to identify the same transactions in
81 different statement files eg. partial mt942 and full mt940
82 Information about tnr uniqueness has been obtained from mBank support,
83 it lacks in mt940 mBank specification.
84 """
86 matches = tnr_re.search(tag_dict[tag.slug])
88 if matches: # pragma no branch
89 tag_dict['tnr'] = matches.groupdict()['tnr']
91 return tag_dict
94# https://www.db-bankline.deutsche-bank.com/download/MT940_Deutschland_Structure2002.pdf
95DETAIL_KEYS = {
96 '': 'transaction_code',
97 '00': 'posting_text',
98 '10': 'prima_nota',
99 '20': 'purpose',
100 '30': 'applicant_bin',
101 '31': 'applicant_iban',
102 '32': 'applicant_name',
103 '34': 'return_debit_notes',
104 '35': 'recipient_name',
105 '60': 'additional_purpose',
106}
108# https://www.hettwer-beratung.de/sepa-spezialwissen/sepa-technische-anforderungen/sepa-gesch%C3%A4ftsvorfallcodes-gvc-mt-940/
109GVC_KEYS = {
110 '': 'purpose',
111 'IBAN': 'gvc_applicant_iban',
112 'BIC ': 'gvc_applicant_bin',
113 'EREF': 'end_to_end_reference',
114 'MREF': 'additional_position_reference',
115 'CRED': 'applicant_creditor_id',
116 'PURP': 'purpose_code',
117 'SVWZ': 'purpose',
118 'MDAT': 'additional_position_date',
119 'ABWA': 'deviate_applicant',
120 'ABWE': 'deviate_recipient',
121 'SQTP': 'FRST_ONE_OFF_RECC',
122 'ORCR': 'old_SEPA_CI',
123 'ORMR': 'old_SEPA_additional_position_reference',
124 'DDAT': 'settlement_tag',
125 'KREF': 'customer_reference',
126 'DEBT': 'debitor_identifier',
127 'COAM': 'compensation_amount',
128 'OAMT': 'original_amount',
129}
132def _parse_mt940_details(detail_str, space=False):
133 result = collections.defaultdict(list)
135 tmp = collections.OrderedDict()
136 segment = ''
137 segment_type = ''
139 for index, char in enumerate(detail_str):
140 if char != '?':
141 segment += char
142 continue
144 if index + 2 >= len(detail_str):
145 break
147 tmp[segment_type] = segment if not segment_type else segment[2:]
148 segment_type = detail_str[index + 1] + detail_str[index + 2]
149 segment = ''
151 if segment_type: # pragma: no branch
152 tmp[segment_type] = segment if not segment_type else segment[2:]
154 for key, value in tmp.items():
155 if key in DETAIL_KEYS:
156 result[DETAIL_KEYS[key]].append(value)
157 elif key == '33':
158 key32 = DETAIL_KEYS['32']
159 result[key32].append(value)
160 elif key.startswith('2'):
161 key20 = DETAIL_KEYS['20']
162 result[key20].append(value)
163 elif key in {'60', '61', '62', '63', '64', '65'}:
164 key60 = DETAIL_KEYS['60']
165 result[key60].append(value)
167 joined_result = dict()
168 for key in DETAIL_KEYS.values():
169 if space:
170 value = ' '.join(result[key])
171 else:
172 value = ''.join(result[key])
174 joined_result[key] = value or None
176 return joined_result
179def _parse_mt940_gvcodes(purpose):
180 result = {}
182 for key, value in GVC_KEYS.items():
183 result[value] = None
185 tmp = {}
186 segment_type = None
187 text = ''
189 for index, char in enumerate(purpose):
190 if char == '+' and purpose[index - 4:index] in GVC_KEYS:
191 if segment_type:
192 tmp[segment_type] = text[:-4]
193 text = ''
194 else:
195 text = ''
196 segment_type = purpose[index - 4:index]
197 else:
198 text += char
200 if segment_type: # pragma: no branch
201 tmp[segment_type] = text
202 else:
203 tmp[''] = text # pragma: no cover
205 for key, value in tmp.items():
206 result[GVC_KEYS[key]] = value
208 return result
211def transaction_details_post_processor(
212 transactions, tag, tag_dict, result, space=False):
213 '''Parse the extra details in some transaction formats such as the 60-65
214 keys.
216 Args:
217 transactions (mt940.models.Transactions): list of transactions
218 tag (mt940.tags.Tag): tag
219 tag_dict (dict): dict with the raw tag details
220 result (dict): the resulting tag dict
221 space (bool): include spaces between lines in the mt940 details
222 '''
223 details = tag_dict['transaction_details']
224 details = ''.join(detail.strip('\n\r') for detail in details.splitlines())
226 # check for e.g. 103?00...
227 if re.match(r'^\d{3}\?\d{2}', details):
228 result.update(_parse_mt940_details(details, space=space))
230 purpose = result.get('purpose')
232 if purpose and any(
233 gvk in purpose for gvk in GVC_KEYS
234 if gvk != ''
235 ): # pragma: no branch
236 result.update(_parse_mt940_gvcodes(result['purpose']))
238 del result['transaction_details']
240 return result
243transaction_details_post_processor_with_space = functools.partial(
244 transaction_details_post_processor, space=True
245)
248def transactions_to_transaction(*keys):
249 '''Copy the global transactions details to the transaction.
251 Args:
252 *keys (str): the keys to copy to the transaction
253 '''
254 def _transactions_to_transaction(transactions, tag, tag_dict, result):
255 '''Copy the global transactions details to the transaction.
257 Args:
258 transactions (mt940.models.Transactions): list of transactions
259 tag (mt940.tags.Tag): tag
260 tag_dict (dict): dict with the raw tag details
261 result (dict): the resulting tag dict
262 '''
263 for key in keys:
264 if key in transactions.data:
265 result[key] = transactions.data[key]
267 return result
269 return _transactions_to_transaction