526 lines
21 KiB
Python
526 lines
21 KiB
Python
# -*- coding: utf-8 -*-
|
|
# 本资源来源于互联网公开渠道，仅可用于个人学习爬虫技术。
|
|
# 严禁将其用于任何商业用途，下载后请于 24 小时内删除，搜索结果均来自源站，本人不承担任何责任。
|
|
|
|
import re,time,json,urllib3
|
|
from base.spider import Spider
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
class Spider(Spider):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.base_url = self.device_material = self.base_headers = None
|
|
self.jx_prefix = [] # 初始化存放解析前缀列表
|
|
|
|
def init(self, extend=''):
|
|
try:
|
|
ext = json.loads(extend.rstrip())
|
|
doh, pkg, ver, md5 = ext['doh'], ext['pkg'], ext['ver'], ext['md5']
|
|
if not (doh and pkg and ver and bool(re.fullmatch(r'[0-9A-F]{32}', md5))): raise ValueError
|
|
self.device_material = f"{pkg}|{ver}|{md5}"
|
|
jx_prefix_ext = ext.get('jxPrefix', '')
|
|
self.jx_prefix = [p.strip() for p in jx_prefix_ext.split(',') if p.strip()]
|
|
self.base_headers = {
|
|
'User-Agent': f"Dalvik/2.1.0 (Linux; U; Android 17; Pixel 10 Build/TQ3A.260701.001); cycdm-android/{ver}",
|
|
'Accept': "application/json,application/protobuf",
|
|
'Accept-Encoding': "gzip",
|
|
'accept-charset': "UTF-8"
|
|
}
|
|
|
|
# 通过 DoH 获取最新域名，并进行多域名 Ping 优选
|
|
# 1. 发起 DoH 请求获取动态域名
|
|
data = self.fetch('https://doh.pub/dns-query', params={'name':doh,'type':"txt"}, headers={'User-Agent':'okhttp/5.1.0','Accept':'application/dns-json'}, verify=False, timeout=5).json()
|
|
|
|
domains = []
|
|
if 'Answer' in data:
|
|
for answer in data['Answer']:
|
|
# 16 代表 TXT 记录
|
|
if answer.get('type') == 16:
|
|
# 提取 data 数据并去除首尾的转义引号
|
|
txt_data = answer.get('data', '').strip('"').strip("'")
|
|
|
|
# 兼容多域名存在的情况 (处理可能的逗号或空格分隔)
|
|
for d in txt_data.replace(',', ' ').split():
|
|
if d.startswith('http'):
|
|
domains.append(d)
|
|
|
|
# 去除重复域名，保持列表唯一性
|
|
domains = list(set(domains))
|
|
|
|
# 2. 检查是否存在多域名，并进行 Ping 优选
|
|
if domains:
|
|
for domain in domains:
|
|
# 如果该域名 Ping 测试通过，则将其设为基准 URL 并跳出循环
|
|
if self._ping_domain(domain):
|
|
self.base_url = domain
|
|
# print(f"初始化成功，当前优选域名: {self.base_url}")
|
|
break
|
|
except Exception:
|
|
self.base_url = ''
|
|
# print(f"Init 获取或校验域名失败")
|
|
|
|
def _ping_domain(self, domain):
|
|
# 域名连通性测试方法 (带有动态签名)
|
|
try:
|
|
# Ping 根域名，由于无需特定 path，传入空字符串即可
|
|
path = ""
|
|
clean_params = {}
|
|
|
|
# 调用原类自带的签名方法，生成时间戳参数与 Auth Header
|
|
final_params, auth_header = self.get_signature_headers(path, clean_params)
|
|
|
|
headers = {**self.base_headers, 'x-cyc-auth': auth_header }
|
|
|
|
# 设置 3 秒短超时，用于快速筛除失效节点
|
|
res = self.fetch(domain, params=final_params, headers=headers, verify=False, timeout=3)
|
|
|
|
# 如果请求响应正常，说明域名存活且签名算法匹配
|
|
if res.status_code == 200:
|
|
return True
|
|
except Exception:
|
|
pass
|
|
|
|
return False
|
|
|
|
def homeContent(self, filter):
|
|
if not self.base_url: return None
|
|
data_dict = self.request_api("/index/nav")
|
|
classes, filters = [], {}
|
|
|
|
categories = self._ensure_list(data_dict.get(3, []))
|
|
for cat in categories:
|
|
if not isinstance(cat, dict): continue
|
|
|
|
type_id = str(cat.get(1, ''))
|
|
raw_type_name = cat.get(2, '')
|
|
if isinstance(raw_type_name, dict) or not raw_type_name: continue
|
|
|
|
type_name = str(raw_type_name)
|
|
if not type_id: continue
|
|
|
|
classes.append({'type_id': type_id, 'type_name': type_name})
|
|
|
|
# 解析筛选条件
|
|
filter_configs = []
|
|
raw_filters = cat.get(3, {})
|
|
if isinstance(raw_filters, dict):
|
|
field_map = {1: ('class', '类型'), 2: ('area', '地区'), 3: ('lang', '语言'), 4: ('year', '年份')}
|
|
for f_num, (f_key, f_name) in field_map.items():
|
|
val_str = raw_filters.get(f_num, '')
|
|
if not val_str or not isinstance(val_str, str): continue
|
|
|
|
values_list = [{"n": "全部", "v": ""}]
|
|
for item in val_str.split(','):
|
|
if item.strip(): values_list.append({"n": item.strip(), "v": item.strip()})
|
|
filter_configs.append({"key": f_key, "name": f_name, "init": "", "value": values_list})
|
|
|
|
# 固定排序筛选
|
|
filter_configs.append({
|
|
"key": "order", "name": "排序", "init": "time",
|
|
"value": [{"n": "最新", "v": "time"}, {"n": "热度最高", "v": "hits"}, {"n": "好评", "v": "score"}]
|
|
})
|
|
filters[type_id] = filter_configs
|
|
|
|
return {'class': classes, 'filters': filters}
|
|
|
|
def homeVideoContent(self):
|
|
if not self.base_url: return None
|
|
data_dict = self.request_api("/index/video")
|
|
videos = []
|
|
|
|
categories = self._ensure_list(data_dict.get(3, []))
|
|
for cat in categories:
|
|
if not isinstance(cat, dict): continue
|
|
cat_videos = self._ensure_list(cat.get(5, []))
|
|
for v in cat_videos:
|
|
if not isinstance(v, dict): continue
|
|
videos.append({
|
|
'vod_id': str(v.get(1, '')), 'vod_name': str(v.get(2, '')),
|
|
'vod_pic': str(v.get(3, '')), 'vod_remarks': str(v.get(5, '')),
|
|
'vod_year': str(v.get(7, '')), 'type_name': str(v.get(9, ''))
|
|
})
|
|
return {'list': videos}
|
|
|
|
def categoryContent(self, tid, pg, filter, extend):
|
|
if not self.base_url: return None
|
|
params = {'tid': str(tid), 'page': str(pg), 'limit': "20", 'order': "time"}
|
|
if extend: params.update({k: str(v) for k, v in extend.items()})
|
|
|
|
data_dict = self.request_api("/v2/video/query", params=params)
|
|
videos, raw_list, total = [], [], 0
|
|
|
|
# 兼容 JSON 与 PB 数据体
|
|
data_body = data_dict.get('data') or data_dict.get(3) or {}
|
|
|
|
if isinstance(data_body, dict):
|
|
total = data_body.get('total') or data_body.get(1) or 0
|
|
raw_list = data_body.get('list') or data_body.get(2) or []
|
|
elif isinstance(data_body, list):
|
|
for item in data_body:
|
|
if isinstance(item, dict):
|
|
raw_list.extend(self._ensure_list(item.get('list') or item.get(5) or []))
|
|
|
|
for v in self._ensure_list(raw_list):
|
|
if not isinstance(v, dict): continue
|
|
|
|
vod_id = v.get('vod_id') or v.get(1) or ''
|
|
vod_name = v.get('name') or v.get(2) or ''
|
|
if vod_id and vod_name:
|
|
videos.append({
|
|
'vod_id': str(vod_id),
|
|
'vod_name': str(vod_name),
|
|
'vod_pic': str(v.get('pic') or v.get(3) or ''),
|
|
'vod_remarks': str(v.get('remarks') or v.get(8) or v.get(5) or '')
|
|
})
|
|
|
|
try:
|
|
total_int = int(total)
|
|
except Exception:
|
|
total_int = 0
|
|
page_count = (total_int + 19) // 20 if total_int > 0 else 1
|
|
|
|
return {'list': videos, 'page': pg, 'pagecount': page_count, 'limit': 20, 'total': total_int}
|
|
|
|
def searchContent(self, key, quick, pg='1'):
|
|
if not self.base_url: return None
|
|
params = {'text': str(key), 'pg': str(pg), 'type_id': "0", 'limit': "20"}
|
|
data_dict = self.request_api("/v2/video/search", params=params)
|
|
videos = []
|
|
|
|
data_body = data_dict.get('data') or data_dict.get(4) or data_dict.get(3) or []
|
|
for v in self._ensure_list(data_body):
|
|
if not isinstance(v, dict): continue
|
|
|
|
vod_id = v.get('vod_id') or v.get(1) or ''
|
|
vod_name = v.get('name') or v.get(2) or ''
|
|
if vod_id and vod_name:
|
|
videos.append({
|
|
'vod_id': str(vod_id),
|
|
'vod_name': str(vod_name),
|
|
'vod_pic': str(v.get('pic') or v.get(3) or v.get(4) or ''),
|
|
'vod_remarks': str(v.get('remarks') or v.get(8) or v.get(5) or '')
|
|
})
|
|
|
|
total = data_dict.get('total') or data_dict.get(3) or data_dict.get(1) or 0
|
|
try:
|
|
total_int = int(total)
|
|
except Exception:
|
|
total_int = 0
|
|
page_count = (total_int + 19) // 20 if total_int > 0 else 1
|
|
|
|
return {'list': videos, 'page': pg, 'pagecount': page_count}
|
|
|
|
def detailContent(self, ids):
|
|
if not self.base_url: return None
|
|
vod_id = ids[0]
|
|
data_dict = self.request_api(f"/v2/video/info/{vod_id}")
|
|
data = data_dict.get('data') or data_dict.get(3) or {}
|
|
|
|
shows, play_urls = [], []
|
|
play_from_list = self._ensure_list(data.get('vod_play_from') or data.get(24) or data.get(20) or [])
|
|
|
|
for pf in play_from_list:
|
|
if not isinstance(pf, dict): continue
|
|
source_code = pf.get('code') or pf.get(1)
|
|
source_name = pf.get('name') or pf.get(2) or source_code
|
|
if not source_code: continue
|
|
|
|
# 【修改处】在详情页直接拉取所有的剧集内容，固定 index='-1'
|
|
params = {'id': str(vod_id), 'from': str(source_code), 'index': '-1'}
|
|
play_data_dict = self.request_api("/video/play_url", params=params)
|
|
|
|
# 取出列表数据，对应 JSON 里的 "3"
|
|
raw_play_list = play_data_dict.get('data') or play_data_dict.get(3) or []
|
|
|
|
eps = []
|
|
for ep in self._ensure_list(raw_play_list):
|
|
if not isinstance(ep, dict): continue
|
|
|
|
ep_name = ep.get('name') or ep.get(1) or '未知'
|
|
ep_url = ep.get('url') or ep.get('play_url') or ep.get(2) or ''
|
|
# 获取新增加的值 3，表示是否 json 解析
|
|
ep_parse = ep.get('parse') or ep.get(3) or 0
|
|
|
|
if ep_url:
|
|
# 用 @@ 分隔传递 URL 和 Parse 标识到 playerContent
|
|
eps.append(f"{ep_name}${ep_url}@@{ep_parse}")
|
|
|
|
if eps:
|
|
shows.append(str(source_name))
|
|
play_urls.append('#'.join(eps))
|
|
|
|
video = {
|
|
'vod_id': str(vod_id),
|
|
'vod_name': str(data.get('vod_name') or data.get(3) or data.get(2) or ''),
|
|
'vod_pic': str(data.get('vod_pic') or data.get(7) or data.get(4) or ''),
|
|
'vod_remarks': str(data.get('vod_remarks') or data.get(12) or data.get(5) or ''),
|
|
'vod_year': str(data.get('vod_year') or data.get(14) or data.get(15) or ''),
|
|
'vod_area': str(data.get('vod_area') or data.get(13) or data.get(14) or ''),
|
|
'vod_actor': str(data.get('vod_actor') or data.get(8) or ''),
|
|
'vod_director': str(data.get('vod_director') or data.get(9) or ''),
|
|
'vod_content': str(data.get('vod_content') or data.get(23) or data.get(15) or ''),
|
|
'type_name': str(data.get('vod_class') or data.get(6) or ''),
|
|
'vod_play_from': '$$$'.join(shows),
|
|
'vod_play_url': '$$$'.join(play_urls)
|
|
}
|
|
return {'list': [video]}
|
|
|
|
def playerContent(self, flag, vid, vip_flags):
|
|
# 【修改处】分离从详情页传递过来的 url 与 parse_flag
|
|
parts = vid.split('@@')
|
|
url = parts[0]
|
|
parse_flag = parts[1] if len(parts) > 1 else '0'
|
|
|
|
# 匹配配置里的多规则解析前缀
|
|
matched_prefix = None
|
|
for prefix in self.jx_prefix:
|
|
if url.startswith(prefix):
|
|
matched_prefix = prefix
|
|
break
|
|
|
|
# 如果获取标识值为 1，或者匹配到了 jx_prefix 前缀，执行 json 解析
|
|
if str(parse_flag) == '1' or matched_prefix:
|
|
try:
|
|
res = self.fetch(url, verify=False, headers={'User-Agent': 'okhttp/4.12.0'})
|
|
json_res = res.json()
|
|
# 尝试获取 json 里的 url 键值，失败则抹除前缀进行直连兜底
|
|
url = json_res.get('url') or (url.replace(matched_prefix, '', 1) if matched_prefix else url)
|
|
except Exception:
|
|
# 解析失败时的兜底逻辑
|
|
url = url.replace(matched_prefix, '', 1) if matched_prefix else url
|
|
|
|
return {'parse': 0, 'url': url, 'header': {'User-Agent': 'libmpv'}}
|
|
|
|
def _ensure_list(self, obj):
|
|
"""确保对象返回列表格式"""
|
|
if not obj: return []
|
|
return obj if isinstance(obj, list) else [obj]
|
|
|
|
# 纯 Python 实现 XXHash64
|
|
def xxh64_hexdigest(self, data, seed=0):
|
|
MASK64 = 0xFFFFFFFFFFFFFFFF
|
|
PRIME64_1 = 0x9E3779B185EBCA87
|
|
PRIME64_2 = 0xC2B2AE3D27D4EB4F
|
|
PRIME64_3 = 0x165667B19E3779F9
|
|
PRIME64_4 = 0x85EBCA77C2B2AE63
|
|
PRIME64_5 = 0x27D4EB2F165667C5
|
|
|
|
def rotl64(x, r):
|
|
return ((x << r) | (x >> (64 - r))) & MASK64
|
|
|
|
def round_calc(acc, input_val):
|
|
acc = (acc + (input_val * PRIME64_2) & MASK64) & MASK64
|
|
acc = rotl64(acc, 31)
|
|
return (acc * PRIME64_1) & MASK64
|
|
|
|
def merge_round(acc, val):
|
|
val = round_calc(0, val)
|
|
acc = (acc ^ val) & MASK64
|
|
return (acc * PRIME64_1 + PRIME64_4) & MASK64
|
|
|
|
b = data if isinstance(data, bytes) else data.encode('utf-8')
|
|
l = len(b)
|
|
|
|
if l >= 32:
|
|
v1 = (seed + PRIME64_1 + PRIME64_2) & MASK64
|
|
v2 = (seed + PRIME64_2) & MASK64
|
|
v3 = seed & MASK64
|
|
v4 = (seed - PRIME64_1) & MASK64
|
|
|
|
p = 0
|
|
limit = l - 32
|
|
while p <= limit:
|
|
v1 = round_calc(v1, int.from_bytes(b[p:p + 8], 'little'))
|
|
p += 8
|
|
v2 = round_calc(v2, int.from_bytes(b[p:p + 8], 'little'))
|
|
p += 8
|
|
v3 = round_calc(v3, int.from_bytes(b[p:p + 8], 'little'))
|
|
p += 8
|
|
v4 = round_calc(v4, int.from_bytes(b[p:p + 8], 'little'))
|
|
p += 8
|
|
|
|
h64 = (rotl64(v1, 1) + rotl64(v2, 7) + rotl64(v3, 12) + rotl64(v4, 18)) & MASK64
|
|
h64 = merge_round(h64, v1)
|
|
h64 = merge_round(h64, v2)
|
|
h64 = merge_round(h64, v3)
|
|
h64 = merge_round(h64, v4)
|
|
else:
|
|
h64 = (seed + PRIME64_5) & MASK64
|
|
|
|
h64 = (h64 + l) & MASK64
|
|
p = l - (l % 32) if l >= 32 else 0
|
|
|
|
while p + 8 <= l:
|
|
k1 = round_calc(0, int.from_bytes(b[p:p + 8], 'little'))
|
|
h64 = (h64 ^ k1) & MASK64
|
|
h64 = rotl64(h64, 27)
|
|
h64 = (h64 * PRIME64_1 + PRIME64_4) & MASK64
|
|
p += 8
|
|
|
|
if p + 4 <= l:
|
|
h64 = (h64 ^ ((int.from_bytes(b[p:p + 4], 'little') * PRIME64_1) & MASK64)) & MASK64
|
|
h64 = rotl64(h64, 23)
|
|
h64 = (h64 * PRIME64_2 + PRIME64_3) & MASK64
|
|
p += 4
|
|
|
|
while p < l:
|
|
h64 = (h64 ^ ((b[p] * PRIME64_5) & MASK64)) & MASK64
|
|
h64 = rotl64(h64, 11)
|
|
h64 = (h64 * PRIME64_1) & MASK64
|
|
p += 1
|
|
|
|
h64 = (h64 ^ (h64 >> 33)) & MASK64
|
|
h64 = (h64 * PRIME64_2) & MASK64
|
|
h64 = (h64 ^ (h64 >> 29)) & MASK64
|
|
h64 = (h64 * PRIME64_3) & MASK64
|
|
h64 = (h64 ^ (h64 >> 32)) & MASK64
|
|
|
|
return f"{h64:016x}"
|
|
|
|
def parse_pb_varint(self, stream, pos):
|
|
result, shift = 0, 0
|
|
while pos < len(stream):
|
|
byte = stream[pos]
|
|
pos += 1
|
|
result |= (byte & 0x7F) << shift
|
|
shift += 7
|
|
if not (byte & 0x80): break
|
|
return result, pos
|
|
|
|
def is_valid_pb_message(self, b):
|
|
p = 0
|
|
while p < len(b):
|
|
try:
|
|
tag_wire, p = self.parse_pb_varint(b, p)
|
|
if p > len(b) or tag_wire == 0: return False
|
|
wt = tag_wire & 0x07
|
|
fn = tag_wire >> 3
|
|
if fn == 0: return False
|
|
if wt == 0:
|
|
_, p = self.parse_pb_varint(b, p)
|
|
elif wt == 1:
|
|
p += 8
|
|
elif wt == 2:
|
|
length, p = self.parse_pb_varint(b, p)
|
|
p += length
|
|
elif wt == 5:
|
|
p += 4
|
|
else:
|
|
return False
|
|
except Exception:
|
|
return False
|
|
return p == len(b) and len(b) > 0
|
|
|
|
def decode_length_delimited(self, data):
|
|
if not data: return ""
|
|
|
|
# 1. 优先尝试解析为 UTF-8 字符串
|
|
try:
|
|
decoded_str = data.decode('utf-8')
|
|
# 增加一个简单的启发式判断：如果字符串中包含大量不可打印字符，则认为它不是普通文本
|
|
# 如果是合法的文本，则直接返回，避免被误判为嵌套消息
|
|
if all(c.isprintable() or c.isspace() for c in decoded_str) and len(decoded_str) > 0:
|
|
return decoded_str
|
|
except UnicodeDecodeError:
|
|
pass
|
|
|
|
# 2. 如果不是有效的字符串，再检查是否为合法的 PB 嵌套结构
|
|
if self.is_valid_pb_message(data):
|
|
try:
|
|
parsed = self.parse_pb_message(data)
|
|
if isinstance(parsed, dict) and len(parsed) > 0:
|
|
return parsed
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. 实在不行则返回 Hex
|
|
return data.hex()
|
|
|
|
def parse_pb_message(self, stream):
|
|
pos, msg = 0, {}
|
|
while pos < len(stream):
|
|
try:
|
|
tag_wire, pos = self.parse_pb_varint(stream, pos)
|
|
if pos > len(stream) or tag_wire == 0: break
|
|
wire_type = tag_wire & 0x07
|
|
field_num = tag_wire >> 3
|
|
|
|
val = None
|
|
if wire_type == 0: # Varint
|
|
val, pos = self.parse_pb_varint(stream, pos)
|
|
elif wire_type == 1: # 64-bit
|
|
val, pos = stream[pos:pos + 8], pos + 8
|
|
elif wire_type == 2: # Length-delimited (关键点)
|
|
length, pos = self.parse_pb_varint(stream, pos)
|
|
# 此处调用改进后的 decode_length_delimited
|
|
val = self.decode_length_delimited(stream[pos:pos + length])
|
|
pos += length
|
|
elif wire_type == 5: # 32-bit
|
|
val, pos = stream[pos:pos + 4], pos + 4
|
|
else:
|
|
break
|
|
|
|
if field_num not in msg: msg[field_num] = []
|
|
msg[field_num].append(val)
|
|
except Exception:
|
|
break
|
|
|
|
# 展开单元素列表
|
|
for k in msg:
|
|
if len(msg[k]) == 1: msg[k] = msg[k][0]
|
|
return msg
|
|
|
|
|
|
def get_signature_headers(self, path, business_params):
|
|
timestamp = int(time.time() * 1000)
|
|
full_params = dict(business_params)
|
|
full_params['timestamp'] = str(timestamp)
|
|
|
|
|
|
hash1 = self.xxh64_hexdigest(self.device_material, seed=0)
|
|
|
|
sign_params = {str(k): [str(v)] for k, v in full_params.items()}
|
|
data = {
|
|
"method": "GET",
|
|
"timestamp": timestamp,
|
|
"path": path,
|
|
"parameters": sign_params,
|
|
"body": ""
|
|
}
|
|
|
|
json_str = json.dumps(data, separators=(',', ':'), ensure_ascii=False)
|
|
hash2 = self.xxh64_hexdigest(json_str, seed=timestamp)
|
|
final_input = f"{timestamp}|{hash2}|{hash1}"
|
|
final_seed = timestamp ^ 0x9E3779B185EBCA87
|
|
final_hash = self.xxh64_hexdigest(final_input, seed=final_seed)
|
|
|
|
return full_params, f"v3:{timestamp}:{final_hash}"
|
|
|
|
def request_api(self, path, params=None):
|
|
if params is None: params = {}
|
|
clean_params = {k: v for k, v in params.items() if v != ""}
|
|
final_params, auth_header = self.get_signature_headers(path, clean_params)
|
|
|
|
headers = { **self.base_headers, 'x-cyc-auth': auth_header }
|
|
response = self.fetch(f"{self.base_url}{path}", params=final_params, verify=False, headers=headers)
|
|
if 'application/protobuf' in response.headers.get('Content-Type', '').lower():
|
|
return self.parse_pb_message(response.content)
|
|
try:
|
|
return response.json()
|
|
except Exception:
|
|
return {}
|
|
|
|
def getName(self):
|
|
pass
|
|
|
|
def isVideoFormat(self, url):
|
|
pass
|
|
|
|
def manualVideoCheck(self):
|
|
pass
|
|
|
|
def destroy(self):
|
|
pass
|
|
|
|
def localProxy(self, param):
|
|
pass |