audrius/bank-statement-import
1
0
Fork 1
mirror of https://github.com/OCA/bank-statement-import.git synced 2025-01-20 12:37:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

[RFR] *_import_adyen: refactor

Browse Source 
Move parsing code to separate class;
Break up large and unwieldy parsing method in separate methods.
This commit is contained in:
Ronald Portier
2022-03-28 17:58:54 +02:00
committed by Ronald Portier (Therp BV)
parent a3f5847791
commit 7a4da01ea0
5 changed files with 294 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
account_bank_statement_clearing_account/__manifest__.py
View File

@@ -6,7 +6,7 @@
"version": "13.0.1.0.0",
"author": "Opener B.V., Vanmoof BV, Odoo Community Association (OCA)",
"category": "Banking addons",
"website": "https://github.com/oca/bank-statement-import",
"website": "https://github.com/OCA/bank-statement-import",
"license": "AGPL-3",
"depends": ["account"],
"installable": True,

2
account_bank_statement_import_adyen/__manifest__.py
View File

@@ -7,7 +7,7 @@
"version": "13.0.1.0.0",
"author": "Opener BV, Vanmoof BV, Odoo Community Association (OCA)",
"category": "Banking addons",
"website": "https://github.com/oca/bank-statement-import",
"website": "https://github.com/OCA/bank-statement-import",
"license": "AGPL-3",
"depends": [
"base_import",

1
account_bank_statement_import_adyen/models/__init__.py
View File

@@ -1,2 +1,3 @@
from . import account_bank_statement_import
from . import account_bank_statement_import_adyen_parser
from . import account_journal

250
account_bank_statement_import_adyen/models/account_bank_statement_import.py
View File

@@ -5,44 +5,11 @@
# pylint: disable=protected-access,no-self-use
import logging
from odoo import _, fields, models
from odoo import _, models
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__) # pylint: disable=invalid-name
COLUMNS = {
"Company Account": 1,
"Merchant Account": 2,
"Psp Reference": 3,
"Merchant Reference": 4,
"Payment Method": 5, # Not used at present
"Creation Date": 6,
"TimeZone": 7, # Not used at present
"Type": 8,
"Modification Reference": 9,
"Gross Currency": 10, # Not used at present
"Gross Debit (GC)": 11, # Not used at present
"Gross Credit (GC)": 12, # Not used at present
"Exchange Rate": 13, # Not used at present
"Net Currency": 14,
"Net Debit (NC)": 15, # Fee or Merchant Payout
"Net Credit (NC)": 16,
"Commission (NC)": 17,
"Markup (NC)": 18,
"Scheme Fees (NC)": 19,
"Interchange (NC)": 20,
"Payment Method Variant": 21,
"Modification Merchant Reference": 22, # Not used at present
"Batch Number": 23,
"Reserved4": 24, # Not used at present
"Reserved5": 25, # Not used at present
"Reserved6": 26, # Not used at present
"Reserved7": 27, # Not used at present
"Reserved8": 28, # Not used at present
"Reserved9": 29, # Not used at present
"Reserved10": 30, # Not used at present
}
class AccountBankStatementImport(models.TransientModel):
"""Add import of Adyen statements."""
@@ -52,7 +19,6 @@ class AccountBankStatementImport(models.TransientModel):
def _parse_file(self, data_file):
"""Parse an Adyen xlsx file and map merchant account strings to journals."""
try:
_logger.debug(_("Try parsing as Adyen settlement details."))
return self._parse_adyen_file(data_file)
except Exception as exc: # pylint: disable=broad-except
message = _("Statement file was not a Adyen settlement details file.")
@@ -62,94 +28,11 @@ class AccountBankStatementImport(models.TransientModel):
return super()._parse_file(data_file)
def _parse_adyen_file(self, data_file):
"""Parse file assuming it is an Adyen file.
An Exception will be thrown if file cannot be parsed.
"""
# pylint: disable=too-many-locals,too-many-branches
statement = None
headers = False
batch_number = False
fees = 0.0
balance = 0.0
payout = 0.0
"""Just parse the adyen file."""
_logger.debug(_("Try parsing as Adyen settlement details."))
parser = self.env["account.bank.statement.import.adyen.parser"]
rows = self._get_rows(data_file)
num_rows = 0
num_transactions = 0
for row in rows:
num_rows += 1
if not row[1]:
continue
if not headers:
on_header_row = self._check_header_row(row)
if not on_header_row:
continue
self._set_columns(row)
headers = True
continue
if len(row) < 24:
raise ValueError(
"Not an Adyen statement. Unexpected row length %s "
"less then minimum of 24" % len(row)
)
if not statement:
merchant_account = self._get_value(row, "Merchant Account")
self._validate_merchant_account(merchant_account)
batch_number = self._get_value(row, "Batch Number")
statement = self._make_statement(row)
currency_code = self._get_value(row, "Net Currency")
else:
self._update_statement(statement, row)
row_type = self._get_value(row, "Type").strip()
if row_type == "MerchantPayout":
payout -= self._balance(row)
else:
balance += self._balance(row)
num_transactions += 1
transaction = self._get_adyen_transaction(row)
transaction["unique_import_id"] = self._get_unique_import_id(statement)
statement["transactions"].append(transaction)
fees += self._sum_fees(row)
if not headers:
raise ValueError(
"Not an Adyen statement. Did not encounter header row in %d rows."
% (num_rows,)
)
if fees:
balance -= fees
self._add_fees_transaction(statement, fees, batch_number)
if statement["transactions"] and not payout:
_logger.info(_("No payout detected in Adyen statement."))
if self.env.user.company_id.currency_id.compare_amounts(balance, payout) != 0:
raise UserError(
_("Parse error. Balance %s not equal to merchant " "payout %s")
% (balance, payout)
)
_logger.info(
_("Processed %d rows from Adyen statement file with %d transactions"),
num_rows,
num_transactions,
)
return currency_code, merchant_account, [statement]
def _validate_merchant_account(self, merchant_account):
"""Check wether merchant account exist, and belongs to the correct journal."""
journal = self.env["account.journal"].search(
[("adyen_merchant_account", "=", merchant_account)], limit=1
)
if not journal:
raise UserError(
_("No journal refers to Merchant Account %s") % merchant_account
)
if self._context.get("journal_id", journal.id) != journal.id:
raise UserError(
_(
"Selected journal Merchant Account does not match "
"the import file Merchant Account "
"column: %s"
)
% merchant_account
)
return parser.parse_rows(rows)
def _get_rows(self, data_file):
"""Get rows from data_file."""
@@ -162,126 +45,3 @@ class AccountBankStatementImport(models.TransientModel):
import_model = self.env["base_import.import"]
importer = import_model.create({"file": data_file, "file_name": filename})
return importer._read_file({"quoting": '"', "separator": ","})
def _check_header_row(self, row):
"""Header row is the first one with a "Company Account" header cell."""
for cell in row:
if cell == "Company Account":
return True
return False
def _set_columns(self, row):
"""Set columns from headers. There MUST be a 'Company Account' header."""
seen_company_account = False
for num, header in enumerate(row):
if not header.strip():
continue # Ignore empty columns.
if header == "Company Account":
seen_company_account = True
if header not in COLUMNS:
_logger.debug(_("Unknown header %s in Adyen statement headers"), header)
else:
COLUMNS[header] = num # Set the right number for the column.
if not seen_company_account:
raise ValueError(
_("Not an Adyen statement. Headers %s do not contain 'Company Account'")
% ", ".join(row)
)
def _get_value(self, row, column):
"""Get the value from the righ column in the row."""
return row[COLUMNS[column]]
def _make_statement(self, row):
"""Make statement on first transaction in file."""
statement = {"transactions": []}
statement["name"] = "{merchant} {year}/{batch}".format(
merchant=self._get_value(row, "Merchant Account"),
year=self._get_value(row, "Creation Date")[:4],
batch=self._get_value(row, "Batch Number"),
)
statement["date"] = self._get_transaction_date(row)
return statement
def _get_transaction_date(self, row):
"""Get transaction date in right format."""
return fields.Date.from_string(self._get_value(row, "Creation Date"))
def _update_statement(self, statement, row):
"""Update statement from transaction row."""
# Statement date is date of earliest transaction in file.
date = self._get_transaction_date(row)
if date < statement.get("date"):
statement["date"] = date
def _balance(self, row):
return (
-self._sum_amount_values(row, ("Net Debit (NC)",))
+ self._sum_amount_values(row, ("Net Credit (NC)",))
+ self._sum_fees(row)
)
def _sum_fees(self, row):
"""Sum the amounts in the fees columns."""
return self._sum_amount_values(
row,
("Commission (NC)", "Markup (NC)", "Scheme Fees (NC)", "Interchange (NC)",),
)
def _sum_amount_values(self, row, columns):
"""Sum the amounts from the columns passed."""
amount = 0.0
for column in columns:
value = self._get_value(row, column)
if value:
amount += float(value)
return amount
def _get_adyen_transaction(self, row):
"""Get transaction from row.
This can easily be overwritten in custom modules to add extra information.
"""
merchant_account = self._get_value(row, "Merchant Account")
psp_reference = self._get_value(row, "Psp Reference")
merchant_reference = self._get_value(row, "Merchant Reference")
payment_method = self._get_value(row, "Payment Method Variant")
modification_reference = self._get_value(row, "Modification Reference")
transaction = {
"date": self._get_transaction_date(row),
"amount": self._balance(row),
}
transaction["note"] = " ".join(
[
part
for part in [
merchant_account,
psp_reference,
merchant_reference,
payment_method,
]
if part
]
)
transaction["name"] = (
merchant_reference or psp_reference or modification_reference
)
transaction["ref"] = (
psp_reference or modification_reference or merchant_reference
)
transaction["transaction_type"] = self._get_value(row, "Type")
return transaction
def _get_unique_import_id(self, statement):
"""get unique import ID for transaction."""
return statement["name"] + str(len(statement["transactions"])).zfill(4)
def _add_fees_transaction(self, statement, fees, batch_number):
"""Single transaction for all fees in statement."""
transaction = dict(
unique_import_id=self._get_unique_import_id(statement),
date=max(t["date"] for t in statement["transactions"]),
amount=-fees,
name="Commission, markup etc. batch %s" % batch_number,
)
statement["transactions"].append(transaction)

286
account_bank_statement_import_adyen/models/account_bank_statement_import_adyen_parser.py Normal file
View File

@@ -0,0 +1,286 @@
# Copyright 2017 Opener BV (<https://opener.amsterdam>)
# Copyright 2021-2022 Therp BV <https://therp.nl>.
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
"""Add import of Adyen statements."""
# pylint: disable=protected-access,no-self-use
import logging
from odoo import _, fields, models
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__) # pylint: disable=invalid-name
COLUMNS = {
"Company Account": 1,
"Merchant Account": 2,
"Psp Reference": 3,
"Merchant Reference": 4,
"Payment Method": 5, # Not used at present
"Creation Date": 6,
"TimeZone": 7, # Not used at present
"Type": 8,
"Modification Reference": 9,
"Gross Currency": 10, # Not used at present
"Gross Debit (GC)": 11, # Not used at present
"Gross Credit (GC)": 12, # Not used at present
"Exchange Rate": 13, # Not used at present
"Net Currency": 14,
"Net Debit (NC)": 15, # Fee or Merchant Payout
"Net Credit (NC)": 16,
"Commission (NC)": 17,
"Markup (NC)": 18,
"Scheme Fees (NC)": 19,
"Interchange (NC)": 20,
"Payment Method Variant": 21,
"Modification Merchant Reference": 22, # Not used at present
"Batch Number": 23,
"Reserved4": 24, # Not used at present
"Reserved5": 25, # Not used at present
"Reserved6": 26, # Not used at present
"Reserved7": 27, # Not used at present
"Reserved8": 28, # Not used at present
"Reserved9": 29, # Not used at present
"Reserved10": 30, # Not used at present
}
class AccountBankStatementImportAdyenParser(models.TransientModel):
"""Parse Adyen statement files for bank import."""
_name = "account.bank.statement.import.adyen.parser"
_description = "Account Bank Statement Import Adyen Parser"
def parse_rows(self, rows):
"""Parse rows generated from an Adyen file.
An Exception will be thrown if file cannot be parsed.
"""
statement = None
fees = 0.0
balance = 0.0
payout = 0.0
num_rows = self._process_headers(rows)
for row in rows:
num_rows += 1
if not self._is_transaction_row(row):
continue
if not statement:
statement = self._make_statement(row)
statement_info = self._get_statement_info(row)
row_type = self._get_value(row, "Type").strip()
if row_type == "MerchantPayout":
payout -= self._balance(row)
else:
balance += self._balance(row)
transaction = self._get_transaction(row)
self._append_transaction(statement, transaction)
fees += self._sum_fees(row)
if fees:
balance -= fees
self._append_fees_transaction(
statement, fees, statement_info["batch_number"]
)
self._validate_statement(statement, payout, balance)
_logger.info(
_("Processed %d rows from Adyen statement file with %d transactions"),
num_rows,
len(statement["transactions"]),
)
return (
statement_info["currency_code"],
statement_info["merchant_account"],
[statement],
)
def _process_headers(self, rows):
"""Process the headers in the generated rows."""
num_rows = 0
for row in rows:
num_rows += 1
if not row[1]:
continue
on_header_row = self._check_header_row(row)
if not on_header_row:
continue
self._set_columns(row)
return num_rows
raise ValueError(
"Not an Adyen statement. Did not encounter header row in %d rows."
% (num_rows,)
)
def _is_transaction_row(self, row):
"""Check wether row is a not empty and valid transaction row."""
if not row[1]:
return False
if len(row) < 24:
raise ValueError(
"Not an Adyen statement. Unexpected row length %s "
"less then minimum of 24" % len(row)
)
return True
def _get_statement_info(self, row):
"""Get general information for statement."""
merchant_account = self._get_value(row, "Merchant Account")
self._validate_merchant_account(merchant_account)
batch_number = self._get_value(row, "Batch Number")
currency_code = self._get_value(row, "Net Currency")
return {
"merchant_account": merchant_account,
"batch_number": batch_number,
"currency_code": currency_code,
}
def _validate_merchant_account(self, merchant_account):
"""Check wether merchant account exist, and belongs to the correct journal."""
journal = self.env["account.journal"].search(
[("adyen_merchant_account", "=", merchant_account)], limit=1
)
if not journal:
raise UserError(
_("No journal refers to Merchant Account %s") % merchant_account
)
if self._context.get("journal_id", journal.id) != journal.id:
raise UserError(
_(
"Selected journal Merchant Account does not match "
"the import file Merchant Account "
"column: %s"
)
% merchant_account
)
def _check_header_row(self, row):
"""Header row is the first one with a "Company Account" header cell."""
for cell in row:
if cell == "Company Account":
return True
return False
def _set_columns(self, row):
"""Set columns from headers. There MUST be a 'Company Account' header."""
seen_company_account = False
for num, header in enumerate(row):
if not header.strip():
continue # Ignore empty columns.
if header == "Company Account":
seen_company_account = True
if header not in COLUMNS:
_logger.debug(_("Unknown header %s in Adyen statement headers"), header)
else:
COLUMNS[header] = num # Set the right number for the column.
if not seen_company_account:
raise ValueError(
_("Not an Adyen statement. Headers %s do not contain 'Company Account'")
% ", ".join(row)
)
def _validate_statement(self, statement, payout, balance):
"""Check wether statement valid: balanced. Log when no payout."""
if statement["transactions"] and not payout:
_logger.info(_("No payout detected in Adyen statement."))
if self.env.user.company_id.currency_id.compare_amounts(balance, payout) != 0:
raise UserError(
_("Parse error. Balance %s not equal to merchant " "payout %s")
% (balance, payout)
)
def _get_value(self, row, column):
"""Get the value from the righ column in the row."""
return row[COLUMNS[column]]
def _make_statement(self, row):
"""Make statement on first transaction in file."""
statement = {"transactions": []}
statement["name"] = "{merchant} {year}/{batch}".format(
merchant=self._get_value(row, "Merchant Account"),
year=self._get_value(row, "Creation Date")[:4],
batch=self._get_value(row, "Batch Number"),
)
statement["date"] = self._get_transaction_date(row)
return statement
def _get_transaction_date(self, row):
"""Get transaction date in right format."""
return fields.Date.from_string(self._get_value(row, "Creation Date"))
def _balance(self, row):
return (
-self._sum_amount_values(row, ("Net Debit (NC)",))
+ self._sum_amount_values(row, ("Net Credit (NC)",))
+ self._sum_fees(row)
)
def _sum_fees(self, row):
"""Sum the amounts in the fees columns."""
return self._sum_amount_values(
row,
("Commission (NC)", "Markup (NC)", "Scheme Fees (NC)", "Interchange (NC)",),
)
def _sum_amount_values(self, row, columns):
"""Sum the amounts from the columns passed."""
amount = 0.0
for column in columns:
value = self._get_value(row, column)
if value:
amount += float(value)
return amount
def _get_transaction(self, row):
"""Get transaction from row.
This can easily be overwritten in custom modules to add extra information.
"""
merchant_account = self._get_value(row, "Merchant Account")
psp_reference = self._get_value(row, "Psp Reference")
merchant_reference = self._get_value(row, "Merchant Reference")
payment_method = self._get_value(row, "Payment Method Variant")
modification_reference = self._get_value(row, "Modification Reference")
transaction = {
"date": self._get_transaction_date(row),
"amount": self._balance(row),
}
transaction["note"] = " ".join(
[
part
for part in [
merchant_account,
psp_reference,
merchant_reference,
payment_method,
]
if part
]
)
transaction["name"] = (
merchant_reference or psp_reference or modification_reference
)
transaction["ref"] = (
psp_reference or modification_reference or merchant_reference
)
transaction["transaction_type"] = self._get_value(row, "Type")
return transaction
def _append_fees_transaction(self, statement, fees, batch_number):
"""Single transaction for all fees in statement."""
max_date = max(t["date"] for t in statement["transactions"])
transaction = {
"date": max_date,
"amount": -fees,
"name": "Commission, markup etc. batch %s" % batch_number,
}
self._append_transaction(statement, transaction)
def _append_transaction(self, statement, transaction):
"""Add transaction with unique import id to statement."""
# Statement date is date of earliest transaction in file.
if transaction["date"] < statement.get("date"):
statement["date"] = transaction["date"]
transaction["unique_import_id"] = self._get_unique_import_id(statement)
statement["transactions"].append(transaction)
def _get_unique_import_id(self, statement):
"""get unique import ID for transaction."""
return statement["name"] + str(len(statement["transactions"])).zfill(4)