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)
        )

    # 从 AIGC 数据查询Daily发文账号信息
    @staticmethod
    async def get_daily_publish_accounts(pool: DatabaseManager) -> List[Dict]:
        query = """
            select distinct t3.name, t3.gh_id, t3.follower_count, t3.create_timestamp as account_init_timestamp,
                t4.service_type_info as account_type, t4.verify_type_info as account_auth, t3.id as account_id,
                group_concat(distinct t5.remark) as account_remark
            from
                publish_plan t1
                join publish_plan_account t2 on t1.id = t2.plan_id
                join publish_account t3 on t2.account_id = t3.id
                left join publish_account_wx_type t4 on t3.id = t4.account_id
                left join publish_account_remark t5 on t3.id = t5.publish_account_id
            where t1.plan_status = 1 and t1.content_modal = 3 and t3.channel = 5
            group by t3.id;
        """
        account_list = await pool.async_fetch(query, db_name="aigc")
        return [i for i in account_list if "自动回复" not in str(i["account_remark"])]