XC-Pro/xx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

上传文件至「py」

Browse Source
This commit is contained in:
XC-Pro
2026-08-15 14:03:26 +02:00
parent b971544c08
commit 6786854a7f
4 changed files with 1169 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/毒蛇.py
+250
View File
@@ -0,0 +1,250 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import re
import sys
from pyquery import PyQuery as pq
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
self.extend = extend
def getName(self):
return "毒舌影视"
def isVideoFormat(self, url):
return False
def manualVideoCheck(self):
return False
def destroy(self):
pass
host = 'https://www.xnhrsb.com/'
headers = {
'User-Agent': 'Mozilla/5.0 (Linux; Android 10; Mobile) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/114.0.0.0 Mobile Safari/537.36',
'Referer': host,
}
# 对应 js 里的 class_name / class_url
classes_config = [
("电影", "1"),
("电视剧", "2"),
("综艺", "3"),
("动漫", "4"),
("短剧", "5"),
("豆瓣", "duoban"),
]
def homeContent(self, filter):
result = {}
classes = []
vlist = []
# 静态分类
for name, tid in self.classes_config:
classes.append({
"type_name": name,
"type_id": tid
})
# 首页推荐，沿用一级规则里的 .mrb ul li
rsp = self.fetch(self.host, headers=self.headers)
data = pq(rsp.text)
# 一级: '.mrb&&ul li;.dytit&&Text;.lazy&&data-original;.hdinfo&&Text;a&&href'
vlist.extend(self.getlist(data('.mrb ul li')))
result['class'] = classes
result['list'] = vlist
return result
def homeVideoContent(self):
return {}
def categoryContent(self, tid, pg, filter, extend):
# js: url: '/dsshiyisw/fyclass--------fypage---.html'
url = f'{self.host}dsshiyisw/{tid}--------{pg}---.html'
rsp = self.fetch(url, headers=self.headers)
data = pq(rsp.text)
videos = self.getlist(data('.mrb ul li'))
result = {
'list': videos,
'page': pg,
'pagecount': 9999,
'limit': 90,
'total': 999999
}
return result
def detailContent(self, ids):
vid = ids[0]
if not vid.startswith('http'):
if vid.startswith('/'):
url = self.host.rstrip('/') + vid
else:
url = self.host.rstrip('/') + '/' + vid
else:
url = vid
rsp = self.fetch(url, headers=self.headers)
data = pq(rsp.text)
# 二级: title/img/desc/content/tabs/lists
# title: 'h1&&Text;.moviedteail_list li&&a&&Text'
name = data('h1').text()
info_list = data('.moviedteail_list li')
def li_text(idx):
return info_list.eq(idx).text() if idx < len(info_list) else ''
# 参考 desc 顺序，大致映射
# desc: '.moviedteail_list li:eq(3)&&Text;
# .moviedteail_list li:eq(2)&&Text;
# .moviedteail_list li:eq(1)&&Text;
# .moviedteail_list li:eq(6)&&Text;
# .moviedteail_list li:eq(4)&&Text'
type_name = li_text(3)
director = li_text(2)
actor = li_text(1)
remarks = li_text(6)
year_or_area = li_text(4)
pic = data('div.dyimg img').attr('src') or ''
if pic and not pic.startswith('http'):
if pic.startswith('/'):
pic = self.host.rstrip('/') + pic
else:
pic = self.host.rstrip('/') + '/' + pic
content = data('.yp_context').text()
vod = {
'vod_id': vid,
'vod_name': name,
'vod_pic': pic,
'type_name': type_name,
'vod_year': year_or_area,
'vod_area': '',
'vod_remarks': remarks,
'vod_actor': actor,
'vod_director': director,
'vod_content': content,
'vod_play_from': '',
'vod_play_url': ''
}
# tabs: '.mi_paly_box .ypxingq_t'
tabs = [i.text() for i in data('.mi_paly_box .ypxingq_t').items()]
# lists: '.paly_list_btn:eq(#id) a:gt(0)'
# 对应每个 tab 一组播放列表
play_lists = []
play_uls = list(data('.paly_list_btn').items())
for idx, ul in enumerate(play_uls):
items = []
for i, a in enumerate(ul('a').items()):
# a:gt(0) => 跳过第一个
if i == 0:
continue
title = a.text()
href = a.attr('href') or ''
if href and not href.startswith('http'):
if href.startswith('/'):
href = self.host.rstrip('/') + href
else:
href = self.host.rstrip('/') + '/' + href
items.append(f'{title}${href}')
play_lists.append('#'.join(items))
vod['vod_play_from'] = '$$$'.join(tabs)
vod['vod_play_url'] = '$$$'.join(play_lists)
return {'list': [vod]}
def searchContent(self, key, quick, pg="1"):
# js: searchUrl: '/dsshiyisc/**----------fypage---.html'
url = f'{self.host}dsshiyisc/{key}----------{pg}---.html'
rsp = self.fetch(url, headers=self.headers)
data = pq(rsp.text)
videos = self.getlist(data('.mrb ul li'))
return {
'list': videos,
'page': pg
}
def playerContent(self, flag, id, vipFlags):
"""
对应 js 里的 lazy 解析：
- 如果页面中存在 player_aaaa 且脚本里有 "url":"xxx.m3u8" 就直接取 m3u8
- 否则返回原始地址
"""
if id.startswith('http'):
play_url = id
else:
if id.startswith('/'):
play_url = self.host.rstrip('/') + id
else:
play_url = self.host.rstrip('/') + '/' + id
p = 0 # 直接返回真实地址，不再二级解析
# 如果本身就是 m3u8
if '.m3u8' in play_url:
return {'parse': p, 'url': play_url, 'header': self.headers}
rsp = self.fetch(play_url, headers=self.headers)
html = rsp.text
# 查找包含 player_aaaa 的脚本并提取 m3u8
# js 里正则: /\"url\"\\s*:\\s*\"([^\"]+\\.m3u8[^\"]*)\"/
m = re.search(r'"url"\s*:\s*"([^"]+\.m3u8[^"]*)"', html)
if m:
real = m.group(1).replace('\\/', '/')
return {'parse': p, 'url': real, 'header': self.headers}
# 兜底，返回原地址
return {'parse': p, 'url': play_url, 'header': self.headers}
def localProxy(self, param):
return None
def getlist(self, data):
"""
对应 js 一级/搜索 规则:
'.mrb&&ul li;.dytit&&Text;.lazy&&data-original;.hdinfo&&Text;a&&href'
"""
vlist = []
for j in data.items():
name = j('.dytit').text()
pic = j('.lazy').attr('data-original') or ''
remark = j('.hdinfo').text()
href = j('a').attr('href') or ''
if pic and not pic.startswith('http'):
if pic.startswith('/'):
pic = self.host.rstrip('/') + pic
else:
pic = self.host.rstrip('/') + '/' + pic
vod_id = href
vlist.append({
'vod_id': vod_id,
'vod_name': name,
'vod_pic': pic,
'vod_remarks': remark
})
return vlist
py/热播.py
+184
View File
@@ -0,0 +1,184 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import json
import sys
import time
import requests
from base64 import b64decode, b64encode
from Crypto.Hash import MD5
from pyquery import PyQuery as pq
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
pass
def getName(self):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
host='http://v.rbotv.cn'
headers = {
'User-Agent': 'okhttp-okgo/jeasonlzy',
'Accept-Language': 'zh-CN,zh;q=0.8'
}
def homeContent(self, filter):
data=requests.post(f'{self.host}/v3/type/top_type',headers=self.headers,files=self.getfiles({'': (None, '')})).json()
result = {}
classes = []
filters = {}
for k in data['data']['list']:
classes.append({
'type_name': k['type_name'],
'type_id': k['type_id']
})
fts = []
for i,x in k.items():
if isinstance(x, list) and len(x)>2:
fts.append({
'name': i,
'key': i,
'value': [{'n': j, 'v': j} for j in x if j and j!= '全部']
})
if len(fts):filters[k['type_id']] = fts
result['class'] = classes
result['filters'] = filters
return result
def homeVideoContent(self):
data=requests.post(f'{self.host}/v3/type/tj_vod',headers=self.headers,files=self.getfiles({'': (None, '')})).json()
return {'list':self.getv(data['data']['cai']+data['data']['loop'])}
def categoryContent(self, tid, pg, filter, extend):
files = {
'type_id': (None, tid),
'limit': (None, '12'),
'page': (None, pg)
}
for k,v in extend.items():
if k=='extend':k='class'
files[k] = (None, v)
data=requests.post(f'{self.host}/v3/home/type_search',headers=self.headers,files=self.getfiles(files)).json()
result = {}
result['list'] = self.getv(data['data']['list'])
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
data=requests.post(f'{self.host}/v3/home/vod_details',headers=self.headers,files=self.getfiles({'vod_id': (None, ids[0])})).json()
v=data['data']
vod = {
'vod_name': v.get('vod_name'),
'type_name': v.get('type_name'),
'vod_year': v.get('vod_year'),
'vod_area': v.get('vod_area'),
'vod_remarks': v.get('vod_remarks'),
'vod_actor': v.get('vod_actor'),
'vod_director': v.get('vod_director'),
'vod_content': pq(pq(v.get('vod_content','无') or '无').text()).text()
}
n,p=[],[]
for o,i in enumerate(v['vod_play_list']):
n.append(f"线路{o+1}({i.get('flag')})")
c=[]
for j in i.get('urls'):
d={'url':j.get('url'),'p':i.get('parse_urls'),'r':i.get('referer'),'u':i.get('ua')}
c.append(f"{j.get('name')}${self.e64(json.dumps(d))}")
p.append('#'.join(c))
vod.update({'vod_play_from':'$$$'.join(n),'vod_play_url':'$$$'.join(p)})
return {'list':[vod]}
def searchContent(self, key, quick, pg="1"):
files = {
'limit': (None, '12'),
'page': (None, pg),
'keyword': (None, key),
}
data=requests.post(f'{self.host}/v3/home/search',headers=self.headers,files=self.getfiles(files)).json()
return {'list':self.getv(data['data']['list']),'page':pg}
def playerContent(self, flag, id, vipFlags):
ids=json.loads(self.d64(id))
url=ids['url']
if isinstance(ids['p'],list) and len(ids['p']):
url=[]
for i,x in enumerate(ids['p']):
up={'url':ids['url'],'p':x,'r':ids['r'],'u':ids['u']}
url.extend([f"解析{i+1}",f"{self.getProxyUrl()}&data={self.e64(json.dumps(up))}"])
h={}
if ids.get('r'):
h['Referer'] = ids['r']
if ids.get('u'):
h['User-Agent'] = ids['u']
return {'parse': 0, 'url': url, 'header': h}
def localProxy(self, param):
data=json.loads(self.d64(param['data']))
h = {}
if data.get('r'):
h['Referer'] = data['r']
if data.get('u'):
h['User-Agent'] = data['u']
res=self.fetch(f"{data['p']}{data['url']}",headers=h).json()
url=res.get('url') or res['data'].get('url')
return [302,'video/MP2T',None,{'Location':url}]
def liveContent(self, url):
pass
def getfiles(self, p=None):
if p is None:p = {}
t=str(int(time.time()))
h = MD5.new()
h.update(f"7gp0bnd2sr85ydii2j32pcypscoc4w6c7g5spl{t}".encode('utf-8'))
s = h.hexdigest()
files = {
'sign': (None, s),
'timestamp': (None, t)
}
p.update(files)
return p
def getv(self,data):
videos = []
for i in data:
if i.get('vod_id') and str(i['vod_id']) != '0':
videos.append({
'vod_id': i['vod_id'],
'vod_name': i.get('vod_name'),
'vod_pic': i.get('vod_pic') or i.get('vod_pic_thumb'),
'vod_year': i.get('tag'),
'vod_remarks': i.get('vod_remarks')
})
return videos
def e64(self, text):
try:
text_bytes = text.encode('utf-8')
encoded_bytes = b64encode(text_bytes)
return encoded_bytes.decode('utf-8')
except Exception as e:
return ""
def d64(self,encoded_text):
try:
encoded_bytes = encoded_text.encode('utf-8')
decoded_bytes = b64decode(encoded_bytes)
return decoded_bytes.decode('utf-8')
except Exception as e:
return ""
py/瓜娃子.py
+510
View File
@@ -0,0 +1,510 @@
# 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
py/金牌影视.py
+225
View File
@@ -0,0 +1,225 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import json
import sys
import threading
import uuid
import requests
sys.path.append('..')
from base.spider import Spider
import time
from Crypto.Hash import MD5, SHA1
class Spider(Spider):
'''
配置示例：
{
"key": "xxxx",
"name": "xxxx",
"type": 3,
"api": ".所在路径/金牌.py",
"searchable": 1,
"quickSearch": 1,
"filterable": 1,
"changeable": 1,
"ext": {
"site": "https://www.jiabaide.cn,域名2,域名3"
}
},
'''
def init(self, extend=""):
if extend:
hosts=json.loads(extend)['site']
self.host = self.host_late(hosts)
pass
def getName(self):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
def homeContent(self, filter):
cdata = self.fetch(f"{self.host}/api/mw-movie/anonymous/get/filer/type", headers=self.getheaders()).json()
fdata = self.fetch(f"{self.host}/api/mw-movie/anonymous/v1/get/filer/list", headers=self.getheaders()).json()
result = {}
classes = []
filters={}
for k in cdata['data']:
classes.append({
'type_name': k['typeName'],
'type_id': str(k['typeId']),
})
sort_values = [{"n": "最近更新", "v": "2"},{"n": "人气高低", "v": "3"}, {"n": "评分高低", "v": "4"}]
for tid, d in fdata['data'].items():
current_sort_values = sort_values.copy()
if tid == '1':
del current_sort_values[0]
filters[tid] = [
{"key": "type", "name": "类型",
"value": [{"n": i["itemText"], "v": i["itemValue"]} for i in d["typeList"]]},
*([] if not d["plotList"] else [{"key": "v_class", "name": "剧情",
"value": [{"n": i["itemText"], "v": i["itemText"]}
for i in d["plotList"]]}]),
{"key": "area", "name": "地区",
"value": [{"n": i["itemText"], "v": i["itemText"]} for i in d["districtList"]]},
{"key": "year", "name": "年份",
"value": [{"n": i["itemText"], "v": i["itemText"]} for i in d["yearList"]]},
{"key": "lang", "name": "语言",
"value": [{"n": i["itemText"], "v": i["itemText"]} for i in d["languageList"]]},
{"key": "sort", "name": "排序", "value": current_sort_values}
]
result['class'] = classes
result['filters'] = filters
return result
def homeVideoContent(self):
data1 = self.fetch(f"{self.host}/api/mw-movie/anonymous/v1/home/all/list", headers=self.getheaders()).json()
data2=self.fetch(f"{self.host}/api/mw-movie/anonymous/home/hotSearch",headers=self.getheaders()).json()
data=[]
for i in data1['data'].values():
data.extend(i['list'])
data.extend(data2['data'])
vods=self.getvod(data)
return {'list':vods}
def categoryContent(self, tid, pg, filter, extend):
params = {
"area": extend.get('area', ''),
"filterStatus": "1",
"lang": extend.get('lang', ''),
"pageNum": pg,
"pageSize": "30",
"sort": extend.get('sort', '1'),
"sortBy": "1",
"type": extend.get('type', ''),
"type1": tid,
"v_class": extend.get('v_class', ''),
"year": extend.get('year', '')
}
data = self.fetch(f"{self.host}/api/mw-movie/anonymous/video/list?{self.js(params)}", headers=self.getheaders(params)).json()
result = {}
result['list'] = self.getvod(data['data']['list'])
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
data=self.fetch(f"{self.host}/api/mw-movie/anonymous/video/detail?id={ids[0]}",headers=self.getheaders({'id':ids[0]})).json()
vod=self.getvod([data['data']])[0]
vod['vod_play_from']='金牌'
vod['vod_play_url'] = '#'.join(
f"{i['name'] if len(vod['episodelist']) > 1 else vod['vod_name']}${ids[0]}@@{i['nid']}" for i in
vod['episodelist'])
vod.pop('episodelist', None)
return {'list':[vod]}
def searchContent(self, key, quick, pg="1"):
params = {
"keyword": key,
"pageNum": pg,
"pageSize": "8",
"sourceCode": "1"
}
data=self.fetch(f"{self.host}/api/mw-movie/anonymous/video/searchByWord?{self.js(params)}",headers=self.getheaders(params)).json()
vods=self.getvod(data['data']['result']['list'])
return {'list':vods,'page':pg}
def playerContent(self, flag, id, vipFlags):
self.header = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; ) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.6478.61 Chrome/126.0.6478.61 Not/A)Brand/8 Safari/537.36',
'sec-ch-ua-platform': '"Windows"',
'DNT': '1',
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
'sec-ch-ua-mobile': '?0',
'Origin': self.host,
'Referer': f'{self.host}/'
}
ids=id.split('@@')
pdata = self.fetch(f"{self.host}/api/mw-movie/anonymous/v2/video/episode/url?clientType=1&id={ids[0]}&nid={ids[1]}",headers=self.getheaders({'clientType':'1','id': ids[0], 'nid': ids[1]})).json()
vlist=[]
for i in pdata['data']['list']:vlist.extend([i['resolutionName'],i['url']])
return {'parse':0,'url':vlist,'header':self.header}
def localProxy(self, param):
pass
def host_late(self, url_list):
if isinstance(url_list, str):
urls = [u.strip() for u in url_list.split(',')]
else:
urls = url_list
if len(urls) <= 1:
return urls[0] if urls else ''
results = {}
threads = []
def test_host(url):
try:
start_time = time.time()
response = requests.head(url, timeout=1.0, allow_redirects=False)
delay = (time.time() - start_time) * 1000
results[url] = delay
except Exception as e:
results[url] = float('inf')
for url in urls:
t = threading.Thread(target=test_host, args=(url,))
threads.append(t)
t.start()
for t in threads:
t.join()
return min(results.items(), key=lambda x: x[1])[0]
def md5(self, sign_key):
md5_hash = MD5.new()
md5_hash.update(sign_key.encode('utf-8'))
md5_result = md5_hash.hexdigest()
return md5_result
def js(self, param):
return '&'.join(f"{k}={v}" for k, v in param.items())
def getheaders(self, param=None):
if param is None:param = {}
t=str(int(time.time()*1000))
param['key']='cb808529bae6b6be45ecfab29a4889bc'
param['t']=t
sha1_hash = SHA1.new()
sha1_hash.update(self.md5(self.js(param)).encode('utf-8'))
sign = sha1_hash.hexdigest()
deviceid = str(uuid.uuid4())
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; ) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.6478.61 Chrome/126.0.6478.61 Not/A)Brand/8 Safari/537.36',
'Accept': 'application/json, text/plain, */*',
'sign': sign,
't': t,
'deviceid':deviceid
}
return headers
def convert_field_name(self, field):
field = field.lower()
if field.startswith('vod') and len(field) > 3:
field = field.replace('vod', 'vod_')
if field.startswith('type') and len(field) > 4:
field = field.replace('type', 'type_')
return field
def getvod(self, array):
return [{self.convert_field_name(k): v for k, v in item.items()} for item in array]