Joey
59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""汇率服务：CNY→RUB。数据源三级降级（FloatRates → 俄央行 → 兜底），服务端缓存。"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import httpx
|
|
|
|
_FALLBACK_RATE = 11.5
|
|
_MIN, _MAX = 5.0, 25.0
|
|
_CACHE_TTL = 3600 # 秒
|
|
|
|
_cache: dict = {"rate": None, "source": "", "ts": 0.0}
|
|
|
|
|
|
def _valid(rate: float) -> bool:
|
|
return _MIN <= rate <= _MAX
|
|
|
|
|
|
async def _fetch_floatrates() -> float | None:
|
|
async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
|
|
resp = await client.get("https://www.floatrates.com/daily/cny.json")
|
|
resp.raise_for_status()
|
|
rub = resp.json().get("rub", {})
|
|
rate = rub.get("rate")
|
|
return float(rate) if rate else None
|
|
|
|
|
|
async def _fetch_cbr() -> float | None:
|
|
async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
|
|
resp = await client.get("https://www.cbr-xml-daily.ru/daily_json.js")
|
|
resp.raise_for_status()
|
|
cny = resp.json().get("Valute", {}).get("CNY", {})
|
|
value = cny.get("Value")
|
|
return float(value) if value else None
|
|
|
|
|
|
async def get_fx_rate() -> dict:
|
|
"""返回 {rate, source, updated_at}。带 1 小时内存缓存。"""
|
|
now = time.time()
|
|
if _cache["rate"] and (now - _cache["ts"]) < _CACHE_TTL:
|
|
return dict(_cache)
|
|
|
|
rate = None
|
|
source = ""
|
|
for name, fn in (("floatrates", _fetch_floatrates), ("cbr", _fetch_cbr)):
|
|
try:
|
|
r = await fn()
|
|
if r is not None and _valid(r):
|
|
rate, source = r, name
|
|
break
|
|
except Exception: # noqa: BLE001 - 数据源失败降级
|
|
continue
|
|
|
|
if rate is None:
|
|
rate, source = _FALLBACK_RATE, "fallback"
|
|
|
|
_cache.update({"rate": rate, "source": source, "ts": now})
|
|
return dict(_cache)
|