LongArticlesJob/getOffVideosDaily.py

"""
@author: luojunhui
@description: GetOffVideos Daily
"""
import json
import time

import requests
import schedule
from tqdm import tqdm

from applications import PQMySQL, Functions, PQAPI, log
from applications.decoratorApi import retryOnTimeout


@retryOnTimeout()
def bot(account_list):
    """
    机器人
    """
    url = "https://open.feishu.cn/open-apis/bot/v2/hook/b44333f2-16c0-4cb1-af01-d135f8704410"
    headers = {"Content-Type": "application/json"}
    payload = {
        "msg_type": "interactive",
        "card": {
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "content": "存在视频下架失败<at id=all></at>\n",
                        "tag": "lark_md",
                    },
                },
                {
                    "tag": "div",
                    "text": {
                        "content": json.dumps(
                            account_list, ensure_ascii=False, indent=4
                        ),
                        "tag": "lark_md",
                    },
                },
            ],
            "header": {"title": {"content": "【重点关注】", "tag": "plain_text"}},
        },
    }
    requests.request("POST", url=url, headers=headers, data=json.dumps(payload), timeout=10)


class AutoGetOffVideos(object):
    """
    自动下架视频
    """
    pqMysql = PQMySQL()
    pqAPI = PQAPI()

    @classmethod
    def getLongArticlesVideos(cls, time_stamp):
        """
        获取待下架的视频
        :return:
        """
        select_sql = f"""
        SELECT video_id
        FROM get_off_videos
        WHERE video_status = 1 and publish_time < {time_stamp};
        """
        result = cls.pqMysql.select(sql=select_sql)
        log(
            task="getOffVideosDaily",
            function="getLongArticlesVideos",
            message="查找到视频 id_list，一共{}条视频".format(len(result))
        )
        return result

    @classmethod
    def updateVideoIdStatus(cls, video_id):
        """
        修改数据库内视频状态
        :param video_id:
        :return:
        """
        time_stamp = int(time.time())
        select_sql = f"""
                UPDATE get_off_videos
                SET video_status = 0, get_off_time = {time_stamp}
                WHERE video_id = %s;
                """
        try:
            cls.pqMysql.update(
                sql=select_sql,
                params=video_id
            )
            log(
                task="getOffVideosDaily",
                function="updateVideoIdStatus",
                message="成功修改视频状态",
                data={"video_id": video_id}
            )
        except Exception as e:
            log(
                task="getOffVideosDaily",
                function="updateVideoIdStatus",
                message="修改视频状态失败--- 推测 sql 问题，报错信息：{}".format(e),
                status="fail"
            )

    @classmethod
    def changeVideoIdStatus(cls, video_id):
        """
        修改视频id状态
        :return:
        """
        response = cls.pqAPI.changeVideoStatus(video_id, 2)
        if response:
            cls.updateVideoIdStatus(video_id=video_id)
        else:
            log(
                task="getOffVideosDaily",
                function="changeVideoIdStatus",
                status="fail",
                message="请求票圈修改状态异常: ---video_id = {}".format(video_id),
            )

    @classmethod
    def task1(cls):
        """
        已经请求超过3天的视频全部下架
        :return:
        """
        now_stamp = int(time.time())
        three_days_before = now_stamp - 3 * 24 * 60 * 60
        video_set = cls.getLongArticlesVideos(time_stamp=three_days_before)
        vid_list = [i[0] for i in video_set]
        for video_id in tqdm(vid_list):
            try:
                cls.changeVideoIdStatus(video_id=video_id)
            except Exception as e:
                log(
                    task="getOffVideosDaily",
                    function="task1",
                    status="fail",
                    message="task1下架单个视频失败，video_id={}, 报错信息={}".format(video_id, e),
                )

    @classmethod
    def task2(cls):
        """
        校验 3 天前发布的视频是否已经下架
        :return:
        """
        three_days_ago = int(time.time()) - 3 * 24 * 3600
        sql = f"""
        SELECT video_id
        FROM get_off_videos
        WHERE publish_time < {three_days_ago} and video_status = 1;
        """
        vid_tuple = cls.pqMysql.select(sql)
        if vid_tuple:
            vid_list = [i[0] for i in vid_tuple]
            for vid in vid_list:
                try:
                    cls.changeVideoIdStatus(video_id=vid)
                except Exception as e:
                    log(
                        task="getOffVideosDaily",
                        function="task2",
                        status="fail",
                        message="task2下架单个视频失败，video_id={}, 报错信息={}".format(vid, e),
                    )
            time.sleep(10)
            vid_tuple2 = cls.pqMysql.select(sql)
            if vid_tuple2:
                vid_list2 = [i[0] for i in vid_tuple2]
                bot(vid_list2)
            else:
                return
        else:
            return


def getOffJob():
    """
    下架任务
    :return:
    """
    AGV = AutoGetOffVideos()
    AGV.task1()


def checkJob():
    """
    校验 3 天前发布的视频是否已经被下架
    :return:
    """
    AGV = AutoGetOffVideos()
    AGV.task2()


if __name__ == '__main__':
    schedule.every().day.at("09:30").do(Functions().job_with_thread, getOffJob)

    schedule.every().day.at("10:30").do(Functions().job_with_thread, checkJob)

    schedule.every().day.at("14:30").do(Functions().job_with_thread, getOffJob)

    schedule.every().day.at("15:00").do(Functions().job_with_thread, checkJob)

    while True:
        schedule.run_pending()
        time.sleep(1)