title_with_video/tasks/history_task.py

"""
@author: luojunhui
"""
import json
import time
import asyncio

from applications.config import Config
from applications.log import logging
from applications.functions.pqFunctions import publish_to_pq
from applications.functions.common import shuffle_list


class historyContentIdTask(object):
    """
    处理已经匹配过小程序的文章
    """

    def __init__(self, mysql_client):
        """
        :param mysql_client:
        """
        self.mysql_client = mysql_client
        self.config = Config()
        self.article_match_video_table = self.config.article_match_video_table
        self.article_text_table = self.config.article_text_table
        self.article_crawler_video_table = self.config.article_crawler_video_table
        self.gh_id_dict = json.loads(self.config.get_config_value("testAccountLevel2"))
        self.history_coroutines = self.config.get_config_value("historyArticleCoroutines")

    async def get_tasks(self):
        """
        获取任务
        :return:
        """
        select_sql1 = f"""
            SELECT 
                ART.trace_id, 
                ART.content_id, 
                ART.flow_pool_level, 
                ART.gh_id,
                ART.process_times
            FROM {self.article_match_video_table} ART
            JOIN (
                select content_id, count(1) as cnt 
                from {self.article_crawler_video_table}
                where download_status = 2
                group by content_id
            ) VID on ART.content_id = VID.content_id and VID.cnt >= 3
            WHERE ART.content_status = 0 and ART.process_times <= 3
            ORDER BY request_timestamp
            LIMIT {self.history_coroutines};
        """
        tasks = await self.mysql_client.async_select(sql=select_sql1)
        task_obj_list = [
            {
                "trace_id": item[0],
                "content_id": item[1],
                "flow_pool_level": item[2],
                "gh_id": item[3],
                "process_times": item[4]
            } for item in tasks
        ]
        logging(
            code="9001",
            info="本次任务获取到 {} 条视频".format(len(task_obj_list)),
            data=task_obj_list
        )
        return task_obj_list

    async def get_video_list(self, content_id):
        """
        content_id
        :return:
        """
        sql = f"""
        SELECT platform, play_count, like_count, video_oss_path, cover_oss_path, user_id
        FROM {self.article_crawler_video_table}
        WHERE content_id = '{content_id}' and download_status = 2;
        """
        res_tuple = await self.mysql_client.async_select(sql)
        if len(res_tuple) >= 3:
            return [
                {
                    "platform": i[0],
                    "play_count": i[1],
                    "like_count": i[2],
                    "video_oss_path": i[3],
                    "cover_oss_path": i[4],
                    "uid": i[5]
                }
                for i in res_tuple]
        else:
            return []

    async def get_kimi_title(self, content_id):
        """
        获取 kimiTitle
        :param content_id:
        :return:
        """
        select_sql = f"""
        select kimi_title from {self.article_text_table} where content_id = '{content_id}';
        """
        res_tuple = await self.mysql_client.async_select(select_sql)
        if res_tuple:
            return res_tuple[0][0]
        else:
            return False

    async def publish_videos_to_pq(self, trace_id, flow_pool_level, kimi_title, gh_id, download_videos, process_times):
        """
        发布至 pq
        :param process_times:
        :param trace_id:
        :param download_videos: 已下载的视频---> list [{}, {}, {}.... ]
        :param gh_id: 公众号 id ---> str
        :param kimi_title: kimi 标题 ---> str
        :param flow_pool_level: 流量池层级 ---> str
        :return:
        """
        published_status = 4
        match flow_pool_level:
            case "autoArticlePoolLevel4":
                # 冷启层， 全量做
                video_list = shuffle_list(download_videos)[:3]
            case "autoArticlePoolLevel3":
                # 次条，只针对具体账号做
                if self.gh_id_dict.get(gh_id):
                    video_list = shuffle_list(download_videos)[:3]
                else:
                    video_list = download_videos[:3]
            case "autoArticlePoolLevel2":
                video_list = []
            case "autoArticlePoolLevel1":
                # 头条，先不做
                video_list = download_videos[:3]
            case _:
                print("未传流量池信息")
                video_list = download_videos[:3]
        L = []
        for video_obj in video_list:
            params = {
                "videoPath": video_obj['video_oss_path'],
                "uid": video_obj['uid'],
                "title": kimi_title
            }
            response = await publish_to_pq(params)
            time.sleep(2)
            obj = {
                "uid": video_obj['uid'],
                "source": video_obj['platform'],
                "kimiTitle": kimi_title,
                "videoId": response['data']['id'],
                "videoCover": response['data']['shareImgPath'],
                "videoPath": response['data']['videoPath'],
                "videoOss": video_obj['video_oss_path']
            }
            L.append(obj)
        update_sql = f"""
           UPDATE {self.article_match_video_table}
           SET content_status = %s, response = %s, process_times = %s
           WHERE trace_id = %s;
           """
        await self.mysql_client.async_insert(
            sql=update_sql,
            params=(published_status, json.dumps(L, ensure_ascii=False), process_times + 1, trace_id)
        )
        logging(
            code="9002",
            info="已经从历史文章更新",
            trace_id=trace_id
        )

    async def process_task(self, params):
        """
        异步执行
        :param params:
        :return:
        """
        content_id = params['content_id']
        trace_id = params['trace_id']
        flow_pool_level = params['flow_pool_level']
        gh_id = params['gh_id']
        process_times = params['process_times']
        download_videos = await self.get_video_list(content_id=content_id)
        if download_videos:
            # 把状态修改为 4
            """
            todo: 加上状态锁，防止多个进程同时处理一个视频， 通过 update_time && content_id来判断
            """
            # update_sql = f"""
            # UPDATE {self.article_crawler_video_table}
            # SET content_status = %s
            # WHERE trace_id = %s;
            # """
            # await self.mysql_client.asyncInsert(
            #     sql=update_sql,
            #     params=(4, trace_id)
            # )

            kimi_title = await self.get_kimi_title(content_id)
            await self.publish_videos_to_pq(
                flow_pool_level=flow_pool_level,
                kimi_title=kimi_title,
                gh_id=gh_id,
                trace_id=trace_id,
                download_videos=download_videos,
                process_times=process_times
            )
        else:
            pass

    async def deal(self):
        """
        处理
        :return:
        """
        task_list = await self.get_tasks()
        logging(
            code="5002",
            info="History content_task Task Got {} this time".format(len(task_list)),
            function="History Contents Task"
        )
        if task_list:
            tasks = [self.process_task(params) for params in task_list]
            await asyncio.gather(*tasks)
        else:
            print("暂时未获得历史已存在文章")