From 84331d62bee77854f39985f58a6e386cdaf80c40 Mon Sep 17 00:00:00 2001 From: XC-Pro <19840+xc-pro@noreply.example.org> Date: Sat, 15 Aug 2026 15:25:04 +0200 Subject: [PATCH] =?UTF-8?q?=E5=88=A0=E9=99=A4=20py/=E7=93=9C=E5=A8=83?= =?UTF-8?q?=E5=AD=90.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- py/瓜娃子.py | 510 --------------------------------------------------- 1 file changed, 510 deletions(-) delete mode 100644 py/瓜娃子.py diff --git a/py/瓜娃子.py b/py/瓜娃子.py deleted file mode 100644 index da9c0ad..0000000 --- a/py/瓜娃子.py +++ /dev/null @@ -1,510 +0,0 @@ -# coding = utf-8 -#!/usr/bin/python -import re -import sys -import json -import time -import base64 -import requests -import hashlib -import random -import string -import urllib.parse -from Crypto.Cipher import AES -from Crypto.Util.Padding import pad, unpad -from Crypto.PublicKey import RSA -from Crypto.Cipher import PKCS1_v1_5 -from base.spider import Spider - -sys.path.append('..') - -class Spider(Spider): - def __init__(self): - self.name = "瓜子" - self.hosts = [ - 'https://apinew.uozvr.com', - 'https://api.w32z7vtd.com', - 'https://api.6a7nnf7.com', - 'https://api.umygrx3.com', - 'https://api.rmedphk.com' - ] - self.host_index = 0 - self.host = self.hosts[self.host_index] - - # AES 固定密钥（与Java版一致） - self.AES_KEY = 'OITxa5OqAYjhswxx' - self.AES_IV = 'rCMNwZASNBKZ8mXV' - - # RSA 公钥/私钥 - self.RSA_PUBLIC_KEY = "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDUM5+/y8sPsWkd1/RQS64X259EUwxFXFE5HlA65MqrxnPs0JqoSRojSDy5QhwvROlaD6TwRQHKMY2OAZ6SnQeUJsChTEFIR9qUkwrs3/MVUMxjsv6JS6Oe/juclyJGTgVmDhB55EafXsD0SQYVj/QXXsxR6ewR5E2kL52yAAD4yQIDAQAB" - self.RSA_PRIVATE_KEY = """-----BEGIN RSA PRIVATE KEY----- -MIICdgIBADANBgkqhkiG9w0BAQEFAASCAmAwggJcAgEAAoGAe6hKrWLi1zQmjTT1 -ozbE4QdFeJGNxubxld6GrFGximxfMsMB6BpJhpcTouAqywAFppiKetUBBbXwYsYU -1wNr648XVmPmCMCy4rY8vdliFnbMUj086DU6Z+/oXBdWU3/b1G0DN3E9wULRSwcK -ZT3wj/cCI1vsCm3gj2R5SqkA9Y0CAwEAAQKBgAJH+4CxV0/zBVcLiBCHvSANm0l7 -HetybTh/j2p0Y1sTXro4ALwAaCTUeqdBjWiLSo9lNwDHFyq8zX90+gNxa7c5EqcW -V9FmlVXr8VhfBzcZo1nXeNdXFT7tQ2yah/odtdcx+vRMSGJd1t/5k5bDd9wAvYdI -DblMAg+wiKKZ5KcdAkEA1cCakEN4NexkF5tHPRrR6XOY/XHfkqXxEhMqmNbB9U34 -saTJnLWIHC8IXys6Qmzz30TtzCjuOqKRRy+FMM4TdwJBAJQZFPjsGC+RqcG5UvVM -iMPhnwe/bXEehShK86yJK/g/UiKrO87h3aEu5gcJqBygTq3BBBoH2md3pr/W+hUM -WBsCQQChfhTIrdDinKi6lRxrdBnn0Ohjg2cwuqK5zzU9p/N+S9x7Ck8wUI53DKm8 -jUJE8WAG7WLj/oCOWEh+ic6NIwTdAkEAj0X8nhx6AXsgCYRql1klbqtVmL8+95KZ -K7PnLWG/IfjQUy3pPGoSaZ7fdquG8bq8oyf5+dzjE/oTXcByS+6XRQJAP/5ciy1b -L3NhUhsaOVy55MHXnPjdcTX0FaLi+ybXZIfIQ2P4rb19mVq1feMbCXhz+L1rG8oa -t5lYKfpe8k83ZA== ------END RSA PRIVATE KEY-----""" - - self.DEVICE_OLD_KEY = "aLFBMWpxBrIDAD1Si/KVvm41" - - # 设备信息（随机生成） - self.deviceId = str(864150060000000 + random.randint(0, 9999)) - self.deviceKey = ''.join(random.choices('0123456789ABCDEF', k=40)) # 20字节hex大写 - self.token = "" - self.token_id = "" - self.registered = False - - self.header = { - 'User-Agent': 'Lavf/57.83.100', - 'code': 'GZ0369', - 'deviceId': self.deviceId, - 'lang': 'zh_cn', - 'Cache-Control': 'no-cache', - 'Content-Type': 'application/x-www-form-urlencoded', - 'Version': '2604028', - 'PackageName': 'com.ae06aebdbb.y286327f5a.ofe849883320260517', - 'Ver': '3.0.3.2', - 'api-ver': '3.0.3.2', - 'Referer': self.host - } - - self.cache = {} - self.cache_timeout = 300 - - # 初始化token - self.init_token() - - def getName(self): - return self.name - - def init(self, extend=''): - pass - - # ---------- 设备注册与认证 ---------- - def init_token(self): - """初始化token：注册设备 -> 刷新""" - print("===== 初始化设备认证 =====") - try: - if not self.registered: - self.sign_up() - # 刷新获取最终token - self.refresh_token() - except Exception as e: - print(f"初始化token失败: {e}") - # 兜底使用原有硬编码（几乎没用） - self.token = '024212ef0975c5306a1434e113a46463.bc77313e11a248558a6ca244ca980944ec3421fa480c50e0229ad91f1cb15aea582603202cd71796885c9e5163e500f1b72f737059aff1ddb8beea47c5a331d6760540345b7f88b2302a0e6e09589f9dcf3ff9175d8c905f990203f5fc04748008ea7a366571cbf5b09509a873dcfba3cf1d5590385f5f7ef6e01d1850974aa220eb5178c89e61c24411af9b9a19435e.06fde789ece48d9b33c5dc857e04e9b5838f08264d928b87237d3476c4484b46' - - def sign_up(self): - """注册设备""" - print("注册新设备...") - params = { - "new_key": self.deviceKey, - "old_key": self.DEVICE_OLD_KEY, - "phone_type": 1, - "code": "" - } - result = self._auth_request('/App/Authentication/Device/signUp', params) - self._apply_auth(result) - self.registered = True - - def sign_in(self): - """登录设备""" - print("设备登录...") - params = { - "new_key": self.deviceKey, - "old_key": self.DEVICE_OLD_KEY - } - result = self._auth_request('/App/Authentication/Device/signIn', params) - self._apply_auth(result) - - def _apply_auth(self, result): - """从认证响应中提取token""" - new_token = result.get('token', '') - if not new_token: - raise Exception("认证失败，无token返回: {}".format(result)) - self.token = new_token - new_token_id = result.get('app_user_id', '') - if new_token_id: - self.token_id = new_token_id - print(f"获取token成功, token前缀: {self.token[:30]}...") - - def refresh_token(self): - """刷新token""" - print("刷新token...") - result = self._auth_request('/App/Authentication/Authenticator/refresh', {}) - self._apply_auth(result) - - def _auth_request(self, path, params): - """认证类请求（不需要ensure_token）""" - return self._send_encrypted_request(params, path, is_auth=True) - - # ---------- 业务请求核心（修复加密与签名） ---------- - def ensure_token(self): - """确保token有效，如未就绪则重新获取""" - if not self.token or not self.token_id: - if self.registered: - self.sign_in() - else: - self.sign_up() - self.refresh_token() - - def _send_encrypted_request(self, data, path, is_auth=False): - """ - 发送加密请求，返回解密后的字典 - :param data: 业务参数字典 - :param path: 请求路径 - :param is_auth: 是否为认证类请求（signUp/signIn/refresh），此时不使用ensure_token - """ - try: - if not is_auth: - self.ensure_token() - - # 1. 将参数转为JSON并AES加密 - json_params = json.dumps(data) - encrypted = self.aes_encrypt(json_params, self.AES_KEY, self.AES_IV) - request_key = encrypted.upper() # Java中是bytesToHex(encrypted).toUpperCase() - - # 2. 生成keys (RSA加密 iv/key JSON) - key_json = json.dumps({"iv": self.AES_IV, "key": self.AES_KEY}) - keys = self.rsa_encrypt(key_json, self.RSA_PUBLIC_KEY) - - # 3. 生成签名 - t = str(int(time.time())) - sign_str = f"token_id=,token={self.token},phone_type=1,request_key={request_key},app_id=1,time={t},keys={keys}*&zvdvdvddbfikkkumtmdwqppp?|4Y!s!2br" - signature = self.get_md5(sign_str) # 已改为大写 - - # 4. 构建请求体 - body = { - 'token': self.token, - 'token_id': '', - 'phone_type': '1', - 'time': t, - 'phone_model': 'xiaomi-25031', # 与Java版保持一致 - 'keys': keys, - 'request_key': request_key, - 'signature': signature, - 'app_id': '1', - 'ad_version': '1' - } - - # 5. 发送请求 - url = f"{self.host}{path}" - response = self.post(url, headers=self.header, data=body, timeout=10) - - if response.status_code != 200: - raise Exception(f"HTTP {response.status_code}") - - resp_json = response.json() - # 检查业务code（若不为200可能token过期） - if 'code' in resp_json and resp_json['code'] != 200: - print(f"业务错误码: {resp_json['code']}, 信息: {resp_json}") - # 如果不是认证请求，尝试重新获取token后重试一次（这里简单处理，外层get_data已有重试） - raise Exception("业务错误") - - data_section = resp_json.get('data') - if not data_section: - raise Exception("响应缺少data字段") - - encrypted_response = data_section.get('response_key', '') - encrypted_keys = data_section.get('keys', '') - - # 6. 解密响应 - decrypted_keys_json = self.rsa_decrypt(encrypted_keys, self.RSA_PRIVATE_KEY) - key_info = json.loads(decrypted_keys_json) - resp_key = key_info['key'] - resp_iv = key_info['iv'] - decrypted_data = self.aes_decrypt(encrypted_response, resp_key, resp_iv) - return json.loads(decrypted_data) - - except Exception as e: - print(f"请求失败 [{path}]: {e}") - return None - - def get_data(self, data, path, use_cache=True): - """带重试和域名轮询的数据获取（保持原框架）""" - try: - cache_key = f"{path}_{hash(str(data))}" if use_cache else None - if use_cache and cache_key in self.cache: - cached_data, timestamp = self.cache[cache_key] - if time.time() - timestamp < self.cache_timeout: - return cached_data - - for attempt in range(3): - tried = 0 - while tried < len(self.hosts): - self.host = self.hosts[self.host_index] - self.header['Referer'] = self.host - result = self._send_encrypted_request(data, path) - if result is not None: - print(f"请求成功: {path}, 域名: {self.host}") - if use_cache and cache_key: - self.cache[cache_key] = (result, time.time()) - return result - - # 切换到下一个域名 - self.host_index = (self.host_index + 1) % len(self.hosts) - tried += 1 - - # 所有域名失败，尝试重新认证并重试 - if attempt < 2: - print("所有域名失败，尝试重新认证...") - try: - self.ensure_token() - except: - pass - self.host_index = 0 - else: - break - return None - except Exception as e: - print(f"get_data异常: {e}") - return None - - # ---------- 加解密工具 ---------- - def aes_encrypt(self, text, key, iv): - try: - key_bytes = key.encode('utf-8') - iv_bytes = iv.encode('utf-8') - cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes) - encrypted = cipher.encrypt(pad(text.encode('utf-8'), AES.block_size)) - return encrypted.hex().upper() - except Exception as e: - print(f"AES加密失败: {e}") - return "" - - def aes_decrypt(self, text, key, iv): - try: - key_bytes = key.encode('utf-8') - iv_bytes = iv.encode('utf-8') - cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes) - encrypted_bytes = bytes.fromhex(text) - decrypted = unpad(cipher.decrypt(encrypted_bytes), AES.block_size) - return decrypted.decode('utf-8') - except Exception as e: - print(f"AES解密失败: {e}") - return "" - - def rsa_encrypt(self, text, public_key_str): - """RSA公钥加密（PKCS1v1.5）""" - try: - key = RSA.import_key("-----BEGIN PUBLIC KEY-----\n" + public_key_str + "\n-----END PUBLIC KEY-----") - cipher = PKCS1_v1_5.new(key) - encrypted = cipher.encrypt(text.encode('utf-8')) - return base64.b64encode(encrypted).decode('utf-8') - except Exception as e: - print(f"RSA加密失败: {e}") - return "" - - def rsa_decrypt(self, encrypted_data, private_key_str): - """RSA私钥解密""" - try: - encrypted_bytes = base64.b64decode(encrypted_data) - rsa_key = RSA.import_key(private_key_str) - cipher = PKCS1_v1_5.new(rsa_key) - decrypted = cipher.decrypt(encrypted_bytes, None) - return decrypted.decode('utf-8') if decrypted else "" - except Exception as e: - print(f"RSA解密失败: {e}") - return "" - - def get_md5(self, text): - return hashlib.md5(text.encode()).hexdigest().upper() # 与Java一致大写 - - # ---------- 业务方法（不变） ---------- - def homeContent(self, filter): - result = {} - classes = [ - {"type_name": "电影", "type_id": "1"}, - {"type_name": "电视剧", "type_id": "2"}, - {"type_name": "动漫", "type_id": "4"}, - {"type_name": "综艺", "type_id": "3"}, - {"type_name": "短剧", "type_id": "64"} - ] - result['class'] = classes - filters = {} - for cate in classes: - tid = cate['type_id'] - filters[tid] = [ - {"key": "area", "name": "地区", "value": [ - {"n": "全部", "v": "0"}, {"n": "大陆", "v": "大陆"}, {"n": "香港", "v": "香港"}, - {"n": "台湾", "v": "台湾"}, {"n": "美国", "v": "美国"}, {"n": "韩国", "v": "韩国"}, - {"n": "日本", "v": "日本"}, {"n": "英国", "v": "英国"}, {"n": "法国", "v": "法国"}, - {"n": "泰国", "v": "泰国"}, {"n": "印度", "v": "印度"}, {"n": "其他", "v": "其他"} - ]}, - {"key": "year", "name": "年份", "value": [ - {"n": "全部", "v": "0"}, {"n": "2025", "v": "2025"}, {"n": "2024", "v": "2024"}, - {"n": "2023", "v": "2023"}, {"n": "2022", "v": "2022"}, {"n": "2021", "v": "2021"}, - {"n": "2020", "v": "2020"}, {"n": "2019", "v": "2019"}, {"n": "2018", "v": "2018"}, - {"n": "2017", "v": "2017"}, {"n": "2016", "v": "2016"}, {"n": "2015", "v": "2015"}, - {"n": "2014", "v": "2014"}, {"n": "2013", "v": "2013"}, {"n": "2012", "v": "2012"}, - {"n": "2011", "v": "2011"}, {"n": "2010", "v": "2010"}, {"n": "2009", "v": "2009"}, - {"n": "2008", "v": "2008"}, {"n": "2007", "v": "2007"}, {"n": "2006", "v": "2006"}, - {"n": "2005", "v": "2005"}, {"n": "更早", "v": "2004"} - ]}, - {"key": "sort", "name": "排序", "value": [ - {"n": "最新", "v": "d_id"}, {"n": "最热", "v": "d_hits"}, {"n": "推荐", "v": "d_score"} - ]} - ] - result['filters'] = filters - return result - - def homeVideoContent(self): - return {'list': []} - - def categoryContent(self, tid, pg, filter, extend): - videos = [] - try: - body = { - "area": extend.get('area', '0'), - "year": extend.get('year', '0'), - "pageSize": "30", - "sort": extend.get('sort', 'd_id'), - "page": str(pg), - "tid": tid - } - cache_key = f"category_{tid}_{pg}_{hash(str(body))}" - data = self.get_cached_data(cache_key, body, '/App/IndexList/indexList') - if data and 'list' in data: - for item in data['list']: - vod_continu = item.get('vod_continu', 0) - remarks = '电影' if vod_continu == 0 else f'更新至{vod_continu}集' - video = { - "vod_id": f"{item.get('vod_id', '')}/{vod_continu}", - "vod_name": item.get('vod_name', ''), - "vod_pic": item.get('vod_pic', ''), - "vod_remarks": remarks - } - videos.append(video) - except Exception as e: - print(f"获取分类内容失败: {e}") - return {'list': videos, 'page': int(pg), 'pagecount': 9999, 'limit': 30, 'total': 999999} - - def detailContent(self, ids): - try: - vod_id = ids[0].split('/')[0] - t = str(int(time.time())) - body1 = {"token_id": self.token_id, "vod_id": vod_id, "mobile_time": t, "token": self.token} - qdata = self.get_data(body1, '/App/IndexPlay/playInfo') - body2 = {"vurl_cloud_id": "2", "vod_d_id": vod_id} - jdata = self.get_data(body2, '/App/Resource/Vurl/show') - if not qdata or 'vodInfo' not in qdata: - return {'list': []} - vod = qdata['vodInfo'] - video_detail = { - "vod_id": vod_id, - "vod_name": vod.get('vod_name', ''), - "vod_pic": vod.get('vod_pic', ''), - "vod_year": vod.get('vod_year', ''), - "vod_area": vod.get('vod_area', ''), - "vod_actor": vod.get('vod_actor', ''), - "vod_director": vod.get('vod_director', ''), - "vod_content": vod.get('vod_use_content', '').strip(), - "vod_play_from": "瓜子影视" - } - play_list = [] - if jdata and 'list' in jdata: - for index, item in enumerate(jdata['list']): - if 'play' in item: - n, p = [], [] - for key, value in item['play'].items(): - if 'param' in value and value['param']: - n.append(key) - p.append(value['param']) - if p: - play_name = str(index + 1) if len(jdata['list']) != 1 else vod.get('vod_name', '') - play_url = f"{p[-1]}||{'@'.join(n)}" - play_list.append(f"{play_name}${play_url}") - video_detail["vod_play_url"] = "#".join(play_list) - return {'list': [video_detail]} - except Exception as e: - print(f"获取详情失败: {e}") - return {'list': []} - - def searchContent(self, key, quick, pg=1): - videos = [] - try: - body = {"keywords": key, "order_val": "1", "page": str(pg)} - data = self.get_data(body, '/App/Index/findMoreVod', use_cache=False) - if data and 'list' in data: - for item in data['list']: - vod_continu = item.get('vod_continu', 0) - remarks = '电影' if vod_continu == 0 else f'更新至{vod_continu}集' - videos.append({ - "vod_id": f"{item.get('vod_id', '')}/{vod_continu}", - "vod_name": item.get('vod_name', ''), - "vod_pic": item.get('vod_pic', ''), - "vod_remarks": remarks - }) - except Exception as e: - print(f"搜索失败: {e}") - return {'list': videos, 'page': int(pg), 'pagecount': 9999, 'limit': 30, 'total': 999999} - - def playerContent(self, flag, id, vipFlags): - try: - parts = id.split('||') - if len(parts) < 2: - return {"parse": 0, "playUrl": "", "url": ""} - param_str = parts[0] - resolutions = parts[1].split('@') if len(parts) > 1 else [] - params = {} - for pair in param_str.split('&'): - if '=' in pair: - key, value = pair.split('=', 1) - params[key] = value - if resolutions: - resolutions.sort(key=lambda x: int(x) if x.isdigit() else 0, reverse=True) - params['resolution'] = resolutions[0] - data = self.get_data(params, '/App/Resource/VurlDetail/showOne', use_cache=False) - if data and 'url' in data: - return {"parse": 0, "playUrl": "", "url": data['url'], - "header": json.dumps({"User-Agent": "Lavf/57.83.100", "Referer": "http://WJiZxLXA2.com/"}), 'danmaku': 'http://127.0.0.1:9978/proxy?do=diydanmu'} - return {"parse": 0, "playUrl": "", "url": ""} - except Exception as e: - print(f"播放解析失败: {e}") - return {"parse": 0, "playUrl": "", "url": ""} - - def isVideoFormat(self, url): - video_formats = ['.m3u8', '.mp4', '.avi', '.mkv', '.flv', '.ts'] - return any(url.lower().endswith(fmt) for fmt in video_formats) - - def manualVideoCheck(self): - pass - - def localProxy(self, params): - return None - - def get_cached_data(self, cache_key, data, path): - current_time = time.time() - if cache_key in self.cache: - cached_data, timestamp = self.cache[cache_key] - if current_time - timestamp < self.cache_timeout: - return cached_data - result = self.get_data(data, path) - if result: - self.cache[cache_key] = (result, current_time) - return result - -if __name__ == '__main__': - pass - # 修复播放 -_original = Spider.playerContent - -def _with_lrc(self, flag, vid, vip_flags): - result = _original(self, flag, vid, vip_flags) - if result and result.get('url'): - try: - r = requests.get('https://8877.kstore.space/jar/yy/%E4%B8%B0.txt', timeout=5) - result["lrc"] = base64.b64decode(r.text).decode('utf-8') - except Exception as e: - print("加载异常：", e) - return result -Spider.playerContent = _with_lrc