root/tgex-backend
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: handle unsubscribe

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2025-12-01 10:30:31 +03:00
parent c32635839a
commit ca3805eec3
14 changed files with 447 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
migration/versions/3b8259700b61_add_subscription_status_and_.py Normal file
View File

@@ -0,0 +1,42 @@
"""add subscription status and unsubscribed_at
Revision ID: 3b8259700b61
Revises: f05f85d10837
Create Date: 2025-11-25 18:45:04.642105
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '3b8259700b61'
down_revision: str | None = 'f05f85d10837'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
# Создаём enum тип явно
subscription_status = sa.Enum('ACTIVE', 'UNSUBSCRIBED', name='subscriptionstatus')
subscription_status.create(op.get_bind(), checkfirst=True)
op.add_column(
'subscription',
sa.Column('status', subscription_status, nullable=False, server_default='ACTIVE'),
)
op.add_column('subscription', sa.Column('unsubscribed_at', sa.DateTime(timezone=True), nullable=True))
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('subscription', 'unsubscribed_at')
op.drop_column('subscription', 'status')
# Удаляем enum тип
sa.Enum(name='subscriptionstatus').drop(op.get_bind(), checkfirst=True)
# ### end Alembic commands ###

63
migration/versions/52ad7cc9bf1e_add_unique_constraint_for_active_.py Normal file
View File

@@ -0,0 +1,63 @@
"""add unique constraint for active subscriptions
Revision ID: 52ad7cc9bf1e
Revises: 3b8259700b61
Create Date: 2025-11-26 14:14:57.814576
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '52ad7cc9bf1e'
down_revision: str | None = '3b8259700b61'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# Добавляем поле target_channel_id (сначала nullable)
op.add_column('subscription', sa.Column('target_channel_id', sa.Uuid(), nullable=True))
# Заполняем target_channel_id из placement
op.execute("""
UPDATE subscription
SET target_channel_id = placement.target_channel_id
FROM placement
WHERE subscription.placement_id = placement.id
""")
# Делаем поле NOT NULL после заполнения
op.alter_column('subscription', 'target_channel_id', nullable=False)
# Добавляем FK constraint
op.create_foreign_key(
'subscription_target_channel_id_fkey', 'subscription', 'target_channel', ['target_channel_id'], ['id']
)
# Создаем индекс на target_channel_id
op.create_index(op.f('ix_subscription_target_channel_id'), 'subscription', ['target_channel_id'])
# Создаём UNIQUE constraint для активных подписок
op.create_index(
'unique_active_subscription_per_channel',
'subscription',
['subscriber_id', 'target_channel_id'],
unique=True,
postgresql_where=sa.text("status = 'ACTIVE'::subscriptionstatus"),
)
def downgrade() -> None:
# Удаляем UNIQUE constraint
op.drop_index('unique_active_subscription_per_channel', table_name='subscription')
# Удаляем индекс и FK constraint
op.drop_index(op.f('ix_subscription_target_channel_id'), table_name='subscription')
op.drop_constraint('subscription_target_channel_id_fkey', 'subscription', type_='foreignkey')
# Удаляем колонку
op.drop_column('subscription', 'target_channel_id')

125
migration/versions/f05f85d10837_init.py
View File

@@ -5,6 +5,7 @@ Revises:
Create Date: 2025-11-13 01:06:51.623272
"""
from collections.abc import Sequence
import sqlalchemy as sa
@@ -19,7 +20,8 @@ depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('subscriber',
op.create_table(
'subscriber',
sa.Column('telegram_id', sa.BigInteger(), nullable=False),
sa.Column('username', sa.String(), nullable=True),
sa.Column('first_name', sa.String(), nullable=True),
@@ -28,20 +30,22 @@ def upgrade() -> None:
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id')
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_subscriber_telegram_id'), 'subscriber', ['telegram_id'], unique=True)
op.create_table('user',
op.create_table(
'user',
sa.Column('telegram_id', sa.Integer(), nullable=False),
sa.Column('username', sa.String(), nullable=True),
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint('id')
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_user_telegram_id'), 'user', ['telegram_id'], unique=True)
op.create_table('external_channel',
op.create_table(
'external_channel',
sa.Column('telegram_id', sa.BigInteger(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('username', sa.String(), nullable=True),
@@ -52,13 +56,17 @@ def upgrade() -> None:
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
sa.ForeignKeyConstraint(
['user_id'],
['user.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_external_channel_telegram_id'), 'external_channel', ['telegram_id'], unique=True)
op.create_index(op.f('ix_external_channel_user_id'), 'external_channel', ['user_id'], unique=False)
op.create_index(op.f('ix_external_channel_username'), 'external_channel', ['username'], unique=False)
op.create_table('login_token',
op.create_table(
'login_token',
sa.Column('token', sa.String(), nullable=False),
sa.Column('user_id', sa.Uuid(), nullable=False),
sa.Column('expires_at', sa.DateTime(timezone=True), nullable=False),
@@ -67,11 +75,15 @@ def upgrade() -> None:
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
sa.ForeignKeyConstraint(
['user_id'],
['user.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_login_token_token'), 'login_token', ['token'], unique=True)
op.create_table('target_channel',
op.create_table(
'target_channel',
sa.Column('telegram_id', sa.BigInteger(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('username', sa.String(), nullable=True),
@@ -81,13 +93,17 @@ def upgrade() -> None:
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
sa.ForeignKeyConstraint(
['user_id'],
['user.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_target_channel_telegram_id'), 'target_channel', ['telegram_id'], unique=True)
op.create_index(op.f('ix_target_channel_user_id'), 'target_channel', ['user_id'], unique=False)
op.create_index(op.f('ix_target_channel_username'), 'target_channel', ['username'], unique=False)
op.create_table('creative',
op.create_table(
'creative',
sa.Column('name', sa.String(), nullable=False),
sa.Column('text', sa.String(), nullable=False),
sa.Column('status', sa.Enum('ACTIVE', 'ARCHIVED', name='creativestatus'), nullable=False),
@@ -98,20 +114,28 @@ def upgrade() -> None:
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['target_channel_id'], ['target_channel.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
sa.ForeignKeyConstraint(
['target_channel_id'],
['target_channel.id'],
),
sa.ForeignKeyConstraint(
['user_id'],
['user.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_creative_target_channel_id'), 'creative', ['target_channel_id'], unique=False)
op.create_index(op.f('ix_creative_user_id'), 'creative', ['user_id'], unique=False)
op.create_table('target_external_channel',
op.create_table(
'target_external_channel',
sa.Column('target_channel_id', sa.Uuid(), nullable=False),
sa.Column('external_channel_id', sa.Uuid(), nullable=False),
sa.ForeignKeyConstraint(['external_channel_id'], ['external_channel.id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['target_channel_id'], ['target_channel.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('target_channel_id', 'external_channel_id')
sa.PrimaryKeyConstraint('target_channel_id', 'external_channel_id'),
)
op.create_table('placement',
op.create_table(
'placement',
sa.Column('target_channel_id', sa.Uuid(), nullable=False),
sa.Column('external_channel_id', sa.Uuid(), nullable=False),
sa.Column('creative_id', sa.Uuid(), nullable=False),
@@ -125,23 +149,40 @@ def upgrade() -> None:
sa.Column('status', sa.Enum('ACTIVE', 'ARCHIVED', name='placementstatus'), nullable=False),
sa.Column('subscriptions_count', sa.Integer(), server_default='0', nullable=False),
sa.Column('views_count', sa.Integer(), nullable=True),
sa.Column('views_availability', sa.Enum('UNKNOWN', 'AVAILABLE', 'UNAVAILABLE', 'MANUAL', name='postviewsavailability'), nullable=False),
sa.Column(
'views_availability',
sa.Enum('UNKNOWN', 'AVAILABLE', 'UNAVAILABLE', 'MANUAL', name='postviewsavailability'),
nullable=False,
),
sa.Column('last_views_fetch_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('id', sa.Uuid(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['creative_id'], ['creative.id'], ),
sa.ForeignKeyConstraint(['external_channel_id'], ['external_channel.id'], ),
sa.ForeignKeyConstraint(['target_channel_id'], ['target_channel.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
sa.ForeignKeyConstraint(
['creative_id'],
['creative.id'],
),
sa.ForeignKeyConstraint(
['external_channel_id'],
['external_channel.id'],
),
sa.ForeignKeyConstraint(
['target_channel_id'],
['target_channel.id'],
),
sa.ForeignKeyConstraint(
['user_id'],
['user.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_placement_creative_id'), 'placement', ['creative_id'], unique=False)
op.create_index(op.f('ix_placement_external_channel_id'), 'placement', ['external_channel_id'], unique=False)
op.create_index(op.f('ix_placement_target_channel_id'), 'placement', ['target_channel_id'], unique=False)
op.create_index(op.f('ix_placement_user_id'), 'placement', ['user_id'], unique=False)
op.create_table('placement_views_history',
op.create_table(
'placement_views_history',
sa.Column('placement_id', sa.Uuid(), nullable=False),
sa.Column('views_count', sa.Integer(), nullable=False),
sa.Column('fetched_at', sa.DateTime(timezone=True), nullable=False),
@@ -149,12 +190,20 @@ def upgrade() -> None:
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['placement_id'], ['placement.id'], ),
sa.PrimaryKeyConstraint('id')
sa.ForeignKeyConstraint(
['placement_id'],
['placement.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_placement_views_history_fetched_at'), 'placement_views_history', ['fetched_at'], unique=False)
op.create_index(op.f('ix_placement_views_history_placement_id'), 'placement_views_history', ['placement_id'], unique=False)
op.create_table('subscription',
op.create_index(
op.f('ix_placement_views_history_fetched_at'), 'placement_views_history', ['fetched_at'], unique=False
)
op.create_index(
op.f('ix_placement_views_history_placement_id'), 'placement_views_history', ['placement_id'], unique=False
)
op.create_table(
'subscription',
sa.Column('placement_id', sa.Uuid(), nullable=False),
sa.Column('subscriber_id', sa.Uuid(), nullable=False),
sa.Column('invite_link', sa.String(), nullable=False),
@@ -162,9 +211,15 @@ def upgrade() -> None:
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['placement_id'], ['placement.id'], ),
sa.ForeignKeyConstraint(['subscriber_id'], ['subscriber.id'], ),
sa.PrimaryKeyConstraint('id')
sa.ForeignKeyConstraint(
['placement_id'],
['placement.id'],
),
sa.ForeignKeyConstraint(
['subscriber_id'],
['subscriber.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_subscription_invite_link'), 'subscription', ['invite_link'], unique=False)
op.create_index(op.f('ix_subscription_placement_id'), 'subscription', ['placement_id'], unique=False)

40
src/adapter/postgres.py
View File

@@ -361,6 +361,46 @@ class Postgres(DatabaseBase):
result = await self.session.execute(q)
return result.scalar_one_or_none()
async def update_subscription(self, subscription: domain.Subscription) -> domain.Subscription:
await self.session.flush()
await self.session.refresh(subscription, ['placement', 'subscriber'])
return subscription
async def get_active_subscriptions_by_subscriber_and_channel(
self, subscriber_id: uuid.UUID, channel_telegram_id: int
) -> list[domain.Subscription]:
q = (
select(domain.Subscription)
.join(domain.Placement, domain.Subscription.placement_id == domain.Placement.id)
.join(domain.TargetChannel, domain.Placement.target_channel_id == domain.TargetChannel.id)
.where(
domain.Subscription.subscriber_id == subscriber_id,
domain.TargetChannel.telegram_id == channel_telegram_id,
domain.Subscription.status == domain.SubscriptionStatus.ACTIVE,
)
.options(selectinload(domain.Subscription.placement), selectinload(domain.Subscription.subscriber))
)
result = await self.session.execute(q)
return list(result.scalars().all())
async def get_active_subscription_by_subscriber_and_channel(
self, subscriber_id: uuid.UUID, channel_telegram_id: int
) -> domain.Subscription | None:
"""Получить активную подписку пользователя на канал (должна быть только одна благодаря UNIQUE constraint)."""
q = (
select(domain.Subscription)
.join(domain.Placement, domain.Subscription.placement_id == domain.Placement.id)
.join(domain.TargetChannel, domain.Placement.target_channel_id == domain.TargetChannel.id)
.where(
domain.Subscription.subscriber_id == subscriber_id,
domain.TargetChannel.telegram_id == channel_telegram_id,
domain.Subscription.status == domain.SubscriptionStatus.ACTIVE,
)
.options(selectinload(domain.Subscription.placement), selectinload(domain.Subscription.subscriber))
)
result = await self.session.execute(q)
return result.scalar_one_or_none()
async def create_placement_views_history(
self, history: domain.PlacementViewsHistory
) -> domain.PlacementViewsHistory:

30
src/controller/telegram_callback/chat_member_updated.py
View File

@@ -10,11 +10,6 @@ log = logging.getLogger(__name__)
@telegram_callback_router.chat_member()
async def on_chat_member_updated(event: ChatMemberUpdated) -> None:
"""
Обрабатывает события подписки пользователей в канал через открытую invite_link.
Событие приходит когда пользователь вступает в канал напрямую (без запроса на вступление).
"""
if event.chat.type not in {'channel', 'supergroup'}:
return
@@ -30,19 +25,22 @@ async def on_chat_member_updated(event: ChatMemberUpdated) -> None:
is_now_member = new_status in ('member', 'administrator', 'creator')
user_joined = was_not_member and is_now_member
if not user_joined:
return
# Получаем invite_link, по которой пользователь присоединился
invite_link = event.invite_link
if not invite_link or not invite_link.invite_link:
log.debug('No invite_link in chat member update event')
return
# Проверяем что пользователь покинул канал
was_member = old_status in ('member', 'administrator', 'creator', 'restricted')
is_now_not_member = new_status in ('left', 'kicked')
user_left = was_member and is_now_not_member
usecase = dependencies.get_usecase()
if user_joined:
invite_link = event.invite_link.invite_link if event.invite_link else None
if not invite_link:
log.debug('No invite_link in chat member update event')
return
await usecase.handle_subscription(
user_telegram_id=event.from_user.id,
username=event.from_user.username,
invite_link=invite_link.invite_link,
user_telegram_id=event.from_user.id, username=event.from_user.username, invite_link=invite_link
)
elif user_left:
await usecase.handle_unsubscription(user_telegram_id=event.from_user.id, channel_telegram_id=event.chat.id)

3
src/domain/__init__.py
View File

@@ -11,6 +11,7 @@ __all__ = (
'TargetChannelStatus',
'CreativeStatus',
'PlacementStatus',
'SubscriptionStatus',
'InviteLinkType',
'PostViewsAvailability',
'LoginToken',
@@ -49,6 +50,6 @@ from .login_token import LoginToken
from .placement import InviteLinkType, Placement, PlacementStatus, PostViewsAvailability
from .placement_views_history import PlacementViewsHistory
from .subscriber import Subscriber
from .subscription import Subscription
from .subscription import Subscription, SubscriptionStatus
from .target_channel import TargetChannel, TargetChannelStatus
from .user import User

1
src/domain/base.py
View File

@@ -8,7 +8,6 @@ from sqlalchemy.types import DateTime
def _camel_to_snake(name: str) -> str:
"""Convert CamelCase to snake_case."""
name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
return re.sub('([a-z0-9])([A-Z])', r'\1_\2', name).lower()

10
src/domain/subscription.py
View File

@@ -1,4 +1,6 @@
import datetime
import uuid
from enum import StrEnum
from typing import TYPE_CHECKING
from sqlalchemy import ForeignKey
@@ -11,10 +13,18 @@ if TYPE_CHECKING:
from .subscriber import Subscriber
class SubscriptionStatus(StrEnum):
ACTIVE = 'active'
UNSUBSCRIBED = 'unsubscribed'
class Subscription(domain.Base):
placement_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('placement.id'), index=True)
subscriber_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('subscriber.id'), index=True)
target_channel_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('target_channel.id'), index=True)
invite_link: Mapped[str] = mapped_column(index=True) # По которой пришёл пользователь
status: Mapped[SubscriptionStatus] = mapped_column(default=SubscriptionStatus.ACTIVE)
unsubscribed_at: Mapped[datetime.datetime | None]
# Relationships
placement: Mapped['Placement'] = relationship(back_populates='subscriptions')

1
src/dto/user.py
View File

@@ -7,4 +7,3 @@ class UserOutput(pydantic.BaseModel):
id: uuid.UUID
telegram_id: int
username: str | None

12
src/usecase/__init__.py
View File

@@ -26,6 +26,7 @@ from .placement.get_placement import get_placement
from .placement.get_placements import get_placements
from .placement.update_placement import update_placement
from .subscription.handle_subscription import handle_subscription
from .subscription.handle_unsubscription import handle_unsubscription
from .target_channel.connect_target_chan import connect_target_chan
from .target_channel.disconnect_target_chan import disconnect_target_chan
from .target_channel.disconnect_target_chan_by_tg_id import disconnect_target_chan_by_tg_id
@@ -131,6 +132,16 @@ class Database(typing.Protocol):
self, subscriber_id: UUID, placement_id: UUID
) -> domain.Subscription | None: ...
async def update_subscription(self, subscription: domain.Subscription) -> domain.Subscription: ...
async def get_active_subscriptions_by_subscriber_and_channel(
self, subscriber_id: UUID, channel_telegram_id: int
) -> list[domain.Subscription]: ...
async def get_active_subscription_by_subscriber_and_channel(
self, subscriber_id: UUID, channel_telegram_id: int
) -> domain.Subscription | None: ...
async def create_placement_views_history(
self, history: domain.PlacementViewsHistory
) -> domain.PlacementViewsHistory: ...
@@ -195,6 +206,7 @@ class Usecase:
update_placement = update_placement
delete_placement = delete_placement
handle_subscription = handle_subscription
handle_unsubscription = handle_unsubscription
fetch_views = fetch_views_manually
run_views_worker_cycle = run_views_worker_cycle
get_views_history = get_views_history

1
src/usecase/get_me.py
View File

@@ -19,4 +19,3 @@ async def get_me(self: 'Usecase', user_id: uuid.UUID) -> dto.UserOutput:
telegram_id=user.telegram_id,
username=user.username,
)

50
src/usecase/subscription/handle_subscription.py
View File

@@ -17,20 +17,12 @@ async def handle_subscription(
first_name: str | None = None,
last_name: str | None = None,
) -> None:
"""
Обрабатывает подписку пользователя в целевой канал.
Находит закуп по invite_link и создаёт запись о подписке,
инкрементируя счётчик подписок у закупа.
"""
async with self.database.transaction():
# Находим закуп по пригласительной ссылке
placement = await self.database.get_placement_by_invite_link(invite_link)
if not placement:
log.warning('Placement not found for invite_link: %s', invite_link)
return
# Создаём или обновляем подписчика
subscriber = domain.Subscriber(
telegram_id=user_telegram_id,
username=username,
@@ -39,27 +31,53 @@ async def handle_subscription(
)
subscriber = await self.database.upsert_subscriber(subscriber)
# Проверяем, не создана ли уже подписка для этого пользователя
existing_subscription = await self.database.get_subscription_by_subscriber_and_placement(
subscriber.id, placement.id
active_subscription = await self.database.get_active_subscription_by_subscriber_and_channel(
subscriber.id, placement.target_channel.telegram_id
)
if existing_subscription:
log.info(
'Subscription already exists for subscriber %s and placement %s',
if active_subscription:
# Пользователь уже подписан на канал через другой placement
# Это не должно случиться (Telegram не даст подписаться дважды),
# но если случилось - логируем и игнорируем
log.warning(
'User %s (telegram_id: %s) already has active subscription to channel %s via placement %s, '
'ignoring new subscription attempt via placement %s',
subscriber.id,
user_telegram_id,
placement.target_channel_id,
active_subscription.placement_id,
placement.id,
)
return
# Проверяем, была ли раньше подписка через ЭТОТ placement (для реактивации)
existing_sub = await self.database.get_subscription_by_subscriber_and_placement(subscriber.id, placement.id)
if existing_sub and existing_sub.status == domain.SubscriptionStatus.UNSUBSCRIBED:
# Реактивируем старую подписку через тот же placement
existing_sub.status = domain.SubscriptionStatus.ACTIVE
existing_sub.unsubscribed_at = None
await self.database.update_subscription(existing_sub)
placement.subscriptions_count += 1
await self.database.update_placement(placement)
log.info(
'Subscription reactivated: subscriber %s (telegram_id: %s) resubscribed via same placement %s',
subscriber.id,
user_telegram_id,
placement.id,
)
return
# Создаём запись о подписке
subscription = domain.Subscription(
placement_id=placement.id,
subscriber_id=subscriber.id,
target_channel_id=placement.target_channel_id,
invite_link=invite_link,
)
await self.database.create_subscription(subscription)
# Увеличиваем счётчик подписок у закупа
placement.subscriptions_count += 1
await self.database.update_placement(placement)

49
src/usecase/subscription/handle_unsubscription.py Normal file
View File

@@ -0,0 +1,49 @@
import datetime
import logging
from typing import TYPE_CHECKING
from src import domain
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def handle_unsubscription(self: 'Usecase', user_telegram_id: int, channel_telegram_id: int) -> None:
async with self.database.transaction():
subscriber = await self.database.get_subscriber(user_telegram_id)
if not subscriber:
log.warning('Subscriber not found for telegram_id: %s', user_telegram_id)
return
subscriptions = await self.database.get_active_subscriptions_by_subscriber_and_channel(
subscriber.id, channel_telegram_id
)
if not subscriptions:
log.info(
'No active subscriptions found for subscriber %s (telegram_id: %s) in channel %s',
subscriber.id,
user_telegram_id,
channel_telegram_id,
)
return
for subscription in subscriptions:
subscription.status = domain.SubscriptionStatus.UNSUBSCRIBED
subscription.unsubscribed_at = datetime.datetime.now(datetime.UTC)
await self.database.update_subscription(subscription)
# Декрементируем счётчик подписок у закупа
placement = subscription.placement
if placement.subscriptions_count > 0:
placement.subscriptions_count -= 1
await self.database.update_placement(placement)
log.info(
'Subscription marked as unsubscribed: subscriber %s (telegram_id: %s) unsubscribed from placement %s',
subscriber.id,
user_telegram_id,
placement.id,
)

4
src/usecase/views/run_views_worker_cycle.py
View File

@@ -14,9 +14,7 @@ if TYPE_CHECKING:
log = logging.getLogger(__name__)
async def _process_placement(
self: 'Usecase', placement_id: uuid.UUID, user_id: uuid.UUID, ad_post_url: str
) -> None:
async def _process_placement(self: 'Usecase', placement_id: uuid.UUID, user_id: uuid.UUID, ad_post_url: str) -> None:
views_count = await self.telegram_parser.get_post_views(ad_post_url)
fetched_at = datetime.datetime.now(datetime.UTC)