Files
g2s-aggregator/app/adapters/delivery_providers/cse/constants.py
T

73 lines
2.4 KiB
Python

"""CSE reference constants and geography resolution.
CSE identifies geography, cargo type and currency by GUID. Geography GUIDs per
city live in ``cities_map`` under the ``cse`` block (see task 037). Cargo-type
GUIDs come from ``GetReferenceData: TypesOfCargo`` and must be configured with
real values before production use.
"""
from app.adapters.delivery_providers.cse.errors import CSERequestError
from app.cities import cities_map
# Provider name surfaced in unified DeliveryPrice/SystemDataTariff.
CSE_PROVIDER_NAME = "cse"
# CSE returns currency short names (e.g. "RUR"); normalize to ISO-4217.
_CURRENCY_NAME_TO_CODE = {
"RUR": "RUB",
"RUB": "RUB",
"USD": "USD",
"EUR": "EUR",
}
# TypeOfCargo GUIDs by parcel type (GetReferenceData: TypesOfCargo).
TYPE_OF_CARGO_BY_PARCEL_TYPE: dict[str, str] = {
"doc": "81dd8a13-8235-494f-84fd-9c04c51d50ec", # Документы
"parcel": "4aab1fc6-fc2b-473a-8728-58bcd4ff79ba", # Груз
}
# Cargo type used when the parcel type is not known (price calculation flow).
DEFAULT_TYPE_OF_CARGO: str = "4aab1fc6-fc2b-473a-8728-58bcd4ff79ba" # Груз
def normalize_currency(currency_name: str | None) -> str:
if not currency_name:
return "RUB"
return _CURRENCY_NAME_TO_CODE.get(currency_name.strip().upper(), "RUB")
def resolve_type_of_cargo(parcel_type: str | None) -> str:
cargo_guid = (
TYPE_OF_CARGO_BY_PARCEL_TYPE.get(parcel_type)
if parcel_type is not None
else DEFAULT_TYPE_OF_CARGO
)
if not cargo_guid:
raise CSERequestError(
f"CSE cargo type is not configured for parcel type {parcel_type!r}."
)
return cargo_guid
def resolve_cse_geography(city_id: int) -> str:
"""Resolve CSE geography GUID from cities_map by city identifier."""
city_entry = cities_map.get(str(city_id))
if not isinstance(city_entry, dict):
raise CSERequestError(
f"CSE geography is not configured for city id {city_id}."
)
cse_data = city_entry.get("cse")
if not isinstance(cse_data, dict):
raise CSERequestError(
f"CSE geography is not configured for city id {city_id}."
)
geography_uid = cse_data.get("geography_uid")
if not isinstance(geography_uid, str) or not geography_uid:
raise CSERequestError(
f"CSE geography is not configured for city id {city_id}."
)
return geography_uid