malenia/services/user_service.py

import logging

from aiogram import Bot
from aiogram.types import Message, ReactionTypeEmoji, User as TelegramUser
from remnawave import RemnawaveSDK
from sqlalchemy.ext.asyncio import AsyncSession

from config import settings
from db.models.tickets import TicketStatus
from db.models.users import User
from keyboards.admins import ticket_actions
from misc import utils
from repositories import UserRepository
from repositories.tickets import TicketRepository
from schemas.subscriptions import SubscriptionPlan
from schemas.tickets import NewTicketDTO
from schemas.users import NewUserDTO, TicketCreator
from services.rw import RWUserInfo, get_user_by_telegram_id
from texts import (
    ADMIN_STANDARD_FALLBACK,
    ADMIN_TICKET_WITH_RW,
    ADMIN_TICKET_WITHOUT_RW,
    SUPPORT_SIGNATURE,
    TICKET_STATUS_EMOJIS,
)

logger = logging.getLogger(__name__)


class UserServiceError(Exception): ...


class UserService:
    def __init__(
        self, user_repository: UserRepository, ticket_repository: TicketRepository
    ) -> None:
        self.user_repository = user_repository
        self.ticket_repository = ticket_repository

    @staticmethod
    def format_ticket_name(
        creator: TicketCreator,
        status: TicketStatus | None = TicketStatus.OPEN,
        ticket_id: int | None = None,
    ):
        return (
            TICKET_STATUS_EMOJIS[status.value]
            + (f" T-{ticket_id:0>3}" if ticket_id else "")
            + " | "
            + (f"@{creator.username}" if creator.username else str(creator.id))
        )

    @staticmethod
    async def format_admin_support_message(
        ticket_number: int,
        user: TelegramUser,
        rw_user: RWUserInfo | None,
        balance: int,
        referal_id: int | None = None,
    ):
        username_display = utils.format_username(user.username)
        referal_display = utils.format_referal_id(referal_id)

        if rw_user is None:
            # Пользователь не найден в Remnawave
            return ADMIN_TICKET_WITHOUT_RW.format(
                ticket_number=ticket_number,
                full_name=user.full_name,
                username_display=username_display,
                telegram_id=user.id,
                referal_id=referal_display,
                balance=balance,
            )

        status_icon, status_text = utils.format_status(rw_user.status)
        expire_date = rw_user.expire_at.strftime("%d.%m.%Y")
        days_left = utils.format_days_left(rw_user.expire_at)
        used_traffic = utils.format_traffic(rw_user.used_traffic_bytes, rw_user.traffic_limit_bytes)
        hwid_limit = utils.format_hwid_limit(rw_user.hwid_device_limit)
        squads = utils.format_squads(rw_user.active_squads)

        return ADMIN_TICKET_WITH_RW.format(
            ticket_number=ticket_number,
            full_name=user.full_name,
            username_display=username_display,
            telegram_id=user.id,
            balance=balance,
            status_icon=status_icon,
            status=status_text,
            expire_date=expire_date,
            days_left=days_left,
            used_traffic=used_traffic,
            hwid_limit=hwid_limit,
            squads=squads,
            remnawave_uuid=rw_user.uuid,
            referal_id=referal_display,
        )

    async def add_user(
        self,
        session: AsyncSession,
        user_id: int,
        rw_sdk: RemnawaveSDK,
        referal: int | None = None,
    ) -> User:
        user = await self.user_repository.get_user_by_id(session, user_id)
        if user:
            return user

        rw_user = await get_user_by_telegram_id(rw_sdk, user_id)
        link = utils.get_subscription_link(rw_user.short_uuid) if rw_user else None
        referal = int(referal) if str(referal).isdigit() and user_id != int(referal) else None

        model = NewUserDTO(id=user_id, referal=referal, link=link)
        return await self.user_repository.create_user(session, model)

    async def create_subscription_ticket(
        self,
        session: AsyncSession,
        user: TelegramUser,
        subscription: SubscriptionPlan,
        bot: Bot,
    ):
        ticket = await self.ticket_repository.get_available_ticket_by_user_id(session, user.id)
        db_user = await self.user_repository.get_user_by_id(session, user.id)

        if ticket:
            try:
                await bot.send_message(
                    chat_id=settings.admin_group_id,
                    message_thread_id=ticket.thread_id,
                    text=utils.format_support_subscription(
                        subscription=subscription,
                        full_name=user.full_name,
                        user_id=user.id,
                        username=user.username,
                        balance=db_user.balance,
                        referal_id=db_user.referal_id,
                    ),
                    reply_markup=ticket_actions,
                )
                return
            except Exception:
                await self.ticket_repository.update_ticket_status(
                    session, ticket, TicketStatus.CLOSED
                )
                raise
        creator = TicketCreator(id=user.id, username=user.username)

        ticket_dto = NewTicketDTO(
            creator_id=user.id,
        )
        ticket = await self.ticket_repository.create_ticket(session, ticket_data=ticket_dto)

        thread = await bot.create_forum_topic(
            settings.admin_group_id,
            self.format_ticket_name(creator, ticket_id=ticket.id, status=TicketStatus.PENDING),
        )
        ticket = await self.ticket_repository.update_ticket_thread_id(
            session, ticket, thread.message_thread_id
        )
        db_user = await self.user_repository.get_user_by_id(session, ticket.creator_id)

        await bot.send_message(
            chat_id=settings.admin_group_id,
            message_thread_id=ticket.thread_id,
            text=utils.format_support_subscription(
                subscription=subscription,
                full_name=user.full_name,
                user_id=user.id,
                username=user.username,
                balance=db_user.balance,
                referal_id=db_user.referal_id,
            ),
            reply_markup=ticket_actions,
        )

    async def support_forward_message(self, session: AsyncSession, msg: Message, rw: RemnawaveSDK):
        user = await self.user_repository.get_user_by_id(session, msg.from_user.id)
        if not user:
            raise UserServiceError("User %d is not found.", msg.from_user.id)

        ticket = await self.ticket_repository.get_available_ticket_by_user_id(session, user.id)

        if ticket:
            try:
                await msg.copy_to(
                    chat_id=settings.admin_group_id, message_thread_id=ticket.thread_id
                )
                return
            except Exception:
                await self.ticket_repository.update_ticket_status(
                    session, ticket, TicketStatus.CLOSED
                )
                raise

        rw_user = await get_user_by_telegram_id(rw, user.id)
        creator = TicketCreator(id=user.id, username=msg.from_user.username)

        ticket_dto = NewTicketDTO(
            creator_id=user.id,
        )
        ticket = await self.ticket_repository.create_ticket(session, ticket_data=ticket_dto)

        thread = await msg.bot.create_forum_topic(
            settings.admin_group_id,
            self.format_ticket_name(creator, ticket_id=ticket.id, status=TicketStatus.PENDING),
        )
        ticket = await self.ticket_repository.update_ticket_thread_id(
            session, ticket, thread.message_thread_id
        )

        await msg.bot.send_message(
            settings.admin_group_id,
            message_thread_id=thread.message_thread_id,
            text=await self.format_admin_support_message(
                ticket_number=ticket.id,
                user=msg.from_user,
                rw_user=rw_user,
                balance=user.balance,
                referal_id=user.referal_id,
            ),
            reply_markup=ticket_actions,
        )
        await msg.copy_to(settings.admin_group_id, message_thread_id=thread.message_thread_id)

    async def support_answer(self, session: AsyncSession, msg: Message):
        thread_id = msg.message_thread_id
        ticket = await self.ticket_repository.get_ticket_by_thread_id(session, thread_id=thread_id)

        if ticket.status == TicketStatus.PENDING:
            await self.ticket_repository.update_ticket_status(session, ticket, TicketStatus.OPEN)
            await msg.bot.edit_forum_topic(
                settings.admin_group_id,
                ticket.thread_id,
                name=self.format_ticket_name(TicketCreator(ticket.creator_id), ticket_id=ticket.id),
            )

        try:
            await msg.bot.send_message(ticket.creator_id, SUPPORT_SIGNATURE)
            await msg.copy_to(ticket.creator_id)
            await msg.react([ReactionTypeEmoji(emoji="⚡")])
        except Exception:
            await msg.reply(ADMIN_STANDARD_FALLBACK)

    async def close_ticket(self, session: AsyncSession, msg: Message):
        thread_id = msg.message_thread_id
        if not thread_id:
            return
        ticket = await self.ticket_repository.get_ticket_by_thread_id(session, thread_id=thread_id)

        await msg.bot.edit_forum_topic(
            settings.admin_group_id,
            ticket.thread_id,
            name=self.format_ticket_name(
                TicketCreator(ticket.creator_id),
                ticket_id=ticket.id,
                status=TicketStatus.CLOSED,
            ),
        )

        await self.ticket_repository.update_ticket_status(session, ticket, TicketStatus.CLOSED)
        await msg.bot.close_forum_topic(settings.admin_group_id, message_thread_id=thread_id)

    async def update_user_link(
        self, session: AsyncSession, user_id: int, rw_sdk: RemnawaveSDK
    ) -> User:
        rw_user = await get_user_by_telegram_id(rw_sdk, user_id)

        user = await self.user_repository.get_user_by_id(session, user_id)

        if not rw_user:
            return user

        link = utils.get_subscription_link(rw_user.short_uuid)

        user = await self.user_repository.update_user_link(session, user_id, link)

        return user

    async def fetch_user_by_thread(self, session: AsyncSession, thread_id: int) -> User | None:
        ticket = await self.ticket_repository.get_ticket_by_thread_id(session, thread_id=thread_id)

        return await self.user_repository.get_user_by_id(session, ticket.creator_id)

    async def fetch_referal_by_thread(self, session: AsyncSession, thread_id: int) -> User | None:
        ticket = await self.ticket_repository.get_ticket_by_thread_id(session, thread_id=thread_id)

        return await self.user_repository.get_referal_by_user_id(session, ticket.creator_id)

    async def send_promotion_to_users(self, session: AsyncSession, msg: Message) -> list[User]:
        users = await self.user_repository.get_users(session)

        sent: list[User] = []
        for user in users:
            try:
                await msg.copy_to(user.id)
                sent.append(user)
            except Exception:
                logger.critical("failed to send a message to user %d.", user.id, exc_info=True)

        return sent