import random import re import time import json import cffi import requests from urllib.parse import urlencode, urlparse from common import 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: 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: 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(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): return list return list except Exception as e: 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 url_video "视频号加密视频解密" @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: 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 new_video return new_video """ 视频号视频下载 """ @classmethod def sph_download_video(cls, video_url, video_path_url, video_id, video): new_video = video_path_url + str(video_id) + '.mp4' 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 new_video video_url = cls.decrypt_video(data=data, decode_key=decode_key, enc_length=enc_length) 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 new_video """票圈/快手创作者""" @classmethod def download_video(cls, video_url, video_path_url, video_id): video = video_path_url + str(video_id) + '.mp4' try: for i in range(3): payload = {} headers = {} response = requests.request("GET", video_url, headers=headers, data=payload) if response.status_code == 200: # 以二进制写入模式打开文件 with open(f"{video}", "wb") as file: # 将响应内容写入文件 file.write(response.content) time.sleep(5) return video return None except Exception: return video """抖音""" @classmethod def download_dy_video(cls, video_url, video_path_url, video_id): video = video_path_url + str(video_id) + '.mp4' try: 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', 'pragma': 'no-cache', 'range': 'bytes=0-', 'referer': f'https://www.douyin.com/video/{video_id}', '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, timeout= 60) 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 video """ 票圈站内视频下载 """ @classmethod def download_video_jpg(cls, video_url, video_path_url, video_id): video = video_path_url + str(video_id) + '.jpg' try: payload = {} headers = {} response = requests.request("GET", video_url, headers=headers, data=payload) if response.status_code == 200: # 以二进制写入模式打开文件 with open(f"{video}", "wb") as file: # 将响应内容写入文件 file.write(response.content) time.sleep(5) return video except Exception: return video 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)