import random import re import time import json import requests from urllib.parse import urlencode from common import Common, AliyunLogger 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' } 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, "票圈") 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, "该视频已改造过", "2001", 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 "" """ 获取封面 """ @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 download_video(cls, video_url, video_path_url, video_id, video, channel_id): url_video = video_url new_video = video_path_url + str(video_id) + '.mp4' 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 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= 'http://111.51.148.73/ksc1/55LDHcm2DuGHbBbzIDfqMh-vfVhaCuIXZG0szwmCZj9pXLmq7otQhUyGXtBtc13Rb0C49a4FzRU5xfUaRQb6De46IO6hQBVqzjOkFArRDfTC7u9BThi0Bsa2w6PP_6dg7WQYKxU6ZIbYI2tnLfOXU5Ira-DEZuGDQbgpZstADnGtvLMkOiB6DImlhfgdxWIZ.mp4?tag=1-1724839019-sp-0-lgl7lhkrnt-8937a5c247b0f313&provider=self&ocid=341&clientCacheKey=3x3br4x5z5hew32_bF.mp4&tt=bF&di=6f31a709&bp=10001' video = {"wx_msg": '{"TypeName":"AddMsg","Appid":"wx_GtAew_wmLbsHMbpDe7Hwk","Data":{"MsgId":1176686077,"FromUserName":{"string":"51757111911@chatroom"},"ToUserName":{"string":"wxid_mr9pdsanc99422"},"MsgType":1,"Content":{"string":"wxid_3379403802612:\n6.41 复制打开抖音，看看【Zooey的作品】被一只青蛙问候了？！ https://v.douyin.com/ihHknvbV/ w@f.bA YZm:/ 06/05 "},"Status":3,"ImgStatus":1,"ImgBuf":{"iLen":0},"CreateTime":1725343506,"MsgSource":"\n\t0\n\t1\n\t\n\t\t3\n\t\t86\n\t\n\t1\n\t4\n\tV1_pVkC6xlw|v1_pVkC6xlw\n\t\n\t\t\n\t\n\n","NewMsgId":6473116084228305363,"MsgSeq":776685968},"Wxid":"wxid_mr9pdsanc99422"}'} a = PQ.download_video(url,'/Users/tzld/Desktop/video_rewriting/path','70100016',video,'单点视频') print(a)