You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

310 lines
12 KiB

  1. # Copyright (C) 2014-2015 Therp BV <http://therp.nl>.
  2. # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
  3. """Generic parser for MT940 files, base for customized versions per bank."""
  4. import re
  5. import logging
  6. from datetime import datetime
  7. def str2amount(sign, amount_str):
  8. """Convert sign (C or D) and amount in string to signed amount (float)."""
  9. factor = (1 if sign == 'C' else -1)
  10. return factor * float(amount_str.replace(',', '.'))
  11. def get_subfields(data, codewords):
  12. """Return dictionary with value array for each codeword in data.
  13. For instance:
  14. data =
  15. /BENM//NAME/Kosten/REMI/Periode 01-10-2013 t/m 31-12-2013/ISDT/20
  16. codewords = ['BENM', 'ADDR', 'NAME', 'CNTP', ISDT', 'REMI']
  17. Then return subfields = {
  18. 'BENM': [],
  19. 'NAME': ['Kosten'],
  20. 'REMI': ['Periode 01-10-2013 t', 'm 31-12-2013'],
  21. 'ISDT': ['20'],
  22. }
  23. """
  24. subfields = {}
  25. current_codeword = None
  26. for word in data.split('/'):
  27. if not word and not current_codeword:
  28. continue
  29. if word in codewords:
  30. current_codeword = word
  31. subfields[current_codeword] = []
  32. continue
  33. if current_codeword in subfields:
  34. subfields[current_codeword].append(word)
  35. return subfields
  36. def get_counterpart(transaction, subfield):
  37. """Get counterpart from transaction.
  38. Counterpart is often stored in subfield of tag 86. The subfield
  39. can be BENM, ORDP, CNTP"""
  40. if not subfield:
  41. return # subfield is empty
  42. if len(subfield) >= 1 and subfield[0]:
  43. transaction.update({'account_number': subfield[0]})
  44. if len(subfield) >= 2 and subfield[1]:
  45. transaction.update({'account_bic': subfield[1]})
  46. if len(subfield) >= 3 and subfield[2]:
  47. transaction.update({'partner_name': subfield[2]})
  48. def handle_common_subfields(transaction, subfields):
  49. """Deal with common functionality for tag 86 subfields."""
  50. # Get counterpart from CNTP, BENM or ORDP subfields:
  51. for counterpart_field in ['CNTP', 'BENM', 'ORDP']:
  52. if counterpart_field in subfields:
  53. get_counterpart(transaction, subfields[counterpart_field])
  54. if not transaction.get('name'):
  55. transaction['name'] = ''
  56. # REMI: Remitter information (text entered by other party on trans.):
  57. if 'REMI' in subfields:
  58. transaction['name'] += (
  59. subfields['REMI'][2]
  60. # this might look like
  61. # /REMI/USTD//<remittance info>/
  62. # or
  63. # /REMI/STRD/CUR/<betalingskenmerk>/
  64. if len(subfields['REMI']) >= 3 and subfields['REMI'][0] in [
  65. 'STRD', 'USTD'
  66. ]
  67. else
  68. '/'.join(x for x in subfields['REMI'] if x)
  69. )
  70. # EREF: End-to-end reference
  71. if 'EREF' in subfields:
  72. transaction['name'] += '/'.join(filter(bool, subfields['EREF']))
  73. # Get transaction reference subfield (might vary):
  74. if transaction.get('ref') in subfields:
  75. transaction['ref'] = ''.join(subfields[transaction['ref']])
  76. class MT940(object):
  77. """Inherit this class in your account_banking.parsers.models.parser,
  78. define functions to handle the tags you need to handle and adjust static
  79. variables as needed.
  80. At least, you should override handle_tag_61 and handle_tag_86.
  81. Don't forget to call super.
  82. handle_tag_* functions receive the remainder of the the line (that is,
  83. without ':XX:') and are supposed to write into self.current_transaction
  84. """
  85. def __init__(self):
  86. """Initialize parser - override at least header_regex.
  87. This in fact uses the ING syntax, override in others."""
  88. self.mt940_type = 'General'
  89. self.header_lines = 3 # Number of lines to skip
  90. self.header_regex = '^0000 01INGBNL2AXXXX|^{1' # Start of header
  91. self.footer_regex = '^-}$|^-XXX$' # Stop processing on seeing this
  92. self.tag_regex = '^:[0-9]{2}[A-Z]*:' # Start of new tag
  93. self.current_statement = None
  94. self.current_transaction = None
  95. self.statements = []
  96. self.currency_code = None
  97. self.account_number = None
  98. def is_mt940(self, line):
  99. """determine if a line is the header of a statement"""
  100. if not bool(re.match(self.header_regex, line)):
  101. raise ValueError(
  102. 'File starting with %s does not seem to be a'
  103. ' valid %s MT940 format bank statement.' %
  104. (line[:12], self.mt940_type)
  105. )
  106. def is_mt940_statement(self, line):
  107. """determine if line is the start of a statement"""
  108. if not bool(line.startswith('{4:')):
  109. raise ValueError(
  110. 'The pre processed match %s does not seem to be a'
  111. ' valid %s MT940 format bank statement. Every statement'
  112. ' should start be a dict starting with {4:..' % line
  113. )
  114. def pre_process_data(self, data):
  115. matches = []
  116. self.is_mt940(line=data)
  117. data = data.replace(
  118. '-}', '}').replace('}{', '}\r\n{').replace('\r\n', '\n')
  119. if data.startswith(':940:'):
  120. for statement in data.replace(':940:', '').split(':20:'):
  121. match = '{4:\n:20:' + statement + '}'
  122. matches.append(match)
  123. else:
  124. tag_re = re.compile(
  125. r'(\{4:[^{}]+\})',
  126. re.MULTILINE)
  127. matches = tag_re.findall(data)
  128. return matches
  129. def parse(self, data, header_lines=None):
  130. """Parse mt940 bank statement file contents."""
  131. data = data.decode()
  132. matches = self.pre_process_data(data)
  133. for match in matches:
  134. self.is_mt940_statement(line=match)
  135. iterator = '\n'.join(
  136. match.split('\n')[1:-1]).split('\n').__iter__()
  137. line = None
  138. record_line = ''
  139. try:
  140. while True:
  141. if not self.current_statement:
  142. self.handle_header(line, iterator,
  143. header_lines=header_lines)
  144. line = iterator.next()
  145. if not self.is_tag(line) and not self.is_footer(line):
  146. record_line = self.add_record_line(line, record_line)
  147. continue
  148. if record_line:
  149. self.handle_record(record_line)
  150. if self.is_footer(line):
  151. self.handle_footer(line, iterator)
  152. record_line = ''
  153. continue
  154. record_line = line
  155. except StopIteration:
  156. pass
  157. if self.current_statement:
  158. if record_line:
  159. self.handle_record(record_line)
  160. record_line = ''
  161. self.statements.append(self.current_statement)
  162. self.current_statement = None
  163. return self.currency_code, self.account_number, self.statements
  164. def add_record_line(self, line, record_line):
  165. record_line += line
  166. return record_line
  167. def is_footer(self, line):
  168. """determine if a line is the footer of a statement"""
  169. return line and bool(re.match(self.footer_regex, line))
  170. def is_tag(self, line):
  171. """determine if a line has a tag"""
  172. return line and bool(re.match(self.tag_regex, line))
  173. def handle_header(self, dummy_line, iterator, header_lines=None):
  174. """skip header lines, create current statement"""
  175. if not header_lines:
  176. header_lines = self.header_lines
  177. for dummy_i in range(header_lines):
  178. iterator.next()
  179. self.current_statement = {
  180. 'name': None,
  181. 'date': None,
  182. 'balance_start': 0.0,
  183. 'balance_end_real': 0.0,
  184. 'transactions': []
  185. }
  186. def handle_footer(self, dummy_line, dummy_iterator):
  187. """add current statement to list, reset state"""
  188. self.statements.append(self.current_statement)
  189. self.current_statement = None
  190. def handle_record(self, line):
  191. """find a function to handle the record represented by line"""
  192. tag_match = re.match(self.tag_regex, line)
  193. tag = tag_match.group(0).strip(':')
  194. if not hasattr(self, 'handle_tag_%s' % tag): # pragma: no cover
  195. logging.error('Unknown tag %s', tag)
  196. logging.error(line)
  197. return
  198. handler = getattr(self, 'handle_tag_%s' % tag)
  199. handler(line[tag_match.end():])
  200. def handle_tag_20(self, data):
  201. """Contains unique ? message ID"""
  202. pass
  203. def handle_tag_25(self, data):
  204. """Handle tag 25: local bank account information."""
  205. data = data.replace('EUR', '').replace('.', '').strip()
  206. self.account_number = data
  207. def handle_tag_28C(self, data):
  208. """Sequence number within batch - normally only zeroes."""
  209. pass
  210. def handle_tag_60F(self, data):
  211. """get start balance and currency"""
  212. # For the moment only first 60F record
  213. # The alternative would be to split the file and start a new
  214. # statement for each 20: tag encountered.
  215. if not self.currency_code:
  216. self.currency_code = data[7:10]
  217. self.current_statement['balance_start'] = str2amount(
  218. data[0],
  219. data[10:]
  220. )
  221. if not self.current_statement['date']:
  222. self.current_statement['date'] = datetime.strptime(data[1:7],
  223. '%y%m%d')
  224. def handle_tag_61(self, data):
  225. """get transaction values"""
  226. self.current_statement['transactions'].append({})
  227. self.current_transaction = self.current_statement['transactions'][-1]
  228. self.current_transaction['date'] = datetime.strptime(
  229. data[:6],
  230. '%y%m%d'
  231. )
  232. def handle_tag_62F(self, data):
  233. """Get ending balance, statement date and id.
  234. We use the date on the last 62F tag as statement date, as the date
  235. on the 60F record (previous end balance) might contain a date in
  236. a previous period.
  237. We generate the statement.id from the local_account and the end-date,
  238. this should normally be unique, provided there is a maximum of
  239. one statement per day.
  240. Depending on the bank, there might be multiple 62F tags in the import
  241. file. The last one counts.
  242. """
  243. self.current_statement['balance_end_real'] = str2amount(
  244. data[0],
  245. data[10:]
  246. )
  247. self.current_statement['date'] = datetime.strptime(data[1:7], '%y%m%d')
  248. # Only replace logically empty (only whitespace or zeroes) id's:
  249. # But do replace statement_id's added before (therefore starting
  250. # with local_account), because we need the date on the last 62F
  251. # record.
  252. statement_name = self.current_statement['name'] or ''
  253. test_empty_id = re.sub(r'[\s0]', '', statement_name)
  254. is_account_number = statement_name.startswith(self.account_number)
  255. if not test_empty_id or is_account_number:
  256. self.current_statement['name'] = '%s-%s' % (
  257. self.account_number,
  258. self.current_statement['date'].strftime('%Y-%m-%d'),
  259. )
  260. def handle_tag_64(self, data):
  261. """get current balance in currency"""
  262. pass
  263. def handle_tag_65(self, data):
  264. """get future balance in currency"""
  265. pass
  266. def handle_tag_86(self, data):
  267. """details for previous transaction, here most differences between
  268. banks occur"""
  269. pass