hexdev
606aa4b89f
- Add `/push` command for admins to broadcast messages to all users with approve/decline confirmation flow (handlers/ads.py) - Add common handler for undefined commands and delete callback - Add `UserRepository.get_users()` to fetch all users - Add `UserService.send_promotion_to_users()` with error logging - Add promotion confirmation keyboard and FSM state - Add PostgreSQL availability check in startup.sh before migrations - Add new text constants for promotion flow
281 lines
10 KiB
Python
281 lines
10 KiB
Python
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 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
|
|
):
|
|
username_display = utils.format_username(user.username)
|
|
|
|
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,
|
|
)
|
|
|
|
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,
|
|
)
|
|
|
|
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,
|
|
),
|
|
)
|
|
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,
|
|
),
|
|
)
|
|
|
|
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,
|
|
),
|
|
)
|
|
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
|
|
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 | None:
|
|
rw_user = await get_user_by_telegram_id(rw_sdk, user_id)
|
|
|
|
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
|