136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
import asyncio
|
|
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 update_creative(
|
|
self: 'Usecase',
|
|
creative_id: uuid.UUID,
|
|
input: dto.UpdateCreativeInput,
|
|
user_id: uuid.UUID,
|
|
workspace_id: uuid.UUID,
|
|
) -> dto.CreativeOutput:
|
|
context = await self.ensure_workspace_permission(workspace_id, user_id, domain.PermissionKey.CREATIVES_WRITE)
|
|
|
|
creative = await self.database.get_creative(workspace_id, creative_id)
|
|
if not creative:
|
|
log.warning('User %s attempted to update unavailable creative %s', user_id, creative_id)
|
|
raise domain.CreativeNotFound(creative_id)
|
|
|
|
context.ensure_project_permission(domain.PermissionKey.CREATIVES_WRITE, creative.project_id)
|
|
|
|
if input.name:
|
|
creative.name = input.name
|
|
if input.text:
|
|
creative.text = domain.replace_invite_link_with_tag(input.text)
|
|
media_items: list[dto.CreativeMediaInput] | None = None
|
|
if input.media_items is not None:
|
|
media_items = input.media_items
|
|
domain.validate_media_items([item.media_type for item in media_items])
|
|
for item in media_items:
|
|
domain.validate_media_size(item.media_data)
|
|
if input.buttons is not None:
|
|
creative.buttons = [button.model_dump() for button in input.buttons]
|
|
if input.status:
|
|
creative.status = input.status
|
|
if input.tag:
|
|
creative.tag = input.tag
|
|
|
|
await self.database.update_creative(creative)
|
|
|
|
if media_items is not None:
|
|
await _replace_media_items(self, creative.id, workspace_id, media_items)
|
|
|
|
placements_count = await self.database.count_placement_posts_by_creative(creative.id)
|
|
creative_media = await _get_media_items(creative)
|
|
|
|
return dto.CreativeOutput(
|
|
id=creative.id,
|
|
name=creative.name,
|
|
text=creative.text,
|
|
media_items=creative_media,
|
|
buttons=creative.buttons,
|
|
project_id=creative.project_id,
|
|
project_channel_title=creative.project.channel.title,
|
|
created_at=creative.created_at,
|
|
status=creative.status,
|
|
tag=creative.tag,
|
|
placements_count=placements_count,
|
|
)
|
|
|
|
|
|
def _get_content_type(media_type: str | None) -> str:
|
|
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],
|
|
) -> None:
|
|
existing_items = await domain.CreativeMedia.filter(creative_id=creative_id).all()
|
|
if existing_items:
|
|
old_media_keys = [item.media_s3_key for item in existing_items if item.media_s3_key]
|
|
await domain.CreativeMedia.filter(creative_id=creative_id).delete()
|
|
|
|
if old_media_keys:
|
|
|
|
async def delete_old_media() -> None:
|
|
for key in old_media_keys:
|
|
try:
|
|
await self.s3.delete(key)
|
|
log.info('Deleted old creative media from S3: %s', key)
|
|
except Exception as e:
|
|
log.warning('Failed to delete old creative media from S3: %s', e)
|
|
|
|
asyncio.create_task(delete_old_media())
|
|
|
|
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 new creative media to S3: %s', media_s3_key)
|
|
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,
|
|
)
|
|
|
|
|
|
async def _get_media_items(creative: domain.Creative) -> list[dto.CreativeMediaItem]:
|
|
media_rel = creative.media_items
|
|
if hasattr(media_rel, 'all'):
|
|
items = await media_rel.all().order_by('position')
|
|
else:
|
|
items = sorted(media_rel, key=lambda item: item.position)
|
|
return [
|
|
dto.CreativeMediaItem(
|
|
media_type=item.media_type,
|
|
media_file_id=item.media_file_id,
|
|
position=item.position,
|
|
)
|
|
for item in items
|
|
]
|