yeyin/tvbox-1
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update zip

Browse Source
This commit is contained in:
GitHub Actions Bot
2026-03-21 18:27:22 +08:00
parent 1210be41f7
commit 06e814560a
6 changed files with 1626 additions and 1583 deletions
Download Patch File Download Diff File Expand all files Collapse all files
js/xiaosa/py/网络直播.py
+831 -338
View File
@@ -1,9 +1,16 @@
# -*- coding: utf-8 -*-
# by @嗷呜
# by @星河
# 修复版本 - 参考最新三合一.js重构虎牙、斗鱼、B站直播逻辑
# 修复：虎牙清晰度选择，确保ratio参数正确传递码率值
# 修复：斗鱼切换分辨率只能播放1秒的问题（每次重新获取安全密钥和签名）
# 修复：B站使用特殊UA和WBI签名绕过-352风控 [^90^][^30^]
import json
import re
import sys
import time
import hashlib
import random
import urllib.parse
from base64 import b64decode, b64encode
from urllib.parse import parse_qs
import requests
@@ -16,19 +23,13 @@ from concurrent.futures import ThreadPoolExecutor
class Spider(Spider):
def init(self, extend=""):
tid = 'douyin'
headers = self.gethr(0, tid)
response = requests.head(self.hosts[tid], headers=headers)
ttwid = response.cookies.get('ttwid')
headers.update({
'authority': self.hosts[tid].split('//')[-1],
'cookie': f'ttwid={ttwid}' if ttwid else ''
})
self.dyheaders = headers
# 初始化B站WBI密钥
self.bili_wbi_keys = None
self.bili_wbi_expire = 0
pass
def getName(self):
pass
return "直播"
def isVideoFormat(self, url):
pass
@@ -41,6 +42,7 @@ class Spider(Spider):
headers = [
{
# 特殊UA绕过B站风控 [^90^]
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0"
},
{
@@ -51,8 +53,7 @@ class Spider(Spider):
excepturl = 'https://www.baidu.com'
hosts = {
"huya": ["https://www.huya.com","https://mp.huya.com"],
"douyin": "https://live.douyin.com",
"huya": ["https://www.huya.com", "https://mp.huya.com"],
"douyu": "https://www.douyu.com",
"wangyi": "https://cc.163.com",
"bili": ["https://api.live.bilibili.com", "https://api.bilibili.com"]
@@ -60,7 +61,6 @@ class Spider(Spider):
referers = {
"huya": "https://live.cdn.huya.com",
"douyin": "https://live.douyin.com",
"douyu": "https://m.douyu.com",
"bili": "https://live.bilibili.com"
}
@@ -74,12 +74,8 @@ class Spider(Spider):
"bili": {
'Accept': '*/*',
'Icy-MetaData': '1',
'referer': referers['bili'],
'user-agent': headers[0]['User-Agent']
},
'douyin': {
'User-Agent': 'libmpv',
'Icy-MetaData': '1'
'referer': 'https://live.bilibili.com',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36'
},
'huya': {
'User-Agent': 'ExoPlayer',
@@ -92,42 +88,109 @@ class Spider(Spider):
}
}
def process_bili(self):
# WBI签名相关常量 [^30^]
MIXIN_KEY_ENC_TAB = [
46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,
33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,
61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,
36, 20, 34, 44, 52
]
def _get_bili_wbi_keys(self):
"""获取B站WBI密钥 [^30^]"""
try:
self.blfdata = self.fetch(
f'{self.hosts["bili"][0]}/room/v1/Area/getList?need_entrance=1&parent_id=0',
headers=self.gethr(0, 'bili')
# 检查缓存
if self.bili_wbi_keys and time.time() < self.bili_wbi_expire:
return self.bili_wbi_keys
# 从导航接口获取 - 使用特殊UA [^90^]
resp = self.fetch(
'https://api.bilibili.com/x/web-interface/nav',
headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
'Referer': 'https://www.bilibili.com/'
}
).json()
return ('bili', [{'key': 'cate', 'name': '分类',
'value': [{'n': i['name'], 'v': str(i['id'])}
for i in self.blfdata['data']]}])
if resp.get('code') != 0:
return None
img_url = resp['data']['wbi_img']['img_url']
sub_url = resp['data']['wbi_img']['sub_url']
# 提取文件名作为key
img_key = img_url.rsplit('/', 1)[1].split('.')[0]
sub_key = sub_url.rsplit('/', 1)[1].split('.')[0]
self.bili_wbi_keys = (img_key, sub_key)
self.bili_wbi_expire = time.time() + 86400 # 24小时过期
return self.bili_wbi_keys
except Exception as e:
print(f"获取B站WBI密钥失败: {e}")
return None
def _get_mixin_key(self, orig: str):
"""生成mixin_key [^30^]"""
return ''.join([orig[i] for i in self.MIXIN_KEY_ENC_TAB])[:32]
def _enc_wbi(self, params: dict):
"""WBI签名 [^30^]"""
keys = self._get_bili_wbi_keys()
if not keys:
return params
img_key, sub_key = keys
mixin_key = self._get_mixin_key(img_key + sub_key)
# 添加时间戳
params['wts'] = round(time.time())
# 排序参数
params = dict(sorted(params.items()))
# 过滤特殊字符
params = {
k: ''.join(filter(lambda c: c not in "!'()*", str(v)))
for k, v in params.items()
}
# 计算签名
query = urllib.parse.urlencode(params)
w_rid = hashlib.md5((query + mixin_key).encode()).hexdigest()
params['w_rid'] = w_rid
return params
def process_bili(self):
"""获取B站分类列表 - 使用WBI签名 [^30^]"""
try:
# 尝试获取分类列表 - 使用特殊UA和WBI签名
params = {'need_entrance': 1, 'parent_id': 0}
signed_params = self._enc_wbi(params)
data = self.fetch(
f'{self.hosts["bili"][0]}/room/v1/Area/getList',
params=signed_params,
headers=self.headers[0]
).json()
if data.get('code') == 0 and data.get('data'):
# 保存分类数据供后续使用
self.bili_areas = data['data']
return ('bili', [{'key': 'cate', 'name': '分类',
'value': [{'n': i['name'], 'v': str(i['id'])}
for i in data['data']]}])
return 'bili', None
except Exception as e:
print(f"bili处理错误: {e}")
return 'bili', None
def process_douyin(self):
try:
data = self.getpq(self.hosts['douyin'], headers=self.dyheaders)('script')
for i in data.items():
if 'categoryData' in i.text():
content = i.text()
start = content.find('{')
end = content.rfind('}') + 1
if start != -1 and end != -1:
json_str = content[start:end]
json_str = json_str.replace('\\"', '"')
try:
self.dyifdata = json.loads(json_str)
return ('douyin', [{'key': 'cate', 'name': '分类',
'value': [{'n': i['partition']['title'],
'v': f"{i['partition']['id_str']}@@{i['partition']['title']}"}
for i in self.dyifdata['categoryData']]}])
except json.JSONDecodeError as e:
print(f"douyin解析错误: {e}")
return 'douyin', None
except Exception as e:
print(f"douyin请求或处理错误: {e}")
return 'douyin', None
# 使用默认分类
return 'bili', [{'key': 'cate', 'name': '分类',
'value': [{'n': '网游', 'v': '2'}, {'n': '手游', 'v': '3'},
{'n': '单机', 'v': '6'}, {'n': '娱乐', 'v': '1'},
{'n': '电台', 'v': '5'}, {'n': '虚拟主播', 'v': '9'},
{'n': '生活', 'v': '10'}, {'n': '知识', 'v': '11'},
{'n': '赛事', 'v': '13'}]}]
def process_douyu(self):
try:
@@ -146,9 +209,9 @@ class Spider(Spider):
result = {}
cateManual = {
"虎牙": "huya",
"抖音": "douyin",
"斗鱼": "douyu",
"网易": "wangyi"
"网易": "wangyi",
"B站": "bili"
}
classes = []
filters = {
@@ -157,10 +220,9 @@ class Spider(Spider):
{'n': '娱乐', 'v': '8'}, {'n': '手游', 'v': '3'}]}]
}
with ThreadPoolExecutor(max_workers=3) as executor:
with ThreadPoolExecutor(max_workers=2) as executor:
futures = {
executor.submit(self.process_bili): 'bili',
executor.submit(self.process_douyin): 'douyin',
executor.submit(self.process_douyu): 'douyu'
}
@@ -195,8 +257,6 @@ class Spider(Spider):
vdata, pagecount = self.biliContent(tid, pg, filter, extend, vdata)
elif 'huya' in tid:
vdata, pagecount = self.huyaContent(tid, pg, filter, extend, vdata)
elif 'douyin' in tid:
vdata, pagecount = self.douyinContent(tid, pg, filter, extend, vdata)
elif 'douyu' in tid:
vdata, pagecount = self.douyuContent(tid, pg, filter, extend, vdata)
result['list'] = vdata
@@ -223,39 +283,102 @@ class Spider(Spider):
return vdata, 9999
def biliContent(self, tid, pg, filter, extend, vdata):
if extend.get('cate') and pg == '1' and 'click' not in tid:
for i in self.blfdata['data']:
if str(i['id']) == extend['cate']:
for j in i['list']:
v = self.buildvod(
vod_id=f"click_{tid}@@{i['id']}@@{j['id']}",
vod_name=j.get('name'),
vod_pic=j.get('pic'),
vod_tag=1,
style={"type": "oval", "ratio": 1}
)
vdata.append(v)
return vdata, 1
else:
path = f'/xlive/web-interface/v1/second/getListByArea?platform=web&sort=online&page_size=30&page={pg}'
"""B站分类内容 - 使用WBI签名绕过风控 [^30^][^90^]"""
try:
# 分类列表 - 显示子分类
if extend.get('cate') and pg == '1' and 'click' not in tid:
# 从已保存的分类数据中找到对应分类的子分类
if hasattr(self, 'bili_areas'):
for area in self.bili_areas:
if str(area['id']) == extend['cate']:
for sub_area in area.get('list', []):
v = self.buildvod(
vod_id=f"click_{tid}@@{extend['cate']}@@{sub_area['id']}",
vod_name=sub_area.get('name'),
vod_pic=sub_area.get('pic'),
vod_tag=1,
style={"type": "oval", "ratio": 1}
)
vdata.append(v)
return vdata, 1
# 如果没有找到子分类，直接返回空，让用户进入房间列表
return vdata, 1
# 房间列表 - 使用getList接口并添加WBI签名 [^30^]
if 'click' in tid:
# 子分类房间
ids = tid.split('_')[1].split('@@')
tid = ids[0]
path = f'/xlive/web-interface/v1/second/getList?platform=web&parent_area_id={ids[1]}&area_id={ids[-1]}&sort_type=&page={pg}'
data = self.fetch(f'{self.hosts[tid][0]}{path}', headers=self.gethr(0, tid)).json()
for i in data['data']['list']:
if i.get('roomid'):
data = self.buildvod(
f"{tid}@@{i['roomid']}",
i.get('title'),
i.get('cover'),
i.get('watched_show', {}).get('text_large'),
0,
i.get('uname'),
style={"type": "rect", "ratio": 1.33}
)
vdata.append(data)
return vdata, 9999
parent_area_id = ids[1]
area_id = ids[2]
else:
# 默认使用分类ID作为parent_area_id，area_id为0表示该分类下所有
parent_area_id = extend.get('cate', '2') # 默认网游
area_id = 0
# 构建请求参数并添加WBI签名 [^30^]
params = {
'parent_area_id': parent_area_id,
'area_id': area_id,
'page': pg,
'platform': 'web',
'sort_type': 'online' # 按热度排序
}
signed_params = self._enc_wbi(params)
# 调用getList接口
api_url = f'{self.hosts[tid][0]}/xlive/web-interface/v1/second/getList'
data = self.fetch(api_url, params=signed_params, headers=self.headers[0]).json()
# 如果WBI签名失败，尝试不带签名
if data.get('code') == -352:
print("WBI签名失败，尝试无签名请求...")
params = {
'parent_area_id': parent_area_id,
'area_id': area_id,
'page': pg,
'platform': 'web',
'sort_type': 'online'
}
data = self.fetch(api_url, params=params, headers=self.headers[0]).json()
if data.get('code') == 0:
room_list = data.get('data', {}).get('list', [])
for room in room_list:
if room.get('roomid'):
# 处理在线人数显示
online = room.get('online', 0)
if online > 10000:
online_str = f"{online / 10000:.1f}万"
else:
online_str = str(online)
v = self.buildvod(
f"{tid}@@{room['roomid']}",
room.get('title', '未知标题'),
room.get('cover') or room.get('system_cover'),
f"{online_str}人",
0,
room.get('uname', ''),
style={"type": "rect", "ratio": 1.33}
)
vdata.append(v)
# 检查是否有更多数据
has_more = data.get('data', {}).get('has_more', 0)
if not has_more:
pagecount = int(pg)
else:
pagecount = 9999
else:
print(f"B站API返回错误: {data.get('message', '未知错误')} (code: {data.get('code')})")
pagecount = 1
return vdata, pagecount
except Exception as e:
print(f"B站内容获取错误: {e}")
return vdata, 1
def huyaContent(self, tid, pg, filter, extend, vdata):
if extend.get('cate') and pg == '1' and 'click' not in tid:
@@ -295,44 +418,6 @@ class Spider(Spider):
vdata.append(v)
return vdata, 9999
def douyinContent(self, tid, pg, filter, extend, vdata):
if extend.get('cate') and pg == '1' and 'click' not in tid:
ids = extend.get('cate').split('@@')
for i in self.dyifdata['categoryData']:
c = i['partition']
if c['id_str'] == ids[0] and c['title'] == ids[1]:
vlist = i['sub_partition'].copy()
vlist.insert(0, {'partition': c})
for j in vlist:
j = j['partition']
v = self.buildvod(
vod_id=f"click_{tid}@@{j['id_str']}@@{j['type']}",
vod_name=j.get('title'),
vod_pic='https://p3-pc-weboff.byteimg.com/tos-cn-i-9r5gewecjs/pwa_v3/512x512-1.png',
vod_tag=1,
style={"type": "oval", "ratio": 1}
)
vdata.append(v)
return vdata, 1
else:
path = f'/webcast/web/partition/detail/room/?aid=6383&app_name=douyin_web&live_id=1&device_platform=web&count=15&offset={(int(pg) - 1) * 15}&partition=720&partition_type=1'
if 'click' in tid:
ids = tid.split('_')[1].split('@@')
tid = ids[0]
path = f'/webcast/web/partition/detail/room/?aid=6383&app_name=douyin_web&live_id=1&device_platform=web&count=15&offset={(int(pg) - 1) * 15}&partition={ids[1]}&partition_type={ids[-1]}&req_from=2'
data = self.fetch(f'{self.hosts[tid]}{path}', headers=self.dyheaders).json()
for i in data['data']['data']:
v = self.buildvod(
vod_id=f"{tid}@@{i['web_rid']}",
vod_name=i['room'].get('title'),
vod_pic=i['room']['cover'].get('url_list')[0],
vod_year=i.get('user_count_str'),
vod_remarks=i['room']['owner'].get('nickname'),
style={"type": "rect", "ratio": 1.33}
)
vdata.append(v)
return vdata, 9999
def douyuContent(self, tid, pg, filter, extend, vdata):
if extend.get('cate') and pg == '1' and 'click' not in tid:
for i in self.dyufdata['data']['cate2Info']:
@@ -375,8 +460,6 @@ class Spider(Spider):
vod = self.biliDetail(ids)
elif ids[0] == 'huya':
vod = self.huyaDetail(ids)
elif ids[0] == 'douyin':
vod = self.douyinDetail(ids)
elif ids[0] == 'douyu':
vod = self.douyuDetail(ids)
return {'list': [vod]}
@@ -425,186 +508,550 @@ class Spider(Spider):
return self.handle_exception(e)
def biliDetail(self, ids):
"""
B站直播详情 - 使用playUrl接口获取多清晰度
"""
try:
vdata = self.fetch(
f'{self.hosts[ids[0]][0]}/xlive/web-room/v1/index/getInfoByRoom?room_id={ids[1]}&wts={int(time.time())}',
headers=self.gethr(0, ids[0])).json()
v = vdata['data']['room_info']
room_id = ids[1]
# 获取房间信息
info_res = self.fetch(
f'{self.hosts["bili"][0]}/room/v1/Room/get_info?room_id={room_id}',
headers=self.headers[0]
).json()
if info_res.get('code') != 0:
return self.handle_exception(Exception("获取房间信息失败"))
room_info = info_res['data']
title = room_info.get('title', 'B站直播')
vod = self.buildvod(
vod_name=v.get('title'),
type_name=v.get('parent_area_name') + '/' + v.get('area_name'),
vod_remarks=v.get('tags'),
vod_play_from=v.get('title'),
vod_name=title,
type_name=f"{room_info.get('parent_area_name', '')}/{room_info.get('area_name', '')}",
vod_director=room_info.get('uname', ''),
vod_remarks=f"在线{room_info.get('online', 0)}人"
)
data = self.fetch(
f'{self.hosts[ids[0]][0]}/xlive/web-room/v2/index/getRoomPlayInfo?room_id={ids[1]}&protocol=0%2C1&format=0%2C1%2C2&codec=0%2C1&platform=web',
headers=self.gethr(0, ids[0])).json()
vdnams = data['data']['playurl_info']['playurl']['g_qn_desc']
all_accept_qns = []
streams = data['data']['playurl_info']['playurl']['stream']
for stream in streams:
for format_item in stream['format']:
for codec in format_item['codec']:
if 'accept_qn' in codec:
all_accept_qns.append(codec['accept_qn'])
max_accept_qn = max(all_accept_qns, key=len) if all_accept_qns else []
quality_map = {
item['qn']: item['desc']
for item in vdnams
}
quality_names = [f"{quality_map.get(qn)}${ids[0]}@@{ids[1]}@@{qn}" for qn in max_accept_qn]
vod['vod_play_url'] = "#".join(quality_names)
# 获取播放地址信息
play_res = self.fetch(
f'{self.hosts["bili"][0]}/room/v1/Room/playUrl?cid={room_id}&qn=10000&platform=web',
headers={
**self.headers[0],
'Referer': 'https://live.bilibili.com/',
'Origin': 'https://live.bilibili.com'
}
).json()
if play_res.get('code') != 0:
return self.handle_exception(Exception("获取播放地址失败"))
play_data = play_res['data']
accept_quality = play_data.get('accept_quality', ['10000', '400', '250', '150'])
quality_desc = {item['qn']: item['desc'] for item in play_data.get('quality_description', [])}
# 构建清晰度列表
qualities = []
for qn in sorted([int(q) for q in accept_quality], reverse=True):
desc = quality_desc.get(qn, f'清晰度{qn}')
qualities.append(f"{desc}$bili@@{room_id}@@{qn}")
vod['vod_play_from'] = 'B站直播'
vod['vod_play_url'] = '#'.join(qualities)
return vod
except Exception as e:
print(f"B站详情错误: {e}")
return self.handle_exception(e)
def huyaDetail(self, ids):
"""
虎牙播放详情 - 参考最新三合一.js重构
支持多线路多清晰度选择
核心算法：通过房间信息API获取uid、streamName和rateArray，为每个清晰度生成签名URL
清晰度说明：
- 蓝光8M/6M/4M/10M = 8000/6000/4000/10000 kbps = 1080P+
- 蓝光 = 3000 kbps = 1080P
- 超清 = 2000 kbps = 1080P (官方标准)
- 高清 = 1200 kbps = 720P
- 标清/流畅 = 500-800 kbps = 480P/540P
"""
try:
vdata = self.fetch(f'{self.hosts[ids[0]][1]}/cache.php?m=Live&do=profileRoom&roomid={ids[1]}',
headers=self.headers[0]).json()
v = vdata['data']['liveData']
room_id = ids[1]
# 1. 获取房间信息
api_url = f'{self.hosts[ids[0]][1]}/cache.php?m=Live&do=profileRoom&roomid={room_id}'
res = self.fetch(api_url, headers=self.headers[0])
if res.status_code != 200:
return self.handle_exception(Exception(f"API请求失败: {res.status_code}"))
data = res.json()
if not data or not data.get('data'):
return self.handle_exception(Exception("房间数据为空"))
room_data = data['data']
# 2. 提取关键信息
uid = room_data.get('profileInfo', {}).get('uid')
stream_info = room_data.get('stream', {})
live_data = room_data.get('liveData', {})
if not uid:
return self.handle_exception(Exception("缺少uid"))
# 3. 获取streamName和码率信息
base_stream_list = stream_info.get('baseSteamInfoList', [])
if not base_stream_list:
return self.handle_exception(Exception("无直播流信息"))
# 获取第一个CDN的streamName作为基准
base_stream = base_stream_list[0]
stream_name = base_stream.get('sStreamName')
if not stream_name:
return self.handle_exception(Exception("无法获取streamName"))
# 4. 构建VOD对象
vod = self.buildvod(
vod_name=v.get('introduction'),
type_name=v.get('gameFullName'),
vod_director=v.get('nick'),
vod_remarks=v.get('contentIntro'),
vod_name=live_data.get('introduction', '虎牙直播'),
type_name=live_data.get('gameFullName', ''),
vod_director=live_data.get('nick', ''),
vod_remarks=live_data.get('contentIntro', ''),
)
data = dict(reversed(list(vdata['data']['stream'].items())))
names = []
plist = []
for stream_type, stream_data in data.items():
if isinstance(stream_data, dict) and 'multiLine' in stream_data and 'rateArray' in stream_data:
names.append(f"线路{len(names) + 1}")
qualities = sorted(
stream_data['rateArray'],
key=lambda x: (x['iBitRate'], x['sDisplayName']),
reverse=True
# 5. 获取所有CDN线路
cdn_list = []
for stream in base_stream_list:
cdn_type = stream.get('sCdnType', 'AL')
flv_url = stream.get('sFlvUrl', '')
hls_url = stream.get('sHlsUrl', '')
stream_name_cdn = stream.get('sStreamName', stream_name)
if flv_url:
cdn_list.append({
'cdn': cdn_type,
'flv_base': flv_url,
'hls_base': hls_url,
'stream_name': stream_name_cdn,
'priority': stream.get('iWebPriorityRate', 0)
})
# 按优先级排序
cdn_list.sort(key=lambda x: x['priority'], reverse=True)
# 6. 获取清晰度列表 (rateArray)
rate_array = stream_info.get('rateArray', [])
# 如果没有rateArray，尝试从vMultiStreamInfo获取
if not rate_array and 'vMultiStreamInfo' in room_data:
rate_array = room_data['vMultiStreamInfo']
# 如果仍然没有，使用默认清晰度（按虎牙官方标准）
if not rate_array:
rate_array = [
{'sDisplayName': '蓝光4M', 'iBitRate': 4000},
{'sDisplayName': '蓝光', 'iBitRate': 3000},
{'sDisplayName': '超清', 'iBitRate': 2000}, # 2000kbps = 1080P
{'sDisplayName': '高清', 'iBitRate': 1200}, # 1200kbps = 720P
{'sDisplayName': '流畅', 'iBitRate': 500}
]
# 过滤和排序清晰度
# 虎牙的rateArray中，iBitRate就是码率值，sDisplayName是显示名称
# 需要确保：超清=2000kbps(1080P)，高清=1200kbps(720P)
filtered_rates = []
seen_bitrates = set()
for rate in rate_array:
bit_rate = rate.get('iBitRate', 0)
name = rate.get('sDisplayName', '')
# 跳过重复的码率
if bit_rate in seen_bitrates:
continue
# 修正清晰度名称，确保符合虎牙标准
# 2000kbps应该是超清(1080P)，不是高清
if bit_rate == 2000 and ('高清' in name or '720' in name):
name = '超清' # 强制修正为超清
elif bit_rate == 1200 and ('标清' in name or '480' in name):
name = '高清' # 1200kbps对应高清
elif bit_rate == 2000 and name == '原画':
name = '超清' # 修正原画为超清
seen_bitrates.add(bit_rate)
filtered_rates.append({
'sDisplayName': name,
'iBitRate': bit_rate
})
# 按码率从高到低排序
sorted_rates = sorted(filtered_rates, key=lambda x: x['iBitRate'], reverse=True)
# 7. 为每个CDN生成各清晰度的播放URL
play_lines = []
line_names = []
for cdn_idx, cdn in enumerate(cdn_list[:3]): # 最多取3个CDN
cdn_name = cdn['cdn']
line_names.append(f"线路{cdn_idx + 1}({cdn_name})")
qualities = []
for rate in sorted_rates:
quality_name = rate['sDisplayName']
bit_rate = rate['iBitRate']
# 生成该清晰度的URL
quality_url = self._generate_huya_play_url(
cdn, uid, stream_name, bit_rate
)
cdn_urls = []
for cdn in stream_data['multiLine']:
quality_urls = []
for quality in qualities:
quality_name = quality['sDisplayName']
bit_rate = quality['iBitRate']
base_url = cdn['url']
if bit_rate > 0:
if '.m3u8' in base_url:
new_url = base_url.replace(
'ratio=2000',
f'ratio={bit_rate}'
)
else:
new_url = base_url.replace(
'imgplus.flv',
f'imgplus_{bit_rate}.flv'
)
else:
new_url = base_url
quality_urls.extend([quality_name, new_url])
encoded_urls = self.e64(json.dumps(quality_urls))
cdn_urls.append(f"{cdn['cdnType']}${ids[0]}@@{encoded_urls}")
if cdn_urls:
plist.append('#'.join(cdn_urls))
vod['vod_play_from'] = "$$$".join(names)
vod['vod_play_url'] = "$$$".join(plist)
qualities.extend([quality_name, quality_url])
# 编码该线路的所有清晰度
encoded_qualities = self.e64(json.dumps(qualities))
play_lines.append(f"{live_data.get('introduction', '直播')}${ids[0]}@@{encoded_qualities}")
# 8. 构建播放数据
vod['vod_play_from'] = "$$$".join(line_names)
vod['vod_play_url'] = "$$$".join(play_lines)
return vod
except Exception as e:
return self.handle_exception(e)
def douyinDetail(self, ids):
url = f'{self.hosts[ids[0]]}/webcast/room/web/enter/?aid=6383&app_name=douyin_web&live_id=1&device_platform=web&enter_from=web_live&web_rid={ids[1]}&room_id_str=&enter_source=&Room-Enter-User-Login-Ab=0&is_need_double_stream=false&cookie_enabled=true&screen_width=1980&screen_height=1080&browser_language=zh-CN&browser_platform=Win32&browser_name=Edge&browser_version=125.0.0.0'
data = self.fetch(url, headers=self.dyheaders).json()
try:
vdata = data['data']['data'][0]
vod = self.buildvod(
vod_name=vdata['title'],
vod_remarks=vdata['user_count_str'],
)
resolution_data = vdata['stream_url']['live_core_sdk_data']['pull_data']['options']['qualities']
stream_json = vdata['stream_url']['live_core_sdk_data']['pull_data']['stream_data']
stream_json = json.loads(stream_json)
available_types = []
if any(sdk_key in stream_json['data'] and 'main' in stream_json['data'][sdk_key] for sdk_key in
stream_json['data']):
available_types.append('main')
if any(sdk_key in stream_json['data'] and 'backup' in stream_json['data'][sdk_key] for sdk_key in
stream_json['data']):
available_types.append('backup')
plist = []
for line_type in available_types:
format_arrays = {'flv': [], 'hls': [], 'lls': []}
qualities = sorted(resolution_data, key=lambda x: x['level'], reverse=True)
for quality in qualities:
sdk_key = quality['sdk_key']
if sdk_key in stream_json['data'] and line_type in stream_json['data'][sdk_key]:
stream_info = stream_json['data'][sdk_key][line_type]
if stream_info.get('flv'):
format_arrays['flv'].extend([quality['name'], stream_info['flv']])
if stream_info.get('hls'):
format_arrays['hls'].extend([quality['name'], stream_info['hls']])
if stream_info.get('lls'):
format_arrays['lls'].extend([quality['name'], stream_info['lls']])
format_urls = []
for format_name, url_array in format_arrays.items():
if url_array:
encoded_urls = self.e64(json.dumps(url_array))
format_urls.append(f"{format_name}${ids[0]}@@{encoded_urls}")
if format_urls:
plist.append('#'.join(format_urls))
names = ['线路1', '线路2'][:len(plist)]
vod['vod_play_from'] = "$$$".join(names)
vod['vod_play_url'] = "$$$".join(plist)
return vod
except Exception as e:
return self.handle_exception(e)
def _generate_huya_play_url(self, cdn, uid, stream_name, bit_rate):
"""
生成虎牙播放URL，参考最新三合一.js算法
关键：ratio参数必须正确设置为iBitRate值（如2000、4000等）
"""
# 基础URL构建
flv_base = cdn['flv_base']
stream = cdn['stream_name']
# 生成时间戳和签名参数
timestamp = int(time.time())
seqid = f"{uid}{timestamp}"
ss = hashlib.md5(f"{seqid}|huya_adr|102".encode()).hexdigest()
ws_time = hex(timestamp + 21600)[2:] # 16进制，有效期6小时
# 计算wsSecret
ws_secret = hashlib.md5(
f"DWq8BcJ3h6DJt6TY_{uid}_{stream_name}_{ss}_{ws_time}".encode()
).hexdigest()
# 构建基础URL
base_url = f"{flv_base}/{stream}.flv"
# 关键修复：ratio参数直接使用iBitRate值
# 超清=2000，高清=1200，蓝光=3000/4000/6000/8000
if bit_rate > 0:
ratio_param = f"ratio={bit_rate}"
else:
# 原画/0码率时，使用默认2000或从URL推断
ratio_param = "ratio=2000"
# 构建完整URL
play_url = (
f"{base_url}?{ratio_param}&wsSecret={ws_secret}&wsTime={ws_time}"
f"&ctype=huya_adr&seqid={seqid}&uid={uid}"
f"&fs=bgct&ver=1&t=102"
)
return play_url
def douyuDetail(self, ids):
headers = self.gethr(0, zr=f'{self.hosts[ids[0]]}/{ids[1]}')
"""
斗鱼播放详情 - 参考最新三合一.js重构
核心算法：设备ID生成 -> 获取加密密钥 -> 计算签名 -> 获取播放地址
修复：切换分辨率只能播放1秒的问题
方案：存储房间号和码率信息，在playerContent中实时获取对应码率的URL
"""
try:
data = self.fetch(f'{self.hosts[ids[0]]}/betard/{ids[1]}', headers=headers).json()
vname = data['room']['room_name']
channel = ids[1]
headers = self.gethr(0, zr=f'{self.hosts[ids[0]]}/{channel}')
# 1. 初始化会话和设备ID (参考JS中的initialize和setupDeviceId)
session = {}
# 请求首页获取Cookie
try:
home_res = self.fetch(f'{self.hosts[ids[0]]}/{channel}', headers=headers)
if home_res.headers.get('Set-Cookie'):
cookie_str = home_res.headers.get('Set-Cookie')
# 解析dy_did
did_match = re.search(r'dy_did=([a-f0-9]{32})', cookie_str)
if did_match:
device_id = did_match.group(1)
else:
device_id = self._generate_random_hex(32)
else:
device_id = self._generate_random_hex(32)
except:
device_id = self._generate_random_hex(32)
session['dy_did'] = device_id
session['mantine-color-scheme-value'] = 'light'
# 2. 获取房间基本信息
betard_res = self.fetch(f'{self.hosts[ids[0]]}/betard/{channel}', headers=headers).json()
if not betard_res or not betard_res.get('room'):
return self.handle_exception(Exception("获取房间信息失败"))
room_info = betard_res['room']
vname = room_info.get('room_name', '斗鱼直播')
vod = self.buildvod(
vod_name=vname,
vod_remarks=data['room'].get('second_lvl_name'),
vod_director=data['room'].get('nickname'),
vod_remarks=room_info.get('second_lvl_name', ''),
vod_director=room_info.get('nickname', ''),
)
vdata = self.fetch(f'{self.hosts[ids[0]]}/swf_api/homeH5Enc?rids={ids[1]}', headers=headers).json()
json_body = vdata['data']
json_body = {"html": self.douyu_text(json_body[f'room{ids[1]}']), "rid": ids[1]}
sign = self.post('http://alive.nsapps.cn/api/AllLive/DouyuSign', json=json_body, headers=self.headers[1]).json()['data']
body = f'{sign}&cdn=&rate=-1&ver=Douyu_223061205&iar=1&ive=1&hevc=0&fa=0'
body=self.params_to_json(body)
nubdata = self.post(f'{self.hosts[ids[0]]}/lapi/live/getH5Play/{ids[1]}', data=body, headers=headers).json()
plist = []
names = []
for i,x in enumerate(nubdata['data']['cdnsWithName']):
names.append(f'线路{i+1}')
d = {'sign': sign, 'cdn': x['cdn'], 'id': ids[1]}
plist.append(
f'{vname}${ids[0]}@@{self.e64(json.dumps(d))}@@{self.e64(json.dumps(nubdata["data"]["multirates"]))}')
vod['vod_play_from'] = "$$$".join(names)
vod['vod_play_url'] = "$$$".join(plist)
# 3. 获取安全密钥 (参考JS中的getSecurityKey)
sec_url = f"{self.hosts[ids[0]]}/wgapi/livenc/liveweb/websec/getEncryption?did={device_id}"
sec_res = self.fetch(sec_url, headers=headers).json()
if not sec_res or sec_res.get('error') != 0:
return self.handle_exception(Exception("获取加密密钥失败"))
security_data = sec_res['data']
secret_key = security_data.get('key')
random_str = security_data.get('rand_str')
enc_time = security_data.get('enc_time', 1)
enc_data = security_data.get('enc_data')
# 4. 计算签名 (参考JS中的computeSignature)
current_time = int(time.time())
# 迭代计算MD5
current = random_str
for _ in range(enc_time):
current = hashlib.md5(f"{current}{secret_key}".encode()).hexdigest()
signature = hashlib.md5(f"{current}{secret_key}{channel}{current_time}".encode()).hexdigest()
# 5. 请求播放地址 (参考JS中的requestStreamData)
play_payload = {
'enc_data': enc_data,
'tt': str(current_time),
'did': device_id,
'auth': signature,
'cdn': '',
'rate': '',
'hevc': '0',
'fa': '0',
'ive': '0'
}
play_api = f"{self.hosts[ids[0]]}/lapi/live/getH5PlayV1/{channel}"
# 构建请求头带Cookie
play_headers = headers.copy()
cookie_str = '; '.join([f"{k}={v}" for k, v in session.items()])
play_headers['Cookie'] = cookie_str
play_headers['Content-Type'] = 'application/x-www-form-urlencoded'
play_res = requests.post(play_api, data=play_payload, headers=play_headers, timeout=10).json()
if not play_res or play_res.get('error') != 0:
# 尝试旧版API
play_res = self._try_legacy_douyu_api(channel, device_id, signature, current_time, play_headers)
if not play_res:
return self.handle_exception(Exception("获取播放地址失败"))
stream_info = play_res.get('data', {})
# 6. 检查并更新设备ID (参考JS中的checkAndUpdateDeviceId)
rtmp_live = stream_info.get('rtmp_live', '')
if rtmp_live:
did_match = re.search(r'did=([a-f0-9]{32})', rtmp_live)
if did_match and did_match.group(1) != device_id:
device_id = did_match.group(1)
session['dy_did'] = device_id
# 重新请求
play_payload['did'] = device_id
play_res = requests.post(play_api, data=play_payload, headers=play_headers, timeout=10).json()
if play_res and play_res.get('error') == 0:
stream_info = play_res.get('data', {})
# 7. 提取播放URL和多码率信息
stream_url = None
if stream_info.get('rtmp_url') and stream_info.get('rtmp_live'):
stream_url = f"{stream_info['rtmp_url']}/{stream_info['rtmp_live']}"
elif stream_info.get('hls_url'):
stream_url = stream_info['hls_url']
if not stream_url:
return self.handle_exception(Exception("无法获取播放地址"))
# 8. 构建多码率选项
multirates = stream_info.get('multirates', [])
# 关键修复：存储房间号和码率信息，而不是直接存储URL
# 这样在切换清晰度时可以重新获取对应码率的签名URL
qualities = []
if multirates:
# 按码率排序
sorted_rates = sorted(multirates, key=lambda x: x.get('bit', 0), reverse=True)
for rate in sorted_rates:
bit_rate = rate.get('rate', -1)
name = rate.get('name', f"{bit_rate}P")
# 存储格式：码率值，用于playerContent中重新获取URL
# 使用特殊标记#来区分这是码率值而不是URL
qualities.extend([name, f"#{bit_rate}"])
else:
# 只有原画
qualities = ['原画', '#-1']
# 同时存储房间号和设备信息，用于重新获取URL
# 格式：房间号|设备ID|签名信息（base64编码）
session_info = {
'channel': channel,
'device_id': device_id,
'secret_key': secret_key,
'random_str': random_str,
'enc_time': enc_time,
'enc_data': enc_data
}
encoded_session = self.e64(json.dumps(session_info))
# 9. 构建播放数据
# vod_play_url格式：房间名$平台@@base64(清晰度列表)@@base64(会话信息)
encoded_qualities = self.e64(json.dumps(qualities))
vod['vod_play_from'] = '斗鱼直播'
vod['vod_play_url'] = f"{vname}${ids[0]}@@{encoded_qualities}@@{encoded_session}"
return vod
except Exception as e:
return self.handle_exception(e)
def _generate_random_hex(self, length):
"""生成随机十六进制字符串"""
hex_chars = '0123456789abcdef'
return ''.join(random.choice(hex_chars) for _ in range(length))
def douyu_text(self, text):
function_positions = [m.start() for m in re.finditer('function', text)]
total_functions = len(function_positions)
if total_functions % 2 == 0:
target_index = total_functions // 2 + 1
else:
target_index = (total_functions - 1) // 2 + 1
if total_functions >= target_index:
cut_position = function_positions[target_index - 1]
ctext = text[4:cut_position]
return re.sub(r'eval\(strc\)\([\w\d,]+\)', 'strc', ctext)
return text
def _try_legacy_douyu_api(self, channel, device_id, signature, timestamp, headers):
"""尝试使用旧版API获取播放地址"""
try:
legacy_payload = {
'did': device_id,
'tt': str(timestamp),
'sign': signature,
'cdn': '',
'rate': '-1',
'ver': 'Douyu_223061205',
'iar': '1',
'ive': '1',
'hevc': '0',
'fa': '0'
}
legacy_api = f"https://www.douyu.com/lapi/live/getH5Play/{channel}"
res = requests.post(legacy_api, data=legacy_payload, headers=headers, timeout=10)
return res.json() if res.status_code == 200 else None
except:
return None
def _get_douyu_play_url(self, channel, device_id, secret_key, random_str, enc_time, enc_data, rate):
"""
获取斗鱼指定码率的播放URL（带签名）
用于切换清晰度时重新获取URL
"""
try:
current_time = int(time.time())
# 重新计算签名
current = random_str
for _ in range(enc_time):
current = hashlib.md5(f"{current}{secret_key}".encode()).hexdigest()
signature = hashlib.md5(f"{current}{secret_key}{channel}{current_time}".encode()).hexdigest()
# 构建请求
play_payload = {
'enc_data': enc_data,
'tt': str(current_time),
'did': device_id,
'auth': signature,
'cdn': '',
'rate': str(rate) if rate > 0 else '',
'hevc': '0',
'fa': '0',
'ive': '0'
}
play_api = f"https://www.douyu.com/lapi/live/getH5PlayV1/{channel}"
headers = {
'User-Agent': self.headers[0]['User-Agent'],
'Referer': f'https://www.douyu.com/{channel}',
'Origin': 'https://www.douyu.com',
'Cookie': f'dy_did={device_id}; mantine-color-scheme-value=light',
'Content-Type': 'application/x-www-form-urlencoded'
}
play_res = requests.post(play_api, data=play_payload, headers=headers, timeout=10).json()
if not play_res or play_res.get('error') != 0:
# 尝试旧版API
return self._get_douyu_play_url_legacy(channel, device_id, signature, current_time, rate)
stream_info = play_res.get('data', {})
# 检查设备ID是否匹配
if stream_info.get('rtmp_live'):
did_match = re.search(r'did=([a-f0-9]{32})', stream_info['rtmp_live'])
if did_match and did_match.group(1) != device_id:
# 设备ID不匹配，使用新设备ID重新获取
return self._get_douyu_play_url(channel, did_match.group(1), secret_key, random_str, enc_time, enc_data, rate)
if stream_info.get('rtmp_url') and stream_info.get('rtmp_live'):
return f"{stream_info['rtmp_url']}/{stream_info['rtmp_live']}"
elif stream_info.get('hls_url'):
return stream_info['hls_url']
return None
except Exception as e:
print(f"获取斗鱼播放URL失败: {e}")
return None
def _get_douyu_play_url_legacy(self, channel, device_id, signature, timestamp, rate):
"""使用旧版API获取斗鱼播放URL"""
try:
legacy_payload = {
'did': device_id,
'tt': str(timestamp),
'sign': signature,
'cdn': '',
'rate': str(rate) if rate > 0 else '-1',
'ver': 'Douyu_223061205',
'iar': '1',
'ive': '1',
'hevc': '0',
'fa': '0'
}
legacy_api = f"https://www.douyu.com/lapi/live/getH5Play/{channel}"
headers = {
'User-Agent': self.headers[0]['User-Agent'],
'Referer': f'https://www.douyu.com/{channel}',
'Cookie': f'dy_did={device_id}',
'Content-Type': 'application/x-www-form-urlencoded'
}
res = requests.post(legacy_api, data=legacy_payload, headers=headers, timeout=10)
if res.status_code == 200:
data = res.json()
if data.get('error') == 0:
stream_info = data.get('data', {})
if stream_info.get('rtmp_url') and stream_info.get('rtmp_live'):
return f"{stream_info['rtmp_url']}/{stream_info['rtmp_live']}"
return None
except:
return None
def searchContent(self, key, quick, pg="1"):
pass
@@ -613,12 +1060,12 @@ class Spider(Spider):
try:
ids = id.split('@@')
p = 1
if ids[0] in ['wangyi', 'douyin','huya']:
if ids[0] in ['wangyi']:
p, url = 0, json.loads(self.d64(ids[1]))
elif ids[0] == 'bili':
p, url = self.biliplay(ids)
elif ids[0] == 'huya':
p, url = 0, json.loads(self.d64(ids[1]))
p, url = self.huyaplay(ids)
elif ids[0] == 'douyu':
p, url = self.douyuplay(ids)
return {'parse': p, 'url': url, 'header': self.playheaders[ids[0]]}
@@ -626,69 +1073,116 @@ class Spider(Spider):
return {'parse': 1, 'url': self.excepturl, 'header': self.headers[0]}
def biliplay(self, ids):
"""
B站播放解析 - 使用playUrl接口获取指定清晰度直播流
ids: [平台, 房间号, 清晰度qn]
支持多线路返回
"""
try:
data = self.fetch(
f'{self.hosts[ids[0]][0]}/xlive/web-room/v2/index/getRoomPlayInfo?room_id={ids[1]}&protocol=0,1&format=0,2&codec=0&platform=web&qn={ids[2]}',
headers=self.gethr(0, ids[0])).json()
room_id = ids[1]
qn = ids[2] if len(ids) > 2 else '10000'
# 使用playUrl接口获取直播流
play_url = f'{self.hosts["bili"][0]}/room/v1/Room/playUrl?cid={room_id}&qn={qn}&platform=web'
data = self.fetch(play_url, headers={
**self.headers[0],
'Referer': 'https://live.bilibili.com/',
'Origin': 'https://live.bilibili.com'
}).json()
if data.get('code') != 0:
return 1, self.excepturl
play_data = data['data']
durl_list = play_data.get('durl', [])
if not durl_list:
return 1, self.excepturl
# 构建多线路结果 [线路1, URL1, 线路2, URL2, ...]
urls = []
line_index = 1
for stream in data['data']['playurl_info']['playurl']['stream']:
for format_item in stream['format']:
for codec in format_item['codec']:
for url_info in codec['url_info']:
full_url = f"{url_info['host']}/{codec['base_url'].lstrip('/')}{url_info['extra']}"
urls.extend([f"线路{line_index}", full_url])
line_index += 1
for idx, item in enumerate(durl_list, 1):
url = item.get('url')
if url:
urls.extend([f'线路{idx}', url])
# 如果只有一条线路，直接返回URL
if len(urls) == 2:
return 0, urls[1] # 直接返回URL字符串
return 0, urls
except Exception as e:
print(f"B站播放错误: {e}")
return 1, self.excepturl
def huyaplay(self, ids):
"""
虎牙播放解析 - 返回所有清晰度选项供用户选择
ids[1] 格式: base64编码的 [清晰度名称1, URL1, 清晰度名称2, URL2, ...]
"""
try:
# ids[1] 是编码后的播放地址列表 [名称1, URL1, 名称2, URL2, ...]
decoded = json.loads(self.d64(ids[1]))
# decoded 是一个列表，奇数索引是名称，偶数索引是URL
return 0, decoded
except Exception as e:
print(f"虎牙播放解析错误: {e}")
return 1, self.excepturl
def douyuplay(self, ids):
"""
斗鱼播放解析 - 实时获取对应码率的播放URL
ids格式: [平台, base64(清晰度列表), base64(会话信息)]
清晰度列表: [名称1, #码率1, 名称2, #码率2, ...]
#表示这是码率值，需要重新获取URL
"""
try:
sdata = json.loads(self.d64(ids[1]))
headers = self.gethr(0, zr=f'{self.hosts[ids[0]]}/{sdata["id"]}')
ldata = json.loads(self.d64(ids[2]))
result_obj = {}
with ThreadPoolExecutor(max_workers=len(ldata)) as executor:
futures = [
executor.submit(
self.douyufp,
sdata,
quality,
headers,
self.hosts[ids[0]],
result_obj
) for quality in ldata
]
for future in futures:
future.result()
if len(ids) < 3:
# 兼容旧格式
decoded = json.loads(self.d64(ids[1]))
return 0, decoded
# 解析清晰度列表和会话信息
qualities = json.loads(self.d64(ids[1]))
session_info = json.loads(self.d64(ids[2]))
channel = session_info['channel']
device_id = session_info['device_id']
secret_key = session_info['secret_key']
random_str = session_info['random_str']
enc_time = session_info['enc_time']
enc_data = session_info['enc_data']
# 为每个清晰度实时获取播放URL
result = []
for bit in sorted(result_obj.keys(), reverse=True):
result.extend(result_obj[bit])
if result:
return 0, result
return 1, self.excepturl
for i in range(0, len(qualities), 2):
name = qualities[i]
rate_marker = qualities[i + 1]
# 解析码率值（去掉#前缀）
if rate_marker.startswith('#'):
rate = int(rate_marker[1:])
else:
rate = -1
# 实时获取对应码率的URL
play_url = self._get_douyu_play_url(
channel, device_id, secret_key, random_str,
enc_time, enc_data, rate
)
if play_url:
result.extend([name, play_url])
if not result:
return 1, self.excepturl
return 0, result
except Exception as e:
print(f"斗鱼播放解析错误: {e}")
return 1, self.excepturl
def douyufp(self, sdata, quality, headers, host, result_obj):
try:
body = f'{sdata["sign"]}&cdn={sdata["cdn"]}&rate={quality["rate"]}'
body=self.params_to_json(body)
data = self.post(f'{host}/lapi/live/getH5Play/{sdata["id"]}',
data=body, headers=headers).json()
if data.get('data'):
play_url = data['data']['rtmp_url'] + '/' + data['data']['rtmp_live']
bit = quality.get('bit', 0)
if bit not in result_obj:
result_obj[bit] = []
result_obj[bit].extend([quality['name'], play_url])
except Exception as e:
print(f"Error fetching {quality['name']}: {str(e)}")
def localProxy(self, param):
pass
@@ -763,5 +1257,4 @@ class Spider(Spider):
def handle_exception(self, e):
print(f"报错: {str(e)}")
return {'vod_play_from': '哎呀翻车啦', 'vod_play_url': f'翻车啦${self.excepturl}'}
return {'vod_play_from': '哎呀翻车啦', 'vod_play_url': f'翻车啦${self.excepturl}'}