LongArticleTaskServer/applications/tasks/monitor_tasks/auto_reply_cards_monitor.py

import json
import time
import uuid
import xml.etree.ElementTree as ET

from datetime import datetime, timedelta
from urllib.parse import unquote, parse_qs

from applications.utils import fetch_from_odps
from applications.utils import AsyncHttpClient
from applications.crawler.wechat import get_article_list_from_account
from applications.crawler.wechat import get_article_detail


class AutoReplyCardsMonitorConst:
    # fetch_status
    FETCH_INIT_STATUS = 0
    FETCH_PROCESSING_STATUS = 1
    FETCH_SUCCESS_STATUS = 2
    FETCH_FAIL_STATUS = 3

    # task_status
    INIT_STATUS = 0
    PROCESSING_STATUS = 1
    SUCCESS_STATUS = 2
    FAIL_STATUS = 99

    # account_status
    VALID_STATUS = 1
    INVALID_STATUS = 0


class AutoReplyCardsMonitorUtils(AutoReplyCardsMonitorConst):
    @staticmethod
    def generate_task_id(task_name, gh_id):
        match task_name:
            case "follow":
                return f"{task_name}_{gh_id}"
            case _:
                return f"{task_name}_{uuid.uuid4()}"

    @staticmethod
    def extract_reply_cards(msg_type, root):
        page_path = root.find(".//pagepath").text
        card_title = root.find(".//title").text
        mini_program = root.find(".//sourcedisplayname").text
        file_id = root.find("appmsg/appattach/cdnthumburl").text
        ase_key = root.find("appmsg/appattach/aeskey").text
        file_size = root.find("appmsg/appattach/cdnthumblength").text
        return {
            "title": card_title,
            "page_path": page_path,
            "msg_type": msg_type,
            "mini_program": mini_program,
            "file_id": file_id,
            "file_size": file_size,
            "ase_key": ase_key,
        }

    @staticmethod
    def extract_reply_articles(msg_type, root):
        title = root.find("appmsg/title").text
        url = root.find("appmsg/url").text
        cover_url = root.find("appmsg/thumburl").text
        account_name = root.find("appmsg/sourcedisplayname").text
        gh_id = root.find("appmsg/sourceusername").text
        desc = root.find("appmsg/des").text
        return {
            "msg_type": msg_type,
            "title": title,
            "url": url,
            "cover_url": cover_url,
            "account_name": account_name,
            "gh_id": gh_id,
            "desc": desc,
        }

    # 解析 xml
    @staticmethod
    def extract_callback_xml(self, xml_text):
        try:
            root = ET.fromstring(xml_text)
            msg_type = root.find("appmsg/type").text
            match msg_type:
                case "5":
                    return self.extract_reply_articles(msg_type, root)

                case "33":
                    return self.extract_reply_cards(msg_type, root)

                case "36":
                    return self.extract_reply_cards(msg_type, root)

                case _:
                    return {
                        "msg_type": msg_type,
                    }

        except Exception as e:
            print(e)
            return {}

    # 解析 page_path
    @staticmethod
    def extract_page_path(page_path):
        pass

    @staticmethod
    async def get_cover_url(aes_key, total_size, file_id):
        url = "http://api.geweapi.com/gewe/v2/api/message/downloadCdn"
        data = {
            "appId": "wx_anFlUnezoUynU3SKcqTWk",
            "aesKey": aes_key,
            "totalSize": total_size,
            "fileId": file_id,
            "type": "3",
            "suffix": "jpg",
        }
        headers = {
            "X-GEWE-TOKEN": "d3fb918f-0f36-4769-b095-410181614231",
            "Content-Type": "application/json",
        }
        async with AsyncHttpClient() as client:
            response = await client.post(url, headers=headers, data=json.dumps(data))

        return response

    @staticmethod
    async def get_sample_url(recent_articles):
        for article in recent_articles:
            link = article["ContentUrl"]
            print(link)
            response = await get_article_detail(article_link=link)
            print(response)
            if not response:
                continue
            code = response["code"]
            if code == 0 or code == 25006:
                return link

        return None

    # 获取检测的账号 list
    @staticmethod
    def get_monitor_account_list():
        yesterday = (datetime.today() - timedelta(days=1)).strftime("%Y%m%d")
        query = f"""
            SELECT  公众号名, ghid, count(DISTINCT mid) AS uv
            FROM    loghubods.opengid_base_data
            WHERE   dt = {yesterday}
            AND     hotsencetype = 1074
            AND     usersharedepth = 0
            AND     channel = '公众号合作-即转-稳定'
            GROUP BY 公众号名, ghid
            HAVING uv > 100
            ORDER BY uv DESC
            LIMIT 10
            ;
        """
        result = fetch_from_odps(query)
        return result

    # 下载封面图片
    async def download_cover(self, url, file_path):
        pass

    # 上传封面至 oss
    async def upload_cover(self, file_path):
        pass


class AutoReplyCardsMonitorMapper(AutoReplyCardsMonitorUtils):
    def __init__(self, pool, log_client):
        self.pool = pool
        self.log_client = log_client

    # 获取自动回复任务结果
    async def get_auto_reply_task_result(self, task_id):
        query = """
            SELECT task_result, task_status, err_msg, update_timestamp
            FROM gzh_msg_record
            WHERE task_id = %s;
        """
        return await self.pool.async_fetch(
            query=query, params=(task_id,), db_name="aigc"
        )

    # 查询账号
    async def fetch_account_status(self, account_name):
        query = """
            SELECT partner_name, partner_id, gh_id, status, follow_status
            FROM cooperate_accounts
            WHERE account_name = %s;
        """
        return await self.pool.async_fetch(query=query, params=(account_name,))

    # 更新账号状态为无效
    async def set_account_as_invalid(self, gh_id):
        query = """
            UPDATE cooperate_accounts SET status = %s WHERE gh_id = %s;
        """
        await self.pool.async_save(query=query, params=(self.INVALID_STATUS, gh_id))

    # 插入AIGC关注公众号任务
    async def insert_aigc_follow_account_task(self, task_id, link):
        timestamp = int(time.time() * 1000)
        query = """
            INSERT INTO gzh_msg_record (task_id, biz_type, task_params, create_timestamp, update_timestamp) VALUES (%s, %s, %s, %s, %s); 
        """
        return await self.pool.async_save(
            query=query,
            params=(task_id, "follow", link, timestamp, timestamp),
            db_name="aigc",
        )

    # 插入AIGC自动回复任务
    async def insert_aigc_auto_reply_task(self, task_id, account_name):
        timestamp = int(time.time() * 1000)
        query = """
            INSERT INTO gzh_msg_record (task_id, task_params, create_timestamp, update_timestamp) VALUES (%s, %s, %s, %s); 
        """
        return await self.pool.async_save(
            query=query,
            params=(task_id, account_name, timestamp, timestamp),
            db_name="aigc",
        )

    # 为账号设置 sample_url
    async def set_sample_url(self, gh_id, sample_url):
        query = """
            UPDATE cooperate_accounts SET sample_link = %s WHERE gh_id = %s;
        """
        return await self.pool.async_save(query=query, params=(sample_url, gh_id))

    # 修改账号的关注状态
    async def update_follow_status(self, gh_id, ori_status, new_status):
        query = """
            UPDATE cooperate_accounts SET follow_status = %s WHERE gh_id = %s and follow_status = %s;
        """
        return await self.pool.async_save(
            query=query, params=(new_status, gh_id, ori_status)
        )

    # 从 aigc 获取关注结果
    async def fetch_follow_account_status(self, gh_id):
        query = """
            SELECT task_status, err_msg 
            FROM gzh_msg_record
            WHERE task_id = %s;
        """
        return await self.pool.async_fetch(
            query=query, params=(f"follow_{gh_id}",), db_name="aigc"
        )

    # 创建自动回复任务
    async def create_auto_reply_task(self, task_id, gh_id):
        query = """
            INSERT INTO cooperate_accounts_task (task_id, gh_id) VALUES (%s, %s);
        """
        return await self.pool.async_save(query=query, params=(task_id, gh_id))

    async def update_auto_reply_task_status(
        self, task_id, status_type, ori_status, new_status
    ):
        task_query = """
            UPDATE cooperate_accounts_task SET task_status = %s WHERE task_id = %s AND task_status = %s;
        """
        extract_query = """
            UPDATE cooperate_accounts_task SET extract_status = %s WHERE task_id = %s AND extract_status = %s;
        """
        match status_type:
            case "task":
                return await self.pool.async_save(
                    query=task_query, params=(new_status, task_id, ori_status)
                )
            case "extract":
                return await self.pool.async_save(
                    query=extract_query, params=(new_status, task_id, ori_status)
                )
            case _:
                print("status_type_error")
                return None

    # 获取正在自动回复卡片的任务 id
    async def fetch_auto_replying_tasks(self):
        query = """
            SELECT task_id FROM cooperate_accounts_task WHERE task_status = %s;
        """
        return await self.pool.async_fetch(
            query=query, params=(self.PROCESSING_STATUS,)
        )

    # 设置自动回复结果
    async def set_auto_reply_result(self, task_id, finish_timestamp, result):
        query = """
            UPDATE cooperate_accounts_task 
            SET finish_timestamp = %s, result = %s, task_status = %s
            WHERE task_id = %s and task_status = %s;
        """
        return await self.pool.async_save(
            query=query,
            params=(
                finish_timestamp,
                result,
                self.SUCCESS_STATUS,
                task_id,
                self.PROCESSING_STATUS,
            ),
        )


class AutoReplyCardsMonitor(AutoReplyCardsMonitorMapper):
    def __init__(self, pool, log_client):
        super().__init__(pool, log_client)

    # 创建单个关注公众号任务
    async def create_follow_single_account_task(self, gh_id):
        response = await get_article_list_from_account(account_id=gh_id)
        code = response.get("code")
        match code:
            case 0:
                recent_articles = response["data"]["data"][0]["AppMsg"]["DetailInfo"]
                article_url = await self.get_sample_url(recent_articles)
                print(article_url)
                if article_url:
                    await self.set_sample_url(gh_id, article_url)

                    task_id = self.generate_task_id(task_name="follow", gh_id=gh_id)
                    affected_rows = await self.insert_aigc_follow_account_task(
                        task_id, article_url
                    )

                    if affected_rows:
                        await self.update_follow_status(
                            gh_id, self.INIT_STATUS, self.PROCESSING_STATUS
                        )

            case 25013:
                await self.set_account_as_invalid(gh_id)

            case _:
                pass

    # 创建单个账号自动回复任务
    async def create_auto_reply_single_account_task(self, gh_id, account_name):
        task_id = self.generate_task_id(task_name="auto_reply", gh_id=gh_id)
        # 先插入 task, 再创建自动回复任务
        create_row = await self.create_auto_reply_task(task_id, gh_id)
        if create_row:
            affected_rows = await self.insert_aigc_auto_reply_task(
                task_id, account_name
            )
            if not affected_rows:
                print("发布任务至 AIGC 失败")
            else:
                await self.update_auto_reply_task_status(
                    task_id, "task", self.INIT_STATUS, self.PROCESSING_STATUS
                )
        else:
            print("创建任务至 DB 失败")

    async def follow_gzh_task(self):
        account_list = self.get_monitor_account_list()
        for account in account_list:
            try:
                fetch_response = await self.fetch_account_status(account.公众号名)
                if not fetch_response:
                    print("账号不存在", account)
                    # todo 没有 gh_id, 暂时无法存储账号
                    # affected_rows =await self.insert_accounts_task(account.公众号名, account.ghid)
                    # if affected_rows:
                    #     await self.create_follow_account_task(account.ghid)

                else:
                    account_detail = fetch_response[0]
                    status = account_detail["status"]
                    follow_status = account_detail["follow_status"]
                    if not status:
                        print("账号已经迁移或者封禁")
                        continue

                    match follow_status:
                        case self.INIT_STATUS:
                            await self.create_follow_single_account_task(
                                account_detail["gh_id"]
                            )

                        case self.PROCESSING_STATUS:
                            fetch_response = await self.fetch_follow_account_status(
                                account_detail["gh_id"]
                            )
                            if not fetch_response:
                                await self.update_follow_status(
                                    account_detail["gh_id"],
                                    self.PROCESSING_STATUS,
                                    self.INIT_STATUS,
                                )

                            task_status = fetch_response[0]["task_status"]
                            match task_status:
                                case self.FETCH_INIT_STATUS:
                                    continue
                                case self.FETCH_PROCESSING_STATUS:
                                    continue
                                case self.FETCH_SUCCESS_STATUS:
                                    await self.update_follow_status(
                                        account_detail["gh_id"],
                                        self.PROCESSING_STATUS,
                                        self.SUCCESS_STATUS,
                                    )
                                case self.FETCH_FAIL_STATUS:
                                    await self.update_follow_status(
                                        account_detail["gh_id"],
                                        self.PROCESSING_STATUS,
                                        self.FAIL_STATUS,
                                    )

                        case self.SUCCESS_STATUS:
                            # 账号已经关注，创建获取自动回复任务
                            await self.create_auto_reply_single_account_task(
                                account_detail["gh_id"], account.公众号名
                            )

                        case _:
                            print(f"{account.公众号名}账号状态异常")

            except Exception as e:
                print(f"处理账号{account.公众号名}异常", e)

    # 异步获取关注结果
    async def get_auto_reply_response(self):
        task_list = await self.fetch_auto_replying_tasks()
        if not task_list:
            return

        for task in task_list:
            try:
                task_id = task["task_id"]
                response = await self.get_auto_reply_task_result(task_id)
                if not response:
                    continue

                task_status = response[0]["task_status"]
                task_result = response[0]["task_result"]
                update_timestamp = response[0]["update_timestamp"]
                match task_status:
                    case self.FETCH_FAIL_STATUS:
                        await self.update_auto_reply_task_status(
                            task_id, "task", self.PROCESSING_STATUS, self.FAIL_STATUS
                        )

                    case self.FETCH_SUCCESS_STATUS:
                        await self.set_auto_reply_result(
                            task_id, update_timestamp, task_result
                        )

                    case _:
                        continue

            except Exception as e:
                print(e)

    # 解析 xml 并且更新数据
    async def extract_task(self):
        pass

    # main function
    async def deal(self, task_name):
        match task_name:
            case "follow_gzh_task":
                await self.follow_gzh_task()

            case "get_auto_reply_task":
                await self.get_auto_reply_response()

            case "extract_task":
                await self.extract_task()

            case _:
                print("task_error")