104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
"""Валидация позиций прайса — правила ТЗ §4.4.
|
|
|
|
Каждое правило возвращает флаги; по их уровню определяется статус позиции:
|
|
ошибка → error, предупреждение → needs_review, иначе → done. Конвертация валют и
|
|
сравнение с прошлой версией (детектор аномалий) тоже здесь.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
|
|
# Ориентировочные курсы к тенге. В проде — курс Нацбанка на дату прайса.
|
|
RATES_TO_KZT: dict[str, float] = {"KZT": 1.0, "USD": 480.0, "RUB": 5.3}
|
|
|
|
ANOMALY_RATIO = 0.5 # изменение цены более чем на 50% к прошлой версии — аномалия
|
|
|
|
|
|
@dataclass
|
|
class Flag:
|
|
"""Замечание валидации по одной позиции."""
|
|
|
|
level: str # info / warning / error
|
|
code: str
|
|
message: str
|
|
|
|
|
|
def validate_name(name: str | None) -> list[Flag]:
|
|
if name and name.strip():
|
|
return []
|
|
return [Flag("error", "empty_name", "пустое название услуги")]
|
|
|
|
|
|
def validate_prices(prices: dict[str, float]) -> list[Flag]:
|
|
flags: list[Flag] = []
|
|
for tier, value in prices.items():
|
|
if value is None or value <= 0:
|
|
flags.append(
|
|
Flag("warning", "price_nonpositive", f"тариф «{tier}»: цена не положительна")
|
|
)
|
|
resident, nonresident = prices.get("resident"), prices.get("nonresident")
|
|
if resident and nonresident and nonresident < resident:
|
|
flags.append(
|
|
Flag("warning", "nonresident_below_resident", "цена нерезидента ниже цены резидента")
|
|
)
|
|
return flags
|
|
|
|
|
|
def validate_date(effective_date: date | None, today: date | None = None) -> list[Flag]:
|
|
if effective_date and today and effective_date > today:
|
|
return [Flag("warning", "future_date", "дата прайса в будущем")]
|
|
return []
|
|
|
|
|
|
def validate_anomaly(new_price: float | None, previous_price: float | None) -> list[Flag]:
|
|
if (
|
|
previous_price
|
|
and new_price
|
|
and abs(new_price - previous_price) / previous_price > ANOMALY_RATIO
|
|
):
|
|
return [
|
|
Flag(
|
|
"warning",
|
|
"price_anomaly",
|
|
f"цена изменилась более чем на {int(ANOMALY_RATIO * 100)}% "
|
|
f"(было {previous_price:g}, стало {new_price:g})",
|
|
)
|
|
]
|
|
return []
|
|
|
|
|
|
def to_kzt(amount: float, currency: str, rates: dict[str, float] | None = None) -> float:
|
|
"""Перевести сумму в тенге по курсу (оригинал сохраняется отдельно)."""
|
|
rates = rates or RATES_TO_KZT
|
|
return amount * rates.get(currency.upper(), 1.0)
|
|
|
|
|
|
def validate_item(
|
|
name: str | None,
|
|
prices: dict[str, float],
|
|
*,
|
|
currency: str = "KZT",
|
|
effective_date: date | None = None,
|
|
previous_resident: float | None = None,
|
|
today: date | None = None,
|
|
) -> tuple[str, list[Flag]]:
|
|
"""Проверить позицию и вернуть (статус, флаги).
|
|
|
|
Статус: error — нельзя сохранять как есть; needs_review — на ручную проверку;
|
|
done — чисто.
|
|
"""
|
|
flags: list[Flag] = []
|
|
flags += validate_name(name)
|
|
flags += validate_prices(prices)
|
|
flags += validate_date(effective_date, today)
|
|
if previous_resident is not None:
|
|
flags += validate_anomaly(prices.get("resident"), previous_resident)
|
|
|
|
if any(f.level == "error" for f in flags):
|
|
return "error", flags
|
|
if any(f.level == "warning" for f in flags):
|
|
return "needs_review", flags
|
|
return "done", flags
|