Раис Юсупалиев
376 lines
11 KiB
Python
376 lines
11 KiB
Python
"""Pure business rules for request normalization and price ordering."""
|
|
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
from decimal import ROUND_HALF_UP, Decimal
|
|
from typing import Protocol
|
|
|
|
DEFAULT_WEIGHT_ROUND_SCALE = 2
|
|
DEFAULT_PROVIDER_PRICE_MULTIPLIER = Decimal("1")
|
|
MAX_ROUND_SCALE = 6
|
|
DIMENSIONS_ROUND_SCALE = 1
|
|
MIN_WEIGHT_KG = Decimal("0.01")
|
|
MIN_DIMENSION_CM = Decimal("0.1")
|
|
INTEGER_PRICE_QUANTIZER = Decimal("1")
|
|
KOPECKS_IN_RUBLE = Decimal("100")
|
|
DOC_SERVICE_MARKERS = ("документ", "document")
|
|
|
|
_MISSING = object()
|
|
|
|
|
|
class RequestLike(Protocol):
|
|
entity: object
|
|
from_city: object
|
|
to_city: object
|
|
weight_kg: object
|
|
length_cm: object
|
|
width_cm: object
|
|
height_cm: object
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class NormalizedDeliveryRequest:
|
|
entity: str
|
|
from_city: int
|
|
to_city: int
|
|
weight_kg: Decimal
|
|
length_cm: Decimal
|
|
width_cm: Decimal
|
|
height_cm: Decimal
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ProviderPrice:
|
|
provider: str
|
|
service_name: str
|
|
price: Decimal
|
|
currency: str
|
|
delivery_days_min: int
|
|
delivery_days_max: int
|
|
tariff_code: str | None = None
|
|
bypass_parcel_type_filter: bool = False
|
|
|
|
|
|
def normalize_delivery_request(
|
|
request: RequestLike,
|
|
*,
|
|
weight_round_scale: int = DEFAULT_WEIGHT_ROUND_SCALE,
|
|
) -> NormalizedDeliveryRequest:
|
|
"""Return normalized request values according to domain rules."""
|
|
|
|
normalized_weight_scale = _clamp_scale(weight_round_scale)
|
|
|
|
return NormalizedDeliveryRequest(
|
|
entity=str(_get_required_attr(request, "entity")),
|
|
from_city=_normalize_city_id(_get_required_attr(request, "from_city")),
|
|
to_city=_normalize_city_id(_get_required_attr(request, "to_city")),
|
|
weight_kg=_normalize_weight(
|
|
_to_decimal(_get_required_attr(request, "weight_kg")),
|
|
normalized_weight_scale,
|
|
),
|
|
length_cm=_normalize_dimension(
|
|
_to_decimal(_get_required_attr(request, "length_cm")),
|
|
),
|
|
width_cm=_normalize_dimension(
|
|
_to_decimal(_get_required_attr(request, "width_cm")),
|
|
),
|
|
height_cm=_normalize_dimension(
|
|
_to_decimal(_get_required_attr(request, "height_cm")),
|
|
),
|
|
)
|
|
|
|
|
|
def filter_valid_prices(
|
|
prices: Iterable[object],
|
|
*,
|
|
price_multiplier: Decimal = DEFAULT_PROVIDER_PRICE_MULTIPLIER,
|
|
) -> list[ProviderPrice]:
|
|
"""Normalize price records and keep only domain-valid entries."""
|
|
|
|
normalized_multiplier = _normalize_price_multiplier(price_multiplier)
|
|
valid_prices: list[ProviderPrice] = []
|
|
for price_entry in prices:
|
|
normalized_price = _normalize_price(
|
|
price_entry,
|
|
price_multiplier=normalized_multiplier,
|
|
)
|
|
if normalized_price is not None:
|
|
valid_prices.append(normalized_price)
|
|
return valid_prices
|
|
|
|
|
|
def sort_prices_by_price(prices: Iterable[ProviderPrice]) -> list[ProviderPrice]:
|
|
"""Sort provider prices by ascending value."""
|
|
|
|
return sorted(prices, key=lambda price_entry: price_entry.price)
|
|
|
|
|
|
def filter_and_sort_prices(
|
|
prices: Iterable[object],
|
|
*,
|
|
price_multiplier: Decimal = DEFAULT_PROVIDER_PRICE_MULTIPLIER,
|
|
parcel_type: object | None = None,
|
|
) -> list[ProviderPrice]:
|
|
"""Apply domain filtering and return prices ordered by value."""
|
|
|
|
return sort_prices_by_price(
|
|
filter_prices_by_parcel_type(
|
|
filter_valid_prices(prices, price_multiplier=price_multiplier),
|
|
parcel_type=parcel_type,
|
|
)
|
|
)
|
|
|
|
|
|
def calculate_expected_payment_amount_kopecks(
|
|
provider_price: object,
|
|
*,
|
|
price_multiplier: Decimal = DEFAULT_PROVIDER_PRICE_MULTIPLIER,
|
|
) -> int | None:
|
|
"""Return expected payment amount in kopecks for a RUB provider price."""
|
|
|
|
currency = _normalize_text(_get_optional_attr(provider_price, "currency")).upper()
|
|
if currency != "RUB":
|
|
return None
|
|
|
|
price = _try_to_decimal(_get_optional_attr(provider_price, "price"))
|
|
if price is None or not price.is_finite() or price <= 0:
|
|
return None
|
|
|
|
adjusted_price = _apply_price_multiplier_and_round(
|
|
price,
|
|
price_multiplier=_normalize_price_multiplier(price_multiplier),
|
|
)
|
|
if adjusted_price is None:
|
|
return None
|
|
|
|
return int(adjusted_price * KOPECKS_IN_RUBLE)
|
|
|
|
|
|
def is_init_payment_price_valid(
|
|
requested_amount_kopecks: object,
|
|
provider_price: object,
|
|
*,
|
|
price_multiplier: Decimal = DEFAULT_PROVIDER_PRICE_MULTIPLIER,
|
|
) -> bool:
|
|
expected_amount_kopecks = calculate_expected_payment_amount_kopecks(
|
|
provider_price,
|
|
price_multiplier=price_multiplier,
|
|
)
|
|
if expected_amount_kopecks is None:
|
|
return False
|
|
|
|
requested_amount = _try_to_int(requested_amount_kopecks)
|
|
return requested_amount == expected_amount_kopecks
|
|
|
|
|
|
def filter_prices_by_parcel_type(
|
|
prices: Iterable[ProviderPrice],
|
|
*,
|
|
parcel_type: object | None = None,
|
|
) -> list[ProviderPrice]:
|
|
"""Filter aggregated prices according to the requested parcel type."""
|
|
|
|
normalized_parcel_type = _normalize_parcel_type(parcel_type)
|
|
if normalized_parcel_type is None:
|
|
return list(prices)
|
|
|
|
return [
|
|
price
|
|
for price in prices
|
|
if price.bypass_parcel_type_filter
|
|
or _matches_parcel_type(price.service_name, normalized_parcel_type)
|
|
]
|
|
|
|
|
|
def _normalize_price(
|
|
candidate: object,
|
|
*,
|
|
price_multiplier: Decimal,
|
|
) -> ProviderPrice | None:
|
|
provider = _normalize_text(_get_optional_attr(candidate, "provider"))
|
|
service_name = _normalize_text(_get_optional_attr(candidate, "service_name"))
|
|
currency = _normalize_text(_get_optional_attr(candidate, "currency")).upper()
|
|
price_raw = _get_optional_attr(candidate, "price")
|
|
min_days_raw = _get_optional_attr(candidate, "delivery_days_min")
|
|
max_days_raw = _get_optional_attr(candidate, "delivery_days_max")
|
|
|
|
if not provider or not service_name:
|
|
return None
|
|
|
|
if len(currency) != 3 or not currency.isalpha():
|
|
return None
|
|
|
|
price = _try_to_decimal(price_raw)
|
|
if price is None or not price.is_finite() or price <= 0:
|
|
return None
|
|
adjusted_price = _apply_price_multiplier_and_round(
|
|
price,
|
|
price_multiplier=price_multiplier,
|
|
)
|
|
if adjusted_price is None:
|
|
return None
|
|
|
|
min_days = _try_to_int(min_days_raw)
|
|
max_days = _try_to_int(max_days_raw)
|
|
if min_days is None or max_days is None:
|
|
return None
|
|
|
|
if min_days < 0 or max_days < min_days:
|
|
return None
|
|
|
|
return ProviderPrice(
|
|
provider=provider,
|
|
service_name=service_name,
|
|
price=adjusted_price,
|
|
currency=currency,
|
|
delivery_days_min=min_days,
|
|
delivery_days_max=max_days,
|
|
tariff_code=_try_to_str(_get_optional_attr(candidate, "tariff_code")),
|
|
bypass_parcel_type_filter=_extract_bypass_parcel_type_filter(candidate),
|
|
)
|
|
|
|
|
|
def _extract_bypass_parcel_type_filter(candidate: object) -> bool:
|
|
value = _get_optional_attr(candidate, "bypass_parcel_type_filter")
|
|
if value is _MISSING or value is None:
|
|
return False
|
|
return bool(value)
|
|
|
|
|
|
def _normalize_weight(weight: Decimal, scale: int) -> Decimal:
|
|
rounded = _round_half_up(weight, scale)
|
|
if rounded < MIN_WEIGHT_KG:
|
|
return MIN_WEIGHT_KG
|
|
return rounded
|
|
|
|
|
|
def _normalize_dimension(value: Decimal) -> Decimal:
|
|
rounded = _round_half_up(value, DIMENSIONS_ROUND_SCALE)
|
|
if rounded < MIN_DIMENSION_CM:
|
|
return MIN_DIMENSION_CM
|
|
return rounded
|
|
|
|
|
|
def _get_required_attr(source: object, name: str) -> object:
|
|
value = _get_optional_attr(source, name)
|
|
if value is _MISSING:
|
|
raise ValueError(f"Missing required attribute: {name}")
|
|
return value
|
|
|
|
|
|
def _get_optional_attr(source: object, name: str) -> object:
|
|
try:
|
|
return getattr(source, name)
|
|
except AttributeError:
|
|
return _MISSING
|
|
|
|
|
|
def _normalize_text(value: object) -> str:
|
|
if value is _MISSING or value is None:
|
|
return ""
|
|
return str(value).strip()
|
|
|
|
|
|
def _try_to_str(value: object) -> str | None:
|
|
if value is _MISSING or value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|
|
|
|
|
|
def _normalize_city_id(value: object) -> int:
|
|
city_id = _try_to_int(value)
|
|
if city_id is None:
|
|
raise ValueError(f"City identifier must be an integer: {value!r}")
|
|
return city_id
|
|
|
|
|
|
def _clamp_scale(scale: int) -> int:
|
|
if scale < 0:
|
|
return 0
|
|
if scale > MAX_ROUND_SCALE:
|
|
return MAX_ROUND_SCALE
|
|
return scale
|
|
|
|
|
|
def _round_half_up(value: Decimal, scale: int) -> Decimal:
|
|
quantizer = Decimal("1").scaleb(-scale)
|
|
return value.quantize(quantizer, rounding=ROUND_HALF_UP)
|
|
|
|
|
|
def _normalize_price_multiplier(value: Decimal) -> Decimal:
|
|
multiplier = _try_to_decimal(value)
|
|
if multiplier is None or not multiplier.is_finite() or multiplier <= 0:
|
|
raise ValueError(f"Price multiplier must be a positive finite Decimal: {value!r}")
|
|
return multiplier
|
|
|
|
|
|
def _normalize_parcel_type(value: object) -> str | None:
|
|
normalized = _normalize_text(value).casefold()
|
|
if normalized in {"doc", "parcel"}:
|
|
return normalized
|
|
return None
|
|
|
|
|
|
def _matches_parcel_type(service_name: str, parcel_type: str) -> bool:
|
|
is_document_tariff = _contains_document_marker(service_name)
|
|
if parcel_type == "doc":
|
|
return is_document_tariff
|
|
return not is_document_tariff
|
|
|
|
|
|
def _contains_document_marker(service_name: str) -> bool:
|
|
normalized_service_name = service_name.casefold()
|
|
return any(
|
|
marker in normalized_service_name
|
|
for marker in DOC_SERVICE_MARKERS
|
|
)
|
|
|
|
|
|
def _apply_price_multiplier_and_round(
|
|
price: Decimal,
|
|
*,
|
|
price_multiplier: Decimal,
|
|
) -> Decimal | None:
|
|
adjusted_price = (price * price_multiplier).quantize(
|
|
INTEGER_PRICE_QUANTIZER,
|
|
rounding=ROUND_HALF_UP,
|
|
)
|
|
if not adjusted_price.is_finite() or adjusted_price <= 0:
|
|
return None
|
|
return adjusted_price
|
|
|
|
|
|
def _to_decimal(value: object) -> Decimal:
|
|
converted = _try_to_decimal(value)
|
|
if converted is None:
|
|
raise ValueError(f"Value cannot be converted to Decimal: {value!r}")
|
|
return converted
|
|
|
|
|
|
def _try_to_decimal(value: object) -> Decimal | None:
|
|
if isinstance(value, Decimal):
|
|
return value
|
|
if value is _MISSING:
|
|
return None
|
|
try:
|
|
return Decimal(str(value))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _try_to_int(value: object) -> int | None:
|
|
if isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, int):
|
|
return value
|
|
if value is _MISSING:
|
|
return None
|
|
try:
|
|
converted = int(value)
|
|
except Exception:
|
|
return None
|
|
if Decimal(str(value)) != Decimal(converted):
|
|
return None
|
|
return converted
|