349 lines
14 KiB
Python
349 lines
14 KiB
Python
"""Load settings from environment / .env file."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from urllib.parse import quote_plus
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
_ENV_FILE = (os.getenv("ENV_FILE") or "").strip()
|
|
load_dotenv(_ENV_FILE or None)
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parent
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class Settings:
|
|
remnawave_base_url: str
|
|
remnawave_token: str
|
|
bot_token: str
|
|
database_url: str
|
|
admin_ids: tuple[int, ...] = ()
|
|
caddy_token: str | None = None
|
|
remnawave_username: str = ""
|
|
remnawave_password: str = ""
|
|
tariffs_path: str = str(ROOT_DIR / "tariffs.json")
|
|
crypto_bot_token: str = ""
|
|
crypto_pay_testnet: bool = False
|
|
heleket_merchant_id: str = ""
|
|
heleket_api_key: str = ""
|
|
heleket_callback_url: str = ""
|
|
heleket_currency: str = "RUB"
|
|
digiseller_seller_id: str = ""
|
|
digiseller_api_key: str = ""
|
|
digiseller_product_ids: tuple[int, ...] = ()
|
|
digiseller_pay_url: str = ""
|
|
digiseller_credit_from: str = "cnt_goods"
|
|
digiseller_callback_url: str = ""
|
|
lava_shop_id: str = ""
|
|
lava_secret_key: str = ""
|
|
lava_additional_key: str = ""
|
|
lava_hook_url: str = ""
|
|
lava_success_url: str = ""
|
|
lava_fail_url: str = ""
|
|
lava_expire_minutes: int = 60
|
|
anypay_api_id: str = ""
|
|
anypay_api_key: str = ""
|
|
anypay_project_id: str = ""
|
|
anypay_currency: str = "RUB"
|
|
anypay_method: str = ""
|
|
anypay_success_url: str = ""
|
|
anypay_fail_url: str = ""
|
|
anypay_email: str = ""
|
|
anypay_check_ip: bool = False
|
|
webapp_url: str = ""
|
|
webapp_host: str = "0.0.0.0"
|
|
webapp_port: int = 8080
|
|
topup_code_pay_url: str = ""
|
|
web_admin_login: str = ""
|
|
web_admin_password: str = ""
|
|
stars_enabled: bool = True
|
|
stars_rub_per_star: float = 1.5
|
|
smtp_host: str = ""
|
|
smtp_port: int = 587
|
|
smtp_user: str = ""
|
|
smtp_password: str = ""
|
|
smtp_from: str = ""
|
|
smtp_tls: bool = True
|
|
smtp_ssl: bool = False
|
|
telegram_client_id: str = ""
|
|
telegram_webhook_enabled: bool = False
|
|
telegram_webhook_path: str = "/telegram/webhook"
|
|
telegram_webhook_secret: str = ""
|
|
telegram_webhook_url: str = ""
|
|
|
|
@property
|
|
def web_admin_enabled(self) -> bool:
|
|
return bool(self.web_admin_login and self.web_admin_password)
|
|
|
|
@property
|
|
def effective_telegram_client_id(self) -> str:
|
|
if (self.telegram_client_id or "").strip():
|
|
return self.telegram_client_id.strip()
|
|
from services.tg_webapp import telegram_client_id_from_token
|
|
|
|
return telegram_client_id_from_token(self.bot_token)
|
|
|
|
@property
|
|
def effective_telegram_webhook_path(self) -> str:
|
|
path = (self.telegram_webhook_path or "/telegram/webhook").strip() or "/telegram/webhook"
|
|
if not path.startswith("/"):
|
|
path = f"/{path}"
|
|
return path.rstrip("/") or "/telegram/webhook"
|
|
|
|
@property
|
|
def effective_telegram_webhook_url(self) -> str:
|
|
explicit = (self.telegram_webhook_url or "").strip().rstrip("/")
|
|
if explicit:
|
|
return explicit
|
|
base = (self.webapp_url or "").strip().rstrip("/")
|
|
if not base:
|
|
return ""
|
|
return f"{base}{self.effective_telegram_webhook_path}"
|
|
|
|
|
|
def _parse_env_secret(raw: str) -> str:
|
|
"""Значение из .env без кавычек и пробелов по краям."""
|
|
s = (raw or "").strip()
|
|
if len(s) >= 2 and s[0] == s[-1] and s[0] in "\"'":
|
|
s = s[1:-1].strip()
|
|
return s
|
|
|
|
|
|
def _parse_admin_ids(raw: str) -> tuple[int, ...]:
|
|
if not raw.strip():
|
|
return ()
|
|
result: list[int] = []
|
|
for part in raw.split(","):
|
|
part = part.strip()
|
|
if not part:
|
|
continue
|
|
result.append(int(part))
|
|
return tuple(result)
|
|
|
|
|
|
def _build_database_url() -> str:
|
|
explicit = (os.getenv("DATABASE_URL") or "").strip()
|
|
if explicit:
|
|
return explicit
|
|
|
|
user = (os.getenv("POSTGRES_USER") or "vpnbot").strip()
|
|
password = (os.getenv("POSTGRES_PASSWORD") or "vpnbot").strip()
|
|
host = (os.getenv("POSTGRES_HOST") or "db").strip()
|
|
port = (os.getenv("POSTGRES_PORT") or "5432").strip()
|
|
name = (os.getenv("POSTGRES_DB") or "vpnbot").strip()
|
|
return f"postgresql://{quote_plus(user)}:{quote_plus(password)}@{host}:{port}/{name}"
|
|
|
|
|
|
def _parse_int_ids(raw: str) -> tuple[int, ...]:
|
|
if not raw.strip():
|
|
return ()
|
|
result: list[int] = []
|
|
for part in raw.replace(";", ",").split(","):
|
|
part = part.strip()
|
|
if not part:
|
|
continue
|
|
result.append(int(part))
|
|
return tuple(result)
|
|
|
|
|
|
def get_settings(*, require_bot_token: bool = False) -> Settings:
|
|
base_url = (os.getenv("REMNAWAVE_BASE_URL") or "").rstrip("/")
|
|
token = (os.getenv("REMNAWAVE_TOKEN") or "").strip()
|
|
bot_token = (os.getenv("BOT_TOKEN") or "").strip()
|
|
caddy = (os.getenv("CADDY_TOKEN") or "").strip() or None
|
|
admin_ids = _parse_admin_ids(os.getenv("ADMIN_IDS") or "")
|
|
tariffs_path = (os.getenv("TARIFFS_PATH") or str(ROOT_DIR / "tariffs.json")).strip()
|
|
crypto_bot_token = (os.getenv("CRYPTO_BOT_TOKEN") or "").strip()
|
|
crypto_pay_testnet = (os.getenv("CRYPTO_PAY_TESTNET") or "").strip().lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
"on",
|
|
}
|
|
remnawave_username = (os.getenv("REMNAWAVE_USERNAME") or "").strip()
|
|
remnawave_password = (os.getenv("REMNAWAVE_PASSWORD") or "").strip()
|
|
webapp_url = (os.getenv("WEBAPP_URL") or "").strip().rstrip("/")
|
|
webapp_host = (os.getenv("WEBAPP_HOST") or "0.0.0.0").strip()
|
|
webapp_port = int((os.getenv("WEBAPP_PORT") or "8080").strip())
|
|
database_url = _build_database_url()
|
|
heleket_merchant_id = (os.getenv("HELEKET_MERCHANT_ID") or "").strip()
|
|
heleket_api_key = (os.getenv("HELEKET_API_KEY") or "").strip()
|
|
heleket_callback_url = (os.getenv("HELEKET_CALLBACK_URL") or "").strip().rstrip("/")
|
|
if not heleket_callback_url and webapp_url:
|
|
heleket_callback_url = f"{webapp_url}/api/payments/heleket"
|
|
heleket_currency = (os.getenv("HELEKET_CURRENCY") or "RUB").strip().upper() or "RUB"
|
|
digiseller_seller_id = (os.getenv("DIGISELLER_SELLER_ID") or "").strip()
|
|
digiseller_api_key = (os.getenv("DIGISELLER_API_KEY") or "").strip()
|
|
digiseller_product_ids = _parse_int_ids(os.getenv("DIGISELLER_PRODUCT_IDS") or "")
|
|
digiseller_pay_url = (os.getenv("DIGISELLER_PAY_URL") or "").strip()
|
|
digiseller_credit_from = (
|
|
os.getenv("DIGISELLER_CREDIT_FROM") or "cnt_goods"
|
|
).strip().lower() or "cnt_goods"
|
|
digiseller_callback_url = (
|
|
os.getenv("DIGISELLER_CALLBACK_URL") or ""
|
|
).strip().rstrip("/")
|
|
if not digiseller_callback_url and webapp_url:
|
|
digiseller_callback_url = f"{webapp_url}/api/payments/digiseller"
|
|
lava_shop_id = (os.getenv("LAVA_SHOP_ID") or "").strip()
|
|
lava_secret_key = (os.getenv("LAVA_SECRET_KEY") or "").strip()
|
|
lava_additional_key = (os.getenv("LAVA_ADDITIONAL_KEY") or "").strip()
|
|
lava_hook_url = (os.getenv("LAVA_HOOK_URL") or "").strip().rstrip("/")
|
|
if not lava_hook_url and webapp_url:
|
|
lava_hook_url = f"{webapp_url}/api/payments/lava"
|
|
lava_success_url = (os.getenv("LAVA_SUCCESS_URL") or "").strip()
|
|
if not lava_success_url and webapp_url:
|
|
lava_success_url = f"{webapp_url}/pay/lava/success"
|
|
lava_fail_url = (os.getenv("LAVA_FAIL_URL") or "").strip()
|
|
if not lava_fail_url and webapp_url:
|
|
lava_fail_url = f"{webapp_url}/pay/lava/fail"
|
|
try:
|
|
lava_expire_minutes = int((os.getenv("LAVA_EXPIRE_MINUTES") or "60").strip())
|
|
except ValueError:
|
|
lava_expire_minutes = 60
|
|
anypay_api_id = (os.getenv("ANYPAY_API_ID") or "").strip()
|
|
anypay_api_key = (os.getenv("ANYPAY_API_KEY") or "").strip()
|
|
anypay_project_id = (os.getenv("ANYPAY_PROJECT_ID") or "").strip()
|
|
anypay_currency = (os.getenv("ANYPAY_CURRENCY") or "RUB").strip().upper() or "RUB"
|
|
anypay_method = (os.getenv("ANYPAY_METHOD") or "").strip().lower()
|
|
anypay_success_url = (os.getenv("ANYPAY_SUCCESS_URL") or "").strip()
|
|
if not anypay_success_url and webapp_url:
|
|
anypay_success_url = f"{webapp_url}/pay/anypay/success"
|
|
anypay_fail_url = (os.getenv("ANYPAY_FAIL_URL") or "").strip()
|
|
if not anypay_fail_url and webapp_url:
|
|
anypay_fail_url = f"{webapp_url}/pay/anypay/fail"
|
|
anypay_email = (os.getenv("ANYPAY_EMAIL") or "").strip()
|
|
anypay_check_ip = (os.getenv("ANYPAY_CHECK_IP") or "").strip().lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
"on",
|
|
}
|
|
topup_code_pay_url = (os.getenv("TOPUP_CODE_PAY_URL") or "").strip()
|
|
web_admin_login = _parse_env_secret(os.getenv("WEB_ADMIN_LOGIN") or "")
|
|
web_admin_password = _parse_env_secret(os.getenv("WEB_ADMIN_PASSWORD") or "")
|
|
stars_enabled = (os.getenv("STARS_ENABLED") or "true").strip().lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
"on",
|
|
}
|
|
try:
|
|
stars_rub_per_star = float((os.getenv("STARS_RUB_PER_STAR") or "1.5").strip())
|
|
except ValueError:
|
|
stars_rub_per_star = 1.5
|
|
if stars_rub_per_star <= 0:
|
|
stars_rub_per_star = 1.5
|
|
smtp_host = (os.getenv("SMTP_HOST") or "").strip()
|
|
try:
|
|
smtp_port = int((os.getenv("SMTP_PORT") or "587").strip())
|
|
except ValueError:
|
|
smtp_port = 587
|
|
smtp_user = (os.getenv("SMTP_USER") or "").strip()
|
|
smtp_password = os.getenv("SMTP_PASSWORD") or ""
|
|
smtp_from = (os.getenv("SMTP_FROM") or "").strip()
|
|
smtp_tls = (os.getenv("SMTP_TLS") or "true").strip().lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
"on",
|
|
}
|
|
smtp_ssl = (os.getenv("SMTP_SSL") or "").strip().lower() in {
|
|
"1",
|
|
"true",
|
|
"yes",
|
|
"on",
|
|
}
|
|
telegram_client_id = (os.getenv("TELEGRAM_CLIENT_ID") or "").strip()
|
|
webhook_env = (os.getenv("TELEGRAM_WEBHOOK") or "").strip().lower()
|
|
if webhook_env in {"1", "true", "yes", "on"}:
|
|
telegram_webhook_enabled = True
|
|
elif webhook_env in {"0", "false", "no", "off"}:
|
|
telegram_webhook_enabled = False
|
|
else:
|
|
# По умолчанию webhook, если кабинет уже на публичном HTTPS
|
|
telegram_webhook_enabled = webapp_url.startswith("https://")
|
|
telegram_webhook_path = (os.getenv("TELEGRAM_WEBHOOK_PATH") or "/telegram/webhook").strip()
|
|
telegram_webhook_secret = _parse_env_secret(os.getenv("TELEGRAM_WEBHOOK_SECRET") or "")
|
|
if not telegram_webhook_secret and bot_token and telegram_webhook_enabled:
|
|
telegram_webhook_secret = hashlib.sha256(
|
|
f"tg-webhook:{bot_token}".encode("utf-8")
|
|
).hexdigest()[:48]
|
|
telegram_webhook_url = (os.getenv("TELEGRAM_WEBHOOK_URL") or "").strip().rstrip("/")
|
|
if not base_url:
|
|
raise SystemExit(
|
|
"REMNAWAVE_BASE_URL не задан. Скопируйте .env.example → .env и заполните."
|
|
)
|
|
if not token and not (remnawave_username and remnawave_password):
|
|
raise SystemExit(
|
|
"Задайте REMNAWAVE_TOKEN или пару REMNAWAVE_USERNAME / REMNAWAVE_PASSWORD."
|
|
)
|
|
if require_bot_token and not bot_token:
|
|
raise SystemExit(
|
|
"BOT_TOKEN не задан. Создайте бота у @BotFather и добавьте токен в .env."
|
|
)
|
|
|
|
return Settings(
|
|
remnawave_base_url=base_url,
|
|
remnawave_token=token,
|
|
bot_token=bot_token,
|
|
database_url=database_url,
|
|
admin_ids=admin_ids,
|
|
caddy_token=caddy,
|
|
remnawave_username=remnawave_username,
|
|
remnawave_password=remnawave_password,
|
|
tariffs_path=tariffs_path,
|
|
crypto_bot_token=crypto_bot_token,
|
|
crypto_pay_testnet=crypto_pay_testnet,
|
|
heleket_merchant_id=heleket_merchant_id,
|
|
heleket_api_key=heleket_api_key,
|
|
heleket_callback_url=heleket_callback_url,
|
|
heleket_currency=heleket_currency,
|
|
digiseller_seller_id=digiseller_seller_id,
|
|
digiseller_api_key=digiseller_api_key,
|
|
digiseller_product_ids=digiseller_product_ids,
|
|
digiseller_pay_url=digiseller_pay_url,
|
|
digiseller_credit_from=digiseller_credit_from,
|
|
digiseller_callback_url=digiseller_callback_url,
|
|
lava_shop_id=lava_shop_id,
|
|
lava_secret_key=lava_secret_key,
|
|
lava_additional_key=lava_additional_key,
|
|
lava_hook_url=lava_hook_url,
|
|
lava_success_url=lava_success_url,
|
|
lava_fail_url=lava_fail_url,
|
|
lava_expire_minutes=lava_expire_minutes,
|
|
anypay_api_id=anypay_api_id,
|
|
anypay_api_key=anypay_api_key,
|
|
anypay_project_id=anypay_project_id,
|
|
anypay_currency=anypay_currency,
|
|
anypay_method=anypay_method,
|
|
anypay_success_url=anypay_success_url,
|
|
anypay_fail_url=anypay_fail_url,
|
|
anypay_email=anypay_email,
|
|
anypay_check_ip=anypay_check_ip,
|
|
webapp_url=webapp_url,
|
|
webapp_host=webapp_host,
|
|
webapp_port=webapp_port,
|
|
topup_code_pay_url=topup_code_pay_url,
|
|
web_admin_login=web_admin_login,
|
|
web_admin_password=web_admin_password,
|
|
smtp_host=smtp_host,
|
|
smtp_port=smtp_port,
|
|
smtp_user=smtp_user,
|
|
smtp_password=smtp_password,
|
|
smtp_from=smtp_from,
|
|
smtp_tls=smtp_tls,
|
|
smtp_ssl=smtp_ssl,
|
|
stars_enabled=stars_enabled,
|
|
stars_rub_per_star=stars_rub_per_star,
|
|
telegram_client_id=telegram_client_id,
|
|
telegram_webhook_enabled=telegram_webhook_enabled,
|
|
telegram_webhook_path=telegram_webhook_path,
|
|
telegram_webhook_secret=telegram_webhook_secret,
|
|
telegram_webhook_url=telegram_webhook_url,
|
|
)
|