diff --git a/.env.example b/.env.example index 4ecfe43..29816ff 100644 --- a/.env.example +++ b/.env.example @@ -61,10 +61,11 @@ TRIBUTE_LINK_12_MONTHS= # API key for verifying Tribute webhook signatures TRIBUTE_API_KEY= -# Subscription Expiration Notifications -SUBSCRIPTION_EXPIRATION_NOTIFICATION_DAYS=7 -SUBSCRIPTION_NOTIFICATION_HOUR_UTC=13 -SUBSCRIPTION_NOTIFICATION_MINUTE_UTC=0 +# Subscription Notifications +SUBSCRIPTION_NOTIFICATIONS_ENABLED=True +SUBSCRIPTION_NOTIFY_ON_EXPIRE=True +SUBSCRIPTION_NOTIFY_AFTER_EXPIRE=True +SUBSCRIPTION_NOTIFY_DAYS_BEFORE=3 # Referral Bonus Days REFERRAL_BONUS_DAYS_1_MONTH=3 @@ -79,6 +80,7 @@ REFEREE_BONUS_DAYS_12_MONTHS=15 # Panel API Configuration PANEL_API_URL=http://your_panel_api_url/api PANEL_API_KEY=your_panel_api_key +PANEL_WEBHOOK_SECRET= # secret used to verify panel webhook signatures # User traffic limits (applied for all users) # 0 means unlimited diff --git a/README.md b/README.md index 9b168c7..9ef64e2 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,7 @@ This Telegram bot is designed to automate the sale and management of subscriptio * **Panel API Settings:** * `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_WEBHOOK_SECRET`: Secret key for verifying webhooks from the Remnawave panel. * `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. diff --git a/bot/main_bot.py b/bot/main_bot.py index a2f799b..1e33528 100644 --- a/bot/main_bot.py +++ b/bot/main_bot.py @@ -16,7 +16,7 @@ from aiogram.client.default import DefaultBotProperties from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application from aiogram.fsm.storage.memory import MemoryStorage from aiohttp import web -from apscheduler.schedulers.asyncio import AsyncIOScheduler +from bot.services.panel_webhook_service import PanelWebhookService, panel_webhook_route from sqlalchemy.orm import sessionmaker from config.settings import Settings @@ -31,7 +31,6 @@ from bot.handlers.user import user_router_aggregate from bot.handlers.admin import admin_router_aggregate from bot.filters.admin_filter import AdminFilter -from bot.services.notification_service import schedule_subscription_notifications from bot.services.yookassa_service import YooKassaService from bot.services.panel_api_service import PanelApiService from bot.services.subscription_service import SubscriptionService @@ -101,26 +100,6 @@ async def on_startup_configured(dispatcher: Dispatcher): logging.info("STARTUP: on_startup_configured executing...") - existing_scheduler: Optional[AsyncIOScheduler] = dispatcher.get("scheduler") - - if existing_scheduler and existing_scheduler.running: - logging.warning("STARTUP: Scheduler already running, skipping initialization.") - else: - scheduler = AsyncIOScheduler(timezone="UTC") - try: - await schedule_subscription_notifications( - bot, - settings, - i18n_instance, - scheduler, - panel_service, - async_session_factory, - ) - scheduler.start() - dispatcher["scheduler"] = scheduler - logging.info("STARTUP: APScheduler started.") - except Exception as e: - logging.error(f"STARTUP: Failed to start APScheduler: {e}", exc_info=True) telegram_webhook_url_to_set = getattr(settings, "TELEGRAM_WEBHOOK_BASE_URL", None) if telegram_webhook_url_to_set: @@ -213,15 +192,6 @@ async def on_startup_configured(dispatcher: Dispatcher): async def on_shutdown_configured(dispatcher: Dispatcher): logging.warning("SHUTDOWN: on_shutdown_configured executing...") - scheduler: Optional[AsyncIOScheduler] = dispatcher.get("scheduler") - if scheduler and scheduler.running: - try: - scheduler.shutdown(wait=False) - logging.info("SHUTDOWN: APScheduler shut down.") - except Exception as e: - logging.error( - f"SHUTDOWN: Error shutting down APScheduler: {e}", exc_info=True - ) panel_service: Optional[PanelApiService] = dispatcher.get("panel_service") if panel_service and hasattr(panel_service, "close_session"): @@ -304,6 +274,12 @@ async def run_bot(settings_param: Settings): subscription_service, referral_service, ) + panel_webhook_service = PanelWebhookService( + bot, + settings_param, + i18n_instance, + local_async_session_factory, + ) dp["i18n_instance"] = i18n_instance dp["yookassa_service"] = yookassa_service @@ -313,6 +289,7 @@ async def run_bot(settings_param: Settings): dp["promo_code_service"] = promo_code_service dp["stars_service"] = stars_service dp["tribute_service"] = tribute_service + dp["panel_webhook_service"] = panel_webhook_service dp["async_session_factory"] = local_async_session_factory dp.update.outer_middleware(DBSessionMiddleware(local_async_session_factory)) @@ -367,6 +344,7 @@ async def run_bot(settings_param: Settings): app["panel_service"] = panel_service app["stars_service"] = stars_service app["tribute_service"] = tribute_service + app["panel_webhook_service"] = panel_webhook_service setup_application(app, dp, bot=bot) @@ -402,6 +380,11 @@ async def run_bot(settings_param: Settings): app.router.add_post(tribute_path, tribute_webhook_route) logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}") + panel_path = settings_param.panel_webhook_path + if panel_path.startswith("/"): + app.router.add_post(panel_path, panel_webhook_route) + logging.info(f"Panel webhook route configured at: [POST] {panel_path}") + web_app_runner = web.AppRunner(app) await web_app_runner.setup() site = web.TCPSite( diff --git a/bot/services/notification_service.py b/bot/services/notification_service.py index d2162ca..37cc534 100644 --- a/bot/services/notification_service.py +++ b/bot/services/notification_service.py @@ -2,138 +2,12 @@ import logging import asyncio from aiogram import Bot from aiogram.utils.text_decorations import html_decoration as hd -from apscheduler.schedulers.asyncio import AsyncIOScheduler from datetime import datetime, timezone from config.settings import Settings from sqlalchemy.orm import sessionmaker from bot.middlewares.i18n import JsonI18n -from bot.services.panel_api_service import PanelApiService -from bot.services.subscription_service import SubscriptionService - - -async def send_expiration_warnings(bot: Bot, settings: Settings, - i18n: JsonI18n, - panel_service: PanelApiService, - async_session_factory: sessionmaker): - - logging.info( - f"Scheduler job 'send_expiration_warnings' started at {datetime.now(timezone.utc)} UTC." - ) - - if async_session_factory is None: - logging.error( - "NotificationService: AsyncSessionFactory not provided to send_expiration_warnings!" - ) - return - - async with async_session_factory() as session: - try: - - sub_service = SubscriptionService(settings, panel_service) - - expiring_subs_details_list = await sub_service.get_subscriptions_ending_soon( - session, settings.SUBSCRIPTION_EXPIRATION_NOTIFICATION_DAYS) - - if not expiring_subs_details_list: - logging.info( - "No subscriptions found ending soon for notification.") - return - - logging.info( - f"Found {len(expiring_subs_details_list)} subscriptions for expiration warning." - ) - - for sub_details in expiring_subs_details_list: - user_id = sub_details['user_id'] - user_lang = sub_details.get('language_code', - settings.DEFAULT_LANGUAGE) - first_name = hd.quote(sub_details.get('first_name', f"User {user_id}")) - end_date_str_for_msg = sub_details.get('end_date_str', "N/A") - days_left_display = sub_details.get('days_left', "N/A") - - subscription_actual_end_date_obj: Optional[ - datetime] = sub_details.get( - 'subscription_end_date_iso_for_update') - - _ = lambda key, **kwargs: i18n.gettext(user_lang, key, **kwargs - ) - message_text = _("subscription_ending_soon_notification", - user_name=first_name, - end_date=end_date_str_for_msg, - days_left=days_left_display) - try: - await bot.send_message(user_id, message_text) - logging.info( - f"Sent expiration warning to user {user_id} for subscription ending {end_date_str_for_msg}." - ) - - if subscription_actual_end_date_obj: - await sub_service.update_last_notification_sent( - session, user_id, subscription_actual_end_date_obj) - else: - logging.warning( - f"Could not find exact subscription end_date_obj for user {user_id} to update notification time." - ) - - except Exception as e: - logging.error( - f"Failed to send expiration warning or update notification status for user {user_id}: {e}", - exc_info=True) - - await asyncio.sleep(0.1) - - await session.commit() - logging.info( - "Finished processing expiration warnings. Session committed.") - - except Exception as e_session: - logging.error( - f"Error during send_expiration_warnings session: {e_session}", - exc_info=True) - await session.rollback() - logging.info( - "Session rolled back due to error in send_expiration_warnings." - ) - - -async def schedule_subscription_notifications( - bot: Bot, settings: Settings, i18n: JsonI18n, - scheduler: AsyncIOScheduler, panel_service: PanelApiService, - async_session_factory: sessionmaker): - - async def job_wrapper(): - - try: - await send_expiration_warnings(bot, settings, i18n, panel_service, - async_session_factory) - except Exception as e: - logging.error( - f"Unhandled error in scheduled job 'send_expiration_warnings' (job_wrapper): {e}", - exc_info=True) - - try: - notification_hour = int(settings.SUBSCRIPTION_NOTIFICATION_HOUR_UTC) - notification_minute = int( - settings.SUBSCRIPTION_NOTIFICATION_MINUTE_UTC) - except (ValueError, TypeError): - logging.warning( - "SUBSCRIPTION_NOTIFICATION_HOUR_UTC or MINUTE_UTC is invalid in settings. Defaulting to 9:00 UTC." - ) - notification_hour = 9 - notification_minute = 0 - - scheduler.add_job(job_wrapper, - 'cron', - hour=notification_hour, - minute=notification_minute, - name="daily_subscription_expiration_warnings_v2", - misfire_grace_time=60 * 15, - replace_existing=True) - logging.info( - 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, diff --git a/bot/services/panel_webhook_service.py b/bot/services/panel_webhook_service.py new file mode 100644 index 0000000..612f536 --- /dev/null +++ b/bot/services/panel_webhook_service.py @@ -0,0 +1,92 @@ +import json +import logging +import hmac +import hashlib +from aiohttp import web +from aiogram import Bot +from sqlalchemy.orm import sessionmaker +from typing import Optional +from config.settings import Settings +from bot.middlewares.i18n import JsonI18n +from db.dal import user_dal + +EVENT_MAP = { + "user.expires_in_72_hours": (3, "subscription_72h_notification"), + "user.expires_in_48_hours": (2, "subscription_48h_notification"), + "user.expires_in_24_hours": (1, "subscription_24h_notification"), +} + +class PanelWebhookService: + def __init__(self, bot: Bot, settings: Settings, i18n: JsonI18n, async_session_factory: sessionmaker): + self.bot = bot + self.settings = settings + self.i18n = i18n + self.async_session_factory = async_session_factory + + async def _send_message(self, user_id: int, lang: str, message_key: str, **kwargs): + _ = lambda k, **kw: self.i18n.gettext(lang, k, **kw) + try: + await self.bot.send_message(user_id, _(message_key, **kwargs)) + except Exception as e: + logging.error(f"Failed to send notification to {user_id}: {e}") + + async def handle_event(self, event_name: str, user_payload: dict): + telegram_id = user_payload.get("telegramId") + if not telegram_id: + logging.warning("Panel webhook without telegramId received") + return + user_id = int(telegram_id) + + if not self.settings.SUBSCRIPTION_NOTIFICATIONS_ENABLED: + return + + async with self.async_session_factory() as session: + db_user = await user_dal.get_user_by_id(session, user_id) + lang = db_user.language_code if db_user and db_user.language_code else self.settings.DEFAULT_LANGUAGE + first_name = db_user.first_name or f"User {user_id}" if db_user else f"User {user_id}" + + if event_name in EVENT_MAP: + days_left, msg_key = EVENT_MAP[event_name] + if days_left == self.settings.SUBSCRIPTION_NOTIFY_DAYS_BEFORE: + await self._send_message( + user_id, + lang, + msg_key, + user_name=first_name, + end_date=user_payload.get("expireAt", "")[:10], + ) + elif event_name == "user.expired" and self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE: + await self._send_message(user_id, lang, "subscription_expired_notification") + elif event_name == "user.expired_24_hours_ago" and self.settings.SUBSCRIPTION_NOTIFY_AFTER_EXPIRE: + await self._send_message(user_id, lang, "subscription_expired_yesterday_notification") + + async def handle_webhook(self, raw_body: bytes, signature_header: Optional[str]) -> web.Response: + if self.settings.PANEL_WEBHOOK_SECRET: + if not signature_header: + return web.Response(status=403, text="no_signature") + expected_sig = hmac.new( + self.settings.PANEL_WEBHOOK_SECRET.encode(), + raw_body, + hashlib.sha256, + ).hexdigest() + if not hmac.compare_digest(expected_sig, signature_header): + return web.Response(status=403, text="invalid_signature") + + try: + payload = json.loads(raw_body.decode()) + except Exception: + return web.Response(status=400, text="bad_request") + + event_name = payload.get("name") or payload.get("event") + user_data = payload.get("payload", {}).get("user") or payload.get("payload", {}) + if not event_name: + return web.Response(status=200, text="ok_no_event") + + await self.handle_event(event_name, user_data) + return web.Response(status=200, text="ok") + +async def panel_webhook_route(request: web.Request): + service: PanelWebhookService = request.app["panel_webhook_service"] + raw = await request.read() + signature_header = request.headers.get("X-Remnawave-Signature") + return await service.handle_webhook(raw, signature_header) diff --git a/bot/services/tribute_service.py b/bot/services/tribute_service.py index b4d8225..4424878 100644 --- a/bot/services/tribute_service.py +++ b/bot/services/tribute_service.py @@ -76,7 +76,11 @@ class TributeService: event_name = payload.get('name') data = payload.get('payload', {}) user_id = data.get('telegram_user_id') - price_val = data.get('price') + price_val = ( + data.get('amount') + or data.get('amount_paid') + or data.get('price') + ) if not user_id or price_val is None: return web.Response(status=200, text="ok_missing_fields") @@ -168,18 +172,6 @@ class TributeService: float(price_rub), currency="RUB", ) - elif event_name == 'cancelled_subscription': - db_user = await user_dal.get_user_by_id(session, user_id) - lang = db_user.language_code if db_user and db_user.language_code else settings.DEFAULT_LANGUAGE - _ = lambda k, **kw: i18n.gettext(lang, k, **kw) - try: - await bot.send_message(user_id, _("subscription_cancelled_notification")) - except Exception as e: - logging.warning( - f"Failed to notify user {user_id} about cancellation: {e}") - await subscription_dal.set_skip_notifications_for_provider( - session, user_id, 'tribute', False) - await session.commit() else: await session.commit() return web.Response(status=200, text="ok") diff --git a/config/settings.py b/config/settings.py index 0429736..5a15223 100644 --- a/config/settings.py +++ b/config/settings.py @@ -61,10 +61,12 @@ class Settings(BaseSettings): TRIBUTE_LINK_6_MONTHS: Optional[str] = Field(default=None) TRIBUTE_LINK_12_MONTHS: Optional[str] = Field(default=None) TRIBUTE_API_KEY: Optional[str] = Field(default=None) + PANEL_WEBHOOK_SECRET: Optional[str] = Field(default=None) - SUBSCRIPTION_EXPIRATION_NOTIFICATION_DAYS: int = Field(default=7) - SUBSCRIPTION_NOTIFICATION_HOUR_UTC: int = Field(default=9) - SUBSCRIPTION_NOTIFICATION_MINUTE_UTC: int = Field(default=0) + SUBSCRIPTION_NOTIFICATIONS_ENABLED: bool = Field(default=True) + SUBSCRIPTION_NOTIFY_ON_EXPIRE: bool = Field(default=True) + SUBSCRIPTION_NOTIFY_AFTER_EXPIRE: bool = Field(default=True) + SUBSCRIPTION_NOTIFY_DAYS_BEFORE: int = Field(default=3) REFERRAL_BONUS_DAYS_INVITER_1_MONTH: Optional[int] = Field( default=3, alias="REFERRAL_BONUS_DAYS_1_MONTH") @@ -184,6 +186,18 @@ class Settings(BaseSettings): return f"{self.YOOKASSA_WEBHOOK_BASE_URL.rstrip('/')}{self.tribute_webhook_path}" return None + @computed_field + @property + def panel_webhook_path(self) -> str: + return "/webhook/panel" + + @computed_field + @property + def panel_full_webhook_url(self) -> Optional[str]: + if self.YOOKASSA_WEBHOOK_BASE_URL: + return f"{self.YOOKASSA_WEBHOOK_BASE_URL.rstrip('/')}{self.panel_webhook_path}" + return None + @computed_field @property def subscription_options(self) -> Dict[int, float]: diff --git a/locales/en.json b/locales/en.json index c363d6c..35ad676 100644 --- a/locales/en.json +++ b/locales/en.json @@ -211,8 +211,11 @@ "error_displaying_statistics": "Error displaying statistics.", "stub_page_display": "Page", - "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_72h_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends in 72 hours on {end_date}.\n\nPlease renew it in the main menu.", + "subscription_48h_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends in 48 hours on {end_date}.\n\nPlease renew it in the main menu.", + "subscription_24h_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends in 24 hours on {end_date}.\n\nPlease renew it in the main menu.", + "subscription_expired_notification": "Your VPN subscription has expired.", + "subscription_expired_yesterday_notification": "Your VPN subscription expired 24 hours ago.", "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}.", diff --git a/locales/ru.json b/locales/ru.json index 5182788..85cb240 100644 --- a/locales/ru.json +++ b/locales/ru.json @@ -211,8 +211,11 @@ "error_displaying_statistics": "Ошибка отображения статистики.", "stub_page_display": "Страница", - "subscription_ending_soon_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает {end_date} (через {days_left} дн.).\n\nЧтобы не потерять доступ, пожалуйста, продлите ее заранее в главном меню бота.", - "subscription_cancelled_notification": "Ваша подписка отменена. Доступ сохранится до конца оплаченного периода.", + "subscription_72h_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает через 72 часа ({end_date}).\n\nПожалуйста, продлите её в главном меню бота.", + "subscription_48h_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает через 48 часов ({end_date}).\n\nПожалуйста, продлите её в главном меню бота.", + "subscription_24h_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает через 24 часа ({end_date}).\n\nПожалуйста, продлите её в главном меню бота.", + "subscription_expired_notification": "Срок вашей подписки истек.", + "subscription_expired_yesterday_notification": "Ваша подписка истекла сутки назад.", "admin_new_trial_notification": "\ud83c\udf21 Пользователь {user_id} активировал пробный период до {end_date}.", "admin_new_payment_notification": "\ud83d\udcb3 Получен платеж от пользователя {user_id}: {months} мес. за {amount} {currency}.", diff --git a/requirements.txt b/requirements.txt index b17d568..d9133af 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,8 +4,7 @@ aiohttp==3.10.11 pydantic==2.7.1 yookassa==3.5.0 pycountry==23.12.11 -apscheduler==3.10.4 pydantic_settings sqlalchemy[asyncio]==2.0.29 asyncpg==0.29.0 -alembic==1.13.1 \ No newline at end of file +alembic==1.13.1