Source code for services.posts.media

from typing import Any, Dict, List

from django.conf import settings
from django.db.models import F

from apps.posts.constants import ContentTypeChoices
from apps.posts.models import Post, PostMedia
from tasks.posts import compress_image, send_video_service




class PostMediaService:
    """Post Media Service"""



    @classmethod
    def create(cls, post: Post, media_list: List):
        """
        Method for create media
        :param post: Post object
        :param media_list: List of image and video
        """
        created_media = cls._create_post_media(post, media_list)
        cls.compress_media(created_media)
        return created_media


    @classmethod
    def _compress_image(cls, media: PostMedia) -> None:
        if settings.IS_TESTING:
            compress_image(media.id)
        else:
            compress_image.delay(media.id)

    @classmethod
    def _compress_video(cls, media: PostMedia) -> None:
        if settings.IS_TESTING:
            send_video_service(media.id)
        else:
            send_video_service.delay(media.id)

    @classmethod
    def _create_post_media(cls, post: Post, media_list: List) -> List[PostMedia]:
        """
        Method for bulk create media
        :param post: Post object
        :param media_list: List of image and video
        """
        media_objects = []
        for item in media_list:
            media = PostMedia(post=post, **item)
            media_objects.append(media)
        return PostMedia.objects.bulk_create(media_objects)



    @classmethod
    def compress_media(cls, medias: List[PostMedia]) -> None:
        for media in medias:
            if media.type == ContentTypeChoices.VIDEO:
                cls._compress_video(media)
            elif media.type == ContentTypeChoices.IMAGE:
                cls._compress_image(media)




    @classmethod
    def update(cls, post_media: PostMedia, data: Dict[str, Any]) -> PostMedia:
        """
        Update post
        :param post_media: PostMedia instance
        :param data: Validated data
        """
        if post_media.type == "image" and "original" in data:
            old_original = F("original")
            if old_original != data["original"]:
                # The 'original' field has been updated
                # You can take any desired action here
                cls._compress_image(post_media)

        for attr, value in data.items():
            setattr(post_media, attr, value)
        post_media.save()
        return post_media




    @classmethod
    def set_formatted_path(cls, post_media: PostMedia, upload_video_path: str, upload_preview_path: str) -> PostMedia:
        """
        Set proccessed video and preview from Video service
        :param post_media: PostMedia instance
        :param upload_video_path: Path to formatted video
        :param upload_preview_path: Path to preview
        """
        data = {"formatted_path": upload_video_path, "preview_path": upload_preview_path}
        instance = cls.update(post_media, data)
        # PostService(post_media.post).check_on_publish()
        return instance