import frappe import requests from frappe import _ from crm.fcrm.doctype.fcrm_settings.fcrm_settings import FCRMSettings @frappe.whitelist() def get_exchange_rate(from_currency: str, to_currency: str, date: str | None = None): if not date: date = "latest" # "latest" is keyed by today's date so tomorrow's call automatically misses the cache cache_date = frappe.utils.today() if date == "latest" else date cache_key = f"exchange_rate_{from_currency}_{to_currency}_{cache_date}" cached_rate = frappe.cache().get_value(cache_key) if cached_rate is not None: return cached_rate rate, api_used = _fetch_exchange_rate(from_currency, to_currency, date) if rate is not None: frappe.cache().set_value(cache_key, rate) return rate _raise_exchange_rate_error(from_currency, to_currency, date, api_used) def _fetch_exchange_rate(from_currency: str, to_currency: str, date: str): settings = frappe.get_single("FCRM Settings") provider = settings.service_provider # Paid providers — no fallback, fail explicitly if provider == "exchangerate.host": return _fetch_from_exchangerate_host(settings, from_currency, to_currency, date), provider if provider == "exchangerate-api": return _fetch_from_exchangerate_api(settings, from_currency, to_currency), provider # Free providers — try both as fallbacks, regardless of which is "primary" if provider == "frankfurter.app": rate = _fetch_from_frankfurter(from_currency, to_currency, date) if rate is not None: return rate, provider # Frankfurter is down — silently fall back to fawaz rate = _fetch_from_fawaz_api(from_currency, to_currency, date) return rate, "fawazahmed-exchange-api" if provider == "fawazahmed-exchange-api": rate = _fetch_from_fawaz_api(from_currency, to_currency, date) if rate is not None: return rate, provider # fawaz is down — silently fall back to frankfurter rate = _fetch_from_frankfurter(from_currency, to_currency, date) return rate, "frankfurter" # Unknown provider — try both free ones rate = _fetch_from_frankfurter(from_currency, to_currency, date) if rate is not None: return rate, "frankfurter" return _fetch_from_fawaz_api(from_currency, to_currency, date), "fawazahmed-exchange-api" def _fetch_from_frankfurter(from_currency: str, to_currency: str, date: str): try: res = requests.get( f"https://api.frankfurter.app/{date}?from={from_currency}&to={to_currency}", timeout=5 ) if res.ok: return res.json().get("rates", {}).get(to_currency) except Exception: pass return None def _fetch_from_fawaz_api(from_currency: str, to_currency: str, date: str): from_lower = from_currency.lower() to_lower = to_currency.lower() date_str = "latest" if date == "latest" else date urls = [ f"https://cdn.jsdelivr.net/npm/@fawazahmed0/currency-api@{date_str}/v1/currencies/{from_lower}.json", f"https://{date_str}.currency-api.pages.dev/v1/currencies/{from_lower}.json", ] for url in urls: try: res = requests.get(url, timeout=5) if res.ok: return res.json()[from_lower][to_lower] except Exception: continue return None def _fetch_from_exchangerate_host(settings: FCRMSettings, from_currency: str, to_currency: str, date: str): if not settings.access_key: frappe.throw( _("Access Key is required for Service Provider: {0}").format( frappe.bold(settings.service_provider) ) ) params = {"access_key": settings.access_key, "from": from_currency, "to": to_currency, "amount": 1} if date != "latest": params["date"] = date res = requests.get("https://api.exchangerate.host/convert", params=params, timeout=5) if res.ok: return res.json()["result"] return None def _fetch_from_exchangerate_api(settings: FCRMSettings, from_currency: str, to_currency: str): if not settings.access_key: frappe.throw( _("Access Key is required for Service Provider: {0}").format( frappe.bold(settings.service_provider) ) ) res = requests.get( f"https://v6.exchangerate-api.com/v6/{settings.access_key}/pair/{from_currency}/{to_currency}", timeout=5, ) if res.ok: data = res.json() if data["result"] == "success": return data["conversion_rate"] return None def _raise_exchange_rate_error(from_currency: str, to_currency: str, date: str, api_used: str): frappe.log_error( title="Exchange Rate Fetch Error", message=f"Failed to fetch exchange rate from {from_currency} to {to_currency} using {api_used} API.", ) if api_used == "frankfurter": user = frappe.session.user is_manager = ( "System Manager" in frappe.get_roles(user) or "Sales Manager" in frappe.get_roles(user) or user == "Administrator" ) if not is_manager: frappe.throw( _( "Ask your manager to set up the Exchange Rate Provider, as default provider does not support currency conversion for {0} to {1}." ).format(from_currency, to_currency) ) frappe.throw( _( "Setup the Exchange Rate Provider other than 'Frankfurter' in settings, as default provider does not support currency conversion for {0} to {1}." ).format(from_currency, to_currency) ) frappe.throw( _( "Failed to fetch exchange rate from {0} to {1} on {2}. Please check your internet connection or try again later." ).format(from_currency, to_currency, date) )