guanggao_video/common/piaoquan_utils.py

import random
import time
import json
import requests
from urllib.parse import urlencode

from common import Common
from common.sql_help import sqlCollect


class PQ:

    """
    获取视频链接
    """
    @classmethod
    def get_pw_url(cls, user_id):
        url = f"https://admin.piaoquantv.com/manager/video/detail/{user_id}"
        payload = {}
        headers = {
            'authority': 'admin.piaoquantv.com',
            'accept': 'application/json, text/plain, */*',
            'accept-language': 'zh-CN,zh;q=0.9',
            'cache-control': 'no-cache',
            'cookie': 'SESSION=YjU3MzgwNTMtM2QyYi00YjljLWI3YWUtZTBjNWYwMGQzYWNl',
            'pragma': 'no-cache',
            'referer': f'https://admin.piaoquantv.com/cms/post-detail/{user_id}/detail',
            'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
            '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/120.0.0.0 Safari/537.36'
        }

        response = requests.request("GET", url, headers=headers, data=payload)
        data = response.json()
        try:
            video_url = data["content"]["transedVideoPath"]
            new_title = data["content"]["title"]
            return video_url, new_title
        except Exception as e:
            Common.logger("video").warning(f"获取视频链接失败:{e}\n")
            return ""

    """
     获取视频链接
    """
    @classmethod
    def get_audio_url(cls, task_mark, user_id, title, mark):
        url = f"https://admin.piaoquantv.com/manager/video/detail/{user_id}"
        payload = {}
        headers = {
            'authority': 'admin.piaoquantv.com',
            'accept': 'application/json, text/plain, */*',
            'accept-language': 'zh-CN,zh;q=0.9',
            'cache-control': 'no-cache',
            'cookie': 'SESSION=YjU3MzgwNTMtM2QyYi00YjljLWI3YWUtZTBjNWYwMGQzYWNl',
            'pragma': 'no-cache',
            'referer': f'https://admin.piaoquantv.com/cms/post-detail/{user_id}/detail',
            'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
            '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/120.0.0.0 Safari/537.36'
        }

        response = requests.request("GET", url, headers=headers, data=payload)
        data = response.json()
        try:
            list = []
            video_id = data["content"]["id"]
            if mark:
                status = sqlCollect.is_used(task_mark, video_id, mark)
            else:
                status = True
            if status:
                if title == '' or title == None:
                    new_title = data["content"]["title"]
                else:
                    if '/' in title:
                        titles = [t for t in title.split('/') if t and t != "None"]
                    else:
                        titles = [title]
                    new_title = random.choice(titles)
                video_url = data["content"]["transedVideoPath"]
                cover = data["content"]["coverImgPath"]
                all_data = {"video_id": video_id, "title": new_title, "cover": cover, "video_url": video_url}
                list.append(all_data)
                return list
            return list
        except Exception as e:
            Common.logger("video").warning(f"获取视频链接失败:{e}\n")
            return ""

    """
    获取用户下的所有视频
    """
    @classmethod
    def get_user_url(cls, task_mark, user_id, number, title, mark):
        url = f"https://admin.piaoquantv.com/manager/video/page?uid={user_id}&pageNum=1&pageSize=100"

        payload = {}
        headers = {
            'accept': 'application/json, text/plain, */*',
            'cookie': 'SESSION=NjRmMGVjNTAtNzJiNi00ODE0LThjYzQtYmZiNTJhMDNiZTcz',
            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'
        }

        response = requests.request("GET", url, headers=headers, data=payload)
        data = response.json()
        try:
            content = data["content"]["objs"]
            list = []
            for url in content:
                video_id = url["id"]
                status = sqlCollect.is_used(task_mark, video_id, mark)
                if status:
                    if title == '' or title == None:
                        new_title = url["title"]
                    else:
                        if '/' in title:
                            titles = title.split('/')
                        else:
                            titles = [title]
                        new_title = random.choice(titles)
                    cover = url["coverImgPath"]
                    video_url = url["transedVideoPath"]
                    all_data = {"video_id": video_id, "title": new_title, "cover": cover, "video_url": video_url}
                    list.append(all_data)
                    if len(list) == int(number):
                        Common.logger("log").info(f"获取视频总数:{len(list)}\n")
                        return list
            Common.logger("log").info(f"获取视频总数:{len(list)}\n")
            return list
        except Exception as e:
            Common.logger("log").warning(f"获取音频视频链接失败:{e}\n")
            return ""

    """
    获取封面
    """
    @classmethod
    def get_cover(cls, uid):
        time.sleep(1)
        url = "https://admin.piaoquantv.com/manager/video/multiCover/listV2"

        payload = json.dumps({
            "videoId": uid,
            "range": "2h"
        })
        headers = {
            'accept': 'application/json',
            'accept-language': 'zh-CN,zh;q=0.9',
            'cache-control': 'no-cache',
            'content-type': 'application/json',
            'cookie': 'SESSION=YjU3MzgwNTMtM2QyYi00YjljLWI3YWUtZTBjNWYwMGQzYWNl',
            'origin': 'https://admin.piaoquantv.com',
            'pragma': 'no-cache',
            'priority': 'u=1, i',
            'sec-ch-ua': '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'
        }

        response = requests.request("POST", url, headers=headers, data=payload)
        data = response.json()
        content = data["content"]
        if len(content) == 1:
            return content[0]["coverUrl"]
        max_share_count = 0
        selected_cover_url = ""
        for item in content:
            share_count = item.get("shareWeight")
            if share_count is not None and share_count > max_share_count:
                max_share_count = share_count
                selected_cover_url = item["coverUrl"]
            elif share_count == max_share_count and item["createUser"] == "用户":
                selected_cover_url = item["coverUrl"]
        return selected_cover_url

    """
    获取标题
    """
    @classmethod
    def get_title(cls, uid):
        url = "https://admin.piaoquantv.com/manager/video/multiTitleV2/listV2"

        payload = json.dumps({
            "videoId": uid,
            "range": "4h"
        })
        headers = {
            'accept': 'application/json',
            'accept-language': 'zh-CN,zh;q=0.9',
            'cache-control': 'no-cache',
            'content-type': 'application/json',
            'cookie': 'SESSION=YjU3MzgwNTMtM2QyYi00YjljLWI3YWUtZTBjNWYwMGQzYWNl',
            'origin': 'https://admin.piaoquantv.com',
            'pragma': 'no-cache',
            'priority': 'u=1, i',
            'sec-ch-ua': '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'
        }
        response = requests.request("POST", url, headers=headers, data=payload)
        data = response.json()
        content = data["content"]
        if len(content) == 1:
            return content[0]["title"]
        max_share_count = 0
        selected_title = ""
        for item in content:
            share_count = item.get("shareWeight")
            if share_count is not None and share_count > max_share_count:
                max_share_count = share_count
                selected_title = item["title"]
            elif share_count == max_share_count and item["createUser"] == "用户":
                selected_title = item["title"]
        return selected_title

    """
    新生成视频上传到对应账号下
    """
    @classmethod
    def insert_piaoquantv(cls, new_video_path, new_title):

        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/x-www-form-urlencoded'
        }
        payload = {
            # 'coverImgPath': cover,
            'deviceToken': '9ef064f2f7869b3fd67d6141f8a899175dddc91240971172f1f2a662ef891408',
            'fileExtensions': 'MP4',
            'loginUid': 70277791,
            'networkType': 'Wi-Fi',
            'platform': 'iOS',
            'requestId': 'fb972cbd4f390afcfd3da1869cd7d001',
            'sessionId': '362290597725ce1fa870d7be4f46dcc2',
            'subSessionId': '362290597725ce1fa870d7be4f46dcc2',
            'title': new_title,
            'token': '524a8bc871dbb0f4d4717895083172ab37c02d2f',
            'uid': 70277791,
            'versionCode': '486',
            'versionName': '3.4.12',
            'videoFromScene': '1',
            'videoPath': new_video_path,
            'viewStatus': '1'
        }
        encoded_payload = urlencode(payload)
        response = requests.request("POST", url, headers=headers, data=encoded_payload)
        data = response.json()
        code = data["code"]
        if code == 0:
            new_video_id = data["data"]["id"]
            return new_video_id
        else:
            return ''


    """
    票圈站内视频下载
    """
    @classmethod
    def download_video(cls, video_url, video_path_url, video_id):
        for i in range(3):
            payload = {}
            headers = {}
            response = requests.request("GET", video_url, headers=headers, data=payload)
            if response.status_code == 200:
                # 以二进制写入模式打开文件
                video = video_path_url + str(video_id) + '.mp4'
                with open(f"{video}", "wb") as file:
                    # 将响应内容写入文件
                    file.write(response.content)
                time.sleep(5)
                return video
        return ''