rate configurator/ purchase of conditions/ fixed payments/ refactoring/ major improvements and fixes

core/settings/tariffs_config.py

@@ -0,0 +1,192 @@
+from math import ceil
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from database.models import Setting
+
+
+TARIFFS_CONFIG: dict[str, Any] = {
+    "ALLOW_DOWNGRADE": True,
+    "KEY_ADDONS_PACK_MODE": "all",
+}
+
+
+async def load_tariffs_config(session: AsyncSession) -> None:
+    """Загружает конфиг тарифов из БД."""
+    stmt = select(Setting).where(Setting.key == "TARIFFS_CONFIG")
+    result = await session.execute(stmt)
+    setting = result.scalar_one_or_none()
+
+    if setting is None:
+        tariffs_config = TARIFFS_CONFIG.copy()
+        setting = Setting(
+            key="TARIFFS_CONFIG",
+            value=tariffs_config,
+            description="Конфигурация тарифов",
+        )
+        session.add(setting)
+    else:
+        stored = setting.value or {}
+        tariffs_config = TARIFFS_CONFIG.copy()
+        tariffs_config.update(stored)
+        setting.value = tariffs_config
+
+    TARIFFS_CONFIG.clear()
+    TARIFFS_CONFIG.update(tariffs_config)
+    await session.flush()
+
+
+async def update_tariffs_config(session: AsyncSession, new_values: dict[str, Any]) -> None:
+    """Обновляет конфиг тарифов."""
+    stmt = select(Setting).where(Setting.key == "TARIFFS_CONFIG")
+    result = await session.execute(stmt)
+    setting = result.scalar_one_or_none()
+
+    if setting is None:
+        setting = Setting(
+            key="TARIFFS_CONFIG",
+            value=new_values,
+            description="Конфигурация тарифов",
+        )
+        session.add(setting)
+    else:
+        setting.value = new_values
+
+    await session.commit()
+
+    tariffs_config = TARIFFS_CONFIG.copy()
+    tariffs_config.update(new_values)
+
+    TARIFFS_CONFIG.clear()
+    TARIFFS_CONFIG.update(tariffs_config)
+
+
+def calc_extra_devices_price(tariff: dict[str, Any], device_limit: int) -> int:
+    base_device_limit = int(tariff.get("device_limit") or 1)
+    extra_devices = max(0, device_limit - base_device_limit)
+    if extra_devices <= 0:
+        return 0
+
+    step_price = int(tariff.get("device_step_rub") or 0)
+    overrides = tariff.get("device_overrides") or {}
+
+    override_total = overrides.get(str(device_limit))
+    if override_total is not None:
+        return int(ceil(float(override_total)))
+
+    return int(ceil(extra_devices * step_price))
+
+
+def calc_extra_traffic_price(tariff: dict[str, Any], traffic_gb: int | None) -> int:
+    if traffic_gb is None:
+        return 0
+
+    traffic_limit_bytes = tariff.get("traffic_limit")
+    if traffic_limit_bytes:
+        base_traffic_gb = ceil(traffic_limit_bytes / 1024 / 1024 / 1024)
+    else:
+        base_traffic_gb = 0
+
+    step_price = int(tariff.get("traffic_step_rub") or 0)
+    overrides = tariff.get("traffic_overrides") or {}
+
+    override_total = overrides.get(str(traffic_gb))
+    if override_total is not None:
+        return int(ceil(float(override_total)))
+
+    if traffic_gb == 0:
+        return 0
+
+    extra_gb = max(0, traffic_gb - base_traffic_gb)
+    if extra_gb <= 0:
+        return 0
+
+    return int(ceil(extra_gb * step_price))
+
+
+def calculate_config_price(
+    tariff: dict[str, Any],
+    duration_days: int,
+    device_limit: int,
+    traffic_gb: int | None,
+) -> int:
+    base_duration = int(tariff.get("duration_days") or 0) or duration_days or 30
+    if base_duration <= 0:
+        base_duration = duration_days or 30
+
+    base_price = int(tariff.get("price_rub") or 0)
+
+    duration_multiplier = duration_days / base_duration
+    base_price_scaled = base_price * duration_multiplier
+
+    extra_devices_price = calc_extra_devices_price(tariff, device_limit)
+    extra_traffic_price = calc_extra_traffic_price(tariff, traffic_gb)
+
+    total = base_price_scaled + extra_devices_price + extra_traffic_price
+    return int(ceil(total))
+
+
+def normalize_tariff_config(tariff: dict[str, Any]) -> dict[str, Any]:
+    raw_duration_options = tariff.get("duration_options") or []
+    duration_options: list[int] = []
+    for value in raw_duration_options:
+        try:
+            v = int(value)
+        except (TypeError, ValueError):
+            continue
+        if v > 0:
+            duration_options.append(v)
+    if not duration_options:
+        base_duration = int(tariff.get("duration_days") or 0) or 30
+        duration_options = [base_duration]
+    duration_options = sorted(set(duration_options))
+
+    raw_device_options = tariff.get("device_options") or []
+    device_options: list[int] = []
+    for value in raw_device_options:
+        try:
+            v = int(value)
+        except (TypeError, ValueError):
+            continue
+        if v > 0:
+            device_options.append(v)
+    if not device_options:
+        base_device_limit = int(tariff.get("device_limit") or 0)
+        if base_device_limit > 0:
+            device_options = [base_device_limit]
+        else:
+            device_options = []
+    device_options = sorted(set(device_options))
+
+    raw_traffic_options = tariff.get("traffic_options_gb")
+    traffic_options_gb: list[int] | None
+    if raw_traffic_options is None:
+        traffic_options_gb = None
+    else:
+        traffic_values: list[int] = []
+        has_unlimited = False
+        for value in raw_traffic_options:
+            try:
+                v = int(value)
+            except (TypeError, ValueError):
+                continue
+            if v == 0:
+                has_unlimited = True
+            elif v > 0:
+                traffic_values.append(v)
+        if not traffic_values and not has_unlimited:
+            traffic_options_gb = None
+        else:
+            unique_values = sorted(set(traffic_values))
+            if has_unlimited:
+                traffic_options_gb = [0] + unique_values
+            else:
+                traffic_options_gb = unique_values
+
+    return {
+        "duration_options": duration_options,
+        "device_options": device_options,
+        "traffic_options_gb": traffic_options_gb,
+    }