github/DouyinLiveRecorder
1
0
Fork 0
mirror of https://github.com/ihmily/DouyinLiveRecorder.git synced 2025-12-26 05:48:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: optimize bilibili live stream fetch

Browse Source
This commit is contained in:
ihmily 2024-07-21 16:11:13 +08:00
parent f615ee3130
commit c86f3a76f3
2 changed files with 66 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
main.py
View File

@ -4,7 +4,7 @@
Author: Hmily
GitHub: https://github.com/ihmily
Date: 2023-07-17 23:52:05
Update: 2024-07-15 22:55:29
Update: 2024-07-20 20:41:12
Copyright (c) 2023-2024 by Hmily, All Rights Reserved.
Function: Record live stream video.
"""
@ -38,6 +38,7 @@ from spider import (
get_douyu_stream_data,
get_yy_stream_data,
get_bilibili_stream_data,
get_bilibili_room_info,
get_xhs_stream_url,
get_bigo_stream_url,
get_blued_stream_url,
@ -539,66 +540,32 @@ def get_yy_stream_url(json_data: dict) -> Dict[str, Any]:
@trace_error_decorator
def get_bilibili_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
if "is_live" in json_data and not json_data['anchor_name']:
return json_data
anchor_name = json_data['anchor_name']
playurl_info = json_data['stream_data']['playurl_info']
result = {
"anchor_name": anchor_name,
"is_live": False,
}
if playurl_info:
# 其中qn=30000为杜比 20000为4K 10000为原画 400蓝光 250超清 150高清 80流畅
quality_list = {'10000': 'bluray', '400': '4000', '250': '2500', '150': '1500', '80': '800'}
format_list = playurl_info['playurl']['stream'][1]['format']
current_qn = format_list[0]['codec'][0]['current_qn']
if int(current_qn) != 10000 and len(format_list) > 1:
stream_data = format_list[1]['codec'][0]
else:
stream_data = format_list[0]['codec'][0]
accept_qn_list = stream_data['accept_qn']
qn_count = len(accept_qn_list)
if 10000 not in accept_qn_list:
new_accept_qn_list = [10000] + accept_qn_list
else:
new_accept_qn_list = [i for i in accept_qn_list]
while len(new_accept_qn_list) < 5:
new_accept_qn_list.append(new_accept_qn_list[-1])
video_quality_options = {
"原画": 0,
"蓝光": 1,
"超清": 2,
"高清": 3,
"标清": 4,
"流畅": 4
anchor_name = json_data["anchor_name"]
if not json_data["live_status"]:
return {
"anchor_name": anchor_name,
"is_live": False
}
select_quality = new_accept_qn_list[video_quality_options[video_quality]]
select_quality = quality_list[str(select_quality)]
base_url = stream_data['base_url']
host = stream_data['url_info'][0]['host']
extra = stream_data['url_info'][0]['extra']
if int(current_qn) != 10000 and qn_count > 1:
if (qn_count == 2 and 10000 not in accept_qn_list) or qn_count > 2:
live_key = base_url.split('/')[3]
live_key_list = live_key.split('_')
if len(live_key_list) == 4:
live_key_list[3] = select_quality
else:
live_key_list.append(select_quality)
room_url = json_data['room_url']
new_live_key = '_'.join(live_key_list)
base_url = re.sub(live_key, new_live_key, base_url)
extra = re.sub(live_key, new_live_key, extra)
time_at = int(time.time()) + 10800
extra = re.sub(r"expires=[0-9]+&len=", f"expires={time_at}&len=", extra)
video_quality_options = {
"原画": '10000',
"蓝光": '400',
"超清": '250',
"高清": '150',
"标清": '80',
"流畅": '80'
}
m3u8_url = host + base_url + extra
result['is_live'] = True
result['record_url'] = m3u8_url # B站使用m3u8链接进行录制
return result
select_quality = video_quality_options[video_quality]
play_url = get_bilibili_stream_data(room_url, qn=select_quality, platform='web', proxy_addr=proxy_addr,
cookies=bili_cookie)
return {
'anchor_name': json_data['anchor_name'],
'is_live': True,
'record_url': play_url
}
@trace_error_decorator
@ -773,7 +740,7 @@ def start_record(url_data: tuple, count_variable: int = -1):
elif record_url.find("https://live.bilibili.com/") > -1:
platform = 'B站直播'
with semaphore:
json_data = get_bilibili_stream_data(
json_data = get_bilibili_room_info(
url=record_url, proxy_addr=proxy_address, cookies=bili_cookie)
port_info = get_bilibili_stream_url(json_data, record_quality)
@ -1898,4 +1865,4 @@ while True:
first_run = False
time.sleep(3)
time.sleep(3)

63
spider.py
View File

@ -4,7 +4,7 @@
Author: Hmily
GitHub: https://github.com/ihmily
Date: 2023-07-15 23:15:00
Update: 2024-07-15 22:55:00
Update: 2024-07-20 20:41:12
Copyright (c) 2023 by Hmily, All Rights Reserved.
Function: Get live stream data.
"""
@ -580,7 +580,7 @@ def get_yy_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: U
@trace_error_decorator
def get_bilibili_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
def get_bilibili_room_info(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0',
@ -590,31 +590,48 @@ def get_bilibili_stream_data(url: str, proxy_addr: Union[str, None] = None, cook
if cookies:
headers['Cookie'] = cookies
def get_data_from_api(rid: str) -> Dict[str, Any]:
api = f'https://api.live.bilibili.com/xlive/web-room/v2/index/getRoomPlayInfo?room_id={rid}&no_playurl=0&mask=1&qn=0&platform=web&protocol=0,1&format=0,1,2&codec=0,1,2&dolby=5&panorama=1'
json_str = get_req(url=api, proxy_addr=proxy_addr, headers=headers)
return json.loads(json_str)
try:
html_str = get_req(url=url, proxy_addr=proxy_addr, headers=headers)
json_str = re.search('<script>window.__NEPTUNE_IS_MY_WAIFU__=(.*?)</script><script>', html_str, re.S)
if json_str:
json_str = json_str.group(1)
json_data = json.loads(json_str)
json_data['anchor_name'] = json_data['roomInfoRes']['data']['anchor_info']['base_info']['uname']
json_data['stream_data'] = json_data['roomInitRes']['data']
else:
room_id = url.split('?')[0].rsplit('/', maxsplit=1)[1]
json_data = get_data_from_api(room_id)
json_data['stream_data'] = json_data['data']
info_api = f'https://api.live.bilibili.com/live_user/v1/UserInfo/get_anchor_in_room?roomid={room_id}'
json_str = get_req(info_api, proxy_addr=proxy_addr, headers=headers)
json_data['anchor_name'] = json.loads(json_str)['data']['info']['uname']
room_id = url.split('?')[0].rsplit('/', maxsplit=1)[1]
json_str = get_req(f'https://api.live.bilibili.com/room/v1/Room/room_init?id={room_id}',
proxy_addr=proxy_addr,headers=headers)
room_info = json.loads(json_str)
uid = room_info['data']['uid']
live_status = True if room_info['data']['live_status'] == 1 else False
return json_data
api = f'https://api.live.bilibili.com/live_user/v1/Master/info?uid={uid}'
json_str2 = get_req(url=api, proxy_addr=proxy_addr, headers=headers)
anchor_info = json.loads(json_str2)
anchor_name = anchor_info['data']['info']['uname']
return {"anchor_name": anchor_name, "live_status": live_status, "room_url":url}
except Exception as e:
print(e)
return {"anchor_name": '', "is_live": False}
return {"anchor_name": '', "live_status": False, "room_url":url}
@trace_error_decorator
def get_bilibili_stream_data(url: str, qn: str = '10000', platform: str = 'web', proxy_addr: Union[str, None] = None,
cookies: Union[str, None] = None) -> str:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2',
}
if cookies:
headers['Cookie'] = cookies
room_id = url.split('?')[0].rsplit('/', maxsplit=1)[1]
params = {
'cid': room_id,
'qn': qn,
'platform': platform,
}
play_api = f'https://api.live.bilibili.com/room/v1/Room/playUrl?{urllib.parse.urlencode(params)}'
json_str = get_req(play_api, proxy_addr=proxy_addr, headers=headers)
json_data = json.loads(json_str)
for i in json_data['data']['durl']:
if 'd1--cn-gotcha' in i['url']:
return i['url']
return json_data['data']['durl'][-1]['url']
@trace_error_decorator