100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
import logging
|
|
import uuid
|
|
from typing import TYPE_CHECKING
|
|
|
|
from src import domain, dto
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
async def create_creative(
|
|
self: 'Usecase', input: dto.CreateCreativeInput, project_id: uuid.UUID, user_id: uuid.UUID, workspace_id: uuid.UUID
|
|
) -> dto.CreativeOutput:
|
|
await self.ensure_workspace_permission(
|
|
workspace_id, user_id, domain.PermissionKey.CREATIVES_WRITE, for_project_id=project_id
|
|
)
|
|
|
|
project = await self.database.get_project(workspace_id, project_id=project_id)
|
|
if not project:
|
|
log.warning('User %s attempted to create creative for unavailable project %s', user_id, project_id)
|
|
raise domain.ProjectNotFound(project_id)
|
|
|
|
creative_text = domain.replace_invite_link_with_tag(input.text)
|
|
media_items = input.media_items or []
|
|
domain.validate_media_items([item.media_type for item in media_items])
|
|
for item in media_items:
|
|
domain.validate_media_size(item.media_data)
|
|
|
|
creative = domain.Creative(
|
|
name=input.name,
|
|
text=creative_text,
|
|
buttons=[button.model_dump() for button in input.buttons],
|
|
status=domain.CreativeStatus.ACTIVE,
|
|
project_id=project.id,
|
|
created_by_user_id=user_id,
|
|
)
|
|
await self.database.create_creative(creative)
|
|
|
|
created_media = await _replace_media_items(self, creative.id, workspace_id, media_items)
|
|
|
|
return dto.CreativeOutput(
|
|
id=creative.id,
|
|
name=creative.name,
|
|
text=creative.text,
|
|
media_items=created_media,
|
|
buttons=creative.buttons,
|
|
project_id=project.id,
|
|
project_channel_title=project.channel.title,
|
|
created_at=creative.created_at,
|
|
status=creative.status,
|
|
placements_count=0,
|
|
)
|
|
|
|
|
|
def _get_content_type(media_type: str | None) -> str:
|
|
"""Map Telegram media type to MIME content type."""
|
|
if not media_type:
|
|
return 'application/octet-stream'
|
|
|
|
mapping = {
|
|
'photo': 'image/jpeg',
|
|
'video': 'video/mp4',
|
|
'animation': 'image/gif',
|
|
}
|
|
return mapping.get(media_type, 'application/octet-stream')
|
|
|
|
|
|
async def _replace_media_items(
|
|
self: 'Usecase',
|
|
creative_id: uuid.UUID,
|
|
workspace_id: uuid.UUID,
|
|
media_items: list[dto.CreativeMediaInput],
|
|
) -> list[dto.CreativeMediaItem]:
|
|
created: list[dto.CreativeMediaItem] = []
|
|
for position, item in enumerate(media_items):
|
|
media_s3_key: str | None = None
|
|
if item.media_data:
|
|
file_id = uuid.uuid4()
|
|
media_s3_key = f'creatives/{workspace_id}/{file_id}'
|
|
content_type = _get_content_type(item.media_type)
|
|
await self.s3.upload(media_s3_key, item.media_data, content_type)
|
|
log.info('Uploaded creative media to S3: %s', media_s3_key)
|
|
media = await domain.CreativeMedia.create(
|
|
creative_id=creative_id,
|
|
media_type=item.media_type,
|
|
media_file_id=item.media_file_id,
|
|
media_s3_key=media_s3_key,
|
|
position=position,
|
|
)
|
|
created.append(
|
|
dto.CreativeMediaItem(
|
|
media_type=media.media_type,
|
|
media_file_id=media.media_file_id,
|
|
position=media.position,
|
|
)
|
|
)
|
|
return created
|