[16.0][FIX] currency_rate_update_xe: use XE.com JSON API (HTML scraping returns 403)

currency_rate_update_xe/__manifest__.py

@@ -3,7 +3,7 @@
 
 {
     "name": "Currency Rate Update: XE.com",
-    "version": "16.0.1.0.0",
+    "version": "16.0.1.1.0",
     "category": "Financial Management/Configuration",
     "summary": "Update exchange rates using XE.com",
     "author": "Tecnativa, Odoo Community Association (OCA)",

currency_rate_update_xe/models/res_currency_rate_provider_XE.py

@@ -1,14 +1,19 @@
 # Copyright 2023 Tecnativa - Ernesto Tejeda
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
 
-from datetime import date, timedelta
+import base64
+from datetime import date
 
 import requests
-from lxml import etree
 
 from odoo import _, fields, models
 from odoo.exceptions import UserError
 
+# Public credentials embedded in xe.com's own front-end, required to reach its
+# JSON converter endpoint.
+XE_API_URL = "https://www.xe.com/api/protected/midmarket-converter/"
+XE_API_TOKEN = base64.b64encode(b"lodestar:pugsnax").decode()
+
 
 class ResCurrencyRateProviderXE(models.Model):
     _inherit = "res.currency.rate.provider"
@@ -221,51 +226,42 @@ def _obtain_rates(self, base_currency, currencies, date_from, date_to):
         self.ensure_one()
         if self.service != "XE":
             return super()._obtain_rates(base_currency, currencies, date_from, date_to)
-        base_url = "http://www.xe.com/currencytables"
-        if date_from < date.today():
-            return self._get_historical_rate(
-                base_url, currencies, date_from, date_to, base_currency
+        # XE.com blocks automated requests to its public HTML pages
+        # (CloudFront answers 403 regardless of the User-Agent sent), so we
+        # read the JSON endpoint that powers xe.com's own converter instead.
+        # That endpoint only exposes the latest mid-market rates, hence we
+        # always return today's rates regardless of the requested range.
+        api_rates = self._get_xe_rates()
+        base_rate = api_rates.get(base_currency)
+        if not base_rate:
+            raise UserError(
+                _("XE.com didn't return a rate for the base currency %s.")
+                % base_currency
             )
-        else:
-            return self._get_latest_rate(base_url, currencies, base_currency)
-
-    def _get_latest_rate(self, base_url, currencies, base_currency):
-        """Get all the exchange rates for today"""
-        url = f"{base_url}/?from={base_currency}"
-        data = self._request_data(url)
-        return {date.today(): self._parse_data(data, currencies)}
+        rates = {}
+        for currency in currencies:
+            if currency == base_currency:
+                continue
+            rate = api_rates.get(currency)
+            if rate:
+                # API rates are USD-based; make them relative to base_currency.
+                rates[currency] = rate / base_rate
+        return {date.today(): rates}
 
-    def _get_historical_rate(
-        self, base_url, currencies, date_from, date_to, base_currency
-    ):
-        """Get all the exchange rates from 'date_from' to 'date_to'"""
-        content = {}
-        current_date = date_from
-        while current_date <= date_to:
-            url = f"{base_url}/?from={base_currency}&date={current_date.strftime('%Y-%m-%d')}"
-            data = self._request_data(url)
-            content[current_date] = self._parse_data(data, currencies)
-            current_date += timedelta(days=1)
-        return content
-
-    def _request_data(
-        self,
-        url,
-    ):
+    def _get_xe_rates(self):
+        """Return the latest mid-market rates (USD-based) from XE.com."""
         try:
-            return requests.request("GET", url, timeout=10)
+            response = requests.get(
+                XE_API_URL,
+                timeout=30,
+                headers={
+                    "Authorization": "Basic %s" % XE_API_TOKEN,
+                    "User-Agent": "Mozilla/5.0",
+                },
+            )
+            response.raise_for_status()
         except Exception as e:
             raise UserError(
                 _("Couldn't fetch data. Please contact your administrator.")
             ) from e
-
-    def _parse_data(self, data, currencies):
-        result = {}
-        html_elem = etree.fromstring(data.content, etree.HTMLParser())
-        rows_elem = html_elem.xpath(".//div[@id='table-section']//tbody/tr")
-        for row_elem in rows_elem:
-            currency_code = "".join(row_elem.find(".//th").itertext()).strip()
-            if currency_code in currencies:
-                rate = float(row_elem.find("td[2]").text.replace(",", ""))
-                result[currency_code] = rate
-        return result
+        return response.json().get("rates", {})

currency_rate_update_xe/tests/test_currency_rate_update_xe.py

@@ -2,6 +2,8 @@
 # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
 
 
+from dateutil.relativedelta import relativedelta
+
 from odoo import fields
 from odoo.tests import common
 
@@ -35,6 +37,10 @@ def setUpClass(cls):
         cls.CurrencyRate.search([]).unlink()
 
     def test_cron(self):
+        # Pretend the provider already ran yesterday so _scheduled_update
+        # fetches a single day. The XE provider always returns today's
+        # mid-market rates from the API, so exactly one rate (USD) is created.
+        self.xe_provider.last_successful_run = self.today - relativedelta(days=1)
         self.xe_provider._scheduled_update()
         rates = self.CurrencyRate.search([])
         self.assertEqual(len(rates), 1)