from __future__ import annotations import secrets import math from dataclasses import dataclass, field from datetime import datetime, timezone from sqlalchemy import delete, func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.config import Settings from app.db.base import utcnow from app.db.models import ( InternalSquad, PromoCode, PromoCodeApplication, ReferralCode, ReferralInvite, RemnawaveUser, RemnawaveUserInternalSquad, SubscriptionRequestLog, TelegramUser, ) from app.schemas.remnawave import RemnawaveUser as RemoteRemnawaveUser from app.schemas.remnawave import SubscriptionRequestHistory from app.services.remnawave_client import RemnawaveApiClient @dataclass(slots=True) class CachedUserView: record: RemnawaveUser internal_squads: list[str] = field(default_factory=list) recent_requests: list[SubscriptionRequestLog] = field(default_factory=list) @dataclass(slots=True) class ReferralSummary: total_invited: int = 0 referral_code: str = "" applied_referral_code: str = "" applied_promo_code: str = "" applied_promo_discount_percent: int = 0 applied_promo_plan_code: str | None = None applied_promo_expires_at: datetime | None = None recent_names: list[str] = field(default_factory=list) @dataclass(slots=True) class PromoCodeView: code: str plan_code: str | None discount_percent: int expires_at: datetime | None is_active: bool created_by_telegram_id: int | None = None created_at: datetime | None = None updated_at: datetime | None = None @property def is_expired(self) -> bool: return self.expires_at is not None and self.expires_at <= utcnow() @property def is_valid(self) -> bool: return self.is_active and not self.is_expired and self.discount_percent > 0 @dataclass(slots=True) class AdminStats: total_telegram_users: int = 0 total_cached_users: int = 0 active_cached_users: int = 0 @dataclass(slots=True) class AdminTelegramUserListItem: telegram_id: int username: str | None first_name: str | None last_name: str | None language_code: str | None is_admin: bool is_blocked: bool last_seen_at: datetime created_at: datetime @property def display_name(self) -> str: value = " ".join(part for part in [self.first_name, self.last_name] if part).strip() return value or self.username or str(self.telegram_id) @dataclass(slots=True) class AdminTelegramUsersPage: items: list[AdminTelegramUserListItem] = field(default_factory=list) page: int = 1 page_size: int = 8 total_items: int = 0 total_pages: int = 1 @dataclass(slots=True) class AdminTelegramUserProfile: user: TelegramUser referral_code: str = "" applied_referral_code: str = "" invited_count: int = 0 recent_invited_names: list[str] = field(default_factory=list) accesses: list[CachedUserView] = field(default_factory=list) class SyncService: def __init__( self, *, settings: Settings, session_factory: async_sessionmaker[AsyncSession], remnawave_client: RemnawaveApiClient, ) -> None: self._settings = settings self._session_factory = session_factory self._remnawave_client = remnawave_client async def is_user_blocked(self, telegram_id: int) -> bool: async with self._session_factory() as session: is_blocked = await session.scalar( select(TelegramUser.is_blocked).where( TelegramUser.telegram_id == telegram_id ) ) return bool(is_blocked) async def block_telegram_user(self, telegram_id: int) -> None: async with self._session_factory() as session: user = await session.scalar( select(TelegramUser).where(TelegramUser.telegram_id == telegram_id) ) if user is not None: user.is_blocked = True user.updated_at = utcnow() await session.commit() async def unblock_telegram_user(self, telegram_id: int) -> None: async with self._session_factory() as session: user = await session.scalar( select(TelegramUser).where(TelegramUser.telegram_id == telegram_id) ) if user is not None: user.is_blocked = False user.updated_at = utcnow() await session.commit() async def get_expiring_users( self, *, days_before: int, tolerance_hours: int = 6, ) -> list[tuple[int, int, datetime]]: """Return (remnawave_user.id, telegram_id, expire_at) for users expiring within window.""" from datetime import timedelta now = utcnow() window_start = now + timedelta(days=days_before) - timedelta(hours=tolerance_hours) window_end = now + timedelta(days=days_before) + timedelta(hours=tolerance_hours) async with self._session_factory() as session: rows = await session.execute( select( RemnawaveUser.id, RemnawaveUser.telegram_id, RemnawaveUser.expire_at, ) .where( RemnawaveUser.status == "ACTIVE", RemnawaveUser.telegram_id.isnot(None), RemnawaveUser.expire_at >= window_start, RemnawaveUser.expire_at <= window_end, ) ) return [(row[0], row[1], row[2]) for row in rows.all()] async def register_telegram_user( self, *, telegram_id: int, username: str | None, first_name: str | None, last_name: str | None, language_code: str | None, ) -> None: async with self._session_factory() as session: await self._upsert_telegram_user( session, telegram_id=telegram_id, username=username, first_name=first_name, last_name=last_name, language_code=language_code, ) await self._get_or_create_referral_code(session, telegram_id=telegram_id) await session.commit() async def create_promo_code( self, *, code: str, plan_code: str | None, discount_percent: int, expires_at: datetime | None, created_by_telegram_id: int, ) -> PromoCodeView: normalized_code = self.normalize_discount_code(code) if not normalized_code: raise ValueError("Введите слово для промокода.") if len(normalized_code) > 32: raise ValueError("Промокод должен быть не длиннее 32 символов.") if not all(ch.isalnum() or ch in "_-" for ch in normalized_code): raise ValueError("Промокод может содержать только буквы, цифры, _ и -.") resolved_discount = int(discount_percent) if resolved_discount <= 0 or resolved_discount > 100: raise ValueError("Скидка должна быть от 1 до 100%.") resolved_plan_code = (plan_code or "").strip() or None if resolved_plan_code and len(resolved_plan_code) > 64: raise ValueError("Код тарифа слишком длинный.") resolved_expires_at = self._normalize_datetime(expires_at) if resolved_expires_at is not None and resolved_expires_at <= utcnow(): raise ValueError("Срок действия промокода должен быть в будущем.") async with self._session_factory() as session: duplicate = await session.scalar( select(PromoCode.id).where(PromoCode.code == normalized_code) ) if duplicate is not None: raise ValueError("Такой промокод уже существует.") promo = PromoCode( code=normalized_code, plan_code=resolved_plan_code, discount_percent=resolved_discount, expires_at=resolved_expires_at, is_active=True, created_by_telegram_id=created_by_telegram_id, ) session.add(promo) await session.commit() await session.refresh(promo) return self._promo_code_view(promo) async def get_admin_promo_codes(self, *, limit: int = 20) -> list[PromoCodeView]: resolved_limit = max(1, min(limit, 100)) async with self._session_factory() as session: rows = list( ( await session.scalars( select(PromoCode) .order_by(PromoCode.created_at.desc(), PromoCode.id.desc()) .limit(resolved_limit) ) ).all() ) return [self._promo_code_view(row) for row in rows] async def get_active_promo_for_user(self, telegram_id: int) -> PromoCodeView | None: async with self._session_factory() as session: promo = await self._get_active_promo_for_user(session, telegram_id=telegram_id) return self._promo_code_view(promo) if promo is not None else None async def consume_discount_for_user(self, telegram_id: int) -> None: """Burn the applied promo / referral discount so it cannot be reused. - For promo codes: deletes the row from ``promo_code_applications``. - For referral invites: sets ``discount_used = True`` so the referral bonus logic (inviter days) is preserved while the invited user no longer receives a price discount. - Sets ``discount_ever_used = True`` on ``telegram_users`` to permanently prevent any future discount code activation. """ async with self._session_factory() as session: # 1. Burn promo-code application await session.execute( delete(PromoCodeApplication).where( PromoCodeApplication.telegram_id == telegram_id ) ) # 2. Mark referral invite discount as used await session.execute( update(ReferralInvite) .where( ReferralInvite.invited_telegram_id == telegram_id, ReferralInvite.discount_used == False, # noqa: E712 ) .values(discount_used=True) ) # 3. Permanently mark the user so no new codes can be activated await session.execute( update(TelegramUser) .where( TelegramUser.telegram_id == telegram_id, TelegramUser.discount_ever_used == False, # noqa: E712 ) .values(discount_ever_used=True) ) await session.commit() async def is_referral_discount_used(self, telegram_id: int) -> bool: """Return True if the referral invite discount was already consumed.""" async with self._session_factory() as session: discount_used = await session.scalar( select(ReferralInvite.discount_used).where( ReferralInvite.invited_telegram_id == telegram_id ) ) return bool(discount_used) async def apply_referral_code( self, *, telegram_id: int, username: str | None, first_name: str | None, last_name: str | None, language_code: str | None, referral_code: str, ) -> str: async with self._session_factory() as session: invited = await self._upsert_telegram_user( session, telegram_id=telegram_id, username=username, first_name=first_name, last_name=last_name, language_code=language_code, ) # Block activation if the user has already used a discount if invited.discount_ever_used: raise ValueError( "Вы уже использовали промокод или реферальный код. " "Повторная активация невозможна." ) await self._get_or_create_referral_code(session, telegram_id=telegram_id) raw_code = self.normalize_discount_code(referral_code) promo = await self._find_promo_code(session, code=raw_code) if promo is not None: self._ensure_promo_can_be_applied(promo) await self._upsert_promo_application( session, telegram_id=telegram_id, promo_code_id=promo.id, ) await session.commit() return promo.code existing = await session.scalar( select(ReferralInvite).where( ReferralInvite.invited_telegram_id == telegram_id ) ) if existing is not None: raise ValueError("Реферальный код уже применен к вашему аккаунту.") inviter_telegram_id = await self._find_inviter_by_referral_code(session, referral_code) if inviter_telegram_id is None: raise ValueError("Реферальный код не найден.") if inviter_telegram_id == telegram_id: raise ValueError("Нельзя использовать собственный реферальный код.") invite_created = await self._upsert_referral_invite( session, invited=invited, referred_by_telegram_id=inviter_telegram_id, ) if not invite_created: raise ValueError("Не удалось применить реферальный код.") await session.execute( delete(PromoCodeApplication).where( PromoCodeApplication.telegram_id == telegram_id ) ) await session.commit() return referral_code.strip().upper() async def get_referral_summary(self, telegram_id: int) -> ReferralSummary: async with self._session_factory() as session: referral_code = await self._get_or_create_referral_code( session, telegram_id=telegram_id, ) await session.commit() total_invited = await session.scalar( select(func.count(ReferralInvite.id)).where( ReferralInvite.inviter_telegram_id == telegram_id ) ) recent_rows = await session.execute( select(ReferralInvite.invited_display_name) .where(ReferralInvite.inviter_telegram_id == telegram_id) .order_by(ReferralInvite.created_at.desc()) .limit(5) ) recent_names = list(recent_rows.scalars().all()) applied_referral_code = await session.scalar( select(ReferralCode.code) .join( ReferralInvite, ReferralInvite.inviter_telegram_id == ReferralCode.telegram_id, ) .where(ReferralInvite.invited_telegram_id == telegram_id) ) applied_promo = await self._get_active_promo_for_user( session, telegram_id=telegram_id, ) return ReferralSummary( total_invited=int(total_invited or 0), referral_code=referral_code, applied_referral_code=applied_referral_code or "", applied_promo_code=applied_promo.code if applied_promo is not None else "", applied_promo_discount_percent=( int(applied_promo.discount_percent) if applied_promo is not None else 0 ), applied_promo_plan_code=applied_promo.plan_code if applied_promo is not None else None, applied_promo_expires_at=applied_promo.expires_at if applied_promo is not None else None, recent_names=recent_names, ) async def get_admin_stats(self) -> AdminStats: async with self._session_factory() as session: total_telegram_users = await session.scalar( select(func.count(TelegramUser.id)) ) total_cached_users = await session.scalar( select(func.count(RemnawaveUser.id)) ) active_cached_users = await session.scalar( select(func.count(RemnawaveUser.id)).where(RemnawaveUser.status == "ACTIVE") ) return AdminStats( total_telegram_users=int(total_telegram_users or 0), total_cached_users=int(total_cached_users or 0), active_cached_users=int(active_cached_users or 0), ) async def get_admin_telegram_users_page( self, *, page: int, page_size: int = 8, ) -> AdminTelegramUsersPage: resolved_page_size = max(1, min(page_size, 20)) async with self._session_factory() as session: total_items = int( await session.scalar(select(func.count(TelegramUser.id))) or 0 ) total_pages = max(1, math.ceil(total_items / resolved_page_size)) if total_items else 1 resolved_page = min(max(page, 1), total_pages) rows = list( ( await session.scalars( select(TelegramUser) .order_by( TelegramUser.last_seen_at.desc(), TelegramUser.created_at.desc(), TelegramUser.id.desc(), ) .offset((resolved_page - 1) * resolved_page_size) .limit(resolved_page_size) ) ).all() ) return AdminTelegramUsersPage( items=[ AdminTelegramUserListItem( telegram_id=row.telegram_id, username=row.username, first_name=row.first_name, last_name=row.last_name, language_code=row.language_code, is_admin=row.is_admin, is_blocked=row.is_blocked, last_seen_at=row.last_seen_at, created_at=row.created_at, ) for row in rows ], page=resolved_page, page_size=resolved_page_size, total_items=total_items, total_pages=total_pages, ) async def get_admin_telegram_user_profile( self, *, telegram_id: int, ) -> AdminTelegramUserProfile | None: async with self._session_factory() as session: user = await session.scalar( select(TelegramUser).where(TelegramUser.telegram_id == telegram_id) ) if user is None: return None referral_code = await session.scalar( select(ReferralCode.code).where(ReferralCode.telegram_id == telegram_id) ) invited_count = await session.scalar( select(func.count(ReferralInvite.id)).where( ReferralInvite.inviter_telegram_id == telegram_id ) ) recent_rows = await session.execute( select(ReferralInvite.invited_display_name) .where(ReferralInvite.inviter_telegram_id == telegram_id) .order_by(ReferralInvite.created_at.desc()) .limit(5) ) recent_invited_names = list(recent_rows.scalars().all()) applied_referral_code = await session.scalar( select(ReferralCode.code) .join( ReferralInvite, ReferralInvite.inviter_telegram_id == ReferralCode.telegram_id, ) .where(ReferralInvite.invited_telegram_id == telegram_id) ) accesses = await self.get_cached_users_for_telegram(telegram_id) return AdminTelegramUserProfile( user=user, referral_code=referral_code or "", applied_referral_code=applied_referral_code or "", invited_count=int(invited_count or 0), recent_invited_names=recent_invited_names, accesses=accesses, ) async def sync_users_for_telegram( self, *, telegram_id: int, username: str | None, first_name: str | None, last_name: str | None, language_code: str | None, ) -> list[CachedUserView]: remote_users = await self._remnawave_client.get_users_by_telegram_id(telegram_id) async with self._session_factory() as session: owner = await self._upsert_telegram_user( session, telegram_id=telegram_id, username=username, first_name=first_name, last_name=last_name, language_code=language_code, ) await self._sync_remote_users_for_telegram( session, telegram_id=telegram_id, remote_users=remote_users, owner=owner, ) await session.commit() return await self.get_cached_users_for_telegram(telegram_id) async def refresh_cached_users_for_telegram(self, *, telegram_id: int) -> list[CachedUserView]: remote_users = await self._remnawave_client.get_users_by_telegram_id(telegram_id) async with self._session_factory() as session: owner = await session.scalar( select(TelegramUser).where(TelegramUser.telegram_id == telegram_id) ) await self._sync_remote_users_for_telegram( session, telegram_id=telegram_id, remote_users=remote_users, owner=owner, ) await session.commit() return await self.get_cached_users_for_telegram(telegram_id) async def link_user_by_short_uuid( self, *, short_uuid: str, telegram_id: int, username: str | None, first_name: str | None, last_name: str | None, language_code: str | None, ) -> CachedUserView: remote_user = await self._remnawave_client.get_user_by_short_uuid(short_uuid) if remote_user.telegram_id and remote_user.telegram_id != telegram_id: raise ValueError("Этот аккаунт уже привязан к другому Telegram ID.") updated_user = await self._remnawave_client.update_user( { "uuid": str(remote_user.uuid), "telegramId": telegram_id, } ) async with self._session_factory() as session: owner = await self._upsert_telegram_user( session, telegram_id=telegram_id, username=username, first_name=first_name, last_name=last_name, language_code=language_code, ) await self._upsert_remnawave_user(session, updated_user, owner) if self._settings.sync_subscription_history: history = await self._remnawave_client.get_user_subscription_history( str(updated_user.uuid) ) await self._upsert_history(session, str(updated_user.uuid), history) await session.commit() user = await self.get_cached_user_by_uuid(str(updated_user.uuid), include_history=True) if user is None: raise ValueError("Не удалось получить локальный кэш пользователя после привязки.") return user async def sync_user_by_identifier( self, *, identifier: str, include_history: bool = True, ) -> CachedUserView: resolved = await self._remnawave_client.resolve_user(identifier) remote_user = await self._remnawave_client.get_user_by_uuid(str(resolved.uuid)) history = None if include_history and self._settings.sync_subscription_history: history = await self._remnawave_client.get_user_subscription_history(str(resolved.uuid)) async with self._session_factory() as session: await self._upsert_remnawave_user(session, remote_user, owner=None) if history is not None: await self._upsert_history(session, str(resolved.uuid), history) await session.commit() user = await self.get_cached_user_by_uuid(str(resolved.uuid), include_history=include_history) if user is None: raise ValueError("Не удалось получить локальный кэш пользователя после синхронизации.") return user async def sync_all_users(self, batch_size: int) -> int: start = 0 total_synced = 0 while True: page = await self._remnawave_client.get_all_users(start=start, size=batch_size) if not page.users: break async with self._session_factory() as session: for remote_user in page.users: await self._upsert_remnawave_user(session, remote_user, owner=None) total_synced += 1 await session.commit() start += len(page.users) if start >= page.total: break return total_synced async def get_cached_users_for_telegram(self, telegram_id: int) -> list[CachedUserView]: async with self._session_factory() as session: rows = await session.execute( select(RemnawaveUser, InternalSquad.name) .outerjoin( RemnawaveUserInternalSquad, RemnawaveUserInternalSquad.user_id == RemnawaveUser.id, ) .outerjoin( InternalSquad, InternalSquad.uuid == RemnawaveUserInternalSquad.squad_uuid, ) .where(RemnawaveUser.telegram_id == telegram_id) .order_by(RemnawaveUser.username.asc()) ) return self._build_views(rows.all()) async def get_cached_user_by_uuid( self, user_uuid: str, *, include_history: bool = False, ) -> CachedUserView | None: async with self._session_factory() as session: rows = await session.execute( select(RemnawaveUser, InternalSquad.name) .outerjoin( RemnawaveUserInternalSquad, RemnawaveUserInternalSquad.user_id == RemnawaveUser.id, ) .outerjoin( InternalSquad, InternalSquad.uuid == RemnawaveUserInternalSquad.squad_uuid, ) .where(RemnawaveUser.rw_uuid == user_uuid) ) views = self._build_views(rows.all()) if not views: return None view = views[0] if include_history: history_rows = await session.execute( select(SubscriptionRequestLog) .where(SubscriptionRequestLog.user_uuid == user_uuid) .order_by(SubscriptionRequestLog.request_at.desc()) .limit(5) ) view.recent_requests = list(history_rows.scalars().all()) return view async def _upsert_telegram_user( self, session: AsyncSession, *, telegram_id: int, username: str | None, first_name: str | None, last_name: str | None, language_code: str | None, ) -> TelegramUser: record = await session.scalar( select(TelegramUser).where(TelegramUser.telegram_id == telegram_id) ) if record is None: record = TelegramUser(telegram_id=telegram_id) session.add(record) record.username = username record.first_name = first_name record.last_name = last_name record.language_code = language_code record.is_admin = self._settings.is_admin(telegram_id) record.last_seen_at = utcnow() record.updated_at = utcnow() await session.flush() return record async def _sync_remote_users_for_telegram( self, session: AsyncSession, *, telegram_id: int, remote_users: list[RemoteRemnawaveUser], owner: TelegramUser | None, ) -> None: active_uuids = {str(remote_user.uuid) for remote_user in remote_users} for remote_user in remote_users: await self._upsert_remnawave_user(session, remote_user, owner) filters = [RemnawaveUser.telegram_id == telegram_id] if owner is not None: filters.append(RemnawaveUser.owner_telegram_user_id == owner.id) stale_records = await session.scalars( select(RemnawaveUser).where(or_(*filters)) ) for record in stale_records: if record.rw_uuid in active_uuids: continue if record.telegram_id == telegram_id: record.telegram_id = None if owner is not None and record.owner_telegram_user_id == owner.id: record.owner_telegram_user_id = None record.synced_at = utcnow() async def _upsert_remnawave_user( self, session: AsyncSession, remote_user: RemoteRemnawaveUser, owner: TelegramUser | None, ) -> RemnawaveUser: owner_id = owner.id if owner is not None else None if owner_id is None and remote_user.telegram_id is not None: with session.no_autoflush: owner_id = await self._find_owner_id(session, remote_user.telegram_id) record = await session.scalar( select(RemnawaveUser).where(RemnawaveUser.rw_uuid == str(remote_user.uuid)) ) if record is None: record = RemnawaveUser(rw_uuid=str(remote_user.uuid)) session.add(record) record.owner_telegram_user_id = owner_id record.rw_id = remote_user.id record.short_uuid = remote_user.short_uuid record.username = remote_user.username record.status = remote_user.status record.traffic_limit_bytes = remote_user.traffic_limit_bytes record.used_traffic_bytes = remote_user.user_traffic.used_traffic_bytes record.lifetime_used_traffic_bytes = remote_user.user_traffic.lifetime_used_traffic_bytes record.expire_at = self._normalize_datetime(remote_user.expire_at) record.telegram_id = remote_user.telegram_id record.email = remote_user.email record.description = remote_user.description record.tag = remote_user.tag record.hwid_device_limit = remote_user.hwid_device_limit record.external_squad_uuid = self._maybe_str(remote_user.external_squad_uuid) record.trojan_password = remote_user.trojan_password record.vless_uuid = str(remote_user.vless_uuid) record.ss_password = remote_user.ss_password record.last_triggered_threshold = remote_user.last_triggered_threshold record.sub_revoked_at = self._normalize_datetime(remote_user.sub_revoked_at) record.sub_last_user_agent = remote_user.sub_last_user_agent record.sub_last_opened_at = self._normalize_datetime(remote_user.sub_last_opened_at) record.last_traffic_reset_at = self._normalize_datetime(remote_user.last_traffic_reset_at) record.subscription_url = remote_user.subscription_url record.online_at = self._normalize_datetime(remote_user.user_traffic.online_at) record.first_connected_at = self._normalize_datetime(remote_user.user_traffic.first_connected_at) record.last_connected_node_uuid = self._maybe_str(remote_user.user_traffic.last_connected_node_uuid) record.remote_created_at = self._normalize_datetime(remote_user.created_at) record.remote_updated_at = self._normalize_datetime(remote_user.updated_at) record.synced_at = utcnow() await session.flush() await session.execute( delete(RemnawaveUserInternalSquad).where( RemnawaveUserInternalSquad.user_id == record.id ) ) for squad in remote_user.active_internal_squads: squad_uuid = str(squad.uuid) squad_record = await session.get(InternalSquad, squad_uuid) if squad_record is None: squad_record = InternalSquad(uuid=squad_uuid, name=squad.name, synced_at=utcnow()) session.add(squad_record) else: squad_record.name = squad.name squad_record.synced_at = utcnow() session.add( RemnawaveUserInternalSquad( user_id=record.id, squad_uuid=squad_uuid, ) ) await session.flush() return record async def _upsert_history( self, session: AsyncSession, user_uuid: str, history: SubscriptionRequestHistory, ) -> None: for item in history.records: existing = await session.scalar( select(SubscriptionRequestLog).where( SubscriptionRequestLog.remote_id == item.id ) ) if existing is None: existing = SubscriptionRequestLog(remote_id=item.id, user_uuid=user_uuid) session.add(existing) existing.user_uuid = user_uuid existing.request_at = self._normalize_datetime(item.request_at) existing.request_ip = item.request_ip existing.user_agent = item.user_agent existing.synced_at = utcnow() async def _upsert_referral_invite( self, session: AsyncSession, *, invited: TelegramUser, referred_by_telegram_id: int | None, ) -> bool: if referred_by_telegram_id is None or referred_by_telegram_id == invited.telegram_id: return False existing = await session.scalar( select(ReferralInvite).where( ReferralInvite.invited_telegram_id == invited.telegram_id ) ) if existing is not None: return False display_name = " ".join( part for part in [invited.first_name, invited.last_name] if part ).strip() or invited.username or str(invited.telegram_id) session.add( ReferralInvite( inviter_telegram_id=referred_by_telegram_id, invited_telegram_id=invited.telegram_id, invited_username=invited.username, invited_display_name=display_name, ) ) return True async def _get_or_create_referral_code( self, session: AsyncSession, *, telegram_id: int, ) -> str: existing = await session.scalar( select(ReferralCode).where(ReferralCode.telegram_id == telegram_id) ) if existing is not None: return existing.code alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" while True: code = "".join(secrets.choice(alphabet) for _ in range(8)) duplicate = await session.scalar( select(ReferralCode.id).where(ReferralCode.code == code) ) if duplicate is None: session.add(ReferralCode(telegram_id=telegram_id, code=code)) await session.flush() return code async def _find_inviter_by_referral_code( self, session: AsyncSession, referred_by_referral_code: str | None, ) -> int | None: raw_code = (referred_by_referral_code or "").strip().upper() if not raw_code: return None return await session.scalar( select(ReferralCode.telegram_id).where(ReferralCode.code == raw_code) ) async def _find_promo_code( self, session: AsyncSession, *, code: str, ) -> PromoCode | None: raw_code = self.normalize_discount_code(code) if not raw_code: return None return await session.scalar( select(PromoCode).where(PromoCode.code == raw_code) ) async def _get_active_promo_for_user( self, session: AsyncSession, *, telegram_id: int, ) -> PromoCode | None: promo = await session.scalar( select(PromoCode) .join( PromoCodeApplication, PromoCodeApplication.promo_code_id == PromoCode.id, ) .where(PromoCodeApplication.telegram_id == telegram_id) ) if promo is None or not self._promo_is_valid(promo): return None return promo async def _upsert_promo_application( self, session: AsyncSession, *, telegram_id: int, promo_code_id: int, ) -> None: application = await session.scalar( select(PromoCodeApplication).where( PromoCodeApplication.telegram_id == telegram_id ) ) if application is None: session.add( PromoCodeApplication( telegram_id=telegram_id, promo_code_id=promo_code_id, ) ) return application.promo_code_id = promo_code_id application.updated_at = utcnow() async def _find_owner_id(self, session: AsyncSession, telegram_id: int | None) -> int | None: if telegram_id is None: return None return await session.scalar( select(TelegramUser.id).where(TelegramUser.telegram_id == telegram_id) ) @staticmethod def _normalize_datetime(value: datetime | None) -> datetime | None: if value is None: return None if value.tzinfo is None: return value return value.astimezone(timezone.utc).replace(tzinfo=None) @staticmethod def _maybe_str(value: object | None) -> str | None: return str(value) if value is not None else None @staticmethod def normalize_discount_code(value: str | None) -> str: return "".join((value or "").strip().upper().split()) @classmethod def _ensure_promo_can_be_applied(cls, promo: PromoCode) -> None: if not promo.is_active: raise ValueError("Промокод отключён.") if cls._promo_is_expired(promo): raise ValueError("Срок действия промокода истёк.") if promo.discount_percent <= 0: raise ValueError("Промокод не даёт скидку.") @classmethod def _promo_is_valid(cls, promo: PromoCode) -> bool: return bool(promo.is_active) and not cls._promo_is_expired(promo) and promo.discount_percent > 0 @staticmethod def _promo_is_expired(promo: PromoCode) -> bool: return promo.expires_at is not None and promo.expires_at <= utcnow() @staticmethod def _promo_code_view(promo: PromoCode) -> PromoCodeView: return PromoCodeView( code=promo.code, plan_code=promo.plan_code, discount_percent=int(promo.discount_percent), expires_at=promo.expires_at, is_active=bool(promo.is_active), created_by_telegram_id=promo.created_by_telegram_id, created_at=promo.created_at, updated_at=promo.updated_at, ) @staticmethod def _build_views(rows: list[tuple[RemnawaveUser, str | None]]) -> list[CachedUserView]: view_map: dict[int, CachedUserView] = {} for user, squad_name in rows: view = view_map.setdefault(user.id, CachedUserView(record=user)) if squad_name and squad_name not in view.internal_squads: view.internal_squads.append(squad_name) return list(view_map.values())