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

feat: subs and views

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2025-11-10 17:08:56 +03:00
parent 2549242b06
commit 0bed5c266d
34 changed files with 1706 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -125,3 +125,5 @@ cython_debug/
# PyPI configuration file
.pypirc
.uv_cache/

51
migration/versions/e7ab0b38a8bf_add_subscriptions_table.py Normal file
View File

@@ -0,0 +1,51 @@
"""add subscriptions table
Revision ID: e7ab0b38a8bf
Revises: 6517eda4a92e
Create Date: 2025-11-10 16:48:25.310928
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = 'e7ab0b38a8bf'
down_revision: str | None = '6517eda4a92e'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
'subscriptions',
sa.Column('purchase_id', sa.Uuid(), nullable=False),
sa.Column('user_telegram_id', sa.BigInteger(), nullable=False),
sa.Column('username', sa.String(), nullable=True),
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(
['purchase_id'],
['purchases.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_subscriptions_invite_link'), 'subscriptions', ['invite_link'], unique=False)
op.create_index(op.f('ix_subscriptions_purchase_id'), 'subscriptions', ['purchase_id'], unique=False)
op.create_index(op.f('ix_subscriptions_user_telegram_id'), 'subscriptions', ['user_telegram_id'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_subscriptions_user_telegram_id'), table_name='subscriptions')
op.drop_index(op.f('ix_subscriptions_purchase_id'), table_name='subscriptions')
op.drop_index(op.f('ix_subscriptions_invite_link'), table_name='subscriptions')
op.drop_table('subscriptions')
# ### end Alembic commands ###

67
migration/versions/fafe143ee0f4_add_views_tracking.py Normal file
View File

@@ -0,0 +1,67 @@
"""add views tracking
Revision ID: fafe143ee0f4
Revises: e7ab0b38a8bf
Create Date: 2025-11-10 16:57:28.652101
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = 'fafe143ee0f4'
down_revision: str | None = 'e7ab0b38a8bf'
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 тип для статуса доступности просмотров ПЕРЕД использованием
op.execute("CREATE TYPE postviewsavailability AS ENUM ('UNKNOWN', 'AVAILABLE', 'UNAVAILABLE', 'MANUAL')")
op.create_table(
'viewssnapshots',
sa.Column('purchase_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(
['purchase_id'],
['purchases.id'],
),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_viewssnapshots_fetched_at'), 'viewssnapshots', ['fetched_at'], unique=False)
op.create_index(op.f('ix_viewssnapshots_purchase_id'), 'viewssnapshots', ['purchase_id'], unique=False)
op.add_column(
'purchases',
sa.Column(
'views_availability',
sa.Enum('UNKNOWN', 'AVAILABLE', 'UNAVAILABLE', 'MANUAL', name='postviewsavailability'),
nullable=False,
server_default='UNKNOWN',
),
)
op.add_column('purchases', sa.Column('last_views_fetch_at', sa.DateTime(timezone=True), nullable=True))
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('purchases', 'last_views_fetch_at')
op.drop_column('purchases', 'views_availability')
op.drop_index(op.f('ix_viewssnapshots_purchase_id'), table_name='viewssnapshots')
op.drop_index(op.f('ix_viewssnapshots_fetched_at'), table_name='viewssnapshots')
op.drop_table('viewssnapshots')
# Удаляём enum тип
op.execute('DROP TYPE IF EXISTS postviewsavailability')
# ### end Alembic commands ###

54
src/adapter/postgres.py
View File

@@ -299,3 +299,57 @@ class Postgres(DatabaseBase):
)
result = await self.session.execute(stmt)
return result.scalar_one_or_none() is not None
async def get_purchase_by_invite_link(self, invite_link: str) -> domain.Purchase | None:
stmt = (
select(domain.Purchase)
.options(
selectinload(domain.Purchase.target_channel),
selectinload(domain.Purchase.external_channel),
selectinload(domain.Purchase.creative),
)
.where(domain.Purchase.invite_link == invite_link)
)
result = await self.session.execute(stmt)
return result.scalar_one_or_none()
async def create_subscription(self, subscription: domain.Subscription) -> domain.Subscription:
self.session.add(subscription)
await self.session.flush()
await self.session.refresh(subscription, ['purchase'])
return subscription
async def get_subscription_by_user_and_purchase(
self, user_telegram_id: int, purchase_id: uuid.UUID
) -> domain.Subscription | None:
stmt = select(domain.Subscription).where(
domain.Subscription.user_telegram_id == user_telegram_id,
domain.Subscription.purchase_id == purchase_id,
)
result = await self.session.execute(stmt)
return result.scalar_one_or_none()
async def create_views_snapshot(self, snapshot: domain.ViewsSnapshot) -> domain.ViewsSnapshot:
self.session.add(snapshot)
await self.session.flush()
await self.session.refresh(snapshot, ['purchase'])
return snapshot
async def get_views_history(
self,
purchase_id: uuid.UUID,
*,
from_date: datetime.datetime | None = None,
to_date: datetime.datetime | None = None,
) -> list[domain.ViewsSnapshot]:
stmt = select(domain.ViewsSnapshot).where(domain.ViewsSnapshot.purchase_id == purchase_id)
if from_date:
stmt = stmt.where(domain.ViewsSnapshot.fetched_at >= from_date)
if to_date:
stmt = stmt.where(domain.ViewsSnapshot.fetched_at <= to_date)
stmt = stmt.order_by(domain.ViewsSnapshot.fetched_at.asc())
result = await self.session.execute(stmt)
return list(result.scalars().all())

31
src/adapter/telegram.py
View File

@@ -26,3 +26,34 @@ class Telegram(TelegramBase):
creates_join_request=requires_approval,
)
return invite_link.invite_link
async def get_post_views(self, post_url: str) -> int | None:
"""
Получить количество просмотров поста по URL.
ЗАГЛУШКА: В текущей реализации возвращает None.
Для реальной работы необходимо:
1. Интегрировать TDLib/MTProto клиент
2. Парсить post_url (формат: t.me/<username>/<message_id> или t.me/c/<chat_id>/<message_id>)
3. Вызывать channels.getMessages через TDLib
4. Извлекать поле views из ответа
Returns:
int | None: Количество просмотров или None если не удалось получить
"""
log.warning(
'get_post_views is not implemented yet (TDLib integration required). URL: %s',
post_url,
)
# TODO: Implement TDLib integration
# Example pseudo-code:
# try:
# username, message_id = parse_telegram_url(post_url)
# channel = await tdlib.resolve_username(username)
# message = await tdlib.get_message(channel.id, message_id)
# return message.views
# except Exception as e:
# log.error('Failed to get post views: %s', e)
# return None
return None

2
src/controller/http/__init__.py
View File

@@ -5,6 +5,7 @@ from src.controller.http.creatives import creatives_router
from src.controller.http.external_channels import external_channels_router
from src.controller.http.purchases import purchases_router
from src.controller.http.target_channels import target_channels_router
from src.controller.http.views import views_router
api_router = APIRouter()
@@ -17,5 +18,6 @@ api_v1_router.include_router(target_channels_router)
api_v1_router.include_router(external_channels_router)
api_v1_router.include_router(creatives_router)
api_v1_router.include_router(purchases_router)
api_v1_router.include_router(views_router)
api_router.include_router(api_v1_router)

59
src/controller/http/views.py Normal file
View File

@@ -0,0 +1,59 @@
import datetime
import uuid
from typing import Annotated
from fastapi import Depends
from fastapi.routing import APIRouter
from src import dependencies, dto
from src.adapter.jwt import JWTPayload
views_router = APIRouter(prefix='/purchases/{purchase_id}/views', tags=['views'])
@views_router.get('/history')
async def get_views_history(
purchase_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
from_date: datetime.datetime | None = None,
to_date: datetime.datetime | None = None,
) -> dto.GetViewsHistoryOutput:
"""Получить историю просмотров для закупа."""
input_data = dto.GetViewsHistoryInput(
purchase_id=purchase_id,
user_id=current_user.user_id,
from_date=from_date,
to_date=to_date,
)
return await dependencies.get_usecase().get_views_history(input=input_data)
@views_router.post('/fetch')
async def fetch_views(
purchase_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> dto.FetchViewsOutput:
"""Получить актуальные просмотры через Telegram API."""
input_data = dto.FetchViewsInput(
purchase_id=purchase_id,
user_id=current_user.user_id,
)
return await dependencies.get_usecase().fetch_views(input=input_data)
@views_router.post('/manual')
async def update_views_manually(
purchase_id: uuid.UUID,
views_count: int,
current_user: Annotated[JWTPayload, Depends(dependencies.get_current_user)],
) -> dto.PurchaseOutput:
"""Обновить просмотры вручную."""
input_data = dto.UpdateViewsManuallyInput(
purchase_id=purchase_id,
user_id=current_user.user_id,
views_count=views_count,
)
return await dependencies.get_usecase().update_views_manually(input=input_data)

2
src/controller/telegram_callback/__init__.py
View File

@@ -3,4 +3,4 @@ from aiogram import Router
telegram_callback_router = Router()
# Import handlers to register them
from . import my_chat_member, start_with_login # noqa: E402, F401
from . import chat_join_request, chat_member_updated, my_chat_member, start_with_login # noqa: E402, F401

37
src/controller/telegram_callback/chat_join_request.py Normal file
View File

@@ -0,0 +1,37 @@
import logging
from aiogram.types import ChatJoinRequest
from src import dependencies
from src.controller.telegram_callback import telegram_callback_router
log = logging.getLogger(__name__)
@telegram_callback_router.chat_join_request()
async def on_chat_join_request(event: ChatJoinRequest) -> None:
"""
Обрабатывает запросы на вступление в канал через приватную invite_link с одобрением.
Событие приходит когда пользователь отправляет запрос на вступление в канал.
"""
if event.chat.type not in {'channel', 'supergroup'}:
return
if not event.from_user:
log.error('Failed to get user data from chat join request')
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 join request event')
return
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,
)

48
src/controller/telegram_callback/chat_member_updated.py Normal file
View File

@@ -0,0 +1,48 @@
import logging
from aiogram.types import ChatMemberUpdated
from src import dependencies
from src.controller.telegram_callback import telegram_callback_router
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
if not event.from_user:
log.error('Failed to get user data from chat member update')
return
old_status = event.old_chat_member.status
new_status = event.new_chat_member.status
# Проверяем что пользователь присоединился к каналу
was_not_member = old_status in ('left', 'kicked')
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
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,
)

7
src/domain/__init__.py
View File

@@ -5,10 +5,13 @@ __all__ = (
'ExternalChannel',
'Creative',
'Purchase',
'Subscription',
'ViewsSnapshot',
'TargetChannelStatus',
'CreativeStatus',
'PurchaseStatus',
'InviteLinkType',
'PostViewsAvailability',
'LoginToken',
'UserNotFound',
'LoginTokenNotFound',
@@ -42,6 +45,8 @@ from .error import (
)
from .external_channel import ExternalChannel
from .login_token import LoginToken
from .purchase import InviteLinkType, Purchase, PurchaseStatus
from .purchase import InviteLinkType, PostViewsAvailability, Purchase, PurchaseStatus
from .subscription import Subscription
from .target_channel import TargetChannel, TargetChannelStatus
from .user import User
from .views_snapshot import ViewsSnapshot

15
src/domain/purchase.py
View File

@@ -24,6 +24,15 @@ class InviteLinkType(StrEnum):
APPROVAL = 'approval' # с одобрением ботом
class PostViewsAvailability(StrEnum):
"""Статус доступности получения просмотров поста."""
UNKNOWN = 'unknown' # Ещё не проверялось
AVAILABLE = 'available' # Просмотры доступны
UNAVAILABLE = 'unavailable' # Нет доступа / битая ссылка / приватный канал
MANUAL = 'manual' # Просмотры вводятся вручную
class Purchase(domain.Base):
# Основные поля
target_channel_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('targetchannels.id'), index=True)
@@ -44,7 +53,11 @@ class Purchase(domain.Base):
# Метрики (будут обновляться из Telegram API)
subscriptions_count: Mapped[int] = mapped_column(default=0, server_default='0')
views_count: Mapped[int | None] # просмотры рекламного поста (если доступны)
views_count: Mapped[int | None] # последние известные просмотры (кэш из последнего snapshot)
# Статус доступности просмотров
views_availability: Mapped[PostViewsAvailability] = mapped_column(default=PostViewsAvailability.UNKNOWN)
last_views_fetch_at: Mapped[datetime.datetime | None] # когда последний раз получали просмотры
# Relationships
target_channel: Mapped['TargetChannel'] = relationship()

22
src/domain/subscription.py Normal file
View File

@@ -0,0 +1,22 @@
import uuid
from typing import TYPE_CHECKING
from sqlalchemy import BigInteger, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src import domain
if TYPE_CHECKING:
from .purchase import Purchase
class Subscription(domain.Base):
"""Подписка пользователя в целевой канал через конкретный закуп."""
purchase_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('purchases.id'), index=True)
user_telegram_id: Mapped[int] = mapped_column(BigInteger, index=True)
username: Mapped[str | None]
invite_link: Mapped[str] = mapped_column(index=True) # По которой пришёл пользователь
# Relationship
purchase: Mapped['Purchase'] = relationship()

22
src/domain/views_snapshot.py Normal file
View File

@@ -0,0 +1,22 @@
import datetime
import uuid
from typing import TYPE_CHECKING
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src import domain
if TYPE_CHECKING:
from .purchase import Purchase
class ViewsSnapshot(domain.Base):
"""Снимок просмотров рекламного поста в определенный момент времени."""
purchase_id: Mapped[uuid.UUID] = mapped_column(ForeignKey('purchases.id'), index=True)
views_count: Mapped[int] # Количество просмотров на момент снимка
fetched_at: Mapped[datetime.datetime] = mapped_column(index=True) # Когда получены данные
# Relationship
purchase: Mapped['Purchase'] = relationship()

14
src/dto/__init__.py
View File

@@ -34,6 +34,12 @@ __all__ = (
'UpdatePurchaseInput',
'ArchivePurchaseInput',
'DeletePurchaseInput',
'ViewsSnapshotOutput',
'GetViewsHistoryInput',
'GetViewsHistoryOutput',
'FetchViewsInput',
'FetchViewsOutput',
'UpdateViewsManuallyInput',
)
from .creative import (
@@ -78,3 +84,11 @@ from .target_channel import (
)
from .telegram_login import TelegramLoginInput
from .validate_login_token import ValidateLoginTokenInput, ValidateLoginTokenOutput
from .views import (
FetchViewsInput,
FetchViewsOutput,
GetViewsHistoryInput,
GetViewsHistoryOutput,
UpdateViewsManuallyInput,
ViewsSnapshotOutput,
)

2
src/dto/purchase.py
View File

@@ -23,6 +23,8 @@ class PurchaseOutput(pydantic.BaseModel):
status: domain.PurchaseStatus
subscriptions_count: int
views_count: int | None
views_availability: domain.PostViewsAvailability
last_views_fetch_at: datetime.datetime | None
created_at: datetime.datetime

56
src/dto/views.py Normal file
View File

@@ -0,0 +1,56 @@
import datetime
import uuid
import pydantic
from src import domain
class ViewsSnapshotOutput(pydantic.BaseModel):
"""Снимок просмотров."""
id: uuid.UUID
purchase_id: uuid.UUID
views_count: int
fetched_at: datetime.datetime
created_at: datetime.datetime
class GetViewsHistoryInput(pydantic.BaseModel):
"""Получить историю просмотров для закупа."""
purchase_id: uuid.UUID
user_id: uuid.UUID
from_date: datetime.datetime | None = None # Фильтр: с какой даты
to_date: datetime.datetime | None = None # Фильтр: по какую дату
class GetViewsHistoryOutput(pydantic.BaseModel):
"""История просмотров."""
snapshots: list[ViewsSnapshotOutput]
class FetchViewsInput(pydantic.BaseModel):
"""Запросить обновление просмотров для закупа."""
purchase_id: uuid.UUID
user_id: uuid.UUID
class FetchViewsOutput(pydantic.BaseModel):
"""Результат получения просмотров."""
purchase_id: uuid.UUID
views_count: int | None # None если не удалось получить
views_availability: domain.PostViewsAvailability
fetched_at: datetime.datetime
error_message: str | None = None # Сообщение об ошибке если не удалось
class UpdateViewsManuallyInput(pydantic.BaseModel):
"""Ручное обновление просмотров."""
purchase_id: uuid.UUID
user_id: uuid.UUID
views_count: int

28
src/usecase/__init__.py
View File

@@ -22,6 +22,7 @@ from .purchase.delete_purchase import delete_purchase
from .purchase.get_purchase import get_purchase
from .purchase.get_purchases import get_purchases
from .purchase.update_purchase import update_purchase
from .subscription.handle_subscription import handle_subscription
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
@@ -29,6 +30,9 @@ from .target_channel.get_user_target_chans import get_user_target_chans
from .target_channel.update_target_chan_permissions import update_target_chan_permissions
from .telegram_login import telegram_login
from .validate_login_token import validate_login_token
from .views.fetch_views import fetch_views
from .views.get_views_history import get_views_history
from .views.update_views_manually import update_views_manually
class Database(typing.Protocol):
@@ -110,6 +114,24 @@ class Database(typing.Protocol):
async def check_creative_in_use(self, creative_id: UUID) -> bool: ...
async def get_purchase_by_invite_link(self, invite_link: str) -> domain.Purchase | None: ...
async def create_subscription(self, subscription: domain.Subscription) -> domain.Subscription: ...
async def get_subscription_by_user_and_purchase(
self, user_telegram_id: int, purchase_id: UUID
) -> domain.Subscription | None: ...
async def create_views_snapshot(self, snapshot: domain.ViewsSnapshot) -> domain.ViewsSnapshot: ...
async def get_views_history(
self,
purchase_id: UUID,
*,
from_date: datetime.datetime | None = None,
to_date: datetime.datetime | None = None,
) -> list[domain.ViewsSnapshot]: ...
class TelegramWriter(typing.Protocol):
async def send_message(self, text: str, chat_id: int) -> None: ...
@@ -120,6 +142,8 @@ class TelegramWriter(typing.Protocol):
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str: ...
async def get_post_views(self, post_url: str) -> int | None: ... # None если не удалось получить
class JWTEncoder(typing.Protocol):
def encode_access_token(self, user_id: UUID, telegram_id: int, username: str | None = None) -> str: ...
@@ -155,3 +179,7 @@ class Usecase:
update_purchase = update_purchase
archive_purchase = archive_purchase
delete_purchase = delete_purchase
handle_subscription = handle_subscription
fetch_views = fetch_views
get_views_history = get_views_history
update_views_manually = update_views_manually

20
src/usecase/creative/archive_creative.py
View File

@@ -20,13 +20,13 @@ async def archive_creative(self: 'Usecase', creative_id: uuid.UUID, user_id: uui
creative.status = domain.CreativeStatus.ARCHIVED
updated = await self.database.update_creative(creative)
return dto.CreativeOutput(
id=updated.id,
name=updated.name,
text=updated.text,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
created_at=updated.created_at,
status=updated.status,
purchases_count=updated.purchases_count,
)
return dto.CreativeOutput(
id=updated.id,
name=updated.name,
text=updated.text,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
created_at=updated.created_at,
status=updated.status,
purchases_count=updated.purchases_count,
)

21
src/usecase/creative/create_creative.py
View File

@@ -22,19 +22,20 @@ async def create_creative(self: 'Usecase', input: dto.CreateCreativeInput, user_
creative = domain.Creative(
name=input.name,
text=input.text,
status=domain.CreativeStatus.ACTIVE,
target_channel_id=target_channel.id,
user_id=user_id,
)
created = await self.database.create_creative(creative)
return dto.CreativeOutput(
id=created.id,
name=created.name,
text=created.text,
target_channel_id=created.target_channel_id,
target_channel_title=created.target_channel.title,
created_at=created.created_at,
status=created.status,
purchases_count=created.purchases_count,
)
return dto.CreativeOutput(
id=created.id,
name=created.name,
text=created.text,
target_channel_id=created.target_channel_id,
target_channel_title=created.target_channel.title,
created_at=created.created_at,
status=created.status,
purchases_count=created.purchases_count,
)

20
src/usecase/creative/get_creative.py
View File

@@ -16,13 +16,13 @@ async def get_creative(self: 'Usecase', input: dto.GetCreativeInput) -> dto.Crea
log.warning('User %s attempted to access unavailable creative %s', input.user_id, input.creative_id)
raise domain.CreativeNotFound(input.creative_id)
return dto.CreativeOutput(
id=creative.id,
name=creative.name,
text=creative.text,
target_channel_id=creative.target_channel_id,
target_channel_title=creative.target_channel.title,
created_at=creative.created_at,
status=creative.status,
purchases_count=creative.purchases_count,
)
return dto.CreativeOutput(
id=creative.id,
name=creative.name,
text=creative.text,
target_channel_id=creative.target_channel_id,
target_channel_title=creative.target_channel.title,
created_at=creative.created_at,
status=creative.status,
purchases_count=creative.purchases_count,
)

20
src/usecase/creative/update_creative.py
View File

@@ -26,13 +26,13 @@ async def update_creative(
updated = await self.database.update_creative(creative)
return dto.CreativeOutput(
id=updated.id,
name=updated.name,
text=updated.text,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
created_at=updated.created_at,
status=updated.status,
purchases_count=updated.purchases_count,
)
return dto.CreativeOutput(
id=updated.id,
name=updated.name,
text=updated.text,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
created_at=updated.created_at,
status=updated.status,
purchases_count=updated.purchases_count,
)

40
src/usecase/purchase/archive_purchase.py
View File

@@ -20,22 +20,24 @@ async def archive_purchase(self: 'Usecase', purchase_id: uuid.UUID, user_id: uui
purchase.status = domain.PurchaseStatus.ARCHIVED
updated = await self.database.update_purchase(purchase)
return dto.PurchaseOutput(
id=updated.id,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
external_channel_id=updated.external_channel_id,
external_channel_title=updated.external_channel.title,
creative_id=updated.creative_id,
creative_name=updated.creative.name,
placement_date=updated.placement_date,
cost=updated.cost,
comment=updated.comment,
ad_post_url=updated.ad_post_url,
invite_link_type=updated.invite_link_type,
invite_link=updated.invite_link,
status=updated.status,
subscriptions_count=updated.subscriptions_count,
views_count=updated.views_count,
created_at=updated.created_at,
)
return dto.PurchaseOutput(
id=updated.id,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
external_channel_id=updated.external_channel_id,
external_channel_title=updated.external_channel.title,
creative_id=updated.creative_id,
creative_name=updated.creative.name,
placement_date=updated.placement_date,
cost=updated.cost,
comment=updated.comment,
ad_post_url=updated.ad_post_url,
invite_link_type=updated.invite_link_type,
invite_link=updated.invite_link,
status=updated.status,
subscriptions_count=updated.subscriptions_count,
views_count=updated.views_count,
views_availability=updated.views_availability,
last_views_fetch_at=updated.last_views_fetch_at,
created_at=updated.created_at,
)

41
src/usecase/purchase/create_purchase.py
View File

@@ -62,22 +62,25 @@ async def create_purchase(self: 'Usecase', input: dto.CreatePurchaseInput, user_
creative.purchases_count += 1
await self.database.update_creative(creative)
return dto.PurchaseOutput(
id=created.id,
target_channel_id=created.target_channel_id,
target_channel_title=created.target_channel.title,
external_channel_id=created.external_channel_id,
external_channel_title=created.external_channel.title,
creative_id=created.creative_id,
creative_name=created.creative.name,
placement_date=created.placement_date,
cost=created.cost,
comment=created.comment,
ad_post_url=created.ad_post_url,
invite_link_type=created.invite_link_type,
invite_link=created.invite_link,
status=created.status,
subscriptions_count=created.subscriptions_count,
views_count=created.views_count,
created_at=created.created_at,
)
# Формируем ответ внутри транзакции, пока объекты в сессии
return dto.PurchaseOutput(
id=created.id,
target_channel_id=created.target_channel_id,
target_channel_title=target_channel.title,
external_channel_id=created.external_channel_id,
external_channel_title=external_channel.title,
creative_id=created.creative_id,
creative_name=creative.name,
placement_date=created.placement_date,
cost=created.cost,
comment=created.comment,
ad_post_url=created.ad_post_url,
invite_link_type=created.invite_link_type,
invite_link=created.invite_link,
status=created.status,
subscriptions_count=created.subscriptions_count,
views_count=created.views_count,
views_availability=created.views_availability,
last_views_fetch_at=created.last_views_fetch_at,
created_at=created.created_at,
)

40
src/usecase/purchase/get_purchase.py
View File

@@ -16,22 +16,24 @@ async def get_purchase(self: 'Usecase', input: dto.GetPurchaseInput) -> dto.Purc
log.warning('Purchase %s not found for user %s', input.purchase_id, input.user_id)
raise domain.PurchaseNotFound(input.purchase_id)
return dto.PurchaseOutput(
id=purchase.id,
target_channel_id=purchase.target_channel_id,
target_channel_title=purchase.target_channel.title,
external_channel_id=purchase.external_channel_id,
external_channel_title=purchase.external_channel.title,
creative_id=purchase.creative_id,
creative_name=purchase.creative.name,
placement_date=purchase.placement_date,
cost=purchase.cost,
comment=purchase.comment,
ad_post_url=purchase.ad_post_url,
invite_link_type=purchase.invite_link_type,
invite_link=purchase.invite_link,
status=purchase.status,
subscriptions_count=purchase.subscriptions_count,
views_count=purchase.views_count,
created_at=purchase.created_at,
)
return dto.PurchaseOutput(
id=purchase.id,
target_channel_id=purchase.target_channel_id,
target_channel_title=purchase.target_channel.title,
external_channel_id=purchase.external_channel_id,
external_channel_title=purchase.external_channel.title,
creative_id=purchase.creative_id,
creative_name=purchase.creative.name,
placement_date=purchase.placement_date,
cost=purchase.cost,
comment=purchase.comment,
ad_post_url=purchase.ad_post_url,
invite_link_type=purchase.invite_link_type,
invite_link=purchase.invite_link,
status=purchase.status,
subscriptions_count=purchase.subscriptions_count,
views_count=purchase.views_count,
views_availability=purchase.views_availability,
last_views_fetch_at=purchase.last_views_fetch_at,
created_at=purchase.created_at,
)

2
src/usecase/purchase/get_purchases.py
View File

@@ -35,6 +35,8 @@ async def get_purchases(self: 'Usecase', input: dto.GetPurchasesInput) -> dto.Ge
status=purchase.status,
subscriptions_count=purchase.subscriptions_count,
views_count=purchase.views_count,
views_availability=purchase.views_availability,
last_views_fetch_at=purchase.last_views_fetch_at,
created_at=purchase.created_at,
)
for purchase in purchases

40
src/usecase/purchase/update_purchase.py
View File

@@ -31,22 +31,24 @@ async def update_purchase(
updated = await self.database.update_purchase(purchase)
return dto.PurchaseOutput(
id=updated.id,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
external_channel_id=updated.external_channel_id,
external_channel_title=updated.external_channel.title,
creative_id=updated.creative_id,
creative_name=updated.creative.name,
placement_date=updated.placement_date,
cost=updated.cost,
comment=updated.comment,
ad_post_url=updated.ad_post_url,
invite_link_type=updated.invite_link_type,
invite_link=updated.invite_link,
status=updated.status,
subscriptions_count=updated.subscriptions_count,
views_count=updated.views_count,
created_at=updated.created_at,
)
return dto.PurchaseOutput(
id=updated.id,
target_channel_id=updated.target_channel_id,
target_channel_title=updated.target_channel.title,
external_channel_id=updated.external_channel_id,
external_channel_title=updated.external_channel.title,
creative_id=updated.creative_id,
creative_name=updated.creative.name,
placement_date=updated.placement_date,
cost=updated.cost,
comment=updated.comment,
ad_post_url=updated.ad_post_url,
invite_link_type=updated.invite_link_type,
invite_link=updated.invite_link,
status=updated.status,
subscriptions_count=updated.subscriptions_count,
views_count=updated.views_count,
views_availability=updated.views_availability,
last_views_fetch_at=updated.last_views_fetch_at,
created_at=updated.created_at,
)

59
src/usecase/subscription/handle_subscription.py Normal file
View File

@@ -0,0 +1,59 @@
import logging
from typing import TYPE_CHECKING
from src import domain
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def handle_subscription(
self: 'Usecase',
user_telegram_id: int,
username: str | None,
invite_link: str,
) -> None:
"""
Обрабатывает подписку пользователя в целевой канал.
Находит закуп по invite_link и создаёт запись о подписке,
инкрементируя счётчик подписок у закупа.
"""
async with self.database.transaction():
# Находим закуп по пригласительной ссылке
purchase = await self.database.get_purchase_by_invite_link(invite_link)
if not purchase:
log.warning('Purchase not found for invite_link: %s', invite_link)
return
# Проверяем, не создана ли уже подписка для этого пользователя
existing_subscription = await self.database.get_subscription_by_user_and_purchase(user_telegram_id, purchase.id)
if existing_subscription:
log.info(
'Subscription already exists for user %s and purchase %s',
user_telegram_id,
purchase.id,
)
return
# Создаём запись о подписке
subscription = domain.Subscription(
purchase_id=purchase.id,
user_telegram_id=user_telegram_id,
username=username,
invite_link=invite_link,
)
await self.database.create_subscription(subscription)
# Увеличиваем счётчик подписок у закупа
purchase.subscriptions_count += 1
await self.database.update_purchase(purchase)
log.info(
'Subscription created: user %s subscribed via purchase %s (invite_link: %s)',
user_telegram_id,
purchase.id,
invite_link,
)

77
src/usecase/views/fetch_views.py Normal file
View File

@@ -0,0 +1,77 @@
import datetime
import logging
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def fetch_views(self: 'Usecase', input: dto.FetchViewsInput) -> dto.FetchViewsOutput:
"""
Получить просмотры для закупа через Telegram API.
Создаёт снимок просмотров и обновляет метрики закупа.
"""
async with self.database.transaction():
purchase = await self.database.get_purchase(input.user_id, input.purchase_id)
if not purchase:
log.warning('Purchase %s not found for user %s', input.purchase_id, input.user_id)
raise domain.PurchaseNotFound(input.purchase_id)
if not purchase.ad_post_url:
log.warning('Purchase %s has no ad_post_url', input.purchase_id)
return dto.FetchViewsOutput(
purchase_id=purchase.id,
views_count=None,
views_availability=domain.PostViewsAvailability.UNAVAILABLE,
fetched_at=datetime.datetime.now(datetime.UTC),
error_message='No ad_post_url specified',
)
# Попытка получить просмотры через Telegram API
views_count = await self.telegram_writer.get_post_views(purchase.ad_post_url)
fetched_at = datetime.datetime.now(datetime.UTC)
if views_count is not None:
# Успешно получены просмотры
purchase.views_count = views_count
purchase.views_availability = domain.PostViewsAvailability.AVAILABLE
purchase.last_views_fetch_at = fetched_at
await self.database.update_purchase(purchase)
# Создаём снимок
snapshot = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=views_count,
fetched_at=fetched_at,
)
await self.database.create_views_snapshot(snapshot)
log.info('Views fetched for purchase %s: %d', purchase.id, views_count)
return dto.FetchViewsOutput(
purchase_id=purchase.id,
views_count=views_count,
views_availability=domain.PostViewsAvailability.AVAILABLE,
fetched_at=fetched_at,
)
else:
# Не удалось получить просмотры
purchase.views_availability = domain.PostViewsAvailability.UNAVAILABLE
purchase.last_views_fetch_at = fetched_at
await self.database.update_purchase(purchase)
log.warning('Failed to fetch views for purchase %s', purchase.id)
return dto.FetchViewsOutput(
purchase_id=purchase.id,
views_count=None,
views_availability=domain.PostViewsAvailability.UNAVAILABLE,
fetched_at=fetched_at,
error_message='Failed to fetch views from Telegram API',
)

37
src/usecase/views/get_views_history.py Normal file
View File

@@ -0,0 +1,37 @@
import logging
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def get_views_history(self: 'Usecase', input: dto.GetViewsHistoryInput) -> dto.GetViewsHistoryOutput:
"""Получить историю просмотров для закупа."""
async with self.database.transaction():
purchase = await self.database.get_purchase(input.user_id, input.purchase_id)
if not purchase:
log.warning('Purchase %s not found for user %s', input.purchase_id, input.user_id)
raise domain.PurchaseNotFound(input.purchase_id)
snapshots = await self.database.get_views_history(
input.purchase_id,
from_date=input.from_date,
to_date=input.to_date,
)
return dto.GetViewsHistoryOutput(
snapshots=[
dto.ViewsSnapshotOutput(
id=snapshot.id,
purchase_id=snapshot.purchase_id,
views_count=snapshot.views_count,
fetched_at=snapshot.fetched_at,
created_at=snapshot.created_at,
)
for snapshot in snapshots
]
)

60
src/usecase/views/update_views_manually.py Normal file
View File

@@ -0,0 +1,60 @@
import datetime
import logging
from typing import TYPE_CHECKING
from src import domain, dto
if TYPE_CHECKING:
from .. import Usecase
log = logging.getLogger(__name__)
async def update_views_manually(self: 'Usecase', input: dto.UpdateViewsManuallyInput) -> dto.PurchaseOutput:
"""Ручное обновление просмотров для закупа."""
async with self.database.transaction():
purchase = await self.database.get_purchase(input.user_id, input.purchase_id)
if not purchase:
log.warning('Purchase %s not found for user %s', input.purchase_id, input.user_id)
raise domain.PurchaseNotFound(input.purchase_id)
fetched_at = datetime.datetime.now(datetime.UTC)
# Обновляем метрики закупа
purchase.views_count = input.views_count
purchase.views_availability = domain.PostViewsAvailability.MANUAL
purchase.last_views_fetch_at = fetched_at
await self.database.update_purchase(purchase)
# Создаём снимок
snapshot = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=input.views_count,
fetched_at=fetched_at,
)
await self.database.create_views_snapshot(snapshot)
log.info('Views manually updated for purchase %s: %d', purchase.id, input.views_count)
# Возвращаем обновлённый закуп
return dto.PurchaseOutput(
id=purchase.id,
target_channel_id=purchase.target_channel_id,
target_channel_title=purchase.target_channel.title,
external_channel_id=purchase.external_channel_id,
external_channel_title=purchase.external_channel.title,
creative_id=purchase.creative_id,
creative_name=purchase.creative.name,
placement_date=purchase.placement_date,
cost=purchase.cost,
comment=purchase.comment,
ad_post_url=purchase.ad_post_url,
invite_link_type=purchase.invite_link_type,
invite_link=purchase.invite_link,
status=purchase.status,
subscriptions_count=purchase.subscriptions_count,
views_count=purchase.views_count,
views_availability=purchase.views_availability,
last_views_fetch_at=purchase.last_views_fetch_at,
created_at=purchase.created_at,
)

63
tests/conftest.py
View File

@@ -19,6 +19,8 @@ class MockDatabase:
self.external_channels: dict[uuid.UUID, domain.ExternalChannel] = {}
self.creatives: dict[uuid.UUID, domain.Creative] = {}
self.purchases: dict[uuid.UUID, domain.Purchase] = {}
self.subscriptions: dict[uuid.UUID, domain.Subscription] = {}
self.views_snapshots: dict[uuid.UUID, domain.ViewsSnapshot] = {}
self.m2m_target_external: dict[uuid.UUID, list[uuid.UUID]] = {}
def transaction(self) -> AbstractAsyncContextManager[None]:
@@ -205,6 +207,8 @@ class MockDatabase:
purchase.subscriptions_count = 0
if not hasattr(purchase, 'status') or purchase.status is None:
purchase.status = domain.PurchaseStatus.ACTIVE
if not hasattr(purchase, 'views_availability') or purchase.views_availability is None:
purchase.views_availability = domain.PostViewsAvailability.UNKNOWN
# Загружаем связанные объекты
target_ch = self.target_channels.get(purchase.target_channel_id)
if target_ch:
@@ -257,6 +261,61 @@ class MockDatabase:
return True
return False
async def get_purchase_by_invite_link(self, invite_link: str) -> domain.Purchase | None:
for purchase in self.purchases.values():
if purchase.invite_link == invite_link:
return purchase
return None
async def create_subscription(self, subscription: domain.Subscription) -> domain.Subscription:
if not subscription.id:
subscription.id = uuid.uuid4()
if not hasattr(subscription, 'created_at') or subscription.created_at is None:
subscription.created_at = datetime.datetime.now(datetime.UTC)
if not hasattr(subscription, 'updated_at') or subscription.updated_at is None:
subscription.updated_at = datetime.datetime.now(datetime.UTC)
# Загружаем связанный объект
purchase = self.purchases.get(subscription.purchase_id)
if purchase:
subscription.purchase = purchase
self.subscriptions[subscription.id] = subscription
return subscription
async def get_subscription_by_user_and_purchase(
self, user_telegram_id: int, purchase_id: uuid.UUID
) -> domain.Subscription | None:
for subscription in self.subscriptions.values():
if subscription.user_telegram_id == user_telegram_id and subscription.purchase_id == purchase_id:
return subscription
return None
async def create_views_snapshot(self, snapshot: domain.ViewsSnapshot) -> domain.ViewsSnapshot:
if not snapshot.id:
snapshot.id = uuid.uuid4()
if not hasattr(snapshot, 'created_at') or snapshot.created_at is None:
snapshot.created_at = datetime.datetime.now(datetime.UTC)
if not hasattr(snapshot, 'updated_at') or snapshot.updated_at is None:
snapshot.updated_at = datetime.datetime.now(datetime.UTC)
purchase = self.purchases.get(snapshot.purchase_id)
if purchase:
snapshot.purchase = purchase
self.views_snapshots[snapshot.id] = snapshot
return snapshot
async def get_views_history(
self,
purchase_id: uuid.UUID,
*,
from_date: datetime.datetime | None = None,
to_date: datetime.datetime | None = None,
) -> list[domain.ViewsSnapshot]:
result = [s for s in self.views_snapshots.values() if s.purchase_id == purchase_id]
if from_date:
result = [s for s in result if s.fetched_at >= from_date]
if to_date:
result = [s for s in result if s.fetched_at <= to_date]
return sorted(result, key=lambda s: s.fetched_at)
class MockTelegramWriter:
def __init__(self) -> None:
@@ -279,6 +338,10 @@ class MockTelegramWriter:
link_type = 'approval' if requires_approval else 'public'
return f'https://t.me/+{link_type}_link_{self.invite_link_counter}_{chat_id}'
async def get_post_views(self, post_url: str) -> int | None:
"""Mock implementation returns None (TDLib not integrated)."""
return None
class MockJWTEncoder:
def encode_access_token(self, user_id: uuid.UUID, telegram_id: int, username: str | None = None) -> str:

342
tests/test_subscriptions.py Normal file
View File

@@ -0,0 +1,342 @@
"""Тесты обработки подписок пользователей через invite_link."""
import datetime
import pytest
from src import domain, usecase
from tests.conftest import MockDatabase
@pytest.mark.asyncio
async def test_handle_subscription_success(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Первая подписка пользователя по invite_link успешно создаётся."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=50000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
invite_link = 'https://t.me/+unique_link_123'
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link=invite_link,
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
subscriptions_count=0,
)
await mock_database.create_purchase(purchase)
# Симулируем подписку пользователя
subscriber_id = 999888777
subscriber_username = 'new_subscriber'
await usecases.handle_subscription(
user_telegram_id=subscriber_id,
username=subscriber_username,
invite_link=invite_link,
)
# Проверяем, что подписка создана
subscription = await mock_database.get_subscription_by_user_and_purchase(subscriber_id, purchase.id)
assert subscription is not None
assert subscription.user_telegram_id == subscriber_id
assert subscription.username == subscriber_username
assert subscription.invite_link == invite_link
# Проверяем, что счётчик подписок у закупа увеличился
updated_purchase = await mock_database.get_purchase(user.id, purchase.id)
assert updated_purchase is not None
assert updated_purchase.subscriptions_count == 1
@pytest.mark.asyncio
async def test_handle_subscription_duplicate_ignored(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Повторная подписка того же пользователя игнорируется - не создаёт дубликат."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=30000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
invite_link = 'https://t.me/+unique_link_456'
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link=invite_link,
invite_link_type=domain.InviteLinkType.APPROVAL,
status=domain.PurchaseStatus.ACTIVE,
subscriptions_count=0,
)
await mock_database.create_purchase(purchase)
subscriber_id = 888777666
subscriber_username = 'duplicate_user'
# Первая подписка
await usecases.handle_subscription(
user_telegram_id=subscriber_id,
username=subscriber_username,
invite_link=invite_link,
)
# Проверяем счётчик
updated_purchase = await mock_database.get_purchase(user.id, purchase.id)
assert updated_purchase is not None
assert updated_purchase.subscriptions_count == 1
# Повторная попытка подписки того же пользователя
await usecases.handle_subscription(
user_telegram_id=subscriber_id,
username=subscriber_username,
invite_link=invite_link,
)
# Проверяем, что счётчик не увеличился
updated_purchase_again = await mock_database.get_purchase(user.id, purchase.id)
assert updated_purchase_again is not None
assert updated_purchase_again.subscriptions_count == 1 # Остался 1, не увеличился
# Проверяем, что в БД только одна подписка этого пользователя
all_subscriptions = [s for s in mock_database.subscriptions.values() if s.user_telegram_id == subscriber_id]
assert len(all_subscriptions) == 1
@pytest.mark.asyncio
async def test_handle_subscription_invalid_invite_link(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Подписка по несуществующему invite_link не создаётся."""
fake_invite_link = 'https://t.me/+fake_link_999'
# Пытаемся обработать подписку по несуществующей ссылке
await usecases.handle_subscription(
user_telegram_id=777666555,
username='unknown_user',
invite_link=fake_invite_link,
)
# Проверяем, что подписки не создано
assert len(mock_database.subscriptions) == 0
@pytest.mark.asyncio
async def test_multiple_users_subscription_same_purchase(
usecases: usecase.Usecase, mock_database: MockDatabase
) -> None:
"""Несколько пользователей подписываются через один закуп - каждый учитывается."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=40000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Viral Creative',
text='Viral content',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
invite_link = 'https://t.me/+popular_link_789'
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link=invite_link,
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
subscriptions_count=0,
)
await mock_database.create_purchase(purchase)
# Подписываются три разных пользователя
users_data = [
(111222333, 'user_one'),
(444555666, 'user_two'),
(777888999, 'user_three'),
]
for telegram_id, username in users_data:
await usecases.handle_subscription(
user_telegram_id=telegram_id,
username=username,
invite_link=invite_link,
)
# Проверяем, что создано 3 подписки
purchase_subscriptions = [s for s in mock_database.subscriptions.values() if s.purchase_id == purchase.id]
assert len(purchase_subscriptions) == 3
# Проверяем, что счётчик правильный
updated_purchase = await mock_database.get_purchase(user.id, purchase.id)
assert updated_purchase is not None
assert updated_purchase.subscriptions_count == 3
@pytest.mark.asyncio
async def test_subscription_tracking_across_different_purchases(
usecases: usecase.Usecase, mock_database: MockDatabase
) -> None:
"""Один пользователь может подписаться через разные закупы - каждая подписка учитывается отдельно."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=25000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
# Создаём два разных закупа с разными invite_link
purchase1 = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link_a',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
subscriptions_count=0,
)
purchase2 = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link_b',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
subscriptions_count=0,
)
await mock_database.create_purchase(purchase1)
await mock_database.create_purchase(purchase2)
subscriber_id = 555444333
# Пользователь подписывается через первый закуп
await usecases.handle_subscription(
user_telegram_id=subscriber_id,
username='multi_subscriber',
invite_link=purchase1.invite_link,
)
# Тот же пользователь "подписывается" через второй закуп
# (технически невозможно, но тестируем логику)
await usecases.handle_subscription(
user_telegram_id=subscriber_id,
username='multi_subscriber',
invite_link=purchase2.invite_link,
)
# Проверяем, что у каждого закупа свой счётчик
updated_purchase1 = await mock_database.get_purchase(user.id, purchase1.id)
updated_purchase2 = await mock_database.get_purchase(user.id, purchase2.id)
assert updated_purchase1 is not None
assert updated_purchase1.subscriptions_count == 1
assert updated_purchase2 is not None
assert updated_purchase2.subscriptions_count == 1
# Проверяем, что создано 2 подписки
user_subscriptions = [s for s in mock_database.subscriptions.values() if s.user_telegram_id == subscriber_id]
assert len(user_subscriptions) == 2

424
tests/test_views.py Normal file
View File

@@ -0,0 +1,424 @@
"""Тесты работы с просмотрами рекламных постов."""
import datetime
import pytest
from src import domain, dto, usecase
from tests.conftest import MockDatabase
@pytest.mark.asyncio
async def test_fetch_views_no_ad_post_url(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Попытка получить просмотры для закупа без ad_post_url возвращает unavailable."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=50000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link1',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
ad_post_url=None, # Нет ссылки на пост
)
await mock_database.create_purchase(purchase)
input_data = dto.FetchViewsInput(
purchase_id=purchase.id,
user_id=user.id,
)
result = await usecases.fetch_views(input=input_data)
assert result.views_count is None
assert result.views_availability == domain.PostViewsAvailability.UNAVAILABLE
assert result.error_message == 'No ad_post_url specified'
@pytest.mark.asyncio
async def test_fetch_views_api_unavailable(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Если Telegram API недоступен (возвращает None), статус = unavailable."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=30000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link2',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
ad_post_url='https://t.me/externalchannel/123',
)
await mock_database.create_purchase(purchase)
input_data = dto.FetchViewsInput(
purchase_id=purchase.id,
user_id=user.id,
)
# Mock telegram writer returns None (TDLib not integrated)
result = await usecases.fetch_views(input=input_data)
assert result.views_count is None
assert result.views_availability == domain.PostViewsAvailability.UNAVAILABLE
assert 'Failed to fetch views' in result.error_message
# Проверяем, что purchase обновлён
updated_purchase = await mock_database.get_purchase(user.id, purchase.id)
assert updated_purchase is not None
assert updated_purchase.views_availability == domain.PostViewsAvailability.UNAVAILABLE
assert updated_purchase.last_views_fetch_at is not None
@pytest.mark.asyncio
async def test_update_views_manually(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Ручное обновление просмотров создаёт снимок и обновляет закуп."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=40000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link3',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
ad_post_url='https://t.me/externalchannel/456',
)
await mock_database.create_purchase(purchase)
input_data = dto.UpdateViewsManuallyInput(
purchase_id=purchase.id,
user_id=user.id,
views_count=15000,
)
result = await usecases.update_views_manually(input=input_data)
# Проверяем возвращённый результат
assert result.views_count == 15000
assert result.views_availability == domain.PostViewsAvailability.MANUAL
assert result.last_views_fetch_at is not None
# Проверяем, что создан снимок
snapshots = [s for s in mock_database.views_snapshots.values() if s.purchase_id == purchase.id]
assert len(snapshots) == 1
assert snapshots[0].views_count == 15000
# Проверяем, что закуп обновлён
updated_purchase = await mock_database.get_purchase(user.id, purchase.id)
assert updated_purchase is not None
assert updated_purchase.views_count == 15000
assert updated_purchase.views_availability == domain.PostViewsAvailability.MANUAL
@pytest.mark.asyncio
async def test_get_views_history_empty(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Получение истории просмотров для закупа без снимков возвращает пустой список."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=25000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link4',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
)
await mock_database.create_purchase(purchase)
input_data = dto.GetViewsHistoryInput(
purchase_id=purchase.id,
user_id=user.id,
)
result = await usecases.get_views_history(input=input_data)
assert len(result.snapshots) == 0
@pytest.mark.asyncio
async def test_get_views_history_with_snapshots(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Получение истории просмотров возвращает снимки в хронологическом порядке."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=35000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link5',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
)
await mock_database.create_purchase(purchase)
# Создаём несколько снимков
now = datetime.datetime.now(datetime.UTC)
snapshot1 = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=1000,
fetched_at=now - datetime.timedelta(hours=2),
)
snapshot2 = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=1500,
fetched_at=now - datetime.timedelta(hours=1),
)
snapshot3 = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=2000,
fetched_at=now,
)
await mock_database.create_views_snapshot(snapshot1)
await mock_database.create_views_snapshot(snapshot2)
await mock_database.create_views_snapshot(snapshot3)
input_data = dto.GetViewsHistoryInput(
purchase_id=purchase.id,
user_id=user.id,
)
result = await usecases.get_views_history(input=input_data)
assert len(result.snapshots) == 3
# Проверяем сортировку по времени (старые первые)
assert result.snapshots[0].views_count == 1000
assert result.snapshots[1].views_count == 1500
assert result.snapshots[2].views_count == 2000
@pytest.mark.asyncio
async def test_get_views_history_with_date_filter(usecases: usecase.Usecase, mock_database: MockDatabase) -> None:
"""Фильтрация истории просмотров по датам работает корректно."""
user = domain.User(telegram_id=123456, username='testuser')
await mock_database.create_user(user)
target_channel = domain.TargetChannel(
telegram_id=-1001234567890,
title='Target Channel',
username='targetchannel',
user_id=user.id,
is_active=True,
)
await mock_database.upsert_target_channel(target_channel)
external_channel = domain.ExternalChannel(
telegram_id=-1009876543210,
title='External Channel',
username='externalchannel',
description=None,
subscribers_count=28000,
user_id=user.id,
)
await mock_database.create_external_channel(external_channel)
creative = domain.Creative(
name='Test Creative',
text='Test text',
target_channel_id=target_channel.id,
user_id=user.id,
status=domain.CreativeStatus.ACTIVE,
)
await mock_database.create_creative(creative)
purchase = domain.Purchase(
target_channel_id=target_channel.id,
external_channel_id=external_channel.id,
creative_id=creative.id,
user_id=user.id,
placement_date=datetime.datetime.now(datetime.UTC),
invite_link='https://t.me/+link6',
invite_link_type=domain.InviteLinkType.PUBLIC,
status=domain.PurchaseStatus.ACTIVE,
)
await mock_database.create_purchase(purchase)
# Создаём снимки с разными датами
now = datetime.datetime.now(datetime.UTC)
snapshot1 = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=1000,
fetched_at=now - datetime.timedelta(days=3),
)
snapshot2 = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=1500,
fetched_at=now - datetime.timedelta(days=1),
)
snapshot3 = domain.ViewsSnapshot(
purchase_id=purchase.id,
views_count=2000,
fetched_at=now,
)
await mock_database.create_views_snapshot(snapshot1)
await mock_database.create_views_snapshot(snapshot2)
await mock_database.create_views_snapshot(snapshot3)
# Запрашиваем историю за последние 2 дня
input_data = dto.GetViewsHistoryInput(
purchase_id=purchase.id,
user_id=user.id,
from_date=now - datetime.timedelta(days=2),
)
result = await usecases.get_views_history(input=input_data)
# Должны вернуться только snapshot2 и snapshot3
assert len(result.snapshots) == 2
assert result.snapshots[0].views_count == 1500
assert result.snapshots[1].views_count == 2000