diff --git a/src/adapter/parser.py b/src/adapter/parser.py index 2ab8199..ed84005 100644 --- a/src/adapter/parser.py +++ b/src/adapter/parser.py @@ -10,12 +10,11 @@ log = logging.getLogger(__name__) class FetchChannelResponse(BaseModel): - id: str telegram_id: int username: str | None title: str | None - access_hash: int - pts: int + access_hash: int | None + pts: int | None class ParserClient(Parser): diff --git a/src/adapter/postgres.py b/src/adapter/postgres.py index 005405e..a3d6f83 100644 --- a/src/adapter/postgres.py +++ b/src/adapter/postgres.py @@ -549,7 +549,15 @@ class Postgres(DatabaseBase): telegram_user_id: uuid.UUID, placement_post_id: uuid.UUID ) -> domain.Subscription | None: return await domain.Subscription.get_or_none( - telegram_user_id=telegram_user_id, placement_post_id=placement_post_id + telegram_user_id=telegram_user_id, placement_id=placement_post_id + ) + + @staticmethod + async def get_subscription_by_subscriber_and_placement( + telegram_user_id: uuid.UUID, placement_id: uuid.UUID + ) -> domain.Subscription | None: + return await domain.Subscription.get_or_none( + telegram_user_id=telegram_user_id, placement_id=placement_id ) @staticmethod @@ -563,10 +571,18 @@ class Postgres(DatabaseBase): date_from: datetime.datetime | None = None, date_to: datetime.datetime | None = None, ) -> list[domain.Subscription]: + """Get subscriptions for placement_posts by finding their parent placements.""" if not placement_post_ids: return [] - query = domain.Subscription.filter(placement_post_id__in=placement_post_ids) + # Get placement_ids from placement_posts + placement_posts = await domain.PlacementPost.filter(id__in=placement_post_ids).all() + placement_ids = [pp.placement_id for pp in placement_posts] + + if not placement_ids: + return [] + + query = domain.Subscription.filter(placement_id__in=placement_ids) if date_from: query = query.filter(created_at__gte=date_from) @@ -636,7 +652,15 @@ class Postgres(DatabaseBase): @staticmethod async def count_subscriptions_by_placement_post(placement_post_id: uuid.UUID) -> int: - return await domain.Subscription.filter(placement_post_id=placement_post_id).count() + """Count subscriptions for a placement_post by finding its parent placement.""" + placement_post = await domain.PlacementPost.get_or_none(id=placement_post_id) + if not placement_post: + return 0 + return await domain.Subscription.filter(placement_id=placement_post.placement_id).count() + + @staticmethod + async def count_subscriptions_by_placement(placement_id: uuid.UUID) -> int: + return await domain.Subscription.filter(placement_id=placement_id).count() @staticmethod async def count_placement_posts_by_creative_batch(creative_ids: list[uuid.UUID]) -> dict[uuid.UUID, int]: @@ -657,20 +681,57 @@ class Postgres(DatabaseBase): @staticmethod async def count_subscriptions_by_placement_post_batch(placement_post_ids: list[uuid.UUID]) -> dict[uuid.UUID, int]: + """Count subscriptions for placement_posts by finding their parent placements.""" if not placement_post_ids: return {} from tortoise.functions import Count + # Get placement_ids from placement_posts + placement_posts = await domain.PlacementPost.filter(id__in=placement_post_ids).all() + placement_id_to_post_ids: dict[uuid.UUID, list[uuid.UUID]] = {} + for pp in placement_posts: + placement_id_to_post_ids.setdefault(pp.placement_id, []).append(pp.id) + + placement_ids = list(placement_id_to_post_ids.keys()) + if not placement_ids: + return dict.fromkeys(placement_post_ids, 0) + + # Count subscriptions by placement results = ( - await domain.Subscription.filter(placement_post_id__in=placement_post_ids) - .group_by('placement_post_id') + await domain.Subscription.filter(placement_id__in=placement_ids) + .group_by('placement_id') .annotate(count=Count('id')) - .values('placement_post_id', 'count') + .values('placement_id', 'count') ) - counts = {row['placement_post_id']: row['count'] for row in results} - return {pid: counts.get(pid, 0) for pid in placement_post_ids} + placement_counts = {row['placement_id']: row['count'] for row in results} + + # Map back to placement_post_ids + post_counts: dict[uuid.UUID, int] = {} + for placement_id, post_ids in placement_id_to_post_ids.items(): + count = placement_counts.get(placement_id, 0) + for post_id in post_ids: + post_counts[post_id] = count + + return {pid: post_counts.get(pid, 0) for pid in placement_post_ids} + + @staticmethod + async def count_subscriptions_by_placement_batch(placement_ids: list[uuid.UUID]) -> dict[uuid.UUID, int]: + if not placement_ids: + return {} + + from tortoise.functions import Count + + results = ( + await domain.Subscription.filter(placement_id__in=placement_ids) + .group_by('placement_id') + .annotate(count=Count('id')) + .values('placement_id', 'count') + ) + + counts = {row['placement_id']: row['count'] for row in results} + return {pid: counts.get(pid, 0) for pid in placement_ids} @staticmethod async def has_placement_posts_for_creative(creative_id: uuid.UUID) -> bool: diff --git a/src/adapter/telegram_bot.py b/src/adapter/telegram_bot.py index 9cc4a79..1c5498a 100644 --- a/src/adapter/telegram_bot.py +++ b/src/adapter/telegram_bot.py @@ -1,7 +1,13 @@ import logging +from collections.abc import Sequence from typing import Any -from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, InputMediaPhoto, InputMediaVideo +from aiogram.types import ( + InlineKeyboardButton, + InlineKeyboardMarkup, + InputMediaPhoto, + InputMediaVideo, +) from shared.telegram_base import TelegramBase from src.usecase import TelegramBotWriter @@ -85,14 +91,14 @@ class TelegramBot(TelegramBase, TelegramBotWriter): async def send_media_group( self, chat_id: int, - media_items: list[Any], + media_items: Sequence[Any], caption: str | None = None, parse_mode: str | None = None, ) -> None: if not media_items: return - group = [] + group: list[InputMediaPhoto | InputMediaVideo] = [] for index, item in enumerate(media_items): item_caption = caption if index == 0 else None if item.media_type == 'photo': @@ -106,7 +112,8 @@ class TelegramBot(TelegramBase, TelegramBotWriter): else: raise ValueError(f'Unsupported media type for group: {item.media_type}') - await self.bot.send_media_group(chat_id=chat_id, media=group) + # Type ignore: aiogram expects Union type but we only use Photo/Video + await self.bot.send_media_group(chat_id=chat_id, media=group) # type: ignore[arg-type] async def edit_message_text(self, text: str, chat_id: int, message_id: int) -> None: await self.bot.edit_message_text(chat_id=chat_id, message_id=message_id, text=text, reply_markup=None) diff --git a/src/domain/creative.py b/src/domain/creative.py index fd01885..6c7e1da 100644 --- a/src/domain/creative.py +++ b/src/domain/creative.py @@ -15,6 +15,7 @@ from .error import ( ) if TYPE_CHECKING: + from .project import Project from .user import User @@ -46,6 +47,7 @@ class Creative(TimestampedModel): if TYPE_CHECKING: project_id: UUID created_by_user_id: UUID | None + media_items: 'fields.ReverseRelation[CreativeMedia]' class Meta: table = 'creative' diff --git a/src/dto/__init__.py b/src/dto/__init__.py index 278b462..efa9b87 100644 --- a/src/dto/__init__.py +++ b/src/dto/__init__.py @@ -125,12 +125,12 @@ from .analytics import ( from .channel import ( AttachChannelToWorkspaceInput, ChannelOutput, + CreateChannelInput, + CreateChannelResult, + CreateChannelsInput, + CreateChannelsOutput, GetChannelInput, GetChannelsInput, - CreateChannelInput, - CreateChannelsInput, - CreateChannelResult, - CreateChannelsOutput, GetChannelsOutput, ) from .creative import ( diff --git a/src/usecase/__init__.py b/src/usecase/__init__.py index 9c3a54d..cf8da6c 100644 --- a/src/usecase/__init__.py +++ b/src/usecase/__init__.py @@ -1,4 +1,5 @@ import typing +from collections.abc import Sequence from dataclasses import dataclass from uuid import UUID @@ -94,7 +95,7 @@ class TelegramBotWriter(typing.Protocol): async def send_media_group( self, chat_id: int, - media_items: list[MediaItem], + media_items: Sequence[MediaItem], caption: str | None = None, parse_mode: str | None = None, ) -> None: ... diff --git a/src/usecase/analytics/get_overview_analytics.py b/src/usecase/analytics/get_overview_analytics.py index b4310ba..2100d5f 100644 --- a/src/usecase/analytics/get_overview_analytics.py +++ b/src/usecase/analytics/get_overview_analytics.py @@ -86,13 +86,17 @@ async def get_overview_analytics( elif placement_date >= previous_period_start and placement_date < input.date_from: previous_placements.append(placement_post) - placement_ids = [p.id for p in placements] + placement_post_ids = [p.id for p in placements] subscriptions = await self.database.get_subscriptions_for_placement_posts( - placement_ids, date_from=previous_period_start, date_to=input.date_to + placement_post_ids, date_from=previous_period_start, date_to=input.date_to ) - subs_per_placement_current: dict[UUID, int] = defaultdict(int) - subs_per_placement_previous: dict[UUID, int] = defaultdict(int) + # Map placement_post_id to placement_id for subscription aggregation + # Since subscriptions link to placement, we need to map back to placement_post + placement_post_to_placement: dict[UUID, UUID] = {p.id: p.placement_id for p in placements} + + subs_per_placement_post_current: dict[UUID, int] = defaultdict(int) + subs_per_placement_post_previous: dict[UUID, int] = defaultdict(int) subs_per_day_current: dict[datetime.date, int] = defaultdict(int) for sub in subscriptions: @@ -100,17 +104,24 @@ async def get_overview_analytics( if created_at is None: continue + # Find which placement_post(s) this subscription belongs to via placement_id + # A placement can have multiple placement_posts, so we count it for each + for pp_id, p_id in placement_post_to_placement.items(): + if p_id == sub.placement_id: + if created_at >= input.date_from and created_at <= input.date_to: + subs_per_placement_post_current[pp_id] += 1 + elif created_at >= previous_period_start and created_at < input.date_from: + subs_per_placement_post_previous[pp_id] += 1 + + # Count each subscription only once for daily stats if created_at >= input.date_from and created_at <= input.date_to: - subs_per_placement_current[sub.placement_post_id] += 1 subs_per_day_current[created_at.date()] += 1 - elif created_at >= previous_period_start and created_at < input.date_from: - subs_per_placement_previous[sub.placement_post_id] += 1 post_ids = [p.post.id for p in placements if p.post] views_map = await self.database.get_latest_views_data_batch(post_ids) if post_ids else {} def _aggregate_totals( - placement_list: list[domain.PlacementPost], subs_per_placement: dict[UUID, int] + placement_list: list[domain.PlacementPost], subs_per_placement_post: dict[UUID, int] ) -> tuple[float, int, int]: total_cost = 0.0 total_subscriptions = 0 @@ -119,7 +130,7 @@ async def get_overview_analytics( for placement_post in placement_list: total_cost += _get_cost(placement_post) - subs = subs_per_placement.get(placement_post.id, 0) + subs = subs_per_placement_post.get(placement_post.id, 0) total_subscriptions += subs if placement_post.post and placement_post.post.id in views_map: @@ -127,8 +138,12 @@ async def get_overview_analytics( return total_cost, total_subscriptions, total_views - current_cost, current_subs, current_views = _aggregate_totals(current_placements, subs_per_placement_current) - previous_cost, previous_subs, previous_views = _aggregate_totals(previous_placements, subs_per_placement_previous) + current_cost, current_subs, current_views = _aggregate_totals( + current_placements, subs_per_placement_post_current + ) + previous_cost, previous_subs, previous_views = _aggregate_totals( + previous_placements, subs_per_placement_post_previous + ) # Apply hide_subscriptions filter if hide_subscriptions: @@ -159,7 +174,7 @@ async def get_overview_analytics( cost_value = _get_cost(placement_post) cost_per_day[placement_day] += cost_value project_spending_map[placement.project_id] += cost_value - subs = subs_per_placement_current.get(placement_post.id, 0) + subs = subs_per_placement_post_current.get(placement_post.id, 0) channel_id = placement.channel_id if channel_id not in channel_stats: diff --git a/src/usecase/project/tg_add_project.py b/src/usecase/project/tg_add_project.py index bd38ad2..44d3e0d 100644 --- a/src/usecase/project/tg_add_project.py +++ b/src/usecase/project/tg_add_project.py @@ -124,7 +124,11 @@ async def tg_add_project(self: 'Usecase', input: dto.ConnectProjectInput) -> dto if channel.telegram_id != telegram_id: channel.telegram_id = telegram_id updated = True - if parser_response and parser_response.access_hash is not None and channel.access_hash != parser_response.access_hash: + if ( + parser_response + and parser_response.access_hash is not None + and channel.access_hash != parser_response.access_hash + ): channel.access_hash = parser_response.access_hash updated = True if parser_response and parser_response.pts is not None and channel.pts != parser_response.pts: diff --git a/src/usecase/subscription/handle_subscription.py b/src/usecase/subscription/handle_subscription.py index 8d6b1b2..7b2430b 100644 --- a/src/usecase/subscription/handle_subscription.py +++ b/src/usecase/subscription/handle_subscription.py @@ -17,11 +17,13 @@ async def handle_subscription( first_name: str | None = None, last_name: str | None = None, ) -> None: - 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) + placement_post = await self.database.get_placement_post_by_invite_link(invite_link) + if not placement_post or not placement_post.placement: + log.warning('PlacementPost not found for invite_link: %s', invite_link) return + placement = placement_post.placement + subscriber = await self.database.get_telegram_user(telegram_id=user_telegram_id) if not subscriber: subscriber = domain.TelegramUser( @@ -57,6 +59,7 @@ async def handle_subscription( return # Проверяем, была ли раньше подписка через ЭТОТ placement (для реактивации) + # Note: Subscription now links to placement directly, not placement_post existing_sub = await self.database.get_subscription_by_subscriber_and_placement( subscriber.id, placement.id )