Skip to content

Commit

Permalink
[IMP] *_online_ponto: forward port 12.0 improvements
Browse files Browse the repository at this point in the history 
Separate retrieval of data from ponto (buffer data) and
creation of statements.
  • Loading branch information
@NL66278
NL66278 committed Jan 24, 2023
1 parent fa59c44 commit c32a65f
Show file tree
Hide file tree
Showing 11 changed files with 433 additions and 507 deletions.
13 changes: 8 additions & 5 deletions account_statement_import_online_ponto/__manifest__.py
View file
Original file line number Diff line number Diff line change
@@ -1,14 +1,17 @@
# Copyright 2020 Florent de Labarre
# Copyright 2020 Tecnativa - João Marques
# Copyright 2020 Florent de Labarre.
# Copyright 2020 Tecnativa - João Marques.
# Copyright 2022-2023 Therp BV <https://therp.nl>.
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
{
"name": "Online Bank Statements: MyPonto.com",
"version": "14.0.1.1.0",
"version": "14.0.2.0.0",
"category": "Account",
"website": "https://github.com/OCA/bank-statement-import",
"author": "Florent de Labarre, Odoo Community Association (OCA)",
"author": "Florent de Labarre, Therp BV, Odoo Community Association (OCA)",
"license": "AGPL-3",
"installable": True,
"depends": ["account_statement_import_online"],
"data": ["view/online_bank_statement_provider.xml"],
"data": [
"views/online_bank_statement_provider.xml",
],
}
2 changes: 1 addition & 1 deletion account_statement_import_online_ponto/models/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).

from . import ponto_interface
from . import online_bank_statement_provider_ponto
261 changes: 70 additions & 191 deletions account_statement_import_online_ponto/models/online_bank_statement_provider_ponto.py
View file
Original file line number Diff line number Diff line change
@@ -1,246 +1,125 @@
# Copyright 2020 Florent de Labarre
# Copyright 2022-2023 Therp BV <https://therp.nl>.
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).

import base64
import json
import logging
import re
import time
from datetime import datetime

import pytz
import requests
from dateutil.relativedelta import relativedelta

from odoo import _, api, fields, models
from odoo.exceptions import UserError

from odoo.addons.base.models.res_bank import sanitize_account_number
from odoo import _, api, models

PONTO_ENDPOINT = "https://api.myponto.com"
_logger = logging.getLogger(__name__)


class OnlineBankStatementProviderPonto(models.Model):
_inherit = "online.bank.statement.provider"

ponto_token = fields.Char(readonly=True)
ponto_token_expiration = fields.Datetime(readonly=True)
ponto_last_identifier = fields.Char(readonly=True)

def ponto_reset_last_identifier(self):
self.write({"ponto_last_identifier": False})

@api.model
def _get_available_services(self):
"""Each provider model must register its service."""
return super()._get_available_services() + [
("ponto", "MyPonto.com"),
]

def _obtain_statement_data(self, date_since, date_until):
"""Check wether called for ponto servide, otherwise pass the buck."""
self.ensure_one()
if self.service != "ponto":
return super()._obtain_statement_data(date_since, date_until)
return self._ponto_obtain_statement_data(date_since, date_until)

#########
# ponto #
#########

def _ponto_header_token(self):
self.ensure_one()
if self.username and self.password:
login = "{}:{}".format(self.username, self.password)
login = base64.b64encode(login.encode("UTF-8")).decode("UTF-8")
return {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
"Authorization": "Basic %s" % login,
}
raise UserError(_("Please fill login and key."))

def _ponto_header(self):
self.ensure_one()
if (
not self.ponto_token
or not self.ponto_token_expiration
or self.ponto_token_expiration <= fields.Datetime.now()
):

url = PONTO_ENDPOINT + "/oauth2/token"
response = requests.post(
url,
params={"grant_type": "client_credentials"},
headers=self._ponto_header_token(),
if self.service != "ponto": # pragma: no cover
return super()._obtain_statement_data(
date_since,
date_until,
)
if response.status_code == 200:
data = json.loads(response.text)
access_token = data.get("access_token", False)
if not access_token:
raise UserError(_("Ponto : no token"))
else:
self.sudo().ponto_token = access_token
expiration_date = fields.Datetime.now() + relativedelta(
seconds=data.get("expires_in", False)
)
self.sudo().ponto_token_expiration = expiration_date
else:
raise UserError(_("%s \n\n %s") % (response.status_code, response.text))
return {
"Accept": "application/json",
"Authorization": "Bearer %s" % self.ponto_token,
}

def _ponto_get_account_ids(self):
url = PONTO_ENDPOINT + "/accounts"
response = requests.get(
url, params={"limit": 100}, headers=self._ponto_header()
)
if response.status_code == 200:
data = json.loads(response.text)
res = {}
for account in data.get("data", []):
iban = sanitize_account_number(
account.get("attributes", {}).get("reference", "")
)
res[iban] = account.get("id")
return res
raise UserError(_("%s \n\n %s") % (response.status_code, response.text))

def _ponto_synchronisation(self, account_id):
url = PONTO_ENDPOINT + "/synchronizations"
data = {
"data": {
"type": "synchronization",
"attributes": {
"resourceType": "account",
"resourceId": account_id,
"subtype": "accountTransactions",
},
}
}
response = requests.post(url, headers=self._ponto_header(), json=data)
if response.status_code in (200, 201, 400):
data = json.loads(response.text)
sync_id = data.get("attributes", {}).get("resourceId", False)
else:
raise UserError(
_("Error during Create Synchronisation %s \n\n %s")
% (response.status_code, response.text)
)

# Check synchronisation
if not sync_id:
return
url = PONTO_ENDPOINT + "/synchronizations/" + sync_id
number = 0
while number == 100:
number += 1
response = requests.get(url, headers=self._ponto_header())
if response.status_code == 200:
data = json.loads(response.text)
status = data.get("status", {})
if status in ("success", "error"):
return
time.sleep(4)

def _ponto_get_transaction(self, account_id, date_since, date_until):
page_url = PONTO_ENDPOINT + "/accounts/" + account_id + "/transactions"
params = {"limit": 100}
page_next = True
last_identifier = self.ponto_last_identifier
if last_identifier:
params["before"] = last_identifier
page_next = False
transaction_lines = []
latest_identifier = False
while page_url:
response = requests.get(
page_url, params=params, headers=self._ponto_header()
)
if response.status_code != 200:
raise UserError(
_("Error during get transaction.\n\n%s \n\n %s")
% (response.status_code, response.text)
)
if params.get("before"):
params.pop("before")
data = json.loads(response.text)
links = data.get("links", {})
if page_next:
page_url = links.get("next", False)
else:
page_url = links.get("prev", False)
transactions = data.get("data", [])
if transactions:
current_transactions = []
for transaction in transactions:
date = self._ponto_date_from_string(
transaction.get("attributes", {}).get("executionDate")
)
if date_since <= date < date_until:
current_transactions.append(transaction)
if current_transactions:
if not page_next or (page_next and not latest_identifier):
latest_identifier = current_transactions[0].get("id")
transaction_lines.extend(current_transactions)
if latest_identifier:
self.ponto_last_identifier = latest_identifier
return transaction_lines

def _ponto_date_from_string(self, date_str):
"""Dates in Ponto are expressed in UTC, so we need to convert them
to supplied tz for proper classification.
"""
dt = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S.%fZ")
dt = dt.replace(tzinfo=pytz.utc).astimezone(pytz.timezone(self.tz or "utc"))
return dt.replace(tzinfo=None)
return self._ponto_obtain_statement_data(date_since, date_until)

def _ponto_obtain_statement_data(self, date_since, date_until):
"""Translate information from Ponto to Odoo bank statement lines."""
self.ensure_one()
account_ids = self._ponto_get_account_ids()
journal = self.journal_id
iban = self.account_number
account_id = account_ids.get(iban)
if not account_id:
raise UserError(
_("Ponto : wrong configuration, unknow account %s")
% journal.bank_account_id.acc_number
)
self._ponto_synchronisation(account_id)
transaction_lines = self._ponto_get_transaction(
account_id, date_since, date_until
_logger.debug(
_("Ponto obtain statement data for journal %s from %s to %s"),
self.journal_id.name,
date_since,
date_until,
)
lines = self._ponto_retrieve_data(date_since)
new_transactions = []
sequence = 0
for transaction in transaction_lines:
for transaction in lines:
date = self._ponto_get_execution_datetime(transaction)
if date < date_since or date > date_until:
continue
sequence += 1
vals_line = self._ponto_get_transaction_vals(transaction, sequence)
new_transactions.append(vals_line)
if new_transactions:
return new_transactions, {}
return
return new_transactions, {}

def _ponto_retrieve_data(self, date_since):
"""Fill buffer with data from Ponto.
We will retrieve data from the latest transactions present in Ponto
backwards, until we find data that has an execution date before date_since.
"""
lines = []
interface_model = self.env["ponto.interface"]
access_data = interface_model._login(self.username, self.password)
interface_model._set_access_account(access_data, self.account_number)
latest_identifier = False
transactions = interface_model._get_transactions(access_data, latest_identifier)
while transactions:
lines.extend(transactions)
latest_identifier = transactions[-1].get("id")
earliest_datetime = self._ponto_get_execution_datetime(transactions[-1])
if earliest_datetime < date_since:
break
transactions = interface_model._get_transactions(
access_data, latest_identifier
)
return lines

def _ponto_get_transaction_vals(self, transaction, sequence):
"""Translate information from Ponto to statement line vals."""
attributes = transaction.get("attributes", {})
ref_list = [
attributes.get(x)
for x in {"description", "counterpartName", "counterpartReference"}
for x in {
"description",
"counterpartName",
"counterpartReference",
}
if attributes.get(x)
]
ref = " ".join(ref_list)
date = self._ponto_date_from_string(attributes.get("executionDate"))
date = self._ponto_get_execution_datetime(transaction)
vals_line = {
"sequence": sequence,
"date": date,
"ref": re.sub(" +", " ", ref) or "/",
"payment_ref": attributes.get("remittanceInformation", ref),
"unique_import_id": transaction["id"],
"amount": attributes["amount"],
"raw_data": transaction,
"raw_data": json.dumps(transaction),
}
if attributes.get("counterpartReference"):
vals_line["account_number"] = attributes["counterpartReference"]
if attributes.get("counterpartName"):
vals_line["partner_name"] = attributes["counterpartName"]
return vals_line

def _ponto_get_execution_datetime(self, transaction):
"""Get execution datetime for a transaction.
Odoo often names variables containing date and time just xxx_date or
date_xxx. We try to avoid this misleading naming by using datetime as
much for variables and fields of type datetime.
"""
attributes = transaction.get("attributes", {})
return self._ponto_datetime_from_string(attributes.get("executionDate"))

def _ponto_datetime_from_string(self, date_str):
"""Dates in Ponto are expressed in UTC, so we need to convert them
to supplied tz for proper classification.
"""
dt = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S.%fZ")
dt = dt.replace(tzinfo=pytz.utc).astimezone(pytz.timezone(self.tz or "utc"))
return dt.replace(tzinfo=None)
Loading

0 comments on commit c32a65f

Please sign in to comment.