275 lines
11 KiB
Python
275 lines
11 KiB
Python
import typing
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from uuid import UUID
|
|
|
|
if typing.TYPE_CHECKING:
|
|
from aiogram.types import InlineKeyboardButton
|
|
|
|
from src.adapter.postgres import Postgres
|
|
|
|
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_overview_analytics import get_overview_analytics
|
|
from .analytics.get_placements_analytics import get_placements_analytics
|
|
from .analytics.get_projects_analytics import get_projects_analytics
|
|
from .analytics.get_spending_analytics import get_spending_analytics
|
|
from .auth.attach_login_token_message import attach_login_token_message
|
|
from .auth.create_telegram_login_token import create_telegram_login_token
|
|
from .auth.get_jwt_by_telegram_id import get_jwt_by_telegram_id
|
|
from .auth.get_me import get_me
|
|
from .auth.validate_login_token import validate_login_token
|
|
from .channel.attach_channel_to_workspace import attach_channel_to_workspace
|
|
from .channel.create_channels import create_channels
|
|
from .channel.get_channel import get_channel
|
|
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.update_creative import update_creative
|
|
from .placement.fetch_placement_post_cycle import fetch_placement_post_cycle
|
|
from .placement.update_post_status_cycle import update_post_status_cycle
|
|
from .project.archive_project import archive_project, unarchive_project
|
|
from .project.delete_project import delete_project
|
|
from .project.disconnect_project_by_tg_id import disconnect_project_by_tg_id
|
|
from .project.get_project import get_project
|
|
from .project.get_workspace_projects import get_workspace_projects
|
|
from .project.move_project_to_workspace import move_project_to_workspace
|
|
from .project.tg_add_project import tg_add_project
|
|
from .project.update_project_invite_link_type import update_project_invite_link_type
|
|
from .project.update_project_permissions import update_project_permissions
|
|
from .purchase.build_placement_creative import build_placement_creative
|
|
from .purchase.create_placements import create_placements
|
|
from .purchase.delete_placement import delete_placement
|
|
from .purchase.get_placement import get_placement_user
|
|
from .purchase.get_placements import get_placements
|
|
from .purchase.update_placement import update_placement
|
|
from .purchase.update_placement_post import update_placement_post
|
|
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.accept_workspace_invite import accept_workspace_invite
|
|
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.delete_workspace_avatar import delete_workspace_avatar
|
|
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_avatar import update_workspace_avatar
|
|
from .workspace.update_workspace_member_permissions import update_workspace_member_permissions
|
|
|
|
|
|
class MediaItem(typing.Protocol):
|
|
media_type: str
|
|
media_file_id: str
|
|
|
|
|
|
class TelegramBotWriter(typing.Protocol):
|
|
async def send_message(
|
|
self,
|
|
text: str,
|
|
chat_id: int,
|
|
parse_mode: str | None = None,
|
|
disable_preview: bool = False,
|
|
reply_to_message_id: int | None = None,
|
|
) -> int: ...
|
|
|
|
async def send_message_with_inline_keyboard(
|
|
self,
|
|
text: str,
|
|
chat_id: int,
|
|
buttons: list[list['InlineKeyboardButton']],
|
|
parse_mode: str | None = None,
|
|
disable_preview: bool = False,
|
|
reply_to_message_id: int | None = None,
|
|
) -> int: ...
|
|
|
|
async def send_media_with_inline_keyboard(
|
|
self,
|
|
text: str,
|
|
chat_id: int,
|
|
media_type: str,
|
|
media_file_id: str,
|
|
buttons: list[list['InlineKeyboardButton']],
|
|
parse_mode: str | None = None,
|
|
reply_to_message_id: int | None = None,
|
|
) -> int: ...
|
|
|
|
async def send_media_group(
|
|
self,
|
|
chat_id: int,
|
|
media_items: Sequence[MediaItem],
|
|
caption: str | None = None,
|
|
parse_mode: str | None = None,
|
|
reply_to_message_id: int | None = None,
|
|
) -> int: ...
|
|
|
|
async def edit_message_text(self, text: str, chat_id: int, message_id: int) -> 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, name: str | None = None
|
|
) -> str: ...
|
|
|
|
|
|
class JWTEncoder(typing.Protocol):
|
|
def encode_access_token(self, user_id: UUID, telegram_id: int, username: str | None = None) -> str: ...
|
|
|
|
|
|
class FetchChannelResponse(typing.Protocol):
|
|
telegram_id: int
|
|
username: str | None
|
|
title: str | None
|
|
access_hash: int | None
|
|
pts: int | None
|
|
|
|
|
|
class Parser(typing.Protocol):
|
|
async def fetch_telegram_channel(self, username: str) -> FetchChannelResponse | None: ...
|
|
async def resolve_telegram_channel_by_invite(self, invite_link: str) -> FetchChannelResponse | None: ...
|
|
|
|
|
|
class S3Storage(typing.Protocol):
|
|
async def upload(self, key: str, data: bytes, content_type: str) -> None: ...
|
|
|
|
async def get(self, key: str) -> bytes: ...
|
|
|
|
async def delete(self, key: str) -> None: ...
|
|
|
|
def public_url(self, key: str) -> str: ...
|
|
|
|
|
|
@dataclass
|
|
class Usecase:
|
|
database: 'Postgres'
|
|
telegram_bot: TelegramBotWriter
|
|
jwt_encoder: JWTEncoder
|
|
parser: Parser
|
|
s3: S3Storage
|
|
|
|
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 ensure_analytics_permission(
|
|
self, workspace_id: UUID, user_id: UUID, *, for_project_id: UUID | None = None
|
|
) -> domain.WorkspacePermissionContext:
|
|
"""Ensure user has any analytics permission."""
|
|
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 not permissions.has_any_analytics_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
|
|
|
|
if not user.telegram_user:
|
|
raise domain.UserNotFound(user.id)
|
|
|
|
telegram_user = user.telegram_user
|
|
workspace_name = telegram_user.username or f'Workspace {telegram_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
|
|
|
|
validate_login_token = validate_login_token
|
|
create_telegram_login_token = create_telegram_login_token
|
|
attach_login_token_message = attach_login_token_message
|
|
get_jwt_by_telegram_id = get_jwt_by_telegram_id
|
|
get_me = get_me
|
|
tg_add_project = tg_add_project
|
|
get_workspace_projects = get_workspace_projects
|
|
get_project = get_project
|
|
archive_project = archive_project
|
|
unarchive_project = unarchive_project
|
|
delete_project = delete_project
|
|
move_project_to_workspace = move_project_to_workspace
|
|
disconnect_project_by_tg_id = disconnect_project_by_tg_id
|
|
update_project_permissions = update_project_permissions
|
|
update_project_invite_link_type = update_project_invite_link_type
|
|
get_channels = get_channels
|
|
create_channels = create_channels
|
|
get_channel = get_channel
|
|
attach_channel_to_workspace = attach_channel_to_workspace
|
|
# Placement (user-managed) use cases
|
|
create_placements = create_placements
|
|
get_placements = get_placements
|
|
get_placement_user = get_placement_user
|
|
build_placement_creative = build_placement_creative
|
|
update_placement = update_placement
|
|
update_placement_post = update_placement_post
|
|
delete_placement = delete_placement
|
|
# Creative use cases
|
|
get_creatives = get_creatives
|
|
get_creative = get_creative
|
|
create_creative = create_creative
|
|
update_creative = update_creative
|
|
delete_creative = delete_creative
|
|
# PlacementPost (system-managed) use cases
|
|
fetch_placement_post_cycle = fetch_placement_post_cycle
|
|
update_post_status_cycle = update_post_status_cycle
|
|
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_projects_analytics = get_projects_analytics
|
|
get_spending_analytics = get_spending_analytics
|
|
get_overview_analytics = get_overview_analytics
|
|
get_workspaces = get_workspaces
|
|
create_workspace = create_workspace
|
|
update_workspace = update_workspace
|
|
delete_workspace = delete_workspace
|
|
update_workspace_avatar = update_workspace_avatar
|
|
delete_workspace_avatar = delete_workspace_avatar
|
|
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
|
|
accept_workspace_invite = accept_workspace_invite
|
|
tg_accept_workspace_invite = tg_accept_workspace_invite
|