title_with_video/tasks/new_task.py

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

from static.config import db_article, db_video
from applications.functions.log import logging
from static.config import mysql_coroutines
from applications.functions.common import async_post


async def publishToPQ(video_obj):
    """
    publish video to pq
    :return:
    """
    oss_path = video_obj['videoPath']
    uid = video_obj['uid']
    title = video_obj['title']
    cover = video_obj['coverPath']
    url = "https://vlogapi.piaoquantv.com/longvideoapi/crawler/video/send"
    headers = {
        "User-Agent": "PQSpeed/486 CFNetwork/1410.1 Darwin/22.6.0",
        "cookie": "JSESSIONID=4DEA2B5173BB9A9E82DB772C0ACDBC9F; JSESSIONID=D02C334150025222A0B824A98B539B78",
        "referer": "http://appspeed.piaoquantv.com",
        "token": "524a8bc871dbb0f4d4717895083172ab37c02d2f",
        "accept-language": "zh-CN,zh-Hans;q=0.9",
        "Content-Type": "application/x-www-form-urlencoded",
    }
    payload = {
        "coverImgPath": cover,
        "deviceToken": "9ef064f2f7869b3fd67d6141f8a899175dddc91240971172f1f2a662ef891408",
        "fileExtensions": "MP4",
        "loginUid": uid,
        "networkType": "Wi-Fi",
        "platform": "iOS",
        "requestId": "fb972cbd4f390afcfd3da1869cd7d001",
        "sessionId": "362290597725ce1fa870d7be4f46dcc2",
        "subSessionId": "362290597725ce1fa870d7be4f46dcc2",
        "title": title,
        "token": "524a8bc871dbb0f4d4717895083172ab37c02d2f",
        "uid": uid,
        "versionCode": "486",
        "versionName": "3.4.12",
        "videoFromScene": "1",
        "videoPath": oss_path,
        "viewStatus": "1",
    }
    response = await async_post(url, headers, payload)
    return response


async def getPQVideoDetail(video_id):
    """
    获取票圈视频详情信息
    :return:
    """
    url = "https://longvideoapi.piaoquantv.com/longvideoapi/openapi/video/batchSelectVideoInfo"
    data = {
        "videoIdList": [video_id]
    }
    header = {
        "Content-Type": "application/json",
    }
    response = await async_post(url, header, json.dumps(data))
    return response


async def getNewVideoIds(video_obj_list):
    """
    video
    :return:
    """
    vid_list = []
    for video_obj in video_obj_list:
        # video_obj 里面的信息对于历史数据可能不全，需要从pq获取
        try:
            if len(vid_list) >= 3:
                return vid_list
            else:
                pq_response = await publishToPQ(video_obj)
                video_id = pq_response['data']['id']
                vid_list.append(video_id)
        except:
            continue
    return vid_list


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

    def __init__(self, mysql_client):
        """
        :param mysql_client:
        """
        self.mysql_client = mysql_client

    async def getTaskList(self):
        """
        获取任务
        :return:
        """
        select_sql1 = f"""
            SELECT trace_id, content_id, gh_id, article_title, article_text, content_status, process_times
            FROM {db_article} 
            WHERE content_status = 0 and process_times <= 3
            ORDER BY request_time_stamp
            ASC
            LIMIT {mysql_coroutines};
        """
        tasks = await self.mysql_client.async_select(sql=select_sql1)
        task_obj_list = [
            {
                "trace_id": item[0],
                "content_id": item[1],
                "gh_id": item[2],
                "title": item[3],
                "text": item[4],
                "content_status": item[5],
                "process_times": item[6]
            } for item in tasks
        ]
        logging(
            code="9001",
            info="本次任务获取到 {} 条视频".format(len(task_obj_list)),
            data=task_obj_list
        )
        return task_obj_list

    async def getHistoryVideoOssPath(self, content_id):
        """
        check whether the contents videos exists
        :param content_id:
        :return:
        """
        select_sql = f"""
            SELECT video_title, uid, video_path, cover_path
            FROM {db_video}
            where content_id = '{content_id}' and oss_status = 1 order by request_time DESC;
        """
        content_videos = await self.mysql_client.async_select(select_sql)
        video_list = [
            {
                "title": line[0],
                "uid": line[1],
                "videoPath": line[2],
                "coverPath": line[3]
            }
            for line in content_videos
        ]
        if len(video_list) >= 3:
            return video_list
        else:
            return None

    async def useExistOssPath(self, video_info_list, params):
        """
        使用已经存在的视频id
        :return:
        """
        trace_id = params['trace_id']
        content_id = params['content_id']
        select_sql = f"""
            SELECT kimi_title
            FROM {db_article}
            WHERE content_id = '{content_id}' and kimi_title is not null limit 1;
        """
        info = await self.mysql_client.async_select(sql=select_sql)
        kimi_title = info[0]
        video_id_list = await getNewVideoIds(video_info_list)
        vid1, vid2, vid3 = video_id_list[0], video_id_list[1], video_id_list[2]
        update_sql = f"""
            UPDATE {db_article}
            SET 
                kimi_title=%s,
                recall_video_id1=%s, 
                recall_video_id2=%s, 
                recall_video_id3=%s,
                content_status=%s,
                process_times = %s
            WHERE  trace_id = %s
        """

        await self.mysql_client.async_insert(
            sql=update_sql,
            params=(
                kimi_title,
                vid1,
                vid2,
                vid3,
                2,
                int(params['process_times']) + 1,
                trace_id
            )
        )
        logging(
            code="9002",
            info="已从历史文章更新，文章id： {}".format(content_id),
            trace_id=trace_id
        )

    async def processTask(self, params):
        """
        异步执行
        :param params:
        :return:
        """
        content_id = params['content_id']
        trace_id = params['trace_id']
        # 判断该篇文章是否存在未下架的视频，且判断是否有3条， 如果没有三条，则启动新抓取任务，后续优化点
        oss_path_list = await self.getHistoryVideoOssPath(content_id=content_id)
        if oss_path_list:
            # 说明已经存在了结果, 将该条记录下的video_oss拿出来
            logging(
                code="9001",
                info="存在历史文章",
                trace_id=trace_id
            )
            await self.useExistOssPath(video_info_list=oss_path_list, params=params)
        else:
            pass

    async def deal(self):
        """
        处理
        :return:
        """
        task_list = await self.getTaskList()
        task_dict = {}
        for task in task_list:
            key = task['content_id']
            task_dict[key] = task
        process_list = []
        for item in task_dict:
            process_list.append(task_dict[item])
        if process_list:
            tasks = [self.processTask(params) for params in process_list]
            await asyncio.gather(*tasks)
        else:
            logging(
                code="9008",
                info="没有要处理的请求"
            )