diff --git a/migration/versions/3b8259700b61_add_subscription_status_and_.py b/migration/versions/3b8259700b61_add_subscription_status_and_.py new file mode 100644 index 0000000..bcf4e59 --- /dev/null +++ b/migration/versions/3b8259700b61_add_subscription_status_and_.py @@ -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 ### diff --git a/migration/versions/52ad7cc9bf1e_add_unique_constraint_for_active_.py b/migration/versions/52ad7cc9bf1e_add_unique_constraint_for_active_.py new file mode 100644 index 0000000..2943905 --- /dev/null +++ b/migration/versions/52ad7cc9bf1e_add_unique_constraint_for_active_.py @@ -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') diff --git a/migration/versions/f05f85d10837_init.py b/migration/versions/f05f85d10837_init.py index efb5a42..4f119f1 100644 --- a/migration/versions/f05f85d10837_init.py +++ b/migration/versions/f05f85d10837_init.py @@ -1,10 +1,11 @@ """init Revision ID: f05f85d10837 -Revises: +Revises: Create Date: 2025-11-13 01:06:51.623272 """ + from collections.abc import Sequence import sqlalchemy as sa @@ -19,152 +20,206 @@ depends_on: str | Sequence[str] | None = None def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### - 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), - sa.Column('last_name', 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') + 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), + sa.Column('last_name', 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'), ) op.create_index(op.f('ix_subscriber_telegram_id'), 'subscriber', ['telegram_id'], unique=True) - 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') + 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'), ) op.create_index(op.f('ix_user_telegram_id'), 'user', ['telegram_id'], unique=True) - 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), - sa.Column('description', sa.String(), nullable=True), - sa.Column('subscribers_count', sa.Integer(), nullable=True), - sa.Column('user_id', sa.Uuid(), nullable=False), - 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(['user_id'], ['user.id'], ), - sa.PrimaryKeyConstraint('id') + 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), + sa.Column('description', sa.String(), nullable=True), + sa.Column('subscribers_count', sa.Integer(), nullable=True), + sa.Column('user_id', sa.Uuid(), nullable=False), + 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( + ['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', - 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), - sa.Column('used_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(['user_id'], ['user.id'], ), - sa.PrimaryKeyConstraint('id') + 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), + sa.Column('used_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( + ['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', - sa.Column('telegram_id', sa.BigInteger(), nullable=False), - sa.Column('title', sa.String(), nullable=False), - sa.Column('username', sa.String(), nullable=True), - sa.Column('status', sa.Enum('ACTIVE', 'INACTIVE', 'ARCHIVED', name='targetchannelstatus'), nullable=False), - sa.Column('user_id', sa.Uuid(), nullable=False), - 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(['user_id'], ['user.id'], ), - sa.PrimaryKeyConstraint('id') + 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), + sa.Column('status', sa.Enum('ACTIVE', 'INACTIVE', 'ARCHIVED', name='targetchannelstatus'), nullable=False), + sa.Column('user_id', sa.Uuid(), nullable=False), + 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( + ['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', - 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), - sa.Column('placements_count', sa.Integer(), server_default='0', nullable=False), - sa.Column('user_id', sa.Uuid(), nullable=False), - sa.Column('target_channel_id', sa.Uuid(), nullable=False), - 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(['target_channel_id'], ['target_channel.id'], ), - sa.ForeignKeyConstraint(['user_id'], ['user.id'], ), - sa.PrimaryKeyConstraint('id') + 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), + sa.Column('placements_count', sa.Integer(), server_default='0', nullable=False), + sa.Column('user_id', sa.Uuid(), nullable=False), + sa.Column('target_channel_id', sa.Uuid(), nullable=False), + 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( + ['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', - 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') + 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'), ) - 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), - sa.Column('user_id', sa.Uuid(), nullable=False), - sa.Column('placement_date', sa.DateTime(timezone=True), nullable=False), - sa.Column('cost', sa.Float(), nullable=True), - sa.Column('comment', sa.String(), nullable=True), - sa.Column('ad_post_url', sa.String(), nullable=True), - sa.Column('invite_link_type', sa.Enum('PUBLIC', 'APPROVAL', name='invitelinktype'), nullable=False), - sa.Column('invite_link', sa.String(), nullable=False), - 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('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') + 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), + sa.Column('user_id', sa.Uuid(), nullable=False), + sa.Column('placement_date', sa.DateTime(timezone=True), nullable=False), + sa.Column('cost', sa.Float(), nullable=True), + sa.Column('comment', sa.String(), nullable=True), + sa.Column('ad_post_url', sa.String(), nullable=True), + sa.Column('invite_link_type', sa.Enum('PUBLIC', 'APPROVAL', name='invitelinktype'), nullable=False), + sa.Column('invite_link', sa.String(), nullable=False), + 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('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'), ) 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', - 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), - 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(['placement_id'], ['placement.id'], ), - sa.PrimaryKeyConstraint('id') + 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), + 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( + ['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', - sa.Column('placement_id', sa.Uuid(), nullable=False), - sa.Column('subscriber_id', sa.Uuid(), nullable=False), - sa.Column('invite_link', sa.String(), nullable=False), - 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(['placement_id'], ['placement.id'], ), - sa.ForeignKeyConstraint(['subscriber_id'], ['subscriber.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', + sa.Column('placement_id', sa.Uuid(), nullable=False), + sa.Column('subscriber_id', sa.Uuid(), nullable=False), + sa.Column('invite_link', sa.String(), nullable=False), + 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( + ['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) diff --git a/src/adapter/postgres.py b/src/adapter/postgres.py index 2e85ea5..283eb6e 100644 --- a/src/adapter/postgres.py +++ b/src/adapter/postgres.py @@ -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: diff --git a/src/controller/telegram_callback/chat_member_updated.py b/src/controller/telegram_callback/chat_member_updated.py index 683610a..6bd5ca4 100644 --- a/src/controller/telegram_callback/chat_member_updated.py +++ b/src/controller/telegram_callback/chat_member_updated.py @@ -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() - await usecase.handle_subscription( - user_telegram_id=event.from_user.id, - username=event.from_user.username, - invite_link=invite_link.invite_link, - ) + 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 + ) + + elif user_left: + await usecase.handle_unsubscription(user_telegram_id=event.from_user.id, channel_telegram_id=event.chat.id) diff --git a/src/domain/__init__.py b/src/domain/__init__.py index 36bd695..55f0464 100644 --- a/src/domain/__init__.py +++ b/src/domain/__init__.py @@ -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 diff --git a/src/domain/base.py b/src/domain/base.py index c124136..ff0af26 100644 --- a/src/domain/base.py +++ b/src/domain/base.py @@ -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() diff --git a/src/domain/subscription.py b/src/domain/subscription.py index 0ff0c3b..ebc748a 100644 --- a/src/domain/subscription.py +++ b/src/domain/subscription.py @@ -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') diff --git a/src/dto/user.py b/src/dto/user.py index 2fed355..f327e9e 100644 --- a/src/dto/user.py +++ b/src/dto/user.py @@ -7,4 +7,3 @@ class UserOutput(pydantic.BaseModel): id: uuid.UUID telegram_id: int username: str | None - diff --git a/src/usecase/__init__.py b/src/usecase/__init__.py index 8fdc954..605deac 100644 --- a/src/usecase/__init__.py +++ b/src/usecase/__init__.py @@ -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 diff --git a/src/usecase/get_me.py b/src/usecase/get_me.py index 86701f1..9a95868 100644 --- a/src/usecase/get_me.py +++ b/src/usecase/get_me.py @@ -19,4 +19,3 @@ async def get_me(self: 'Usecase', user_id: uuid.UUID) -> dto.UserOutput: telegram_id=user.telegram_id, username=user.username, ) - diff --git a/src/usecase/subscription/handle_subscription.py b/src/usecase/subscription/handle_subscription.py index 02d4de1..83626b2 100644 --- a/src/usecase/subscription/handle_subscription.py +++ b/src/usecase/subscription/handle_subscription.py @@ -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) diff --git a/src/usecase/subscription/handle_unsubscription.py b/src/usecase/subscription/handle_unsubscription.py new file mode 100644 index 0000000..3268c9b --- /dev/null +++ b/src/usecase/subscription/handle_unsubscription.py @@ -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, + ) diff --git a/src/usecase/views/run_views_worker_cycle.py b/src/usecase/views/run_views_worker_cycle.py index ceff28e..71c09d0 100644 --- a/src/usecase/views/run_views_worker_cycle.py +++ b/src/usecase/views/run_views_worker_cycle.py @@ -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)