# -*- coding: utf-8 -*- # 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 from pyquery import PyQuery as pq sys.path.append('..') from base.spider import Spider from concurrent.futures import ThreadPoolExecutor class Spider(Spider): def init(self, extend=""): # 初始化B站WBI密钥 self.bili_wbi_keys = None self.bili_wbi_expire = 0 pass def getName(self): return "直播" def isVideoFormat(self, url): pass def manualVideoCheck(self): pass def destroy(self): pass 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" }, { "User-Agent": "Dart/3.4 (dart:io)" } ] excepturl = 'https://www.baidu.com' hosts = { "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"] } referers = { "huya": "https://live.cdn.huya.com", "douyu": "https://m.douyu.com", "bili": "https://live.bilibili.com" } playheaders = { "wangyi": { "User-Agent": "ExoPlayer", "Connection": "Keep-Alive", "Icy-MetaData": "1" }, "bili": { 'Accept': '*/*', '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', 'Connection': 'Keep-Alive', 'Icy-MetaData': '1' }, 'douyu': { 'User-Agent': 'libmpv', 'Icy-MetaData': '1' } } # 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: # 检查缓存 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() 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', [{'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: self.dyufdata = self.fetch( f'{self.referers["douyu"]}/api/cate/list', headers=self.headers[1] ).json() return ('douyu', [{'key': 'cate', 'name': '分类', 'value': [{'n': i['cate1Name'], 'v': str(i['cate1Id'])} for i in self.dyufdata['data']['cate1Info']]}]) except Exception as e: print(f"douyu错误: {e}") return 'douyu', None def homeContent(self, filter): result = {} cateManual = { "虎牙": "huya", "斗鱼": "douyu", "网易": "wangyi", "B站": "bili" } classes = [] filters = { 'huya': [{'key': 'cate', 'name': '分类', 'value': [{'n': '网游', 'v': '1'}, {'n': '单机', 'v': '2'}, {'n': '娱乐', 'v': '8'}, {'n': '手游', 'v': '3'}]}] } with ThreadPoolExecutor(max_workers=2) as executor: futures = { executor.submit(self.process_bili): 'bili', executor.submit(self.process_douyu): 'douyu' } for future in futures: platform, filter_data = future.result() if filter_data: filters[platform] = filter_data for k in cateManual: classes.append({ 'type_name': k, 'type_id': cateManual[k] }) result['class'] = classes result['filters'] = filters return result def homeVideoContent(self): pass def categoryContent(self, tid, pg, filter, extend): vdata = [] result = {} pagecount = 9999 result['page'] = pg result['limit'] = 90 result['total'] = 999999 if tid == 'wangyi': vdata, pagecount = self.wyccContent(tid, pg, filter, extend, vdata) elif 'bili' in tid: vdata, pagecount = self.biliContent(tid, pg, filter, extend, vdata) elif 'huya' in tid: vdata, pagecount = self.huyaContent(tid, pg, filter, extend, vdata) elif 'douyu' in tid: vdata, pagecount = self.douyuContent(tid, pg, filter, extend, vdata) result['list'] = vdata result['pagecount'] = pagecount return result def wyccContent(self, tid, pg, filter, extend, vdata): params = { 'format': 'json', 'start': (int(pg) - 1) * 20, 'size': '20', } response = self.fetch(f'{self.hosts[tid]}/api/category/live/', params=params, headers=self.headers[0]).json() for i in response['lives']: if i.get('cuteid'): bvdata = self.buildvod( vod_id=f"{tid}@@{i['cuteid']}", vod_name=i.get('title'), vod_pic=i.get('cover'), vod_remarks=i.get('nickname'), style={"type": "rect", "ratio": 1.33} ) vdata.append(bvdata) return vdata, 9999 def biliContent(self, tid, pg, filter, extend, vdata): """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] 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: id = extend.get('cate') data = self.fetch(f'{self.referers[tid]}/liveconfig/game/bussLive?bussType={id}', headers=self.headers[1]).json() for i in data['data']: v = self.buildvod( vod_id=f"click_{tid}@@{int(i['gid'])}", vod_name=i.get('gameFullName'), vod_pic=f'https://huyaimg.msstatic.com/cdnimage/game/{int(i["gid"])}-MS.jpg', vod_tag=1, style={"type": "oval", "ratio": 1} ) vdata.append(v) return vdata, 1 else: gid = '' if 'click' in tid: ids = tid.split('_')[1].split('@@') tid = ids[0] gid = f'&gameId={ids[1]}' data = self.fetch(f'{self.hosts[tid][0]}/cache.php?m=LiveList&do=getLiveListByPage&tagAll=0{gid}&page={pg}', headers=self.headers[1]).json() for i in data['data']['datas']: if i.get('profileRoom'): v = self.buildvod( f"{tid}@@{i['profileRoom']}", i.get('introduction'), i.get('screenshot'), str(int(i.get('totalCount', '1')) / 10000) + '万', 0, i.get('nick'), 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']: if str(i['cate1Id']) == extend['cate']: v = self.buildvod( vod_id=f"click_{tid}@@{i['cate2Id']}", vod_name=i.get('cate2Name'), vod_pic=i.get('icon'), vod_remarks=i.get('count'), vod_tag=1, style={"type": "oval", "ratio": 1} ) vdata.append(v) return vdata, 1 else: path = f'/japi/weblist/apinc/allpage/6/{pg}' if 'click' in tid: ids = tid.split('_')[1].split('@@') tid = ids[0] path = f'/gapi/rkc/directory/mixList/2_{ids[1]}/{pg}' url = f'{self.hosts[tid]}{path}' data = self.fetch(url, headers=self.headers[1]).json() for i in data['data']['rl']: v = self.buildvod( vod_id=f"{tid}@@{i['rid']}", vod_name=i.get('rn'), vod_pic=i.get('rs16'), vod_year=str(int(i.get('ol', 1)) / 10000) + '万', vod_remarks=i.get('nn'), style={"type": "rect", "ratio": 1.33} ) vdata.append(v) return vdata, 9999 def detailContent(self, ids): ids = ids[0].split('@@') if ids[0] == 'wangyi': vod = self.wyccDetail(ids) elif ids[0] == 'bili': vod = self.biliDetail(ids) elif ids[0] == 'huya': vod = self.huyaDetail(ids) elif ids[0] == 'douyu': vod = self.douyuDetail(ids) return {'list': [vod]} def wyccDetail(self, ids): try: vdata = self.getpq(f'{self.hosts[ids[0]]}/{ids[1]}', self.headers[0])('script').eq(-1).text() def get_quality_name(vbr): if vbr <= 600: return "标清" elif vbr <= 1000: return "高清" elif vbr <= 2000: return "超清" else: return "蓝光" data = json.loads(vdata)['props']['pageProps']['roomInfoInitData'] name = data['live'].get('title', ids[0]) vod = self.buildvod(vod_name=data.get('keywords_suffix'), vod_remarks=data['live'].get('title'), vod_content=data.get('description_suffix')) resolution_data = data['live']['quickplay']['resolution'] all_streams = {} sorted_qualities = sorted(resolution_data.items(), key=lambda x: x[1]['vbr'], reverse=True) for quality, data in sorted_qualities: vbr = data['vbr'] quality_name = get_quality_name(vbr) for cdn_name, url in data['cdn'].items(): if cdn_name not in all_streams and type(url) == str and url.startswith('http'): all_streams[cdn_name] = [] if isinstance(url, str) and url.startswith('http'): all_streams[cdn_name].extend([quality_name, url]) plists = [] names = [] for i, (cdn_name, stream_list) in enumerate(all_streams.items(), 1): names.append(f'线路{i}') pstr = f"{name}${ids[0]}@@{self.e64(json.dumps(stream_list))}" plists.append(pstr) vod['vod_play_from'] = "$$$".join(names) vod['vod_play_url'] = "$$$".join(plists) return vod except Exception as e: return self.handle_exception(e) def biliDetail(self, ids): """ B站直播详情 - 使用playUrl接口获取多清晰度 """ try: 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=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)}人" ) # 获取播放地址信息 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: 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=live_data.get('introduction', '虎牙直播'), type_name=live_data.get('gameFullName', ''), vod_director=live_data.get('nick', ''), vod_remarks=live_data.get('contentIntro', ''), ) # 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 ) 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 _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): """ 斗鱼播放详情 - 参考最新三合一.js重构 核心算法：设备ID生成 -> 获取加密密钥 -> 计算签名 -> 获取播放地址 修复：切换分辨率只能播放1秒的问题 方案：存储房间号和码率信息，在playerContent中实时获取对应码率的URL """ try: 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=room_info.get('second_lvl_name', ''), vod_director=room_info.get('nickname', ''), ) # 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 _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 def playerContent(self, flag, id, vipFlags): try: ids = id.split('@@') p = 1 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 = self.huyaplay(ids) elif ids[0] == 'douyu': p, url = self.douyuplay(ids) return {'parse': p, 'url': url, 'header': self.playheaders[ids[0]]} except Exception as e: return {'parse': 1, 'url': self.excepturl, 'header': self.headers[0]} def biliplay(self, ids): """ B站播放解析 - 使用playUrl接口获取指定清晰度直播流 ids: [平台, 房间号, 清晰度qn] 支持多线路返回 """ try: 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 = [] 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: 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 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 localProxy(self, param): pass 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: print(f"Base64编码错误: {str(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: print(f"Base64解码错误: {str(e)}") return "" def josn_to_params(self, params, skip_empty=False): query = [] for k, v in params.items(): if skip_empty and not v: continue query.append(f"{k}={v}") return "&".join(query) def params_to_json(self, query_string): parsed_data = parse_qs(query_string) result = {key: value[0] for key, value in parsed_data.items()} return result def buildvod(self, vod_id='', vod_name='', vod_pic='', vod_year='', vod_tag='', vod_remarks='', style='', type_name='', vod_area='', vod_actor='', vod_director='', vod_content='', vod_play_from='', vod_play_url=''): vod = { 'vod_id': vod_id, 'vod_name': vod_name, 'vod_pic': vod_pic, 'vod_year': vod_year, 'vod_tag': 'folder' if vod_tag else '', 'vod_remarks': vod_remarks, 'style': style, 'type_name': type_name, 'vod_area': vod_area, 'vod_actor': vod_actor, 'vod_director': vod_director, 'vod_content': vod_content, 'vod_play_from': vod_play_from, 'vod_play_url': vod_play_url } vod = {key: value for key, value in vod.items() if value} return vod def getpq(self, url, headers=None, cookies=None): data = self.fetch(url, headers=headers, cookies=cookies).text try: return pq(data) except Exception as e: print(f"解析页面错误: {str(e)}") return pq(data.encode('utf-8')) def gethr(self, index, rf='', zr=''): headers = self.headers[index] if zr: headers['referer'] = zr else: headers['referer'] = f"{self.referers[rf]}/" return headers def handle_exception(self, e): print(f"报错: {str(e)}") return {'vod_play_from': '哎呀翻车啦', 'vod_play_url': f'翻车啦${self.excepturl}'}