1233 lines
46 KiB
Python
1233 lines
46 KiB
Python
import logging
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from typing import Optional, Union
|
|
|
|
from aiogram import Bot, F, Router, types
|
|
from aiogram.exceptions import TelegramAPIError, TelegramBadRequest, TelegramForbiddenError
|
|
from aiogram.filters import Command, CommandStart
|
|
from aiogram.fsm.context import FSMContext
|
|
from aiogram.utils.text_decorations import html_decoration as hd
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from bot.keyboards.inline.user_keyboards import (
|
|
get_bot_interface_inline_keyboard,
|
|
get_channel_subscription_keyboard,
|
|
get_information_links_keyboard,
|
|
get_language_selection_keyboard,
|
|
get_main_menu_inline_keyboard,
|
|
)
|
|
from bot.middlewares.i18n import JsonI18n, normalize_locale_language_code
|
|
from bot.services.panel_api_service import PanelApiService
|
|
from bot.services.promo_code_service import PromoCodeService
|
|
from bot.services.referral_service import ReferralService
|
|
from bot.services.subscription_service import SubscriptionService
|
|
from bot.services.telegram_notifications import TELEGRAM_NOTIFICATIONS_ENABLED
|
|
from bot.utils.callback_answer import safe_answer_callback
|
|
from bot.utils.channel_subscription import (
|
|
is_required_channel_access_error,
|
|
normalize_required_channel_id,
|
|
resolve_required_channel_link,
|
|
)
|
|
from bot.utils.install_links import (
|
|
append_install_share_link_text,
|
|
ensure_user_install_guide_links,
|
|
)
|
|
from bot.utils.text_sanitizer import sanitize_display_name, sanitize_username
|
|
from config.settings import Settings
|
|
from db.dal import user_dal
|
|
from db.models import User
|
|
|
|
router = Router(name="user_start_router")
|
|
|
|
|
|
def _remnashop_referral_compat_enabled(settings: Settings) -> bool:
|
|
return bool(getattr(settings, "MIGRATION_REMNASHOP_REFERRAL_CODE_COMPAT_ENABLED", False))
|
|
|
|
|
|
def _referral_code_lookup_candidates(
|
|
raw_ref_value: str,
|
|
*,
|
|
remnashop_compat: bool,
|
|
) -> list[str]:
|
|
value = str(raw_ref_value or "").strip()
|
|
if not value:
|
|
return []
|
|
|
|
candidates = [value]
|
|
if value and value[0].lower() == "u":
|
|
stripped_current_prefix = value[1:]
|
|
if remnashop_compat:
|
|
candidates.append(stripped_current_prefix)
|
|
else:
|
|
candidates = [stripped_current_prefix]
|
|
|
|
unique: list[str] = []
|
|
for candidate in candidates:
|
|
candidate = candidate.strip()
|
|
if candidate and candidate not in unique:
|
|
unique.append(candidate)
|
|
return unique
|
|
|
|
|
|
async def _resolve_referrer_from_start_ref(
|
|
session: AsyncSession,
|
|
raw_ref_value: str,
|
|
*,
|
|
settings: Settings,
|
|
current_user_id: int,
|
|
) -> Optional[int]:
|
|
ref_user: Optional[User] = None
|
|
if raw_ref_value.isdigit() and settings.LEGACY_REFS:
|
|
potential_referrer_id = int(raw_ref_value)
|
|
if potential_referrer_id != current_user_id:
|
|
ref_user = await user_dal.get_user_by_id(session, potential_referrer_id)
|
|
|
|
include_legacy = _remnashop_referral_compat_enabled(settings)
|
|
if not ref_user:
|
|
for code in _referral_code_lookup_candidates(
|
|
raw_ref_value,
|
|
remnashop_compat=include_legacy,
|
|
):
|
|
ref_user = await user_dal.get_user_by_referral_code(
|
|
session,
|
|
code,
|
|
include_legacy=include_legacy,
|
|
)
|
|
if ref_user:
|
|
break
|
|
|
|
if ref_user and ref_user.user_id != current_user_id:
|
|
return int(ref_user.user_id)
|
|
return None
|
|
|
|
|
|
async def should_show_trial_button(
|
|
settings: Settings,
|
|
subscription_service: SubscriptionService,
|
|
session: AsyncSession,
|
|
user_id: int,
|
|
) -> bool:
|
|
if not settings.TRIAL_ENABLED:
|
|
return False
|
|
|
|
if hasattr(subscription_service, "has_trial_blocking_subscription") and callable(
|
|
getattr(subscription_service, "has_trial_blocking_subscription")
|
|
):
|
|
return not await subscription_service.has_trial_blocking_subscription(session, user_id)
|
|
|
|
logging.error("Method has_trial_blocking_subscription is missing in SubscriptionService!")
|
|
return False
|
|
|
|
|
|
async def send_main_menu(
|
|
target_event: Union[types.Message, types.CallbackQuery],
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
subscription_service: SubscriptionService,
|
|
session: AsyncSession,
|
|
is_edit: bool = False,
|
|
):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
|
|
user_id = target_event.from_user.id
|
|
user_full_name = hd.quote(target_event.from_user.full_name)
|
|
|
|
if not i18n:
|
|
logging.error(f"i18n_instance missing in send_main_menu for user {user_id}")
|
|
err_msg_fallback = "Error: Language service unavailable. Please try again later."
|
|
if isinstance(target_event, types.CallbackQuery):
|
|
try:
|
|
await target_event.answer(err_msg_fallback, show_alert=True)
|
|
except Exception:
|
|
pass
|
|
elif isinstance(target_event, types.Message):
|
|
try:
|
|
await target_event.answer(err_msg_fallback)
|
|
except Exception:
|
|
pass
|
|
return
|
|
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
|
|
show_trial_button_in_menu = await should_show_trial_button(
|
|
settings, subscription_service, session, user_id
|
|
)
|
|
|
|
text = _(key="main_menu_greeting", user_name=user_full_name)
|
|
reply_markup = get_main_menu_inline_keyboard(
|
|
current_lang, i18n, settings, show_trial_button_in_menu
|
|
)
|
|
|
|
target_message_obj: Optional[types.Message] = None
|
|
if isinstance(target_event, types.Message):
|
|
target_message_obj = target_event
|
|
elif isinstance(target_event, types.CallbackQuery) and target_event.message:
|
|
target_message_obj = target_event.message
|
|
|
|
if not target_message_obj:
|
|
logging.error(f"send_main_menu: target_message_obj is None for event from user {user_id}.")
|
|
if isinstance(target_event, types.CallbackQuery):
|
|
await safe_answer_callback(
|
|
target_event,
|
|
_("error_displaying_menu"),
|
|
show_alert=True,
|
|
)
|
|
return
|
|
|
|
try:
|
|
if is_edit:
|
|
await target_message_obj.edit_text(text, reply_markup=reply_markup)
|
|
else:
|
|
await target_message_obj.answer(text, reply_markup=reply_markup)
|
|
|
|
if isinstance(target_event, types.CallbackQuery):
|
|
await safe_answer_callback(target_event)
|
|
except Exception as e_send_edit:
|
|
logging.warning(
|
|
f"Failed to send/edit main menu (user: {user_id}, is_edit: {is_edit}): {type(e_send_edit).__name__} - {e_send_edit}." # noqa: E501
|
|
)
|
|
if is_edit and target_message_obj:
|
|
try:
|
|
await target_message_obj.answer(text, reply_markup=reply_markup)
|
|
except Exception as e_send_new:
|
|
logging.error(
|
|
f"Also failed to send new main menu message for user {user_id}: {e_send_new}"
|
|
)
|
|
if isinstance(target_event, types.CallbackQuery):
|
|
await safe_answer_callback(
|
|
target_event,
|
|
_("error_occurred_try_again") if is_edit else None,
|
|
)
|
|
|
|
|
|
async def send_bot_interface_menu(
|
|
target_event: Union[types.Message, types.CallbackQuery],
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
subscription_service: SubscriptionService,
|
|
session: AsyncSession,
|
|
is_edit: bool = False,
|
|
):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
|
|
if not i18n:
|
|
logging.error("i18n_instance missing in send_bot_interface_menu.")
|
|
return
|
|
|
|
user_id = target_event.from_user.id
|
|
show_trial_button_in_menu = await should_show_trial_button(
|
|
settings, subscription_service, session, user_id
|
|
)
|
|
|
|
text = i18n.gettext(current_lang, "bot_interface_menu_title")
|
|
if settings.SUBSCRIPTION_MINI_APP_URL:
|
|
text = f"{text}\n\n{i18n.gettext(current_lang, 'bot_interface_menu_webapp_hint')}"
|
|
reply_markup = get_bot_interface_inline_keyboard(
|
|
current_lang, i18n, settings, show_trial_button_in_menu
|
|
)
|
|
|
|
target_message_obj: Optional[types.Message] = None
|
|
if isinstance(target_event, types.Message):
|
|
target_message_obj = target_event
|
|
elif isinstance(target_event, types.CallbackQuery) and target_event.message:
|
|
target_message_obj = target_event.message
|
|
|
|
if not target_message_obj:
|
|
logging.error(
|
|
"send_bot_interface_menu: target_message_obj is None for user %s.",
|
|
user_id,
|
|
)
|
|
return
|
|
|
|
try:
|
|
if is_edit:
|
|
await target_message_obj.edit_text(text, reply_markup=reply_markup)
|
|
else:
|
|
await target_message_obj.answer(text, reply_markup=reply_markup)
|
|
|
|
if isinstance(target_event, types.CallbackQuery):
|
|
await safe_answer_callback(target_event)
|
|
except Exception as e_send_edit:
|
|
logging.warning(
|
|
"Failed to send/edit bot interface menu (user: %s, is_edit: %s): %s - %s.",
|
|
user_id,
|
|
is_edit,
|
|
type(e_send_edit).__name__,
|
|
e_send_edit,
|
|
)
|
|
if is_edit:
|
|
try:
|
|
await target_message_obj.answer(text, reply_markup=reply_markup)
|
|
except Exception as e_send_new:
|
|
logging.error(
|
|
"Also failed to send new bot interface menu for user %s: %s",
|
|
user_id,
|
|
e_send_new,
|
|
)
|
|
|
|
|
|
async def ensure_required_channel_subscription(
|
|
event: Union[types.Message, types.CallbackQuery],
|
|
settings: Settings,
|
|
i18n: Optional[JsonI18n],
|
|
current_lang: str,
|
|
session: AsyncSession,
|
|
db_user: Optional[User] = None,
|
|
) -> bool:
|
|
"""
|
|
Verify that the user is a member of the required channel (if configured).
|
|
Returns True when access can proceed, False when user must subscribe first.
|
|
"""
|
|
required_channel_id = normalize_required_channel_id(settings.REQUIRED_CHANNEL_ID)
|
|
if not required_channel_id:
|
|
return True
|
|
|
|
if isinstance(event, types.CallbackQuery):
|
|
user_id = event.from_user.id
|
|
bot_instance: Optional[Bot] = getattr(event, "bot", None)
|
|
if bot_instance is None and event.message:
|
|
bot_instance = event.message.bot
|
|
message_obj: Optional[types.Message] = event.message
|
|
else:
|
|
user_id = event.from_user.id
|
|
bot_instance = event.bot if hasattr(event, "bot") else None
|
|
message_obj = event
|
|
|
|
if bot_instance is None:
|
|
logging.error("Channel subscription check: bot instance missing for user %s.", user_id)
|
|
return False
|
|
|
|
if user_id in settings.ADMIN_IDS:
|
|
return True
|
|
|
|
if db_user is None:
|
|
try:
|
|
db_user = await user_dal.get_user_by_id(session, user_id)
|
|
except Exception as fetch_error:
|
|
logging.error(
|
|
"Channel subscription check: failed to fetch user %s: %s",
|
|
user_id,
|
|
fetch_error,
|
|
exc_info=True,
|
|
)
|
|
return False
|
|
|
|
if not db_user:
|
|
logging.warning(
|
|
"Required channel check skipped because user %s is not persisted yet.",
|
|
user_id,
|
|
)
|
|
return True
|
|
|
|
if (
|
|
db_user.channel_subscription_verified
|
|
and db_user.channel_subscription_verified_for == required_channel_id
|
|
):
|
|
return True
|
|
|
|
def translate(key: str, **kwargs) -> str:
|
|
if i18n:
|
|
return i18n.gettext(current_lang, key, **kwargs)
|
|
return key
|
|
|
|
now = datetime.now(timezone.utc)
|
|
is_member = False
|
|
status_value = None
|
|
|
|
try:
|
|
member = await bot_instance.get_chat_member(required_channel_id, user_id)
|
|
status = getattr(member, "status", None)
|
|
status_value = getattr(status, "value", status)
|
|
allowed_statuses = {"creator", "administrator", "member", "restricted"}
|
|
if status_value in allowed_statuses:
|
|
is_member = True
|
|
except TelegramBadRequest as bad_request:
|
|
if is_required_channel_access_error(bad_request):
|
|
logging.error(
|
|
"Required channel check failed due to channel access/configuration error "
|
|
"(configured=%s, normalized=%s): %s",
|
|
settings.REQUIRED_CHANNEL_ID,
|
|
required_channel_id,
|
|
bad_request,
|
|
)
|
|
error_text = translate("channel_subscription_check_failed")
|
|
if isinstance(event, types.CallbackQuery):
|
|
try:
|
|
await event.answer(error_text, show_alert=True)
|
|
except Exception:
|
|
pass
|
|
if message_obj:
|
|
try:
|
|
await message_obj.answer(error_text)
|
|
except Exception:
|
|
pass
|
|
else:
|
|
await event.answer(error_text)
|
|
return False
|
|
|
|
logging.info(
|
|
"Required channel check: user %s not subscribed (details: %s)",
|
|
user_id,
|
|
bad_request,
|
|
)
|
|
except TelegramForbiddenError as forbidden_error:
|
|
logging.error(
|
|
"Required channel check failed due to insufficient permissions: %s",
|
|
forbidden_error,
|
|
)
|
|
error_text = translate("channel_subscription_check_failed")
|
|
if isinstance(event, types.CallbackQuery):
|
|
try:
|
|
await event.answer(error_text, show_alert=True)
|
|
except Exception:
|
|
pass
|
|
if message_obj:
|
|
try:
|
|
await message_obj.answer(error_text)
|
|
except Exception:
|
|
pass
|
|
else:
|
|
await event.answer(error_text)
|
|
return False
|
|
except TelegramAPIError as api_error:
|
|
logging.error(
|
|
"Required channel check failed for user %s: %s",
|
|
user_id,
|
|
api_error,
|
|
exc_info=True,
|
|
)
|
|
error_text = translate("channel_subscription_check_failed")
|
|
if isinstance(event, types.CallbackQuery):
|
|
try:
|
|
await event.answer(error_text, show_alert=True)
|
|
except Exception:
|
|
pass
|
|
if message_obj:
|
|
try:
|
|
await message_obj.answer(error_text)
|
|
except Exception:
|
|
pass
|
|
else:
|
|
await event.answer(error_text)
|
|
return False
|
|
|
|
update_payload = {
|
|
"channel_subscription_checked_at": now,
|
|
"channel_subscription_verified_for": required_channel_id,
|
|
"channel_subscription_verified": is_member,
|
|
}
|
|
try:
|
|
await user_dal.update_user(session, user_id, update_payload)
|
|
except Exception as update_error:
|
|
logging.error(
|
|
"Failed to persist channel verification result for user %s: %s",
|
|
user_id,
|
|
update_error,
|
|
exc_info=True,
|
|
)
|
|
|
|
if is_member:
|
|
logging.info(
|
|
"User %s confirmed as member of required channel %s (status=%s).",
|
|
user_id,
|
|
required_channel_id,
|
|
status_value,
|
|
)
|
|
return True
|
|
|
|
channel_link = await resolve_required_channel_link(
|
|
bot_instance,
|
|
required_channel_id,
|
|
settings.REQUIRED_CHANNEL_LINK,
|
|
)
|
|
keyboard = get_channel_subscription_keyboard(current_lang, i18n, channel_link) if i18n else None
|
|
|
|
prompt_text = translate("channel_subscription_required")
|
|
|
|
if isinstance(event, types.CallbackQuery):
|
|
if keyboard and event.message:
|
|
try:
|
|
await event.message.edit_text(prompt_text, reply_markup=keyboard)
|
|
except Exception as edit_error:
|
|
logging.debug(
|
|
"Failed to edit prompt message for user %s: %s",
|
|
user_id,
|
|
edit_error,
|
|
)
|
|
if keyboard is None and message_obj:
|
|
try:
|
|
await message_obj.answer(prompt_text)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
await event.answer(prompt_text, show_alert=True)
|
|
except Exception:
|
|
pass
|
|
else:
|
|
await event.answer(prompt_text, reply_markup=keyboard)
|
|
|
|
return False
|
|
|
|
|
|
@router.message(CommandStart())
|
|
@router.message(
|
|
CommandStart(
|
|
magic=F.args.regexp(r"^ref_([A-Za-z0-9_-]{1,64})$").as_("ref_match")
|
|
)
|
|
)
|
|
@router.message(CommandStart(magic=F.args.regexp(r"^promo_([A-Za-z0-9_-]{1,100})$").as_("promo_match")))
|
|
@router.message(CommandStart(magic=F.args.regexp(r"^admin_user_(\d+)$").as_("admin_user_match")))
|
|
@router.message(CommandStart(magic=F.args.regexp(r"^ticket_(\d+)$").as_("ticket_match")))
|
|
@router.message(CommandStart(magic=F.args.regexp(r"^notifications$").as_("notifications_match")))
|
|
@router.message(CommandStart(magic=F.args.regexp(r"^page_ref$").as_("page_ref_match")))
|
|
@router.message(
|
|
CommandStart(
|
|
magic=F.args.regexp(
|
|
r"^(?!ref_|promo_|admin_user_|ticket_|notifications$|page_ref$|webapp_auth_)([A-Za-z0-9_\-]{2,64})$"
|
|
).as_("ad_param_match")
|
|
)
|
|
)
|
|
async def start_command_handler(
|
|
message: types.Message,
|
|
state: FSMContext,
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
subscription_service: SubscriptionService,
|
|
referral_service: ReferralService,
|
|
session: AsyncSession,
|
|
ref_match: Optional[re.Match] = None,
|
|
promo_match: Optional[re.Match] = None,
|
|
page_ref_match: Optional[re.Match] = None,
|
|
ad_param_match: Optional[re.Match] = None,
|
|
admin_user_match: Optional[re.Match] = None,
|
|
ticket_match: Optional[re.Match] = None,
|
|
notifications_match: Optional[re.Match] = None,
|
|
):
|
|
await state.clear()
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if i18n else key
|
|
|
|
user = message.from_user
|
|
user_id = user.id
|
|
|
|
if admin_user_match and user_id in settings.ADMIN_IDS:
|
|
target_user_id = int(admin_user_match.group(1))
|
|
target_user = await user_dal.get_user_by_id(session, target_user_id)
|
|
if not target_user:
|
|
await message.answer(_("admin_user_not_found", input=hd.quote(str(target_user_id))))
|
|
return
|
|
|
|
try:
|
|
from bot.handlers.admin.user_management import (
|
|
_send_with_profile_link_fallback,
|
|
format_user_card,
|
|
get_user_card_keyboard,
|
|
)
|
|
|
|
referral_service = ReferralService(settings, subscription_service, message.bot, i18n)
|
|
user_card_text = await format_user_card(
|
|
target_user,
|
|
session,
|
|
subscription_service,
|
|
i18n,
|
|
current_lang,
|
|
referral_service,
|
|
)
|
|
keyboard = get_user_card_keyboard(
|
|
target_user.user_id,
|
|
i18n,
|
|
current_lang,
|
|
target_user.referred_by_id,
|
|
)
|
|
|
|
await _send_with_profile_link_fallback(
|
|
message.answer,
|
|
text=user_card_text,
|
|
markup=keyboard.as_markup(),
|
|
user_id=target_user.user_id,
|
|
parse_mode="HTML",
|
|
)
|
|
return
|
|
except Exception as e_admin_card:
|
|
logging.error(
|
|
"Failed to open admin user card via deep-link for %s: %s",
|
|
target_user_id,
|
|
e_admin_card,
|
|
exc_info=True,
|
|
)
|
|
await message.answer(_("admin_user_card_error"))
|
|
return
|
|
|
|
if ticket_match:
|
|
ticket_id = int(ticket_match.group(1))
|
|
base_url = (settings.SUBSCRIPTION_MINI_APP_URL or "").strip()
|
|
if base_url:
|
|
ticket_url = f"{base_url.rstrip('/')}/support/{ticket_id}"
|
|
keyboard = types.InlineKeyboardMarkup(
|
|
inline_keyboard=[
|
|
[
|
|
types.InlineKeyboardButton(
|
|
text=i18n.gettext(current_lang, "wa_support_open_ticket")
|
|
if i18n
|
|
else "Открыть тикет",
|
|
web_app=types.WebAppInfo(url=ticket_url),
|
|
)
|
|
]
|
|
]
|
|
)
|
|
await message.answer(
|
|
i18n.gettext(current_lang, "wa_support_open_ticket_hint")
|
|
if i18n
|
|
else "Откройте тикет в Mini App.",
|
|
reply_markup=keyboard,
|
|
)
|
|
return
|
|
|
|
referred_by_user_id: Optional[int] = None
|
|
promo_code_to_apply: Optional[str] = None
|
|
should_open_referral_from_start = False
|
|
ad_start_param: Optional[str] = None
|
|
notifications_start_requested = bool(notifications_match)
|
|
|
|
if ref_match:
|
|
raw_ref_value = ref_match.group(1)
|
|
referred_by_user_id = await _resolve_referrer_from_start_ref(
|
|
session,
|
|
raw_ref_value,
|
|
settings=settings,
|
|
current_user_id=user_id,
|
|
)
|
|
elif promo_match:
|
|
promo_code_to_apply = promo_match.group(1)
|
|
logging.info(f"User {user_id} started with promo code: {promo_code_to_apply}")
|
|
elif notifications_start_requested:
|
|
logging.info("User %s started bot from notifications deep-link.", user_id)
|
|
elif page_ref_match:
|
|
should_open_referral_from_start = True
|
|
logging.info(f"User {user_id} started with page_ref deep-link.")
|
|
elif ad_param_match:
|
|
ad_start_param = ad_param_match.group(1)
|
|
logging.info(f"User {user_id} started with ad start param: {ad_start_param}")
|
|
|
|
sanitized_username = sanitize_username(user.username)
|
|
sanitized_first_name = sanitize_display_name(user.first_name)
|
|
sanitized_last_name = sanitize_display_name(user.last_name)
|
|
notification_status_now = datetime.now(timezone.utc)
|
|
|
|
db_user = await user_dal.get_user_by_id(session, user_id)
|
|
is_existing_user = db_user is not None
|
|
if not db_user:
|
|
user_data_to_create = {
|
|
"user_id": user_id,
|
|
"telegram_id": user_id,
|
|
"username": sanitized_username,
|
|
"first_name": sanitized_first_name,
|
|
"last_name": sanitized_last_name,
|
|
"language_code": current_lang,
|
|
"referred_by_id": referred_by_user_id,
|
|
"registration_date": datetime.now(timezone.utc),
|
|
"telegram_notifications_status": TELEGRAM_NOTIFICATIONS_ENABLED,
|
|
"telegram_notifications_checked_at": notification_status_now,
|
|
"telegram_notifications_enabled_at": notification_status_now,
|
|
"telegram_notifications_blocked_at": None,
|
|
}
|
|
try:
|
|
db_user, created = await user_dal.create_user(session, user_data_to_create)
|
|
|
|
if created:
|
|
try:
|
|
await session.commit()
|
|
except Exception as commit_error:
|
|
await session.rollback()
|
|
logging.error(
|
|
f"Failed to commit new user {user_id}: {commit_error}",
|
|
exc_info=True,
|
|
)
|
|
await message.answer(_("error_occurred_processing_request"))
|
|
return
|
|
|
|
logging.info(
|
|
f"New user {user_id} added to session. Referred by: {referred_by_user_id or 'N/A'}." # noqa: E501
|
|
)
|
|
|
|
# Auto-grant referral welcome bonus to newly registered referred users.
|
|
referral_welcome_days = max(
|
|
0, int(getattr(settings, "REFERRAL_WELCOME_BONUS_DAYS", 0) or 0)
|
|
)
|
|
if referred_by_user_id and referral_welcome_days > 0:
|
|
try:
|
|
referral_bonus_end_date = (
|
|
await subscription_service.extend_active_subscription_days(
|
|
session,
|
|
user_id,
|
|
referral_welcome_days,
|
|
reason="referral_welcome_bonus",
|
|
)
|
|
)
|
|
if referral_bonus_end_date:
|
|
await session.commit()
|
|
logging.info(
|
|
"Referral welcome bonus applied: user %s got %s days, new end date %s.", # noqa: E501
|
|
user_id,
|
|
referral_welcome_days,
|
|
referral_bonus_end_date.isoformat(),
|
|
)
|
|
await message.answer(
|
|
_(
|
|
"referral_welcome_bonus_applied",
|
|
days=referral_welcome_days,
|
|
end_date=referral_bonus_end_date.strftime("%d.%m.%Y %H:%M:%S"),
|
|
),
|
|
parse_mode="HTML",
|
|
)
|
|
else:
|
|
await session.rollback()
|
|
logging.warning(
|
|
"Referral welcome bonus was not applied for user %s (referred by %s).", # noqa: E501
|
|
user_id,
|
|
referred_by_user_id,
|
|
)
|
|
except Exception as referral_bonus_error:
|
|
await session.rollback()
|
|
logging.error(
|
|
"Failed to apply referral welcome bonus for user %s: %s",
|
|
user_id,
|
|
referral_bonus_error,
|
|
exc_info=True,
|
|
)
|
|
|
|
# Send notification about new user registration
|
|
try:
|
|
from bot.services.notification_service import NotificationService
|
|
|
|
notification_service = NotificationService(message.bot, settings, i18n)
|
|
await notification_service.notify_new_user_registration(
|
|
user_id=user_id,
|
|
username=sanitized_username,
|
|
first_name=sanitized_first_name,
|
|
referred_by_id=referred_by_user_id,
|
|
)
|
|
except Exception as e:
|
|
logging.error(f"Failed to send new user notification: {e}")
|
|
except Exception as e_create:
|
|
logging.error(f"Failed to add new user {user_id} to session: {e_create}", exc_info=True)
|
|
await message.answer(_("error_occurred_processing_request"))
|
|
return
|
|
else:
|
|
update_payload = {}
|
|
if db_user.language_code != current_lang:
|
|
update_payload["language_code"] = current_lang
|
|
if db_user.telegram_id != user_id:
|
|
update_payload["telegram_id"] = user_id
|
|
if db_user.telegram_notifications_status != TELEGRAM_NOTIFICATIONS_ENABLED:
|
|
update_payload["telegram_notifications_status"] = TELEGRAM_NOTIFICATIONS_ENABLED
|
|
update_payload["telegram_notifications_checked_at"] = notification_status_now
|
|
update_payload["telegram_notifications_enabled_at"] = notification_status_now
|
|
update_payload["telegram_notifications_blocked_at"] = None
|
|
# Set referral only if not already set AND user is not currently active.
|
|
# This allows previously subscribed but currently inactive users to be attributed.
|
|
if referred_by_user_id and db_user.referred_by_id is None:
|
|
try:
|
|
is_active_now = await subscription_service.has_active_subscription(session, user_id)
|
|
except Exception:
|
|
is_active_now = False
|
|
if not is_active_now:
|
|
update_payload["referred_by_id"] = referred_by_user_id
|
|
if sanitized_username != db_user.username:
|
|
update_payload["username"] = sanitized_username
|
|
if sanitized_first_name != db_user.first_name:
|
|
update_payload["first_name"] = sanitized_first_name
|
|
if sanitized_last_name != db_user.last_name:
|
|
update_payload["last_name"] = sanitized_last_name
|
|
|
|
if update_payload:
|
|
try:
|
|
await user_dal.update_user(session, user_id, update_payload)
|
|
|
|
logging.info(f"Updated existing user {user_id} in session: {update_payload}")
|
|
except Exception as e_update:
|
|
logging.error(
|
|
f"Failed to update existing user {user_id} in session: {e_update}",
|
|
exc_info=True,
|
|
)
|
|
|
|
# Attribute user to ad campaign if start param provided
|
|
if ad_start_param:
|
|
try:
|
|
from db.dal import ad_dal as _ad_dal
|
|
|
|
campaign = await _ad_dal.get_campaign_by_start_param(session, ad_start_param)
|
|
if campaign and campaign.is_active:
|
|
await _ad_dal.ensure_attribution(
|
|
session, user_id=user_id, campaign_id=campaign.ad_campaign_id
|
|
)
|
|
await session.commit()
|
|
except Exception as e_attr:
|
|
logging.error(f"Failed to attribute user {user_id} to ad '{ad_start_param}': {e_attr}")
|
|
try:
|
|
await session.rollback()
|
|
except Exception:
|
|
pass
|
|
|
|
if not await ensure_required_channel_subscription(
|
|
message, settings, i18n, current_lang, session, db_user
|
|
):
|
|
return
|
|
|
|
open_referral_page_for_existing_user = should_open_referral_from_start and is_existing_user
|
|
|
|
# Send welcome message if not disabled
|
|
if (
|
|
not settings.DISABLE_WELCOME_MESSAGE
|
|
and not open_referral_page_for_existing_user
|
|
and not notifications_start_requested
|
|
):
|
|
await message.answer(_(key="welcome", user_name=hd.quote(user.full_name)))
|
|
|
|
if notifications_start_requested:
|
|
await message.answer(_("telegram_notifications_started"), parse_mode="HTML")
|
|
|
|
# Auto-apply promo code if provided via start parameter
|
|
if promo_code_to_apply:
|
|
try:
|
|
from bot.services.promo_code_service import PromoCodeService
|
|
|
|
promo_code_service = PromoCodeService(settings, subscription_service, message.bot, i18n)
|
|
|
|
success, result = await promo_code_service.apply_promo_code(
|
|
session, user_id, promo_code_to_apply, current_lang
|
|
)
|
|
|
|
if success:
|
|
await session.commit()
|
|
logging.info(f"Auto-applied promo code '{promo_code_to_apply}' for user {user_id}")
|
|
|
|
# Get updated subscription details
|
|
active = await subscription_service.get_active_subscription_details(
|
|
session, user_id
|
|
)
|
|
config_link_display = active.get("config_link") if active else None
|
|
connect_button_url = active.get("connect_button_url") if active else None
|
|
config_link_text = config_link_display or _("config_link_not_available")
|
|
|
|
new_end_date = result if isinstance(result, datetime) else None
|
|
|
|
promo_success_text = _(
|
|
"promo_code_applied_success_full",
|
|
end_date=(
|
|
new_end_date.strftime("%d.%m.%Y %H:%M:%S") if new_end_date else "N/A"
|
|
),
|
|
config_link=config_link_text,
|
|
)
|
|
install_links = await ensure_user_install_guide_links(session, settings, user_id)
|
|
install_share_url = install_links.public_share_url
|
|
if install_share_url:
|
|
try:
|
|
await session.commit()
|
|
promo_success_text = append_install_share_link_text(
|
|
promo_success_text,
|
|
_,
|
|
install_share_url,
|
|
)
|
|
except Exception:
|
|
await session.rollback()
|
|
logging.exception(
|
|
"Failed to persist install guide share token for promo user %s.",
|
|
user_id,
|
|
)
|
|
install_share_url = None
|
|
|
|
from bot.keyboards.inline.user_keyboards import get_connect_and_main_keyboard
|
|
|
|
await message.answer(
|
|
promo_success_text,
|
|
reply_markup=get_connect_and_main_keyboard(
|
|
current_lang,
|
|
i18n,
|
|
settings,
|
|
config_link_display,
|
|
connect_button_url=connect_button_url,
|
|
install_share_url=install_share_url,
|
|
),
|
|
parse_mode="HTML",
|
|
)
|
|
|
|
# Don't show main menu if promo was successfully applied
|
|
return
|
|
else:
|
|
await session.commit()
|
|
logging.warning(
|
|
f"Failed to auto-apply promo code '{promo_code_to_apply}' for user {user_id}: {result}" # noqa: E501
|
|
)
|
|
await message.answer(str(result), parse_mode="HTML")
|
|
# Continue to show main menu if promo failed
|
|
|
|
except Exception as e:
|
|
logging.error(
|
|
f"Error auto-applying promo code '{promo_code_to_apply}' for user {user_id}: {e}"
|
|
)
|
|
await session.rollback()
|
|
|
|
if open_referral_page_for_existing_user:
|
|
from . import referral as user_referral_handlers
|
|
|
|
await user_referral_handlers.referral_command_handler(
|
|
message, settings, i18n_data, referral_service, message.bot, session
|
|
)
|
|
return
|
|
|
|
await send_main_menu(message, settings, i18n_data, subscription_service, session, is_edit=False)
|
|
|
|
|
|
@router.message(Command("tg"))
|
|
async def tg_interface_command_handler(
|
|
message: types.Message,
|
|
state: FSMContext,
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
subscription_service: SubscriptionService,
|
|
session: AsyncSession,
|
|
):
|
|
await state.clear()
|
|
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
db_user = await user_dal.get_user_by_id(session, message.from_user.id)
|
|
if not await ensure_required_channel_subscription(
|
|
message, settings, i18n, current_lang, session, db_user
|
|
):
|
|
return
|
|
|
|
await send_bot_interface_menu(
|
|
message, settings, i18n_data, subscription_service, session, is_edit=False
|
|
)
|
|
|
|
|
|
@router.callback_query(F.data == "channel_subscription:verify")
|
|
async def verify_channel_subscription_callback(
|
|
callback: types.CallbackQuery,
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
subscription_service: SubscriptionService,
|
|
session: AsyncSession,
|
|
):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
|
|
db_user = await user_dal.get_user_by_id(session, callback.from_user.id)
|
|
|
|
verified = await ensure_required_channel_subscription(
|
|
callback, settings, i18n, current_lang, session, db_user
|
|
)
|
|
if not verified:
|
|
return
|
|
|
|
if db_user and db_user.language_code:
|
|
current_lang = db_user.language_code
|
|
i18n_data["current_language"] = current_lang
|
|
|
|
if i18n:
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
|
|
else:
|
|
_ = lambda key, **kwargs: key
|
|
|
|
if not settings.DISABLE_WELCOME_MESSAGE:
|
|
welcome_text = _(key="welcome", user_name=hd.quote(callback.from_user.full_name))
|
|
if callback.message:
|
|
await callback.message.answer(welcome_text)
|
|
else:
|
|
fallback_bot: Optional[Bot] = getattr(callback, "bot", None)
|
|
if fallback_bot:
|
|
await fallback_bot.send_message(callback.from_user.id, welcome_text)
|
|
|
|
try:
|
|
await safe_answer_callback(
|
|
callback,
|
|
_(key="channel_subscription_verified_success"),
|
|
show_alert=True,
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
await send_main_menu(
|
|
callback, settings, i18n_data, subscription_service, session, is_edit=bool(callback.message)
|
|
)
|
|
|
|
|
|
@router.message(Command("language"))
|
|
@router.callback_query(F.data == "main_action:language")
|
|
async def language_command_handler(
|
|
event: Union[types.Message, types.CallbackQuery],
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
back_callback: str = "main_action:back_to_main",
|
|
):
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if i18n else key
|
|
|
|
text_to_send = _(key="choose_language")
|
|
reply_markup = get_language_selection_keyboard(
|
|
i18n,
|
|
current_lang,
|
|
back_callback=back_callback,
|
|
)
|
|
|
|
target_message_obj = event.message if isinstance(event, types.CallbackQuery) else event
|
|
if not target_message_obj:
|
|
if isinstance(event, types.CallbackQuery):
|
|
await safe_answer_callback(
|
|
event,
|
|
_("error_occurred_try_again"),
|
|
show_alert=True,
|
|
)
|
|
return
|
|
|
|
if isinstance(event, types.CallbackQuery):
|
|
if event.message:
|
|
try:
|
|
await event.message.edit_text(text_to_send, reply_markup=reply_markup)
|
|
except Exception:
|
|
await target_message_obj.answer(text_to_send, reply_markup=reply_markup)
|
|
await safe_answer_callback(event)
|
|
else:
|
|
await target_message_obj.answer(text_to_send, reply_markup=reply_markup)
|
|
|
|
|
|
@router.callback_query(F.data.startswith("set_lang_"))
|
|
async def select_language_callback_handler(
|
|
callback: types.CallbackQuery,
|
|
i18n_data: dict,
|
|
settings: Settings,
|
|
subscription_service: SubscriptionService,
|
|
session: AsyncSession,
|
|
):
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
if not i18n or not callback.message:
|
|
await safe_answer_callback(
|
|
callback,
|
|
"Service error or message context lost.",
|
|
show_alert=True,
|
|
)
|
|
return
|
|
|
|
try:
|
|
lang_payload = callback.data.split("_", 2)[2]
|
|
raw_lang_code, _, return_target = lang_payload.partition(":")
|
|
lang_code = normalize_locale_language_code(
|
|
raw_lang_code,
|
|
set(i18n.locales_data.keys()),
|
|
prefer_known_base=True,
|
|
)
|
|
except IndexError:
|
|
await safe_answer_callback(
|
|
callback,
|
|
"Error processing language selection.",
|
|
show_alert=True,
|
|
)
|
|
return
|
|
if lang_code not in i18n.locales_data:
|
|
await safe_answer_callback(
|
|
callback,
|
|
"Unsupported language.",
|
|
show_alert=True,
|
|
)
|
|
return
|
|
|
|
user_id = callback.from_user.id
|
|
try:
|
|
updated = await user_dal.update_user_language(session, user_id, lang_code)
|
|
if updated:
|
|
i18n_data["current_language"] = lang_code
|
|
_ = lambda key, **kwargs: i18n.gettext(lang_code, key, **kwargs)
|
|
await safe_answer_callback(callback, _(key="language_set_alert"))
|
|
logging.info(f"User {user_id} language updated to {lang_code} in session.")
|
|
else:
|
|
await safe_answer_callback(
|
|
callback,
|
|
"Could not set language.",
|
|
show_alert=True,
|
|
)
|
|
return
|
|
except Exception as e_lang_update:
|
|
logging.error(f"Error updating lang for user {user_id}: {e_lang_update}", exc_info=True)
|
|
await safe_answer_callback(callback, "Error setting language.", show_alert=True)
|
|
return
|
|
if return_target == "bot":
|
|
await send_bot_interface_menu(
|
|
callback, settings, i18n_data, subscription_service, session, is_edit=True
|
|
)
|
|
else:
|
|
await send_main_menu(
|
|
callback, settings, i18n_data, subscription_service, session, is_edit=True
|
|
)
|
|
|
|
|
|
@router.callback_query(F.data.startswith("main_action:"))
|
|
async def main_action_callback_handler(
|
|
callback: types.CallbackQuery,
|
|
state: FSMContext,
|
|
settings: Settings,
|
|
i18n_data: dict,
|
|
bot: Bot,
|
|
subscription_service: SubscriptionService,
|
|
referral_service: ReferralService,
|
|
panel_service: PanelApiService,
|
|
promo_code_service: PromoCodeService,
|
|
session: AsyncSession,
|
|
):
|
|
action = callback.data.split(":")[1]
|
|
|
|
if action in {"back_to_main", "back_to_main_keep", "bot_interface"}:
|
|
await state.clear()
|
|
|
|
from . import promo_user as user_promo_handlers
|
|
from . import referral as user_referral_handlers
|
|
from . import subscription as user_subscription_handlers
|
|
from . import trial_handler as user_trial_handlers
|
|
|
|
if not callback.message:
|
|
await safe_answer_callback(
|
|
callback,
|
|
"Error: message context lost.",
|
|
show_alert=True,
|
|
)
|
|
return
|
|
|
|
if action == "subscribe":
|
|
await user_subscription_handlers.display_subscription_options(
|
|
callback, i18n_data, settings, session
|
|
)
|
|
elif action == "bot_subscribe":
|
|
await user_subscription_handlers.display_subscription_options(
|
|
callback,
|
|
i18n_data,
|
|
settings,
|
|
session,
|
|
back_callback="main_action:bot_interface",
|
|
)
|
|
elif action == "my_subscription":
|
|
await user_subscription_handlers.my_subscription_command_handler(
|
|
callback, i18n_data, settings, panel_service, subscription_service, session, bot
|
|
)
|
|
elif action == "bot_my_subscription":
|
|
await user_subscription_handlers.my_subscription_command_handler(
|
|
callback,
|
|
i18n_data,
|
|
settings,
|
|
panel_service,
|
|
subscription_service,
|
|
session,
|
|
bot,
|
|
back_callback="main_action:bot_interface",
|
|
)
|
|
elif action == "my_devices":
|
|
await user_subscription_handlers.my_devices_command_handler(
|
|
callback, i18n_data, settings, panel_service, subscription_service, session, bot
|
|
)
|
|
elif action == "referral":
|
|
await user_referral_handlers.referral_command_handler(
|
|
callback, settings, i18n_data, referral_service, bot, session
|
|
)
|
|
elif action == "bot_referral":
|
|
await user_referral_handlers.referral_command_handler(
|
|
callback,
|
|
settings,
|
|
i18n_data,
|
|
referral_service,
|
|
bot,
|
|
session,
|
|
back_callback="main_action:bot_interface",
|
|
)
|
|
elif action == "apply_promo":
|
|
await user_promo_handlers.prompt_promo_code_input(
|
|
callback, state, i18n_data, settings, session
|
|
)
|
|
elif action == "bot_apply_promo":
|
|
await user_promo_handlers.prompt_promo_code_input(
|
|
callback,
|
|
state,
|
|
i18n_data,
|
|
settings,
|
|
session,
|
|
back_callback="main_action:bot_interface",
|
|
)
|
|
elif action == "request_trial":
|
|
await user_trial_handlers.request_trial_confirmation_handler(
|
|
callback, settings, i18n_data, subscription_service, session
|
|
)
|
|
elif action == "language":
|
|
await language_command_handler(callback, i18n_data, settings)
|
|
elif action == "bot_language":
|
|
await language_command_handler(
|
|
callback,
|
|
i18n_data,
|
|
settings,
|
|
back_callback="main_action:bot_interface",
|
|
)
|
|
elif action == "bot_interface":
|
|
await send_bot_interface_menu(
|
|
callback, settings, i18n_data, subscription_service, session, is_edit=True
|
|
)
|
|
elif action in {"info", "bot_info"}:
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
|
|
if not i18n:
|
|
await safe_answer_callback(
|
|
callback,
|
|
"Language service error.",
|
|
show_alert=True,
|
|
)
|
|
return
|
|
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if i18n else key
|
|
|
|
privacy_url = settings.PRIVACY_POLICY_URL
|
|
user_agreement_url = settings.USER_AGREEMENT_URL or settings.TERMS_OF_SERVICE_URL
|
|
|
|
if not privacy_url and not user_agreement_url:
|
|
await safe_answer_callback(
|
|
callback,
|
|
_("error_occurred_try_again"),
|
|
show_alert=True,
|
|
)
|
|
return
|
|
|
|
reply_markup = get_information_links_keyboard(
|
|
current_lang,
|
|
i18n,
|
|
privacy_url,
|
|
user_agreement_url,
|
|
back_callback=(
|
|
"main_action:bot_interface"
|
|
if callback.data == "main_action:bot_info"
|
|
else "main_action:back_to_main"
|
|
),
|
|
)
|
|
try:
|
|
await callback.message.edit_text(_(key="info_links_message"), reply_markup=reply_markup)
|
|
except Exception:
|
|
await callback.message.answer(_(key="info_links_message"), reply_markup=reply_markup)
|
|
await safe_answer_callback(callback)
|
|
elif action == "back_to_main":
|
|
await send_main_menu(
|
|
callback, settings, i18n_data, subscription_service, session, is_edit=True
|
|
)
|
|
elif action == "back_to_main_keep":
|
|
await send_main_menu(
|
|
callback, settings, i18n_data, subscription_service, session, is_edit=False
|
|
)
|
|
else:
|
|
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
|
|
_ = lambda key, **kwargs: (
|
|
i18n.gettext(i18n_data.get("current_language"), key, **kwargs) if i18n else key
|
|
)
|
|
await safe_answer_callback(
|
|
callback,
|
|
_("main_menu_unknown_action"),
|
|
show_alert=True,
|
|
)
|