LongArticleTaskServer/applications/tasks/data_recycle_tasks/recycle_outside_account_articles.py

import time, json

import datetime
import traceback
import urllib.parse
from tqdm.asyncio import tqdm

from .recycle_daily_publish_articles import UpdateRootSourceIdAndUpdateTimeTask
from .recycle_daily_publish_articles import Const
from applications.crawler.wechat import get_article_list_from_account
from applications.crawler.wechat import get_article_detail
from applications.pipeline import insert_outside_article_into_recycle_pool
from applications.api import feishu_robot


class RecycleOutsideAccountArticlesTask(Const):
    def __init__(self, pool, log_client, date_string):
        self.pool = pool
        self.log_client = log_client
        self.date_string = date_string

    async def get_outside_accounts(self):
        query = """
            select
                t2.group_source_name as account_source,
                t3.name as name,
                t3.gh_id as gh_id
            from wx_statistics_group_source t1
                join wx_statistics_group_source_account t2 on t2.group_source_name = t1.account_source_name
                join publish_account t3 on t3.id = t2.account_id
            where
                t1.mode_type = '代运营服务号' and
                (
                    t2.group_source_name like '%云誉%'
                        or t2.group_source_name like '%微小盟%'
                        or t2.group_source_name like '%阿雅达%'
                        or t2.group_source_name like '%创易%'
                    )
            and t3.status = 1 and t3.name != '';
        """
        return await self.pool.async_fetch(query=query, db_name="aigc")

    async def recycle_single_account(self, account):
        """recycle single account"""
        query = """
                select max(update_time) as publish_timestamp \
                from outside_account_articles
                where gh_id = %s;
                """
        response = await self.pool.async_fetch(query=query, params=(account["gh_id"],))
        if response:
            max_publish_timestamp = response[0]["publish_timestamp"]
        else:
            max_publish_timestamp = int(time.time()) - self.NEW_ACCOUNT_CRAWL_PERIOD

        cursor = None
        while True:
            response = await get_article_list_from_account(
                account_id=account["gh_id"], index=cursor
            )
            response_code = response["code"]
            match response_code:
                case self.ACCOUNT_FORBIDDEN_CODE:
                    # await feishu_robot.bot(
                    #     title="发布账号封禁",
                    #     detail={
                    #         "账号名称": account["name"],
                    #         "账号id": account["gh_id"],
                    #     },
                    # )
                    return
                case self.ARTICLE_SUCCESS_CODE:
                    msg_list = response.get("data", {}).get("data", [])
                    if not msg_list:
                        return

                    await insert_outside_article_into_recycle_pool(
                        self.pool, self.log_client, msg_list, account
                    )

                    # check last article
                    last_article = msg_list[-1]
                    last_publish_timestamp = last_article["AppMsg"]["BaseInfo"][
                        "UpdateTime"
                    ]
                    if last_publish_timestamp <= max_publish_timestamp:
                        return

                    cursor = response["data"].get("next_cursor")
                    if not cursor:
                        return
                case self.CRAWL_CRASH_CODE:
                    await self.log_client.log(
                        contents={
                            "task": "recycle_daily_publish_articles",
                            "data": {
                                "gh_id": account["gh_id"],
                            },
                            "message": "爬虫挂掉",
                            "status": "fail",
                        }
                    )
                case _:
                    return

    async def deal(self):
        subscription_accounts = await self.get_outside_accounts()
        for account in tqdm(subscription_accounts, desc="recycle each account"):
            try:
                await self.recycle_single_account(account)

            except Exception as e:
                print(
                    f"{account['name']}\t{account['gh_id']}: recycle account error:", e
                )


class UpdateOutsideRootSourceIdAndUpdateTimeTask(UpdateRootSourceIdAndUpdateTimeTask):
    def __init__(self, pool, log_client):
        super().__init__(pool, log_client)

    async def get_outside_article_list_v2(self) -> list[dict]:
        query = """
                select content_url, wx_sn 
                from outside_account_articles where publish_timestamp in %s
                order by update_time desc;
            """
        article_list = await self.pool.async_fetch(
            query=query, params=(tuple([0, -1]),)
        )
        return article_list

    async def check_each_article(self, article: dict):
        url = article["content_url"]
        wx_sn = article["wx_sn"]
        try:
            response = await get_article_detail(url)
            response_code = response["code"]

            if response_code == self.ARTICLE_DELETE_CODE:
                publish_timestamp_s = self.DELETE_STATUS
                root_source_id_list = []
            elif response_code == self.ARTICLE_ILLEGAL_CODE:
                publish_timestamp_s = self.ILLEGAL_STATUS
                root_source_id_list = []
            elif response_code == self.ARTICLE_SUCCESS_CODE:
                data = response["data"]["data"]
                publish_timestamp_ms = data["publish_timestamp"]
                publish_timestamp_s = int(publish_timestamp_ms / 1000)
                mini_program = data.get("mini_program", [])
                if mini_program:
                    root_source_id_list = [
                        urllib.parse.parse_qs(urllib.parse.unquote(i["path"])).get(
                            "rootSourceId", [""]
                        )[0]
                        for i in mini_program
                    ]
                else:
                    root_source_id_list = []
            else:
                publish_timestamp_s = self.UNKNOWN_STATUS
                root_source_id_list = []
        except Exception as e:
            publish_timestamp_s = self.REQUEST_FAIL_STATUS
            root_source_id_list = None
            error_msg = traceback.format_exc()
            await self.log_client.log(
                contents={
                    "task": "get_official_article_detail",
                    "data": {
                        "url": url,
                        "wx_sn": wx_sn,
                        "error_msg": error_msg,
                        "error": str(e),
                    },
                    "function": "check_each_article",
                    "status": "fail",
                }
            )
        query = """
            update outside_account_articles set publish_timestamp = %s, root_source_id_list = %s
            where wx_sn = %s;
        """
        await self.pool.async_save(
            query=query,
            params=(
                publish_timestamp_s,
                json.dumps(root_source_id_list, ensure_ascii=False),
                wx_sn,
            ),
        )
        if publish_timestamp_s == self.REQUEST_FAIL_STATUS:
            article["wx_sn"] = wx_sn
            return article
        else:
            return None

    async def fallback_mechanism(self):
        # # 通过msgId 来修改publish_timestamp
        # update_sql = f"""
        #     update outside_account_articles oaa
        #     join (
        #         select gh_id, app_msg_id, max(publish_timestamp) as publish_timestamp
        #         from outside_account_articles
        #         where publish_timestamp > %s
        #         group by gh_id, app_msg_id
        #     ) vv on oaa.app_msg_id = vv.app_msg_id and oaa.gh_id = vv.gh_id
        #     set oaa.publish_timestamp = vv.publish_timestamp
        #     where oaa.publish_timestamp <= %s;
        # """
        # affected_rows_1 = await self.pool.async_save(query=update_sql, params=(0, 0))

        # 若还是无 publish_timestamp，用update_time当作 publish_timestamp
        update_sql_2 = f"""
            update outside_account_articles
            set publish_timestamp = update_time
            where publish_timestamp < %s;
        """
        affected_rows_2 = await self.pool.async_save(query=update_sql_2, params=(0,))
        if affected_rows_2:
            await feishu_robot.bot(
                title="执行兜底修改发布时间戳",
                detail={
                    # "通过msgId修改": affected_rows_1,
                    "通过create_timestamp修改": affected_rows_2,
                },
                mention=False,
            )

    async def deal(self):
        task_list = await self.get_outside_article_list_v2()
        for task in tqdm(task_list, desc="get article detail step1: "):
            try:
                await self.check_each_article(task)
            except Exception as e:
                try:
                    await self.log_client.log(
                        contents={
                            "task": "get_official_article_detail_step1",
                            "data": {
                                "detail": {
                                    "url": task["ContentUrl"],
                                    "wx_sn": task["wx_sn"],
                                },
                                "error_msg": traceback.format_exc(),
                                "error": str(e),
                            },
                            "function": "check_each_article",
                            "status": "fail",
                        }
                    )
                except Exception as e:
                    print(e)
                    print(traceback.format_exc())

        # # process_failed_task_reproduce
        # fail_tasks = await self.get_article_list()
        # fail_list = []
        # for fail_task in tqdm(fail_tasks, desc="get article detail step2: "):
        #     try:
        #         res = await self.check_each_article(fail_task)
        #         if res:
        #             fail_list.append(res)
        #     except Exception as e:
        #         await self.log_client.log(
        #             contents={
        #                 "task": "get_official_article_detail_step2",
        #                 "data": {
        #                     "detail": {
        #                         "url": fail_task["ContentUrl"],
        #                         "wx_sn": fail_task["wx_sn"].decode("utf-8"),
        #                     },
        #                     "error_msg": traceback.format_exc(),
        #                     "error": str(e),
        #                 },
        #                 "function": "check_each_article",
        #                 "status": "fail",
        #             }
        #         )
        # if fail_list:
        #     await feishu_robot.bot(title="更新文章，获取detail失败", detail=fail_list)
        #
        #     current_hour = datetime.datetime.now().hour
        #     if current_hour >= 21:
        #         await self.fallback_mechanism()