1# encoding=utf-8

2 import re

3 import functools

4 import calendar

5 import collections

6

7

8 def 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

18 def 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

35 def 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

42 def 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

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

56

57

58 def 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

71 tnr_re = re.compile(

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

73 flags=re.MULTILINE | re.UNICODE

74)

75

76

77 def 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

95 DETAIL_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/

109 GVC_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

132 def _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

179 def _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

211 def 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

243 transaction_details_post_processor_with_space = functools.partial(

244 transaction_details_post_processor, space=True

245)

246

247

248 def 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