LongArticlesJob/applications/api/piaoquan_api.py

import json
import requests
from tenacity import retry
from requests.exceptions import RequestException
from typing import Optional, Dict, List

from applications.utils import request_retry

retry_desc = request_retry(retry_times=3, min_retry_delay=2, max_retry_delay=30)


@retry(**retry_desc)
def fetch_piaoquan_video_list_detail(video_id_list: List[int]) -> Optional[Dict]:
    """
    获取票圈视频详情信息
    :param: video_list: 视频id 列表
    :return: Detail
    """
    url = "https://longvideoapi.piaoquantv.com/longvideoapi/openapi/video/batchSelectVideoInfo"
    data = {"videoIdList": video_id_list}
    header = {
        "Content-Type": "application/json",
    }
    try:
        response = requests.post(url, headers=header, json=data, timeout=60)
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        print(f"API请求失败: {e}")
    except json.JSONDecodeError as e:
        print(f"响应解析失败: {e}")
    return None


@retry(**retry_desc)
def fetch_piaoquan_account_video_list(account_id: str, page_id: int, page_size: int) -> Optional[Dict]:
    """
    获取票圈账号视频信息
    :param: account_id: 账号id
    :return: account videos
    """
    import requests

    url = f"https://admin.piaoquantv.com/manager/video/page?uid={account_id}&pageNum={page_id}&pageSize={page_size}&categoryId=55&muid=7"
    headers = {
        'accept': 'application/json, text/plain, */*',
        'accept-language': 'zh,zh-CN;q=0.9',
        'priority': 'u=1, i',
        'sec-ch-ua': '"Chromium";v="136", "Google Chrome";v="136", "Not.A/Brand";v="99"',
        '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/136.0.0.0 Safari/537.36',
        'Cookie': 'SESSION=NGMyMzhkNzQtZmUzMS00ZTAwLTkzMTEtZTYwZThiN2JhNWE3'
    }
    try:
        response = requests.get(url, headers=headers, timeout=60)
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        print(f"API请求失败: {e}")
    except json.JSONDecodeError as e:
        print(f"响应解析失败: {e}")
    return None