import asyncio
import hashlib
import hmac
import ipaddress
import json
import logging
import re
import secrets
import socket
import time
from collections import deque
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from aiohttp import ClientSession, ClientTimeout, web
from aiogram import Bot, Dispatcher
from aiogram.types import LabeledPrice
from pydantic import BaseModel, ConfigDict, EmailStr, ValidationError, constr, field_validator
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
from bot.app.web.webapp_auth import (
create_webapp_session_token,
validate_telegram_login_widget_data,
validate_telegram_webapp_init_data,
verify_webapp_session_token,
)
from bot.services.crypto_pay_service import CryptoPayService
from bot.services.email_auth_service import EmailAuthService, normalize_email
from bot.services.email_templates import render_account_merged
from bot.services.freekassa_service import FreeKassaService
from bot.services.platega_service import PlategaService
from bot.services.promo_code_service import PromoCodeService
from bot.services.referral_service import ReferralService
from bot.services.severpay_service import SeverPayService
from bot.services.subscription_service import SubscriptionService
from bot.services.yookassa_service import YooKassaService
from bot.utils.text_sanitizer import sanitize_display_name, sanitize_username
from bot.utils.request_security import request_client_ip
from config.settings import Settings
from db.dal import payment_dal, subscription_dal, user_dal
from db.dal.user_dal import UserMergeConflictError
from db.models import Payment, User
logger = logging.getLogger(__name__)
TEMPLATE_PATH = Path(__file__).resolve().parent / "templates" / "subscription_webapp.html"
ASSET_DIR = TEMPLATE_PATH.parent
TELEGRAM_WEB_APP_SDK_URL = "https://telegram.org/js/telegram-web-app.js"
TELEGRAM_WEB_APP_SDK_PATH = ASSET_DIR / "telegram-web-app.js"
TELEGRAM_WIDGET_SDK_URL = "https://telegram.org/js/telegram-widget.js?23"
TELEGRAM_WIDGET_SDK_PATH = ASSET_DIR / "telegram-widget.js"
WEBAPP_LOGO_PROXY_PATH = "/webapp-logo"
_UNPATCHED_WIDGET_ORIGIN_SNIPPET = """ if (origin == 'https://telegram.org') {\n origin = default_origin;\n } else if (origin == 'https://telegram-js.azureedge.net' || origin == 'https://tg.dev') {\n origin = dev_origin;\n }\n"""
_PATCHED_WIDGET_ORIGIN_SNIPPET = """ if (origin == 'https://telegram.org') {\n origin = default_origin;\n } else if (origin == 'https://telegram-js.azureedge.net' || origin == 'https://tg.dev') {\n origin = dev_origin;\n } else {\n origin = default_origin;\n }\n"""
WEBAPP_CONFIG_PLACEHOLDER = ""
WEBAPP_I18N_PLACEHOLDER = ""
WEBAPP_JS_PLACEHOLDER = ""
DEV_MOCK_START_MARKER = ""
DEV_MOCK_END_MARKER = ""
WEBAPP_RATE_LIMIT_WINDOW_SECONDS = 60
WEBAPP_RATE_LIMIT_MAX_REQUESTS = 30
WEBAPP_LOGO_MAX_BYTES = 2 * 1024 * 1024
WEBAPP_SESSION_COOKIE_NAME = "rw_webapp_session"
WEBAPP_CSRF_COOKIE_NAME = "rw_webapp_csrf"
WEBAPP_CSRF_HEADER_NAME = "X-CSRF-Token"
WEBAPP_STATE_CHANGING_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
WEBAPP_CSRF_EXEMPT_PATHS = {
"/api/auth/token",
"/api/auth/email/request",
"/api/auth/email/verify",
"/api/auth/logout",
}
class WebAppEmailPayload(BaseModel):
model_config = ConfigDict(extra="ignore")
email: EmailStr
@field_validator("email")
@classmethod
def _normalize_and_limit_email(cls, value: EmailStr) -> str:
normalized = normalize_email(str(value))
if len(normalized) > 254:
raise ValueError("email_too_long")
return normalized
class WebAppEmailCodePayload(WebAppEmailPayload):
code: str = ""
class WebAppPaymentCreatePayload(BaseModel):
model_config = ConfigDict(extra="ignore")
method: str = ""
months: Any = None
description: Optional[constr(max_length=4096)] = None
comment: Optional[constr(max_length=4096)] = None
note: Optional[constr(max_length=4096)] = None
_SHARED_HTTP_SESSION: Optional[ClientSession] = None
_SHARED_HTTP_SESSION_LOCK = asyncio.Lock()
def create_subscription_webapp_application(
dp: Dispatcher,
bot: Bot,
settings: Settings,
async_session_factory: sessionmaker,
) -> web.Application:
app = web.Application(middlewares=[_security_headers_middleware, _csrf_protection_middleware])
app["bot"] = bot
app["dp"] = dp
app["settings"] = settings
app["async_session_factory"] = async_session_factory
app["i18n"] = dp.get("i18n_instance")
app["email_auth_service"] = EmailAuthService(settings)
app["webapp_logo_cache"] = None
app["webapp_logo_cache_lock"] = asyncio.Lock()
app["webapp_settings_cache"] = {"ts": 0.0, "data": {}}
app["webapp_rate_limit_buckets"] = {}
app["webapp_rate_limit_lock"] = asyncio.Lock()
async def _startup(app_obj: web.Application) -> None:
await _ensure_shared_http_session()
async def _shutdown(app_obj: web.Application) -> None:
await _close_shared_http_session()
app.on_startup.append(_startup)
app.on_shutdown.append(_shutdown)
for key in (
"subscription_service",
"yookassa_service",
"freekassa_service",
"cryptopay_service",
"platega_service",
"severpay_service",
"promo_code_service",
"referral_service",
):
if hasattr(dp, "workflow_data") and key in dp.workflow_data: # type: ignore[attr-defined]
app[key] = dp.workflow_data[key] # type: ignore[index]
if hasattr(dp, "workflow_data") and "bot_username" in dp.workflow_data: # type: ignore[attr-defined]
app["bot_username"] = dp.workflow_data["bot_username"] # type: ignore[index]
setup_subscription_webapp_routes(app)
return app
def setup_subscription_webapp_routes(app: web.Application) -> None:
app.router.add_get("/", index_route)
app.router.add_get("/health", health_route)
app.router.add_get("/telegram-web-app.js", telegram_web_app_asset_route)
app.router.add_get("/telegram-widget.js", telegram_widget_asset_route)
app.router.add_get(WEBAPP_LOGO_PROXY_PATH, webapp_logo_route)
app.router.add_get("/subscription_webapp.css", css_asset_route)
app.router.add_get("/subscription_webapp.min.{asset_hash}.js", js_asset_route)
app.router.add_get("/subscription_webapp.js", js_asset_route)
app.router.add_post("/api/auth/token", auth_token_route)
app.router.add_post("/api/auth/email/request", email_auth_request_route)
app.router.add_post("/api/auth/email/verify", email_auth_verify_route)
app.router.add_post("/api/auth/logout", logout_route)
app.router.add_get("/api/me", me_route)
app.router.add_post("/api/account/email/request", account_email_request_route)
app.router.add_post("/api/account/email/verify", account_email_verify_route)
app.router.add_post("/api/account/telegram/link", account_telegram_link_route)
app.router.add_post("/api/promo/apply", apply_promo_route)
app.router.add_post("/api/payments", create_payment_route)
app.router.add_get("/api/payments/{payment_id}", payment_status_route)
async def health_route(request: web.Request) -> web.Response:
return web.json_response({"ok": True})
async def css_asset_route(request: web.Request) -> web.Response:
return await _serve_template_asset(request, "subscription_webapp.css", "text/css")
def _resolve_webapp_logo_url(settings: Settings) -> str:
raw_logo_url = (settings.WEBAPP_LOGO_URL or "").strip()
if not raw_logo_url:
return ""
parsed_logo_url = urlsplit(raw_logo_url)
if parsed_logo_url.scheme == "https" and parsed_logo_url.hostname:
return WEBAPP_LOGO_PROXY_PATH
return ""
async def webapp_logo_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
raw_logo_url = (settings.WEBAPP_LOGO_URL or "").strip()
if not raw_logo_url:
raise web.HTTPNotFound(text="webapp_logo_not_configured")
parsed_logo_url = urlsplit(raw_logo_url)
if parsed_logo_url.scheme != "https" or not parsed_logo_url.hostname:
raise web.HTTPNotFound(text="webapp_logo_not_proxied")
if not await _hostname_resolves_to_public_address(parsed_logo_url.hostname):
raise web.HTTPNotFound(text="webapp_logo_not_proxied")
source_logo_url = raw_logo_url
logo_cache: Optional[Tuple[bytes, str]] = request.app.get("webapp_logo_cache")
if logo_cache is None:
cache_lock: asyncio.Lock = request.app["webapp_logo_cache_lock"]
async with cache_lock:
logo_cache = request.app.get("webapp_logo_cache")
if logo_cache is None:
logo_cache = await _fetch_webapp_logo(source_logo_url)
request.app["webapp_logo_cache"] = logo_cache
if not logo_cache:
raise web.HTTPNotFound(text="webapp_logo_unavailable")
body, content_type = logo_cache
response = web.Response(body=body, content_type=content_type)
response.headers["Cache-Control"] = "no-cache"
return response
async def _fetch_webapp_logo(logo_url: str) -> Optional[Tuple[bytes, str]]:
"""Fetch and cache the configured logo on the server side."""
try:
session = await _get_shared_http_session()
timeout = ClientTimeout(total=5)
async with session.get(logo_url, allow_redirects=False, timeout=timeout) as response:
if response.status != 200:
logger.warning(
"WEBAPP_LOGO_URL returned HTTP %s; keeping the logo hidden.",
response.status,
)
return None
content_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
if content_type and not content_type.startswith("image/"):
logger.warning(
"WEBAPP_LOGO_URL returned non-image content type %s; keeping the logo hidden.",
content_type,
)
return None
body = bytearray()
async for chunk in response.content.iter_chunked(64 * 1024):
body.extend(chunk)
if len(body) > WEBAPP_LOGO_MAX_BYTES:
logger.warning("WEBAPP_LOGO_URL exceeded the 2 MiB limit.")
return None
if not body:
logger.warning("WEBAPP_LOGO_URL returned an empty response body.")
return None
return bytes(body), content_type or "image/png"
except Exception as exc:
logger.warning("Failed to fetch WEBAPP_LOGO_URL: %s", exc)
return None
async def _get_shared_http_session() -> ClientSession:
global _SHARED_HTTP_SESSION
async with _SHARED_HTTP_SESSION_LOCK:
if _SHARED_HTTP_SESSION is None or _SHARED_HTTP_SESSION.closed:
_SHARED_HTTP_SESSION = ClientSession(
timeout=ClientTimeout(total=30),
headers={
"User-Agent": "Mozilla/5.0",
"Accept": "application/javascript,text/javascript,*/*;q=0.8",
},
)
return _SHARED_HTTP_SESSION
async def _ensure_shared_http_session() -> None:
await _get_shared_http_session()
async def _close_shared_http_session() -> None:
global _SHARED_HTTP_SESSION
async with _SHARED_HTTP_SESSION_LOCK:
if _SHARED_HTTP_SESSION and not _SHARED_HTTP_SESSION.closed:
await _SHARED_HTTP_SESSION.close()
_SHARED_HTTP_SESSION = None
async def _hostname_resolves_to_public_address(hostname: str) -> bool:
if not hostname:
return False
try:
ip_obj = ipaddress.ip_address(hostname)
return not (
ip_obj.is_private
or ip_obj.is_loopback
or ip_obj.is_link_local
or ip_obj.is_unspecified
or ip_obj.is_reserved
)
except ValueError:
pass
loop = asyncio.get_running_loop()
try:
resolved = await loop.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
except Exception:
return False
found_public_ip = False
for entry in resolved:
sockaddr = entry[4]
if not sockaddr:
continue
candidate = sockaddr[0]
try:
ip_obj = ipaddress.ip_address(candidate)
except ValueError:
continue
if (
ip_obj.is_private
or ip_obj.is_loopback
or ip_obj.is_link_local
or ip_obj.is_unspecified
or ip_obj.is_reserved
):
return False
found_public_ip = True
return found_public_ip
@web.middleware
async def _security_headers_middleware(request: web.Request, handler):
request["csp_nonce"] = secrets.token_urlsafe(16)
try:
response = await handler(request)
except web.HTTPException as exc:
response = exc
nonce = request.get("csp_nonce", "")
response.headers.setdefault(
"Content-Security-Policy",
(
"default-src 'self'; "
f"script-src 'self' 'nonce-{nonce}' 'unsafe-eval' https://telegram.org; "
"frame-src https://oauth.telegram.org; "
"frame-ancestors https://web.telegram.org https://t.me; "
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
"font-src 'self' https://fonts.gstatic.com https://cdn.jsdelivr.net data:; "
"img-src 'self' data: https:; "
"connect-src 'self'; "
"object-src 'none'; "
"base-uri 'self'; "
"form-action 'self'"
),
)
response.headers.setdefault("Referrer-Policy", "no-referrer")
response.headers.setdefault("X-Content-Type-Options", "nosniff")
response.headers.setdefault(
"Permissions-Policy",
(
"accelerometer=(), autoplay=(), camera=(), display-capture=(), "
"encrypted-media=(), geolocation=(), gyroscope=(), magnetometer=(), "
"microphone=(), midi=(), payment=(), usb=()"
),
)
return response
@web.middleware
async def _csrf_protection_middleware(request: web.Request, handler):
settings: Settings = request.app["settings"]
header = request.headers.get("Authorization", "")
prefix = "Bearer "
if header.startswith(prefix):
if verify_webapp_session_token(settings, header[len(prefix):].strip()):
return await handler(request)
if (
request.method in WEBAPP_STATE_CHANGING_METHODS
and request.path not in WEBAPP_CSRF_EXEMPT_PATHS
and request.cookies.get(WEBAPP_SESSION_COOKIE_NAME)
):
csrf_cookie = request.cookies.get(WEBAPP_CSRF_COOKIE_NAME, "")
csrf_header = request.headers.get(WEBAPP_CSRF_HEADER_NAME, "")
if (
not csrf_cookie
or not csrf_header
or not hmac.compare_digest(csrf_header, csrf_cookie)
):
return _json_error(403, "csrf_failed", "Invalid CSRF token")
return await handler(request)
def _get_cached_webapp_settings(request: web.Request) -> Dict[str, Any]:
settings: Settings = request.app["settings"]
cache = request.app["webapp_settings_cache"]
now = time.monotonic()
if now - float(cache.get("ts", 0.0)) >= 60 or not cache.get("data"):
cache["data"] = {
"logo_url": _resolve_webapp_logo_url(settings),
"subscription_options": settings.subscription_options,
"stars_subscription_options": settings.stars_subscription_options,
"support_url": settings.SUPPORT_LINK or "",
"terms_url": settings.TERMS_OF_SERVICE_URL or "",
"privacy_policy_url": settings.PRIVACY_POLICY_URL or "",
"user_agreement_url": settings.USER_AGREEMENT_URL or "",
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"email_auth_enabled": settings.email_auth_configured,
"language": _normalize_language(settings.DEFAULT_LANGUAGE),
}
cache["ts"] = now
return cache["data"]
async def _enforce_webapp_rate_limit(
request: web.Request,
*,
user_id: int,
action: str,
) -> Optional[web.Response]:
settings: Settings = request.app["settings"]
ip_address = request_client_ip(request, trusted_proxies=settings.trusted_proxies) or request.remote or "unknown"
key = f"{action}:{ip_address}:{int(user_id)}"
buckets: Dict[str, deque[float]] = request.app["webapp_rate_limit_buckets"]
lock: asyncio.Lock = request.app["webapp_rate_limit_lock"]
now = time.monotonic()
async with lock:
bucket = buckets.setdefault(key, deque())
while bucket and now - bucket[0] >= WEBAPP_RATE_LIMIT_WINDOW_SECONDS:
bucket.popleft()
if not bucket:
buckets.pop(key, None)
bucket = buckets.setdefault(key, deque())
if len(bucket) >= WEBAPP_RATE_LIMIT_MAX_REQUESTS:
retry_after = max(
1,
int(WEBAPP_RATE_LIMIT_WINDOW_SECONDS - (now - bucket[0])),
) if bucket else WEBAPP_RATE_LIMIT_WINDOW_SECONDS
return web.json_response(
{
"ok": False,
"error": "rate_limited",
"retry_after": retry_after,
},
status=429,
headers={"Retry-After": str(retry_after)},
)
bucket.append(now)
return None
async def telegram_web_app_asset_route(request: web.Request) -> web.Response:
if not TELEGRAM_WEB_APP_SDK_PATH.exists():
await refresh_telegram_web_app_sdk()
try:
response = await _serve_template_asset(
request,
"telegram-web-app.js",
"application/javascript",
)
except FileNotFoundError:
logger.exception(
"Telegram Web App SDK is unavailable at %s",
TELEGRAM_WEB_APP_SDK_PATH,
)
raise web.HTTPServiceUnavailable(text="telegram_web_app_sdk_unavailable")
response.headers["Cache-Control"] = "no-cache"
return response
async def telegram_widget_asset_route(request: web.Request) -> web.Response:
if not TELEGRAM_WIDGET_SDK_PATH.exists():
await refresh_telegram_login_widget_sdk()
try:
data = TELEGRAM_WIDGET_SDK_PATH.read_bytes()
except FileNotFoundError:
logger.exception(
"Telegram Login Widget SDK is unavailable at %s",
TELEGRAM_WIDGET_SDK_PATH,
)
raise web.HTTPServiceUnavailable(text="telegram_widget_sdk_unavailable")
data = _normalize_telegram_login_widget_sdk(data)
response = web.Response(body=data, content_type="application/javascript")
response.headers["Cache-Control"] = "no-cache"
return response
async def refresh_telegram_web_app_sdk() -> bool:
"""Best-effort refresh of the vendored Telegram Web App SDK."""
try:
session = await _get_shared_http_session()
async with session.get(TELEGRAM_WEB_APP_SDK_URL) as response:
if response.status != 200:
logger.warning(
"Telegram Web App SDK refresh returned HTTP %s; keeping the bundled copy.",
response.status,
)
return False
data = await response.read()
except Exception as exc:
logger.warning("Failed to refresh Telegram Web App SDK: %s", exc)
return False
try:
TELEGRAM_WEB_APP_SDK_PATH.parent.mkdir(parents=True, exist_ok=True)
existing_data = (
TELEGRAM_WEB_APP_SDK_PATH.read_bytes()
if TELEGRAM_WEB_APP_SDK_PATH.exists()
else None
)
if existing_data == data:
logger.info("Telegram Web App SDK is already up to date.")
return True
temp_path = TELEGRAM_WEB_APP_SDK_PATH.with_name(
f"{TELEGRAM_WEB_APP_SDK_PATH.name}.tmp"
)
temp_path.write_bytes(data)
temp_path.replace(TELEGRAM_WEB_APP_SDK_PATH)
logger.info(
"Telegram Web App SDK updated at %s (%d bytes).",
TELEGRAM_WEB_APP_SDK_PATH,
len(data),
)
return True
except Exception as exc:
logger.warning("Failed to store Telegram Web App SDK locally: %s", exc)
return False
async def refresh_telegram_login_widget_sdk() -> bool:
"""Best-effort refresh of the vendored Telegram Login Widget SDK."""
try:
session = await _get_shared_http_session()
async with session.get(TELEGRAM_WIDGET_SDK_URL) as response:
if response.status != 200:
logger.warning(
"Telegram Login Widget SDK refresh returned HTTP %s; keeping the bundled copy.",
response.status,
)
return False
data = await response.read()
except Exception as exc:
logger.warning("Failed to refresh Telegram Login Widget SDK: %s", exc)
return False
try:
data = _normalize_telegram_login_widget_sdk(data)
TELEGRAM_WIDGET_SDK_PATH.parent.mkdir(parents=True, exist_ok=True)
existing_data = (
TELEGRAM_WIDGET_SDK_PATH.read_bytes()
if TELEGRAM_WIDGET_SDK_PATH.exists()
else None
)
if existing_data == data:
logger.info("Telegram Login Widget SDK is already up to date.")
return True
temp_path = TELEGRAM_WIDGET_SDK_PATH.with_name(
f"{TELEGRAM_WIDGET_SDK_PATH.name}.tmp"
)
temp_path.write_bytes(data)
temp_path.replace(TELEGRAM_WIDGET_SDK_PATH)
logger.info(
"Telegram Login Widget SDK updated at %s (%d bytes).",
TELEGRAM_WIDGET_SDK_PATH,
len(data),
)
return True
except Exception as exc:
logger.warning("Failed to store Telegram Login Widget SDK locally: %s", exc)
return False
def _normalize_telegram_login_widget_sdk(data: bytes) -> bytes:
# Keep the vendored widget pointing to Telegram's OAuth host instead of the local origin.
text = data.decode("utf-8")
normalized = text.replace(
_UNPATCHED_WIDGET_ORIGIN_SNIPPET,
_PATCHED_WIDGET_ORIGIN_SNIPPET,
1,
)
return normalized.encode("utf-8")
async def js_asset_route(request: web.Request) -> web.Response:
asset_hash = request.match_info.get("asset_hash")
filename = (
f"subscription_webapp.min.{asset_hash}.js"
if asset_hash
else "subscription_webapp.js"
)
response = await _serve_template_asset(
request,
filename,
"application/javascript",
strip_dev_mock=not asset_hash,
)
response.headers["Cache-Control"] = (
"public, max-age=31536000, immutable" if asset_hash else "no-cache"
)
return response
async def index_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
if not settings.WEBAPP_ENABLED:
raise web.HTTPNotFound(text="webapp_disabled")
html = TEMPLATE_PATH.read_text(encoding="utf-8")
cached = _get_cached_webapp_settings(request)
config = {
"title": settings.WEBAPP_TITLE,
"primaryColor": settings.WEBAPP_PRIMARY_COLOR,
"logoUrl": cached["logo_url"],
"apiBase": "/api",
"telegramLoginBotUsername": request.app.get("bot_username") or "",
"supportUrl": cached["support_url"],
"termsUrl": cached["terms_url"],
"privacyPolicyUrl": cached["privacy_policy_url"],
"userAgreementUrl": cached["user_agreement_url"],
"currency": cached["currency"],
"language": cached["language"],
"emailAuthEnabled": cached["email_auth_enabled"],
}
html = _strip_marked_block(html, DEV_MOCK_START_MARKER, DEV_MOCK_END_MARKER)
i18n_instance: Optional[object] = request.app.get("i18n")
i18n_payload = getattr(i18n_instance, "locales_data", {}) if i18n_instance else {}
nonce = request.get("csp_nonce", "")
html = html.replace(
WEBAPP_CONFIG_PLACEHOLDER,
(
f""
),
)
html = html.replace(
WEBAPP_I18N_PLACEHOLDER,
(
f""
),
)
html = html.replace(
WEBAPP_JS_PLACEHOLDER,
f'',
)
return web.Response(text=html, content_type="text/html", charset="utf-8")
async def _serve_template_asset(
request: web.Request,
filename: str,
content_type: str,
*,
strip_dev_mock: bool = False,
) -> web.Response:
settings: Settings = request.app["settings"]
if not settings.WEBAPP_ENABLED:
raise web.HTTPNotFound(text="webapp_disabled")
path = ASSET_DIR / filename
text = path.read_text(encoding="utf-8")
if strip_dev_mock:
text = _strip_marked_block(
text,
"/* WEBAPP_DEV_MOCK_START */",
"/* WEBAPP_DEV_MOCK_END */",
)
return web.Response(text=text, content_type=content_type, charset="utf-8")
def _resolve_webapp_js_asset_name() -> str:
minified_assets = []
for path in ASSET_DIR.glob("subscription_webapp.min.*.js"):
try:
minified_assets.append((path.stat().st_mtime, path.name))
except OSError:
continue
if minified_assets:
minified_assets.sort(reverse=True)
return minified_assets[0][1]
return "subscription_webapp.js"
def _strip_marked_block(html: str, start_marker: str, end_marker: str) -> str:
start = html.find(start_marker)
if start == -1:
return html
end = html.find(end_marker, start)
if end == -1:
return html[:start]
return html[:start] + html[end + len(end_marker):]
async def auth_token_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
payload = await _read_json(request)
init_data = str(payload.get("init_data") or "")
auth_data = payload.get("auth_data")
referral_param = str(payload.get("referral_code") or payload.get("start_param") or "")
telegram_user = None
if init_data:
telegram_user = validate_telegram_webapp_init_data(
init_data,
settings.BOT_TOKEN,
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
elif auth_data is not None:
telegram_user = validate_telegram_login_widget_data(
auth_data,
settings.BOT_TOKEN,
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
if not telegram_user:
return _json_error(401, "invalid_auth", "Invalid Telegram auth data")
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=int(telegram_user.get("id") or 0),
action="auth_token",
)
if rate_limit_response:
return rate_limit_response
async_session_factory: sessionmaker = request.app["async_session_factory"]
authenticated_user_id: Optional[int] = None
async with async_session_factory() as session:
try:
db_user = await _ensure_user_from_telegram(
session,
telegram_user,
settings,
referral_param=referral_param,
)
if db_user.is_banned:
await session.rollback()
return _json_error(403, "banned", "Access denied")
referral_applied = await _apply_referral_to_existing_user(
request,
session,
db_user,
referral_param or telegram_user.get("start_param"),
)
if getattr(db_user, "_webapp_created", False) or referral_applied:
await _apply_referral_welcome_bonus_if_needed(
request,
session,
db_user,
referral_param or telegram_user.get("start_param"),
)
authenticated_user_id = int(db_user.user_id)
await session.commit()
except Exception as exc:
await session.rollback()
logger.exception("WebApp auth failed")
return _json_error(500, "auth_failed", "Auth failed")
token = create_webapp_session_token(settings, int(authenticated_user_id))
return _build_webapp_auth_response(settings, {"ok": True}, token=token)
async def logout_route(request: web.Request) -> web.Response:
response = web.json_response({"ok": True})
_clear_webapp_auth_cookies(response)
return response
async def email_auth_request_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailPayload, payload)
if validation_error:
return validation_error
email = email_payload.email
lang = _normalize_language(str(payload.get("language") or settings.DEFAULT_LANGUAGE))
return await _request_email_code(
request,
email=email,
purpose="login",
language_code=lang,
target_user_id=None,
)
async def email_auth_verify_route(request: web.Request) -> web.Response:
settings: Settings = request.app["settings"]
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailCodePayload, payload)
if validation_error:
return validation_error
email = email_payload.email
code = str(email_payload.code or "")
referral_param = str(payload.get("referral_code") or payload.get("start_param") or "")
email_service: EmailAuthService = request.app["email_auth_service"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
created_user = False
new_user_referrer_id: Optional[int] = None
async with async_session_factory() as session:
try:
verify_result = await email_service.verify_code(
session,
email=email,
purpose="login",
code=code,
target_user_id=None,
)
if not verify_result.ok:
await session.commit()
status = 429 if verify_result.error == "rate_limited" else 400
return web.json_response(
{
"ok": False,
"error": verify_result.error or "invalid_code",
"retry_after": verify_result.retry_after,
"message": "Invalid code",
},
status=status,
)
db_user = await user_dal.get_user_by_email(session, email)
if not db_user:
referred_by_id = await _resolve_referrer_id(
session,
referral_param,
current_user_id=None,
)
db_user, _ = await user_dal.create_email_user(
session,
email=email,
language_code=_normalize_language(settings.DEFAULT_LANGUAGE),
email_verified_at=datetime.now(timezone.utc),
referred_by_id=referred_by_id,
)
created_user = True
new_user_referrer_id = referred_by_id
elif not db_user.email_verified_at:
db_user.email_verified_at = datetime.now(timezone.utc)
referral_applied = await _apply_referral_to_existing_user(
request,
session,
db_user,
referral_param,
)
if created_user or referral_applied:
await _apply_referral_welcome_bonus_if_needed(
request,
session,
db_user,
referral_param,
)
if db_user.is_banned:
await session.rollback()
return _json_error(403, "banned", "Access denied")
await session.commit()
except Exception as exc:
await session.rollback()
logger.exception("Email WebApp auth failed")
return _json_error(500, "auth_failed", "Auth failed")
if created_user:
try:
from bot.services.notification_service import NotificationService
bot: Bot = request.app["bot"]
notification_service = NotificationService(
bot,
settings,
request.app.get("i18n"),
)
await notification_service.notify_new_email_user_registration(
user_id=int(db_user.user_id),
email=email,
referred_by_id=new_user_referrer_id,
)
except Exception:
logger.exception("Failed to send new email user notification")
token = create_webapp_session_token(settings, int(db_user.user_id))
return _build_webapp_auth_response(
settings,
{
"ok": True,
"user_id": int(db_user.user_id),
"telegram_id": _telegram_id_for_user(db_user),
},
token=token,
)
async def account_email_request_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailPayload, payload)
if validation_error:
return validation_error
email = email_payload.email
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
if db_user.email == email and db_user.email_verified_at:
return web.json_response({"ok": True, "already_linked": True})
lang = _normalize_language(db_user.language_code or settings.DEFAULT_LANGUAGE)
return await _request_email_code(
request,
email=email,
purpose="link_email",
language_code=lang,
target_user_id=user_id,
)
async def account_email_verify_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=user_id,
action="account_email_verify",
)
if rate_limit_response:
return rate_limit_response
payload = await _read_json(request)
email_payload, validation_error = _validate_model_payload(WebAppEmailCodePayload, payload)
if validation_error:
return validation_error
email = email_payload.email
code = str(email_payload.code or "")
email_service: EmailAuthService = request.app["email_auth_service"]
settings: Settings = request.app["settings"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
merge_notice: Optional[Dict[str, Any]] = None
source_panel_uuid: Optional[str] = None
final_user_id = user_id
final_email = email
final_panel_uuid: Optional[str] = None
async with async_session_factory() as session:
try:
verify_result = await email_service.verify_code(
session,
email=email,
purpose="link_email",
code=code,
target_user_id=user_id,
)
if not verify_result.ok:
await session.commit()
status = 429 if verify_result.error == "rate_limited" else 400
return web.json_response(
{
"ok": False,
"error": verify_result.error or "invalid_code",
"retry_after": verify_result.retry_after,
"message": "Invalid code",
},
status=status,
)
current_user = await user_dal.get_user_by_id(session, user_id)
if not current_user or current_user.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
existing_email_user = await user_dal.get_user_by_email(session, email)
if existing_email_user and existing_email_user.user_id != current_user.user_id:
source_panel_uuid = existing_email_user.panel_user_uuid
current_user = await user_dal.merge_users(
session,
source_user_id=existing_email_user.user_id,
target_user_id=current_user.user_id,
)
merge_notice = await _build_account_merge_notice(
session,
merged_user=current_user,
source_user_id=existing_email_user.user_id,
source_panel_uuid=source_panel_uuid,
settings=settings,
)
current_user.email = email
current_user.email_verified_at = datetime.now(timezone.utc)
await _sync_panel_identity_for_user(request, current_user)
await session.commit()
final_user_id = int(current_user.user_id)
final_panel_uuid = current_user.panel_user_uuid
if merge_notice:
merge_end_date_raw = merge_notice.get("final_end_date")
merge_end_date = (
datetime.fromisoformat(merge_end_date_raw)
if merge_end_date_raw
else None
)
await _sync_panel_identity_for_user(
request,
current_user,
expire_at=merge_end_date,
)
# Best-effort cleanup of the removed panel account after the DB merge.
if source_panel_uuid and final_panel_uuid and source_panel_uuid != final_panel_uuid:
subscription_service: SubscriptionService = request.app.get("subscription_service")
if subscription_service and subscription_service.panel_service:
try:
await subscription_service.panel_service.delete_user_from_panel(
source_panel_uuid,
log_response=False,
)
except Exception as exc:
logger.warning(
"Failed to delete merged source panel user %s: %s",
source_panel_uuid,
exc,
)
email_service: EmailAuthService = request.app.get("email_auth_service")
if email_service and final_email:
email_content = render_account_merged(
settings,
language_code=merge_notice.get("language") or settings.DEFAULT_LANGUAGE,
primary_user_id=merge_notice.get("primary_user_id"),
removed_user_id=merge_notice.get("removed_user_id"),
final_end_date_text=str(
merge_notice.get("final_end_date_text")
or merge_notice.get("final_end_date")
or ""
),
)
try:
await email_service.send_rendered_email(
email=final_email,
content=email_content,
)
except Exception as exc:
logger.warning(
"Failed to send account merge email to %s: %s",
final_email,
exc,
)
except UserMergeConflictError as exc:
await session.rollback()
return _json_error(409, "account_merge_conflict", str(exc))
except Exception as exc:
await session.rollback()
logger.exception("Email account link failed")
return _json_error(500, "link_failed", "Link failed")
token = create_webapp_session_token(settings, int(final_user_id))
response_payload: Dict[str, Any] = {"ok": True}
if merge_notice:
response_payload["account_merge"] = merge_notice
response_payload["user_id"] = final_user_id
return _build_webapp_auth_response(settings, response_payload, token=token)
async def account_telegram_link_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
settings: Settings = request.app["settings"]
payload = await _read_json(request)
init_data = str(payload.get("init_data") or "")
auth_data = payload.get("auth_data")
telegram_user = None
if init_data:
telegram_user = validate_telegram_webapp_init_data(
init_data,
settings.BOT_TOKEN,
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
elif auth_data is not None:
telegram_user = validate_telegram_login_widget_data(
auth_data,
settings.BOT_TOKEN,
max_age_seconds=settings.WEBAPP_AUTH_MAX_AGE_SECONDS,
)
if not telegram_user:
return _json_error(401, "invalid_auth", "Invalid Telegram auth data")
async_session_factory: sessionmaker = request.app["async_session_factory"]
merge_notice: Optional[Dict[str, Any]] = None
source_panel_uuid: Optional[str] = None
final_user_id = user_id
final_telegram_id: Optional[int] = None
final_email: Optional[str] = None
final_panel_uuid: Optional[str] = None
async with async_session_factory() as session:
try:
current_user_before_link = await user_dal.get_user_by_id(session, user_id)
if not current_user_before_link or current_user_before_link.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
source_panel_uuid = current_user_before_link.panel_user_uuid
db_user = await _link_telegram_to_user(
request,
session,
current_user_id=user_id,
telegram_user=telegram_user,
settings=settings,
)
if db_user.is_banned:
await session.rollback()
return _json_error(403, "banned", "Access denied")
final_user_id = int(db_user.user_id)
final_telegram_id = _telegram_id_for_user(db_user)
final_email = db_user.email
final_panel_uuid = db_user.panel_user_uuid
if final_user_id != user_id:
merge_notice = await _build_account_merge_notice(
session,
merged_user=db_user,
source_user_id=user_id,
source_panel_uuid=source_panel_uuid,
settings=settings,
)
await session.commit()
if merge_notice:
merge_end_date_raw = merge_notice.get("final_end_date")
merge_end_date = (
datetime.fromisoformat(merge_end_date_raw)
if merge_end_date_raw
else None
)
await _sync_panel_identity_for_user(
request,
db_user,
expire_at=merge_end_date,
)
# Best-effort cleanup of the removed panel account after the DB merge.
if source_panel_uuid and final_panel_uuid and source_panel_uuid != final_panel_uuid:
subscription_service: SubscriptionService = request.app.get("subscription_service")
if subscription_service and subscription_service.panel_service:
try:
await subscription_service.panel_service.delete_user_from_panel(
source_panel_uuid,
log_response=False,
)
except Exception as exc:
logger.warning(
"Failed to delete merged source panel user %s: %s",
source_panel_uuid,
exc,
)
email_service: EmailAuthService = request.app.get("email_auth_service")
if email_service and final_email:
email_content = render_account_merged(
settings,
language_code=merge_notice.get("language") or settings.DEFAULT_LANGUAGE,
primary_user_id=merge_notice.get("primary_user_id"),
removed_user_id=merge_notice.get("removed_user_id"),
final_end_date_text=str(
merge_notice.get("final_end_date_text")
or merge_notice.get("final_end_date")
or ""
),
)
try:
await email_service.send_rendered_email(
email=final_email,
content=email_content,
)
except Exception as exc:
logger.warning(
"Failed to send account merge email to %s: %s",
final_email,
exc,
)
except UserMergeConflictError as exc:
await session.rollback()
return _json_error(409, "account_merge_conflict", str(exc))
except Exception as exc:
await session.rollback()
logger.exception("Telegram account link failed")
return _json_error(500, "link_failed", "Link failed")
token = create_webapp_session_token(settings, int(final_user_id))
response_payload: Dict[str, Any] = {
"ok": True,
"user_id": int(final_user_id),
"telegram_id": final_telegram_id,
}
if merge_notice:
response_payload["account_merge"] = merge_notice
return _build_webapp_auth_response(settings, response_payload, token=token)
async def me_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
data = await _build_user_payload(request, user_id)
return web.json_response({"ok": True, **data})
async def apply_promo_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
payload = await _read_json(request)
code = str(payload.get("code") or "").strip()
if not code:
return _json_error(400, "empty_code", "Promo code is empty")
settings: Settings = request.app["settings"]
promo_code_service: PromoCodeService = request.app.get("promo_code_service")
if not promo_code_service:
return _json_error(503, "service_unavailable", "Promo service unavailable")
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
try:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
await session.rollback()
return _json_error(403, "access_denied", "Access denied")
lang = _normalize_language(db_user.language_code or settings.DEFAULT_LANGUAGE)
success, result = await promo_code_service.apply_promo_code(
session,
user_id,
code,
lang,
)
if not success:
await session.commit()
return _json_error(400, "promo_apply_failed", str(result))
await session.commit()
end_date = result if isinstance(result, datetime) else None
return web.json_response(
{
"ok": True,
"end_date": end_date.isoformat() if end_date else None,
"end_date_text": end_date.strftime("%d.%m.%Y %H:%M") if end_date else None,
}
)
except Exception as exc:
await session.rollback()
logger.exception("WebApp promo apply failed")
return _json_error(500, "promo_apply_failed", "Promo apply failed")
async def create_payment_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
rate_limit_response = await _enforce_webapp_rate_limit(
request,
user_id=user_id,
action="payments_create",
)
if rate_limit_response:
return rate_limit_response
payload = await _read_json(request)
payment_payload, validation_error = _validate_model_payload(WebAppPaymentCreatePayload, payload)
if validation_error:
return validation_error
method = str(payment_payload.method or "").strip().lower()
try:
months = int(float(payment_payload.months))
except (TypeError, ValueError):
return _json_error(400, "invalid_plan", "Invalid subscription period")
settings: Settings = request.app["settings"]
cached = _get_cached_webapp_settings(request)
price = cached["subscription_options"].get(months)
stars_price = cached["stars_subscription_options"].get(months)
if price is None and method != "stars":
return _json_error(400, "invalid_plan", "Subscription period is not available")
if method == "stars" and (stars_price is None or int(stars_price) <= 0):
return _json_error(400, "invalid_plan", "Stars price is not configured")
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
return _json_error(403, "access_denied", "Access denied")
lang = db_user.language_code or settings.DEFAULT_LANGUAGE
return await _create_subscription_payment(
request=request,
session=session,
user_id=user_id,
method=method,
months=months,
price=float(price or 0),
stars_price=stars_price,
lang=lang,
)
async def payment_status_route(request: web.Request) -> web.Response:
user_id = _require_user_id(request)
try:
payment_id = int(request.match_info["payment_id"])
except (TypeError, ValueError):
return _json_error(400, "invalid_payment", "Invalid payment id")
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
payment = await payment_dal.get_payment_by_db_id(session, payment_id)
if not payment or payment.user_id != user_id:
return _json_error(404, "not_found", "Payment not found")
return web.json_response(
{
"ok": True,
"payment_id": payment.payment_id,
"status": payment.status,
"paid": payment.status == "succeeded",
}
)
async def _read_json(request: web.Request) -> Dict[str, Any]:
try:
data = await request.json()
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _json_error(status: int, code: str, message: str) -> web.Response:
return web.json_response(
{"ok": False, "error": code, "message": message},
status=status,
)
def _validation_error_response(exc: ValidationError) -> web.Response:
for error in exc.errors():
loc = error.get("loc") or ()
field = str(loc[0]) if loc else ""
error_type = str(error.get("type") or "")
message = str(error.get("msg") or "")
message_lower = message.lower()
if field == "email":
if ("too_long" in message_lower or "too long" in message_lower or error_type == "string_too_long"):
return _json_error(400, "email_too_long", "Email is too long")
return _json_error(400, "invalid_email", "Invalid email")
if field in {"description", "comment", "note"} and error_type == "string_too_long":
return _json_error(400, f"{field}_too_long", f"{field.capitalize()} is too long")
if error_type == "string_too_long":
return _json_error(400, "text_too_long", "Text is too long")
return _json_error(400, "invalid_request", "Invalid request")
def _validate_model_payload(
model_cls: type[BaseModel],
payload: Dict[str, Any],
) -> tuple[Optional[BaseModel], Optional[web.Response]]:
try:
return model_cls.model_validate(payload), None
except ValidationError as exc:
return None, _validation_error_response(exc)
def _set_webapp_auth_cookies(
response: web.StreamResponse,
settings: Settings,
session_token: str,
csrf_token: str,
) -> None:
max_age = max(60, int(settings.WEBAPP_SESSION_TTL_SECONDS))
response.set_cookie(
WEBAPP_SESSION_COOKIE_NAME,
session_token,
httponly=True,
secure=True,
samesite="None",
path="/",
max_age=max_age,
)
response.set_cookie(
WEBAPP_CSRF_COOKIE_NAME,
csrf_token,
httponly=False,
secure=True,
samesite="None",
path="/",
max_age=max_age,
)
def _clear_webapp_auth_cookies(response: web.StreamResponse) -> None:
response.set_cookie(
WEBAPP_SESSION_COOKIE_NAME,
"",
httponly=True,
secure=True,
samesite="None",
path="/",
max_age=0,
)
response.set_cookie(
WEBAPP_CSRF_COOKIE_NAME,
"",
httponly=False,
secure=True,
samesite="None",
path="/",
max_age=0,
)
def _build_webapp_auth_response(
settings: Settings,
payload: Dict[str, Any],
*,
token: str,
csrf_token: Optional[str] = None,
) -> web.Response:
response_payload = dict(payload)
response_payload["ok"] = True
response_payload["token"] = token
csrf_value = csrf_token or secrets.token_hex(32)
response_payload["csrf_token"] = csrf_value
response = web.json_response(response_payload)
_set_webapp_auth_cookies(response, settings, token, csrf_value)
return response
def _extract_authenticated_user_id(request: web.Request) -> Optional[int]:
settings: Settings = request.app["settings"]
header = request.headers.get("Authorization", "")
prefix = "Bearer "
if header.startswith(prefix):
user_id = verify_webapp_session_token(settings, header[len(prefix):].strip())
if user_id:
return user_id
cookie_token = request.cookies.get(WEBAPP_SESSION_COOKIE_NAME, "")
if cookie_token:
return verify_webapp_session_token(settings, cookie_token)
return None
def _require_user_id(request: web.Request) -> int:
user_id = _extract_authenticated_user_id(request)
if not user_id:
raise web.HTTPUnauthorized(
text=json.dumps({"ok": False, "error": "unauthorized"}),
content_type="application/json",
)
return user_id
async def _request_email_code(
request: web.Request,
*,
email: str,
purpose: str,
language_code: str,
target_user_id: Optional[int],
) -> web.Response:
email_service: EmailAuthService = request.app["email_auth_service"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
async with async_session_factory() as session:
try:
result = await email_service.request_code(
session,
email=email,
purpose=purpose,
language_code=language_code,
target_user_id=target_user_id,
)
if not result.ok:
await session.rollback()
status = 429 if result.error == "rate_limited" else 400
if result.error == "email_auth_not_configured":
status = 503
return web.json_response(
{
"ok": False,
"error": result.error,
"retry_after": result.retry_after,
},
status=status,
)
await session.commit()
return web.json_response({"ok": True})
except Exception as exc:
await session.rollback()
logger.exception("Failed to send email verification code")
return _json_error(502, "email_send_failed", "Failed to send email")
def _telegram_id_for_user(user: User) -> Optional[int]:
if user.telegram_id:
return int(user.telegram_id)
if user.user_id and int(user.user_id) > 0:
return int(user.user_id)
return None
def _panel_description_for_user(user: User) -> str:
lines = [
user.email or "",
user.username or "",
user.first_name or "",
user.last_name or "",
]
return "\n".join(line for line in lines if line).strip()
async def _sync_panel_identity_for_user(
request: web.Request,
user: User,
*,
expire_at: Optional[datetime] = None,
) -> bool:
if not user.panel_user_uuid:
return False
subscription_service: SubscriptionService = request.app.get("subscription_service")
if not subscription_service or not subscription_service.panel_service:
return False
payload: Dict[str, Any] = {
"description": _panel_description_for_user(user),
}
telegram_id = _telegram_id_for_user(user)
if telegram_id:
payload["telegramId"] = telegram_id
if user.email:
payload["email"] = user.email
if expire_at is not None:
payload["expireAt"] = expire_at.isoformat(timespec="milliseconds").replace("+00:00", "Z")
try:
await subscription_service.panel_service.update_user_details_on_panel(
user.panel_user_uuid,
payload,
log_response=False,
)
return True
except Exception as exc:
logger.warning(
"Failed to sync linked identities to panel for user %s: %s",
user.user_id,
exc,
)
return False
def _format_webapp_datetime(value: Optional[datetime]) -> Optional[str]:
if not value:
return None
normalized = value if value.tzinfo else value.replace(tzinfo=timezone.utc)
return normalized.strftime("%d.%m.%Y %H:%M")
async def _build_account_merge_notice(
session: AsyncSession,
*,
merged_user: User,
source_user_id: int,
source_panel_uuid: Optional[str],
settings: Settings,
) -> Dict[str, Any]:
merged_subscription = None
if merged_user.panel_user_uuid:
merged_subscription = await subscription_dal.get_active_subscription_by_user_id(
session,
merged_user.user_id,
merged_user.panel_user_uuid,
)
if not merged_subscription:
merged_subscription = await subscription_dal.get_active_subscription_by_user_id(
session,
merged_user.user_id,
)
final_end_date = merged_subscription.end_date if merged_subscription else None
if final_end_date and final_end_date.tzinfo is None:
final_end_date = final_end_date.replace(tzinfo=timezone.utc)
return {
"merged": True,
"language": _normalize_language(merged_user.language_code or settings.DEFAULT_LANGUAGE),
"primary_user_id": int(merged_user.user_id),
"removed_user_id": int(source_user_id),
"primary_panel_user_uuid": merged_user.panel_user_uuid,
"removed_panel_user_uuid": source_panel_uuid,
"final_end_date": final_end_date.isoformat() if final_end_date else None,
"final_end_date_text": _format_webapp_datetime(final_end_date),
}
def _telegram_photo_url_value(telegram_user: Dict[str, Any]) -> Optional[str]:
raw_value = telegram_user.get("photo_url")
if not raw_value:
return None
value = str(raw_value).strip()
return value or None
def _apply_telegram_profile_to_user(
user: User,
telegram_user: Dict[str, Any],
settings: Settings,
) -> None:
language_code = telegram_user.get("language_code") or user.language_code or settings.DEFAULT_LANGUAGE
if language_code not in {"ru", "en"}:
language_code = user.language_code or settings.DEFAULT_LANGUAGE
user.telegram_id = int(telegram_user["id"])
user.username = sanitize_username(telegram_user.get("username"))
user.first_name = sanitize_display_name(telegram_user.get("first_name"))
user.last_name = sanitize_display_name(telegram_user.get("last_name"))
user.language_code = language_code
telegram_photo_url = _telegram_photo_url_value(telegram_user)
if telegram_photo_url:
user.telegram_photo_url = telegram_photo_url
async def _link_telegram_to_user(
request: web.Request,
session: AsyncSession,
*,
current_user_id: int,
telegram_user: Dict[str, Any],
settings: Settings,
) -> User:
telegram_id = int(telegram_user["id"])
current_user = await user_dal.get_user_by_id(session, current_user_id)
if not current_user:
raise ValueError("Current user not found.")
existing_telegram_user = await user_dal.get_user_by_telegram_id(session, telegram_id)
if not existing_telegram_user:
existing_telegram_user = await user_dal.get_user_by_id(session, telegram_id)
if existing_telegram_user and existing_telegram_user.user_id != current_user.user_id:
if (
current_user.email
and existing_telegram_user.email
and current_user.email != existing_telegram_user.email
):
raise UserMergeConflictError(
"Telegram account is already linked to a different email."
)
merged_user = await user_dal.merge_users(
session,
source_user_id=current_user.user_id,
target_user_id=existing_telegram_user.user_id,
)
_apply_telegram_profile_to_user(merged_user, telegram_user, settings)
await session.flush()
await _sync_panel_identity_for_user(request, merged_user)
return merged_user
if not existing_telegram_user and int(current_user.user_id) < 0:
language_code = telegram_user.get("language_code") or current_user.language_code or settings.DEFAULT_LANGUAGE
if language_code not in {"ru", "en"}:
language_code = current_user.language_code or settings.DEFAULT_LANGUAGE
target_user, _ = await user_dal.create_user(
session,
{
"user_id": telegram_id,
"telegram_id": telegram_id,
"username": sanitize_username(telegram_user.get("username")),
"first_name": sanitize_display_name(telegram_user.get("first_name")),
"last_name": sanitize_display_name(telegram_user.get("last_name")),
"language_code": language_code,
"registration_date": current_user.registration_date or datetime.now(timezone.utc),
},
)
target_user.referral_code = None
await session.flush()
merged_user = await user_dal.merge_users(
session,
source_user_id=current_user.user_id,
target_user_id=target_user.user_id,
)
_apply_telegram_profile_to_user(merged_user, telegram_user, settings)
await session.flush()
await _sync_panel_identity_for_user(request, merged_user)
return merged_user
if current_user.telegram_id and int(current_user.telegram_id) != telegram_id:
raise UserMergeConflictError("Current account is already linked to Telegram.")
_apply_telegram_profile_to_user(current_user, telegram_user, settings)
await session.flush()
await _sync_panel_identity_for_user(request, current_user)
return current_user
def _normalize_referral_param(raw: Optional[str]) -> Optional[str]:
value = (raw or "").strip()
if not value:
return None
value_lower = value.lower()
if value_lower.startswith("ref_u"):
value = value[5:]
elif value_lower.startswith("ref_"):
value = value[4:]
elif value and value[0].lower() == "u" and len(value) == 10:
value = value[1:]
if not re.fullmatch(r"[A-Za-z0-9]{1,32}", value):
return None
return value.upper()
async def _resolve_referrer_id(
session: AsyncSession,
raw_referral_param: Optional[str],
*,
current_user_id: Optional[int],
) -> Optional[int]:
normalized = _normalize_referral_param(raw_referral_param)
if not normalized:
return None
ref_user = None
if normalized.isdigit():
ref_user = await user_dal.get_user_by_id(session, int(normalized))
if not ref_user:
ref_user = await user_dal.get_user_by_referral_code(session, normalized)
if not ref_user:
return None
if current_user_id is not None and int(ref_user.user_id) == int(current_user_id):
return None
return int(ref_user.user_id)
async def _apply_referral_to_existing_user(
request: web.Request,
session: AsyncSession,
user: User,
raw_referral_param: Optional[str],
) -> bool:
if not raw_referral_param or user.referred_by_id is not None:
return False
referred_by_id = await _resolve_referrer_id(
session,
raw_referral_param,
current_user_id=int(user.user_id),
)
if not referred_by_id:
return False
subscription_service: SubscriptionService = request.app["subscription_service"]
try:
is_active_now = await subscription_service.has_active_subscription(
session,
int(user.user_id),
)
except Exception:
is_active_now = False
if is_active_now:
return False
user.referred_by_id = referred_by_id
await session.flush()
return True
async def _apply_referral_welcome_bonus_if_needed(
request: web.Request,
session: AsyncSession,
user: User,
raw_referral_param: Optional[str],
) -> Optional[datetime]:
if not raw_referral_param or not user.referred_by_id:
return None
settings: Settings = request.app["settings"]
referral_welcome_days = max(
0,
int(getattr(settings, "REFERRAL_WELCOME_BONUS_DAYS", 0) or 0),
)
if referral_welcome_days <= 0:
return None
subscription_service: SubscriptionService = request.app["subscription_service"]
try:
if await subscription_service.has_active_subscription(session, int(user.user_id)):
return None
except Exception:
pass
return await subscription_service.extend_active_subscription_days(
session,
int(user.user_id),
referral_welcome_days,
reason="referral_welcome_bonus",
)
async def _ensure_user_from_telegram(
session: AsyncSession,
telegram_user: Dict[str, Any],
settings: Settings,
*,
referral_param: Optional[str] = None,
) -> User:
user_id = int(telegram_user["id"])
language_code = telegram_user.get("language_code") or settings.DEFAULT_LANGUAGE
if language_code not in {"ru", "en"}:
language_code = settings.DEFAULT_LANGUAGE
update_data = {
"telegram_id": user_id,
"username": sanitize_username(telegram_user.get("username")),
"first_name": sanitize_display_name(telegram_user.get("first_name")),
"last_name": sanitize_display_name(telegram_user.get("last_name")),
"language_code": language_code,
}
telegram_photo_url = _telegram_photo_url_value(telegram_user)
if telegram_photo_url:
update_data["telegram_photo_url"] = telegram_photo_url
db_user = await user_dal.get_user_by_telegram_id(session, user_id)
if not db_user:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user:
referred_by_id = await _resolve_referrer_id(
session,
referral_param or telegram_user.get("start_param"),
current_user_id=user_id,
)
db_user, created = await user_dal.create_user(
session,
{
"user_id": user_id,
**update_data,
"referred_by_id": referred_by_id,
"registration_date": datetime.now(timezone.utc),
},
)
setattr(db_user, "_webapp_created", bool(created))
return db_user
changed = {
key: value
for key, value in update_data.items()
if getattr(db_user, key) != value
}
if changed:
db_user = await user_dal.update_user(session, db_user.user_id, changed) or db_user
return db_user
async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, Any]:
settings: Settings = request.app["settings"]
async_session_factory: sessionmaker = request.app["async_session_factory"]
subscription_service: SubscriptionService = request.app["subscription_service"]
cached = _get_cached_webapp_settings(request)
async with async_session_factory() as session:
db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or db_user.is_banned:
raise web.HTTPForbidden(
text=json.dumps({"ok": False, "error": "access_denied"}),
content_type="application/json",
)
active = await subscription_service.get_active_subscription_details(
session, user_id
)
referral_code = await user_dal.ensure_referral_code(session, db_user)
referral_service: Optional[ReferralService] = request.app.get("referral_service")
bot_username = request.app.get("bot_username") or ""
referral_link = None
if referral_service and bot_username:
referral_link = await referral_service.generate_referral_link(
session,
bot_username,
user_id,
)
webapp_referral_link = _build_webapp_referral_link(
request.app["settings"].SUBSCRIPTION_MINI_APP_URL,
referral_code,
)
referral_stats = (
await referral_service.get_referral_stats(session, user_id)
if referral_service
else {"invited_count": 0, "purchased_count": 0}
)
local_sub = await subscription_dal.get_active_subscription_by_user_id(
session,
user_id,
db_user.panel_user_uuid,
) if db_user.panel_user_uuid else None
try:
await session.commit()
except Exception:
await session.rollback()
lang = _normalize_language(db_user.language_code or settings.DEFAULT_LANGUAGE)
return {
"user": {
"id": user_id,
"username": db_user.username,
"email": db_user.email,
"email_verified": bool(db_user.email_verified_at),
"telegram_id": db_user.telegram_id,
"telegram_linked": bool(_telegram_id_for_user(db_user)),
"telegram_photo_url": db_user.telegram_photo_url,
"first_name": db_user.first_name,
"language_code": lang,
},
"subscription": _serialize_subscription(active, local_sub, lang),
"referral": {
"code": referral_code,
"bot_link": referral_link,
"webapp_link": webapp_referral_link,
"invited_count": referral_stats.get("invited_count", 0),
"purchased_count": referral_stats.get("purchased_count", 0),
"bonus_details": _serialize_referral_bonus_details(settings, lang),
},
"plans": _serialize_plans(
settings,
lang,
subscription_options=cached["subscription_options"],
stars_subscription_options=cached["stars_subscription_options"],
),
"payment_methods": _serialize_payment_methods(settings, request.app),
"settings": {
"support_url": settings.SUPPORT_LINK,
"traffic_mode": bool(settings.traffic_sale_mode),
"email_auth_enabled": settings.email_auth_configured,
},
}
def _serialize_referral_bonus_details(settings: Settings, lang: str) -> List[Dict[str, Any]]:
if getattr(settings, "traffic_sale_mode", False):
return []
details: List[Dict[str, Any]] = []
for months, _price in sorted(settings.subscription_options.items()):
inviter_days = settings.referral_bonus_inviter.get(months)
friend_days = settings.referral_bonus_referee.get(months)
if inviter_days is None and friend_days is None:
continue
details.append(
{
"months": int(months),
"title": _format_months_title(int(months), lang),
"inviter_days": int(inviter_days or 0),
"friend_days": int(friend_days or 0),
}
)
return details
def _build_webapp_referral_link(
base_url: Optional[str],
referral_code: Optional[str],
) -> Optional[str]:
if not base_url or not referral_code:
return None
parts = urlsplit(base_url)
query = dict(parse_qsl(parts.query, keep_blank_values=True))
query["ref"] = f"u{referral_code}"
return urlunsplit(
(
parts.scheme,
parts.netloc,
parts.path or "/",
urlencode(query),
parts.fragment,
)
)
def _serialize_subscription(
active: Optional[Dict[str, Any]],
local_sub: Optional[Any],
lang: str,
) -> Dict[str, Any]:
if not active:
return {
"active": False,
"status": "INACTIVE",
"remaining_text": _format_remaining(0, lang),
"days_left": 0,
"config_link": None,
"connect_url": None,
}
end_date = active.get("end_date")
if end_date and end_date.tzinfo is None:
end_date = end_date.replace(tzinfo=timezone.utc)
seconds_left = 0
if end_date:
seconds_left = max(
0,
int((end_date - datetime.now(timezone.utc)).total_seconds()),
)
return {
"active": seconds_left > 0,
"status": active.get("status_from_panel") or "UNKNOWN",
"end_date": end_date.isoformat() if end_date else None,
"end_date_text": end_date.strftime("%d.%m.%Y %H:%M") if end_date else "N/A",
"days_left": seconds_left // 86400,
"remaining_text": _format_remaining(seconds_left, lang),
"config_link": active.get("config_link"),
"connect_url": active.get("connect_button_url") or active.get("config_link"),
"traffic_limit": _format_bytes(active.get("traffic_limit_bytes")),
"traffic_used": _format_bytes(active.get("traffic_used_bytes")),
"traffic_limit_bytes": _coerce_int_or_none(active.get("traffic_limit_bytes")),
"traffic_used_bytes": _coerce_int_or_none(active.get("traffic_used_bytes")),
"auto_renew_enabled": bool(getattr(local_sub, "auto_renew_enabled", False)),
"provider": getattr(local_sub, "provider", None),
}
def _serialize_plans(
settings: Settings,
lang: str,
*,
subscription_options: Optional[Dict[int, float]] = None,
stars_subscription_options: Optional[Dict[int, int]] = None,
) -> List[Dict[str, Any]]:
active_subscription_options = subscription_options or settings.subscription_options
active_stars_subscription_options = stars_subscription_options or settings.stars_subscription_options
plans: List[Dict[str, Any]] = []
for months, price in sorted(active_subscription_options.items()):
plan = {
"months": int(months),
"price": float(price),
"currency": settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
"title": _format_months_title(int(months), lang),
}
stars_price = active_stars_subscription_options.get(months)
if stars_price is not None and int(stars_price) > 0:
plan["stars_price"] = int(stars_price)
plans.append(plan)
return plans
def _serialize_payment_methods(
settings: Settings,
app: web.Application,
) -> List[Dict[str, Any]]:
labels = {
"severpay": "SeverPay",
"freekassa": "FreeKassa / СБП",
"platega_sbp": "Platega · СБП",
"platega_crypto": "Platega · Crypto",
"yookassa": "Банковская карта",
"stars": "Telegram Stars",
"cryptopay": "CryptoPay",
}
methods: List[Dict[str, Any]] = []
for method in settings.payment_methods_order:
method = method.lower()
if method == "severpay" and _service_configured(app, "severpay_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "freekassa" and _service_configured(app, "freekassa_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "platega_sbp" and settings.PLATEGA_SBP_ENABLED and _service_configured(app, "platega_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "platega_crypto" and settings.PLATEGA_CRYPTO_ENABLED and _service_configured(app, "platega_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "yookassa" and _service_configured(app, "yookassa_service"):
methods.append({"id": method, "name": labels[method]})
elif method == "stars" and settings.STARS_ENABLED:
methods.append({"id": method, "name": labels[method]})
elif method == "cryptopay" and _service_configured(app, "cryptopay_service"):
methods.append({"id": method, "name": labels[method]})
return methods
def _service_configured(app: web.Application, key: str) -> bool:
service = app.get(key)
return bool(service and getattr(service, "configured", False))
async def _create_subscription_payment(
*,
request: web.Request,
session: AsyncSession,
user_id: int,
method: str,
months: int,
price: float,
stars_price: Optional[int],
lang: str,
) -> web.Response:
settings: Settings = request.app["settings"]
description = _payment_description(months, lang)
if method == "yookassa":
return await _create_yookassa_payment(
request, session, user_id, months, price, description
)
if method == "freekassa":
return await _create_freekassa_payment(
request, session, user_id, months, price, description
)
if method in ("platega", "platega_sbp", "platega_crypto"):
return await _create_platega_payment(
request, session, user_id, months, price, description, variant=method
)
if method == "severpay":
return await _create_severpay_payment(
request, session, user_id, months, price, description
)
if method == "cryptopay":
service: CryptoPayService = request.app["cryptopay_service"]
if not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
url = await service.create_invoice(
session=session,
user_id=user_id,
months=months,
amount=price,
description=description,
sale_mode="subscription",
url_kind="web",
)
if not url:
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{"ok": True, "action": "open_link", "payment_url": url, "payment_id": None}
)
if method == "stars":
if not settings.STARS_ENABLED or stars_price is None:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
return await _create_stars_payment(
request, session, user_id, months, int(stars_price), description
)
return _json_error(400, "payment_unavailable", "Payment method unavailable")
async def _create_base_payment_record(
session: AsyncSession,
*,
user_id: int,
amount: float,
currency: str,
status: str,
description: str,
months: int,
provider: str,
) -> Payment:
payment = await payment_dal.create_payment_record(
session,
{
"user_id": user_id,
"amount": amount,
"currency": currency,
"status": status,
"description": description,
"subscription_duration_months": months,
"provider": provider,
},
)
await session.commit()
return payment
async def _create_yookassa_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: int,
price: float,
description: str,
) -> web.Response:
settings: Settings = request.app["settings"]
service: YooKassaService = request.app["yookassa_service"]
if not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
try:
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency="RUB",
status="pending_yookassa",
description=description,
months=months,
provider="yookassa",
)
response = await service.create_payment(
amount=price,
currency="RUB",
description=description,
metadata={
"user_id": str(user_id),
"subscription_months": str(months),
"payment_db_id": str(payment.payment_id),
"sale_mode": "subscription",
"source": "webapp",
},
receipt_email=settings.YOOKASSA_DEFAULT_RECEIPT_EMAIL,
save_payment_method=bool(
settings.yookassa_autopayments_active
and settings.YOOKASSA_AUTOPAYMENTS_REQUIRE_CARD_BINDING
),
)
payment_url = response.get("confirmation_url") if response else None
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
await payment_dal.update_payment_status_by_db_id(
session,
payment.payment_id,
response.get("status", "pending"),
yk_payment_id=response.get("id"),
)
await session.commit()
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("YooKassa WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_freekassa_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: int,
price: float,
description: str,
) -> web.Response:
settings: Settings = request.app["settings"]
service: FreeKassaService = request.app["freekassa_service"]
if not service or not service.configured or not service.payment_method_id:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
try:
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency=service.default_currency,
status="pending_freekassa",
description=description,
months=months,
provider="freekassa",
)
success, response_data = await service.create_order(
payment_db_id=payment.payment_id,
user_id=user_id,
months=months,
amount=price,
currency=service.default_currency,
payment_method_id=service.payment_method_id,
ip_address=service.server_ip,
extra_params={"us_method": service.payment_method_id},
)
payment_url = response_data.get("location") if success else None
provider_id = response_data.get("orderHash") or response_data.get("orderId")
if provider_id:
await payment_dal.update_provider_payment_and_status(
session, payment.payment_id, str(provider_id), payment.status
)
await session.commit()
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("FreeKassa WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_platega_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: int,
price: float,
description: str,
variant: str = "platega_sbp",
) -> web.Response:
settings: Settings = request.app["settings"]
service: PlategaService = request.app["platega_service"]
if not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
if variant == "platega_crypto":
if not settings.PLATEGA_CRYPTO_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
platega_method_id = settings.PLATEGA_CRYPTO_METHOD
else:
if variant == "platega_sbp" and not settings.PLATEGA_SBP_ENABLED:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
platega_method_id = settings.platega_sbp_method_resolved
try:
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
status="pending_platega",
description=description,
months=months,
provider="platega",
)
payload = json.dumps(
{
"payment_db_id": payment.payment_id,
"user_id": user_id,
"months": months,
"sale_mode": "subscription",
"source": "webapp",
"platega_variant": "crypto" if variant == "platega_crypto" else "sbp",
}
)
success, response_data = await service.create_transaction(
payment_db_id=payment.payment_id,
user_id=user_id,
months=months,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
description=description,
payload=payload,
payment_method=platega_method_id,
)
payment_url = (
response_data.get("redirect")
or response_data.get("url")
or response_data.get("paymentUrl")
) if success else None
provider_id = response_data.get("transactionId") or response_data.get("id")
if provider_id:
await payment_dal.update_provider_payment_and_status(
session,
payment.payment_id,
str(provider_id),
str(response_data.get("status", payment.status)),
)
await session.commit()
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("Platega WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_severpay_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: int,
price: float,
description: str,
) -> web.Response:
settings: Settings = request.app["settings"]
service: SeverPayService = request.app["severpay_service"]
if not service or not service.configured:
return _json_error(400, "payment_unavailable", "Payment method unavailable")
try:
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
status="pending_severpay",
description=description,
months=months,
provider="severpay",
)
success, response_data = await service.create_payment(
payment_db_id=payment.payment_id,
user_id=user_id,
months=months,
amount=price,
currency=settings.DEFAULT_CURRENCY_SYMBOL or "RUB",
description=description,
)
payment_url = (
response_data.get("url")
or response_data.get("payment_url")
or response_data.get("paymentUrl")
) if success else None
provider_id = response_data.get("id") or response_data.get("uid")
if provider_id:
await payment_dal.update_provider_payment_and_status(
session, payment.payment_id, str(provider_id), payment.status
)
await session.commit()
if not payment_url:
await payment_dal.update_payment_status_by_db_id(
session, payment.payment_id, "failed_creation"
)
await session.commit()
return _json_error(502, "payment_failed", "Failed to create payment")
return web.json_response(
{
"ok": True,
"action": "open_link",
"payment_url": payment_url,
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("SeverPay WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create payment")
async def _create_stars_payment(
request: web.Request,
session: AsyncSession,
user_id: int,
months: int,
stars_price: int,
description: str,
) -> web.Response:
bot: Bot = request.app["bot"]
try:
payment = await _create_base_payment_record(
session,
user_id=user_id,
amount=float(stars_price),
currency="XTR",
status="pending_stars",
description=description,
months=months,
provider="telegram_stars",
)
payload = f"{payment.payment_id}:{months}:subscription"
prices = [LabeledPrice(label=description, amount=stars_price)]
create_invoice_link = getattr(bot, "create_invoice_link", None)
if callable(create_invoice_link):
invoice_url = await create_invoice_link(
title=description,
description=description,
payload=payload,
provider_token="", # Required to be empty for Telegram Stars (XTR) per Telegram Bot API.
currency="XTR",
prices=prices,
)
return web.json_response(
{
"ok": True,
"action": "open_invoice",
"payment_url": invoice_url,
"payment_id": payment.payment_id,
}
)
await bot.send_invoice(
chat_id=user_id,
title=description,
description=description,
payload=payload,
provider_token="",
currency="XTR",
prices=prices,
)
return web.json_response(
{
"ok": True,
"action": "invoice_sent",
"payment_id": payment.payment_id,
}
)
except Exception as exc:
await session.rollback()
logger.exception("Stars WebApp payment failed")
return _json_error(502, "payment_failed", "Failed to create invoice")
def _normalize_language(lang: Optional[str]) -> str:
value = (lang or "ru").split("-")[0].lower()
return value if value in {"ru", "en"} else "ru"
def _format_remaining(seconds: int, lang: str) -> str:
if seconds <= 0:
if lang == "en":
return "Subscription inactive"
return "Подписка не активна"
days, rem = divmod(seconds, 86400)
hours, rem = divmod(rem, 3600)
minutes = rem // 60
if lang == "en":
if days > 0:
return f"{days} d. {hours} h."
if hours > 0:
return f"{hours} h. {minutes} min."
return f"{max(1, minutes)} min."
if days > 0:
return f"{days} д. {hours} ч."
if hours > 0:
return f"{hours} ч. {minutes} мин."
return f"{max(1, minutes)} мин."
def _coerce_int_or_none(value: Optional[Any]) -> Optional[int]:
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _format_bytes(value: Optional[Any]) -> str:
if value is None:
return "N/A"
try:
size = float(value)
except (TypeError, ValueError):
return str(value)
if size <= 0:
return "∞"
units = ["B", "KB", "MB", "GB", "TB"]
index = 0
while size >= 1024 and index < len(units) - 1:
size /= 1024
index += 1
return f"{size:.2f} {units[index]}"
def _format_months_title(months: int, lang: str) -> str:
if lang == "en":
if months == 1:
return "1 month"
return f"{months} months"
if months == 1:
return "1 месяц"
if 2 <= months <= 4:
return f"{months} месяца"
return f"{months} месяцев"
def _payment_description(months: int, lang: str) -> str:
if lang == "en":
return f"Subscription for {_format_months_title(months, lang)}"
return f"Подписка на {_format_months_title(months, lang)}"