LongArticlesJob/getOffVideosDaily.py

"""
@author: luojunhui
@description: GetOffVideos Daily
"""
import time
import datetime
import traceback

from tqdm import tqdm

from applications import PQMySQL, PQAPI, log, bot

EXPIRE_TIME = 3 * 24 * 60 * 60


class AutoGetOffVideos(object):
    """
    自动下架视频
    """

    def __init__(self):
        self.db_client = None
        self.pq_api = None

    def base_init(self):
        """
        初始化数据库和票圈公共方法
        :return:
        """
        try:
            self.db_client = PQMySQL()
        except Exception as e:
            error_msg = traceback.format_exc()
            bot(
                title="自动下架视频任务，数据库初始化失败",
                detail={
                    "error": str(e),
                    "error_msg": error_msg
                }
            )
            return False

        try:
            self.pq_api = PQAPI()
        except Exception as e:
            error_msg = traceback.format_exc()
            bot(
                title="自动下架视频任务，pq公共方法连接失败",
                detail={
                    "error": str(e),
                    "error_msg": error_msg
                }
            )
            return False
        return True

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

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

    def change_video_id_status(self, video_id):
        """
        修改视频id状态
        :return:
        """
        response = self.pq_api.changeVideoStatus(video_id, 2)
        if response:
            self.update_video_id_status(video_id=video_id)
        else:
            log(
                task="getOffVideosDaily",
                function="change_video_id_status",
                status="fail",
                message="请求票圈修改状态异常: ---video_id = {}".format(video_id),
            )
            bot(
                title="get_off_videos 下架视频异常",
                detail={
                    "video_id": video_id
                },
                mention=False
            )

    def get_off_job(self):
        """
        已经请求超过3天的视频全部下架
        :return:
        """
        now_stamp = int(time.time())
        three_days_before = now_stamp - EXPIRE_TIME
        video_tuple = self.get_long_articles_videos(timestamp=three_days_before)
        vid_list = [i[0] for i in video_tuple]
        for video_id in tqdm(vid_list):
            try:
                self.change_video_id_status(video_id=video_id)
            except Exception as e:
                log(
                    task="getOffVideosDaily",
                    function="get_off_job",
                    status="fail",
                    message="get_off_job下架单个视频失败，video_id={}, 报错信息={}".format(video_id, e),
                )

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


def main():
    """
    main function
    :return:
    """
    auto_get_off_job = AutoGetOffVideos()
    if auto_get_off_job.base_init():
        auto_get_off_job.get_off_job()
        time.sleep(60)
        auto_get_off_job.check_job()
        bot(
            title="get_off_jobs任务执行完成通知",
            mention=False,
            detail={
                "finish_time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
        )


if __name__ == '__main__':
    main()