LongArticleTaskServer/applications/utils/get_cover.py

from aiomysql import DictCursor


# fetch
async def fetch_channel_info(pools, content_id):
    """use content id to get channel_content_id && channel code"""
    fetch_query = f"""
        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 = '{content_id}';
    """
    fetch_response = await pools.async_fetch(
        query=fetch_query, db_name="aigc_db_pool", cursor_type=DictCursor
    )
    return fetch_response


async def fetch_aigc_cover(pools, channel_content_id):
    """
    use channel_content_id to find article cover
    """
    fetch_query = f"""
        select image_url, oss_object_key
        from crawler_content_image
        where channel_content_id = '{channel_content_id}' and image_type = 2;
    """
    fetch_response = await pools.async_fetch(
        query=fetch_query, db_name="aigc_db_pool", cursor_type=DictCursor
    )

    return fetch_response


async def fetch_long_video_cover(pools, channel_content_id):
    """
    use channel_content_id to find long video cover
    """
    fetch_query = f"""
        select image_path
        from video_cover_snapshots
        where video_id = '{channel_content_id}';
    """
    fetch_response = await pools.async_fetch(
        query=fetch_query, db_name="long_video_db_pool", cursor_type=DictCursor
    )
    return fetch_response