Replace forex-python and fetch rates directly from EU website

https://theforexapi.com/ is frequently down which breaks the tests. The EU
website is more reliable.
As forex-python is just also used to set the symbol of currencies when
importing, we can just fill the symbol when it is empty with the alpha-3 code.

issue11731
review439631003
(grafted from 2097ae9311186817cd70c172c478d102435ad2bf)

--HG--
branch : 6.0

currency.py

@@ -10,12 +10,6 @@ from dateutil.relativedelta import relativedelta
 from sql import Window
 from sql.functions import NthValue
 
-try:
-    from forex_python.converter import CurrencyRates, RatesNotAvailableError
-    get_rates = CurrencyRates(force_decimal=True).get_rates
-except ImportError:
-    CurrencyRates = get_rates = None
-
 from trytond.i18n import gettext
 from trytond.model import (
     ModelView, ModelSQL, DeactivableMixin, fields, Unique, Check, SymbolMixin)
@@ -24,6 +18,7 @@ from trytond.pool import Pool
 from trytond.rpc import RPC
 from trytond.pyson import Eval, If
 
+from .ecb import RatesNotAvailableError, get_rates
 from .exceptions import RateError
 from .ir import rate_decimal
 
@@ -299,7 +294,8 @@ class Cron(ModelSQL, ModelView):
     __name__ = 'currency.cron'
 
     source = fields.Selection(
-        [], "Source", required=True,
+        [('ecb', "European Central Bank")],
+        "Source", required=True,
         help="The external source for rates.")
     frequency = fields.Selection([
             ('daily', "Daily"),
@@ -339,8 +335,6 @@ class Cron(ModelSQL, ModelView):
         cls._buttons.update({
                 'run': {},
                 })
-        if CurrencyRates:
-            cls.source.selection.append(('ecb', "European Central Bank"))
 
     @classmethod
     def default_frequency(cls):

ecb.py

@@ -0,0 +1,98 @@
+# This file is part of Tryton.  The COPYRIGHT file at the top level of
+# this repository contains the full copyright notices and license terms.
+
+import datetime as dt
+import ssl
+import sys
+
+try:
+    from lxml import etree as ET
+except ImportError:
+    try:
+        import xml.etree.cElementTree as ET
+    except ImportError:
+        import xml.etree.ElementTree as ET
+
+from decimal import Decimal
+from urllib.error import HTTPError
+from urllib.request import urlopen
+
+_URL = 'https://www.ecb.europa.eu/stats/eurofxref/'
+_URL_TODAY = _URL + 'eurofxref-daily.xml'
+_URL_90 = _URL + 'eurofxref-hist-90d.xml'
+_URL_HIST = _URL + 'eurofxref-hist.xml'
+_START_DATE = dt.date(1999, 1, 4)
+_CUBE_TAG = '{http://www.ecb.int/vocabulary/2002-08-01/eurofxref}Cube'
+
+
+class RatesNotAvailableError(Exception):
+    pass
+
+
+class UnsupportedCurrencyError(Exception):
+    pass
+
+
+def _parse_time(time):
+    return dt.datetime.strptime(time, '%Y-%m-%d').date()
+
+
+def _find_time(source, time=None):
+    for _, element in ET.iterparse(source):
+        if element.tag == _CUBE_TAG and 'time' in element.attrib:
+            if time and _parse_time(element.attrib.get('time')) <= time:
+                return element
+            elif time is None:
+                return element
+            element.clear()
+
+
+def get_rates(currency='EUR', date=None):
+    if date is None:
+        date = dt.date.today()
+    if date < _START_DATE:
+        date = _START_DATE
+    context = ssl.create_default_context()
+    try:
+        with urlopen(_URL_TODAY, context=context) as response:
+            element = _find_time(response)
+            last_date = _parse_time(element.attrib['time'])
+            if last_date < date:
+                raise RatesNotAvailableError()
+            elif last_date == date:
+                return _compute_rates(element, currency, date)
+
+        if last_date - date < dt.timedelta(days=90):
+            url = _URL_90
+        else:
+            url = _URL_HIST
+        with urlopen(url, context=context) as response:
+            element = _find_time(response, date)
+            return _compute_rates(element, currency, date)
+    except HTTPError as e:
+        raise RatesNotAvailableError() from e
+
+
+def _compute_rates(element, currency, date):
+    currencies = {}
+    for cur in element:
+        currencies[cur.attrib['currency']] = Decimal(cur.attrib['rate'])
+    if currency != 'EUR':
+        currencies['EUR'] = Decimal(1)
+        try:
+            base_rate = currencies.pop(currency)
+        except KeyError:
+            raise UnsupportedCurrencyError(f'{currency} is not available')
+        for cur, rate in currencies.items():
+            currencies[cur] = (rate / base_rate).quantize(Decimal('.0001'))
+    return currencies
+
+
+if __name__ == '__main__':
+    currency = 'EUR'
+    if len(sys.argv) > 1:
+        currency = sys.argv[1]
+    date = None
+    if len(sys.argv) > 2:
+        date = dt.datetime.strptime(sys.argv[2], '%Y-%m-%d').date()
+    print(get_rates(currency, date))

scripts/import_currencies.py

@@ -7,7 +7,6 @@ import sys
 from argparse import ArgumentParser
 
 import pycountry
-from forex_python.converter import CurrencyCodes
 
 try:
     from progressbar import ProgressBar, Bar, ETA, SimpleProgress
@@ -53,7 +52,6 @@ def get_currencies():
 def update_currencies(currencies):
     print("Update currencies", file=sys.stderr)
     Currency = Model.get('currency.currency')
-    codes = CurrencyCodes()
 
     records = []
     for currency in _progress(pycountry.currencies):
@@ -61,10 +59,11 @@ def update_currencies(currencies):
         if code in currencies:
             record = currencies[code]
         else:
-            record = Currency(code=code)
+            record = Currency(code=code, symbol=None)
         record.name = _remove_forbidden_chars(currency.name)
         record.numeric_code = currency.numeric
-        record.symbol = codes.get_symbol(currency.alpha_3) or currency.alpha_3
+        if not record.symbol:
+            record.symbol = currency.alpha_3
         records.append(record)
 
     Currency.save(records)

setup.py

@@ -66,7 +66,7 @@ for dep in info.get('depends', []):
         requires.append(get_require_version('trytond_%s' % dep))
 requires.append(get_require_version('trytond'))
 
-tests_require = [get_require_version('proteus'), 'pycountry', 'forex-python']
+tests_require = [get_require_version('proteus'), 'pycountry']
 dependency_links = []
 if minor_version % 2:
     dependency_links.append(
@@ -143,9 +143,7 @@ setup(name=name,
     python_requires='>=3.6',
     install_requires=requires,
     extras_require={
-        'data': [
-            'pycountry', 'forex-python', get_require_version('proteus')],
-        'ecb_rate': ['forex-python'],
+        'data': ['pycountry', get_require_version('proteus')],
         },
     dependency_links=dependency_links,
     zip_safe=False,

tests/test_currency.py

@@ -5,6 +5,8 @@ import doctest
 import unittest
 from decimal import Decimal
 import trytond.tests.test_tryton
+from trytond.modules.currency.ecb import (
+    RatesNotAvailableError, UnsupportedCurrencyError, get_rates)
 from trytond.transaction import Transaction
 from trytond.tests.test_tryton import ModuleTestCase, with_transaction
 from trytond.tests.test_tryton import doctest_teardown
@@ -368,10 +370,81 @@ class CurrencyTestCase(ModuleTestCase):
         self.assertSetEqual(data, result)
 
 
+class ECBtestCase(unittest.TestCase):
+
+    def test_rate_EUR_week_ago(self):
+        "Fetch EUR rates a week ago"
+        week_ago = datetime.date.today() - datetime.timedelta(days=7)
+
+        rates = get_rates('EUR', week_ago)
+
+        self.assertNotIn('EUR', rates)
+        self.assertIn('USD', rates)
+
+    def test_rate_USD_week_ago(self):
+        "Fetch USD rates a week ago"
+        week_ago = datetime.date.today() - datetime.timedelta(days=7)
+
+        rates = get_rates('USD', week_ago)
+
+        self.assertIn('EUR', rates)
+        self.assertNotIn('USD', rates)
+
+    def test_rate_EUR_on_weekend(self):
+        "Fetch EUR rates on weekend"
+
+        rates_fr = get_rates('EUR', datetime.date(2022, 9, 30))
+        rates_sa = get_rates('EUR', datetime.date(2022, 10, 2))
+        rates_su = get_rates('EUR', datetime.date(2022, 10, 2))
+
+        self.assertEqual(rates_sa, rates_fr)
+        self.assertEqual(rates_su, rates_fr)
+
+    def test_rate_USD_on_weekend(self):
+        "Fetch USD rates on weekend"
+
+        rates_fr = get_rates('USD', datetime.date(2022, 9, 30))
+        rates_sa = get_rates('USD', datetime.date(2022, 10, 2))
+        rates_su = get_rates('USD', datetime.date(2022, 10, 2))
+
+        self.assertEqual(rates_sa, rates_fr)
+        self.assertEqual(rates_su, rates_fr)
+
+    def test_rate_EUR_start_date(self):
+        "Fetch EUR rates at start date"
+
+        rates = get_rates('EUR', datetime.date(1999, 1, 4))
+
+        self.assertEqual(rates['USD'], Decimal('1.1789'))
+
+    def test_rate_USD_start_date(self):
+        "Fetch USD rates at start date"
+
+        rates = get_rates('USD', datetime.date(1999, 1, 4))
+
+        self.assertEqual(rates['EUR'], Decimal('0.8482'))
+
+    def test_rate_in_future(self):
+        "Fetch rate in future raise an error"
+        future = datetime.date.today() + datetime.timedelta(days=2)
+
+        with self.assertRaises(RatesNotAvailableError):
+            get_rates('USD', future)
+
+    def test_rate_unsupported_currency(self):
+        "Fetch rate for unsupported currency"
+        date = datetime.date.today() - datetime.timedelta(days=2)
+
+        with self.assertRaises(UnsupportedCurrencyError):
+            get_rates('XXX', date)
+
+
 def suite():
     suite = trytond.tests.test_tryton.suite()
     suite.addTests(unittest.TestLoader().loadTestsFromTestCase(
             CurrencyTestCase))
+    suite.addTests(unittest.TestLoader().loadTestsFromTestCase(
+            ECBtestCase))
     suite.addTests(doctest.DocFileSuite(
             'scenario_currency_compute.rst',
             tearDown=doctest_teardown, encoding='utf-8',