213 lines
7.2 KiB
Python
213 lines
7.2 KiB
Python
# Copyright 2011 Akretion
|
|
# Copyright 2011-2019 Camptocamp SA
|
|
# Copyright 2013 Savoir-faire Linux
|
|
# Copyright 2014 ACSONE SA/NV
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
|
|
import base64
|
|
import csv
|
|
from openerp import _, fields
|
|
|
|
|
|
def UnicodeDictReader(utf8_data, **kwargs):
|
|
sniffer = csv.Sniffer()
|
|
pos = utf8_data.tell()
|
|
sample_data = utf8_data.read(2048)
|
|
utf8_data.seek(pos)
|
|
if not kwargs.get('dialect'):
|
|
dialect = sniffer.sniff(sample_data, delimiters=',;\t')
|
|
del kwargs['dialect']
|
|
else:
|
|
dialect = kwargs.pop('dialect')
|
|
csv_reader = csv.DictReader(utf8_data, dialect=dialect, **kwargs)
|
|
for row in csv_reader:
|
|
yield dict([(str(key or ''),
|
|
str(value or ''))
|
|
for key, value in row.items()])
|
|
|
|
|
|
class AccountMoveImportParser(object):
|
|
|
|
"""
|
|
Generic abstract class for defining parser for different files and
|
|
format to import in a bank statement. Inherit from it to create your
|
|
own. If your file is a .csv or .xls format, you should consider inheirt
|
|
from the FileParser instead.
|
|
"""
|
|
|
|
def __init__(self, journal, *args, **kwargs):
|
|
# The name of the parser as it will be called
|
|
self.parser_name = journal.import_type
|
|
# The result as a list of row. One row per line of data in the file,
|
|
# but not the commission one!
|
|
self.result_row_list = None
|
|
# The file buffer on which to work on
|
|
self.filebuffer = None
|
|
# The profile record to access its parameters in any parser method
|
|
self.journal = journal
|
|
self.move_date = None
|
|
self.move_name = None
|
|
self.move_ref = None
|
|
self.support_multi_moves = None
|
|
|
|
@classmethod
|
|
def parser_for(cls, parser_name):
|
|
"""Override this method for every new parser, so that
|
|
new_bank_statement_parser can return the good class from his name.
|
|
"""
|
|
return False
|
|
|
|
def _decode_64b_stream(self):
|
|
"""Decode self.filebuffer in base 64 and override it"""
|
|
self.filebuffer = base64.b64decode(self.filebuffer)
|
|
return True
|
|
|
|
def _format(self, decode_base_64=True, **kwargs):
|
|
"""Decode into base 64 if asked and Format the given filebuffer by
|
|
calling _custom_format method.
|
|
"""
|
|
if decode_base_64:
|
|
self._decode_64b_stream()
|
|
self._custom_format(kwargs)
|
|
return True
|
|
|
|
def _custom_format(self, *args, **kwargs):
|
|
"""Implement a method in your parser to convert format, encoding and so
|
|
on before starting to work on datas. Work on self.filebuffer
|
|
"""
|
|
return NotImplementedError
|
|
|
|
def _pre(self, *args, **kwargs):
|
|
"""Implement a method in your parser to make a pre-treatment on datas
|
|
before parsing them, like concatenate stuff, and so... Work on
|
|
self.filebuffer
|
|
"""
|
|
return NotImplementedError
|
|
|
|
def _parse(self, *args, **kwargs):
|
|
"""Implement a method in your parser to save the result of parsing
|
|
self.filebuffer in self.result_row_list instance property.
|
|
"""
|
|
return NotImplementedError
|
|
|
|
def _validate(self, *args, **kwargs):
|
|
"""Implement a method in your parser to validate the
|
|
self.result_row_list instance property and raise an error if not valid.
|
|
"""
|
|
return NotImplementedError
|
|
|
|
def _post(self, *args, **kwargs):
|
|
"""Implement a method in your parser to make some last changes on the
|
|
result of parsing the datas, like converting dates, computing
|
|
commission, ...
|
|
"""
|
|
return NotImplementedError
|
|
|
|
def get_move_vals(self):
|
|
"""This method return a dict of vals that ca be passed to create method
|
|
of statement.
|
|
:return: dict of vals that represent additional infos for the statement
|
|
"""
|
|
return {
|
|
'name': self.move_name or '/',
|
|
'date': self.move_date or fields.Datetime.now(),
|
|
'ref': self.move_ref or '/'
|
|
}
|
|
|
|
def get_move_line_vals(self, line, *args, **kwargs):
|
|
"""Implement a method in your parser that must return a dict of vals
|
|
that can be passed to create method of statement line in order to
|
|
record it. It is the responsibility of every parser to give this dict
|
|
of vals, so each one can implement his own way of recording the lines.
|
|
|
|
:param: line: a dict of vals that represent a line of result_row_list
|
|
:return: dict of values to give to the create method of statement line,
|
|
it MUST contain at least:
|
|
{
|
|
'name':value,
|
|
'date':value,
|
|
'amount':value,
|
|
'ref':value,
|
|
}
|
|
"""
|
|
return NotImplementedError
|
|
|
|
def parse(self, filebuffer, *args, **kwargs):
|
|
"""This will be the method that will be called by wizard, button and so
|
|
to parse a filebuffer by calling successively all the private method
|
|
that need to be define for each parser.
|
|
Return:
|
|
[] of rows as {'key':value}
|
|
|
|
Note: The row_list must contain only value that are present in the
|
|
account.bank.statement.line object !!!
|
|
"""
|
|
if filebuffer:
|
|
self.filebuffer = filebuffer
|
|
else:
|
|
raise Exception(_('No buffer file given.'))
|
|
self._format(*args, **kwargs)
|
|
self._pre(*args, **kwargs)
|
|
if self.support_multi_moves:
|
|
while self._parse(*args, **kwargs):
|
|
self._validate(*args, **kwargs)
|
|
self._post(*args, **kwargs)
|
|
yield self.result_row_list
|
|
else:
|
|
self._parse(*args, **kwargs)
|
|
self._validate(*args, **kwargs)
|
|
self._post(*args, **kwargs)
|
|
yield self.result_row_list
|
|
|
|
|
|
def itersubclasses(cls, _seen=None):
|
|
"""
|
|
itersubclasses(cls)
|
|
|
|
Generator over all subclasses of a given class, in depth first order.
|
|
|
|
>>> list(itersubclasses(int)) == [bool]
|
|
True
|
|
>>> class A(object): pass
|
|
>>> class B(A): pass
|
|
>>> class C(A): pass
|
|
>>> class D(B,C): pass
|
|
>>> class E(D): pass
|
|
>>>
|
|
>>> for cls in itersubclasses(A):
|
|
... print(cls.__name__)
|
|
B
|
|
D
|
|
E
|
|
C
|
|
>>> # get ALL (new-style) classes currently defined
|
|
>>> [cls.__name__ for cls in itersubclasses(object)] #doctest: +ELLIPSIS
|
|
['type', ...'tuple', ...]
|
|
"""
|
|
if not isinstance(cls, type):
|
|
raise TypeError('itersubclasses must be called with '
|
|
'new-style classes, not %.100r' % cls)
|
|
if _seen is None:
|
|
_seen = set()
|
|
try:
|
|
subs = cls.__subclasses__()
|
|
except TypeError: # fails only when cls is type
|
|
subs = cls.__subclasses__(cls)
|
|
for sub in subs:
|
|
if sub not in _seen:
|
|
_seen.add(sub)
|
|
yield sub
|
|
for sub in itersubclasses(sub, _seen):
|
|
yield sub
|
|
|
|
|
|
def new_move_parser(journal, *args, **kwargs):
|
|
"""Return an instance of the good parser class based on the given profile.
|
|
|
|
:param profile: browse_record of import profile.
|
|
:return: class instance for given profile import type.
|
|
"""
|
|
for cls in itersubclasses(AccountMoveImportParser):
|
|
if cls.parser_for(journal.import_type):
|
|
return cls(journal, *args, **kwargs)
|
|
raise ValueError
|