LongArticleTaskServer/applications/tasks/data_recycle_tasks/recycle_mini_program_detail.py

import json
import traceback
from typing import Any
from datetime import datetime, timedelta


from applications.crawler.wechat import get_article_detail
from applications.utils import extract_root_source_id
from applications.utils import run_tasks_with_asyncio_task_group


class MiniProgramConst:
    ARTICLE_SUCCESS_CODE = 0
    # 记录默认状态
    DEFAULT_STATUS = 0
    # 请求接口失败状态
    REQUEST_FAIL_STATUS = -1
    # 文章被删除状态
    DELETE_STATUS = -2
    # 未知原因无信息返回状态
    UNKNOWN_STATUS = -3
    # 文章违规状态
    ILLEGAL_STATUS = -4

    TASK_NAME = "recycle_mini_program_detail_daily"
    ARTICLE_TABLE = "official_articles_v2"
    DETAIL_TABLE = "long_articles_detail_info"

    # 更新文章周期天数
    ARTICLE_RECYCLE_DAYS = 1
    # 更新root_source_id天数
    ROOT_SOURCE_ID_UPDATE_DAYS = 3

    # DEFAULT VIDEO ID
    DEFAULT_VIDEO_ID = 1
    ZERO_VIDEO_ID = 0


class RecycleMiniProgramDetailBase(MiniProgramConst):
    def __init__(self, pool, log_client, trace_id):
        self.pool = pool
        self.log_client = log_client
        self.trace_id = trace_id

    # 构造小程序信息对象
    def create_mini_info(self, response: list[dict]) -> list[dict]:
        return [
            {
                "app_id": "wx89e7eb06478361d7",
                "avatar": "https://rescdn.yishihui.com/0temp/logo.png",
                "image_url": "",
                "nike_name": "票圈 l 3亿人喜欢的视频平台",
                "root_source_id": item["root_source_id"],
                "video_id": item["video_id"]
                if item["video_id"]
                else self.DEFAULT_VIDEO_ID,
                "service_type": "0",
                "title": "",
                "type": "card",
            }
            for item in response
        ]

    # 获取单个 root_source_id 的首层，裂变详情
    async def fetch_root_source_id_result(self, root_source_id, dt) -> list[dict]:
        query = f"""
            SELECT first_uv, split0, split0_head, split0_recommend, split1, split1_head, split1_recommend, split2, split2_head, split2_recommend
            FROM changwen_data_rootsourceid
            WHERE root_source_id = %s AND dt = %s;
        """
        return await self.pool.async_fetch(query=query, params=(root_source_id, dt))

    # 获取制定天发布的文章
    async def fetch_published_articles(
        self, run_date: str, date_delta: int
    ) -> list[dict]:
        query = f"""
            SELECT ContentUrl, wx_sn, publish_timestamp, accountName, title, root_source_id_list
            FROM official_articles_v2
            WHERE FROM_UNIXTIME(publish_timestamp)
            BETWEEN DATE_SUB(%s, INTERVAL %s DAY) AND DATE_SUB(%s, INTERVAL 1 SECOND);
        """
        return await self.pool.async_fetch(
            query=query,
            db_name="piaoquan_crawler",
            params=(run_date, date_delta, run_date),
        )

    # 从数据库获取root_source_id信息
    async def fetch_root_source_id_from_db(
        self, root_source_id_list: list[str]
    ) -> list[dict[str, str]]:
        placeholders = ", ".join(["%s"] * len(root_source_id_list))
        query = f"""
            SELECT video_id, root_source_id FROM long_articles_root_source_id WHERE root_source_id in ({placeholders});
        """
        return await self.pool.async_fetch(
            query=query, params=tuple(root_source_id_list)
        )

    # 获取制定几天前的root_source_id
    async def fetch_root_source_id_in_last_days(
        self, run_date: str, date_delta: int
    ) -> list[dict[str, str]]:
        query = f"""
            SELECT recall_dt, root_source_id
            FROM {self.DETAIL_TABLE}
            WHERE publish_dt
            BETWEEN DATE_SUB(%s, INTERVAL %s DAY) AND DATE_SUB(%s, INTERVAL 1 SECOND);
        """
        return await self.pool.async_fetch(
            query=query,
            db_name="piaoquan_crawler",
            params=(run_date, date_delta, run_date),
        )

    # 插入新的root_source_id
    async def insert_each_root_source_id(self, params_list: list[tuple]) -> None:
        # 记录root_source_id信息
        query = f"""
            INSERT INTO {self.DETAIL_TABLE}
            (wx_sn, mini_title, mini_name, cover_url, video_index, root_source_id, video_id, publish_dt, recall_dt)
            values
            (%s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE video_id = VALUES(video_id);
        """
        return await self.pool.async_save(
            query=query, db_name="piaoquan_crawler", params=params_list, batch=True
        )

    # 更新root_source_id 首层 && 裂变信息
    async def update_each_root_source_id(self, item: dict) -> None:
        recall_dt = item["recall_dt"]
        root_source_id = item["root_source_id"]
        mini_program_detail = await self.fetch_root_source_id_result(
            root_source_id, recall_dt
        )
        if not mini_program_detail:
            return

        mini_program_detail = mini_program_detail[0]
        query = f"""
            UPDATE {self.DETAIL_TABLE}
            SET first_level = %s, 
                fission_0 = %s, fission_0_head = %s, fission_0_recommend = %s, 
                fission_1 = %s, fission_1_head = %s, fission_1_recommend = %s, 
                fission_2 = %s, fission_2_head = %s, fission_2_recommend = %s
            WHERE root_source_id = %s and recall_dt = %s;
        """
        await self.pool.async_save(
            query=query,
            db_name="piaoquan_crawler",
            params=(
                mini_program_detail["first_uv"],
                mini_program_detail["split0"],
                mini_program_detail["split0_head"],
                mini_program_detail["split0_recommend"],
                mini_program_detail["split1"],
                mini_program_detail["split1_head"],
                mini_program_detail["split1_recommend"],
                mini_program_detail["split2"],
                mini_program_detail["split2_head"],
                mini_program_detail["split2_recommend"],
                root_source_id,
                recall_dt,
            ),
        )


class RecycleMiniProgramDetailTask(RecycleMiniProgramDetailBase):
    # 业务层
    def __init__(self, pool, log_client, trace_id):
        super().__init__(pool, log_client, trace_id)

    async def handle_single_task(self, article):
        """
        record each article into long_articles_detail_info table
        """
        url, publish_timestamp, wx_sn = (
            article["ContentUrl"],
            article["publish_timestamp"],
            article["wx_sn"].decode(),
        )
        root_source_id_list = (
            json.loads(article["root_source_id_list"])
            if article["root_source_id_list"]
            else []
        )
        mini_detail = []

        # 获取小程序信息
        if root_source_id_list:
            root_source_id_detail = await self.fetch_root_source_id_from_db(
                root_source_id_list
            )
            mini_detail = self.create_mini_info(root_source_id_detail)
        else:
            # no root_source_id_list in article
            try:
                article_detail = await get_article_detail(article_link=url)
                response_code = article_detail["code"]
                if response_code == self.ARTICLE_SUCCESS_CODE:
                    mini_detail = article_detail["data"]["data"].get("mini_program", [])

            except Exception as e:
                await self.log_client.log(
                    contents={
                        "task": self.TASK_NAME,
                        "trace_id": article["trace_id"],
                        "message": "get article detail error",
                        "data": {
                            "error": str(e),
                            "traceback": traceback.format_exc(),
                        },
                    }
                )

        # 将root_source_id信息插入mysql表格
        if not mini_detail:
            return

        publish_date = datetime.fromtimestamp(publish_timestamp)
        publish_dt = publish_date.strftime("%Y-%m-%d")
        # 记录T+0, T+1, T+2 三天的数据
        recall_dt_list = [
            (publish_date + timedelta(days=i)).strftime("%Y-%m-%d") for i in range(3)
        ]
        params_list = []
        for dt in recall_dt_list:
            for video_index, mini_item in enumerate(mini_detail, start=1):
                image_url = mini_item["image_url"]
                nickname = mini_item["nike_name"]
                # 将video_id为0的替换为1
                if mini_item.get("video_id") == self.ZERO_VIDEO_ID:
                    mini_item["video_id"] = self.DEFAULT_VIDEO_ID

                if mini_item.get("root_source_id") and mini_item.get("video_id"):
                    root_source_id, video_id = (
                        mini_item["root_source_id"],
                        mini_item["video_id"],
                    )
                else:
                    id_info = extract_root_source_id(mini_item["path"])
                    root_source_id, video_id = (
                        id_info["root_source_id"],
                        id_info["video_id"],
                    )

                kimi_title = mini_item["title"]
                params_list.append(
                    (
                        wx_sn,
                        kimi_title,
                        nickname,
                        image_url,
                        video_index,
                        root_source_id,
                        video_id,
                        publish_dt,
                        dt,
                    )
                )

        await self.insert_each_root_source_id(params_list)

    # 记录每日发文 && root_source_id 信息
    async def record_published_articles_job(
        self,
        date_delta: int,
        run_date: str,
        *,
        max_concurrency: int = 20,  # 并发上限
        fail_fast: bool = False,  # 是否出现错误就退出
    ) -> dict[str, Any]:
        # get tasks
        task_list = await self.fetch_published_articles(run_date, date_delta)
        return await run_tasks_with_asyncio_task_group(
            task_list=task_list,
            handler=self.handle_single_task,
            max_concurrency=max_concurrency,
            fail_fast=fail_fast,
            description="持久化文章&&rootSourceId",
            unit="article",
        )

    # 更新root_source_id的首层，裂变信息
    async def update_mini_program_detail_job(
        self,
        date_delta: int,
        run_date: str,
        *,
        max_concurrency: int = 20,
        fail_fast: bool = False,
    ) -> dict[str, Any]:
        task_list = await self.fetch_root_source_id_in_last_days(run_date, date_delta)
        return await run_tasks_with_asyncio_task_group(
            task_list=task_list,
            handler=self.update_each_root_source_id,
            max_concurrency=max_concurrency,
            fail_fast=fail_fast,
            description="更新rootSourceId裂变表现",
            unit="id",
        )

    # 业务入口
    async def deal(self, params: dict):
        # 解析请求参数
        run_date = params.get("run_date")
        if not run_date:
            run_date = datetime.today().strftime("%Y-%m-%d")

        record_date_delta = params.get("record_date_delta", self.ARTICLE_RECYCLE_DAYS)
        update_date_delta = params.get(
            "update_date_delta", self.ROOT_SOURCE_ID_UPDATE_DAYS
        )

        # 业务执行
        await self.record_published_articles_job(
            date_delta=record_date_delta, run_date=run_date
        )
        await self.update_mini_program_detail_job(
            date_delta=update_date_delta, run_date=run_date
        )