LongArticleTaskServer/applications/service/get_cover.py

from applications.service.response import Response
from applications.utils import fetch_channel_info
from applications.utils import fetch_aigc_cover
from applications.utils import fetch_long_video_cover


class GetCoverService(Response):
    def __init__(self, pool, params):
        self.pool = pool
        self.params = params
        self.aigc_prefix = "http://res.cybertogether.net/"
        self.pq_prefix = "https://rescdn.yishihui.com/"
        self.suffix = "?x-oss-process=image/resize,m_fill,w_450,h_360/format,jpg"

    async def montage_cover(self, oss_key, pool_name):
        match pool_name:
            case "aigc_db_pool":
                return self.aigc_prefix + oss_key + self.suffix
            case "long_video_db_pool":
                return self.pq_prefix + oss_key + self.suffix
            case _:
                return None

    async def get_cover(self, content_id: str):
        channel_info = await fetch_channel_info(self.pool, content_id)
        channel_content_id = channel_info[0]["channel_content_id"]
        channel_type = channel_info[0]["channel"]
        match channel_type:
            case 5:
                fetch_response = await fetch_aigc_cover(self.pool, channel_content_id)
                if fetch_response:
                    image_oss = fetch_response[0]["oss_object_key"]
                    cover = await self.montage_cover(image_oss, "aigc_db_pool")
                else:
                    return self.error_response(
                        error_code="402",
                        error_message="can't find cover in aigc system",
                    )
            case 6:
                fetch_response = await fetch_aigc_cover(self.pool, channel_content_id)
                if fetch_response:
                    image_oss = fetch_response[0]["oss_object_key"]
                    cover = await self.montage_cover(image_oss, "aigc_db_pool")
                else:
                    return self.error_response(
                        error_code="402",
                        error_message="can't find cover in aigc system",
                    )
            case 10:
                fetch_response = await fetch_long_video_cover(
                    self.pool, channel_content_id
                )
                if fetch_response:
                    image_oss = fetch_response[1]["image_path"]
                    cover = await self.montage_cover(image_oss, "long_video_db_pool")
                else:
                    return self.error_response(
                        error_code="402",
                        error_message="can't find cover in pq",
                    )
            case _:
                return self.error_response(
                    error_code="403",
                    error_message="channel_type is not supported",
                )

        return self.success_response(
            data={
                "cover": cover,
            }
        )

    async def deal(self):
        content_id = self.params.get("content_id")
        if not content_id:
            return self.error_response(
                error_code="401",
                error_message="content_id is required",
            )
        else:
            return await self.get_cover(content_id)