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)