ScheduledTasks/tasks/auto_getoff_videos.py

"""
@author: luojunhui
"""
import json
import os.path
import time

from concurrent.futures.thread import ThreadPoolExecutor

import requests
import pymysql

from applications.aliyunLog import read_log


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

    @classmethod
    def get_long_articles_video_set(cls, time_stamp):
        """
        获取长文视频list
        :return:
        """
        spider_connection = pymysql.connect(
            host="rm-bp1159bu17li9hi94.mysql.rds.aliyuncs.com",
            port=3306,
            user="crawler",
            passwd="crawler123456@",
            db="piaoquan-crawler",
            charset="utf8mb4"
        )
        select_sql = f"""
        SELECT video_id
        FROM article_match_videos
        WHERE video_status = 1 and request_time < %s;
        """
        cursor = spider_connection.cursor()
        cursor.execute(select_sql, time_stamp)
        data = cursor.fetchall()
        return data

    @classmethod
    def update_mysql_status(cls, video_id):
        """
        修改数据库内视频状态
        :param video_id:
        :return:
        """
        spider_connection = pymysql.connect(
            host="rm-bp1159bu17li9hi94.mysql.rds.aliyuncs.com",
            port=3306,
            user="crawler",
            passwd="crawler123456@",
            db="piaoquan-crawler",
            charset="utf8mb4"
        )
        select_sql = f"""
                UPDATE article_match_videos
                SET video_status = 0
                WHERE video_id = %s;
                """
        cursor = spider_connection.cursor()
        cursor.execute(select_sql, video_id)
        spider_connection.commit()

    @classmethod
    def check_video_by_id(cls, video_id):
        """
        通过视频id来判断首次播放时间和现在时间戳大于3天的视频
        :param video_id:
        :return:
        """
        end_time = int(time.time())
        start_time = end_time - 7 * 24 * 3600
        res = read_log(
            start_time=start_time, end_time=end_time, query='businessType : videoPlay and videoId: {}'.format(video_id)
        )
        if res:
            cls.change_status(video_id=video_id)
        else:
            print("No need to change")

    @classmethod
    def change_status(cls, video_id):
        """
        修改视频规则
        :return:
        """
        cls.update_mysql_status(video_id=video_id)
        path = "./static/{}.json".format(video_id)
        if os.path.exists(path):
            print("File already Exists")
        else:
            url = "https://admin.piaoquantv.com/manager/video/audit/v2/updateAuditStatus"
            payload = "videoId={}&auditStatus=2&updateReasonJson=&rejectReasonJson=%5B%7B%22reason%22%3A%22%E9%95%BF%E6%96%87%E8%87%AA%E5%8A%A8%E4%B8%8B%E6%9E%B6%22%2C%22reasonId%22%3A-1%7D%5D&adminUid=206".format(
                video_id)
            headers = {
                'accept': 'application/json',
                'accept-language': 'en,zh;q=0.9,zh-CN;q=0.8',
                'content-type': 'application/x-www-form-urlencoded;charset=UTF-8',
                'cookie': 'SESSION=NTljNTg2YjktMTU0MC00YWQ5LWE4ZTktNDFhODY0NzM3NTcx',
                'origin': 'https://admin.piaoquantv.com',
                'priority': 'u=1, i',
                'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
                'sec-ch-ua-mobile': '?0',
                'sec-ch-ua-platform': '"macOS"',
                'sec-fetch-dest': 'empty',
                'sec-fetch-mode': 'cors',
                'sec-fetch-site': 'same-origin',
                'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36'
            }
            response = requests.request(
                "POST",
                url,
                headers=headers,
                data=payload
            )
            # print(response.json())
            if response.json()['code'] == 0:
                with open(path, "w") as f:
                    f.write(json.dumps({"time": int(time.time())}))
            else:
                print("失败的请求")

    @classmethod
    def task2(cls):
        """
        main function process
        :return:
        """
        now_time_stamp = int(time.time())
        three_days_before = now_time_stamp - 4 * 24 * 3600
        vid_set = cls.get_long_articles_video_set(time_stamp=three_days_before)
        vid_list = [i[0] for i in vid_set]
        with ThreadPoolExecutor(max_workers=8) as Pool:
            Pool.map(cls.check_video_by_id, vid_list)

    @classmethod
    def task1(cls):
        """
        已经请求超过7天的视频全部下架
        :return:
        """
        now_stamp = int(time.time())
        seven_days_before = now_stamp - 8 * 24 * 3600
        video_set = cls.get_long_articles_video_set(time_stamp=seven_days_before)
        vid_list = [i[0] for i in video_set]
        with ThreadPoolExecutor(max_workers=8) as Pool1:
            Pool1.map(cls.change_status, vid_list)