add get_off_videos task

applications/api/__init__.py

@@ -1,6 +1,11 @@
-# feishu_sheet
+# feishu_api
 from .async_feishu_api import FeishuBotApi
 from .async_feishu_api import FeishuSheetApi
 
+# piaoquan api
+from .async_piaoquan_api import change_video_audit_status
+from .async_piaoquan_api import publish_video_to_piaoquan
+from .async_piaoquan_api import fetch_piaoquan_video_list_detail
+
 feishu_robot = FeishuBotApi()
 feishu_sheet = FeishuSheetApi()

applications/api/async_piaoquan_api.py

@@ -0,0 +1,78 @@
+from typing import Optional, Dict, List
+
+from applications.utils import AsyncHttPClient
+
+
+async def fetch_piaoquan_video_list_detail(video_list: List[int]) -> Optional[Dict]:
+    """async fetch piaoquan video detail"""
+    url = "https://longvideoapi.piaoquantv.com/longvideoapi/openapi/video/batchSelectVideoInfo"
+    data = {"videoIdList": video_list}
+    header = {
+        "Content-Type": "application/json",
+    }
+    async with AsyncHttPClient() as client:
+        response = await client.post(url, json=data, headers=header)
+
+    return response
+
+
+async def change_video_audit_status(video_id: int, status_code: int = 5) -> Dict:
+    url = "https://admin.piaoquantv.com/manager/video/audit/v2/updateAuditStatus"
+    payload = "videoId={}&auditStatus={}&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, status_code
+    )
+    headers = {
+        "accept": "application/json",
+        "accept-language": "en,zh;q=0.9,zh-CN;q=0.8",
+        "Content-Type": "application/json",
+        "cookie": "",
+        "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",
+    }
+    async with AsyncHttPClient() as client:
+        response = await client.post(url, data=payload, headers=headers)
+
+    return response
+
+
+async def publish_video_to_piaoquan(oss_path: str, uid: str, title: str) -> Dict:
+    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/json",
+    }
+    payload = {
+        "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",
+        "appType": 888888,
+        "repeatStatus": 1,
+    }
+    async with AsyncHttPClient() as client:
+        response = await client.post(url, data=payload, headers=headers)
+
+    return response

applications/tasks/monitor_tasks/__init__.py

@@ -1 +1,2 @@
 from .kimi_balance import check_kimi_balance
+from .get_off_videos import GetOffVideos

applications/tasks/monitor_tasks/get_off_videos.py

@@ -0,0 +1,91 @@
+import time
+import traceback
+
+from tqdm import tqdm
+
+from applications.api import change_video_audit_status
+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
+
+    # Task code
+    TASK_SUCCESS_STATUS = 2
+    TASK_FAILED_STATUS = 99
+
+
+class GetOffVideos(GetOffVideosConst):
+    def __init__(self, db_client, log_client):
+        self.db_client = db_client
+        self.log_client = log_client
+        self.table = "get_off_videos"
+
+    async def get_task_list(self, earliest_timestamp_threshold, expire_timestamp_threshold):
+        """get videos which need get off"""
+        earliest_timestamp_threshold = 0
+        query = f"""
+            select video_id from {self.table} where video_status = %s and publish_time between %s and %s;
+        """
+        video_list, error = await self.db_client.async_fetch(query, params=(self.VIDEO_AVAILABLE_STATUS, earliest_timestamp_threshold, expire_timestamp_threshold))
+        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)
+        for task in tqdm(task_list):
+            video_id = task['video_id']
+            try:
+                await self.update_video_audit_status(video_id)
+
+            except Exception as e:
+                self.log_client.log(
+                    contents={
+                        "task": "get_off_videos",
+                        "function": "get_off_job",
+                        "status": "fail",
+                        "message": "get off video fail",
+                        "data": {
+                            "video_id": video_id,
+                            "error": str(e),
+                            "traceback": traceback.format_exc()
+                        }
+                    }
+                )
+
+    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

applications/tasks/task_scheduler.py

@@ -4,6 +4,7 @@ from datetime import datetime
 from applications.api import feishu_robot
 from applications.utils import task_schedule_response
 from applications.tasks.monitor_tasks import check_kimi_balance
+from applications.tasks.monitor_tasks import GetOffVideos
 
 
 class TaskScheduler:
@@ -108,6 +109,15 @@ class TaskScheduler:
                     task_name=task_name, data=response
                 )
 
+            case "get_off_videos":
+                sub_task = GetOffVideos(self.db_client, self.log_client)
+                await sub_task.get_off_job()
+                response_code = await sub_task.check()
+                await self.release_task(task_name, date_string, response_code)
+                return await task_schedule_response.success_response(
+                    task_name=task_name, data={"code": response_code, "message": "get off_videos finished"}
+                )
+
             case _:
                 await self.log_client.log(
                     contents={