Compare commits

...
27 Commits
Author SHA1 Message Date
Machka PaslaandGitHub 4d8cf45bda Merge pull request #31 from machka-pasla/codex/remove-auto-renew-for-attribute
Disable tribute auto-renew and add admin alerts
2025-07-06 13:31:14 +07:00
Machka Pasla df987fc52a Remove tribute auto renew and add admin notifications 2025-07-06 13:31:04 +07:00
Machka PaslaandGitHub 07cb68457f Merge pull request #30 from machka-pasla/codex/уточнить-работу-автопродления-с-tribute
Implement auto-renewal for Tribute
2025-07-05 00:05:51 +07:00
Machka Pasla 1e666f90d3 Improve Tribute auto-renew 2025-07-05 00:05:33 +07:00
Machka PaslaandGitHub 6241f766a2 Merge pull request #29 from machka-pasla/codex/add-description-support-for-/start-command
Fix mini app registration
2025-07-04 22:45:19 +07:00
Machka Pasla 451581f792 Fix mini app registration 2025-07-04 22:43:04 +07:00
Machka PaslaandGitHub a015946b20 Merge pull request #28 from machka-pasla/codex/add-description-support-for-/start-command
Enable mini app via API without menu change
2025-07-04 22:35:48 +07:00
Machka Pasla d1d701bff3 feat: allow /start description and keep menu default 2025-07-04 22:35:13 +07:00
Machka PaslaandGitHub ea50567a42 Merge pull request #26 from machka-pasla/codex/добавить-поддержку-mini-app-для-подписки
Support subscription mini app
2025-07-04 20:45:59 +07:00
Machka Pasla a8b62ccc03 Add subscription mini app support 2025-07-04 20:45:27 +07:00
Machka PaslaandGitHub 0e03d72dd3 Merge pull request #25 from machka-pasla/codex/refactor-.env-variables-for-consistency
Fix panel user sync and clarify traffic settings
2025-06-29 00:53:44 +03:00
Machka Pasla aa5eb74d75 Rename user traffic env vars 2025-06-29 00:52:27 +03:00
Machka PaslaandGitHub fd2d0dd570 Merge pull request #24 from machka-pasla/codex/fix-typeerror-with-unsupported-operand-types-for
Fix admin state filter
2025-06-27 20:58:44 +03:00
Machka Pasla abaaf65da9 Fix state filter in promo codes handler 2025-06-27 20:58:00 +03:00
Machka PaslaandGitHub 699964adc0 Merge pull request #23 from machka-pasla/dev
Dev
2025-06-27 20:44:01 +03:00
Machka PaslaandGitHub 74b5587c77 Merge pull request #22 from machka-pasla/codex/refactor-code-for-readability-and-consistency
Add promo management and unify user lookup
2025-06-27 20:43:25 +03:00
Machka Pasla d196801e2e Add promo management and unified user lookup 2025-06-27 20:38:03 +03:00
Machka PaslaandGitHub c12958e9a2 Merge pull request #21 from machka-pasla/codex/fix-activeuserinbounds-filter-in-api-request
Fix default inbounds not applied on user creation
2025-06-27 14:29:46 +03:00
Machka Pasla f4bb61dd1b Use default inbound UUIDs when creating panel user 2025-06-27 14:29:16 +03:00
Machka PaslaandGitHub 24a7e6cc6a Merge pull request #20 from machka-pasla/codex/fix-payment-error-with-tribute
Fix Tribute duplicate payment error
2025-06-27 00:17:27 +03:00
Machka Pasla c6f5e06eba Handle duplicate Tribute payments 2025-06-27 00:17:09 +03:00
Machka PaslaandGitHub ca35567012 Fix typo 2025-06-26 01:36:43 +03:00
Machka PaslaandGitHub cdefe27f89 Merge pull request #18 from machka-pasla/codex/implement-user-creation-and-subscription-updates
Improve panel user handling and promo code activation
2025-06-26 01:28:36 +03:00
Machka Pasla 7200d6e342 Handle missing panel users and allow promo codes 2025-06-26 01:26:55 +03:00
Machka PaslaandGitHub fd4b9b828b Merge pull request #16 from machka-pasla/codex/refactor-panel-unreachable-handling
Fix panel unreachable handling
2025-06-26 00:59:03 +03:00
Machka Pasla b2a0ea5bd2 Remove panel unreachable placeholders and enhance sync 2025-06-26 00:57:48 +03:00
Machka PaslaandGitHub b97774c5b5 Update README.md 2025-06-25 22:15:23 +03:00
25 changed files with 1242 additions and 590 deletions
+8 -6
View File
@@ -16,6 +16,8 @@ DEFAULT_CURRENCY_SYMBOL="RUB" # e.g., RUB, USD, EUR
SUPPORT_LINK=https://t.me/your_support_link SUPPORT_LINK=https://t.me/your_support_link
SERVER_STATUS_URL=https://status.yourdomain.tld/status/your_service SERVER_STATUS_URL=https://status.yourdomain.tld/status/your_service
TERMS_OF_SERVICE_URL=https://example.com/tos TERMS_OF_SERVICE_URL=https://example.com/tos
SUBSCRIPTION_MINI_APP_URL=
START_COMMAND_DESCRIPTION=
# YooKassa Payment Gateway Configuration # YooKassa Payment Gateway Configuration
YOOKASSA_SHOP_ID=your_shop_id YOOKASSA_SHOP_ID=your_shop_id
@@ -78,13 +80,13 @@ REFEREE_BONUS_DAYS_12_MONTHS=15
PANEL_API_URL=http://your_panel_api_url/api PANEL_API_URL=http://your_panel_api_url/api
PANEL_API_KEY=your_panel_api_key PANEL_API_KEY=your_panel_api_key
# Default settings for NEW panel users # User traffic limits (applied for all users)
PANEL_USER_DEFAULT_EXPIRE_DAYS=1 # 0 means unlimited
PANEL_USER_DEFAULT_TRAFFIC_BYTES=0 USER_TRAFFIC_LIMIT_GB=0
PANEL_USER_DEFAULT_TRAFFIC_STRATEGY="NO_RESET" USER_TRAFFIC_STRATEGY="NO_RESET"
# Default Inbounds for Panel Users (Optional, comma-separated UUIDs) # Default Inbounds for Users (Optional, comma-separated UUIDs)
PANEL_USER_DEFAULT_INBOUND_UUIDS=uuid1,uuid2,uuid3 USER_INBOUND_UUIDS=uuid1,uuid2,uuid3
# Trial Settings # Trial Settings
TRIAL_ENABLED=True TRIAL_ENABLED=True
+5 -2
View File
@@ -83,6 +83,8 @@ This Telegram bot is designed to automate the sale and management of subscriptio
* `DEFAULT_CURRENCY_SYMBOL`: e.g., `RUB`, `USD`. * `DEFAULT_CURRENCY_SYMBOL`: e.g., `RUB`, `USD`.
* `SUPPORT_LINK`: (Optional) URL for a support chat/contact (e.g., `https://t.me/your_support`). * `SUPPORT_LINK`: (Optional) URL for a support chat/contact (e.g., `https://t.me/your_support`).
* `SERVER_STATUS_URL`: (Optional) URL to a server status page (e.g., Uptime Kuma). * `SERVER_STATUS_URL`: (Optional) URL to a server status page (e.g., Uptime Kuma).
* `SUBSCRIPTION_MINI_APP_URL`: (Optional) URL of the Telegram mini app for viewing subscription details. If set, the "My Subscription" button will open this mini app and the bot will register it automatically via API.
* `START_COMMAND_DESCRIPTION`: (Optional) Description for the `/start` command shown in the bot's menu.
* **YooKassa Settings:** * **YooKassa Settings:**
* `YOOKASSA_SHOP_ID`: Your shop ID from YooKassa. * `YOOKASSA_SHOP_ID`: Your shop ID from YooKassa.
* `YOOKASSA_SECRET_KEY`: Your secret key from YooKassa. * `YOOKASSA_SECRET_KEY`: Your secret key from YooKassa.
@@ -98,9 +100,10 @@ This Telegram bot is designed to automate the sale and management of subscriptio
`1_MONTH_ENABLED`, `RUB_PRICE_1_MONTH`, `STARS_PRICE_1_MONTH`, `TRIBUTE_LINK_1_MONTH` `1_MONTH_ENABLED`, `RUB_PRICE_1_MONTH`, `STARS_PRICE_1_MONTH`, `TRIBUTE_LINK_1_MONTH`
(and corresponding variables for `3_MONTHS`, `6_MONTHS`, `12_MONTHS`). (and corresponding variables for `3_MONTHS`, `6_MONTHS`, `12_MONTHS`).
* **Panel API Settings:** * **Panel API Settings:**
* `PANEL_API_URL`: Full URL to your Remnawave panel's API (e.g., `http://localhost:3000/api` or `https://panel.yourdomain.com/api`). * `PANEL_API_URL`: Full URL to your Remnawave panel's API (e.g., `http://remnawave:3000/api` or `https://panel.yourdomain.com/api`).
* `PANEL_API_KEY`: API Key for authenticating with the Remnawave panel. * `PANEL_API_KEY`: API Key for authenticating with the Remnawave panel.
* `PANEL_USER_DEFAULT_INBOUND_UUIDS`: (Optional) Comma-separated list of inbound UUIDs from your panel to assign to users. If empty, `activateAllInbounds: true` (panel default) is used for new users. * `USER_INBOUND_UUIDS`: (Optional) Comma-separated list of inbound UUIDs from your panel to assign to users. If empty, `activateAllInbounds: true` (panel default) is used for new users.
* `USER_TRAFFIC_LIMIT_GB` and `USER_TRAFFIC_STRATEGY`: Default traffic limit in gigabytes (0 for unlimited) and the reset strategy applied when updating users on the panel.
* `TRIAL_ENABLED`, `TRIAL_DURATION_DAYS`, `TRIAL_TRAFFIC_LIMIT_GB`: Settings for the trial period. * `TRIAL_ENABLED`, `TRIAL_DURATION_DAYS`, `TRIAL_TRAFFIC_LIMIT_GB`: Settings for the trial period.
* `WEB_SERVER_HOST`, `WEB_SERVER_PORT`: Host and port for the bot's internal webhook server. * `WEB_SERVER_HOST`, `WEB_SERVER_PORT`: Host and port for the bot's internal webhook server.
* `LOGS_PAGE_SIZE`: For admin panel log pagination. * `LOGS_PAGE_SIZE`: For admin panel log pagination.
+5
View File
@@ -24,6 +24,7 @@ router = Router(name="admin_common_router")
@router.message(Command("admin")) @router.message(Command("admin"))
async def admin_panel_command_handler( async def admin_panel_command_handler(
message: types.Message, message: types.Message,
state: FSMContext,
settings: Settings, settings: Settings,
i18n_data: dict, i18n_data: dict,
): ):
@@ -34,6 +35,7 @@ async def admin_panel_command_handler(
await message.answer("Language service error.") await message.answer("Language service error.")
return return
await state.clear()
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
await message.answer(_(key="admin_panel_title"), await message.answer(_(key="admin_panel_title"),
reply_markup=get_admin_panel_keyboard( reply_markup=get_admin_panel_keyboard(
@@ -73,6 +75,9 @@ async def admin_panel_actions_callback_handler(
elif action == "create_promo": elif action == "create_promo":
await admin_promo_handlers.create_promo_prompt_handler( await admin_promo_handlers.create_promo_prompt_handler(
callback, state, i18n_data, settings, session) callback, state, i18n_data, settings, session)
elif action == "manage_promos":
await admin_promo_handlers.manage_promo_codes_handler(
callback, i18n_data, settings, session)
elif action == "view_promos": elif action == "view_promos":
await admin_promo_handlers.view_promo_codes_handler( await admin_promo_handlers.view_promo_codes_handler(
callback, i18n_data, settings, session) callback, i18n_data, settings, session)
+148 -2
View File
@@ -1,5 +1,6 @@
import logging import logging
from aiogram import Router, F, types, Bot from aiogram import Router, F, types, Bot
from aiogram.filters import StateFilter
from aiogram.fsm.context import FSMContext from aiogram.fsm.context import FSMContext
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Optional from typing import Optional
@@ -11,6 +12,7 @@ from db.dal import promo_code_dal
from bot.states.admin_states import AdminStates from bot.states.admin_states import AdminStates
from bot.keyboards.inline.admin_keyboards import get_back_to_admin_panel_keyboard, get_admin_panel_keyboard from bot.keyboards.inline.admin_keyboards import get_back_to_admin_panel_keyboard, get_admin_panel_keyboard
from aiogram.utils.keyboard import InlineKeyboardBuilder, InlineKeyboardButton
from bot.middlewares.i18n import JsonI18n from bot.middlewares.i18n import JsonI18n
router = Router(name="admin_promo_codes_router") router = Router(name="admin_promo_codes_router")
@@ -194,8 +196,152 @@ async def view_promo_codes_handler(callback: types.CallbackQuery,
await callback.answer() await callback.answer()
@router.callback_query(F.data == "admin_action:main", @router.callback_query(F.data == "admin_action:manage_promos")
AdminStates.waiting_for_promo_details) async def manage_promo_codes_handler(callback: types.CallbackQuery,
i18n_data: dict, settings: Settings,
session: AsyncSession):
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not callback.message:
await callback.answer("Error displaying promo codes.", show_alert=True)
return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
promo_models = await promo_code_dal.get_all_active_promo_codes(session,
limit=20,
offset=0)
if not promo_models:
await callback.message.edit_text(
_("admin_no_active_promos"),
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
await callback.answer()
return
kb = InlineKeyboardBuilder()
for promo in promo_models:
kb.row(
InlineKeyboardButton(
text=promo.code,
callback_data=f"promo_edit:{promo.promo_code_id}"),
InlineKeyboardButton(
text=_("admin_promo_delete_button"),
callback_data=f"promo_delete:{promo.promo_code_id}"),
)
kb.row(
InlineKeyboardButton(text=_("back_to_admin_panel_button"),
callback_data="admin_action:main"))
await callback.message.edit_text(
_("admin_manage_promos_title"),
reply_markup=kb.as_markup())
await callback.answer()
@router.callback_query(F.data.startswith("promo_edit:"))
async def promo_edit_select_handler(callback: types.CallbackQuery, state: FSMContext,
i18n_data: dict, settings: Settings,
session: AsyncSession):
promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.get_promo_code_by_id(session, promo_id)
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not promo or not callback.message:
await callback.answer("Error", show_alert=True)
return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
await state.update_data(edit_promo_id=promo_id)
await state.set_state(AdminStates.waiting_for_promo_edit_details)
await callback.message.edit_text(
_("admin_promo_edit_prompt", code=promo.code),
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
parse_mode="HTML")
await callback.answer()
@router.message(AdminStates.waiting_for_promo_edit_details, F.text)
async def process_promo_edit_details(message: types.Message, state: FSMContext,
i18n_data: dict, settings: Settings,
session: AsyncSession):
data = await state.get_data()
promo_id = data.get("edit_promo_id")
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not promo_id:
await message.answer("Error")
await state.clear()
return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
parts = message.text.strip().split()
if not (3 <= len(parts) <= 4):
await message.answer(_("admin_promo_invalid_format"))
return
try:
code_str = parts[0].upper()
bonus = int(parts[1])
max_act = int(parts[2])
valid_until = None
if len(parts) == 4:
days = int(parts[3])
valid_until = datetime.now(timezone.utc) + timedelta(days=days)
except Exception:
await message.answer(_("admin_promo_invalid_format_general"))
return
update_data = {
"code": code_str,
"bonus_days": bonus,
"max_activations": max_act,
"valid_until": valid_until,
}
updated = await promo_code_dal.update_promo_code(session, promo_id, update_data)
if updated:
await session.commit()
await message.answer(
_("admin_promo_updated_success", code=code_str),
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
parse_mode="HTML",
)
else:
await session.rollback()
await message.answer(_("admin_promo_not_found"))
await state.clear()
@router.callback_query(F.data.startswith("promo_delete:"))
async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict,
settings: Settings, session: AsyncSession):
promo_id = int(callback.data.split(":")[1])
promo = await promo_code_dal.delete_promo_code(session, promo_id)
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n or not callback.message:
await callback.answer("Error", show_alert=True)
return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
if promo:
await session.commit()
await callback.message.edit_text(
_("admin_promo_deleted_success", code=promo.code),
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n),
parse_mode="HTML")
else:
await session.rollback()
await callback.message.edit_text(
_("admin_promo_not_found"),
reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n))
await callback.answer()
@router.callback_query(
F.data == "admin_action:main",
StateFilter(
AdminStates.waiting_for_promo_details,
AdminStates.waiting_for_promo_edit_details,
),
)
async def cancel_promo_creation_state_to_menu(callback: types.CallbackQuery, async def cancel_promo_creation_state_to_menu(callback: types.CallbackQuery,
state: FSMContext, state: FSMContext,
settings: Settings, settings: Settings,
+142 -102
View File
@@ -16,11 +16,14 @@ router = Router(name="admin_sync_router")
@router.message(Command("sync")) @router.message(Command("sync"))
async def sync_command_handler(message_event: Union[types.Message, async def sync_command_handler(
types.CallbackQuery], message_event: Union[types.Message, types.CallbackQuery],
bot: Bot, settings: Settings, i18n_data: dict, bot: Bot,
panel_service: PanelApiService, settings: Settings,
session: AsyncSession): i18n_data: dict,
panel_service: PanelApiService,
session: AsyncSession,
):
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n: if not i18n:
@@ -33,14 +36,15 @@ async def sync_command_handler(message_event: Union[types.Message,
return return
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
target_chat_id = message_event.chat.id if isinstance( target_chat_id = (
message_event, types.Message) else ( message_event.chat.id
message_event.message.chat.id if message_event.message else None) if isinstance(message_event, types.Message)
else (message_event.message.chat.id if message_event.message else None)
)
if not target_chat_id: if not target_chat_id:
logging.error("Sync handler: could not determine target_chat_id.") logging.error("Sync handler: could not determine target_chat_id.")
if isinstance(message_event, types.CallbackQuery): if isinstance(message_event, types.CallbackQuery):
await message_event.answer("Error initiating sync.", await message_event.answer("Error initiating sync.", show_alert=True)
show_alert=True)
return return
if isinstance(message_event, types.Message): if isinstance(message_event, types.Message):
@@ -59,21 +63,21 @@ async def sync_command_handler(message_event: Union[types.Message,
if panel_users_data is None: if panel_users_data is None:
error_msg = "Failed to fetch users from panel or panel API issue." error_msg = "Failed to fetch users from panel or panel API issue."
sync_errors.append(error_msg) sync_errors.append(error_msg)
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(session, "failed", error_msg)
session, "failed", error_msg)
await session.commit() await session.commit()
await bot.send_message(target_chat_id, await bot.send_message(target_chat_id, _("sync_failed", details=error_msg))
_("sync_failed", details=error_msg))
return return
if not panel_users_data: if not panel_users_data:
status_msg = "No users found in the panel to sync." status_msg = "No users found in the panel to sync."
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(
session, "success", status_msg, 0, 0) session, "success", status_msg, 0, 0
)
await session.commit() await session.commit()
await bot.send_message( await bot.send_message(
target_chat_id, target_chat_id,
_("sync_completed", status="Success", details=status_msg)) _("sync_completed", status="Success", details=status_msg),
)
return return
total_panel_users = len(panel_users_data) total_panel_users = len(panel_users_data)
@@ -81,9 +85,9 @@ async def sync_command_handler(message_event: Union[types.Message,
for panel_user_dict in panel_users_data: for panel_user_dict in panel_users_data:
users_processed_count += 1 users_processed_count += 1
panel_uuid = panel_user_dict.get('uuid') panel_uuid = panel_user_dict.get("uuid")
telegram_id_from_panel_str = panel_user_dict.get('telegramId') telegram_id_from_panel_str = panel_user_dict.get("telegramId")
panel_username = panel_user_dict.get('username') panel_username = panel_user_dict.get("username")
if not panel_uuid: if not panel_uuid:
logging.warning( logging.warning(
@@ -111,26 +115,22 @@ async def sync_command_handler(message_event: Union[types.Message,
continue continue
bot_user = await user_dal.get_user_by_id(session, bot_user = await user_dal.get_user_by_id(session, telegram_id_from_panel)
telegram_id_from_panel)
if not bot_user: if not bot_user:
user_data_to_create = { user_data_to_create = {
"user_id": "user_id": telegram_id_from_panel,
telegram_id_from_panel, "username": panel_username,
"username": "panel_user_uuid": panel_uuid,
panel_username, "language_code": settings.DEFAULT_LANGUAGE,
"panel_user_uuid": "registration_date": (
panel_uuid, datetime.fromisoformat(
"language_code": panel_user_dict["createdAt"].replace("Z", "+00:00")
settings.DEFAULT_LANGUAGE, )
"registration_date": if panel_user_dict.get("createdAt")
datetime.fromisoformat( else datetime.now(timezone.utc)
panel_user_dict['createdAt'].replace("Z", "+00:00")) ),
if panel_user_dict.get('createdAt') else datetime.now(
timezone.utc)
} }
bot_user = await user_dal.create_user(session, bot_user = await user_dal.create_user(session, user_data_to_create)
user_data_to_create)
logging.info( logging.info(
f"Sync: Created new local user {telegram_id_from_panel} from panel data {panel_uuid}." f"Sync: Created new local user {telegram_id_from_panel} from panel data {panel_uuid}."
) )
@@ -142,8 +142,12 @@ async def sync_command_handler(message_event: Union[types.Message,
) )
conflicting_user = await user_dal.get_user_by_panel_uuid( conflicting_user = await user_dal.get_user_by_panel_uuid(
session, panel_uuid) session, panel_uuid
if conflicting_user and conflicting_user.user_id != telegram_id_from_panel: )
if (
conflicting_user
and conflicting_user.user_id != telegram_id_from_panel
):
sync_errors.append( sync_errors.append(
f"Panel UUID {panel_uuid} for TG {telegram_id_from_panel} already linked to another TG user {conflicting_user.user_id}." f"Panel UUID {panel_uuid} for TG {telegram_id_from_panel} already linked to another TG user {conflicting_user.user_id}."
) )
@@ -151,60 +155,64 @@ async def sync_command_handler(message_event: Union[types.Message,
continue continue
await user_dal.update_user( await user_dal.update_user(
session, telegram_id_from_panel, { session,
"panel_user_uuid": panel_uuid, telegram_id_from_panel,
"username": panel_username {"panel_user_uuid": panel_uuid, "username": panel_username},
}) )
logging.info( logging.info(
f"Sync: Updated panel_uuid for local user {telegram_id_from_panel} to {panel_uuid}." f"Sync: Updated panel_uuid for local user {telegram_id_from_panel} to {panel_uuid}."
) )
panel_sub_link_id = panel_user_dict.get( panel_sub_link_id = panel_user_dict.get(
'subscriptionUuid') or panel_user_dict.get('shortUuid') "subscriptionUuid"
) or panel_user_dict.get("shortUuid")
if panel_sub_link_id: if panel_sub_link_id:
end_date_str = panel_user_dict.get('expireAt') end_date_str = panel_user_dict.get("expireAt")
start_date_str = panel_user_dict.get('createdAt') start_date_str = panel_user_dict.get("createdAt")
if end_date_str: if end_date_str:
try: try:
end_date_obj = datetime.fromisoformat( end_date_obj = datetime.fromisoformat(
end_date_str.replace("Z", "+00:00")) end_date_str.replace("Z", "+00:00")
start_date_obj = datetime.fromisoformat( )
start_date_str.replace( start_date_obj = (
"Z", datetime.fromisoformat(
"+00:00")) if start_date_str else datetime.now( start_date_str.replace("Z", "+00:00")
timezone.utc) )
if start_date_str
else datetime.now(timezone.utc)
)
status_from_panel = panel_user_dict.get( status_from_panel = panel_user_dict.get(
'status', 'UNKNOWN').upper() "status", "UNKNOWN"
is_active_flag = 1 if status_from_panel == 'ACTIVE' and end_date_obj > datetime.now( ).upper()
timezone.utc) else 0 is_active_flag = (
1
if status_from_panel == "ACTIVE"
and end_date_obj > datetime.now(timezone.utc)
else 0
)
sub_payload = { sub_payload = {
"user_id": "user_id": telegram_id_from_panel,
telegram_id_from_panel, "panel_user_uuid": panel_uuid,
"panel_user_uuid": "panel_subscription_uuid": panel_sub_link_id,
panel_uuid, "start_date": start_date_obj,
"panel_subscription_uuid": "end_date": end_date_obj,
panel_sub_link_id, "is_active": is_active_flag,
"start_date": "status_from_panel": status_from_panel,
start_date_obj, "traffic_limit_bytes": panel_user_dict.get(
"end_date": "trafficLimitBytes"
end_date_obj, ),
"is_active": "traffic_used_bytes": panel_user_dict.get(
is_active_flag, "usedTrafficBytes"
"status_from_panel": ),
status_from_panel,
"traffic_limit_bytes":
panel_user_dict.get('trafficLimitBytes'),
"traffic_used_bytes":
panel_user_dict.get('usedTrafficBytes')
} }
await subscription_dal.deactivate_other_active_subscriptions( await subscription_dal.deactivate_other_active_subscriptions(
session, panel_uuid, panel_sub_link_id) session, panel_uuid, panel_sub_link_id
await subscription_dal.upsert_subscription( )
session, sub_payload) await subscription_dal.upsert_subscription(session, sub_payload)
subscriptions_synced_count += 1 subscriptions_synced_count += 1
users_synced_successfully += 1 users_synced_successfully += 1
except ValueError as e_date: except ValueError as e_date:
@@ -217,7 +225,8 @@ async def sync_command_handler(message_event: Union[types.Message,
except Exception as e_sub_sync: except Exception as e_sub_sync:
logging.error( logging.error(
f"Sync: Error syncing subscription for panel user {panel_uuid} (TG ID: {telegram_id_from_panel}): {e_sub_sync}", f"Sync: Error syncing subscription for panel user {panel_uuid} (TG ID: {telegram_id_from_panel}): {e_sub_sync}",
exc_info=True) exc_info=True,
)
sync_errors.append( sync_errors.append(
f"Sub sync error for panel user {panel_uuid} (TG ID: {telegram_id_from_panel})." f"Sub sync error for panel user {panel_uuid} (TG ID: {telegram_id_from_panel})."
) )
@@ -228,7 +237,8 @@ async def sync_command_handler(message_event: Union[types.Message,
else: else:
await subscription_dal.deactivate_other_active_subscriptions( await subscription_dal.deactivate_other_active_subscriptions(
session, panel_uuid, None) session, panel_uuid, None
)
logging.info( logging.info(
f"Sync: Panel user {panel_uuid} (TG ID: {telegram_id_from_panel}) has no subscription link on panel. Deactivated local subs if any." f"Sync: Panel user {panel_uuid} (TG ID: {telegram_id_from_panel}) has no subscription link on panel. Deactivated local subs if any."
) )
@@ -239,6 +249,17 @@ async def sync_command_handler(message_event: Union[types.Message,
f"Sync progress: {users_processed_count}/{total_panel_users} users processed from panel." f"Sync progress: {users_processed_count}/{total_panel_users} users processed from panel."
) )
panel_uuid_set = {u.get("uuid") for u in panel_users_data if u.get("uuid")}
local_users_with_uuid = await user_dal.get_all_users_with_panel_uuid(session)
for local_user in local_users_with_uuid:
if local_user.panel_user_uuid not in panel_uuid_set:
await subscription_dal.deactivate_other_active_subscriptions(
session, local_user.panel_user_uuid, None
)
logging.info(
f"Sync: Local user {local_user.user_id} with panel UUID {local_user.panel_user_uuid} not found on panel. Deactivated local subs."
)
status_msg_key = "sync_completed_details" status_msg_key = "sync_completed_details"
final_status_type = "success" final_status_type = "success"
@@ -251,36 +272,51 @@ async def sync_command_handler(message_event: Union[types.Message,
details_for_db = f"Successfully processed {users_processed_count} users. Synced {subscriptions_synced_count} subscriptions." details_for_db = f"Successfully processed {users_processed_count} users. Synced {subscriptions_synced_count} subscriptions."
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(
session, final_status_type, details_for_db, users_processed_count, session,
subscriptions_synced_count) final_status_type,
details_for_db,
users_processed_count,
subscriptions_synced_count,
)
await session.commit() await session.commit()
final_user_message = _(status_msg_key, final_user_message = _(
total_checked=total_panel_users, status_msg_key,
users_synced=users_synced_successfully, total_checked=total_panel_users,
subs_synced=subscriptions_synced_count, users_synced=users_synced_successfully,
errors_count=len(sync_errors), subs_synced=subscriptions_synced_count,
error_details_preview=error_preview errors_count=len(sync_errors),
if sync_errors else _("no_errors_placeholder")) error_details_preview=(
error_preview if sync_errors else _("no_errors_placeholder")
),
)
await bot.send_message(target_chat_id, final_user_message) await bot.send_message(target_chat_id, final_user_message)
except Exception as e_sync_global: except Exception as e_sync_global:
await session.rollback() await session.rollback()
logging.error(f"Global error during /sync command: {e_sync_global}", logging.error(
exc_info=True) f"Global error during /sync command: {e_sync_global}", exc_info=True
error_detail_for_db = f"An unexpected error occurred during sync: {str(e_sync_global)[:200]}" )
error_detail_for_db = (
f"An unexpected error occurred during sync: {str(e_sync_global)[:200]}"
)
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(
session, "failed", error_detail_for_db, users_processed_count, session,
subscriptions_synced_count) "failed",
error_detail_for_db,
users_processed_count,
subscriptions_synced_count,
)
await bot.send_message(target_chat_id, await bot.send_message(
_("sync_failed", details=error_detail_for_db)) target_chat_id, _("sync_failed", details=error_detail_for_db)
)
@router.message(Command("syncstatus")) @router.message(Command("syncstatus"))
async def sync_status_command_handler(message: types.Message, i18n_data: dict, async def sync_status_command_handler(
settings: Settings, message: types.Message, i18n_data: dict, settings: Settings, session: AsyncSession
session: AsyncSession): ):
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
if not i18n: if not i18n:
@@ -292,13 +328,16 @@ async def sync_status_command_handler(message: types.Message, i18n_data: dict,
response_text = "" response_text = ""
if status_record_model: if status_record_model:
last_time_val = status_record_model.last_sync_time last_time_val = status_record_model.last_sync_time
last_time_str = last_time_val.strftime( last_time_str = (
'%Y-%m-%d %H:%M:%S UTC') if last_time_val else "N/A" last_time_val.strftime("%Y-%m-%d %H:%M:%S UTC") if last_time_val else "N/A"
)
details_val = status_record_model.details details_val = status_record_model.details
details_str = (details_val[:200] + details_str = (
"...") if details_val and len(details_val) > 200 else ( (details_val[:200] + "...")
details_val or "N/A") if details_val and len(details_val) > 200
else (details_val or "N/A")
)
response_text = ( response_text = (
f"<b>{_('admin_stats_last_sync_header')}</b>\n" f"<b>{_('admin_stats_last_sync_header')}</b>\n"
@@ -306,7 +345,8 @@ async def sync_status_command_handler(message: types.Message, i18n_data: dict,
f" {_('admin_stats_sync_status')}: {status_record_model.status}\n" f" {_('admin_stats_sync_status')}: {status_record_model.status}\n"
f" {_('admin_stats_sync_users_processed')}: {status_record_model.users_processed_from_panel}\n" f" {_('admin_stats_sync_users_processed')}: {status_record_model.users_processed_from_panel}\n"
f" {_('admin_stats_sync_subs_synced')}: {status_record_model.subscriptions_synced}\n" f" {_('admin_stats_sync_subs_synced')}: {status_record_model.subscriptions_synced}\n"
f" {_('admin_stats_sync_details_label')}: {details_str}") f" {_('admin_stats_sync_details_label')}: {details_str}"
)
else: else:
response_text = _("admin_sync_status_never_run") response_text = _("admin_sync_status_never_run")
+7 -11
View File
@@ -28,20 +28,16 @@ USERNAME_REGEX = re.compile(r"^[a-zA-Z0-9_]{5,32}$")
async def _get_user_model_by_input(session: AsyncSession, async def _get_user_model_by_input(session: AsyncSession,
input_text: str) -> Optional[User]: input_text: str) -> Optional[User]:
user_model: Optional[User] = None
if input_text.isdigit(): if input_text.isdigit():
try: try:
user_model = await user_dal.get_user_by_id(session, return await user_dal.get_user(session, user_id=int(input_text))
int(input_text))
except ValueError: except ValueError:
pass return None
elif input_text.startswith("@") and USERNAME_REGEX.match(input_text[1:]): if input_text.startswith("@") and USERNAME_REGEX.match(input_text[1:]):
user_model = await user_dal.get_user_by_username( return await user_dal.get_user(session, username=input_text[1:])
session, input_text[1:]) if USERNAME_REGEX.match(input_text):
elif USERNAME_REGEX.match(input_text): return await user_dal.get_user(session, username=input_text)
user_model = await user_dal.get_user_by_username(session, input_text) return None
return user_model
async def ban_user_prompt_handler(callback: types.CallbackQuery, async def ban_user_prompt_handler(callback: types.CallbackQuery,
+10
View File
@@ -20,6 +20,7 @@ from bot.services.panel_api_service import PanelApiService
from bot.services.yookassa_service import YooKassaService from bot.services.yookassa_service import YooKassaService
from bot.middlewares.i18n import JsonI18n from bot.middlewares.i18n import JsonI18n
from config.settings import Settings from config.settings import Settings
from bot.services.notification_service import notify_admin_new_payment
payment_processing_lock = asyncio.Lock() payment_processing_lock = asyncio.Lock()
@@ -175,6 +176,15 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
f"Failed to send final payment success message to user {user_id}: {e_notify}" f"Failed to send final payment success message to user {user_id}: {e_notify}"
) )
await notify_admin_new_payment(
bot,
settings,
i18n,
user_id,
subscription_months,
payment_value,
)
except Exception as e_process: except Exception as e_process:
logging.error( logging.error(
f"Error during process_successful_payment main try block for user {user_id}: {e_process}", f"Error during process_successful_payment main try block for user {user_id}: {e_process}",
+114 -77
View File
@@ -7,7 +7,11 @@ from datetime import datetime
from config.settings import Settings from config.settings import Settings
from bot.services.subscription_service import SubscriptionService from bot.services.subscription_service import SubscriptionService
from bot.services.panel_api_service import PanelApiService from bot.services.panel_api_service import PanelApiService
from bot.keyboards.inline.user_keyboards import get_trial_confirmation_keyboard, get_main_menu_inline_keyboard from bot.services.notification_service import notify_admin_new_trial
from bot.keyboards.inline.user_keyboards import (
get_trial_confirmation_keyboard,
get_main_menu_inline_keyboard,
)
from bot.middlewares.i18n import JsonI18n from bot.middlewares.i18n import JsonI18n
from .start import send_main_menu from .start import send_main_menu
@@ -15,13 +19,16 @@ router = Router(name="user_trial_router")
async def request_trial_confirmation_handler( async def request_trial_confirmation_handler(
callback: types.CallbackQuery, settings: Settings, i18n_data: dict, callback: types.CallbackQuery,
subscription_service: SubscriptionService, session: AsyncSession): settings: Settings,
i18n_data: dict,
subscription_service: SubscriptionService,
session: AsyncSession,
):
user_id = callback.from_user.id user_id = callback.from_user.id
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if i18n else key
) if i18n else key
if not i18n or not callback.message: if not i18n or not callback.message:
await callback.answer(_("error_occurred_try_again"), show_alert=True) await callback.answer(_("error_occurred_try_again"), show_alert=True)
@@ -29,50 +36,60 @@ async def request_trial_confirmation_handler(
show_trial_btn_in_menu_if_fail = False show_trial_btn_in_menu_if_fail = False
if settings.TRIAL_ENABLED: if settings.TRIAL_ENABLED:
if not await subscription_service.has_had_any_subscription( if not await subscription_service.has_had_any_subscription(session, user_id):
session, user_id):
show_trial_btn_in_menu_if_fail = True show_trial_btn_in_menu_if_fail = True
if not settings.TRIAL_ENABLED: if not settings.TRIAL_ENABLED:
await callback.message.edit_text( await callback.message.edit_text(
_("trial_feature_disabled"), _("trial_feature_disabled"),
reply_markup=get_main_menu_inline_keyboard(current_lang, i18n, reply_markup=get_main_menu_inline_keyboard(
settings, False)) current_lang, i18n, settings, False
),
)
await callback.answer() await callback.answer()
return return
if await subscription_service.has_had_any_subscription(session, user_id): if await subscription_service.has_had_any_subscription(session, user_id):
await callback.message.edit_text( await callback.message.edit_text(
_("trial_already_had_subscription_or_trial"), _("trial_already_had_subscription_or_trial"),
reply_markup=get_main_menu_inline_keyboard(current_lang, i18n, reply_markup=get_main_menu_inline_keyboard(
settings, False)) current_lang, i18n, settings, False
),
)
await callback.answer() await callback.answer()
return return
traffic_gb_display = str( traffic_gb_display = (
settings.TRIAL_TRAFFIC_LIMIT_GB str(settings.TRIAL_TRAFFIC_LIMIT_GB)
) if settings.TRIAL_TRAFFIC_LIMIT_GB and settings.TRIAL_TRAFFIC_LIMIT_GB > 0 else _( if settings.TRIAL_TRAFFIC_LIMIT_GB and settings.TRIAL_TRAFFIC_LIMIT_GB > 0
"traffic_unlimited") else _("traffic_unlimited")
)
await callback.message.edit_text( await callback.message.edit_text(
text=_("trial_confirm_prompt", text=_(
days=settings.TRIAL_DURATION_DAYS, "trial_confirm_prompt",
traffic_gb=traffic_gb_display), days=settings.TRIAL_DURATION_DAYS,
reply_markup=get_trial_confirmation_keyboard(current_lang, i18n)) traffic_gb=traffic_gb_display,
),
reply_markup=get_trial_confirmation_keyboard(current_lang, i18n),
)
await callback.answer() await callback.answer()
@router.callback_query(F.data == "trial_action:confirm_activate") @router.callback_query(F.data == "trial_action:confirm_activate")
async def confirm_activate_trial_handler( async def confirm_activate_trial_handler(
callback: types.CallbackQuery, settings: Settings, i18n_data: dict, callback: types.CallbackQuery,
subscription_service: SubscriptionService, settings: Settings,
panel_service: PanelApiService, session: AsyncSession): i18n_data: dict,
subscription_service: SubscriptionService,
panel_service: PanelApiService,
session: AsyncSession,
):
user_id = callback.from_user.id user_id = callback.from_user.id
current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
_ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if i18n else key
) if i18n else key
if not i18n or not callback.message: if not i18n or not callback.message:
await callback.answer(_("error_occurred_try_again"), show_alert=True) await callback.answer(_("error_occurred_try_again"), show_alert=True)
@@ -81,26 +98,22 @@ async def confirm_activate_trial_handler(
if not settings.TRIAL_ENABLED: if not settings.TRIAL_ENABLED:
await callback.answer(_("trial_feature_disabled"), show_alert=True) await callback.answer(_("trial_feature_disabled"), show_alert=True)
await send_main_menu(callback, await send_main_menu(
settings, callback, settings, i18n_data, subscription_service, session, is_edit=True
i18n_data, )
subscription_service,
session,
is_edit=True)
return return
if await subscription_service.has_had_any_subscription(session, user_id): if await subscription_service.has_had_any_subscription(session, user_id):
await callback.answer(_("trial_already_had_subscription_or_trial"), await callback.answer(
show_alert=True) _("trial_already_had_subscription_or_trial"), show_alert=True
await send_main_menu(callback, )
settings, await send_main_menu(
i18n_data, callback, settings, i18n_data, subscription_service, session, is_edit=True
subscription_service, )
session,
is_edit=True)
return return
activation_result = await subscription_service.activate_trial_subscription( activation_result = await subscription_service.activate_trial_subscription(
session, user_id) session, user_id
)
final_message_text_in_chat = "" final_message_text_in_chat = ""
show_trial_button_after_action = False show_trial_button_after_action = False
@@ -110,35 +123,43 @@ async def confirm_activate_trial_handler(
end_date_obj = activation_result.get("end_date") end_date_obj = activation_result.get("end_date")
config_link_for_trial = activation_result.get("subscription_url") or _( config_link_for_trial = activation_result.get("subscription_url") or _(
"config_link_not_available") "config_link_not_available"
)
if config_link_for_trial == _( traffic_gb_val = activation_result.get(
"config_link_not_available") and activation_result.get( "traffic_gb", settings.TRIAL_TRAFFIC_LIMIT_GB
"panel_short_uuid"): )
generated_link = await panel_service.get_subscription_link( traffic_display = (
activation_result["panel_short_uuid"]) f"{traffic_gb_val} GB"
if generated_link: config_link_for_trial = generated_link if traffic_gb_val and traffic_gb_val > 0
else _("traffic_unlimited")
traffic_gb_val = activation_result.get("traffic_gb", )
settings.TRIAL_TRAFFIC_LIMIT_GB)
traffic_display = f"{traffic_gb_val} GB" if traffic_gb_val and traffic_gb_val > 0 else _(
"traffic_unlimited")
final_message_text_in_chat = _( final_message_text_in_chat = _(
"trial_activated_details_message", "trial_activated_details_message",
days=activation_result.get("days", settings.TRIAL_DURATION_DAYS), days=activation_result.get("days", settings.TRIAL_DURATION_DAYS),
end_date=end_date_obj.strftime('%Y-%m-%d') if isinstance( end_date=(
end_date_obj, datetime) else "N/A", end_date_obj.strftime("%Y-%m-%d")
if isinstance(end_date_obj, datetime)
else "N/A"
),
config_link=config_link_for_trial, config_link=config_link_for_trial,
traffic_gb=traffic_display) traffic_gb=traffic_display,
)
else: else:
message_key_from_service = activation_result.get( message_key_from_service = (
"message_key", "trial_activation_failed" activation_result.get("message_key", "trial_activation_failed")
) if activation_result else "trial_activation_failed" if activation_result
else "trial_activation_failed"
)
final_message_text_in_chat = _(message_key_from_service) final_message_text_in_chat = _(message_key_from_service)
await callback.answer(final_message_text_in_chat, show_alert=True) await callback.answer(final_message_text_in_chat, show_alert=True)
if settings.TRIAL_ENABLED and not await subscription_service.has_had_any_subscription( if (
session, user_id): settings.TRIAL_ENABLED
and not await subscription_service.has_had_any_subscription(
session, user_id
)
):
show_trial_button_after_action = True show_trial_button_after_action = True
try: try:
@@ -146,31 +167,47 @@ async def confirm_activate_trial_handler(
final_message_text_in_chat, final_message_text_in_chat,
parse_mode="HTML", parse_mode="HTML",
reply_markup=get_main_menu_inline_keyboard( reply_markup=get_main_menu_inline_keyboard(
current_lang, i18n, settings, show_trial_button_after_action), current_lang, i18n, settings, show_trial_button_after_action
disable_web_page_preview=True) ),
disable_web_page_preview=True,
)
except Exception as e_edit: except Exception as e_edit:
logging.warning( logging.warning(
f"Could not edit trial result message: {e_edit}. Sending new one.") f"Could not edit trial result message: {e_edit}. Sending new one."
)
if callback.message and hasattr(callback.message, if (
'chat') and callback.message.chat: callback.message
and hasattr(callback.message, "chat")
and callback.message.chat
):
await callback.message.chat.send_message( await callback.message.chat.send_message(
final_message_text_in_chat, final_message_text_in_chat,
parse_mode="HTML", parse_mode="HTML",
reply_markup=get_main_menu_inline_keyboard( reply_markup=get_main_menu_inline_keyboard(
current_lang, i18n, settings, current_lang, i18n, settings, show_trial_button_after_action
show_trial_button_after_action), ),
disable_web_page_preview=True) disable_web_page_preview=True,
)
if activation_result and activation_result.get("activated") and end_date_obj:
await notify_admin_new_trial(
callback.bot,
settings,
i18n,
user_id,
end_date_obj,
)
@router.callback_query(F.data == "main_action:cancel_trial") @router.callback_query(F.data == "main_action:cancel_trial")
async def cancel_trial_activation(callback: types.CallbackQuery, async def cancel_trial_activation(
settings: Settings, i18n_data: dict, callback: types.CallbackQuery,
subscription_service: SubscriptionService, settings: Settings,
session: AsyncSession): i18n_data: dict,
await send_main_menu(callback, subscription_service: SubscriptionService,
settings, session: AsyncSession,
i18n_data, ):
subscription_service, await send_main_menu(
session, callback, settings, i18n_data, subscription_service, session, is_edit=True
is_edit=True) )
+2
View File
@@ -18,6 +18,8 @@ def get_admin_panel_keyboard(i18n_instance, lang: str,
callback_data="admin_action:broadcast") callback_data="admin_action:broadcast")
builder.button(text=_(key="admin_create_promo_button"), builder.button(text=_(key="admin_create_promo_button"),
callback_data="admin_action:create_promo") callback_data="admin_action:create_promo")
builder.button(text=_(key="admin_manage_promos_button"),
callback_data="admin_action:manage_promos")
builder.button(text=_(key="admin_view_promos_button"), builder.button(text=_(key="admin_view_promos_button"),
callback_data="admin_action:view_promos") callback_data="admin_action:view_promos")
builder.button(text=_(key="admin_ban_user_button"), builder.button(text=_(key="admin_ban_user_button"),
+15 -4
View File
@@ -1,5 +1,5 @@
from aiogram.utils.keyboard import InlineKeyboardBuilder, InlineKeyboardButton from aiogram.utils.keyboard import InlineKeyboardBuilder, InlineKeyboardButton
from aiogram.types import InlineKeyboardMarkup from aiogram.types import InlineKeyboardMarkup, WebAppInfo
from typing import Dict, Optional, List from typing import Dict, Optional, List
from config.settings import Settings from config.settings import Settings
@@ -21,9 +21,20 @@ def get_main_menu_inline_keyboard(
builder.row( builder.row(
InlineKeyboardButton(text=_(key="menu_subscribe_inline"), InlineKeyboardButton(text=_(key="menu_subscribe_inline"),
callback_data="main_action:subscribe")) callback_data="main_action:subscribe"))
builder.row( if settings.SUBSCRIPTION_MINI_APP_URL:
InlineKeyboardButton(text=_(key="menu_my_subscription_inline"), builder.row(
callback_data="main_action:my_subscription")) InlineKeyboardButton(
text=_(key="menu_my_subscription_inline"),
web_app=WebAppInfo(url=settings.SUBSCRIPTION_MINI_APP_URL),
)
)
else:
builder.row(
InlineKeyboardButton(
text=_(key="menu_my_subscription_inline"),
callback_data="main_action:my_subscription",
)
)
referral_button = InlineKeyboardButton( referral_button = InlineKeyboardButton(
text=_(key="menu_referral_inline"), text=_(key="menu_referral_inline"),
+148 -85
View File
@@ -3,7 +3,13 @@ import asyncio
from typing import Callable, Dict, Any, Awaitable, Optional from typing import Callable, Dict, Any, Awaitable, Optional
from aiogram import Bot, Dispatcher, BaseMiddleware, Router, F from aiogram import Bot, Dispatcher, BaseMiddleware, Router, F
from aiogram.types import Update from aiogram.types import (
Update,
MenuButtonDefault,
MenuButtonWebApp,
WebAppInfo,
BotCommand,
)
from aiogram.enums import ParseMode from aiogram.enums import ParseMode
from aiogram.filters import CommandStart, Command from aiogram.filters import CommandStart, Command
from aiogram.client.default import DefaultBotProperties from aiogram.client.default import DefaultBotProperties
@@ -43,14 +49,17 @@ class DBSessionMiddleware(BaseMiddleware):
super().__init__() super().__init__()
self.async_session_factory = async_session_factory self.async_session_factory = async_session_factory
async def __call__(self, handler: Callable[[Update, Dict[str, Any]], async def __call__(
Awaitable[Any]], event: Update, self,
data: Dict[str, Any]) -> Any: handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]],
event: Update,
data: Dict[str, Any],
) -> Any:
if self.async_session_factory is None: if self.async_session_factory is None:
logging.critical( logging.critical("DBSessionMiddleware: async_session_factory is None!")
"DBSessionMiddleware: async_session_factory is None!")
raise RuntimeError( raise RuntimeError(
"async_session_factory not provided to DBSessionMiddleware") "async_session_factory not provided to DBSessionMiddleware"
)
async with self.async_session_factory() as session: async with self.async_session_factory() as session:
data["session"] = session data["session"] = session
@@ -62,8 +71,8 @@ class DBSessionMiddleware(BaseMiddleware):
except Exception: except Exception:
await session.rollback() await session.rollback()
logging.error( logging.error(
"DBSessionMiddleware: Exception caused rollback.", "DBSessionMiddleware: Exception caused rollback.", exc_info=True
exc_info=True) )
raise raise
@@ -95,25 +104,29 @@ async def on_startup_configured(dispatcher: Dispatcher):
existing_scheduler: Optional[AsyncIOScheduler] = dispatcher.get("scheduler") existing_scheduler: Optional[AsyncIOScheduler] = dispatcher.get("scheduler")
if existing_scheduler and existing_scheduler.running: if existing_scheduler and existing_scheduler.running:
logging.warning( logging.warning("STARTUP: Scheduler already running, skipping initialization.")
"STARTUP: Scheduler already running, skipping initialization.")
else: else:
scheduler = AsyncIOScheduler(timezone="UTC") scheduler = AsyncIOScheduler(timezone="UTC")
try: try:
await schedule_subscription_notifications( await schedule_subscription_notifications(
bot, settings, i18n_instance, scheduler, panel_service, bot,
async_session_factory) settings,
i18n_instance,
scheduler,
panel_service,
async_session_factory,
)
scheduler.start() scheduler.start()
dispatcher["scheduler"] = scheduler dispatcher["scheduler"] = scheduler
logging.info("STARTUP: APScheduler started.") logging.info("STARTUP: APScheduler started.")
except Exception as e: except Exception as e:
logging.error( logging.error(f"STARTUP: Failed to start APScheduler: {e}", exc_info=True)
f"STARTUP: Failed to start APScheduler: {e}", exc_info=True)
telegram_webhook_url_to_set = getattr(settings, telegram_webhook_url_to_set = getattr(settings, "TELEGRAM_WEBHOOK_BASE_URL", None)
'TELEGRAM_WEBHOOK_BASE_URL', None)
if telegram_webhook_url_to_set: if telegram_webhook_url_to_set:
full_telegram_webhook_url = f"{str(telegram_webhook_url_to_set).rstrip('/')}/{settings.BOT_TOKEN}" full_telegram_webhook_url = (
f"{str(telegram_webhook_url_to_set).rstrip('/')}/{settings.BOT_TOKEN}"
)
logging.info( logging.info(
f"STARTUP: Attempting to set Telegram webhook to: {full_telegram_webhook_url if full_telegram_webhook_url != 'ERROR_URL_TOKEN_DETECTED' else 'HIDDEN DUE TO TOKEN'}" f"STARTUP: Attempting to set Telegram webhook to: {full_telegram_webhook_url if full_telegram_webhook_url != 'ERROR_URL_TOKEN_DETECTED' else 'HIDDEN DUE TO TOKEN'}"
@@ -129,7 +142,8 @@ async def on_startup_configured(dispatcher: Dispatcher):
set_success = await bot.set_webhook( set_success = await bot.set_webhook(
url=full_telegram_webhook_url, url=full_telegram_webhook_url,
drop_pending_updates=True, drop_pending_updates=True,
allowed_updates=dispatcher.resolve_used_update_types()) allowed_updates=dispatcher.resolve_used_update_types(),
)
if set_success: if set_success:
logging.info( logging.info(
f"STARTUP: bot.set_webhook to {full_telegram_webhook_url} returned SUCCESS (True)." f"STARTUP: bot.set_webhook to {full_telegram_webhook_url} returned SUCCESS (True)."
@@ -151,7 +165,8 @@ async def on_startup_configured(dispatcher: Dispatcher):
except Exception as e_setwebhook: except Exception as e_setwebhook:
logging.error( logging.error(
f"STARTUP: EXCEPTION during set/get Telegram webhook: {e_setwebhook}", f"STARTUP: EXCEPTION during set/get Telegram webhook: {e_setwebhook}",
exc_info=True) exc_info=True,
)
else: else:
logging.error( logging.error(
"STARTUP: Skipped setting Telegram webhook due to security or configuration error." "STARTUP: Skipped setting Telegram webhook due to security or configuration error."
@@ -162,6 +177,36 @@ async def on_startup_configured(dispatcher: Dispatcher):
) )
await bot.delete_webhook(drop_pending_updates=True) await bot.delete_webhook(drop_pending_updates=True)
if settings.SUBSCRIPTION_MINI_APP_URL:
try:
menu_text = i18n_instance.gettext(
settings.DEFAULT_LANGUAGE,
"menu_my_subscription_inline",
)
await bot.set_chat_menu_button(
menu_button=MenuButtonWebApp(
text=menu_text,
web_app=WebAppInfo(url=settings.SUBSCRIPTION_MINI_APP_URL),
)
)
await bot.set_chat_menu_button(menu_button=MenuButtonDefault())
logging.info(
"STARTUP: Mini app domain registered and default menu button restored."
)
except Exception as e:
logging.error(
f"STARTUP: Failed to register mini app domain: {e}", exc_info=True
)
if settings.START_COMMAND_DESCRIPTION:
try:
await bot.set_my_commands([
BotCommand(command="start", description=settings.START_COMMAND_DESCRIPTION)
])
logging.info("STARTUP: /start command description set.")
except Exception as e:
logging.error(f"STARTUP: Failed to set bot commands: {e}", exc_info=True)
logging.info("STARTUP: Bot on_startup_configured completed.") logging.info("STARTUP: Bot on_startup_configured completed.")
@@ -174,11 +219,12 @@ async def on_shutdown_configured(dispatcher: Dispatcher):
scheduler.shutdown(wait=False) scheduler.shutdown(wait=False)
logging.info("SHUTDOWN: APScheduler shut down.") logging.info("SHUTDOWN: APScheduler shut down.")
except Exception as e: except Exception as e:
logging.error(f"SHUTDOWN: Error shutting down APScheduler: {e}", logging.error(
exc_info=True) f"SHUTDOWN: Error shutting down APScheduler: {e}", exc_info=True
)
panel_service: Optional[PanelApiService] = dispatcher.get("panel_service") panel_service: Optional[PanelApiService] = dispatcher.get("panel_service")
if panel_service and hasattr(panel_service, 'close_session'): if panel_service and hasattr(panel_service, "close_session"):
await panel_service.close_session() await panel_service.close_session()
logging.info("Panel API service session closed on shutdown.") logging.info("Panel API service session closed on shutdown.")
@@ -191,6 +237,7 @@ async def on_shutdown_configured(dispatcher: Dispatcher):
logging.warning(f"SHUTDOWN: Failed to close bot session: {e}") logging.warning(f"SHUTDOWN: Failed to close bot session: {e}")
from db.database_setup import async_engine as global_async_engine from db.database_setup import async_engine as global_async_engine
if global_async_engine: if global_async_engine:
logging.info("SHUTDOWN: Disposing SQLAlchemy engine...") logging.info("SHUTDOWN: Disposing SQLAlchemy engine...")
await global_async_engine.dispose() await global_async_engine.dispose()
@@ -223,28 +270,40 @@ async def run_bot(settings_param: Settings):
f"Failed to get bot info (e.g., for YooKassa default URL): {e}. Using fallback: {actual_bot_username}" f"Failed to get bot info (e.g., for YooKassa default URL): {e}. Using fallback: {actual_bot_username}"
) )
i18n_instance = get_i18n_instance(path="locales", i18n_instance = get_i18n_instance(
default=settings_param.DEFAULT_LANGUAGE) path="locales", default=settings_param.DEFAULT_LANGUAGE
)
yookassa_service = YooKassaService( yookassa_service = YooKassaService(
shop_id=settings_param.YOOKASSA_SHOP_ID, shop_id=settings_param.YOOKASSA_SHOP_ID,
secret_key=settings_param.YOOKASSA_SECRET_KEY, secret_key=settings_param.YOOKASSA_SECRET_KEY,
configured_return_url=settings_param.YOOKASSA_RETURN_URL, configured_return_url=settings_param.YOOKASSA_RETURN_URL,
bot_username_for_default_return=actual_bot_username, bot_username_for_default_return=actual_bot_username,
settings_obj=settings_param) settings_obj=settings_param,
)
panel_service = PanelApiService(settings_param) panel_service = PanelApiService(settings_param)
subscription_service = SubscriptionService(settings_param, panel_service) subscription_service = SubscriptionService(
referral_service = ReferralService(settings_param, subscription_service, settings_param, panel_service, bot, i18n_instance
bot, i18n_instance) )
promo_code_service = PromoCodeService(settings_param, subscription_service, referral_service = ReferralService(
bot, i18n_instance) settings_param, subscription_service, bot, i18n_instance
stars_service = StarsService(bot, settings_param, i18n_instance, )
subscription_service, referral_service) promo_code_service = PromoCodeService(
tribute_service = TributeService(bot, settings_param, i18n_instance, settings_param, subscription_service, bot, i18n_instance
local_async_session_factory, )
panel_service, subscription_service, stars_service = StarsService(
referral_service) bot, settings_param, i18n_instance, subscription_service, referral_service
)
tribute_service = TributeService(
bot,
settings_param,
i18n_instance,
local_async_session_factory,
panel_service,
subscription_service,
referral_service,
)
dp["i18n_instance"] = i18n_instance dp["i18n_instance"] = i18n_instance
dp["yookassa_service"] = yookassa_service dp["yookassa_service"] = yookassa_service
@@ -256,13 +315,13 @@ async def run_bot(settings_param: Settings):
dp["tribute_service"] = tribute_service dp["tribute_service"] = tribute_service
dp["async_session_factory"] = local_async_session_factory dp["async_session_factory"] = local_async_session_factory
dp.update.outer_middleware(DBSessionMiddleware(local_async_session_factory))
dp.update.outer_middleware( dp.update.outer_middleware(
DBSessionMiddleware(local_async_session_factory)) I18nMiddleware(i18n=i18n_instance, settings=settings_param)
)
dp.update.outer_middleware( dp.update.outer_middleware(
I18nMiddleware(i18n=i18n_instance, settings=settings_param)) BanCheckMiddleware(settings=settings_param, i18n_instance=i18n_instance)
dp.update.outer_middleware( )
BanCheckMiddleware(settings=settings_param,
i18n_instance=i18n_instance))
dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings_param)) dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings_param))
dp.startup.register(on_startup_configured) dp.startup.register(on_startup_configured)
@@ -270,13 +329,12 @@ async def run_bot(settings_param: Settings):
await register_all_routers(dp, settings_param) await register_all_routers(dp, settings_param)
tg_webhook_base = getattr(settings_param, 'TELEGRAM_WEBHOOK_BASE_URL', tg_webhook_base = getattr(settings_param, "TELEGRAM_WEBHOOK_BASE_URL", None)
None) yk_webhook_base = getattr(settings_param, "YOOKASSA_WEBHOOK_BASE_URL", None)
yk_webhook_base = getattr(settings_param, 'YOOKASSA_WEBHOOK_BASE_URL',
None)
should_run_aiohttp_server = bool(tg_webhook_base) or ( should_run_aiohttp_server = bool(tg_webhook_base) or (
bool(yk_webhook_base) and bool(settings_param.yookassa_webhook_path)) bool(yk_webhook_base) and bool(settings_param.yookassa_webhook_path)
)
telegram_uses_webhook_mode = bool(tg_webhook_base) telegram_uses_webhook_mode = bool(tg_webhook_base)
run_telegram_polling = not telegram_uses_webhook_mode run_telegram_polling = not telegram_uses_webhook_mode
@@ -297,27 +355,28 @@ async def run_bot(settings_param: Settings):
if should_run_aiohttp_server: if should_run_aiohttp_server:
app = web.Application() app = web.Application()
app['bot'] = bot app["bot"] = bot
app['dp'] = dp app["dp"] = dp
app['settings'] = settings_param app["settings"] = settings_param
app['i18n'] = i18n_instance app["i18n"] = i18n_instance
app['async_session_factory'] = local_async_session_factory app["async_session_factory"] = local_async_session_factory
app['yookassa_service'] = yookassa_service app["yookassa_service"] = yookassa_service
app['subscription_service'] = subscription_service app["subscription_service"] = subscription_service
app['referral_service'] = referral_service app["referral_service"] = referral_service
app['panel_service'] = panel_service app["panel_service"] = panel_service
app['stars_service'] = stars_service app["stars_service"] = stars_service
app['tribute_service'] = tribute_service app["tribute_service"] = tribute_service
setup_application(app, dp, bot=bot) setup_application(app, dp, bot=bot)
if telegram_uses_webhook_mode: if telegram_uses_webhook_mode:
telegram_webhook_path = f"/{settings_param.BOT_TOKEN}" telegram_webhook_path = f"/{settings_param.BOT_TOKEN}"
if not telegram_webhook_path.startswith('/'): if not telegram_webhook_path.startswith("/"):
telegram_webhook_path = '/' + telegram_webhook_path telegram_webhook_path = "/" + telegram_webhook_path
app.router.add_post(telegram_webhook_path, app.router.add_post(
SimpleRequestHandler(dispatcher=dp, bot=bot)) telegram_webhook_path, SimpleRequestHandler(dispatcher=dp, bot=bot)
)
logging.info( logging.info(
f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)" f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)"
) )
@@ -328,48 +387,52 @@ async def run_bot(settings_param: Settings):
logging.error( logging.error(
f"YooKassa webhook path is invalid or not configured in settings: {yk_path}. Skipping YooKassa webhook setup." f"YooKassa webhook path is invalid or not configured in settings: {yk_path}. Skipping YooKassa webhook setup."
) )
elif not yk_path.startswith('/'): elif not yk_path.startswith("/"):
logging.error( logging.error(
f"CRITICAL: YooKassa webhook path '{yk_path}' from settings does not start with '/'. Correct settings.py or .env. Skipping YooKassa webhook." f"CRITICAL: YooKassa webhook path '{yk_path}' from settings does not start with '/'. Correct settings.py or .env. Skipping YooKassa webhook."
) )
else: else:
app.router.add_post( app.router.add_post(
yk_path, yk_path, user_payment_webhook_module.yookassa_webhook_route
user_payment_webhook_module.yookassa_webhook_route) )
logging.info( logging.info(f"YooKassa webhook route configured at: [POST] {yk_path}")
f"YooKassa webhook route configured at: [POST] {yk_path}")
tribute_path = settings_param.tribute_webhook_path tribute_path = settings_param.tribute_webhook_path
if tribute_path.startswith('/'): if tribute_path.startswith("/"):
app.router.add_post( app.router.add_post(tribute_path, tribute_webhook_route)
tribute_path, logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}")
tribute_webhook_route)
logging.info(
f"Tribute webhook route configured at: [POST] {tribute_path}")
web_app_runner = web.AppRunner(app) web_app_runner = web.AppRunner(app)
await web_app_runner.setup() await web_app_runner.setup()
site = web.TCPSite(web_app_runner, site = web.TCPSite(
host=settings_param.WEB_SERVER_HOST, web_app_runner,
port=settings_param.WEB_SERVER_PORT) host=settings_param.WEB_SERVER_HOST,
port=settings_param.WEB_SERVER_PORT,
)
async def web_server_task(): async def web_server_task():
await site.start() await site.start()
logging.info( logging.info(
f"AIOHTTP server started on http://{settings_param.WEB_SERVER_HOST}:{settings_param.WEB_SERVER_PORT}" f"AIOHTTP server started on http://{settings_param.WEB_SERVER_HOST}:{settings_param.WEB_SERVER_PORT}"
) )
await asyncio.Event().wait( (
) if not run_telegram_polling else await asyncio.sleep(31536000) await asyncio.Event().wait()
if not run_telegram_polling
else await asyncio.sleep(31536000)
)
main_tasks.append( main_tasks.append(
asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")) asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")
)
if run_telegram_polling: if run_telegram_polling:
logging.info("Starting bot in Telegram Polling mode...") logging.info("Starting bot in Telegram Polling mode...")
main_tasks.append( main_tasks.append(
asyncio.create_task(dp.start_polling( asyncio.create_task(
bot, allowed_updates=dp.resolve_used_update_types()), dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()),
name="TelegramPollingTask")) name="TelegramPollingTask",
)
)
if not main_tasks: if not main_tasks:
logging.error( logging.error(
@@ -385,8 +448,7 @@ async def run_bot(settings_param: Settings):
try: try:
await asyncio.gather(*main_tasks) await asyncio.gather(*main_tasks)
except (KeyboardInterrupt, SystemExit, asyncio.CancelledError) as e: except (KeyboardInterrupt, SystemExit, asyncio.CancelledError) as e:
logging.info( logging.info(f"Main bot loop interrupted/cancelled: {type(e).__name__} - {e}")
f"Main bot loop interrupted/cancelled: {type(e).__name__} - {e}")
finally: finally:
logging.info("Initiating final bot shutdown sequence...") logging.info("Initiating final bot shutdown sequence...")
for task in main_tasks: for task in main_tasks:
@@ -401,7 +463,8 @@ async def run_bot(settings_param: Settings):
except Exception as e_task_cancel: except Exception as e_task_cancel:
logging.error( logging.error(
f"Error during cancellation of task '{task.get_name()}': {e_task_cancel}", f"Error during cancellation of task '{task.get_name()}': {e_task_cancel}",
exc_info=True) exc_info=True,
)
if web_app_runner: if web_app_runner:
await web_app_runner.cleanup() await web_app_runner.cleanup()
+43
View File
@@ -133,3 +133,46 @@ async def schedule_subscription_notifications(
logging.info( logging.info(
f"Subscription expiration warning job scheduled daily at {notification_hour:02d}:{notification_minute:02d} UTC." f"Subscription expiration warning job scheduled daily at {notification_hour:02d}:{notification_minute:02d} UTC."
) )
async def notify_admins(bot: Bot, settings: Settings, i18n: JsonI18n,
message_key: str, parse_mode: str | None = None,
**kwargs) -> None:
if not settings.ADMIN_IDS:
return
admin_lang = settings.DEFAULT_LANGUAGE
msg = i18n.gettext(admin_lang, message_key, **kwargs)
for admin_id in settings.ADMIN_IDS:
try:
await bot.send_message(admin_id, msg, parse_mode=parse_mode)
except Exception as e:
logging.error(f"Failed to send admin notification to {admin_id}: {e}")
async def notify_admin_new_trial(bot: Bot, settings: Settings, i18n: JsonI18n,
user_id: int, end_date: datetime) -> None:
end_date_str = end_date.strftime('%Y-%m-%d') if isinstance(end_date, datetime) else str(end_date)
await notify_admins(
bot,
settings,
i18n,
"admin_new_trial_notification",
user_id=user_id,
end_date=end_date_str,
)
async def notify_admin_new_payment(bot: Bot, settings: Settings, i18n: JsonI18n,
user_id: int, months: int, amount: float,
currency: str | None = None) -> None:
currency_symbol = currency or settings.DEFAULT_CURRENCY_SYMBOL
await notify_admins(
bot,
settings,
i18n,
"admin_new_payment_notification",
user_id=user_id,
months=months,
amount=f"{amount:.2f}",
currency=currency_symbol,
)
+22
View File
@@ -226,6 +226,28 @@ class PanelApiService:
return None return None
async def get_user(
self,
*,
uuid: Optional[str] = None,
telegram_id: Optional[int] = None,
username: Optional[str] = None,
email: Optional[str] = None,
log_response: bool = True,
) -> Optional[Dict[str, Any]]:
if uuid:
return await self.get_user_by_uuid(uuid, log_response=log_response)
users = await self.get_users_by_filter(
telegram_id=telegram_id,
username=username,
email=email,
log_response=log_response,
)
if users:
return users[0]
return None
async def get_users_by_filter( async def get_users_by_filter(
self, self,
telegram_id: Optional[int] = None, telegram_id: Optional[int] = None,
+1 -6
View File
@@ -5,7 +5,7 @@ from aiogram import Bot
from config.settings import Settings from config.settings import Settings
from db.dal import promo_code_dal, user_dal, subscription_dal from db.dal import promo_code_dal, user_dal
from db.models import PromoCode, User from db.models import PromoCode, User
from .subscription_service import SubscriptionService from .subscription_service import SubscriptionService
@@ -40,11 +40,6 @@ class PromoCodeService:
return False, _("promo_code_already_used_by_user", return False, _("promo_code_already_used_by_user",
code=code_input_upper) code=code_input_upper)
active_sub = await subscription_dal.get_active_subscription_by_user_id(
session, user_id)
if not active_sub:
return False, _("promo_code_no_active_subscription")
bonus_days = promo_data.bonus_days bonus_days = promo_data.bonus_days
new_end_date = await self.subscription_service.extend_active_subscription_days( new_end_date = await self.subscription_service.extend_active_subscription_days(
+1 -2
View File
@@ -137,8 +137,7 @@ class ReferralService:
"status_from_panel": "status_from_panel":
"ACTIVE_BONUS", "ACTIVE_BONUS",
"traffic_limit_bytes": "traffic_limit_bytes":
self.settings. self.settings.user_traffic_limit_bytes,
PANEL_USER_DEFAULT_TRAFFIC_BYTES,
} }
try: try:
await subscription_dal.deactivate_other_active_subscriptions( await subscription_dal.deactivate_other_active_subscriptions(
+360 -234
View File
@@ -2,6 +2,8 @@ import logging
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any, List, Tuple from typing import Optional, Dict, Any, List, Tuple
from aiogram import Bot
from bot.middlewares.i18n import JsonI18n
from db.dal import user_dal, subscription_dal, promo_code_dal, payment_dal from db.dal import user_dal, subscription_dal, promo_code_dal, payment_dal
from db.models import User, Subscription from db.models import User, Subscription
@@ -12,26 +14,48 @@ from .panel_api_service import PanelApiService
class SubscriptionService: class SubscriptionService:
def __init__(self, settings: Settings, panel_service: PanelApiService): def __init__(
self,
settings: Settings,
panel_service: PanelApiService,
bot: Optional[Bot] = None,
i18n: Optional[JsonI18n] = None,
):
self.settings = settings self.settings = settings
self.panel_service = panel_service self.panel_service = panel_service
self.bot = bot
self.i18n = i18n
async def get_user_language(self, session: AsyncSession, async def get_user_language(self, session: AsyncSession, user_id: int) -> str:
user_id: int) -> str:
user_record = await user_dal.get_user_by_id(session, user_id) user_record = await user_dal.get_user_by_id(session, user_id)
return user_record.language_code if user_record and user_record.language_code else self.settings.DEFAULT_LANGUAGE return (
user_record.language_code
if user_record and user_record.language_code
else self.settings.DEFAULT_LANGUAGE
)
async def has_had_any_subscription(self, session: AsyncSession, async def has_had_any_subscription(
user_id: int) -> bool: self, session: AsyncSession, user_id: int
) -> bool:
return await subscription_dal.has_any_subscription_for_user( return await subscription_dal.has_any_subscription_for_user(session, user_id)
session, user_id)
async def _notify_admin_panel_user_creation_failed(self, user_id: int):
if not self.bot or not self.i18n or not self.settings.ADMIN_IDS:
return
admin_lang = self.settings.DEFAULT_LANGUAGE
_adm = lambda k, **kw: self.i18n.gettext(admin_lang, k, **kw)
msg = _adm("admin_panel_user_creation_failed", user_id=user_id)
for admin_id in self.settings.ADMIN_IDS:
try:
await self.bot.send_message(admin_id, msg)
except Exception as e:
logging.error(
f"Failed to notify admin {admin_id} about panel user creation failure: {e}"
)
async def _get_or_create_panel_user_link_details( async def _get_or_create_panel_user_link_details(
self, self, session: AsyncSession, user_id: int, db_user: Optional[User] = None
session: AsyncSession,
user_id: int,
db_user: Optional[User] = None
) -> Tuple[Optional[str], Optional[str], Optional[str], bool]: ) -> Tuple[Optional[str], Optional[str], Optional[str], bool]:
if not db_user: if not db_user:
db_user = await user_dal.get_user_by_id(session, user_id) db_user = await user_dal.get_user_by_id(session, user_id)
@@ -49,7 +73,8 @@ class SubscriptionService:
panel_user_created_or_linked_now = False panel_user_created_or_linked_now = False
panel_users_by_tg_id_list = await self.panel_service.get_users_by_filter( panel_users_by_tg_id_list = await self.panel_service.get_users_by_filter(
telegram_id=user_id) telegram_id=user_id
)
if panel_users_by_tg_id_list and len(panel_users_by_tg_id_list) == 1: if panel_users_by_tg_id_list and len(panel_users_by_tg_id_list) == 1:
panel_user_obj_from_api = panel_users_by_tg_id_list[0] panel_user_obj_from_api = panel_users_by_tg_id_list[0]
logging.info( logging.info(
@@ -68,11 +93,32 @@ class SubscriptionService:
f"User {user_id} (local panel_uuid: {current_local_panel_uuid}) not found on panel by TG ID. Fetching by panel_uuid." f"User {user_id} (local panel_uuid: {current_local_panel_uuid}) not found on panel by TG ID. Fetching by panel_uuid."
) )
panel_user_obj_from_api = await self.panel_service.get_user_by_uuid( panel_user_obj_from_api = await self.panel_service.get_user_by_uuid(
current_local_panel_uuid) current_local_panel_uuid
)
if not panel_user_obj_from_api: if not panel_user_obj_from_api:
logging.warning( logging.warning(
f"Local panel_uuid {current_local_panel_uuid} for TG user {user_id} also not found on panel. User might be deleted from panel or UUID desynced." f"Local panel_uuid {current_local_panel_uuid} for TG user {user_id} also not found on panel. User might be deleted from panel or UUID desynced."
) )
logging.info(
f"Creating new panel user '{panel_username_on_panel_standard}' for TG user {user_id}."
)
creation_response = await self.panel_service.create_panel_user(
username_on_panel=panel_username_on_panel_standard,
telegram_id=user_id,
specific_inbound_uuids=self.settings.parsed_user_inbound_uuids,
default_traffic_limit_bytes=self.settings.user_traffic_limit_bytes,
default_traffic_limit_strategy=self.settings.USER_TRAFFIC_STRATEGY,
)
if (
creation_response
and not creation_response.get("error")
and creation_response.get("response")
):
panel_user_obj_from_api = creation_response.get("response")
panel_user_created_or_linked_now = True
else:
await self._notify_admin_panel_user_creation_failed(user_id)
return None, None, None, False
else: else:
@@ -82,27 +128,35 @@ class SubscriptionService:
creation_response = await self.panel_service.create_panel_user( creation_response = await self.panel_service.create_panel_user(
username_on_panel=panel_username_on_panel_standard, username_on_panel=panel_username_on_panel_standard,
telegram_id=user_id, telegram_id=user_id,
specific_inbound_uuids=self.settings.parsed_user_inbound_uuids,
default_traffic_limit_bytes=self.settings.user_traffic_limit_bytes,
default_traffic_limit_strategy=self.settings.USER_TRAFFIC_STRATEGY,
) )
if creation_response and not creation_response.get( if (
"error") and creation_response.get("response"): creation_response
and not creation_response.get("error")
and creation_response.get("response")
):
panel_user_obj_from_api = creation_response.get("response") panel_user_obj_from_api = creation_response.get("response")
panel_user_created_or_linked_now = True panel_user_created_or_linked_now = True
elif creation_response and creation_response.get( elif creation_response and creation_response.get("errorCode") == "A019":
"errorCode") == "A019":
logging.warning( logging.warning(
f"Panel user '{panel_username_on_panel_standard}' already exists (errorCode A019). Fetching by username." f"Panel user '{panel_username_on_panel_standard}' already exists (errorCode A019). Fetching by username."
) )
fetched_by_username_list = await self.panel_service.get_users_by_filter( fetched_by_username_list = (
username=panel_username_on_panel_standard) await self.panel_service.get_users_by_filter(
if fetched_by_username_list and len( username=panel_username_on_panel_standard
fetched_by_username_list) == 1: )
)
if fetched_by_username_list and len(fetched_by_username_list) == 1:
panel_user_obj_from_api = fetched_by_username_list[0] panel_user_obj_from_api = fetched_by_username_list[0]
if not panel_user_obj_from_api: if not panel_user_obj_from_api:
logging.error( logging.error(
f"Failed to create or link panel user for TG_ID {user_id} with panel username '{panel_username_on_panel_standard}'. Response: {creation_response if 'creation_response' in locals() else 'N/A'}" f"Failed to create or link panel user for TG_ID {user_id} with panel username '{panel_username_on_panel_standard}'. Response: {creation_response if 'creation_response' in locals() else 'N/A'}"
) )
await self._notify_admin_panel_user_creation_failed(user_id)
return None, None, None, False return None, None, None, False
if not panel_user_obj_from_api: if not panel_user_obj_from_api:
@@ -110,33 +164,47 @@ class SubscriptionService:
f"Could not obtain panel user object for TG user {user_id} after all checks." f"Could not obtain panel user object for TG user {user_id} after all checks."
) )
return current_local_panel_uuid if current_local_panel_uuid else None, None, None, panel_user_created_or_linked_now return (
current_local_panel_uuid if current_local_panel_uuid else None,
None,
None,
panel_user_created_or_linked_now,
)
actual_panel_uuid_from_api = panel_user_obj_from_api.get('uuid') actual_panel_uuid_from_api = panel_user_obj_from_api.get("uuid")
actual_panel_username_from_api = panel_user_obj_from_api.get( actual_panel_username_from_api = panel_user_obj_from_api.get("username")
'username') panel_telegram_id_from_api = panel_user_obj_from_api.get("telegramId")
panel_telegram_id_from_api = panel_user_obj_from_api.get('telegramId')
if not actual_panel_uuid_from_api: if not actual_panel_uuid_from_api:
logging.error( logging.error(
f"Panel user object for TG user {user_id} does not contain 'uuid'. Data: {panel_user_obj_from_api}" f"Panel user object for TG user {user_id} does not contain 'uuid'. Data: {panel_user_obj_from_api}"
) )
return current_local_panel_uuid, None, None, panel_user_created_or_linked_now return (
current_local_panel_uuid,
None,
None,
panel_user_created_or_linked_now,
)
needs_local_panel_uuid_update = False needs_local_panel_uuid_update = False
if current_local_panel_uuid is None and actual_panel_uuid_from_api: if current_local_panel_uuid is None and actual_panel_uuid_from_api:
needs_local_panel_uuid_update = True needs_local_panel_uuid_update = True
elif current_local_panel_uuid is not None and current_local_panel_uuid != actual_panel_uuid_from_api: elif (
current_local_panel_uuid is not None
and current_local_panel_uuid != actual_panel_uuid_from_api
):
logging.warning( logging.warning(
f"Local panel_uuid for user {user_id} ('{current_local_panel_uuid}') " f"Local panel_uuid for user {user_id} ('{current_local_panel_uuid}') "
f"differs from panel's UUID ('{actual_panel_uuid_from_api}') for their telegramId. " f"differs from panel's UUID ('{actual_panel_uuid_from_api}') for their telegramId. "
f"Will attempt to update local to panel's version.") f"Will attempt to update local to panel's version."
)
needs_local_panel_uuid_update = True needs_local_panel_uuid_update = True
if needs_local_panel_uuid_update: if needs_local_panel_uuid_update:
conflicting_user_record = await user_dal.get_user_by_panel_uuid( conflicting_user_record = await user_dal.get_user_by_panel_uuid(
session, actual_panel_uuid_from_api) session, actual_panel_uuid_from_api
)
if conflicting_user_record and conflicting_user_record.user_id != user_id: if conflicting_user_record and conflicting_user_record.user_id != user_id:
logging.error( logging.error(
f"CRITICAL CONFLICT: Panel UUID {actual_panel_uuid_from_api} (from panel for TG ID {user_id}) " f"CRITICAL CONFLICT: Panel UUID {actual_panel_uuid_from_api} (from panel for TG ID {user_id}) "
@@ -151,14 +219,20 @@ class SubscriptionService:
"panel_user_uuid": actual_panel_uuid_from_api "panel_user_uuid": actual_panel_uuid_from_api
} }
if actual_panel_username_from_api and actual_panel_username_from_api != panel_username_on_panel_standard and ( if (
actual_panel_username_from_api
and actual_panel_username_from_api
!= panel_username_on_panel_standard
and (
db_user.username is None db_user.username is None
or db_user.username != actual_panel_username_from_api): or db_user.username != actual_panel_username_from_api
update_data_for_local_user[ )
"username"] = actual_panel_username_from_api ):
update_data_for_local_user["username"] = (
actual_panel_username_from_api
)
await user_dal.update_user(session, user_id, await user_dal.update_user(session, user_id, update_data_for_local_user)
update_data_for_local_user)
db_user.panel_user_uuid = actual_panel_uuid_from_api db_user.panel_user_uuid = actual_panel_uuid_from_api
if "username" in update_data_for_local_user: if "username" in update_data_for_local_user:
db_user.username = update_data_for_local_user["username"] db_user.username = update_data_for_local_user["username"]
@@ -175,69 +249,79 @@ class SubscriptionService:
except ValueError: except ValueError:
pass pass
if panel_user_obj_from_api and current_local_panel_uuid and panel_telegram_id_int != user_id: if (
panel_user_obj_from_api
and current_local_panel_uuid
and panel_telegram_id_int != user_id
):
logging.info( logging.info(
f"Panel user {current_local_panel_uuid} has telegramId '{panel_telegram_id_from_api}'. Updating on panel to '{user_id}'." f"Panel user {current_local_panel_uuid} has telegramId '{panel_telegram_id_from_api}'. Updating on panel to '{user_id}'."
) )
await self.panel_service.update_user_details_on_panel( await self.panel_service.update_user_details_on_panel(
current_local_panel_uuid, {"telegramId": user_id}) current_local_panel_uuid, {"telegramId": user_id}
)
panel_sub_link_id = panel_user_obj_from_api.get( panel_sub_link_id = panel_user_obj_from_api.get(
'subscriptionUuid') or panel_user_obj_from_api.get('shortUuid') "subscriptionUuid"
panel_short_uuid = panel_user_obj_from_api.get('shortUuid') ) or panel_user_obj_from_api.get("shortUuid")
panel_short_uuid = panel_user_obj_from_api.get("shortUuid")
if not panel_sub_link_id and current_local_panel_uuid: if not panel_sub_link_id and current_local_panel_uuid:
logging.warning( logging.warning(
f"No subscriptionUuid or shortUuid found on panel for panel_user_uuid {current_local_panel_uuid} (TG ID: {user_id})." f"No subscriptionUuid or shortUuid found on panel for panel_user_uuid {current_local_panel_uuid} (TG ID: {user_id})."
) )
return current_local_panel_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_or_linked_now return (
current_local_panel_uuid,
panel_sub_link_id,
panel_short_uuid,
panel_user_created_or_linked_now,
)
async def activate_trial_subscription( async def activate_trial_subscription(
self, session: AsyncSession, self, session: AsyncSession, user_id: int
user_id: int) -> Optional[Dict[str, Any]]: ) -> Optional[Dict[str, Any]]:
if not self.settings.TRIAL_ENABLED or self.settings.TRIAL_DURATION_DAYS <= 0: if not self.settings.TRIAL_ENABLED or self.settings.TRIAL_DURATION_DAYS <= 0:
return { return {
"eligible": False, "eligible": False,
"activated": False, "activated": False,
"message_key": "trial_feature_disabled" "message_key": "trial_feature_disabled",
} }
db_user = await user_dal.get_user_by_id(session, user_id) db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user: if not db_user:
logging.error( logging.error(f"User {user_id} not found in DB, cannot activate trial.")
f"User {user_id} not found in DB, cannot activate trial.")
return { return {
"eligible": False, "eligible": False,
"activated": False, "activated": False,
"message_key": "user_not_found_for_trial" "message_key": "user_not_found_for_trial",
} }
if await self.has_had_any_subscription(session, user_id): if await self.has_had_any_subscription(session, user_id):
return { return {
"eligible": False, "eligible": False,
"activated": False, "activated": False,
"message_key": "trial_already_had_subscription_or_trial" "message_key": "trial_already_had_subscription_or_trial",
} }
panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = await self._get_or_create_panel_user_link_details( panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = (
session, user_id, db_user) await self._get_or_create_panel_user_link_details(session, user_id, db_user)
)
if not panel_user_uuid or not panel_sub_link_id: if not panel_user_uuid or not panel_sub_link_id:
logging.error( logging.error(f"Failed to get panel link details for trial user {user_id}.")
f"Failed to get panel link details for trial user {user_id}.")
return { return {
"eligible": True, "eligible": True,
"activated": False, "activated": False,
"message_key": "trial_activation_failed_panel_link" "message_key": "trial_activation_failed_panel_link",
} }
start_date = datetime.now(timezone.utc) start_date = datetime.now(timezone.utc)
end_date = start_date + timedelta( end_date = start_date + timedelta(days=self.settings.TRIAL_DURATION_DAYS)
days=self.settings.TRIAL_DURATION_DAYS)
await subscription_dal.deactivate_other_active_subscriptions( await subscription_dal.deactivate_other_active_subscriptions(
session, panel_user_uuid, panel_sub_link_id) session, panel_user_uuid, panel_sub_link_id
)
trial_sub_data = { trial_sub_data = {
"user_id": user_id, "user_id": user_id,
@@ -255,34 +339,34 @@ class SubscriptionService:
except Exception as e_upsert: except Exception as e_upsert:
logging.error( logging.error(
f"Failed to upsert trial subscription for user {user_id}: {e_upsert}", f"Failed to upsert trial subscription for user {user_id}: {e_upsert}",
exc_info=True) exc_info=True,
)
await session.rollback() await session.rollback()
return { return {
"eligible": True, "eligible": True,
"activated": False, "activated": False,
"message_key": "trial_activation_failed_db" "message_key": "trial_activation_failed_db",
} }
panel_update_payload: Dict[str, Any] = { panel_update_payload: Dict[str, Any] = {
"uuid": "uuid": panel_user_uuid,
panel_user_uuid, "expireAt": end_date.isoformat(timespec="milliseconds").replace(
"expireAt": "+00:00", "Z"
end_date.isoformat(timespec='milliseconds').replace('+00:00', 'Z'), ),
"status": "status": "ACTIVE",
"ACTIVE", "trafficLimitBytes": self.settings.trial_traffic_limit_bytes,
"trafficLimitBytes": "trafficLimitStrategy": self.settings.USER_TRAFFIC_STRATEGY,
self.settings.trial_traffic_limit_bytes,
"trafficLimitStrategy":
self.settings.PANEL_USER_DEFAULT_TRAFFIC_STRATEGY,
} }
if self.settings.parsed_default_panel_user_inbound_uuids: if self.settings.parsed_user_inbound_uuids:
panel_update_payload[ panel_update_payload["activeUserInbounds"] = (
"activeUserInbounds"] = self.settings.parsed_default_panel_user_inbound_uuids self.settings.parsed_user_inbound_uuids
)
elif panel_user_created_now: elif panel_user_created_now:
panel_update_payload["activateAllInbounds"] = True panel_update_payload["activateAllInbounds"] = True
updated_panel_user = await self.panel_service.update_user_details_on_panel( updated_panel_user = await self.panel_service.update_user_details_on_panel(
panel_user_uuid, panel_update_payload) panel_user_uuid, panel_update_payload
)
if not updated_panel_user or updated_panel_user.get("error"): if not updated_panel_user or updated_panel_user.get("error"):
logging.warning( logging.warning(
f"Panel user details update FAILED for trial user {panel_user_uuid}. Response: {updated_panel_user}" f"Panel user details update FAILED for trial user {panel_user_uuid}. Response: {updated_panel_user}"
@@ -291,14 +375,13 @@ class SubscriptionService:
return { return {
"eligible": True, "eligible": True,
"activated": False, "activated": False,
"message_key": "trial_activation_failed_panel_update" "message_key": "trial_activation_failed_panel_update",
} }
await session.commit() await session.commit()
final_subscription_url = updated_panel_user.get('subscriptionUrl') final_subscription_url = updated_panel_user.get("subscriptionUrl")
final_panel_short_uuid = updated_panel_user.get( final_panel_short_uuid = updated_panel_user.get("shortUuid", panel_short_uuid)
'shortUuid', panel_short_uuid)
return { return {
"eligible": True, "eligible": True,
@@ -308,7 +391,7 @@ class SubscriptionService:
"traffic_gb": self.settings.TRIAL_TRAFFIC_LIMIT_GB, "traffic_gb": self.settings.TRIAL_TRAFFIC_LIMIT_GB,
"panel_user_uuid": panel_user_uuid, "panel_user_uuid": panel_user_uuid,
"panel_short_uuid": final_panel_short_uuid, "panel_short_uuid": final_panel_short_uuid,
"subscription_url": final_subscription_url "subscription_url": final_subscription_url,
} }
async def activate_subscription( async def activate_subscription(
@@ -319,7 +402,7 @@ class SubscriptionService:
payment_amount: float, payment_amount: float,
payment_db_id: int, payment_db_id: int,
promo_code_id_from_payment: Optional[int] = None, promo_code_id_from_payment: Optional[int] = None,
provider: str = "yookassa" provider: str = "yookassa",
) -> Optional[Dict[str, Any]]: ) -> Optional[Dict[str, Any]]:
db_user = await user_dal.get_user_by_id(session, user_id) db_user = await user_dal.get_user_by_id(session, user_id)
@@ -329,8 +412,9 @@ class SubscriptionService:
) )
return None return None
panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = await self._get_or_create_panel_user_link_details( panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = (
session, user_id, db_user) await self._get_or_create_panel_user_link_details(session, user_id, db_user)
)
if not panel_user_uuid or not panel_sub_link_id: if not panel_user_uuid or not panel_sub_link_id:
logging.error( logging.error(
@@ -339,9 +423,14 @@ class SubscriptionService:
return None return None
current_active_sub = await subscription_dal.get_active_subscription_by_user_id( current_active_sub = await subscription_dal.get_active_subscription_by_user_id(
session, user_id, panel_user_uuid) session, user_id, panel_user_uuid
)
start_date = datetime.now(timezone.utc) start_date = datetime.now(timezone.utc)
if current_active_sub and current_active_sub.end_date and current_active_sub.end_date > start_date: if (
current_active_sub
and current_active_sub.end_date
and current_active_sub.end_date > start_date
):
start_date = current_active_sub.end_date start_date = current_active_sub.end_date
duration_days_total = months * 30 duration_days_total = months * 30
@@ -349,8 +438,13 @@ class SubscriptionService:
if promo_code_id_from_payment: if promo_code_id_from_payment:
promo_model = await promo_code_dal.get_promo_code_by_id( promo_model = await promo_code_dal.get_promo_code_by_id(
session, promo_code_id_from_payment) session, promo_code_id_from_payment
if promo_model and promo_model.is_active and promo_model.current_activations < promo_model.max_activations: )
if (
promo_model
and promo_model.is_active
and promo_model.current_activations < promo_model.max_activations
):
applied_promo_bonus_days = promo_model.bonus_days applied_promo_bonus_days = promo_model.bonus_days
duration_days_total += applied_promo_bonus_days duration_days_total += applied_promo_bonus_days
@@ -358,10 +452,12 @@ class SubscriptionService:
session, session,
promo_code_id_from_payment, promo_code_id_from_payment,
user_id, user_id,
payment_id=payment_db_id) payment_id=payment_db_id,
)
if activation: if activation:
await promo_code_dal.increment_promo_code_usage( await promo_code_dal.increment_promo_code_usage(
session, promo_code_id_from_payment) session, promo_code_id_from_payment
)
else: else:
logging.warning( logging.warning(
f"Promo code {promo_code_id_from_payment} was already activated by user {user_id}, but bonus applied via payment {payment_db_id}." f"Promo code {promo_code_id_from_payment} was already activated by user {user_id}, but bonus applied via payment {payment_db_id}."
@@ -374,7 +470,8 @@ class SubscriptionService:
final_end_date = start_date + timedelta(days=duration_days_total) final_end_date = start_date + timedelta(days=duration_days_total)
await subscription_dal.deactivate_other_active_subscriptions( await subscription_dal.deactivate_other_active_subscriptions(
session, panel_user_uuid, panel_sub_link_id) session, panel_user_uuid, panel_sub_link_id
)
sub_payload = { sub_payload = {
"user_id": user_id, "user_id": user_id,
@@ -385,50 +482,48 @@ class SubscriptionService:
"duration_months": months, "duration_months": months,
"is_active": True, "is_active": True,
"status_from_panel": "ACTIVE", "status_from_panel": "ACTIVE",
"traffic_limit_bytes": "traffic_limit_bytes": self.settings.user_traffic_limit_bytes,
self.settings.PANEL_USER_DEFAULT_TRAFFIC_BYTES,
"provider": provider, "provider": provider,
"skip_notifications": provider == "tribute", "skip_notifications": provider == "tribute",
} }
try: try:
new_or_updated_sub = await subscription_dal.upsert_subscription( new_or_updated_sub = await subscription_dal.upsert_subscription(
session, sub_payload) session, sub_payload
)
except Exception as e_upsert_sub: except Exception as e_upsert_sub:
logging.error( logging.error(
f"Failed to upsert paid subscription for user {user_id}: {e_upsert_sub}", f"Failed to upsert paid subscription for user {user_id}: {e_upsert_sub}",
exc_info=True) exc_info=True,
)
return None return None
panel_update_payload = { panel_update_payload = {
"uuid": "uuid": panel_user_uuid,
panel_user_uuid, "expireAt": final_end_date.isoformat(timespec="milliseconds").replace(
"expireAt": "+00:00", "Z"
final_end_date.isoformat(timespec='milliseconds').replace( ),
'+00:00', 'Z'), "status": "ACTIVE",
"status": "trafficLimitBytes": self.settings.user_traffic_limit_bytes,
"ACTIVE", "trafficLimitStrategy": self.settings.USER_TRAFFIC_STRATEGY,
"trafficLimitBytes":
self.settings.PANEL_USER_DEFAULT_TRAFFIC_BYTES,
"trafficLimitStrategy":
self.settings.PANEL_USER_DEFAULT_TRAFFIC_STRATEGY
} }
if self.settings.parsed_default_panel_user_inbound_uuids: if self.settings.parsed_user_inbound_uuids:
panel_update_payload[ panel_update_payload["activeUserInbounds"] = (
"activeUserInbounds"] = self.settings.parsed_default_panel_user_inbound_uuids self.settings.parsed_user_inbound_uuids
)
elif panel_user_created_now: elif panel_user_created_now:
panel_update_payload["activateAllInbounds"] = True panel_update_payload["activateAllInbounds"] = True
updated_panel_user = await self.panel_service.update_user_details_on_panel( updated_panel_user = await self.panel_service.update_user_details_on_panel(
panel_user_uuid, panel_update_payload) panel_user_uuid, panel_update_payload
)
if not updated_panel_user or updated_panel_user.get("error"): if not updated_panel_user or updated_panel_user.get("error"):
logging.warning( logging.warning(
f"Panel user details update FAILED for paid sub user {panel_user_uuid}. Response: {updated_panel_user}" f"Panel user details update FAILED for paid sub user {panel_user_uuid}. Response: {updated_panel_user}"
) )
return None return None
final_subscription_url = updated_panel_user.get('subscriptionUrl') final_subscription_url = updated_panel_user.get("subscriptionUrl")
final_panel_short_uuid = updated_panel_user.get( final_panel_short_uuid = updated_panel_user.get("shortUuid", panel_short_uuid)
'shortUuid', panel_short_uuid)
return { return {
"subscription_id": new_or_updated_sub.subscription_id, "subscription_id": new_or_updated_sub.subscription_id,
@@ -437,48 +532,84 @@ class SubscriptionService:
"panel_user_uuid": panel_user_uuid, "panel_user_uuid": panel_user_uuid,
"panel_short_uuid": final_panel_short_uuid, "panel_short_uuid": final_panel_short_uuid,
"subscription_url": final_subscription_url, "subscription_url": final_subscription_url,
"applied_promo_bonus_days": applied_promo_bonus_days "applied_promo_bonus_days": applied_promo_bonus_days,
} }
async def extend_active_subscription_days( async def extend_active_subscription_days(
self, self,
session: AsyncSession, session: AsyncSession,
user_id: int, user_id: int,
bonus_days: int, bonus_days: int,
reason: str = "bonus") -> Optional[datetime]: reason: str = "bonus",
) -> Optional[datetime]:
user = await user_dal.get_user_by_id(session, user_id) user = await user_dal.get_user_by_id(session, user_id)
if not user or not user.panel_user_uuid: if not user:
logging.warning( logging.warning(
f"Cannot extend subscription for user {user_id}: User or panel_user_uuid not found." f"Cannot extend subscription for user {user_id}: user not found."
)
return None
panel_uuid, panel_sub_uuid, _, _ = await self._get_or_create_panel_user_link_details(
session, user_id, user
)
if not panel_uuid or not panel_sub_uuid:
logging.error(
f"Failed to ensure panel user for subscription extension of user {user_id}."
) )
return None return None
active_sub = await subscription_dal.get_active_subscription_by_user_id( active_sub = await subscription_dal.get_active_subscription_by_user_id(
session, user_id, user.panel_user_uuid) session, user_id, panel_uuid
)
if not active_sub or not active_sub.end_date: if not active_sub or not active_sub.end_date:
logging.info( logging.info(
f"No active extendable subscription found for user {user_id} (panel: {user.panel_user_uuid}) for reason: {reason}." f"No active subscription found for user {user_id}. Creating new one for {bonus_days} days."
) )
return None start_date = datetime.now(timezone.utc)
new_end_date_obj = start_date + timedelta(days=bonus_days)
bonus_sub_payload = {
"user_id": user_id,
"panel_user_uuid": panel_uuid,
"panel_subscription_uuid": panel_sub_uuid,
"start_date": start_date,
"end_date": new_end_date_obj,
"duration_months": 0,
"is_active": True,
"status_from_panel": "ACTIVE_BONUS",
"traffic_limit_bytes": self.settings.user_traffic_limit_bytes,
}
await subscription_dal.deactivate_other_active_subscriptions(
session, panel_uuid, panel_sub_uuid
)
updated_sub_model = await subscription_dal.upsert_subscription(
session, bonus_sub_payload
)
else:
current_end_date = active_sub.end_date
now_utc = datetime.now(timezone.utc)
start_point_for_bonus = (
current_end_date if current_end_date > now_utc else now_utc
)
new_end_date_obj = start_point_for_bonus + timedelta(days=bonus_days)
current_end_date = active_sub.end_date updated_sub_model = await subscription_dal.update_subscription_end_date(
now_utc = datetime.now(timezone.utc) session, active_sub.subscription_id, new_end_date_obj
start_point_for_bonus = current_end_date if current_end_date > now_utc else now_utc )
new_end_date_obj = start_point_for_bonus + timedelta(days=bonus_days)
updated_sub_model = await subscription_dal.update_subscription_end_date(
session, active_sub.subscription_id, new_end_date_obj)
if updated_sub_model: if updated_sub_model:
panel_update_success = await self.panel_service.update_user_details_on_panel( panel_update_success = (
user.panel_user_uuid, { await self.panel_service.update_user_details_on_panel(
"expireAt": panel_uuid,
new_end_date_obj.isoformat( {
timespec='milliseconds').replace('+00:00', 'Z') "expireAt": new_end_date_obj.isoformat(
}) timespec="milliseconds"
).replace("+00:00", "Z")
},
)
)
if not panel_update_success: if not panel_update_success:
logging.warning( logging.warning(
f"Panel expiry update failed for {user.panel_user_uuid} after {reason} bonus. Local DB was updated to {new_end_date_obj}." f"Panel expiry update failed for {panel_uuid} after {reason} bonus. Local DB was updated to {new_end_date_obj}."
) )
logging.info( logging.info(
@@ -492,8 +623,8 @@ class SubscriptionService:
return None return None
async def get_active_subscription_details( async def get_active_subscription_details(
self, session: AsyncSession, self, session: AsyncSession, user_id: int
user_id: int) -> Optional[Dict[str, Any]]: ) -> Optional[Dict[str, Any]]:
db_user = await user_dal.get_user_by_id(session, user_id) db_user = await user_dal.get_user_by_id(session, user_id)
if not db_user or not db_user.panel_user_uuid: if not db_user or not db_user.panel_user_uuid:
logging.info( logging.info(
@@ -503,136 +634,131 @@ class SubscriptionService:
panel_user_uuid = db_user.panel_user_uuid panel_user_uuid = db_user.panel_user_uuid
local_active_sub = await subscription_dal.get_active_subscription_by_user_id( local_active_sub = await subscription_dal.get_active_subscription_by_user_id(
session, user_id, panel_user_uuid) session, user_id, panel_user_uuid
panel_user_data = await self.panel_service.get_user_by_uuid( )
panel_user_uuid) panel_user_data = await self.panel_service.get_user_by_uuid(panel_user_uuid)
if not panel_user_data: if not panel_user_data:
logging.warning( logging.warning(
f"Panel user {panel_user_uuid} not found on panel for user {user_id}. Using local data if available." f"Panel user {panel_user_uuid} not found on panel for user {user_id}. Clearing local linkage."
) )
if local_active_sub and local_active_sub.end_date and local_active_sub.end_date > datetime.now( await subscription_dal.deactivate_all_user_subscriptions(session, user_id)
timezone.utc): await user_dal.update_user(session, user_id, {"panel_user_uuid": None})
return {
"end_date":
local_active_sub.end_date,
"status_from_panel":
(local_active_sub.status_from_panel or "UNKNOWN") +
" (Panel Unreachable)",
"config_link":
await self.panel_service.get_subscription_link(
local_active_sub.panel_subscription_uuid)
if local_active_sub.panel_subscription_uuid else "N/A",
"traffic_limit_bytes":
local_active_sub.traffic_limit_bytes,
"traffic_used_bytes":
local_active_sub.traffic_used_bytes,
"user_bot_username":
db_user.username,
"is_panel_data":
False
}
return None return None
if local_active_sub: if local_active_sub:
update_payload_local = {} update_payload_local = {}
panel_status = panel_user_data.get('status', 'UNKNOWN').upper() panel_status = panel_user_data.get("status", "UNKNOWN").upper()
panel_expire_at_str = panel_user_data.get('expireAt') panel_expire_at_str = panel_user_data.get("expireAt")
panel_traffic_used = panel_user_data.get('usedTrafficBytes') panel_traffic_used = panel_user_data.get("usedTrafficBytes")
panel_traffic_limit = panel_user_data.get('trafficLimitBytes') panel_traffic_limit = panel_user_data.get("trafficLimitBytes")
panel_sub_uuid_from_panel = panel_user_data.get( panel_sub_uuid_from_panel = panel_user_data.get(
'subscriptionUuid') or panel_user_data.get('shortUuid') "subscriptionUuid"
) or panel_user_data.get("shortUuid")
if local_active_sub.status_from_panel != panel_status: if local_active_sub.status_from_panel != panel_status:
update_payload_local['status_from_panel'] = panel_status update_payload_local["status_from_panel"] = panel_status
if panel_expire_at_str: if panel_expire_at_str:
panel_expire_dt = datetime.fromisoformat( panel_expire_dt = datetime.fromisoformat(
panel_expire_at_str.replace("Z", "+00:00")) panel_expire_at_str.replace("Z", "+00:00")
)
if local_active_sub.end_date.replace( if local_active_sub.end_date.replace(
microsecond=0) != panel_expire_dt.replace( microsecond=0
microsecond=0): ) != panel_expire_dt.replace(microsecond=0):
update_payload_local['end_date'] = panel_expire_dt update_payload_local["end_date"] = panel_expire_dt
update_payload_local['last_notification_sent'] = None update_payload_local["last_notification_sent"] = None
if panel_traffic_used is not None and local_active_sub.traffic_used_bytes != panel_traffic_used: if (
update_payload_local['traffic_used_bytes'] = panel_traffic_used panel_traffic_used is not None
if panel_traffic_limit is not None and local_active_sub.traffic_limit_bytes != panel_traffic_limit: and local_active_sub.traffic_used_bytes != panel_traffic_used
update_payload_local[ ):
'traffic_limit_bytes'] = panel_traffic_limit update_payload_local["traffic_used_bytes"] = panel_traffic_used
if panel_sub_uuid_from_panel and local_active_sub.panel_subscription_uuid != panel_sub_uuid_from_panel: if (
update_payload_local[ panel_traffic_limit is not None
'panel_subscription_uuid'] = panel_sub_uuid_from_panel and local_active_sub.traffic_limit_bytes != panel_traffic_limit
):
update_payload_local["traffic_limit_bytes"] = panel_traffic_limit
if (
panel_sub_uuid_from_panel
and local_active_sub.panel_subscription_uuid
!= panel_sub_uuid_from_panel
):
update_payload_local["panel_subscription_uuid"] = (
panel_sub_uuid_from_panel
)
is_active_based_on_panel = panel_status == "ACTIVE" and ( is_active_based_on_panel = panel_status == "ACTIVE" and (
panel_expire_dt > datetime.now(timezone.utc) panel_expire_dt > datetime.now(timezone.utc)
if panel_expire_dt else False) if panel_expire_dt
else False
)
if local_active_sub.is_active != is_active_based_on_panel: if local_active_sub.is_active != is_active_based_on_panel:
update_payload_local['is_active'] = is_active_based_on_panel update_payload_local["is_active"] = is_active_based_on_panel
if update_payload_local: if update_payload_local:
await subscription_dal.update_subscription( await subscription_dal.update_subscription(
session, local_active_sub.subscription_id, session, local_active_sub.subscription_id, update_payload_local
update_payload_local) )
panel_end_date = datetime.fromisoformat( panel_end_date = (
panel_user_data['expireAt'].replace( datetime.fromisoformat(panel_user_data["expireAt"].replace("Z", "+00:00"))
"Z", "+00:00")) if panel_user_data.get('expireAt') else None if panel_user_data.get("expireAt")
else None
)
return { return {
"end_date": "end_date": panel_end_date,
panel_end_date, "status_from_panel": panel_user_data.get("status", "UNKNOWN").upper(),
"status_from_panel": "config_link": panel_user_data.get("subscriptionUrl"),
panel_user_data.get('status', 'UNKNOWN').upper(), "traffic_limit_bytes": panel_user_data.get("trafficLimitBytes"),
"config_link": "traffic_used_bytes": panel_user_data.get("usedTrafficBytes"),
panel_user_data.get('subscriptionUrl') "user_bot_username": db_user.username,
or (await self.panel_service.get_subscription_link( "is_panel_data": True,
panel_user_data['shortUuid'])
if panel_user_data.get('shortUuid') else "N/A"),
"traffic_limit_bytes":
panel_user_data.get('trafficLimitBytes'),
"traffic_used_bytes":
panel_user_data.get('usedTrafficBytes'),
"user_bot_username":
db_user.username,
"is_panel_data":
True
} }
async def get_subscriptions_ending_soon( async def get_subscriptions_ending_soon(
self, session: AsyncSession, self, session: AsyncSession, days_threshold: int
days_threshold: int) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
subs_models_with_users = await subscription_dal.get_subscriptions_near_expiration( subs_models_with_users = (
session, days_threshold) await subscription_dal.get_subscriptions_near_expiration(
session, days_threshold
)
)
results = [] results = []
for sub_model in subs_models_with_users: for sub_model in subs_models_with_users:
if sub_model.user and sub_model.end_date and not sub_model.skip_notifications: if (
days_left = (sub_model.end_date - datetime.now( sub_model.user
timezone.utc)).total_seconds() / (24 * 3600) and sub_model.end_date
results.append({ and not sub_model.skip_notifications
'user_id': ):
sub_model.user_id, days_left = (
'first_name': sub_model.end_date - datetime.now(timezone.utc)
sub_model.user.first_name or f"User {sub_model.user_id}", ).total_seconds() / (24 * 3600)
'language_code': results.append(
sub_model.user.language_code {
or self.settings.DEFAULT_LANGUAGE, "user_id": sub_model.user_id,
'end_date_str': "first_name": sub_model.user.first_name
sub_model.end_date.strftime('%Y-%m-%d'), or f"User {sub_model.user_id}",
'days_left': "language_code": sub_model.user.language_code
max(0, int(round(days_left))), or self.settings.DEFAULT_LANGUAGE,
'subscription_end_date_iso_for_update': "end_date_str": sub_model.end_date.strftime("%Y-%m-%d"),
sub_model.end_date "days_left": max(0, int(round(days_left))),
}) "subscription_end_date_iso_for_update": sub_model.end_date,
}
)
return results return results
async def update_last_notification_sent(self, session: AsyncSession, async def update_last_notification_sent(
user_id: int, self, session: AsyncSession, user_id: int, subscription_end_date: datetime
subscription_end_date: datetime): ):
sub_to_update = await subscription_dal.find_subscription_for_notification_update( sub_to_update = (
session, user_id, subscription_end_date) await subscription_dal.find_subscription_for_notification_update(
session, user_id, subscription_end_date
)
)
if sub_to_update: if sub_to_update:
await subscription_dal.update_subscription_notification_time( await subscription_dal.update_subscription_notification_time(
session, sub_to_update.subscription_id, session, sub_to_update.subscription_id, datetime.now(timezone.utc)
datetime.now(timezone.utc)) )
logging.info( logging.info(
f"Updated last_notification_sent for user {user_id}, sub_id {sub_to_update.subscription_id}" f"Updated last_notification_sent for user {user_id}, sub_id {sub_to_update.subscription_id}"
) )
+23 -13
View File
@@ -70,19 +70,29 @@ class TributeService:
async with async_session_factory() as session: async with async_session_factory() as session:
if event_name == 'new_subscription': if event_name == 'new_subscription':
payment_record = await payment_dal.create_payment_record( provider_payment_id = str(data.get('subscription_id'))
session, existing_payment = await payment_dal.get_payment_by_provider_payment_id(
{ session, provider_payment_id)
'user_id': user_id, if existing_payment:
'amount': float(price_rub), logging.info(
'currency': 'RUB', "Duplicate Tribute payment webhook ignored for provider_payment_id %s",
'status': 'succeeded', provider_payment_id,
'description': 'Tribute subscription', )
'subscription_duration_months': months, payment_record = existing_payment
'provider_payment_id': str(data.get('subscription_id')), else:
'provider': 'tribute', payment_record = await payment_dal.create_payment_record(
}, session,
) {
'user_id': user_id,
'amount': float(price_rub),
'currency': 'RUB',
'status': 'succeeded',
'description': 'Tribute subscription',
'subscription_duration_months': months,
'provider_payment_id': provider_payment_id,
'provider': 'tribute',
},
)
activation_details = await subscription_service.activate_subscription( activation_details = await subscription_service.activate_subscription(
session, session,
user_id, user_id,
+1
View File
@@ -6,6 +6,7 @@ class AdminStates(StatesGroup):
waiting_for_broadcast_message = State() waiting_for_broadcast_message = State()
confirming_broadcast = State() confirming_broadcast = State()
waiting_for_promo_details = State() waiting_for_promo_details = State()
waiting_for_promo_edit_details = State()
waiting_for_user_id_to_ban = State() waiting_for_user_id_to_ban = State()
waiting_for_user_id_to_unban = State() waiting_for_user_id_to_unban = State()
+17 -7
View File
@@ -86,10 +86,9 @@ class Settings(BaseSettings):
PANEL_API_URL: Optional[str] = None PANEL_API_URL: Optional[str] = None
PANEL_API_KEY: Optional[str] = None PANEL_API_KEY: Optional[str] = None
PANEL_USER_DEFAULT_EXPIRE_DAYS: int = Field(default=1) USER_TRAFFIC_LIMIT_GB: Optional[float] = Field(default=0.0)
PANEL_USER_DEFAULT_TRAFFIC_BYTES: int = Field(default=0) USER_TRAFFIC_STRATEGY: str = Field(default="NO_RESET")
PANEL_USER_DEFAULT_TRAFFIC_STRATEGY: str = Field(default="NO_RESET") USER_INBOUND_UUIDS: Optional[str] = Field(
PANEL_USER_DEFAULT_INBOUND_UUIDS: Optional[str] = Field(
default=None, default=None,
description= description=
"Comma-separated UUIDs of inbounds to activate for new panel users") "Comma-separated UUIDs of inbounds to activate for new panel users")
@@ -102,6 +101,10 @@ class Settings(BaseSettings):
WEB_SERVER_PORT: int = Field(default=8080) WEB_SERVER_PORT: int = Field(default=8080)
LOGS_PAGE_SIZE: int = Field(default=10) LOGS_PAGE_SIZE: int = Field(default=10)
SUBSCRIPTION_MINI_APP_URL: Optional[str] = Field(default=None)
START_COMMAND_DESCRIPTION: Optional[str] = Field(default=None)
@computed_field @computed_field
@property @property
def DATABASE_URL(self) -> str: def DATABASE_URL(self) -> str:
@@ -139,11 +142,18 @@ class Settings(BaseSettings):
@computed_field @computed_field
@property @property
def parsed_default_panel_user_inbound_uuids(self) -> Optional[List[str]]: def user_traffic_limit_bytes(self) -> int:
if self.PANEL_USER_DEFAULT_INBOUND_UUIDS: if self.USER_TRAFFIC_LIMIT_GB is None or self.USER_TRAFFIC_LIMIT_GB <= 0:
return 0
return int(self.USER_TRAFFIC_LIMIT_GB * (1024**3))
@computed_field
@property
def parsed_user_inbound_uuids(self) -> Optional[List[str]]:
if self.USER_INBOUND_UUIDS:
return [ return [
uuid.strip() uuid.strip()
for uuid in self.PANEL_USER_DEFAULT_INBOUND_UUIDS.split(',') for uuid in self.USER_INBOUND_UUIDS.split(',')
if uuid.strip() if uuid.strip()
] ]
return None return None
+22
View File
@@ -46,6 +46,15 @@ async def get_payment_by_yookassa_id(
return result.scalar_one_or_none() return result.scalar_one_or_none()
async def get_payment_by_provider_payment_id(
session: AsyncSession, provider_payment_id: str) -> Optional[Payment]:
"""Fetch a payment by provider-specific identifier."""
stmt = select(Payment).where(
Payment.provider_payment_id == provider_payment_id)
result = await session.execute(stmt)
return result.scalar_one_or_none()
async def get_payment_by_db_id(session: AsyncSession, async def get_payment_by_db_id(session: AsyncSession,
payment_db_id: int) -> Optional[Payment]: payment_db_id: int) -> Optional[Payment]:
@@ -87,6 +96,19 @@ async def update_payment_status_by_db_id(
return payment return payment
async def user_has_successful_payment_for_provider(
session: AsyncSession, user_id: int, provider: str) -> bool:
"""Check if a user has at least one successful payment for the provider."""
stmt = (select(Payment.payment_id)
.where(Payment.user_id == user_id,
Payment.provider == provider,
Payment.status == 'succeeded')
.limit(1))
result = await session.execute(stmt)
return result.scalar_one_or_none() is not None
async def update_payment_status_by_yk_id(session: AsyncSession, async def update_payment_status_by_yk_id(session: AsyncSession,
yookassa_payment_id: str, yookassa_payment_id: str,
new_status: str) -> Optional[Payment]: new_status: str) -> Optional[Payment]:
+21
View File
@@ -49,6 +49,27 @@ async def get_all_active_promo_codes(session: AsyncSession,
return result.scalars().all() return result.scalars().all()
async def update_promo_code(session: AsyncSession, promo_id: int,
update_data: Dict[str, Any]) -> Optional[PromoCode]:
promo = await get_promo_code_by_id(session, promo_id)
if not promo:
return None
for key, value in update_data.items():
setattr(promo, key, value)
await session.flush()
await session.refresh(promo)
return promo
async def delete_promo_code(session: AsyncSession, promo_id: int) -> Optional[PromoCode]:
promo = await get_promo_code_by_id(session, promo_id)
if not promo:
return None
await session.delete(promo)
await session.flush()
return promo
async def increment_promo_code_usage( async def increment_promo_code_usage(
session: AsyncSession, promo_code_id: int) -> Optional[PromoCode]: session: AsyncSession, promo_code_id: int) -> Optional[PromoCode]:
promo = await get_promo_code_by_id(session, promo_code_id) promo = await get_promo_code_by_id(session, promo_code_id)
+37
View File
@@ -129,6 +129,21 @@ async def deactivate_other_active_subscriptions(
) )
async def deactivate_all_user_subscriptions(
session: AsyncSession, user_id: int) -> int:
stmt = (
update(Subscription)
.where(Subscription.user_id == user_id, Subscription.is_active == True)
.values(is_active=False, status_from_panel="INACTIVE_USER_NOT_FOUND")
)
result = await session.execute(stmt)
if result.rowcount > 0:
logging.info(
f"Deactivated {result.rowcount} subscriptions for user {user_id} due to missing panel user."
)
return result.rowcount
async def update_subscription_end_date( async def update_subscription_end_date(
session: AsyncSession, subscription_id: int, session: AsyncSession, subscription_id: int,
new_end_date: datetime) -> Optional[Subscription]: new_end_date: datetime) -> Optional[Subscription]:
@@ -216,3 +231,25 @@ async def set_skip_notifications_for_provider(
Subscription.provider == provider).values(skip_notifications=skip)) Subscription.provider == provider).values(skip_notifications=skip))
result = await session.execute(stmt) result = await session.execute(stmt)
return result.rowcount return result.rowcount
async def get_active_subscriptions_for_autorenew(
session: AsyncSession, provider: str,
days_threshold: int = 1,
require_skip_flag: bool = True) -> List[Subscription]:
"""Fetch active subscriptions nearing expiration for auto-renew logic."""
now_utc = datetime.now(timezone.utc)
threshold_date = now_utc + timedelta(days=days_threshold)
conditions = [
Subscription.provider == provider,
Subscription.is_active == True,
Subscription.end_date <= threshold_date,
]
if require_skip_flag:
conditions.append(Subscription.skip_notifications == True)
stmt = select(Subscription).where(*conditions)
result = await session.execute(stmt)
return result.scalars().all()
+66 -39
View File
@@ -9,33 +9,47 @@ from datetime import datetime
from ..models import User, Subscription from ..models import User, Subscription
async def get_user_by_id(session: AsyncSession, async def get_user_by_id(session: AsyncSession, user_id: int) -> Optional[User]:
user_id: int) -> Optional[User]:
stmt = select(User).where(User.user_id == user_id) stmt = select(User).where(User.user_id == user_id)
result = await session.execute(stmt) result = await session.execute(stmt)
return result.scalar_one_or_none() return result.scalar_one_or_none()
async def get_user_by_username(session: AsyncSession, async def get_user_by_username(session: AsyncSession, username: str) -> Optional[User]:
username: str) -> Optional[User]: clean_username = username.lstrip("@").lower()
clean_username = username.lstrip('@').lower()
stmt = select(User).where(func.lower(User.username) == clean_username) stmt = select(User).where(func.lower(User.username) == clean_username)
result = await session.execute(stmt) result = await session.execute(stmt)
return result.scalar_one_or_none() return result.scalar_one_or_none()
async def get_user_by_panel_uuid(session: AsyncSession, async def get_user_by_panel_uuid(
panel_uuid: str) -> Optional[User]: session: AsyncSession, panel_uuid: str
) -> Optional[User]:
stmt = select(User).where(User.panel_user_uuid == panel_uuid) stmt = select(User).where(User.panel_user_uuid == panel_uuid)
result = await session.execute(stmt) result = await session.execute(stmt)
return result.scalar_one_or_none() return result.scalar_one_or_none()
async def create_user(session: AsyncSession, user_data: Dict[str, async def get_user(
Any]) -> User: session: AsyncSession,
*,
user_id: Optional[int] = None,
username: Optional[str] = None,
panel_uuid: Optional[str] = None,
) -> Optional[User]:
if user_id is not None:
return await get_user_by_id(session, user_id)
if username is not None:
return await get_user_by_username(session, username)
if panel_uuid is not None:
return await get_user_by_panel_uuid(session, panel_uuid)
return None
if 'registration_date' not in user_data:
user_data['registration_date'] = datetime.now() async def create_user(session: AsyncSession, user_data: Dict[str, Any]) -> User:
if "registration_date" not in user_data:
user_data["registration_date"] = datetime.now()
new_user = User(**user_data) new_user = User(**user_data)
session.add(new_user) session.add(new_user)
@@ -47,8 +61,9 @@ async def create_user(session: AsyncSession, user_data: Dict[str,
return new_user return new_user
async def update_user(session: AsyncSession, user_id: int, async def update_user(
update_data: Dict[str, Any]) -> Optional[User]: session: AsyncSession, user_id: int, update_data: Dict[str, Any]
) -> Optional[User]:
user = await get_user_by_id(session, user_id) user = await get_user_by_id(session, user_id)
if user: if user:
for key, value in update_data.items(): for key, value in update_data.items():
@@ -58,16 +73,17 @@ async def update_user(session: AsyncSession, user_id: int,
return user return user
async def update_user_language(session: AsyncSession, user_id: int, async def update_user_language(
lang_code: str) -> bool: session: AsyncSession, user_id: int, lang_code: str
stmt = update(User).where(User.user_id == user_id).values( ) -> bool:
language_code=lang_code) stmt = update(User).where(User.user_id == user_id).values(language_code=lang_code)
result = await session.execute(stmt) result = await session.execute(stmt)
return result.rowcount > 0 return result.rowcount > 0
async def set_user_ban_status(session: AsyncSession, user_id: int, async def set_user_ban_status(
is_banned: bool) -> bool: session: AsyncSession, user_id: int, is_banned: bool
) -> bool:
user = await get_user_by_id(session, user_id) user = await get_user_by_id(session, user_id)
if user: if user:
user.is_banned = is_banned user.is_banned = is_banned
@@ -77,23 +93,27 @@ async def set_user_ban_status(session: AsyncSession, user_id: int,
return False return False
async def get_banned_users_paginated(session: AsyncSession, limit: int, async def get_banned_users_paginated(
offset: int) -> Tuple[List[User], int]: session: AsyncSession, limit: int, offset: int
stmt_users = select(User).where(User.is_banned == True).order_by( ) -> Tuple[List[User], int]:
User.registration_date.desc()).limit(limit).offset(offset) stmt_users = (
select(User)
.where(User.is_banned == True)
.order_by(User.registration_date.desc())
.limit(limit)
.offset(offset)
)
result_users = await session.execute(stmt_users) result_users = await session.execute(stmt_users)
users_list = result_users.scalars().all() users_list = result_users.scalars().all()
stmt_count = select( stmt_count = select(func.count()).select_from(User).where(User.is_banned == True)
func.count()).select_from(User).where(User.is_banned == True)
result_count = await session.execute(stmt_count) result_count = await session.execute(stmt_count)
total_banned = result_count.scalar_one() total_banned = result_count.scalar_one()
return users_list, total_banned return users_list, total_banned
async def get_all_active_user_ids_for_broadcast( async def get_all_active_user_ids_for_broadcast(session: AsyncSession) -> List[int]:
session: AsyncSession) -> List[int]:
stmt = select(User.user_id).where(User.is_banned == False) stmt = select(User.user_id).where(User.is_banned == False)
result = await session.execute(stmt) result = await session.execute(stmt)
return result.scalars().all() return result.scalars().all()
@@ -101,24 +121,31 @@ async def get_all_active_user_ids_for_broadcast(
async def get_user_count_stats_dal(session: AsyncSession) -> Dict[str, int]: async def get_user_count_stats_dal(session: AsyncSession) -> Dict[str, int]:
total_users_stmt = select(func.count(User.user_id)).select_from(User) total_users_stmt = select(func.count(User.user_id)).select_from(User)
banned_users_stmt = select(func.count( banned_users_stmt = (
User.user_id)).select_from(User).where(User.is_banned == True) select(func.count(User.user_id)).select_from(User).where(User.is_banned == True)
)
active_subs_stmt = (select(func.count( active_subs_stmt = (
func.distinct(Subscription.user_id))).join( select(func.count(func.distinct(Subscription.user_id)))
User, Subscription.user_id == User.user_id).where( .join(User, Subscription.user_id == User.user_id)
Subscription.is_active == True).where( .where(Subscription.is_active == True)
Subscription.end_date > datetime.now())) .where(Subscription.end_date > datetime.now())
)
total_users = (await total_users = (await session.execute(total_users_stmt)).scalar_one_or_none() or 0
session.execute(total_users_stmt)).scalar_one_or_none() or 0 banned_users = (await session.execute(banned_users_stmt)).scalar_one_or_none() or 0
banned_users = (
await session.execute(banned_users_stmt)).scalar_one_or_none() or 0
active_subs_users = ( active_subs_users = (
await session.execute(active_subs_stmt)).scalar_one_or_none() or 0 await session.execute(active_subs_stmt)
).scalar_one_or_none() or 0
return { return {
"total_users": total_users, "total_users": total_users,
"banned_users": banned_users, "banned_users": banned_users,
"users_with_active_subscriptions": active_subs_users, "users_with_active_subscriptions": active_subs_users,
} }
async def get_all_users_with_panel_uuid(session: AsyncSession) -> List[User]:
stmt = select(User).where(User.panel_user_uuid.is_not(None))
result = await session.execute(stmt)
return result.scalars().all()
+12
View File
@@ -90,6 +90,7 @@
"admin_stats_button": "📊 Statistics", "admin_stats_button": "📊 Statistics",
"admin_broadcast_button": "📢 Broadcast", "admin_broadcast_button": "📢 Broadcast",
"admin_create_promo_button": "🎁 Create Promo", "admin_create_promo_button": "🎁 Create Promo",
"admin_manage_promos_button": "🛠 Manage Promos",
"admin_view_promos_button": "👀 Promo List", "admin_view_promos_button": "👀 Promo List",
"admin_ban_user_button": "🚫 Ban User", "admin_ban_user_button": "🚫 Ban User",
"admin_unban_user_button": "✅ Unban User", "admin_unban_user_button": "✅ Unban User",
@@ -141,6 +142,13 @@
"admin_promo_list_item": "<code>{code}</code>: +{bonus}d, {current}/{max} used, valid until {valid_until}", "admin_promo_list_item": "<code>{code}</code>: +{bonus}d, {current}/{max} used, valid until {valid_until}",
"admin_promo_valid_indefinitely": "indefinite", "admin_promo_valid_indefinitely": "indefinite",
"admin_promo_valid_until_display": "until {date}", "admin_promo_valid_until_display": "until {date}",
"admin_manage_promos_title": "Manage Promo Codes",
"admin_promo_edit_button": "✏️ Edit",
"admin_promo_delete_button": "🗑 Delete",
"admin_promo_edit_prompt": "Send new details for <code>{code}</code> in format: CODE BONUS_DAYS MAX_USES [VALIDITY_DAYS]",
"admin_promo_updated_success": "Promo <code>{code}</code> updated.",
"admin_promo_deleted_success": "Promo <code>{code}</code> deleted.",
"admin_promo_not_found": "Promo not found.",
"admin_ban_user_prompt": "Enter user ID or @username to ban:", "admin_ban_user_prompt": "Enter user ID or @username to ban:",
"admin_user_not_found_in_bot_db": "User <code>{user_id}</code> not found in bot database.", "admin_user_not_found_in_bot_db": "User <code>{user_id}</code> not found in bot database.",
@@ -198,6 +206,7 @@
"sync_completed_with_errors_details": "Checked: {total_checked} entries.\nUsers synced/updated: {users_synced}.\nSubscriptions synced/updated: {subs_synced}.\nErrors: {errors_count}.\n\nFirst errors:\n{error_details_preview}", "sync_completed_with_errors_details": "Checked: {total_checked} entries.\nUsers synced/updated: {users_synced}.\nSubscriptions synced/updated: {subs_synced}.\nErrors: {errors_count}.\n\nFirst errors:\n{error_details_preview}",
"no_errors_placeholder": "none", "no_errors_placeholder": "none",
"admin_sync_initiated_from_panel": "Sync initiated...", "admin_sync_initiated_from_panel": "Sync initiated...",
"admin_panel_user_creation_failed": "❌ Failed to create panel user for TG ID {user_id}. Panel unreachable?",
"error_displaying_logs_too_long": "Error: logs too long to display in one message. Try viewing logs for a specific user.", "error_displaying_logs_too_long": "Error: logs too long to display in one message. Try viewing logs for a specific user.",
"error_displaying_statistics": "Error displaying statistics.", "error_displaying_statistics": "Error displaying statistics.",
"stub_page_display": "Page", "stub_page_display": "Page",
@@ -205,5 +214,8 @@
"subscription_ending_soon_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends on {end_date} (in {days_left} days).\n\nTo avoid interruption, please renew it in the main menu.", "subscription_ending_soon_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends on {end_date} (in {days_left} days).\n\nTo avoid interruption, please renew it in the main menu.",
"subscription_cancelled_notification": "Your recurring subscription was cancelled. You will keep access until the paid period ends.", "subscription_cancelled_notification": "Your recurring subscription was cancelled. You will keep access until the paid period ends.",
"admin_new_trial_notification": "\ud83c\udf21 User {user_id} activated a free trial until {end_date}.",
"admin_new_payment_notification": "\ud83d\udcb3 Payment received from user {user_id}: {months} mo. for {amount} {currency}.",
"error_unknown": "An unknown error occurred." "error_unknown": "An unknown error occurred."
} }
+12
View File
@@ -90,6 +90,7 @@
"admin_stats_button": "📊 Статистика", "admin_stats_button": "📊 Статистика",
"admin_broadcast_button": "📢 Рассылка", "admin_broadcast_button": "📢 Рассылка",
"admin_create_promo_button": "🎁 Создать промо", "admin_create_promo_button": "🎁 Создать промо",
"admin_manage_promos_button": "🛠 Управление промо",
"admin_view_promos_button": "👀 Список промо", "admin_view_promos_button": "👀 Список промо",
"admin_ban_user_button": "🚫 Забанить", "admin_ban_user_button": "🚫 Забанить",
"admin_unban_user_button": "✅ Разбанить", "admin_unban_user_button": "✅ Разбанить",
@@ -141,6 +142,13 @@
"admin_promo_list_item": "<code>{code}</code>: +{bonus}дн, {current}/{max} акт., до {valid_until}", "admin_promo_list_item": "<code>{code}</code>: +{bonus}дн, {current}/{max} акт., до {valid_until}",
"admin_promo_valid_indefinitely": "бессрочно", "admin_promo_valid_indefinitely": "бессрочно",
"admin_promo_valid_until_display": "до {date}", "admin_promo_valid_until_display": "до {date}",
"admin_manage_promos_title": "Управление промокодами",
"admin_promo_edit_button": "✏️ Изменить",
"admin_promo_delete_button": "🗑 Удалить",
"admin_promo_edit_prompt": "Отправьте новые данные для <code>{code}</code> в формате: КОД ДНИ_БОНУСА МАКС_АКТИВАЦИЙ [СРОК]",
"admin_promo_updated_success": "Промокод <code>{code}</code> обновлен.",
"admin_promo_deleted_success": "Промокод <code>{code}</code> удален.",
"admin_promo_not_found": "Промокод не найден.",
"admin_ban_user_prompt": "Введите ID или @username пользователя для блокировки:", "admin_ban_user_prompt": "Введите ID или @username пользователя для блокировки:",
"admin_user_not_found_in_bot_db": "Пользователь <code>{user_id}</code> не найден в базе данных бота.", "admin_user_not_found_in_bot_db": "Пользователь <code>{user_id}</code> не найден в базе данных бота.",
@@ -198,6 +206,7 @@
"sync_completed_with_errors_details": "Проверено: {total_checked} записей.\nПользователей синхронизировано/обновлено: {users_synced}.\nПодписок синхронизировано/обновлено: {subs_synced}.\nОшибок: {errors_count}.\n\nПервые ошибки:\n{error_details_preview}", "sync_completed_with_errors_details": "Проверено: {total_checked} записей.\nПользователей синхронизировано/обновлено: {users_synced}.\nПодписок синхронизировано/обновлено: {subs_synced}.\nОшибок: {errors_count}.\n\nПервые ошибки:\n{error_details_preview}",
"no_errors_placeholder": "нет", "no_errors_placeholder": "нет",
"admin_sync_initiated_from_panel": "Синхронизация запущена...", "admin_sync_initiated_from_panel": "Синхронизация запущена...",
"admin_panel_user_creation_failed": "❌ Не удалось создать пользователя на панели для TG ID {user_id}. Панель недоступна?",
"error_displaying_logs_too_long": "Ошибка: логи слишком длинные для отображения одним сообщением. Попробуйте найти логи по конкретному пользователю.", "error_displaying_logs_too_long": "Ошибка: логи слишком длинные для отображения одним сообщением. Попробуйте найти логи по конкретному пользователю.",
"error_displaying_statistics": "Ошибка отображения статистики.", "error_displaying_statistics": "Ошибка отображения статистики.",
"stub_page_display": "Страница", "stub_page_display": "Страница",
@@ -205,5 +214,8 @@
"subscription_ending_soon_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает {end_date} (через {days_left} дн.).\n\nЧтобы не потерять доступ, пожалуйста, продлите ее заранее в главном меню бота.", "subscription_ending_soon_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает {end_date} (через {days_left} дн.).\n\nЧтобы не потерять доступ, пожалуйста, продлите ее заранее в главном меню бота.",
"subscription_cancelled_notification": "Ваша подписка отменена. Доступ сохранится до конца оплаченного периода.", "subscription_cancelled_notification": "Ваша подписка отменена. Доступ сохранится до конца оплаченного периода.",
"admin_new_trial_notification": "\ud83c\udf21 Пользователь {user_id} активировал пробный период до {end_date}.",
"admin_new_payment_notification": "\ud83d\udcb3 Получен платеж от пользователя {user_id}: {months} мес. за {amount} {currency}.",
"error_unknown": "Произошла неизвестная ошибка." "error_unknown": "Произошла неизвестная ошибка."
} }