feat: add tariff config and database schema

config/settings.py

@@ -5,6 +5,7 @@ from typing import Optional, List, Dict, Any
 
 from pydantic import BaseModel, Field, ValidationError, computed_field, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
+from config.tariffs_config import TariffsConfig, load_tariffs_config
 
 
 def _split_csv(value: Optional[str]) -> List[str]:
@@ -259,6 +260,7 @@ class Settings(BaseSettings):
         default=None,
         description="Comma-separated list of traffic packages priced in Stars, e.g. '5:500,20:1500'",
     )
+    TARIFFS_CONFIG_PATH: str = Field(default="config/tariffs.json")
 
     SUBSCRIPTION_NOTIFICATIONS_ENABLED: bool = Field(default=True)
     SUBSCRIPTION_NOTIFY_ON_EXPIRE: bool = Field(default=True)
@@ -731,8 +733,15 @@ class Settings(BaseSettings):
     @property
     def traffic_sale_mode(self) -> bool:
         """When true, the bot sells traffic packages instead of time-based subscriptions."""
+        if self.tariffs_config is not None:
+            return False
         return bool(self.traffic_packages or self.stars_traffic_packages)
 
+    @computed_field
+    @property
+    def tariffs_config(self) -> Optional[TariffsConfig]:
+        return load_tariffs_config(self.TARIFFS_CONFIG_PATH)
+
     @computed_field
     @property
     def referral_bonus_inviter(self) -> Dict[int, int]:

config/tariffs.example.json

@@ -0,0 +1,49 @@
+{
+  "default_tariff": "standard",
+  "topup_packages_default": {
+    "rub": [
+      { "gb": 10, "price": 99 },
+      { "gb": 50, "price": 399 },
+      { "gb": 200, "price": 1299 }
+    ],
+    "stars": [
+      { "gb": 10, "price": 2500 }
+    ]
+  },
+  "tariffs": [
+    {
+      "key": "standard",
+      "names": { "ru": "Стандарт", "en": "Standard" },
+      "descriptions": { "ru": "Базовый набор серверов", "en": "Base server pool" },
+      "squad_uuids": ["uuid-1", "uuid-2"],
+      "billing_model": "period",
+      "monthly_gb": 500,
+      "prices_rub": { "1": 150, "3": 400, "6": 750, "12": 1400 },
+      "prices_stars": { "1": 0, "3": 0, "6": 0, "12": 0 },
+      "enabled_periods": [1, 3, 6, 12],
+      "enabled": true
+    },
+    {
+      "key": "traffic_basic",
+      "names": { "ru": "Гигабайты", "en": "Gigabytes" },
+      "descriptions": {
+        "ru": "Пакеты трафика без ограничения по времени",
+        "en": "Traffic packages with no time limit"
+      },
+      "squad_uuids": ["uuid-1", "uuid-2"],
+      "billing_model": "traffic",
+      "conversion_rate_rub_per_gb": 20,
+      "traffic_packages": {
+        "rub": [
+          { "gb": 10, "price": 199 },
+          { "gb": 50, "price": 799 },
+          { "gb": 200, "price": 1999 }
+        ],
+        "stars": [
+          { "gb": 10, "price": 2500 }
+        ]
+      },
+      "enabled": true
+    }
+  ]
+}

config/tariffs_config.py

@@ -0,0 +1,173 @@
+import json
+import logging
+from pathlib import Path
+from typing import Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field, ValidationError, model_validator
+
+
+Currency = Literal["rub", "stars"]
+BillingModel = Literal["period", "traffic"]
+
+
+class TrafficPackage(BaseModel):
+    gb: float
+    price: float
+
+    @model_validator(mode="after")
+    def validate_values(self) -> "TrafficPackage":
+        if self.gb <= 0:
+            raise ValueError("package gb must be greater than zero")
+        if self.price < 0:
+            raise ValueError("package price must be non-negative")
+        return self
+
+
+class PackageSet(BaseModel):
+    rub: List[TrafficPackage] = Field(default_factory=list)
+    stars: List[TrafficPackage] = Field(default_factory=list)
+
+    def for_currency(self, currency: Currency) -> List[TrafficPackage]:
+        return list(getattr(self, currency) or [])
+
+    def has_any(self) -> bool:
+        return bool(self.rub or self.stars)
+
+
+class Tariff(BaseModel):
+    key: str
+    names: Dict[str, str] = Field(default_factory=dict)
+    descriptions: Dict[str, str] = Field(default_factory=dict)
+    squad_uuids: List[str] = Field(default_factory=list)
+    billing_model: BillingModel
+    enabled: bool = True
+
+    monthly_gb: Optional[float] = None
+    prices_rub: Dict[str, float] = Field(default_factory=dict)
+    prices_stars: Dict[str, float] = Field(default_factory=dict)
+    enabled_periods: List[int] = Field(default_factory=list)
+    topup_packages: Optional[PackageSet] = None
+
+    traffic_packages: Optional[PackageSet] = None
+    conversion_rate_rub_per_gb: Optional[float] = None
+
+    @model_validator(mode="after")
+    def validate_tariff(self) -> "Tariff":
+        if not self.key.strip():
+            raise ValueError("tariff key must not be empty")
+        self.key = self.key.strip()
+        self.squad_uuids = [uuid.strip() for uuid in self.squad_uuids if uuid.strip()]
+
+        if self.billing_model == "period":
+            if self.monthly_gb is None or self.monthly_gb < 0:
+                raise ValueError(f"period tariff {self.key}: monthly_gb must be >= 0")
+            if not self.enabled_periods:
+                raise ValueError(f"period tariff {self.key}: enabled_periods is required")
+            for months in self.enabled_periods:
+                if months <= 0:
+                    raise ValueError(f"period tariff {self.key}: enabled periods must be positive")
+                rub_price = self.prices_rub.get(str(months), 0) or 0
+                stars_price = self.prices_stars.get(str(months), 0) or 0
+                if rub_price <= 0 and stars_price <= 0:
+                    raise ValueError(
+                        f"period tariff {self.key}: period {months} needs a non-zero rub or stars price"
+                    )
+            return self
+
+        if not self.traffic_packages or not self.traffic_packages.has_any():
+            raise ValueError(f"traffic tariff {self.key}: traffic_packages is required")
+        if self.conversion_rate_rub_per_gb is not None and self.conversion_rate_rub_per_gb <= 0:
+            raise ValueError(f"traffic tariff {self.key}: conversion_rate_rub_per_gb must be > 0")
+        if not self.traffic_packages.rub and self.conversion_rate_rub_per_gb is None:
+            raise ValueError(
+                f"traffic tariff {self.key}: conversion_rate_rub_per_gb is required without RUB packages"
+            )
+        return self
+
+    def name(self, lang: str, fallback: str = "ru") -> str:
+        return self.names.get(lang) or self.names.get(fallback) or self.key
+
+    def description(self, lang: str, fallback: str = "ru") -> str:
+        return self.descriptions.get(lang) or self.descriptions.get(fallback) or ""
+
+    @property
+    def monthly_bytes(self) -> int:
+        if self.monthly_gb is None or self.monthly_gb <= 0:
+            return 0
+        return int(float(self.monthly_gb) * (1024**3))
+
+    def period_price(self, months: int, currency: Currency = "rub") -> Optional[float]:
+        source = self.prices_rub if currency == "rub" else self.prices_stars
+        value = source.get(str(months))
+        return float(value) if value is not None else None
+
+    def min_period_price_rub(self) -> Optional[float]:
+        prices = [
+            float(self.prices_rub[str(months)])
+            for months in self.enabled_periods
+            if self.prices_rub.get(str(months), 0) and self.prices_rub.get(str(months), 0) > 0
+        ]
+        return min(prices) if prices else None
+
+    def min_traffic_package_rub(self) -> Optional[TrafficPackage]:
+        packages = self.traffic_packages.rub if self.traffic_packages else []
+        return min(packages, key=lambda pkg: pkg.price) if packages else None
+
+    def rub_per_gb_for_conversion(self) -> float:
+        if self.conversion_rate_rub_per_gb:
+            return float(self.conversion_rate_rub_per_gb)
+        packages = self.traffic_packages.rub if self.traffic_packages else []
+        return min(float(pkg.price) / float(pkg.gb) for pkg in packages)
+
+
+class TariffsConfig(BaseModel):
+    default_tariff: str
+    topup_packages_default: Optional[PackageSet] = None
+    tariffs: List[Tariff]
+
+    @model_validator(mode="after")
+    def validate_config(self) -> "TariffsConfig":
+        keys = [tariff.key for tariff in self.tariffs]
+        if len(keys) != len(set(keys)):
+            raise ValueError("tariff keys must be unique")
+        active = [tariff for tariff in self.tariffs if tariff.enabled]
+        if not active:
+            raise ValueError("at least one enabled tariff is required")
+        active_keys = {tariff.key for tariff in active}
+        if self.default_tariff not in active_keys:
+            raise ValueError("default_tariff must reference an enabled tariff")
+        return self
+
+    @property
+    def enabled_tariffs(self) -> List[Tariff]:
+        return [tariff for tariff in self.tariffs if tariff.enabled]
+
+    def get(self, key: str) -> Optional[Tariff]:
+        return next((tariff for tariff in self.tariffs if tariff.key == key), None)
+
+    def require(self, key: str) -> Tariff:
+        tariff = self.get(key)
+        if not tariff or not tariff.enabled:
+            raise KeyError(f"Unknown or disabled tariff: {key}")
+        return tariff
+
+    @property
+    def default(self) -> Tariff:
+        return self.require(self.default_tariff)
+
+    def topup_packages_for(self, tariff: Tariff) -> Optional[PackageSet]:
+        if tariff.billing_model == "traffic":
+            return tariff.traffic_packages
+        return tariff.topup_packages if tariff.topup_packages is not None else self.topup_packages_default
+
+
+def load_tariffs_config(path: str | Path) -> Optional[TariffsConfig]:
+    config_path = Path(path)
+    if not config_path.exists():
+        return None
+    try:
+        data = json.loads(config_path.read_text(encoding="utf-8"))
+        return TariffsConfig.model_validate(data)
+    except (OSError, json.JSONDecodeError, ValidationError, ValueError) as exc:
+        logging.critical("Failed to load tariffs config from %s: %s", config_path, exc)
+        raise