import datetime import typing from dataclasses import dataclass from uuid import UUID if typing.TYPE_CHECKING: from aiogram.types import InlineKeyboardButton from src import domain from .analytics.get_channel_analytics import get_channel_analytics from .analytics.get_creatives_analytics import get_creatives_analytics from .analytics.get_placements_analytics import get_placements_analytics from .analytics.get_spending_analytics import get_spending_analytics from .auth.get_me import get_me from .auth.telegram_login import telegram_login from .auth.telegram_start import telegram_start from .auth.validate_login_token import validate_login_token from .channel.get_channels import get_channels from .creative.create_creative import create_creative from .creative.delete_creative import delete_creative from .creative.get_creative import get_creative from .creative.get_creatives import get_creatives from .creative.tg_create_creative_1 import tg_create_creative_1 from .creative.tg_create_creative_2 import tg_create_creative_2 from .creative.tg_create_creative_3 import tg_create_creative_3 from .creative.tg_create_creative_4 import tg_create_creative_4 from .creative.update_creative import update_creative from .placement.create_placement import create_placement from .placement.delete_placement import delete_placement from .placement.get_placement import get_placement from .placement.get_placements import get_placements from .placement.update_placement import update_placement from .project.disconnect_project_by_tg_id import disconnect_project_by_tg_id from .project.get_workspace_projects import get_workspace_projects from .project.tg_add_project_1 import tg_add_project_1 from .project.tg_add_project_2 import tg_add_project_2 from .project.update_project_permissions import update_project_permissions from .purchase_plan.attach_channel_to_purchase_plan import attach_channel_to_purchase_plan from .purchase_plan.get_purchase_plan_channels import get_purchase_plan_channels from .purchase_plan.remove_purchase_plan_channel import remove_purchase_plan_channel from .purchase_plan.update_purchase_plan_channel import update_purchase_plan_channel from .subscription.handle_subscription import handle_subscription from .subscription.handle_unsubscription import handle_unsubscription from .views.get_views_history import get_views_history from .workspace.create_workspace import create_workspace from .workspace.create_workspace_invite import create_workspace_invite from .workspace.delete_workspace import delete_workspace from .workspace.get_workspace_invites import get_workspace_invites from .workspace.get_workspace_members import get_workspace_members from .workspace.get_workspaces import get_workspaces from .workspace.tg_accept_workspace_invite import tg_accept_workspace_invite from .workspace.update_workspace import update_workspace from .workspace.update_workspace_member_permissions import update_workspace_member_permissions class Database(typing.Protocol): def transaction(self) -> typing.AsyncContextManager[None]: ... async def get_user( self, user_id: UUID | None = None, telegram_id: int | None = None, ) -> domain.User | None: ... async def get_user_by_username(self, username: str) -> domain.User | None: ... async def create_user(self, user: domain.User) -> None: ... async def create_workspace(self, workspace: domain.Workspace) -> None: ... async def update_workspace(self, workspace: domain.Workspace) -> None: ... async def delete_workspace(self, workspace_id: UUID) -> None: ... async def add_user_to_workspace(self, workspace_id: UUID, user_id: UUID) -> domain.WorkspaceUser: ... async def set_workspace_user_permissions( self, workspace_user_id: UUID, global_permissions: set[domain.PermissionKey], scoped_permissions: list[tuple[domain.PermissionKey, domain.PermissionScopeType, UUID]], ) -> None: ... async def update_workspace_user(self, workspace_user: domain.WorkspaceUser) -> None: ... async def get_user_workspaces(self, user_id: UUID) -> list[domain.WorkspaceUser]: ... async def get_workspace(self, workspace_id: UUID) -> domain.Workspace | None: ... async def get_workspace_for_user(self, workspace_id: UUID, user_id: UUID) -> domain.Workspace | None: ... async def get_default_workspace_for_user(self, user_id: UUID) -> domain.Workspace | None: ... async def get_workspace_membership(self, workspace_id: UUID, user_id: UUID) -> domain.WorkspaceUser | None: ... async def get_workspace_members(self, workspace_id: UUID) -> list[domain.WorkspaceUser]: ... async def get_workspace_member(self, workspace_user_id: UUID) -> domain.WorkspaceUser | None: ... async def create_workspace_invite(self, invite: domain.WorkspaceInvite) -> None: ... async def update_workspace_invite(self, invite: domain.WorkspaceInvite) -> None: ... async def get_workspace_invite(self, invite_id: UUID) -> domain.WorkspaceInvite | None: ... async def get_workspace_invite_by_user( self, workspace_id: UUID, user_id: UUID ) -> domain.WorkspaceInvite | None: ... async def get_workspace_invites(self, workspace_id: UUID) -> list[domain.WorkspaceInvite]: ... async def create_login_token(self, login_token: domain.LoginToken) -> None: ... async def create_creative(self, creative: domain.Creative) -> None: ... async def get_login_token(self, token: str) -> domain.LoginToken | None: ... async def mark_token_as_used(self, token: str) -> None: ... async def update_user(self, user: domain.User) -> None: ... async def get_channel( self, channel_id: UUID | None = None, telegram_id: int | None = None, username: str | None = None, ) -> domain.Channel | None: ... async def create_channel(self, channel: domain.Channel) -> None: ... async def update_channel(self, channel: domain.Channel) -> None: ... async def delete_channel(self, channel_id: UUID) -> None: ... async def search_channels(self, username_query: str | None = None) -> list[domain.Channel]: ... async def get_project( self, workspace_id: UUID, project_id: UUID | None = None, channel_id: UUID | None = None ) -> domain.Project | None: ... async def get_project_for_user_by_telegram( self, user_id: UUID, channel_telegram_id: int ) -> domain.Project | None: ... async def get_project_by_channel_telegram(self, channel_telegram_id: int) -> domain.Project | None: ... async def create_project(self, project: domain.Project) -> None: ... async def update_project(self, project: domain.Project) -> None: ... async def get_workspace_projects( self, workspace_id: UUID, allowed_project_ids: set[UUID] | None = None ) -> list[domain.Project]: ... async def get_active_purchase_plan(self, project_id: UUID) -> domain.PurchasePlan | None: ... async def get_purchase_plan(self, workspace_id: UUID, plan_id: UUID) -> domain.PurchasePlan | None: ... async def create_purchase_plan(self, plan: domain.PurchasePlan) -> None: ... async def get_purchase_plan_channels(self, plan_id: UUID) -> list[domain.PurchasePlanChannel]: ... async def get_purchase_plan_channel( self, plan_channel_id: UUID, plan_id: UUID ) -> domain.PurchasePlanChannel | None: ... async def add_channel_to_purchase_plan( self, plan_id: UUID, channel_id: UUID, *, status: domain.PurchasePlanChannelStatus | None = None, planned_cost: float | None = None, comment: str | None = None, ) -> domain.PurchasePlanChannel: ... async def update_purchase_plan_channel(self, plan_channel: domain.PurchasePlanChannel) -> None: ... async def remove_channel_from_purchase_plan(self, plan_id: UUID, channel_id: UUID) -> None: ... async def get_creative(self, workspace_id: UUID, creative_id: UUID) -> domain.Creative | None: ... async def update_creative(self, creative: domain.Creative) -> None: ... async def get_workspace_creatives( self, workspace_id: UUID, project_id: UUID | None = None, include_archived: bool = False, allowed_project_ids: set[UUID] | None = None, ) -> list[domain.Creative]: ... async def delete_creative(self, creative_id: UUID) -> None: ... async def get_placement(self, workspace_id: UUID, placement_id: UUID) -> domain.Placement | None: ... async def create_placement(self, placement: domain.Placement) -> None: ... async def update_placement(self, placement: domain.Placement) -> None: ... async def get_workspace_placements( self, workspace_id: UUID, project_id: UUID | None = None, placement_channel_id: UUID | None = None, creative_id: UUID | None = None, include_archived: bool = False, allowed_project_ids: set[UUID] | None = None, ) -> list[domain.Placement]: ... async def delete_placement(self, placement_id: UUID) -> None: ... async def get_placement_by_invite_link(self, invite_link: str) -> domain.Placement | None: ... async def create_subscription(self, subscription: domain.Subscription) -> None: ... async def get_subscription_by_subscriber_and_placement( self, subscriber_id: UUID, placement_id: UUID ) -> domain.Subscription | None: ... async def update_subscription(self, subscription: domain.Subscription) -> None: ... async def get_active_subscriptions_by_subscriber_and_project( self, subscriber_id: UUID, project_id: UUID ) -> list[domain.Subscription]: ... async def get_active_subscription_by_subscriber_and_project( self, subscriber_id: UUID, project_id: UUID ) -> domain.Subscription | None: ... # Post methods async def create_post(self, post: domain.Post) -> None: ... async def get_post(self, post_id: UUID) -> domain.Post | None: ... async def get_post_by_channel_and_message(self, channel_id: UUID, message_id: int) -> domain.Post | None: ... # Post views history async def create_post_views_history(self, history: domain.PostViewsHistory) -> None: ... async def get_views_history( self, post_id: UUID, *, from_date: datetime.datetime | None = None, to_date: datetime.datetime | None = None, ) -> list[domain.PostViewsHistory]: ... async def get_latest_views_data(self, post_id: UUID) -> tuple[int, datetime.datetime] | None: ... async def get_latest_views_data_batch(self, post_ids: list[UUID]) -> dict[UUID, tuple[int, datetime.datetime]]: ... async def get_telegram_state(self, telegram_id: int) -> domain.TelegramState | None: ... async def set_telegram_state( self, telegram_id: int, state: domain.TelegramStateEnum, context: dict[str, typing.Any] ) -> domain.TelegramState: ... async def clear_telegram_state(self, telegram_id: int) -> None: ... class TelegramBotWriter(typing.Protocol): async def send_message(self, text: str, chat_id: int) -> None: ... async def send_message_with_button(self, text: str, chat_id: int, button_text: str, button_url: str) -> None: ... async def send_message_with_inline_keyboard( self, text: str, chat_id: int, buttons: list[list['InlineKeyboardButton']] ) -> None: ... async def edit_message_reply_markup(self, chat_id: int, message_id: int) -> None: ... async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str: ... class JWTEncoder(typing.Protocol): def encode_access_token(self, user_id: UUID, telegram_id: int, username: str | None = None) -> str: ... @dataclass class Usecase: database: Database telegram_bot: TelegramBotWriter jwt_encoder: JWTEncoder async def ensure_workspace_access(self, workspace_id: UUID, user_id: UUID) -> domain.Workspace: workspace = await self.database.get_workspace_for_user(workspace_id, user_id) if not workspace: raise domain.WorkspaceNotFound(workspace_id) return workspace async def ensure_workspace_permission( self, workspace_id: UUID, user_id: UUID, permission: domain.PermissionKey, *, for_project_id: UUID | None = None, ) -> domain.WorkspacePermissionContext: membership = await self.database.get_workspace_membership(workspace_id, user_id) if not membership or not membership.workspace: raise domain.WorkspaceNotFound(workspace_id) permissions = domain.WorkspacePermissions.from_membership(membership) if for_project_id is not None: has_permission = permissions.has_permission( permission, scope_type=domain.PermissionScopeType.PROJECT, scope_id=for_project_id, ) else: has_permission = permissions.has_any(permission) if not has_permission: raise domain.WorkspaceAccessDenied(workspace_id) return domain.build_workspace_permission_context(membership) async def get_or_create_personal_workspace(self, user: domain.User) -> domain.Workspace: workspace = await self.database.get_default_workspace_for_user(user.id) if workspace: return workspace workspace_name = user.username or f'Workspace {user.telegram_id}' workspace = domain.Workspace(name=workspace_name) async with self.database.transaction(): await self.database.create_workspace(workspace) membership = await self.database.add_user_to_workspace(workspace.id, user.id) await self.database.set_workspace_user_permissions( membership.id, global_permissions={domain.PermissionKey.ADMIN_FULL}, scoped_permissions=[], ) return workspace async def ensure_active_purchase_plan(self, project: domain.Project) -> domain.PurchasePlan: plan = await self.database.get_active_purchase_plan(project.id) if plan: return plan plan = domain.PurchasePlan(workspace_id=project.workspace_id, project_id=project.id) await self.database.create_purchase_plan(plan) return plan validate_login_token = validate_login_token telegram_login = telegram_login telegram_start = telegram_start get_me = get_me tg_add_project_1 = tg_add_project_1 tg_add_project_2 = tg_add_project_2 get_workspace_projects = get_workspace_projects disconnect_project_by_tg_id = disconnect_project_by_tg_id update_project_permissions = update_project_permissions get_channels = get_channels attach_channel_to_purchase_plan = attach_channel_to_purchase_plan get_purchase_plan_channels = get_purchase_plan_channels update_purchase_plan_channel = update_purchase_plan_channel remove_purchase_plan_channel = remove_purchase_plan_channel get_creatives = get_creatives get_creative = get_creative create_creative = create_creative update_creative = update_creative delete_creative = delete_creative tg_create_creative_1 = tg_create_creative_1 tg_create_creative_2 = tg_create_creative_2 tg_create_creative_3 = tg_create_creative_3 tg_create_creative_4 = tg_create_creative_4 get_placements = get_placements get_placement = get_placement create_placement = create_placement update_placement = update_placement delete_placement = delete_placement handle_subscription = handle_subscription handle_unsubscription = handle_unsubscription get_views_history = get_views_history get_placements_analytics = get_placements_analytics get_creatives_analytics = get_creatives_analytics get_channel_analytics = get_channel_analytics get_spending_analytics = get_spending_analytics get_workspaces = get_workspaces create_workspace = create_workspace update_workspace = update_workspace delete_workspace = delete_workspace get_workspace_members = get_workspace_members update_workspace_member_permissions = update_workspace_member_permissions create_workspace_invite = create_workspace_invite get_workspace_invites = get_workspace_invites tg_accept_workspace_invite = tg_accept_workspace_invite