LongArticleTaskServer/app/infra/mapper/piaoquan_crawler_mapper.py

from typing import List, Dict, Set

from app.core.database import DatabaseManager


class PiaoquanCrawlerDatabaseMapper:
    # 兜底修改发文时间戳, 使用相同群发的msgID
    @staticmethod
    async def fallback_mechanism_by_msg_id(pool: DatabaseManager) -> int:
        # 通过msgId 来修改publish_timestamp
        query = """
            UPDATE official_articles_v2 oav 
            JOIN (
                    SELECT ghId, appMsgId, MAX(publish_timestamp) AS publish_timestamp 
                    FROM official_articles_v2
                    WHERE publish_timestamp > %s 
                    GROUP by ghId, appMsgId
                ) vv 
                ON oav.appMsgId = vv.appMsgId AND oav.ghId = vv.ghId
            SET oav.publish_timestamp = vv.publish_timestamp
            WHERE oav.publish_timestamp <= %s;
        """
        return await pool.async_save(
            query=query, params=(0, 0), db_name="piaoquan_crawler"
        )

    # 兜底修改发文时间戳，使用 update_time
    @staticmethod
    async def fallback_mechanism_by_update_time(pool: DatabaseManager) -> int:
        query = """
            UPDATE official_articles_v2
            SET publish_timestamp = updateTime
            WHERE publish_timestamp < %s;
        """
        return await pool.async_save(
            query=query, params=(0,), db_name="piaoquan_crawler"
        )

    # 获取账号的历史发文
    @staticmethod
    async def get_published_articles(pool: DatabaseManager, gh_id: str) -> Set[str]:
        query = """
               SELECT title FROM official_articles_v2 WHERE ghId = %s AND ItemIndex = %s;
           """
        response = await pool.async_fetch(
            query=query, db_name="piaoquan_crawler", params=(gh_id, 1)
        )
        return set([i["title"] for i in response])