From c86f3a76f356608158a51645dced8a224d231f75 Mon Sep 17 00:00:00 2001
From: ihmily <961532186@qq.com>
Date: Sun, 21 Jul 2024 16:11:13 +0800
Subject: [PATCH] feat: optimize bilibili live stream fetch

---
 main.py   | 85 +++++++++++++++++--------------------------------------
 spider.py | 63 ++++++++++++++++++++++++++---------------
 2 files changed, 66 insertions(+), 82 deletions(-)

diff --git a/main.py b/main.py
index ae73156..ac1046a 100644
--- a/main.py
+++ b/main.py
@@ -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)
\ No newline at end of file
+    time.sleep(3)
diff --git a/spider.py b/spider.py
index 5d623a3..a4666ec 100644
--- a/spider.py
+++ b/spider.py
@@ -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