OCA
/
bank-statement-import
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
Browse Source

[FIX] temporary fix for qonto

12.0_fix_qonto_2022_date
Nicolas JEUDY 2 years ago
parent
88193f394c
commit
3ff4ad81b1
1 changed files with 70 additions and 57 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 127
      account_bank_statement_import_online_qonto/models/online_bank_statement_provider_qonto.py

127
account_bank_statement_import_online_qonto/models/online_bank_statement_provider_qonto.py
View File

@ -9,21 +9,21 @@ from odoo import api, models, _
from odoo.exceptions import UserError
from odoo.addons.base.models.res_bank import sanitize_account_number
QONTO_ENDPOINT = 'https://thirdparty.qonto.eu/v2'
QONTO_ENDPOINT = "https://thirdparty.qonto.eu/v2"
class OnlineBankStatementProviderQonto(models.Model):
_inherit = 'online.bank.statement.provider'
_inherit = "online.bank.statement.provider"
@api.model
def _get_available_services(self):
return super()._get_available_services() + [
('qonto', 'Qonto.eu'),
("qonto", "Qonto.eu"),
]
def _obtain_statement_data(self, date_since, date_until):
self.ensure_one()
if self.service != 'qonto':
if self.service != "qonto":
return super()._obtain_statement_data(
date_since,
date_until,
@ -32,12 +32,12 @@ class OnlineBankStatementProviderQonto(models.Model):
def _get_statement_date(self, date_since, date_until):
self.ensure_one()
if self.service != 'qonto':
if self.service != "qonto":
return super()._get_statement_date(
date_since,
date_until,
)
return date_since.astimezone(pytz.timezone('Europe/Paris')).date()
return date_since.astimezone(pytz.timezone("Europe/Paris")).date()
#########
# qonto #
@ -46,85 +46,96 @@ class OnlineBankStatementProviderQonto(models.Model):
def _qonto_header(self):
self.ensure_one()
if self.username and self.password:
return {'Authorization': '%s:%s' % (self.username, self.password)}
raise UserError(_('Please fill login and key'))
return {"Authorization": "%s:%s" % (self.username, self.password)}
raise UserError(_("Please fill login and key"))
def _qonto_get_slug(self):
self.ensure_one()
url = QONTO_ENDPOINT + '/organizations/%7Bid%7D'
url = QONTO_ENDPOINT + "/organizations/%7Bid%7D"
response = requests.get(url, verify=False, headers=self._qonto_header())
if response.status_code == 200:
data = json.loads(response.text)
res = {}
for account in data.get('organization', {}).get('bank_accounts', []):
iban = sanitize_account_number(account.get('iban', ''))
res[iban] = account.get('slug')
for account in data.get("organization", {}).get("bank_accounts", []):
iban = sanitize_account_number(account.get("iban", ""))
res[iban] = account.get("slug")
return res
raise UserError(_('%s \n\n %s') % (response.status_code, response.text))
raise UserError(_("%s \n\n %s") % (response.status_code, response.text))
def _qonto_obtain_transactions(self, slug, date_since, date_until):
self.ensure_one()
url = QONTO_ENDPOINT + '/transactions'
params = {'slug': slug, 'iban': self.account_number}
url = QONTO_ENDPOINT + "/transactions"
params = {"slug": slug, "iban": self.account_number}
if date_since:
params['settled_at_from'] = date_since.replace(
microsecond=0).isoformat() + 'Z'
params["settled_at_from"] = (
date_since.replace(microsecond=0).isoformat() + "Z"
)
if date_until:
params['settled_at_to'] = date_until.replace(
microsecond=0).isoformat() + 'Z'
if date_until == datetime(2022, 1, 1):
date_until = datetime(2021, 12, 31)
params["settled_at_to"] = (
date_until.replace(microsecond=0).isoformat() + "Z"
)
transactions = []
current_page = 1
total_pages = 1
while current_page <= total_pages:
params['current_page'] = current_page
params["current_page"] = current_page
data = self._qonto_get_transactions(url, params)
transactions.extend(data.get('transactions', []))
total_pages = data['meta']['total_pages']
transactions.extend(data.get("transactions", []))
total_pages = data["meta"]["total_pages"]
current_page += 1
return transactions
def _qonto_get_transactions(self, url, params):
response = requests.get(url, verify=False, params=params,
headers=self._qonto_header())
response = requests.get(
url, verify=False, params=params, headers=self._qonto_header()
)
if response.status_code == 200:
return json.loads(response.text)
raise UserError(_('%s \n\n %s') % (response.status_code, response.text))
raise UserError(_("%s \n\n %s") % (response.status_code, response.text))
def _qonto_prepare_statement_line(
self, transaction, sequence, journal_currency, currencies_code2id):
date = datetime.strptime(transaction['settled_at'], '%Y-%m-%dT%H:%M:%S.%fZ')
side = 1 if transaction['side'] == 'credit' else -1
name = transaction['label'] or '/'
if transaction['reference']:
name = '%s %s' % (name, transaction['reference'])
self, transaction, sequence, journal_currency, currencies_code2id
):
date = datetime.strptime(transaction["settled_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
side = 1 if transaction["side"] == "credit" else -1
name = transaction["label"] or "/"
if transaction["reference"]:
name = "%s %s" % (name, transaction["reference"])
vals_line = {
'sequence': sequence,
'date': date,
'name': name,
'ref': transaction['reference'],
'unique_import_id': transaction['transaction_id'],
'amount': transaction['amount'] * side,
"sequence": sequence,
"date": date,
"name": name,
"ref": transaction["reference"],
"unique_import_id": transaction["transaction_id"],
"amount": transaction["amount"] * side,
}
if not transaction['local_currency']:
raise UserError(_(
"Transaction ID %s has not local_currency. "
"This should never happen.") % transaction['transaction_id'])
if transaction['local_currency'] not in currencies_code2id:
raise UserError(_(
"Currency %s used in transaction ID %s doesn't exist "
"in Odoo.") % (
transaction['local_currency'],
transaction['transaction_id']))
if not transaction["local_currency"]:
raise UserError(
_(
"Transaction ID %s has not local_currency. "
"This should never happen."
)
% transaction["transaction_id"]
)
if transaction["local_currency"] not in currencies_code2id:
raise UserError(
_("Currency %s used in transaction ID %s doesn't exist " "in Odoo.")
% (transaction["local_currency"], transaction["transaction_id"])
)
line_currency_id = currencies_code2id[transaction['local_currency']]
line_currency_id = currencies_code2id[transaction["local_currency"]]
if journal_currency.id != line_currency_id:
vals_line.update({
'currency_id': line_currency_id,
'amount_currency': transaction['local_amount'] * side,
})
vals_line.update(
{
"currency_id": line_currency_id,
"amount_currency": transaction["local_amount"] * side,
}
)
return vals_line
def _qonto_obtain_statement_data(self, date_since, date_until):
@ -135,21 +146,23 @@ class OnlineBankStatementProviderQonto(models.Model):
slug = slugs.get(self.account_number)
if not slug:
raise UserError(
_('Qonto : wrong configuration, unknow account %s')
% journal.bank_account_id.acc_number)
_("Qonto : wrong configuration, unknow account %s")
% journal.bank_account_id.acc_number
)
transactions = self._qonto_obtain_transactions(slug, date_since, date_until)
journal_currency = journal.currency_id or journal.company_id.currency_id
all_currencies = self.env['res.currency'].search_read([], ['name'])
currencies_code2id = dict([(x['name'], x['id']) for x in all_currencies])
all_currencies = self.env["res.currency"].search_read([], ["name"])
currencies_code2id = dict([(x["name"], x["id"]) for x in all_currencies])
new_transactions = []
sequence = 0
for transaction in transactions:
sequence += 1
vals_line = self._qonto_prepare_statement_line(
transaction, sequence, journal_currency, currencies_code2id)
transaction, sequence, journal_currency, currencies_code2id
)
new_transactions.append(vals_line)
if new_transactions:

Write Preview
Loading…
Cancel
Save