79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
from __future__ import annotations
|
|
import os
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import yaml
|
|
from pydantic import BaseModel
|
|
|
|
class ProviderConfig(BaseModel):
|
|
type: str
|
|
enabled: bool = True
|
|
model_config = {"extra": "allow"}
|
|
|
|
def extra(self) -> dict[str, Any]:
|
|
return dict(self.__pydantic_extra__) if self.__pydantic_extra__ else {}
|
|
|
|
class RoutingRule(BaseModel):
|
|
name: str
|
|
prefixes: list[str]
|
|
provider: str
|
|
fallback: str | None = None
|
|
|
|
def matches(self, phone: str) -> bool:
|
|
normalized = phone if phone.startswith("+") else f"+{phone}"
|
|
for prefix in sorted(self.prefixes, key=len, reverse=True):
|
|
if normalized.startswith(prefix):
|
|
return True
|
|
return False
|
|
|
|
class RoutingConfig(BaseModel):
|
|
rules: list[RoutingRule] = []
|
|
default_provider: str = "lk_api"
|
|
default_fallback: str | None = None
|
|
|
|
class RateLimitSettings(BaseModel):
|
|
enabled: bool = True
|
|
max_attempts: int = 3
|
|
window_seconds: int = 600
|
|
|
|
class AppSettings(BaseModel):
|
|
log_codes: bool = True
|
|
code_ttl_seconds: int = 300
|
|
rate_limit: RateLimitSettings = RateLimitSettings()
|
|
|
|
class RedisConfig(BaseModel):
|
|
host: str = "redis"
|
|
port: int = 6379
|
|
db: int = 0
|
|
password: str | None = None
|
|
|
|
def url(self) -> str:
|
|
if self.password:
|
|
return f"redis://:{self.password}@{self.host}:{self.port}/{self.db}"
|
|
return f"redis://{self.host}:{self.port}/{self.db}"
|
|
|
|
class Config(BaseModel):
|
|
providers: dict[str, ProviderConfig]
|
|
routing: RoutingConfig
|
|
settings: AppSettings = AppSettings()
|
|
redis: RedisConfig = RedisConfig()
|
|
|
|
def resolve_provider(self, phone: str) -> tuple[str, str | None]:
|
|
for rule in self.routing.rules:
|
|
if rule.matches(phone):
|
|
return rule.provider, rule.fallback
|
|
return self.routing.default_provider, self.routing.default_fallback
|
|
|
|
@lru_cache(maxsize=1)
|
|
def load_config() -> Config:
|
|
path = Path(os.getenv("CONFIG_PATH", "config.yaml"))
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Конфиг не найден: {path}")
|
|
with open(path, encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
return Config.model_validate(raw)
|
|
|
|
def reload_config() -> Config:
|
|
load_config.cache_clear()
|
|
return load_config() |