100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
from __future__ import annotations
|
||
|
||
import xml.etree.ElementTree as ET
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime
|
||
from decimal import Decimal, InvalidOperation
|
||
|
||
from app.core.config import settings
|
||
from app.integrations.gateway import GatewayClient
|
||
|
||
CNY_CODE = "R01375"
|
||
|
||
|
||
class CbrDataError(RuntimeError):
|
||
"""Ответ ЦБ получен, но разобрать его не удалось."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FxRate:
|
||
rate_date: date
|
||
nominal: int
|
||
value: Decimal # курс за nominal единиц, как публикует ЦБ
|
||
rate_per_unit: Decimal # value / nominal
|
||
|
||
|
||
def _cbr_decimal(text: str | None, what: str) -> Decimal:
|
||
# ЦБ пишет десятичную запятую: "12,7516".
|
||
try:
|
||
return Decimal((text or "").strip().replace(",", "."))
|
||
except InvalidOperation as exc:
|
||
raise CbrDataError(f"ЦБ вернул некорректное значение {what}: {text!r}") from exc
|
||
|
||
|
||
def parse_dynamic(content: bytes) -> list[FxRate]:
|
||
"""XML_dynamic.asp -> список курсов по датам.
|
||
|
||
Парсим байты, а не текст: кодировку (windows-1251) берёт из XML-декларации
|
||
сам парсер. ElementTree — как в SFM-parser: ответ приходит только от ЦБ
|
||
через доверенный шлюз.
|
||
"""
|
||
try:
|
||
root = ET.fromstring(content)
|
||
except ET.ParseError as exc:
|
||
raise CbrDataError(f"Не удалось разобрать XML ЦБ: {exc}") from exc
|
||
if root.tag != "ValCurs":
|
||
raise CbrDataError(f"Неожиданный ответ ЦБ: корневой элемент <{root.tag}>.")
|
||
|
||
rates: list[FxRate] = []
|
||
for record in root.iter("Record"):
|
||
raw_date = record.get("Date", "")
|
||
try:
|
||
rate_date = datetime.strptime(raw_date, "%d.%m.%Y").date()
|
||
except ValueError as exc:
|
||
raise CbrDataError(f"ЦБ вернул некорректную дату: {raw_date!r}") from exc
|
||
nominal = _cbr_decimal(record.findtext("Nominal"), "номинала")
|
||
value = _cbr_decimal(record.findtext("Value"), "курса")
|
||
if nominal <= 0:
|
||
raise CbrDataError(f"ЦБ вернул номинал {nominal} на {raw_date}.")
|
||
rates.append(
|
||
FxRate(
|
||
rate_date=rate_date,
|
||
nominal=int(nominal),
|
||
value=value,
|
||
rate_per_unit=value / nominal,
|
||
)
|
||
)
|
||
return sorted(rates, key=lambda rate: rate.rate_date)
|
||
|
||
|
||
class CbrClient:
|
||
"""Курсы ЦБ через шлюз: {GATEWAY_URL}{CBR_PATH}/scripts/XML_dynamic.asp."""
|
||
|
||
def __init__(self, gateway: GatewayClient) -> None:
|
||
self.gateway = gateway
|
||
|
||
@classmethod
|
||
def from_settings(cls) -> CbrClient:
|
||
gateway = GatewayClient(
|
||
source="cbr",
|
||
base_url=settings.GATEWAY_URL,
|
||
path_prefix=settings.CBR_PATH,
|
||
api_key=settings.CBR_API_KEY,
|
||
timeout_seconds=settings.HTTP_TIMEOUT_SECONDS,
|
||
retries=settings.HTTP_RETRIES,
|
||
)
|
||
return cls(gateway)
|
||
|
||
def get_rates(
|
||
self, date_from: date, date_to: date, code: str = CNY_CODE
|
||
) -> list[FxRate]:
|
||
response = self.gateway.get(
|
||
"/scripts/XML_dynamic.asp",
|
||
{
|
||
"date_req1": f"{date_from:%d/%m/%Y}",
|
||
"date_req2": f"{date_to:%d/%m/%Y}",
|
||
"VAL_NM_RQ": code,
|
||
},
|
||
)
|
||
return parse_dynamic(response.content)
|