1028 lines
38 KiB
Python
1028 lines
38 KiB
Python
import datetime
|
|
import logging
|
|
import typing
|
|
import uuid
|
|
|
|
from tortoise import timezone
|
|
from tortoise.transactions import in_transaction
|
|
|
|
from shared.datebase_base import DatabaseBase
|
|
from src import domain
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class Postgres(DatabaseBase):
|
|
@staticmethod
|
|
def transaction() -> typing.AsyncContextManager[None]:
|
|
return in_transaction()
|
|
|
|
@staticmethod
|
|
async def create_user(user: domain.User) -> None:
|
|
await user.save()
|
|
|
|
@staticmethod
|
|
async def get_telegram_user(
|
|
telegram_user_id: uuid.UUID | None = None, telegram_id: int | None = None
|
|
) -> domain.TelegramUser | None:
|
|
if telegram_user_id:
|
|
return await domain.TelegramUser.get_or_none(id=telegram_user_id)
|
|
if telegram_id:
|
|
return await domain.TelegramUser.get_or_none(telegram_id=telegram_id)
|
|
|
|
raise ValueError('Either telegram_user_id or telegram_id must be provided')
|
|
|
|
@staticmethod
|
|
async def create_telegram_user(telegram_user: domain.TelegramUser) -> None:
|
|
await telegram_user.save()
|
|
|
|
@staticmethod
|
|
async def update_telegram_user(telegram_user: domain.TelegramUser) -> None:
|
|
await telegram_user.save()
|
|
|
|
@staticmethod
|
|
async def create_workspace(workspace: domain.Workspace) -> None:
|
|
await workspace.save()
|
|
|
|
@staticmethod
|
|
async def update_workspace(workspace: domain.Workspace) -> None:
|
|
await workspace.save()
|
|
|
|
@staticmethod
|
|
async def delete_workspace(workspace_id: uuid.UUID) -> None:
|
|
await domain.Workspace.filter(id=workspace_id).delete()
|
|
|
|
@staticmethod
|
|
async def add_user_to_workspace(workspace_id: uuid.UUID, user_id: uuid.UUID) -> domain.WorkspaceUser:
|
|
return await domain.WorkspaceUser.create(
|
|
workspace_id=workspace_id,
|
|
user_id=user_id,
|
|
status=domain.WorkspaceUserStatus.ACTIVE,
|
|
)
|
|
|
|
@staticmethod
|
|
async def update_workspace_user(workspace_user: domain.WorkspaceUser) -> None:
|
|
await workspace_user.save()
|
|
|
|
@staticmethod
|
|
async def get_user_workspaces(user_id: uuid.UUID) -> list[domain.WorkspaceUser]:
|
|
return (
|
|
await domain.WorkspaceUser.filter(user_id=user_id)
|
|
.prefetch_related('workspace')
|
|
.order_by('created_at')
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_workspace(workspace_id: uuid.UUID) -> domain.Workspace | None:
|
|
return await domain.Workspace.get_or_none(id=workspace_id)
|
|
|
|
@staticmethod
|
|
async def get_workspace_for_user(workspace_id: uuid.UUID, user_id: uuid.UUID) -> domain.Workspace | None:
|
|
membership = (
|
|
await domain.WorkspaceUser.filter(workspace_id=workspace_id, user_id=user_id)
|
|
.prefetch_related('workspace')
|
|
.first()
|
|
)
|
|
return membership.workspace if membership else None
|
|
|
|
@staticmethod
|
|
async def get_default_workspace_for_user(user_id: uuid.UUID) -> domain.Workspace | None:
|
|
membership = (
|
|
await domain.WorkspaceUser.filter(user_id=user_id)
|
|
.prefetch_related('workspace')
|
|
.order_by('created_at')
|
|
.first()
|
|
)
|
|
return membership.workspace if membership else None
|
|
|
|
@staticmethod
|
|
async def get_workspace_membership(workspace_id: uuid.UUID, user_id: uuid.UUID) -> domain.WorkspaceUser | None:
|
|
return (
|
|
await domain.WorkspaceUser.filter(workspace_id=workspace_id, user_id=user_id)
|
|
.prefetch_related('workspace', 'user', 'user__telegram_user', 'permissions', 'permission_scopes')
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_workspace_members(workspace_id: uuid.UUID) -> list[domain.WorkspaceUser]:
|
|
return (
|
|
await domain.WorkspaceUser.filter(workspace_id=workspace_id)
|
|
.prefetch_related('workspace', 'user__telegram_user', 'permissions', 'permission_scopes')
|
|
.order_by('created_at')
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_workspace_member(workspace_user_id: uuid.UUID) -> domain.WorkspaceUser | None:
|
|
return (
|
|
await domain.WorkspaceUser.filter(id=workspace_user_id)
|
|
.prefetch_related('workspace', 'user__telegram_user', 'permissions', 'permission_scopes')
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
async def create_workspace_invite(invite: domain.WorkspaceInvite) -> None:
|
|
await invite.save()
|
|
|
|
@staticmethod
|
|
async def update_workspace_invite(invite: domain.WorkspaceInvite) -> None:
|
|
await invite.save()
|
|
|
|
@staticmethod
|
|
async def get_workspace_invite(invite_id: uuid.UUID) -> domain.WorkspaceInvite | None:
|
|
return (
|
|
await domain.WorkspaceInvite.filter(id=invite_id)
|
|
.prefetch_related('workspace', 'user__telegram_user', 'invited_by__telegram_user')
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_workspace_invite_by_user(
|
|
workspace_id: uuid.UUID, user_id: uuid.UUID
|
|
) -> domain.WorkspaceInvite | None:
|
|
return await domain.WorkspaceInvite.get_or_none(workspace_id=workspace_id, user_id=user_id)
|
|
|
|
@staticmethod
|
|
async def get_workspace_invites(workspace_id: uuid.UUID) -> list[domain.WorkspaceInvite]:
|
|
return (
|
|
await domain.WorkspaceInvite.filter(workspace_id=workspace_id)
|
|
.prefetch_related('user__telegram_user', 'invited_by__telegram_user')
|
|
.order_by('created_at')
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def set_workspace_user_permissions(
|
|
workspace_user_id: uuid.UUID,
|
|
global_permissions: set[domain.PermissionKey],
|
|
scoped_permissions: list[tuple[domain.PermissionKey, domain.PermissionScopeType, uuid.UUID]],
|
|
) -> None:
|
|
await domain.WorkspaceUserPermission.filter(workspace_user_id=workspace_user_id).delete()
|
|
await domain.WorkspaceUserPermissionScope.filter(workspace_user_id=workspace_user_id).delete()
|
|
|
|
if global_permissions:
|
|
await domain.WorkspaceUserPermission.bulk_create(
|
|
[
|
|
domain.WorkspaceUserPermission(workspace_user_id=workspace_user_id, permission=permission)
|
|
for permission in global_permissions
|
|
]
|
|
)
|
|
|
|
if scoped_permissions:
|
|
scope_objects = []
|
|
for permission, scope_type, scope_id in scoped_permissions:
|
|
kwargs = {
|
|
'workspace_user_id': workspace_user_id,
|
|
'permission': permission,
|
|
}
|
|
if scope_type == domain.PermissionScopeType.PROJECT:
|
|
kwargs['project_id'] = scope_id
|
|
elif scope_type == domain.PermissionScopeType.CREATIVE:
|
|
kwargs['creative_id'] = scope_id
|
|
elif scope_type == domain.PermissionScopeType.PLACEMENT:
|
|
kwargs['placement_id'] = scope_id
|
|
elif scope_type == domain.PermissionScopeType.CHANNEL:
|
|
kwargs['channel_id'] = scope_id
|
|
|
|
scope_objects.append(domain.WorkspaceUserPermissionScope(**kwargs))
|
|
|
|
await domain.WorkspaceUserPermissionScope.bulk_create(scope_objects)
|
|
|
|
@staticmethod
|
|
async def create_login_token(login_token: domain.LoginToken) -> None:
|
|
await login_token.save()
|
|
|
|
@staticmethod
|
|
async def create_creative(creative: domain.Creative) -> None:
|
|
await creative.save()
|
|
|
|
@staticmethod
|
|
async def get_user(user_id: uuid.UUID | None = None, telegram_id: int | None = None) -> domain.User | None:
|
|
if user_id:
|
|
return await domain.User.filter(id=user_id).prefetch_related('telegram_user').first()
|
|
elif telegram_id:
|
|
return (
|
|
await domain.User.filter(telegram_user__telegram_id=telegram_id)
|
|
.prefetch_related('telegram_user')
|
|
.first()
|
|
)
|
|
|
|
raise ValueError('Either user_id or telegram_id must be provided')
|
|
|
|
@staticmethod
|
|
async def get_user_by_username(username: str) -> domain.User | None:
|
|
return (
|
|
await domain.User.filter(telegram_user__username__iexact=username).prefetch_related('telegram_user').first()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_login_token(token: str) -> domain.LoginToken | None:
|
|
return await domain.LoginToken.get_or_none(token=token)
|
|
|
|
@staticmethod
|
|
async def mark_token_as_used(token: str) -> None:
|
|
updated = await domain.LoginToken.filter(token=token, used_at__isnull=True).update(used_at=timezone.now())
|
|
|
|
if updated == 0:
|
|
raise domain.LoginTokenAlreadyUsed() # либо нет токена, либо он уже использован
|
|
|
|
@staticmethod
|
|
async def update_login_token_message_id(token: str, message_id: int) -> None:
|
|
updated = await domain.LoginToken.filter(token=token).update(message_id=message_id)
|
|
if updated == 0:
|
|
raise domain.LoginTokenNotFound()
|
|
|
|
@staticmethod
|
|
async def get_channel(
|
|
channel_id: uuid.UUID | None = None, telegram_id: int | None = None, username: str | None = None
|
|
) -> domain.Channel | None:
|
|
if channel_id:
|
|
return await domain.Channel.get_or_none(id=channel_id)
|
|
if telegram_id:
|
|
return await domain.Channel.get_or_none(telegram_id=telegram_id)
|
|
if username:
|
|
return await domain.Channel.filter(username__iexact=username).first()
|
|
return None
|
|
|
|
@staticmethod
|
|
async def create_channel(channel: domain.Channel) -> None:
|
|
await channel.save()
|
|
|
|
@staticmethod
|
|
async def update_channel(channel: domain.Channel) -> None:
|
|
await channel.save()
|
|
|
|
@staticmethod
|
|
async def search_channels(username_query: str | None = None) -> list[domain.Channel]:
|
|
query = domain.Channel.all()
|
|
|
|
if username_query:
|
|
# Частичный поиск (case-insensitive)
|
|
query = query.filter(username__icontains=username_query)
|
|
|
|
return await query.all()
|
|
|
|
@staticmethod
|
|
async def get_project(
|
|
workspace_id: uuid.UUID, project_id: uuid.UUID | None = None, channel_id: uuid.UUID | None = None
|
|
) -> domain.Project | None:
|
|
query = domain.Project.filter(workspace_id=workspace_id, deleted_at__isnull=True).prefetch_related('channel')
|
|
if project_id:
|
|
query = query.filter(id=project_id)
|
|
if channel_id:
|
|
query = query.filter(channel_id=channel_id)
|
|
return await query.first()
|
|
|
|
@staticmethod
|
|
async def get_project_for_user_by_telegram(user_id: uuid.UUID, channel_telegram_id: int) -> domain.Project | None:
|
|
return (
|
|
await domain.Project.filter(
|
|
channel__telegram_id=channel_telegram_id, workspace__workspace_users__user_id=user_id
|
|
)
|
|
.prefetch_related('channel')
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_project_by_channel_telegram(channel_telegram_id: int) -> domain.Project | None:
|
|
return await domain.Project.filter(channel__telegram_id=channel_telegram_id).prefetch_related('channel').first()
|
|
|
|
@staticmethod
|
|
async def create_project(project: domain.Project) -> None:
|
|
await project.save()
|
|
|
|
@staticmethod
|
|
async def update_project(project: domain.Project) -> None:
|
|
await project.save()
|
|
|
|
@staticmethod
|
|
async def get_workspace_projects(
|
|
workspace_id: uuid.UUID,
|
|
allowed_project_ids: set[uuid.UUID] | None = None,
|
|
include_archived: bool = False,
|
|
) -> list[domain.Project]:
|
|
query = domain.Project.filter(workspace_id=workspace_id, deleted_at__isnull=True).prefetch_related('channel')
|
|
|
|
if allowed_project_ids is not None:
|
|
if not allowed_project_ids:
|
|
return []
|
|
query = query.filter(id__in=list(allowed_project_ids))
|
|
|
|
if not include_archived:
|
|
query = query.filter(status=domain.ProjectStatus.ACTIVE)
|
|
|
|
return await query.order_by('created_at').all()
|
|
|
|
@staticmethod
|
|
async def archive_project(workspace_id: uuid.UUID, project_id: uuid.UUID) -> None:
|
|
project = await domain.Project.get_or_none(id=project_id, workspace_id=workspace_id)
|
|
if not project:
|
|
raise domain.ProjectNotFound()
|
|
project.status = domain.ProjectStatus.ARCHIVED
|
|
await project.save()
|
|
|
|
@staticmethod
|
|
async def unarchive_project(workspace_id: uuid.UUID, project_id: uuid.UUID) -> None:
|
|
project = await domain.Project.get_or_none(id=project_id, workspace_id=workspace_id)
|
|
if not project:
|
|
raise domain.ProjectNotFound()
|
|
project.status = domain.ProjectStatus.ACTIVE
|
|
await project.save()
|
|
|
|
@staticmethod
|
|
async def delete_project(workspace_id: uuid.UUID, project_id: uuid.UUID) -> None:
|
|
project = await domain.Project.get_or_none(id=project_id, workspace_id=workspace_id)
|
|
if not project:
|
|
raise domain.ProjectNotFound()
|
|
|
|
project.deleted_at = timezone.now()
|
|
await project.save()
|
|
|
|
@staticmethod
|
|
async def check_channel_exists_in_workspace(channel_id: uuid.UUID, workspace_id: uuid.UUID) -> bool:
|
|
project = await domain.Project.get_or_none(channel_id=channel_id, workspace_id=workspace_id)
|
|
return project is not None
|
|
|
|
@staticmethod
|
|
async def get_creative(workspace_id: uuid.UUID, creative_id: uuid.UUID) -> domain.Creative | None:
|
|
return await domain.Creative.get_or_none(id=creative_id, project__workspace_id=workspace_id).prefetch_related(
|
|
'project', 'project__channel', 'media_items'
|
|
)
|
|
|
|
@staticmethod
|
|
async def update_creative(creative: domain.Creative) -> None:
|
|
await creative.save()
|
|
|
|
@staticmethod
|
|
async def get_workspace_creatives(
|
|
workspace_id: uuid.UUID,
|
|
project_id: uuid.UUID | None = None,
|
|
include_archived: bool = False,
|
|
allowed_project_ids: set[uuid.UUID] | None = None,
|
|
created_by_user_id: uuid.UUID | None = None,
|
|
tag: domain.CreativeTag | None = None,
|
|
) -> list[domain.Creative]:
|
|
query = domain.Creative.filter(project__workspace_id=workspace_id)
|
|
|
|
if project_id:
|
|
query = query.filter(project_id=project_id)
|
|
elif allowed_project_ids is not None:
|
|
if not allowed_project_ids:
|
|
return []
|
|
query = query.filter(project_id__in=list(allowed_project_ids))
|
|
|
|
if created_by_user_id is not None:
|
|
query = query.filter(created_by_user_id=created_by_user_id)
|
|
|
|
if tag is not None:
|
|
query = query.filter(tag=tag)
|
|
|
|
if not include_archived:
|
|
query = query.filter(status=domain.CreativeStatus.ACTIVE)
|
|
|
|
return await query.prefetch_related('project', 'project__channel', 'media_items').order_by('-created_at').all()
|
|
|
|
@staticmethod
|
|
async def delete_creative(creative_id: uuid.UUID) -> None:
|
|
await domain.Creative.filter(id=creative_id).delete()
|
|
|
|
@staticmethod
|
|
async def create_placement(placement: domain.Placement) -> None:
|
|
await placement.save()
|
|
|
|
@staticmethod
|
|
async def get_placement(workspace_id: uuid.UUID, placement_id: uuid.UUID) -> domain.Placement | None:
|
|
return await domain.Placement.get_or_none(id=placement_id, project__workspace_id=workspace_id).prefetch_related(
|
|
'project', 'project__channel', 'channel', 'creative'
|
|
)
|
|
|
|
@staticmethod
|
|
async def count_placements_by_project_and_channel(
|
|
project_id: uuid.UUID,
|
|
channel_id: uuid.UUID,
|
|
) -> int:
|
|
"""Count placements for a project in a specific channel."""
|
|
return await domain.Placement.filter(
|
|
project_id=project_id,
|
|
channel_id=channel_id,
|
|
).count()
|
|
|
|
@staticmethod
|
|
async def update_placement(placement: domain.Placement) -> None:
|
|
await placement.save()
|
|
|
|
@staticmethod
|
|
async def delete_placement(placement_id: uuid.UUID) -> None:
|
|
await domain.Placement.filter(id=placement_id).delete()
|
|
|
|
@staticmethod
|
|
async def get_project_placements(
|
|
workspace_id: uuid.UUID, project_id: uuid.UUID, include_archived: bool = False
|
|
) -> list[domain.Placement]:
|
|
query = domain.Placement.filter(project__workspace_id=workspace_id, project_id=project_id)
|
|
|
|
if not include_archived:
|
|
query = query.filter(
|
|
status__in=[
|
|
domain.PlacementStatus.NO_STATUS,
|
|
domain.PlacementStatus.WRITE,
|
|
domain.PlacementStatus.WAITING_RESPONSE,
|
|
domain.PlacementStatus.TERMS_APPROVAL,
|
|
domain.PlacementStatus.TO_PAY,
|
|
domain.PlacementStatus.PAID,
|
|
]
|
|
)
|
|
|
|
return (
|
|
await query.prefetch_related('project', 'project__channel', 'channel', 'creative')
|
|
.order_by('-placement_at', '-created_at')
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_placement_post(workspace_id: uuid.UUID, placement_post_id: uuid.UUID) -> domain.PlacementPost | None:
|
|
return await domain.PlacementPost.get_or_none(
|
|
id=placement_post_id, placement__project__workspace_id=workspace_id
|
|
).prefetch_related(
|
|
'placement',
|
|
'placement__project',
|
|
'placement__project__channel',
|
|
'placement__channel',
|
|
'placement__creative',
|
|
'post',
|
|
'post__channel',
|
|
)
|
|
|
|
@staticmethod
|
|
async def count_placement_posts_by_placement(placement_id: uuid.UUID) -> int:
|
|
return await domain.PlacementPost.filter(placement_id=placement_id).count()
|
|
|
|
@staticmethod
|
|
async def get_workspace_placement_posts(
|
|
workspace_id: uuid.UUID,
|
|
project_id: uuid.UUID | None = None,
|
|
placement_channel_id: uuid.UUID | None = None,
|
|
creative_id: uuid.UUID | None = None,
|
|
placement_id: uuid.UUID | None = None,
|
|
include_archived: bool = False,
|
|
allowed_project_ids: set[uuid.UUID] | None = None,
|
|
date_from: datetime.datetime | None = None,
|
|
date_to: datetime.datetime | None = None,
|
|
has_post: bool = False,
|
|
) -> list[domain.PlacementPost]:
|
|
query = domain.PlacementPost.filter(placement__project__workspace_id=workspace_id)
|
|
|
|
if project_id:
|
|
query = query.filter(placement__project_id=project_id)
|
|
elif allowed_project_ids is not None:
|
|
if not allowed_project_ids:
|
|
return []
|
|
query = query.filter(placement__project_id__in=list(allowed_project_ids))
|
|
if placement_channel_id:
|
|
query = query.filter(placement__channel_id=placement_channel_id)
|
|
if creative_id:
|
|
query = query.filter(placement__creative_id=creative_id)
|
|
if placement_id:
|
|
query = query.filter(placement_id=placement_id)
|
|
|
|
if date_from:
|
|
query = query.filter(created_at__gte=date_from)
|
|
if date_to:
|
|
query = query.filter(created_at__lte=date_to)
|
|
|
|
if has_post:
|
|
query = query.filter(post_id__isnull=False)
|
|
|
|
return (
|
|
await query.prefetch_related(
|
|
'placement',
|
|
'placement__project',
|
|
'placement__project__channel',
|
|
'placement__channel',
|
|
'placement__creative',
|
|
'post',
|
|
'post__channel',
|
|
)
|
|
.order_by('-created_at')
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_placement_posts_by_placement_ids(
|
|
workspace_id: uuid.UUID,
|
|
placement_ids: list[uuid.UUID],
|
|
include_archived: bool = False,
|
|
) -> list[domain.PlacementPost]:
|
|
if not placement_ids:
|
|
return []
|
|
|
|
query = domain.PlacementPost.filter(
|
|
placement__project__workspace_id=workspace_id,
|
|
placement_id__in=placement_ids,
|
|
)
|
|
|
|
return (
|
|
await query.prefetch_related(
|
|
'placement',
|
|
'placement__project',
|
|
'placement__project__channel',
|
|
'placement__channel',
|
|
'placement__creative',
|
|
'post',
|
|
'post__channel',
|
|
)
|
|
.order_by('-created_at')
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def create_placement_post(placement_post: domain.PlacementPost) -> None:
|
|
await placement_post.save()
|
|
|
|
@staticmethod
|
|
async def get_placement_post_by_invite_link(invite_link: str) -> domain.PlacementPost | None:
|
|
return await domain.PlacementPost.get_or_none(placement__invite_link=invite_link).prefetch_related(
|
|
'placement',
|
|
'placement__project',
|
|
'placement__project__channel',
|
|
'placement__channel',
|
|
'placement__creative',
|
|
)
|
|
|
|
# Subscription methods
|
|
@staticmethod
|
|
async def create_subscription(subscription: domain.Subscription) -> None:
|
|
await subscription.save()
|
|
|
|
@staticmethod
|
|
async def get_subscription_by_subscriber_and_placement_post(
|
|
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_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
|
|
async def update_subscription(subscription: domain.Subscription) -> None:
|
|
await subscription.save()
|
|
|
|
@staticmethod
|
|
async def get_subscriptions_for_placement_posts(
|
|
placement_post_ids: list[uuid.UUID],
|
|
*,
|
|
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 []
|
|
|
|
# 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)
|
|
if date_to:
|
|
query = query.filter(created_at__lte=date_to)
|
|
|
|
return await query.all()
|
|
|
|
@staticmethod
|
|
async def get_active_subscriptions_by_subscriber_and_project(
|
|
telegram_user_id: uuid.UUID, project_id: uuid.UUID
|
|
) -> list[domain.Subscription]:
|
|
return (
|
|
await domain.Subscription.filter(
|
|
telegram_user_id=telegram_user_id,
|
|
placement_post__placement__project_id=project_id,
|
|
status=domain.SubscriptionStatus.ACTIVE,
|
|
)
|
|
.prefetch_related('placement_post', 'telegram_user')
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_active_subscription_by_subscriber_and_project(
|
|
telegram_user_id: uuid.UUID, project_id: uuid.UUID
|
|
) -> domain.Subscription | None:
|
|
return (
|
|
await domain.Subscription.filter(
|
|
telegram_user_id=telegram_user_id,
|
|
placement_post__placement__project_id=project_id,
|
|
status=domain.SubscriptionStatus.ACTIVE,
|
|
)
|
|
.prefetch_related('placement_post', 'telegram_user')
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_views_history(
|
|
post_id: uuid.UUID, *, from_date: datetime.datetime | None = None, to_date: datetime.datetime | None = None
|
|
) -> list[domain.PostViewsHistory]:
|
|
query = domain.PostViewsHistory.filter(post_id=post_id)
|
|
|
|
if from_date:
|
|
query = query.filter(fetched_at__gte=from_date)
|
|
if to_date:
|
|
query = query.filter(fetched_at__lte=to_date)
|
|
|
|
return await query.order_by('fetched_at').all()
|
|
|
|
@staticmethod
|
|
async def get_latest_views_data_batch(post_ids: list[uuid.UUID]) -> dict[uuid.UUID, tuple[int, datetime.datetime]]:
|
|
if not post_ids:
|
|
return {}
|
|
|
|
results: dict[uuid.UUID, tuple[int, datetime.datetime]] = {}
|
|
for post_id in post_ids:
|
|
latest = await domain.PostViewsHistory.filter(post_id=post_id).order_by('-fetched_at').first()
|
|
if latest:
|
|
results[post_id] = (latest.views_count, latest.fetched_at)
|
|
|
|
return results
|
|
|
|
# Count methods
|
|
@staticmethod
|
|
async def count_placement_posts_by_creative(creative_id: uuid.UUID) -> int:
|
|
return await domain.PlacementPost.filter(placement__creative_id=creative_id).count()
|
|
|
|
@staticmethod
|
|
async def count_subscriptions_by_placement_post(placement_post_id: uuid.UUID) -> int:
|
|
"""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]:
|
|
if not creative_ids:
|
|
return {}
|
|
|
|
from tortoise.functions import Count
|
|
|
|
results = (
|
|
await domain.PlacementPost.filter(placement__creative_id__in=creative_ids)
|
|
.group_by('placement__creative_id')
|
|
.annotate(count=Count('id'))
|
|
.values('placement__creative_id', 'count')
|
|
)
|
|
|
|
counts = {row['placement__creative_id']: row['count'] for row in results}
|
|
return {cid: counts.get(cid, 0) for cid in creative_ids}
|
|
|
|
@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_id__in=placement_ids)
|
|
.group_by('placement_id')
|
|
.annotate(count=Count('id'))
|
|
.values('placement_id', 'count')
|
|
)
|
|
|
|
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 count_unsubscriptions_by_placement_post_batch(
|
|
placement_post_ids: list[uuid.UUID],
|
|
) -> dict[uuid.UUID, int]:
|
|
"""Count unsubscriptions 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 unsubscriptions by placement (filter by UNSUBSCRIBED status)
|
|
results = (
|
|
await domain.Subscription.filter(
|
|
placement_id__in=placement_ids, status=domain.SubscriptionStatus.UNSUBSCRIBED
|
|
)
|
|
.group_by('placement_id')
|
|
.annotate(count=Count('id'))
|
|
.values('placement_id', 'count')
|
|
)
|
|
|
|
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 has_placement_posts_for_creative(creative_id: uuid.UUID) -> bool:
|
|
return await domain.PlacementPost.filter(placement__creative_id=creative_id).exists()
|
|
|
|
@staticmethod
|
|
async def get_next_post_after(channel_id: uuid.UUID, message_id: int) -> domain.Post | None:
|
|
"""Get first post in channel after the given message_id."""
|
|
return (
|
|
await domain.Post.filter(
|
|
channel_id=channel_id,
|
|
message_id__gt=message_id,
|
|
deleted_from_channel_at__isnull=True,
|
|
)
|
|
.order_by('message_id')
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
async def get_next_posts_after_batch(
|
|
channel_message_pairs: list[tuple[uuid.UUID, int]],
|
|
) -> dict[tuple[uuid.UUID, int], domain.Post]:
|
|
"""Batch version: get next post for each (channel_id, message_id) pair."""
|
|
if not channel_message_pairs:
|
|
return {}
|
|
|
|
results: dict[tuple[uuid.UUID, int], domain.Post] = {}
|
|
for channel_id, message_id in channel_message_pairs:
|
|
next_post = (
|
|
await domain.Post.filter(
|
|
channel_id=channel_id,
|
|
message_id__gt=message_id,
|
|
deleted_from_channel_at__isnull=True,
|
|
)
|
|
.order_by('message_id')
|
|
.first()
|
|
)
|
|
if next_post:
|
|
results[(channel_id, message_id)] = next_post
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
async def get_workspace_placements_for_analytics(
|
|
workspace_id: uuid.UUID,
|
|
project_ids: list[uuid.UUID] | None = None,
|
|
channel_ids: list[uuid.UUID] | None = None,
|
|
creative_ids: list[uuid.UUID] | None = None,
|
|
status_list: list[str] | None = None,
|
|
cost_types: list[str] | None = None,
|
|
placement_types: list[str] | None = None,
|
|
invite_link_types: list[str] | None = None,
|
|
cost_min: float | None = None,
|
|
cost_max: float | None = None,
|
|
views_min: int | None = None,
|
|
views_max: int | None = None,
|
|
subscriptions_min: int | None = None,
|
|
subscriptions_max: int | None = None,
|
|
cpm_min: float | None = None,
|
|
cpm_max: float | None = None,
|
|
channel_title_contains: str | None = None,
|
|
creative_name_contains: str | None = None,
|
|
comment_contains: str | None = None,
|
|
placement_date_from: datetime.datetime | None = None,
|
|
placement_date_to: datetime.datetime | None = None,
|
|
sort_by: str = 'created_at',
|
|
sort_direction: str = 'desc',
|
|
offset: int = 0,
|
|
limit: int = 50,
|
|
include_archived: bool = False,
|
|
allowed_project_ids: set[uuid.UUID] | None = None,
|
|
) -> list[domain.Placement]:
|
|
"""Get placements for analytics with flexible filtering, sorting, and pagination."""
|
|
from tortoise.queryset import QuerySet
|
|
|
|
query: QuerySet[domain.Placement] = domain.Placement.filter(
|
|
project__workspace_id=workspace_id,
|
|
)
|
|
|
|
if project_ids:
|
|
query = query.filter(project_id__in=project_ids)
|
|
elif allowed_project_ids is not None:
|
|
if not allowed_project_ids:
|
|
return []
|
|
query = query.filter(project_id__in=list(allowed_project_ids))
|
|
|
|
if channel_ids:
|
|
query = query.filter(channel_id__in=channel_ids)
|
|
|
|
if creative_ids:
|
|
query = query.filter(creative_id__in=creative_ids)
|
|
|
|
if status_list:
|
|
query = query.filter(status__in=status_list)
|
|
|
|
if cost_types:
|
|
query = query.filter(cost_type__in=cost_types)
|
|
|
|
if placement_types:
|
|
query = query.filter(placement_type__in=placement_types)
|
|
|
|
if invite_link_types:
|
|
query = query.filter(invite_link_type__in=invite_link_types)
|
|
|
|
if cost_min is not None:
|
|
query = query.filter(cost_value__gte=cost_min)
|
|
if cost_max is not None:
|
|
query = query.filter(cost_value__lte=cost_max)
|
|
|
|
if placement_date_from:
|
|
query = query.filter(placement_at__gte=placement_date_from)
|
|
if placement_date_to:
|
|
query = query.filter(placement_at__lte=placement_date_to)
|
|
|
|
if channel_title_contains:
|
|
query = query.filter(channel__title__icontains=channel_title_contains)
|
|
|
|
if creative_name_contains:
|
|
query = query.filter(creative__name__icontains=creative_name_contains)
|
|
|
|
if comment_contains:
|
|
query = query.filter(comment__icontains=comment_contains)
|
|
|
|
elif not include_archived:
|
|
query = query.filter(
|
|
status__in=[
|
|
domain.PlacementStatus.NO_STATUS,
|
|
domain.PlacementStatus.WRITE,
|
|
domain.PlacementStatus.WAITING_RESPONSE,
|
|
domain.PlacementStatus.TERMS_APPROVAL,
|
|
domain.PlacementStatus.TO_PAY,
|
|
domain.PlacementStatus.PAID,
|
|
]
|
|
)
|
|
|
|
valid_sort_fields = ['created_at', 'updated_at', 'placement_at', 'cost_value']
|
|
if sort_by not in valid_sort_fields:
|
|
sort_by = 'created_at'
|
|
|
|
if sort_direction == 'asc':
|
|
query = query.order_by(sort_by)
|
|
else:
|
|
query = query.order_by(f'-{sort_by}')
|
|
|
|
return (
|
|
await query.prefetch_related(
|
|
'project',
|
|
'project__channel',
|
|
'channel',
|
|
'creative',
|
|
'placement_posts',
|
|
'placement_posts__post',
|
|
'placement_posts__post__channel',
|
|
)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
async def count_workspace_placements_for_analytics(
|
|
workspace_id: uuid.UUID,
|
|
project_ids: list[uuid.UUID] | None = None,
|
|
channel_ids: list[uuid.UUID] | None = None,
|
|
creative_ids: list[uuid.UUID] | None = None,
|
|
status_list: list[str] | None = None,
|
|
cost_types: list[str] | None = None,
|
|
placement_types: list[str] | None = None,
|
|
invite_link_types: list[str] | None = None,
|
|
cost_min: float | None = None,
|
|
cost_max: float | None = None,
|
|
views_min: int | None = None,
|
|
views_max: int | None = None,
|
|
subscriptions_min: int | None = None,
|
|
subscriptions_max: int | None = None,
|
|
cpm_min: float | None = None,
|
|
cpm_max: float | None = None,
|
|
channel_title_contains: str | None = None,
|
|
creative_name_contains: str | None = None,
|
|
comment_contains: str | None = None,
|
|
placement_date_from: datetime.datetime | None = None,
|
|
placement_date_to: datetime.datetime | None = None,
|
|
include_archived: bool = False,
|
|
allowed_project_ids: set[uuid.UUID] | None = None,
|
|
) -> int:
|
|
"""Count placements for analytics with flexible filtering."""
|
|
query = domain.Placement.filter(project__workspace_id=workspace_id)
|
|
|
|
if project_ids:
|
|
query = query.filter(project_id__in=project_ids)
|
|
elif allowed_project_ids is not None:
|
|
if not allowed_project_ids:
|
|
return 0
|
|
query = query.filter(project_id__in=list(allowed_project_ids))
|
|
|
|
if channel_ids:
|
|
query = query.filter(channel_id__in=channel_ids)
|
|
|
|
if creative_ids:
|
|
query = query.filter(creative_id__in=creative_ids)
|
|
|
|
if status_list:
|
|
query = query.filter(status__in=status_list)
|
|
|
|
if cost_types:
|
|
query = query.filter(cost_type__in=cost_types)
|
|
|
|
if placement_types:
|
|
query = query.filter(placement_type__in=placement_types)
|
|
|
|
if invite_link_types:
|
|
query = query.filter(invite_link_type__in=invite_link_types)
|
|
|
|
if cost_min is not None:
|
|
query = query.filter(cost_value__gte=cost_min)
|
|
if cost_max is not None:
|
|
query = query.filter(cost_value__lte=cost_max)
|
|
|
|
if placement_date_from:
|
|
query = query.filter(placement_at__gte=placement_date_from)
|
|
if placement_date_to:
|
|
query = query.filter(placement_at__lte=placement_date_to)
|
|
|
|
if channel_title_contains:
|
|
query = query.filter(channel__title__icontains=channel_title_contains)
|
|
|
|
if creative_name_contains:
|
|
query = query.filter(creative__name__icontains=creative_name_contains)
|
|
|
|
if comment_contains:
|
|
query = query.filter(comment__icontains=comment_contains)
|
|
|
|
elif not include_archived:
|
|
query = query.filter(
|
|
status__in=[
|
|
domain.PlacementStatus.NO_STATUS,
|
|
domain.PlacementStatus.WRITE,
|
|
domain.PlacementStatus.WAITING_RESPONSE,
|
|
domain.PlacementStatus.TERMS_APPROVAL,
|
|
domain.PlacementStatus.TO_PAY,
|
|
domain.PlacementStatus.PAID,
|
|
]
|
|
)
|
|
|
|
return await query.count()
|