LongArticleTaskServer/app/infra/mapper/aigc_mapper.py

from typing import List, Dict

from app.core.database import DatabaseManager


class AigcDatabaseMapper:
    # 从 aigc 数据库查询 channel_content_id && channel 信息
    @staticmethod
    async def fetch_channel_info(pool: DatabaseManager, content_id: str) -> List[Dict]:
        query = """
            SELECT t1.channel_content_id, t2.channel
            FROM produce_plan_exe_record t1 JOIN crawler_content t2 ON t1.channel_content_id = t2.channel_content_id
            WHERE plan_exe_id = %s;
        """
        return await pool.async_fetch(query=query, db_name="aigc", params=(content_id,))

    # 从 aigc 如何查询文章封面信息
    @staticmethod
    async def fetch_aigc_cover(
        pool: DatabaseManager, channel_content_id: str
    ) -> List[Dict]:
        """
        use channel_content_id to find article cover
        """
        COVER_TYPE = 2
        query = """
            SELECT image_url, oss_object_key
            FROM crawler_content_image
            WHERE channel_content_id = %s AND image_type = %s;
        """
        return await pool.async_fetch(
            query=query, db_name="aigc", params=(channel_content_id, COVER_TYPE)
        )