video_rewriting/data_channel/piaoquan.py

import random
import re
import time
import json

import cffi
import requests
from urllib.parse import urlencode, urlparse

from common import Common, AliyunLogger, Feishu
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"]
            return video_url
        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, "rule": "无", "old_title": data["content"]["title"]}
                list.append(all_data)
                return list
            return list
        except Exception as e:
            Common.logger("video").warning(f"获取视频链接失败:{e}\n")
            return ""

    """
    获取用户下的所有视频
    """
    @classmethod
    def get_pq_url(cls, task_mark, user_id, number, mark, channel_id, name):
        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'
        }


        list = []
        try:
            response = requests.request("GET", url, headers=headers, data=payload)
            data = response.json()
            content = data["content"]["objs"]

            for url in content:
                video_id = url["id"]
                status = sqlCollect.is_used(task_mark, video_id, mark, "票圈")

                cover = url["coverImgPath"]
                video_url = url["transedVideoPath"]
                old_title = url["title"]
                log_data = f"user:{user_id},,video_id:{video_id},,video_url:{video_url},,original_title:{old_title}"
                AliyunLogger.logging(channel_id, name, user_id, video_id, "扫描到一条视频", "2001", log_data)

                if status:
                    AliyunLogger.logging(channel_id, name, user_id, video_id, "该视频已改造过", "2002", log_data)
                    continue
                AliyunLogger.logging(channel_id, name, user_id, video_id, "符合规则等待改造", "2004", log_data)
                all_data = {"video_id": video_id, "cover": cover, "video_url": video_url, "rule": "无", "old_title": old_title}
                list.append(all_data)
                if len(list) == int(number):
                    Common.logger("pq").info(f"获取视频总数:{len(list)}\n")
                    return list
            Common.logger("pq").info(f"获取票圈视频总数:{len(list)}\n")
            return list
        except Exception as e:
            Common.logger("pq").warning(f"获取音频视频链接失败:{e}\n")
            return list

    """
    获取封面
    """
    @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, cover, n_id):
        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 = {
            'deviceToken': '9ef064f2f7869b3fd67d6141f8a899175dddc91240971172f1f2a662ef891408',
            'fileExtensions': 'MP4',
            'loginUid': n_id,
            'networkType': 'Wi-Fi',
            'platform': 'iOS',
            'requestId': 'fb972cbd4f390afcfd3da1869cd7d001',
            'sessionId': '362290597725ce1fa870d7be4f46dcc2',
            'subSessionId': '362290597725ce1fa870d7be4f46dcc2',
            'title': new_title,
            'token': '524a8bc871dbb0f4d4717895083172ab37c02d2f',
            'uid': n_id,
            'versionCode': '486',
            'versionName': '3.4.12',
            'videoFromScene': '1',
            'videoPath': new_video_path,
            'viewStatus': '1'
        }
        if cover:
            payload['coverImgPath'] = cover
        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 get_dd_video_url(cls, wx_msg):
        try:
            url = "http://8.217.190.241:8888/api/wei_xin/msg/callback"

            # payload = json.loads(wx_msg)
            headers = {
                'Content-Type': 'application/json'
            }
            response = requests.request("POST", url, headers=headers, data=wx_msg.encode())
            response = response.json()
            url_video = response['video_url']
            return url_video
        except Exception as e:
            print(e)
            return None

    "视频号加密视频解密"
    @classmethod
    def decrypt_video(cls, data: bytes, decode_key: int, enc_length: int = 131072) -> bytes:
        try:
            ffi = cffi.FFI()
            # lib = ffi.dlopen(r'/Users/tzld/Desktop/video_rewriting/libsph_decrypt.dylib')
            lib = ffi.dlopen(r'/root/video_rewriting/libsph_decrypt.so')
            ffi.cdef('void decrypt(unsigned char *data, const size_t data_length, const uint32_t key);')

            c_data = ffi.new('unsigned char[]', list(data))
            lib.decrypt(c_data, enc_length, decode_key)
            data = bytes(ffi.buffer(c_data, len(data))[:])
            return data
        except Exception as e:
            Common.logger("dd-sph").info(f"decrypt_video获取异常，异常信息{e}")
            return data

    """
    单点视频号视频下载
    """
    @classmethod
    def dd_sph_download_video(cls, video_url, video_path_url, video_id, video, channel_id):
        new_video = video_path_url + str(video_id) + '.mp4'
        if channel_id == '单点视频':
            is_encrypted = video.get('is_encrypted', "0")
            if video['source'] == "视频号" and int(is_encrypted) == 1:
                decode_key = video['decode_key']
                if decode_key == None:
                    return new_video
                decode_key = int(video['decode_key'])
                data, enc_length = None, 0
                for i in range(3):
                    try:
                        response = requests.get(url=video_url, timeout=10)
                        data = response.content
                        data_length = int(response.headers.get("Content-Range", '0').split('/')[-1])
                        enc_length = int(response.headers.get('X-enclen', 131072))
                        if len(data) == data_length:
                            break
                    except TimeoutError:
                        continue
                if not data:
                    v_id = video["video_id"]
                    sqlCollect.update_shp_dd_vid_4(v_id)
                    from_user_name = video['from_user_name']  # 来源用户
                    from_group_name = video['from_group_name']  # 来源群组
                    source = video['source']  # 渠道
                    text = (
                        f"**渠道**: {source}\n"
                        f"**来源用户**: {from_user_name}\n"
                        f"**来源群组**: {from_group_name}\n"
                        f"**原视频链接**: {video['video_url']}\n"
                        f"**原视频封面**: {video['cover']}\n"
                        f"**原视频标题**: {video['old_title']}\n"
                    )
                    Feishu.finish_bot(text,
                                      "https://open.feishu.cn/open-apis/bot/v2/hook/493b3d4c-5fae-4a9d-980b-1dd86636524e",
                                      "【 视频下载失败 】")
                    return new_video
                video_url = cls.decrypt_video(data=data, decode_key=decode_key, enc_length=enc_length)
                with open(f"{new_video}", 'wb') as f:
                    f.write(video_url)
                    return new_video

        url_video = video_url
        for i in range(3):
            try:
                payload = {}
                headers = {}
                response = requests.request("GET", url_video, headers=headers, data=payload)
                if response.status_code == 200:
                    with open(f"{new_video}", "wb") as file:
                        # 将响应内容写入文件
                        file.write(response.content)
                    time.sleep(5)
                    return new_video
                else:
                    if channel_id == '单点视频':
                        wx_msg = video['wx_msg']
                        if wx_msg:
                            url_videos = cls.get_dd_video_url(wx_msg)
                            if url_videos:
                                url_video = url_videos
            except Exception:
                if channel_id == '单点视频':
                    wx_msg = video['wx_msg']
                    if wx_msg:
                        url_videos = cls.get_dd_video_url(wx_msg)
                        if url_videos:
                            url_video = url_videos
                if i == 3:
                    return None
        return None

    """
    视频号视频下载
    """
    @classmethod
    def sph_download_video(cls, video_url, video_path_url, video_id, video):
        decode_key = int(video['decode_key'])
        data, enc_length = None, 0
        for i in range(3):
            try:
                response = requests.get(url=video_url, timeout=10)
                data = response.content
                data_length = int(response.headers.get("Content-Range", '0').split('/')[-1])
                enc_length = int(response.headers.get('X-enclen', 131072))
                if len(data) == data_length:
                    break
            except TimeoutError:
                continue
        if not data:
            return None
        video_url = cls.decrypt_video(data=data, decode_key=decode_key, enc_length=enc_length)
        new_video = video_path_url + str(video_id) + '.mp4'
        for i in range(3):
            try:
                with open(f"{new_video}", 'wb') as f:
                    f.write(video_url)
                    return new_video
            except Exception:
                if i == 3:
                    return None

    """票圈/快手创作者"""
    @classmethod
    def download_video(cls, video_url, video_path_url, video_id):
        try:
            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 None
        except Exception:
            return None

    """抖音"""
    @classmethod
    def download_dy_video(cls, video_url, video_path_url, video_id):
        try:
            video = video_path_url + str(video_id) + '.mp4'
            for i in range(3):
                payload = {}
                headers = {
                    'accept': '*/*',
                    'accept-encoding': 'identity;q=1, *;q=0',
                    'accept-language': 'zh-CN,zh;q=0.9',
                    'cache-control': 'no-cache',
                    'connection': 'keep-alive',
                    'host': urlparse(video_url).netloc,
                    'pragma': 'no-cache',
                    'range': 'bytes=0-',
                    'referer': video_url,
                    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
                }
                response = requests.request("GET", video_url, headers=headers, data=payload)
                if response.status_code == 206:
                    # 以二进制写入模式打开文件
                    with open(f"{video}", "wb") as file:
                        # 将响应内容写入文件
                        file.write(response.content)
                    time.sleep(5)
                    return video
            return video
        except Exception as e:
            print(e)
            return None

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


if __name__ == '__main__':
    url ='https://v5-dy-o-abtest.zjcdn.com/24958f9e7ef091a31c409c0dd1e83a6f/675fc6bc/video/tos/cn/tos-cn-ve-15/oIB5AwLndiATiANQIziB1TfHED3igPr1AD5Kje/?a=1128&ch=11&cr=3&dr=0&lr=all&cd=0%7C0%7C0%7C3&cv=1&br=515&bt=515&cs=0&ds=6&ft=J33IIDDhNF5VQsYesNousauzXzySY~bbUBvThbLfK&mime_type=video_mp4&qs=0&rc=aTg2M2Y3ZjU1NGg3ZTU8aUBpamVoZ285cjh0djMzNGkzM0AxNjIzYjUuNV4xXmFiLTRiYSMwM2FlMmRjY3BgLS1kLTBzcw%3D%3D&btag=80010e000b0001&cc=46&cquery=105E_103Q_103W_103Y_100b&dy_q=1734326142&feature_id=f0150a16a324336cda5d6dd0b69ed299&l=20241216131541780499AB5185B80CA219&req_cdn_type='
    a = PQ.download_dy_video(url,'/Users/z/Downloads/','70100016')
    print(a)