LongArticleTaskServer/app/domains/data_recycle_tasks/article_detail_stat.py

import time
import traceback
from datetime import datetime, timedelta

from app.infra.crawler.wechat import get_gzh_stat_daily
from app.infra.crawler.wechat import get_access_token

from app.infra.shared import run_tasks_with_asyncio_task_group


class ArticleDetailStatConst:
    # Task Status
    INIT_STATUS = 0
    PROCESSING_STATUS = 1
    SUCCESS_STATUS = 2
    FAILED_STATUS = 99

    # Account Status
    ACCOUNT_VALID_STATUS = 1
    ACCOUNT_INVALID_STATUS = 0

    # Cookie Status
    COOKIE_VALID_STATUS = 1
    COOKIE_INVALID_STATUS = 0

    # Gap Time
    GAP_DURATION = 300

    # 小数点保留位数
    DECIMAL_PRECISION = 7


class ArticleDetailStatMapper(ArticleDetailStatConst):
    def __init__(self, pool, log_service):
        self.pool = pool
        self.log_service = log_service

    # 获取账号信息
    async def fetch_monitor_accounts(self):
        query = """
            SELECT gh_id, account_name, app_id, app_secret 
            FROM gzh_account_info WHERE status = %s;
        """
        return await self.pool.async_fetch(
            query=query, params=(self.ACCOUNT_VALID_STATUS,)
        )

    # 更新 access_token
    async def set_access_token_for_each_account(
        self, gh_id, access_token, expire_timestamp
    ):
        query = """
            UPDATE gzh_cookie_info 
            SET access_token = %s, access_token_status = %s, expire_timestamp = %s
            WHERE gh_id = %s;
        """
        return await self.pool.async_save(
            query=query,
            params=(access_token, self.COOKIE_VALID_STATUS, expire_timestamp, gh_id),
        )

    # 从数据库获取 access_token
    async def get_access_token_from_database(self, gh_id):
        query = """
            SELECT access_token, expire_timestamp FROM gzh_cookie_info where gh_id = %s;
        """
        return await self.pool.async_fetch(query=query, params=(gh_id,))

    # 从official_articles_v2 获取文章详情
    async def fetch_article_detail(self, gh_id, msg_id):
        query = """
            SELECT wx_sn, title, ContentUrl, ItemIndex, title_md5
            FROM official_articles_v2
            WHERE ghId = %s AND appMsgId = %s;
        """
        return await self.pool.async_fetch(
            query=query, db_name="piaoquan_crawler", params=(gh_id, msg_id)
        )

    # 存储文章的各个表现
    async def save_article_performance(self, params: tuple):
        query = """
            INSERT INTO article_daily_performance 
            (
                account_name, gh_id, wx_sn, app_msg_id, msg_id, position, title, content_url, stat_date,
                read_user, share_user, read_subscribe_user, read_delivery_rate, read_finish_rate, 
                read_avg_activetime, zaikan_user, like_user, comment_count, collection_user
            ) VALUES (
                %s, %s, %s, %s, %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
            )
            ON DUPLICATE KEY UPDATE
                read_user = VALUES(read_user),
                share_user = VALUES(share_user),
                read_subscribe_user = VALUES(read_subscribe_user),
                read_delivery_rate = VALUES(read_delivery_rate),
                read_finish_rate = VALUES(read_finish_rate),
                read_avg_activetime = VALUES(read_avg_activetime),
                zaikan_user = VALUES(zaikan_user),
                like_user = VALUES(like_user),
                comment_count = VALUES(comment_count),
                collection_user = VALUES(collection_user),
                updated_at = CURRENT_TIMESTAMP;
        """
        return await self.pool.async_save(query=query, params=params)

    # 存储文章的阅读分布信息
    async def save_article_read_distribution(self, params: tuple):
        query = """
            INSERT INTO article_read_distribution 
            (
                wx_sn, jump_pos_1_rate, jump_pos_2_rate, jump_pos_3_rate, jump_pos_4_rate, jump_pos_5_rate,
                src_mp_message_user, src_mp_home_user, src_chat_user, src_moments_user, src_recommend_user,
                src_search_user, src_other_user, src_total_user
            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
                jump_pos_1_rate = VALUES(jump_pos_1_rate),
                jump_pos_2_rate = VALUES(jump_pos_2_rate),
                jump_pos_3_rate = VALUES(jump_pos_3_rate),
                jump_pos_4_rate = VALUES(jump_pos_4_rate),
                jump_pos_5_rate = VALUES(jump_pos_5_rate),
                src_mp_message_user = VALUES(src_mp_message_user),
                src_mp_home_user = VALUES(src_mp_home_user),
                src_chat_user = VALUES(src_chat_user),
                src_moments_user = VALUES(src_moments_user),
                src_recommend_user = VALUES(src_recommend_user),
                src_search_user = VALUES(src_search_user),
                src_other_user = VALUES(src_other_user),
                src_total_user = VALUES(src_total_user),
                updated_at = CURRENT_TIMESTAMP;
        """
        return await self.pool.async_save(query=query, params=params)


class ArticleDetailStat(ArticleDetailStatMapper):
    def __init__(self, pool, log_service):
        super().__init__(pool, log_service)

    # 存储账号信息
    async def save_account_details(self, account, fetch_response):
        article_list = fetch_response.get("list")
        if not article_list:
            return 0
        is_delay = fetch_response["is_delay"]
        if is_delay:
            print("数据延迟，重新处理")
            return 0

        async def _process_article_detail(_article_list):
            _msg_id_set = set()
            for _article in _article_list:
                _msg_id = _article["msgid"].split("_")[0]
                _msg_id_set.add(_msg_id)

            _article_mapper = {}
            for _msg_id in _msg_id_set:
                _publish_articles = await self.fetch_article_detail(
                    gh_id=account["gh_id"], msg_id=_msg_id
                )
                for _publish_article in _publish_articles:
                    _key = f"{_msg_id}_{_publish_article['ItemIndex']}"
                    _value = {
                        "wx_sn": _publish_article["wx_sn"].decode("utf-8"),
                        "title": _publish_article["title"],
                        "content_url": _publish_article["ContentUrl"],
                    }
                    _article_mapper[_key] = _value

            return _article_mapper

        account_published_articles = await _process_article_detail(article_list)

        async def _update_single_article(_article):
            # 文章基本信息
            app_msg_id = _article["msgid"]
            msg_id = app_msg_id.split("_")[0]
            position = app_msg_id.split("_")[1]
            wx_sn = account_published_articles.get(app_msg_id)["wx_sn"]

            # 文章表现
            article_performance = _article["detail_list"][0]
            await self.save_article_performance(
                params=(
                    account["account_name"],
                    account["gh_id"],
                    wx_sn,
                    app_msg_id,
                    msg_id,
                    position,
                    account_published_articles.get(app_msg_id)["title"],
                    account_published_articles.get(app_msg_id)["content_url"],
                    article_performance["stat_date"],
                    article_performance["read_user"],
                    article_performance["share_user"],
                    article_performance["read_subscribe_user"],
                    round(
                        article_performance["read_delivery_rate"],
                        self.DECIMAL_PRECISION,
                    ),
                    round(
                        article_performance["read_finish_rate"], self.DECIMAL_PRECISION
                    ),
                    round(
                        article_performance["read_avg_activetime"],
                        self.DECIMAL_PRECISION,
                    ),
                    article_performance["zaikan_user"],
                    article_performance["like_user"],
                    article_performance["comment_count"],
                    article_performance["collection_user"],
                )
            )

            # 表现分布信息
            jump_rate_map = {
                i["position"]: round(i["rate"], self.DECIMAL_PRECISION)
                for i in article_performance["read_jump_position"]
            }
            source_map = {
                i["scene_desc"]: i["user_count"]
                for i in article_performance["read_user_source"]
            }
            await self.save_article_read_distribution(
                params=(
                    wx_sn,
                    jump_rate_map[1],
                    jump_rate_map[2],
                    jump_rate_map[3],
                    jump_rate_map[4],
                    jump_rate_map[5],
                    source_map["公众号消息"],
                    source_map["公众号主页"],
                    source_map["聊天会话"],
                    source_map["朋友圈"],
                    source_map["推荐"],
                    source_map["搜一搜"],
                    source_map["其他"],
                    source_map["全部"],
                )
            )

        return await run_tasks_with_asyncio_task_group(
            task_list=article_list,
            handler=_update_single_article,
            description="批量更新文章阅读信息",
            unit="article",
        )

    # 处理单个账号
    async def process_single_account(self, account: dict):
        gh_id = account["gh_id"]
        token_info = await self.get_access_token_from_database(gh_id)
        if not token_info:
            return

        # 更新 token
        async def update_token(_new_token_info):
            _access_token = _new_token_info["access_token"]
            _expires_in = _new_token_info["expires_in"]

            await self.set_access_token_for_each_account(
                gh_id=account["gh_id"],
                access_token=_access_token,
                expire_timestamp=_expires_in + int(time.time()) - self.GAP_DURATION,
            )
            print(f"{account['account_name']} access_token updated to database")

        expire_timestamp = token_info[0]["expire_timestamp"] or 0
        if int(time.time()) >= expire_timestamp:
            print(f"{account['account_name']} access_token expired")
            new_token_info = await get_access_token(
                account["app_id"], account["app_secret"]
            )
            access_token = new_token_info["access_token"]
            await update_token(_new_token_info=new_token_info)

        else:
            access_token = token_info[0]["access_token"]

        # yesterday_string = datetime.strftime(datetime.now() - timedelta(days=5), "%Y-%m-%d")
        dt_list = [
            (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
            for i in range(1, 2)
        ]
        for dt in dt_list:
            print(f"{account['account_name']} crawl {dt} read_data")
            fetch_response = await get_gzh_stat_daily(
                access_token=access_token, date_string=dt
            )

            # 请求失败的情况
            if fetch_response.get("errcode") == 40001:
                fetch_new_token_info = await get_access_token(
                    account["app_id"], account["app_secret"]
                )
                await update_token(_new_token_info=fetch_new_token_info)
                break

            # 处理并且落表
            await self.save_account_details(account, fetch_response)

    # 入口函数
    async def deal(self):
        accounts = await self.fetch_monitor_accounts()
        if not accounts:
            return

        for account in accounts:
            try:
                await self.process_single_account(account)
                await self.log_service.log(
                    contents={
                        "task": "article_detail_stat",
                        "account_name": account["account_name"],
                        "status": "success",
                    }
                )
            except Exception as e:
                await self.log_service.log(
                    contents={
                        "task": "article_detail_stat",
                        "account_name": account["account_name"],
                        "error": str(e),
                        "traceback": traceback.format_exc(),
                        "status": "fail",
                    }
                )