remnawave-minishop/bot/services/notification_service.py

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,
                        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,
    )


async def notify_admin_promo_activation(bot: Bot, settings: Settings,
                                        i18n: JsonI18n, user_id: int,
                                        code: str,
                                        bonus_days: int) -> None:
    await notify_admins(
        bot,
        settings,
        i18n,
        "admin_promo_activation_notification",
        user_id=user_id,
        code=code,
        bonus_days=bonus_days,
    )