Coverage for mt940/processors.py: 0%

111 statements  

« prev     ^ index     » next       coverage.py v7.3.2, created at 2024-05-04 05:32 +0000

1# encoding=utf-8 

2import re 

3import functools 

4import calendar 

5import collections 

6 

7 

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 

12 

13 return tag_dict 

14 

15 return _add_currency_pre_processor 

16 

17 

18def date_fixup_pre_processor(transactions, tag, tag_dict, *args): 

19 """ 

20 Replace illegal February 29, 30 dates with the last day of February. 

21 

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) 

31 

32 return tag_dict 

33 

34 

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) 

38 

39 return result 

40 

41 

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 ) 

51 

52 return tag_dict 

53 

54 

55iph_id_re = re.compile(r' ID IPH: X*(?P<iph_id>\d{0,14});') 

56 

57 

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]) 

64 

65 if matches: # pragma no branch 

66 tag_dict['iph_id'] = matches.groupdict()['iph_id'] 

67 

68 return tag_dict 

69 

70 

71tnr_re = re.compile( 

72 r'TNR:[ \n](?P<tnr>\d+\.\d+)', 

73 flags=re.MULTILINE | re.UNICODE 

74) 

75 

76 

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 """ 

85 

86 matches = tnr_re.search(tag_dict[tag.slug]) 

87 

88 if matches: # pragma no branch 

89 tag_dict['tnr'] = matches.groupdict()['tnr'] 

90 

91 return tag_dict 

92 

93 

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} 

107 

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} 

130 

131 

132def _parse_mt940_details(detail_str, space=False): 

133 result = collections.defaultdict(list) 

134 

135 tmp = collections.OrderedDict() 

136 segment = '' 

137 segment_type = '' 

138 

139 for index, char in enumerate(detail_str): 

140 if char != '?': 

141 segment += char 

142 continue 

143 

144 if index + 2 >= len(detail_str): 

145 break 

146 

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 = '' 

150 

151 if segment_type: # pragma: no branch 

152 tmp[segment_type] = segment if not segment_type else segment[2:] 

153 

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) 

166 

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]) 

173 

174 joined_result[key] = value or None 

175 

176 return joined_result 

177 

178 

179def _parse_mt940_gvcodes(purpose): 

180 result = {} 

181 

182 for key, value in GVC_KEYS.items(): 

183 result[value] = None 

184 

185 tmp = {} 

186 segment_type = None 

187 text = '' 

188 

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 

199 

200 if segment_type: # pragma: no branch 

201 tmp[segment_type] = text 

202 else: 

203 tmp[''] = text # pragma: no cover 

204 

205 for key, value in tmp.items(): 

206 result[GVC_KEYS[key]] = value 

207 

208 return result 

209 

210 

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. 

215 

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()) 

225 

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)) 

229 

230 purpose = result.get('purpose') 

231 

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'])) 

237 

238 del result['transaction_details'] 

239 

240 return result 

241 

242 

243transaction_details_post_processor_with_space = functools.partial( 

244 transaction_details_post_processor, space=True 

245) 

246 

247 

248def transactions_to_transaction(*keys): 

249 '''Copy the global transactions details to the transaction. 

250 

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. 

256 

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] 

266 

267 return result 

268 

269 return _transactions_to_transaction