import logging import os import secrets from typing import Any, Dict, List, Optional 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]: if not value: return [] return [item.strip() for item in value.split(",") if item.strip()] class DBSettings(BaseModel): user: str password: str host: str port: int database: str class PaymentSettings(BaseModel): yookassa_enabled: bool yookassa_shop_id: Optional[str] yookassa_secret_key: Optional[str] yookassa_return_url: Optional[str] yookassa_default_receipt_email: Optional[str] yookassa_vat_code: int yookassa_payment_mode: str yookassa_payment_subject: str yookassa_autopayments_enabled: bool yookassa_autopayments_require_card_binding: bool freekassa_enabled: bool freekassa_merchant_id: Optional[str] freekassa_second_secret: Optional[str] freekassa_api_key: Optional[str] freekassa_payment_ip: Optional[str] freekassa_payment_method_id: Optional[int] freekassa_trusted_ips: List[str] platega_enabled: bool platega_base_url: str platega_merchant_id: Optional[str] platega_secret: Optional[str] platega_payment_method: int platega_sbp_enabled: bool platega_crypto_enabled: bool platega_sbp_method: int platega_crypto_method: int platega_return_url: Optional[str] platega_failed_url: Optional[str] severpay_enabled: bool severpay_mid: Optional[int] severpay_token: Optional[str] severpay_return_url: Optional[str] severpay_base_url: str severpay_lifetime_minutes: Optional[int] cryptopay_enabled: bool cryptopay_token: Optional[str] cryptopay_network: str cryptopay_currency_type: str cryptopay_asset: str class EmailSettings(BaseModel): smtp_host: str smtp_port: int smtp_fallback_ports: Optional[str] smtp_timeout_seconds: int smtp_username: Optional[str] smtp_password: Optional[str] smtp_from_email: Optional[str] smtp_from_name: Optional[str] smtp_starttls: bool smtp_use_ssl: bool email_code_ttl_seconds: int email_code_resend_seconds: int email_code_max_attempts: int brute_force_max_failures: int brute_force_window_seconds: int brute_force_lock_seconds: int class WebAppSettings(BaseModel): title: str primary_color: str logo_url: Optional[str] logo_emoji: str logo_emoji_font: str session_ttl_seconds: int session_secret: str webhook_secret_token: str auth_max_age_seconds: int login_token_ttl_seconds: int server_host: str server_port: int enabled: bool trusted_proxies: List[str] class Settings(BaseSettings): BOT_TOKEN: str ADMIN_IDS_STR: str = Field( default="", alias="ADMIN_IDS", description="Comma-separated list of admin Telegram User IDs" ) POSTGRES_USER: str = Field(...) POSTGRES_PASSWORD: str = Field(...) POSTGRES_HOST: str = Field(default="localhost") POSTGRES_PORT: int = Field(default=5432) POSTGRES_DB: str = Field(default="vpn_shop_db") DEFAULT_LANGUAGE: str = Field(default="ru") DEFAULT_CURRENCY_SYMBOL: str = Field(default="RUB") SUPPORT_LINK: Optional[str] = Field(default=None) SERVER_STATUS_URL: Optional[str] = Field(default=None) TERMS_OF_SERVICE_URL: Optional[str] = Field(default=None) PRIVACY_POLICY_URL: Optional[str] = Field(default=None) USER_AGREEMENT_URL: Optional[str] = Field(default=None) REQUIRED_CHANNEL_ID: Optional[int] = Field( default=None, description="Telegram channel ID the user must join to access the bot" ) REQUIRED_CHANNEL_LINK: Optional[str] = Field( default=None, description="Public username or invite link to the required channel for join button", ) YOOKASSA_SHOP_ID: Optional[str] = None YOOKASSA_SECRET_KEY: Optional[str] = None YOOKASSA_RETURN_URL: Optional[str] = None YOOKASSA_DEFAULT_RECEIPT_EMAIL: Optional[str] = Field(default=None) YOOKASSA_VAT_CODE: int = Field(default=1) # Deprecated: explicit receipt fields are now derived from YOOKASSA_AUTOPAYMENTS_ENABLED YOOKASSA_PAYMENT_MODE: str = Field(default="full_prepayment") YOOKASSA_PAYMENT_SUBJECT: str = Field(default="service") # Single toggle to enable recurring payments (saving cards, managing payment methods, auto-renew) # noqa: E501 YOOKASSA_AUTOPAYMENTS_ENABLED: bool = Field(default=False) YOOKASSA_AUTOPAYMENTS_REQUIRE_CARD_BINDING: bool = Field( default=True, description="When true, new YooKassa payments in autopay mode force card binding without a user checkbox.", # noqa: E501 ) LKNPD_INN: Optional[str] = Field( default=None, alias="NALOGO_INN", description="INN for lknpd.nalog.ru (self-employed) authentication", ) LKNPD_PASSWORD: Optional[str] = Field( default=None, alias="NALOGO_PASSWORD", description="Password for lknpd.nalog.ru (self-employed) authentication", ) LKNPD_API_URL: str = Field( default="https://lknpd.nalog.ru/api", alias="NALOGO_API_URL", description="Base URL for LKNPD API (can be overridden for proxies)", ) LKNPD_RECEIPT_NAME_SUBSCRIPTION: str = Field( default="subscription {months} months", alias="NALOGO_RECEIPT_NAME_SUBSCRIPTION", description="Receipt item name for time-based subscriptions. Use {months} placeholder for duration.", # noqa: E501 ) LKNPD_RECEIPT_NAME_TRAFFIC: str = Field( default="traffic package {gb} GB", alias="NALOGO_RECEIPT_NAME_TRAFFIC", description="Receipt item name for traffic packages. Use {gb} placeholder for traffic amount.", # noqa: E501 ) WEBHOOK_BASE_URL: Optional[str] = None TRUSTED_PROXIES: Optional[str] = Field( default="127.0.0.1,::1", description="Comma-separated list of reverse proxy IPs or CIDRs trusted to forward X-Forwarded-For.", # noqa: E501 ) CRYPTOPAY_TOKEN: Optional[str] = None CRYPTOPAY_NETWORK: str = Field(default="mainnet") CRYPTOPAY_CURRENCY_TYPE: str = Field(default="fiat") CRYPTOPAY_ASSET: str = Field(default="RUB") CRYPTOPAY_ENABLED: bool = Field(default=True) PLATEGA_ENABLED: bool = Field(default=False) PLATEGA_BASE_URL: str = Field(default="https://app.platega.io") PLATEGA_MERCHANT_ID: Optional[str] = None PLATEGA_SECRET: Optional[str] = None PLATEGA_PAYMENT_METHOD: int = Field( default=2, description="Legacy Platega payment method ID. Used as fallback for PLATEGA_SBP_METHOD when the new field is unset.", # noqa: E501 ) PLATEGA_SBP_ENABLED: bool = Field( default=False, description="Show a separate Platega SBP payment button.", ) PLATEGA_CRYPTO_ENABLED: bool = Field( default=False, description="Show a separate Platega crypto payment button.", ) PLATEGA_SBP_METHOD: int = Field( default=2, description="Platega method ID for SBP QR (default 2).", ) PLATEGA_CRYPTO_METHOD: int = Field( default=13, description="Platega method ID for crypto (default 13).", ) PLATEGA_RETURN_URL: Optional[str] = Field(default=None) PLATEGA_FAILED_URL: Optional[str] = Field(default=None) FREEKASSA_ENABLED: bool = Field(default=False) FREEKASSA_MERCHANT_ID: Optional[str] = None FREEKASSA_FIRST_SECRET: Optional[str] = None FREEKASSA_SECOND_SECRET: Optional[str] = None FREEKASSA_PAYMENT_URL: str = Field(default="https://pay.freekassa.ru/") FREEKASSA_API_KEY: Optional[str] = None FREEKASSA_PAYMENT_IP: Optional[str] = None FREEKASSA_PAYMENT_METHOD_ID: Optional[int] = None FREEKASSA_TRUSTED_IPS: str = Field( default="168.119.157.136,168.119.60.227,178.154.197.79,51.250.54.238", description="Comma-separated FreeKassa webhook IP allowlist.", ) SEVERPAY_ENABLED: bool = Field(default=False) SEVERPAY_MID: Optional[int] = None SEVERPAY_TOKEN: Optional[str] = None SEVERPAY_RETURN_URL: Optional[str] = None SEVERPAY_BASE_URL: str = Field(default="https://severpay.io/api/merchant") SEVERPAY_LIFETIME_MINUTES: Optional[int] = Field( default=None, description="Lifetime of the payment link in minutes (30-4320, defaults to provider value)", ) YOOKASSA_ENABLED: bool = Field(default=True) STARS_ENABLED: bool = Field(default=True) PAYMENT_METHODS_ORDER: Optional[str] = Field( default=None, description="Comma-separated list of payment methods to show (e.g., severpay,freekassa,yookassa,platega,stars,cryptopay)", # noqa: E501 ) MONTH_1_ENABLED: bool = Field(default=True, alias="1_MONTH_ENABLED") MONTH_3_ENABLED: bool = Field(default=True, alias="3_MONTHS_ENABLED") MONTH_6_ENABLED: bool = Field(default=True, alias="6_MONTHS_ENABLED") MONTH_12_ENABLED: bool = Field(default=True, alias="12_MONTHS_ENABLED") RUB_PRICE_1_MONTH: Optional[int] = Field(default=None) RUB_PRICE_3_MONTHS: Optional[int] = Field(default=None) RUB_PRICE_6_MONTHS: Optional[int] = Field(default=None) RUB_PRICE_12_MONTHS: Optional[int] = Field(default=None) STARS_PRICE_1_MONTH: Optional[int] = Field(default=None) STARS_PRICE_3_MONTHS: Optional[int] = Field(default=None) STARS_PRICE_6_MONTHS: Optional[int] = Field(default=None) STARS_PRICE_12_MONTHS: Optional[int] = Field(default=None) PANEL_WEBHOOK_SECRET: Optional[str] = Field(default=None) TRAFFIC_PACKAGES: Optional[str] = Field( default=None, description="Comma-separated list of traffic packages in the format ':', e.g. '10:199,50:799'", # noqa: E501 ) STARS_TRAFFIC_PACKAGES: Optional[str] = Field( default=None, description="Comma-separated list of traffic packages priced in Stars, e.g. '5:500,20:1500'", # noqa: E501 ) TARIFFS_CONFIG_PATH: str = Field(default="data/tariffs.json") TARIFF_TRAFFIC_WARNING_LEVELS: str = Field( default="85,90,95", description="Comma-separated traffic usage warning levels for tariff traffic limits, e.g. '85,90,95'", # noqa: E501 ) SUBSCRIPTION_NOTIFICATIONS_ENABLED: bool = Field(default=True) SUBSCRIPTION_NOTIFY_ON_EXPIRE: bool = Field(default=True) SUBSCRIPTION_NOTIFY_AFTER_EXPIRE: bool = Field(default=True) SUBSCRIPTION_NOTIFY_DAYS_BEFORE: int = Field(default=3) REFERRAL_BONUS_DAYS_INVITER_1_MONTH: Optional[int] = Field( default=3, alias="REFERRAL_BONUS_DAYS_1_MONTH" ) REFERRAL_BONUS_DAYS_INVITER_3_MONTHS: Optional[int] = Field( default=7, alias="REFERRAL_BONUS_DAYS_3_MONTHS" ) REFERRAL_BONUS_DAYS_INVITER_6_MONTHS: Optional[int] = Field( default=15, alias="REFERRAL_BONUS_DAYS_6_MONTHS" ) REFERRAL_BONUS_DAYS_INVITER_12_MONTHS: Optional[int] = Field( default=30, alias="REFERRAL_BONUS_DAYS_12_MONTHS" ) REFERRAL_BONUS_DAYS_REFEREE_1_MONTH: Optional[int] = Field( default=1, alias="REFEREE_BONUS_DAYS_1_MONTH" ) REFERRAL_BONUS_DAYS_REFEREE_3_MONTHS: Optional[int] = Field( default=3, alias="REFEREE_BONUS_DAYS_3_MONTHS" ) REFERRAL_BONUS_DAYS_REFEREE_6_MONTHS: Optional[int] = Field( default=7, alias="REFEREE_BONUS_DAYS_6_MONTHS" ) REFERRAL_BONUS_DAYS_REFEREE_12_MONTHS: Optional[int] = Field( default=15, alias="REFEREE_BONUS_DAYS_12_MONTHS" ) # Referral program configuration REFERRAL_ONE_BONUS_PER_REFEREE: bool = Field( default=True, description="When true, referral bonuses (for inviter and referee) are applied only once per invited user - on their first successful payment.", # noqa: E501 ) REFERRAL_WELCOME_BONUS_DAYS: int = Field( default=3, description="Welcome bonus days granted to a newly registered user who joined via referral link.", # noqa: E501 ) LEGACY_REFS: bool = Field( default=True, description="Allow legacy referral links like ref_ to continue working. Defaults to True when unset.", # noqa: E501 ) PANEL_API_URL: Optional[str] = None PANEL_API_KEY: Optional[str] = None USER_TRAFFIC_LIMIT_GB: Optional[float] = Field(default=0.0) USER_TRAFFIC_STRATEGY: str = Field(default="NO_RESET") USER_SQUAD_UUIDS: Optional[str] = Field( default=None, description="Comma-separated UUIDs of internal squads to assign to new panel users", ) USER_EXTERNAL_SQUAD_UUID: Optional[str] = Field( default=None, description="UUID of the external squad to assign to new panel users (optional)", ) TRIAL_ENABLED: bool = Field(default=True) TRIAL_DURATION_DAYS: int = Field(default=3) TRIAL_TRAFFIC_LIMIT_GB: Optional[float] = Field(default=5.0) TRIAL_TRAFFIC_STRATEGY: str = Field(default="NO_RESET") CRYPT4_ENABLED: bool = Field( default=False, description="Enable happ crypt4 encryption for subscription URLs" ) CRYPT4_REDIRECT_URL: Optional[str] = Field( default=None, description="Base redirect URL used for the connect button when crypt4 is enabled", ) WEB_SERVER_HOST: str = Field(default="0.0.0.0") WEB_SERVER_PORT: int = Field(default=8080) WEBAPP_ENABLED: bool = Field( default=True, description="Run the subscription Mini App in the same container on a separate port.", ) WEBAPP_SERVER_HOST: str = Field(default="0.0.0.0") WEBAPP_SERVER_PORT: int = Field(default=8081) WEBAPP_TITLE: str = Field(default="Моя подписка") WEBAPP_PRIMARY_COLOR: str = Field(default="#00fe7a") WEBAPP_LOGO_URL: Optional[str] = Field(default=None) WEBAPP_LOGO_EMOJI: str = Field(default="🫥") WEBAPP_LOGO_EMOJI_FONT: str = Field( default="system", description=( "Emoji font for logo fallback: system, noto-color, noto-color-animated, " "noto-emoji, twemoji, openmoji, apple, segoe, noto-local" ), ) WEBAPP_SESSION_SECRET: str = Field(default_factory=lambda: secrets.token_urlsafe(32)) WEBHOOK_SECRET_TOKEN: str = Field(default_factory=lambda: secrets.token_urlsafe(32)) WEBAPP_SESSION_TTL_SECONDS: int = Field(default=24 * 60 * 60) WEBAPP_AUTH_MAX_AGE_SECONDS: int = Field(default=24 * 60 * 60) WEBAPP_LOGIN_TOKEN_TTL_SECONDS: int = Field(default=10 * 60) TELEGRAM_OAUTH_CLIENT_ID: Optional[int] = Field( default=None, description="Telegram Web Login Client ID from BotFather. Defaults to the numeric bot ID from BOT_TOKEN.", # noqa: E501 ) TELEGRAM_OAUTH_CLIENT_SECRET: Optional[str] = Field( default=None, description="Telegram Web Login Client Secret from BotFather. Reserved for full OIDC authorization code integrations.", # noqa: E501 ) TELEGRAM_OAUTH_REQUEST_ACCESS: Optional[str] = Field( default="write", description="Comma-separated Telegram Login permissions to request: write,phone. Leave empty to request only OpenID profile.", # noqa: E501 ) SMTP_HOST: str = Field(default="smtp-relay.brevo.com") SMTP_PORT: int = Field(default=587) SMTP_FALLBACK_PORTS: Optional[str] = Field(default="2525,465") SMTP_TIMEOUT_SECONDS: int = Field(default=30) SMTP_USERNAME: Optional[str] = Field(default=None) SMTP_PASSWORD: Optional[str] = Field(default=None) SMTP_FROM_EMAIL: Optional[str] = Field(default=None) SMTP_FROM_NAME: Optional[str] = Field(default=None) SMTP_STARTTLS: bool = Field(default=True) SMTP_USE_SSL: bool = Field(default=False) EMAIL_CODE_TTL_SECONDS: int = Field(default=10 * 60) EMAIL_CODE_RESEND_SECONDS: int = Field(default=60) EMAIL_CODE_MAX_ATTEMPTS: int = Field(default=5) BRUTE_FORCE_MAX_FAILURES: int = Field( default=5, description="Maximum failed code attempts allowed within the throttle window before a temporary lockout is applied.", # noqa: E501 ) BRUTE_FORCE_WINDOW_SECONDS: int = Field( default=15 * 60, description="Rolling window used to count failed email and promo code attempts.", ) BRUTE_FORCE_LOCK_SECONDS: int = Field( default=30 * 60, description="Temporary lockout duration applied after too many failed code attempts.", ) LOGS_PAGE_SIZE: int = Field(default=10) LOG_ADMIN_ACTIONS: bool = Field( default=True, description="Log updates/events triggered by users from ADMIN_IDS.", ) SUBSCRIPTION_MINI_APP_URL: Optional[str] = Field(default=None) START_COMMAND_DESCRIPTION: Optional[str] = Field(default=None) DISABLE_WELCOME_MESSAGE: bool = Field( default=False, description="Disable welcome message on /start command" ) MY_DEVICES_SECTION_ENABLED: bool = Field( default=False, description="Enable the My Devices section in the subscription menu" ) USER_HWID_DEVICE_LIMIT: Optional[int] = Field( default=None, description="Default hardware device limit for panel users (0 = unlimited)" ) # Inline mode thumbnail URLs INLINE_REFERRAL_THUMBNAIL_URL: str = Field( default="https://cdn-icons-png.flaticon.com/512/1077/1077114.png" ) INLINE_USER_STATS_THUMBNAIL_URL: str = Field( default="https://cdn-icons-png.flaticon.com/512/681/681494.png" ) INLINE_FINANCIAL_STATS_THUMBNAIL_URL: str = Field( default="https://cdn-icons-png.flaticon.com/512/2769/2769339.png" ) INLINE_SYSTEM_STATS_THUMBNAIL_URL: str = Field( default="https://cdn-icons-png.flaticon.com/512/2920/2920277.png" ) @computed_field @property def DATABASE_URL(self) -> str: return f"postgresql+asyncpg://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}" @computed_field @property def db_settings(self) -> DBSettings: return DBSettings( user=self.POSTGRES_USER, password=self.POSTGRES_PASSWORD, host=self.POSTGRES_HOST, port=self.POSTGRES_PORT, database=self.POSTGRES_DB, ) @computed_field @property def payment_settings(self) -> PaymentSettings: return PaymentSettings( yookassa_enabled=self.YOOKASSA_ENABLED, yookassa_shop_id=self.YOOKASSA_SHOP_ID, yookassa_secret_key=self.YOOKASSA_SECRET_KEY, yookassa_return_url=self.YOOKASSA_RETURN_URL, yookassa_default_receipt_email=self.YOOKASSA_DEFAULT_RECEIPT_EMAIL, yookassa_vat_code=self.YOOKASSA_VAT_CODE, yookassa_payment_mode=self.YOOKASSA_PAYMENT_MODE, yookassa_payment_subject=self.YOOKASSA_PAYMENT_SUBJECT, yookassa_autopayments_enabled=self.YOOKASSA_AUTOPAYMENTS_ENABLED, yookassa_autopayments_require_card_binding=self.YOOKASSA_AUTOPAYMENTS_REQUIRE_CARD_BINDING, freekassa_enabled=self.FREEKASSA_ENABLED, freekassa_merchant_id=self.FREEKASSA_MERCHANT_ID, freekassa_second_secret=self.FREEKASSA_SECOND_SECRET, freekassa_api_key=self.FREEKASSA_API_KEY, freekassa_payment_ip=self.FREEKASSA_PAYMENT_IP, freekassa_payment_method_id=self.FREEKASSA_PAYMENT_METHOD_ID, freekassa_trusted_ips=self.freekassa_trusted_ips, platega_enabled=self.PLATEGA_ENABLED, platega_base_url=self.PLATEGA_BASE_URL, platega_merchant_id=self.PLATEGA_MERCHANT_ID, platega_secret=self.PLATEGA_SECRET, platega_payment_method=self.PLATEGA_PAYMENT_METHOD, platega_sbp_enabled=self.PLATEGA_SBP_ENABLED, platega_crypto_enabled=self.PLATEGA_CRYPTO_ENABLED, platega_sbp_method=self.platega_sbp_method_resolved, platega_crypto_method=self.PLATEGA_CRYPTO_METHOD, platega_return_url=self.PLATEGA_RETURN_URL, platega_failed_url=self.PLATEGA_FAILED_URL, severpay_enabled=self.SEVERPAY_ENABLED, severpay_mid=self.SEVERPAY_MID, severpay_token=self.SEVERPAY_TOKEN, severpay_return_url=self.SEVERPAY_RETURN_URL, severpay_base_url=self.SEVERPAY_BASE_URL, severpay_lifetime_minutes=self.SEVERPAY_LIFETIME_MINUTES, cryptopay_enabled=self.CRYPTOPAY_ENABLED, cryptopay_token=self.CRYPTOPAY_TOKEN, cryptopay_network=self.CRYPTOPAY_NETWORK, cryptopay_currency_type=self.CRYPTOPAY_CURRENCY_TYPE, cryptopay_asset=self.CRYPTOPAY_ASSET, ) @computed_field @property def email_settings(self) -> EmailSettings: return EmailSettings( smtp_host=self.SMTP_HOST, smtp_port=self.SMTP_PORT, smtp_fallback_ports=self.SMTP_FALLBACK_PORTS, smtp_timeout_seconds=self.SMTP_TIMEOUT_SECONDS, smtp_username=self.SMTP_USERNAME, smtp_password=self.SMTP_PASSWORD, smtp_from_email=self.SMTP_FROM_EMAIL, smtp_from_name=self.SMTP_FROM_NAME, smtp_starttls=self.SMTP_STARTTLS, smtp_use_ssl=self.SMTP_USE_SSL, email_code_ttl_seconds=self.EMAIL_CODE_TTL_SECONDS, email_code_resend_seconds=self.EMAIL_CODE_RESEND_SECONDS, email_code_max_attempts=self.EMAIL_CODE_MAX_ATTEMPTS, brute_force_max_failures=self.BRUTE_FORCE_MAX_FAILURES, brute_force_window_seconds=self.BRUTE_FORCE_WINDOW_SECONDS, brute_force_lock_seconds=self.BRUTE_FORCE_LOCK_SECONDS, ) @computed_field @property def webapp_settings(self) -> WebAppSettings: return WebAppSettings( title=self.WEBAPP_TITLE, primary_color=self.WEBAPP_PRIMARY_COLOR, logo_url=self.WEBAPP_LOGO_URL, logo_emoji=self.WEBAPP_LOGO_EMOJI, logo_emoji_font=self.WEBAPP_LOGO_EMOJI_FONT, session_ttl_seconds=self.WEBAPP_SESSION_TTL_SECONDS, session_secret=self.WEBAPP_SESSION_SECRET, webhook_secret_token=self.WEBHOOK_SECRET_TOKEN, auth_max_age_seconds=self.WEBAPP_AUTH_MAX_AGE_SECONDS, login_token_ttl_seconds=self.WEBAPP_LOGIN_TOKEN_TTL_SECONDS, server_host=self.WEBAPP_SERVER_HOST, server_port=self.WEBAPP_SERVER_PORT, enabled=self.WEBAPP_ENABLED, trusted_proxies=self.trusted_proxies, ) @computed_field @property def ADMIN_IDS(self) -> List[int]: if self.ADMIN_IDS_STR: try: return [ int(admin_id.strip()) for admin_id in self.ADMIN_IDS_STR.split(",") if admin_id.strip().isdigit() ] except ValueError: logging.error( f"Invalid ADMIN_IDS_STR format: '{self.ADMIN_IDS_STR}'. Expected comma-separated integers." # noqa: E501 ) return [] return [] @computed_field @property def PRIMARY_ADMIN_ID(self) -> Optional[int]: ids = self.ADMIN_IDS return ids[0] if ids else None @computed_field @property def trial_traffic_limit_bytes(self) -> int: if self.TRIAL_TRAFFIC_LIMIT_GB is None or self.TRIAL_TRAFFIC_LIMIT_GB <= 0: return 0 return int(self.TRIAL_TRAFFIC_LIMIT_GB * (1024**3)) @computed_field @property def user_traffic_limit_bytes(self) -> int: if self.USER_TRAFFIC_LIMIT_GB is None or self.USER_TRAFFIC_LIMIT_GB <= 0: return 0 return int(self.USER_TRAFFIC_LIMIT_GB * (1024**3)) @computed_field @property def parsed_user_squad_uuids(self) -> Optional[List[str]]: if self.USER_SQUAD_UUIDS: return [uuid.strip() for uuid in self.USER_SQUAD_UUIDS.split(",") if uuid.strip()] return None @computed_field @property def parsed_user_external_squad_uuid(self) -> Optional[str]: if self.USER_EXTERNAL_SQUAD_UUID: cleaned = self.USER_EXTERNAL_SQUAD_UUID.strip() if cleaned: return cleaned return None @computed_field @property def trusted_proxies(self) -> List[str]: return _split_csv(self.TRUSTED_PROXIES) @computed_field @property def freekassa_trusted_ips(self) -> List[str]: return _split_csv(self.FREEKASSA_TRUSTED_IPS) @computed_field @property def telegram_webhook_path(self) -> str: return "/tg/webhook" @computed_field @property def yookassa_webhook_path(self) -> str: return "/webhook/yookassa" @computed_field @property def yookassa_full_webhook_url(self) -> Optional[str]: base = self.WEBHOOK_BASE_URL if base: return f"{base.rstrip('/')}{self.yookassa_webhook_path}" return None @computed_field @property def panel_webhook_path(self) -> str: return "/webhook/panel" @computed_field @property def panel_full_webhook_url(self) -> Optional[str]: base = self.WEBHOOK_BASE_URL if base: return f"{base.rstrip('/')}{self.panel_webhook_path}" return None @computed_field @property def cryptopay_webhook_path(self) -> str: return "/webhook/cryptopay" @computed_field @property def cryptopay_full_webhook_url(self) -> Optional[str]: base = self.WEBHOOK_BASE_URL if base: return f"{base.rstrip('/')}{self.cryptopay_webhook_path}" return None @computed_field @property def freekassa_webhook_path(self) -> str: return "/webhook/freekassa" @computed_field @property def freekassa_full_webhook_url(self) -> Optional[str]: base = self.WEBHOOK_BASE_URL if base: return f"{base.rstrip('/')}{self.freekassa_webhook_path}" return None @computed_field @property def severpay_webhook_path(self) -> str: return "/webhook/severpay" @computed_field @property def severpay_full_webhook_url(self) -> Optional[str]: base = self.WEBHOOK_BASE_URL if base: return f"{base.rstrip('/')}{self.severpay_webhook_path}" return None @computed_field @property def platega_webhook_path(self) -> str: return "/webhook/platega" @computed_field @property def platega_full_webhook_url(self) -> Optional[str]: base = self.WEBHOOK_BASE_URL if base: return f"{base.rstrip('/')}{self.platega_webhook_path}" return None # Computed YooKassa receipt fields based on recurring toggle @computed_field @property def yk_receipt_payment_mode(self) -> str: # If autopayments are enabled, use service; otherwise full prepayment return "service" if self.YOOKASSA_AUTOPAYMENTS_ENABLED else "full_prepayment" @computed_field @property def yk_receipt_payment_subject(self) -> str: # If autopayments are enabled, use full_payment; otherwise payment return "full_payment" if self.YOOKASSA_AUTOPAYMENTS_ENABLED else "payment" @computed_field @property def subscription_options(self) -> Dict[int, float]: options: Dict[int, float] = {} if self.MONTH_1_ENABLED and self.RUB_PRICE_1_MONTH is not None: options[1] = float(self.RUB_PRICE_1_MONTH) if self.MONTH_3_ENABLED and self.RUB_PRICE_3_MONTHS is not None: options[3] = float(self.RUB_PRICE_3_MONTHS) if self.MONTH_6_ENABLED and self.RUB_PRICE_6_MONTHS is not None: options[6] = float(self.RUB_PRICE_6_MONTHS) if self.MONTH_12_ENABLED and self.RUB_PRICE_12_MONTHS is not None: options[12] = float(self.RUB_PRICE_12_MONTHS) return options @computed_field @property def stars_subscription_options(self) -> Dict[int, int]: options: Dict[int, int] = {} if self.STARS_ENABLED and self.MONTH_1_ENABLED and self.STARS_PRICE_1_MONTH is not None: options[1] = self.STARS_PRICE_1_MONTH if self.STARS_ENABLED and self.MONTH_3_ENABLED and self.STARS_PRICE_3_MONTHS is not None: options[3] = self.STARS_PRICE_3_MONTHS if self.STARS_ENABLED and self.MONTH_6_ENABLED and self.STARS_PRICE_6_MONTHS is not None: options[6] = self.STARS_PRICE_6_MONTHS if self.STARS_ENABLED and self.MONTH_12_ENABLED and self.STARS_PRICE_12_MONTHS is not None: options[12] = self.STARS_PRICE_12_MONTHS return options @computed_field @property def traffic_packages(self) -> Dict[float, float]: """ Mapping of traffic size in GB to price in the default currency. """ packages: Dict[float, float] = {} raw = (self.TRAFFIC_PACKAGES or "").strip() if not raw: return packages for part in raw.split(","): chunk = part.strip() if not chunk or ":" not in chunk: continue size_str, price_str = chunk.split(":", 1) try: size_gb = float(size_str.strip()) price_val = float(price_str.strip()) if size_gb > 0 and price_val >= 0: packages[size_gb] = price_val except ValueError: logging.warning("Invalid TRAFFIC_PACKAGES entry skipped: %s", chunk) continue return packages @computed_field @property def stars_traffic_packages(self) -> Dict[float, int]: """ Mapping of traffic size in GB to price in Telegram Stars. """ packages: Dict[float, int] = {} raw = (self.STARS_TRAFFIC_PACKAGES or "").strip() if not raw: return packages for part in raw.split(","): chunk = part.strip() if not chunk or ":" not in chunk: continue size_str, price_str = chunk.split(":", 1) try: size_gb = float(size_str.strip()) price_val = int(float(price_str.strip())) if size_gb > 0 and price_val >= 0: packages[size_gb] = price_val except ValueError: logging.warning("Invalid STARS_TRAFFIC_PACKAGES entry skipped: %s", chunk) continue return packages @computed_field @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 tariff_traffic_warning_levels(self) -> List[int]: levels: List[int] = [] for part in (self.TARIFF_TRAFFIC_WARNING_LEVELS or "").split(","): chunk = part.strip() if not chunk: continue try: level = int(float(chunk)) except ValueError: logging.warning("Invalid TARIFF_TRAFFIC_WARNING_LEVELS entry skipped: %s", chunk) continue if 0 < level < 100 and level not in levels: levels.append(level) return sorted(levels) or [85, 90, 95] @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]: bonuses: Dict[int, int] = {} if self.REFERRAL_BONUS_DAYS_INVITER_1_MONTH is not None: bonuses[1] = self.REFERRAL_BONUS_DAYS_INVITER_1_MONTH if self.REFERRAL_BONUS_DAYS_INVITER_3_MONTHS is not None: bonuses[3] = self.REFERRAL_BONUS_DAYS_INVITER_3_MONTHS if self.REFERRAL_BONUS_DAYS_INVITER_6_MONTHS is not None: bonuses[6] = self.REFERRAL_BONUS_DAYS_INVITER_6_MONTHS if self.REFERRAL_BONUS_DAYS_INVITER_12_MONTHS is not None: bonuses[12] = self.REFERRAL_BONUS_DAYS_INVITER_12_MONTHS return bonuses @computed_field @property def referral_bonus_referee(self) -> Dict[int, int]: bonuses: Dict[int, int] = {} if self.REFERRAL_BONUS_DAYS_REFEREE_1_MONTH is not None: bonuses[1] = self.REFERRAL_BONUS_DAYS_REFEREE_1_MONTH if self.REFERRAL_BONUS_DAYS_REFEREE_3_MONTHS is not None: bonuses[3] = self.REFERRAL_BONUS_DAYS_REFEREE_3_MONTHS if self.REFERRAL_BONUS_DAYS_REFEREE_6_MONTHS is not None: bonuses[6] = self.REFERRAL_BONUS_DAYS_REFEREE_6_MONTHS if self.REFERRAL_BONUS_DAYS_REFEREE_12_MONTHS is not None: bonuses[12] = self.REFERRAL_BONUS_DAYS_REFEREE_12_MONTHS return bonuses @computed_field @property def yookassa_autopayments_active(self) -> bool: """Autopay features are available only when YooKassa itself is enabled.""" return bool(self.YOOKASSA_ENABLED and self.YOOKASSA_AUTOPAYMENTS_ENABLED) @computed_field @property def payment_methods_order(self) -> List[str]: """ Ordered list of payment providers to show in the subscription payment keyboard. """ default_order = [ "freekassa", "platega_sbp", "platega_crypto", "severpay", "yookassa", "stars", "cryptopay", ] if not self.PAYMENT_METHODS_ORDER: return default_order methods: List[str] = [] for item in self.PAYMENT_METHODS_ORDER.split(","): slug = item.strip().lower() if not slug: continue if slug == "platega": # Legacy slug — expand to the new sub-methods preserving order if "platega_sbp" not in methods: methods.append("platega_sbp") if "platega_crypto" not in methods: methods.append("platega_crypto") continue methods.append(slug) return methods or default_order @computed_field @property def platega_sbp_method_resolved(self) -> int: """SBP method ID, falling back to legacy PLATEGA_PAYMENT_METHOD when SBP-specific value is the default.""" # noqa: E501 if self.PLATEGA_SBP_METHOD != 2: return self.PLATEGA_SBP_METHOD return self.PLATEGA_PAYMENT_METHOD or 2 @computed_field @property def email_auth_configured(self) -> bool: return bool( self.SMTP_HOST and self.SMTP_PORT and self.SMTP_USERNAME and self.SMTP_PASSWORD and self.SMTP_FROM_EMAIL ) @computed_field @property def smtp_ports_to_try(self) -> List[int]: ports: List[int] = [] def add_port(value: Any) -> None: try: port = int(str(value).strip()) except (TypeError, ValueError): return if 0 < port <= 65535 and port not in ports: ports.append(port) add_port(self.SMTP_PORT) for item in (self.SMTP_FALLBACK_PORTS or "").split(","): add_port(item) return ports # Logging Configuration LOG_LEVEL: str = Field( default="INFO", description="Global log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)", ) LOG_CHAT_ID: Optional[int] = Field( default=None, description="Telegram chat/group ID for sending notifications" ) LOG_THREAD_ID: Optional[int] = Field( default=None, description="Thread ID for supergroup messages (optional)" ) @field_validator("LOG_LEVEL", mode="before") @classmethod def normalize_log_level(cls, v): if isinstance(v, str): v = v.strip().upper() if not v: return "INFO" return v @field_validator("POSTGRES_USER", "POSTGRES_PASSWORD", mode="before") @classmethod def validate_required_db_credentials(cls, v): if isinstance(v, str): v = v.strip() if not v: raise ValueError("must not be empty") return v @field_validator("WEBAPP_SESSION_SECRET", "WEBHOOK_SECRET_TOKEN", mode="before") @classmethod def normalize_webapp_secrets(cls, v): if isinstance(v, str): v = v.strip() if v: return v if v: return v return secrets.token_urlsafe(32) @field_validator("LOG_CHAT_ID", "LOG_THREAD_ID", mode="before") @classmethod def validate_optional_int_fields(cls, v): """Convert empty strings to None for optional integer fields""" if isinstance(v, str) and v.strip() == "": return None return v @field_validator( "REQUIRED_CHANNEL_LINK", "PLATEGA_RETURN_URL", "PLATEGA_FAILED_URL", "SEVERPAY_RETURN_URL", "CRYPT4_REDIRECT_URL", "PRIVACY_POLICY_URL", "USER_AGREEMENT_URL", "SUBSCRIPTION_MINI_APP_URL", "WEBAPP_LOGO_URL", "TELEGRAM_OAUTH_CLIENT_SECRET", "TELEGRAM_OAUTH_REQUEST_ACCESS", "SMTP_USERNAME", "SMTP_PASSWORD", "SMTP_FROM_EMAIL", "SMTP_FROM_NAME", "SMTP_FALLBACK_PORTS", mode="before", ) @classmethod def sanitize_optional_link(cls, v): if isinstance(v, str) and not v.strip(): return None return v @field_validator( "USER_HWID_DEVICE_LIMIT", "SEVERPAY_MID", "SEVERPAY_LIFETIME_MINUTES", mode="before" ) @classmethod def validate_optional_int(cls, v): if isinstance(v, str): v = v.strip() if not v: return None return v # Notification types LOG_NEW_USERS: bool = Field( default=True, description="Send notifications for new user registrations" ) LOG_PAYMENTS: bool = Field( default=True, description="Send notifications for successful payments" ) LOG_PROMO_ACTIVATIONS: bool = Field( default=True, description="Send notifications for promo code activations" ) LOG_TRIAL_ACTIVATIONS: bool = Field( default=True, description="Send notifications for trial activations" ) LOG_SUSPICIOUS_ACTIVITY: bool = Field( default=True, description="Send notifications for suspicious promo attempts" ) model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore", populate_by_name=True ) _settings_instance: Optional[Settings] = None def get_settings() -> Settings: global _settings_instance if _settings_instance is None: try: _settings_instance = Settings() if not _settings_instance.ADMIN_IDS: logging.warning( "CRITICAL: ADMIN_IDS not set or contains no valid integer IDs in .env. " "Admin functionality will be restricted." ) if not _settings_instance.PANEL_API_URL: logging.warning( "CRITICAL: PANEL_API_URL is not set. Panel integration will not work." ) if not os.getenv("WEBAPP_SESSION_SECRET"): logging.warning( "WEBAPP_SESSION_SECRET is not set. A generated secret will be used for this process only." # noqa: E501 ) if not os.getenv("WEBHOOK_SECRET_TOKEN"): logging.warning( "WEBHOOK_SECRET_TOKEN is not set. A generated secret will be used for this process only." # noqa: E501 ) if ( not _settings_instance.YOOKASSA_SHOP_ID or not _settings_instance.YOOKASSA_SECRET_KEY ): logging.warning( "CRITICAL: YooKassa credentials (SHOP_ID or SECRET_KEY) are not set. Payments will not work." # noqa: E501 ) if (_settings_instance.LKNPD_INN or _settings_instance.LKNPD_PASSWORD) and not ( _settings_instance.LKNPD_INN and _settings_instance.LKNPD_PASSWORD ): logging.warning( "WARNING: LKNPD credentials are incomplete. Receipt sending will be disabled." ) if _settings_instance.FREEKASSA_ENABLED: if ( not _settings_instance.FREEKASSA_MERCHANT_ID or not _settings_instance.FREEKASSA_API_KEY ): logging.warning( "CRITICAL: FreeKassa is enabled but SHOP_ID or API key is missing. FreeKassa payments will not work." # noqa: E501 ) if not _settings_instance.FREEKASSA_SECOND_SECRET: logging.warning( "WARNING: FreeKassa second secret is not set. Incoming payment notifications cannot be verified." # noqa: E501 ) if not _settings_instance.subscription_options: logging.warning( "CRITICAL: FreeKassa is enabled but no subscription prices are configured (RUB_PRICE_*). Users will not see payment buttons." # noqa: E501 ) if _settings_instance.PLATEGA_ENABLED: if ( not _settings_instance.PLATEGA_MERCHANT_ID or not _settings_instance.PLATEGA_SECRET ): logging.warning( "CRITICAL: Platega is enabled but merchant credentials (PLATEGA_MERCHANT_ID/PLATEGA_SECRET) are missing. Platega payments will not work." # noqa: E501 ) if _settings_instance.SEVERPAY_ENABLED: if not _settings_instance.SEVERPAY_MID or not _settings_instance.SEVERPAY_TOKEN: logging.warning( "CRITICAL: SeverPay is enabled but MID or TOKEN is missing. SeverPay payments will not work." # noqa: E501 ) except ValidationError as e: logging.critical(f"Pydantic validation error while loading settings: {e}") raise SystemExit( f"CRITICAL SETTINGS ERROR: {e}. Please check your .env file and Settings model." ) return _settings_instance