RemnaWaveBOT/app/config.py

from __future__ import annotations

from dataclasses import dataclass
from functools import cached_property
from urllib.parse import quote_plus, urlparse

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


@dataclass(frozen=True, slots=True)
class PaymentPlanSettings:
    code: str
    days: int
    amount_rub: int
    title: str = ""
    description: str = ""
    squad_groups: tuple[str, ...] = ()


PAYMENT_PRODUCT_PLAN_SPECS: dict[str, tuple[str, str, tuple[str, ...]]] = {
    "vpn_white": (
        "VPN + Белые списки",
        "VPN и белые списки в одном тарифе",
        ("vpn", "white"),
    ),
    "white": (
        "Белые списки",
        "Доступ только к белым спискам",
        ("white",),
    ),
    "vpn": (
        "VPN",
        "Доступ только к VPN без белых списков",
        ("vpn",),
    ),
}


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    bot_token: str = Field(validation_alias="BOT_TOKEN")
    bot_admin_ids_raw: str = Field(default="", validation_alias="BOT_ADMIN_IDS")
    bot_moderator_ids_raw: str = Field(default="", validation_alias="BOT_MODERATOR_IDS")
    telegram_proxy_url: str = Field(default="", validation_alias="TELEGRAM_PROXY_URL")
    bot_brand_name: str = Field(default="OREOL VPN", validation_alias="BOT_BRAND_NAME")
    bot_public_username: str = Field(default="", validation_alias="BOT_PUBLIC_USERNAME")
    bot_support_url: str = Field(default="", validation_alias="BOT_SUPPORT_URL")
    bot_terms_url: str = Field(default="", validation_alias="BOT_TERMS_URL")
    bot_support_ticket_link: str = Field(default="", validation_alias="BOT_SUPPORT_TICKET_LINK")
    bot_support_ticket_chat_id_raw: str = Field(
        default="",
        validation_alias="BOT_SUPPORT_TICKET_CHAT_ID",
    )
    bot_support_ticket_thread_id: int = Field(
        default=0,
        validation_alias="BOT_SUPPORT_TICKET_THREAD_ID",
    )
    bot_start_image_enabled: bool = Field(
        default=True,
        validation_alias="BOT_START_IMAGE_ENABLED",
    )
    bot_start_image_path: str = Field(default="", validation_alias="BOT_START_IMAGE_PATH")

    remnawave_base_url: str = Field(validation_alias="REMNAWAVE_BASE_URL")
    remnawave_api_token: str = Field(default="", validation_alias="REMNAWAVE_API_TOKEN")
    remnawave_caddy_api_key: str = Field(default="", validation_alias="REMNAWAVE_CADDY_API_KEY")
    remnawave_timeout_seconds: float = Field(
        default=20.0,
        validation_alias="REMNAWAVE_TIMEOUT_SECONDS",
    )

    db_host: str = Field(default="127.0.0.1", validation_alias="DB_HOST")
    db_port: int = Field(default=3306, validation_alias="DB_PORT")
    db_name: str = Field(default="telegabot", validation_alias="DB_NAME")
    db_user: str = Field(default="telegabot", validation_alias="DB_USER")
    db_password: str = Field(default="", validation_alias="DB_PASSWORD")
    db_echo: bool = Field(default=False, validation_alias="DB_ECHO")
    create_database_on_start: bool = Field(
        default=True,
        validation_alias="CREATE_DATABASE_ON_START",
    )
    create_tables_on_start: bool = Field(
        default=True,
        validation_alias="CREATE_TABLES_ON_START",
    )
    sync_subscription_history: bool = Field(
        default=True,
        validation_alias="SYNC_SUBSCRIPTION_HISTORY",
    )
    sync_batch_size: int = Field(default=100, validation_alias="SYNC_BATCH_SIZE")
    log_level: str = Field(default="INFO", validation_alias="LOG_LEVEL")

    payment_plans_raw: str = Field(
        default="30:250,180:600,365:1000",
        validation_alias="PAYMENT_PLANS",
    )
    payment_product_plans_raw: str = Field(
        default="",
        validation_alias="PAYMENT_PRODUCT_PLANS",
    )
    payment_transfer_text: str = Field(
        default="Переведите оплату по указанным реквизитам и отправьте чек в бот.",
        validation_alias="PAYMENT_TRANSFER_TEXT",
    )
    payment_success_text: str = Field(
        default="Доступ выдан. Сохраните ссылку и используйте её для подключения.",
        validation_alias="PAYMENT_SUCCESS_TEXT",
    )
    payment_review_link: str = Field(default="", validation_alias="PAYMENT_REVIEW_LINK")
    payment_review_chat_id_raw: str = Field(
        default="",
        validation_alias="PAYMENT_REVIEW_CHAT_ID",
    )
    payment_review_thread_id: int = Field(
        default=0,
        validation_alias="PAYMENT_REVIEW_THREAD_ID",
    )
    broadcast_results_link: str = Field(
        default="https://t.me/c/3646494169/67/68",
        validation_alias="BROADCAST_RESULTS_LINK",
    )
    broadcast_results_chat_id_raw: str = Field(
        default="",
        validation_alias="BROADCAST_RESULTS_CHAT_ID",
    )
    broadcast_results_thread_id: int = Field(
        default=0,
        validation_alias="BROADCAST_RESULTS_THREAD_ID",
    )
    referral_discount_percent: int = Field(
        default=5,
        validation_alias="REFERRAL_DISCOUNT_PERCENT",
    )
    referral_bonus_days: int = Field(
        default=7,
        validation_alias="REFERRAL_BONUS_DAYS",
    )
    payment_plan_code: str = Field(default="vpn_30d", validation_alias="PAYMENT_PLAN_CODE")
    payment_plan_title: str = Field(
        default="OREOL VPN на 30 дней",
        validation_alias="PAYMENT_PLAN_TITLE",
    )
    payment_plan_description: str = Field(
        default="Доступ к VPN на 30 дней",
        validation_alias="PAYMENT_PLAN_DESCRIPTION",
    )
    payment_plan_price_stars: int = Field(default=500, validation_alias="PAYMENT_PLAN_PRICE_STARS")
    payment_plan_duration_days: int = Field(
        default=30,
        validation_alias="PAYMENT_PLAN_DURATION_DAYS",
    )
    payment_plan_traffic_limit_gb: int = Field(
        default=0,
        validation_alias="PAYMENT_PLAN_TRAFFIC_LIMIT_GB",
    )
    payment_plan_traffic_reset_period: str = Field(
        default="NO_RESET",
        validation_alias="PAYMENT_PLAN_TRAFFIC_RESET_PERIOD",
    )
    payment_internal_squad_uuids_raw: str = Field(
        default="",
        validation_alias="PAYMENT_INTERNAL_SQUAD_UUIDS",
    )
    payment_vpn_squad_uuids_raw: str = Field(
        default="",
        validation_alias="PAYMENT_VPN_SQUAD_UUIDS",
    )
    payment_white_squad_uuids_raw: str = Field(
        default="",
        validation_alias="PAYMENT_WHITE_SQUAD_UUIDS",
    )
    payment_external_squad_uuid: str = Field(
        default="",
        validation_alias="PAYMENT_EXTERNAL_SQUAD_UUID",
    )
    payment_support_text: str = Field(
        default="Если оплата прошла, но доступ не выдался, напишите в поддержку.",
        validation_alias="PAYMENT_SUPPORT_TEXT",
    )
    payment_username_prefix: str = Field(
        default="Oreol",
        validation_alias="PAYMENT_USERNAME_PREFIX",
    )
    payment_user_tag: str = Field(default="BOT", validation_alias="PAYMENT_USER_TAG")

    @cached_property
    def bot_admin_ids(self) -> set[int]:
        values: set[int] = set()
        for raw_part in self.bot_admin_ids_raw.split(","):
            part = raw_part.strip()
            if part:
                values.add(int(part))
        return values

    @cached_property
    def bot_moderator_ids(self) -> set[int]:
        values: set[int] = set()
        for raw_part in self.bot_moderator_ids_raw.split(","):
            part = raw_part.strip()
            if part:
                values.add(int(part))
        return values

    @property
    def database_url(self) -> str:
        user = quote_plus(self.db_user)
        password = quote_plus(self.db_password)
        return (
            f"mysql+aiomysql://{user}:{password}"
            f"@{self.db_host}:{self.db_port}/{self.db_name}?charset=utf8mb4"
        )

    @property
    def database_server_url(self) -> str:
        user = quote_plus(self.db_user)
        password = quote_plus(self.db_password)
        return (
            f"mysql+aiomysql://{user}:{password}"
            f"@{self.db_host}:{self.db_port}/?charset=utf8mb4"
        )

    @cached_property
    def support_ticket_chat_id(self) -> int | None:
        if self.bot_support_ticket_chat_id_raw.strip():
            return int(self.bot_support_ticket_chat_id_raw.strip())

        parsed_chat_id, _ = self._parse_private_topic_link(self.bot_support_ticket_link)
        return parsed_chat_id

    @cached_property
    def support_ticket_message_thread_id(self) -> int | None:
        if self.bot_support_ticket_thread_id > 0:
            return self.bot_support_ticket_thread_id

        _, parsed_thread_id = self._parse_private_topic_link(self.bot_support_ticket_link)
        return parsed_thread_id

    @cached_property
    def payment_review_chat_id(self) -> int | None:
        if self.payment_review_chat_id_raw.strip():
            return int(self.payment_review_chat_id_raw.strip())

        parsed_chat_id, _ = self._parse_private_topic_link(self.payment_review_link)
        return parsed_chat_id

    @cached_property
    def payment_review_message_thread_id(self) -> int | None:
        if self.payment_review_thread_id > 0:
            return self.payment_review_thread_id

        _, parsed_thread_id = self._parse_private_topic_link(self.payment_review_link)
        return parsed_thread_id

    @cached_property
    def broadcast_results_chat_id(self) -> int | None:
        if self.broadcast_results_chat_id_raw.strip():
            return int(self.broadcast_results_chat_id_raw.strip())

        parsed_chat_id, _ = self._parse_private_topic_link(self.broadcast_results_link)
        return parsed_chat_id

    @cached_property
    def broadcast_results_message_thread_id(self) -> int | None:
        if self.broadcast_results_thread_id > 0:
            return self.broadcast_results_thread_id

        _, parsed_thread_id = self._parse_private_topic_link(self.broadcast_results_link)
        return parsed_thread_id

    @staticmethod
    def _parse_private_topic_link(link: str) -> tuple[int | None, int | None]:
        cleaned = link.strip()
        if not cleaned:
            return None, None

        parsed = urlparse(cleaned)
        if parsed.netloc not in {"t.me", "telegram.me", "www.t.me", "www.telegram.me"}:
            return None, None

        parts = [part for part in parsed.path.split("/") if part]
        if len(parts) < 3 or parts[0] != "c" or not parts[1].isdigit():
            return None, None

        chat_id = int(f"-100{parts[1]}")
        thread_id = int(parts[2]) if len(parts) >= 4 and parts[2].isdigit() else None
        return chat_id, thread_id

    def is_admin(self, telegram_id: int | None) -> bool:
        return telegram_id is not None and telegram_id in self.bot_admin_ids

    def is_moderator(self, telegram_id: int | None) -> bool:
        return telegram_id is not None and telegram_id in self.bot_moderator_ids

    @property
    def bot_public_username_normalized(self) -> str:
        return self.bot_public_username.strip().removeprefix("@")

    @cached_property
    def payment_internal_squad_uuids(self) -> list[str]:
        return self._parse_csv_list(self.payment_internal_squad_uuids_raw)

    @cached_property
    def payment_vpn_squad_uuids(self) -> list[str]:
        return self._parse_csv_list(self.payment_vpn_squad_uuids_raw)

    @cached_property
    def payment_white_squad_uuids(self) -> list[str]:
        return self._parse_csv_list(self.payment_white_squad_uuids_raw)

    @cached_property
    def payment_external_squad_uuid_normalized(self) -> str | None:
        value = self.payment_external_squad_uuid.strip()
        return value or None

    @cached_property
    def payment_username_prefix_normalized(self) -> str:
        normalized = "".join(
            ch for ch in self.payment_username_prefix.strip() if ch.isalnum() or ch in "_-"
        )
        return normalized[:12] or "Oreol"

    @cached_property
    def payment_user_tag_normalized(self) -> str | None:
        normalized = "".join(
            ch for ch in self.payment_user_tag.strip().upper() if ch.isalnum() or ch == "_"
        )
        return normalized[:16] or None

    @cached_property
    def payment_plans(self) -> list[PaymentPlanSettings]:
        product_plans = self._parse_product_plans(
            self.payment_product_plans_raw,
            default_days=max(self.payment_plan_duration_days, 1),
        )
        if product_plans:
            return product_plans

        values: list[PaymentPlanSettings] = []
        for raw_part in self.payment_plans_raw.split(","):
            part = raw_part.strip()
            if not part:
                continue

            fragments = [fragment.strip() for fragment in part.split(":")]
            if len(fragments) != 2 or not fragments[0].isdigit() or not fragments[1].isdigit():
                continue

            days = int(fragments[0])
            amount_rub = int(fragments[1])
            if days <= 0 or amount_rub <= 0:
                continue

            values.append(
                PaymentPlanSettings(
                    code=f"{days}d",
                    days=days,
                    amount_rub=amount_rub,
                )
            )

        if values:
            return sorted(values, key=lambda item: item.days)

        fallback_amount = max(self.payment_plan_price_stars, 0)
        fallback_days = max(self.payment_plan_duration_days, 0)
        if fallback_amount > 0 and fallback_days > 0:
            return [
                PaymentPlanSettings(
                    code=self.payment_plan_code.strip() or f"{fallback_days}d",
                    days=fallback_days,
                    amount_rub=fallback_amount,
                )
            ]

        return []

    @staticmethod
    def _parse_csv_list(raw_value: str) -> list[str]:
        values: list[str] = []
        for raw_part in raw_value.split(","):
            part = raw_part.strip()
            if part:
                values.append(part)
        return values

    @staticmethod
    def _parse_product_plans(
        raw_value: str,
        *,
        default_days: int,
    ) -> list[PaymentPlanSettings]:
        values: list[PaymentPlanSettings] = []
        seen_codes: set[str] = set()

        for raw_part in raw_value.split(","):
            part = raw_part.strip()
            if not part:
                continue

            fragments = [fragment.strip() for fragment in part.split(":")]

            # Legacy format: code:amount (2 fields) → single duration using default_days
            if len(fragments) == 2:
                code, raw_amount = fragments
                pairs = [(str(default_days), raw_amount)]
            # Standard format: code:days:amount (3 fields) → single duration
            elif len(fragments) == 3:
                code = fragments[0]
                pairs = [(fragments[1], fragments[2])]
            # Extended format: code:days1:amount1:days2:amount2:... (odd number ≥ 5)
            elif len(fragments) >= 5 and len(fragments) % 2 == 1:
                code = fragments[0]
                pairs = [
                    (fragments[i], fragments[i + 1])
                    for i in range(1, len(fragments), 2)
                ]
            else:
                continue

            normalized_code = code.lower()
            spec = PAYMENT_PRODUCT_PLAN_SPECS.get(normalized_code)
            if spec is None or normalized_code in seen_codes:
                continue

            title, description, squad_groups = spec
            use_duration_suffix = len(pairs) > 1
            for raw_days, raw_amount in pairs:
                if not raw_days.isdigit() or not raw_amount.isdigit():
                    continue

                days = int(raw_days)
                amount_rub = int(raw_amount)
                if days <= 0 or amount_rub <= 0:
                    continue

                plan_code = f"{normalized_code}_{days}" if use_duration_suffix else normalized_code
                values.append(
                    PaymentPlanSettings(
                        code=plan_code,
                        days=days,
                        amount_rub=amount_rub,
                        title=title,
                        description=description,
                        squad_groups=squad_groups,
                    )
                )

            seen_codes.add(normalized_code)

        return values