LongArticleTaskServer/applications/tasks/monitor_tasks/get_off_videos.py

import time
import traceback
from typing import List, Optional

from tqdm import tqdm

from applications.api import AsyncApolloApi
from applications.api import change_video_audit_status
from applications.api import fetch_piaoquan_video_list_detail
from applications.api import feishu_robot


class GetOffVideosConst:
    EXPIRE_TIME = 3 * 24 * 3600
    EARLIEST_TIME = 7 * 24 * 3600
    VIDEO_AVAILABLE_STATUS = 1
    VIDEO_DISABLE_STATUS = 0

    # VIDEO_AUDIT
    VIDEO_AUDIT_FAIL_STATUS = 2
    VIDEO_AUDIT_SUCCESS_STATUS = 5

    # Task code
    TASK_SUCCESS_STATUS = 2
    TASK_FAILED_STATUS = 99

    # check status
    CHECK_INIT_STATUS = 0
    CHECK_FINISHED_STATUS = 1

    # table
    table = "get_off_videos"


class GetOffVideos(GetOffVideosConst, AsyncApolloApi):
    def __init__(self, db_client, log_client, trace_id):
        super().__init__()
        self.db_client = db_client
        self.log_client = log_client
        self.trace_id = trace_id

    async def get_no_need_get_off_videos() -> set:
        """get video_ids which don't need to get off"""
        # video_ids = await self.get_config_value(key="white_video_ids")
        video_ids = [61719952]
        return set(video_ids)

    async def get_task_list(
        self, earliest_timestamp_threshold: int, expire_timestamp_threshold: int
    ):
        """get videos which need get off"""
        query = f"""
            select video_id from {self.table} where video_status = %s and publish_time between %s and %s
            and video_id > 1;
        """
        video_list = await self.db_client.async_fetch(
            query,
            params=(
                self.VIDEO_AVAILABLE_STATUS,
                earliest_timestamp_threshold,
                expire_timestamp_threshold,
            ),
        )
        no_need_get_off_videos = await self.get_no_need_get_off_videos()
        video_list = [
            video for video in video_list if video["video_id"] not in no_need_get_off_videos
        ]
        return video_list

    async def update_video_status(self, video_id):
        query = f"""
            update {self.table} set video_status = %s, get_off_time = %s where video_id = %s;
        """
        return await self.db_client.async_save(
            query, params=(self.VIDEO_DISABLE_STATUS, int(time.time()), video_id)
        )

    async def update_video_audit_status(self, video_id):
        """use pq api to update video status"""
        response = await change_video_audit_status(
            video_id, self.VIDEO_AUDIT_FAIL_STATUS
        )
        await self.update_video_status(video_id)
        return response

    async def get_off_job(self):
        """get off videos out of expire time"""
        expire_timestamp_threshold = int(time.time()) - self.EXPIRE_TIME
        earliest_timestamp_threshold = int(time.time()) - self.EARLIEST_TIME
        task_list = await self.get_task_list(
            earliest_timestamp_threshold, expire_timestamp_threshold
        )
        await self.log_client.log(
            contents={
                "task": "get_off_videos",
                "trace_id": self.trace_id,
                "message": f"获取{len(task_list)}条待下架视频",
                "data": task_list,
            }
        )
        success_count = 0
        failed_count = 0
        for task in tqdm(task_list):
            video_id = task["video_id"]
            if video_id in (0, 1):
                continue
            try:
                await self.update_video_audit_status(video_id)
                success_count += 1
            except Exception as e:
                await self.log_client.log(
                    contents={
                        "task": "get_off_videos",
                        "function": "get_off_job",
                        "status": "fail",
                        "message": "get off video fail",
                        "trace_id": self.trace_id,
                        "data": {
                            "video_id": video_id,
                            "error": str(e),
                            "traceback": traceback.format_exc(),
                        },
                    }
                )
                failed_count += 1
        if success_count or failed_count:
            await feishu_robot.bot(
                title="自动下架任务完成",
                detail={
                    "成功下架视频数量": success_count,
                    "失败数量": failed_count,
                },
                mention=False,
            )

    async def check(self):
        earliest_timestamp = int(time.time()) - self.EARLIEST_TIME
        expire_timestamp = int(time.time()) - self.EXPIRE_TIME
        task_list = await self.get_task_list(earliest_timestamp, expire_timestamp)
        if task_list:
            await feishu_robot.bot(
                title="自动下架视频失败",
                detail={
                    "total_video": len(task_list),
                    "video_list": [i["video_id"] for i in task_list],
                },
                mention=False,
            )
            return self.TASK_FAILED_STATUS
        else:
            return self.TASK_SUCCESS_STATUS

    async def deal(self):
        await self.get_off_job()
        task_status = await self.check()
        await self.log_client.log(
            contents={
                "task": "get_off_videos",
                "function": "deal",
                "trace_id": self.trace_id,
                "message": "任务执行完成",
            }
        )
        return task_status


class CheckVideoAuditStatus(GetOffVideosConst):
    def __init__(self, db_client, log_client, trace_id):
        self.db_client = db_client
        self.log_client = log_client
        self.trace_id = trace_id

    async def get_video_list_status(self, video_list: List[int]):
        response = await fetch_piaoquan_video_list_detail(video_list)
        video_detail_list = response.get("data", [])
        if video_detail_list:
            bad_video_list = [
                i["id"]
                for i in video_detail_list
                if i["auditStatus"] != self.VIDEO_AUDIT_SUCCESS_STATUS
            ]
        else:
            bad_video_list = []

        return bad_video_list

    async def get_unchecked_video_list(self) -> Optional[List[int]]:
        """find unchecked videos"""
        query = f"""
            select video_id from {self.table} where check_status = %s and video_status = %s and video_id > 1 limit 1000;
        """
        video_id_list = await self.db_client.async_fetch(
            query, params=(self.CHECK_INIT_STATUS, self.VIDEO_AVAILABLE_STATUS)
        )
        if video_id_list:
            return [i["video_id"] for i in video_id_list]
        else:
            return None

    async def update_check_status(self, video_list: List[int]):
        query = f"""update {self.table} set check_status = %s where video_id in %s;"""
        return await self.db_client.async_save(
            query, params=(self.CHECK_FINISHED_STATUS, tuple(video_list))
        )

    async def deal(self):
        def chuck_iterator(arr, chunk_size):
            for i in range(0, len(arr), chunk_size):
                yield arr[i : i + chunk_size]

        video_id_list = await self.get_unchecked_video_list()
        if video_id_list:
            await self.log_client.log(
                contents={
                    "task": "check_video_audit_status",
                    "function": "deal",
                    "trace_id": self.trace_id,
                    "message": f"一共获取{len(video_id_list)}条视频",
                }
            )
        else:
            return self.TASK_SUCCESS_STATUS

        video_chunks = chuck_iterator(video_id_list, 10)

        bad_videos_count = 0
        fail_list = []
        batch = 0
        for video_chunk in video_chunks:
            batch += 1
            bad_video_id_list = await self.get_video_list_status(video_chunk)
            if bad_video_id_list:
                bad_videos_count += len(bad_video_id_list)
                await self.log_client.log(
                    contents={
                        "task": "check_video_audit_status",
                        "function": "deal",
                        "trace_id": self.trace_id,
                        "message": f"batch: {batch} has {len(bad_video_id_list)} bad videos",
                        "data": bad_video_id_list,
                    }
                )
                for bad_video_id in tqdm(bad_video_id_list):
                    response = await change_video_audit_status(bad_video_id)
                    if not response:
                        fail_list.append(bad_video_id)

            await self.update_check_status(video_chunk)
            await self.log_client.log(
                contents={
                    "task": "check_video_audit_status",
                    "function": "deal",
                    "trace_id": self.trace_id,
                    "message": f"finish process batch: {batch}",
                }
            )

        if fail_list:
            await feishu_robot.bot(
                title="校验已发布视频状态出现错误", detail=fail_list, mention=False
            )
            return self.TASK_FAILED_STATUS
        else:
            return self.TASK_SUCCESS_STATUS