From ac85f1194e9a9fef93ee3b0b49178234c699211f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=94=84=E6=96=B0?= Date: Fri, 17 Jul 2026 11:40:34 +0800 Subject: [PATCH] fix(bilibili): use API metadata and expose DASH audio --- src/you_get/common.py | 20 +- src/you_get/extractor.py | 2 +- src/you_get/extractors/bilibili.py | 292 ++++++++++++++++++++--------- tests/test_bilibili.py | 205 ++++++++++++++++++++ tests/test_common.py | 10 + tests/test_extractor.py | 37 ++++ 6 files changed, 471 insertions(+), 95 deletions(-) create mode 100644 tests/test_bilibili.py create mode 100644 tests/test_extractor.py diff --git a/src/you_get/common.py b/src/you_get/common.py index c99e13da60..9f749ea7c1 100755 --- a/src/you_get/common.py +++ b/src/you_get/common.py @@ -555,17 +555,23 @@ def post_content(url, headers={}, post_data={}, decoded=True, **kwargs): return data -def url_size(url, faker=False, headers={}): +def url_size(url, faker=False, headers={}, timeout=None): if faker: - response = urlopen_with_retry( - request.Request(url, headers=fake_headers) - ) + req = request.Request(url, headers=fake_headers) elif headers: - response = urlopen_with_retry(request.Request(url, headers=headers)) + req = request.Request(url, headers=headers) + else: + req = url + + if timeout is None: + response = urlopen_with_retry(req) else: - response = urlopen_with_retry(url) + response = urlopen_with_retry(req, timeout=timeout) - size = response.headers['content-length'] + try: + size = response.headers['content-length'] + finally: + response.close() return int(size) if size is not None else float('inf') diff --git a/src/you_get/extractor.py b/src/you_get/extractor.py index bd71717e72..f44c9ac9b8 100644 --- a/src/you_get/extractor.py +++ b/src/you_get/extractor.py @@ -224,7 +224,7 @@ def download(self, **kwargs): ext = self.dash_streams[stream_id]['container'] total_size = self.dash_streams[stream_id]['size'] - if ext == 'm3u8' or ext == 'm4a': + if ext == 'm3u8': ext = 'mp4' if not urls: diff --git a/src/you_get/extractors/bilibili.py b/src/you_get/extractors/bilibili.py index ea67f92fcf..3e4d1ccddb 100644 --- a/src/you_get/extractors/bilibili.py +++ b/src/you_get/extractors/bilibili.py @@ -1,5 +1,6 @@ #!/usr/bin/env python +from .. import common from ..common import * from ..extractor import VideoExtractor @@ -40,6 +41,14 @@ class Bilibili(VideoExtractor): {'id': 'mp4', 'quality': 0}, {'id': 'jpg', 'quality': 0}, + + # Standalone DASH audio formats, in descending quality order. + {'id': 'dash-audio-30280', 'quality': 30280, + 'container': 'M4A', 'desc': '高音质音频'}, + {'id': 'dash-audio-30232', 'quality': 30232, + 'container': 'M4A', 'desc': '中音质音频'}, + {'id': 'dash-audio-30216', 'quality': 30216, + 'container': 'M4A', 'desc': '低音质音频'}, ] codecids = {7: 'AVC', 12: 'HEVC', 13: 'AV1'} @@ -61,8 +70,8 @@ def height_to_quality(height, qn): @staticmethod def bilibili_headers(referer=None, cookie=None): - # a reasonable UA - ua = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.84 Safari/537.36' + # Keep this in sync with the current browser UA used elsewhere in you-get. + ua = fake_headers['User-Agent'] headers = {'Accept': '*/*', 'Accept-Language': 'en-US,en;q=0.5', 'User-Agent': ua} if referer is not None: headers.update({'Referer': referer}) @@ -74,6 +83,12 @@ def bilibili_headers(referer=None, cookie=None): def bilibili_api(avid, cid, qn=0): return 'https://api.bilibili.com/x/player/playurl?avid=%s&cid=%s&qn=%s&type=&otype=json&fnver=0&fnval=4048&fourk=1' % (avid, cid, qn) + @staticmethod + def bilibili_view_api(video_id): + if video_id.lower().startswith('av'): + return 'https://api.bilibili.com/x/web-interface/view?aid=%s' % video_id[2:] + return 'https://api.bilibili.com/x/web-interface/view?bvid=%s' % video_id + @staticmethod def bilibili_audio_api(sid): return 'https://www.bilibili.com/audio/music-service-c/web/url?sid=%s' % sid @@ -143,36 +158,107 @@ def bilibili_vc_api(video_id): def bilibili_h_api(doc_id): return 'https://api.vc.bilibili.com/link_draw/v1/doc/detail?doc_id=%s' % doc_id + @staticmethod + def video_id_from_url(url): + match = re.search(r'/video/(av\d+|BV[0-9A-Za-z]+)', url, re.IGNORECASE) + return match.group(1) if match else None + + def normalize_video_url(self): + match = re.match( + r'https?://(?:www\.)?bilibili\.com/watchlater/#/(av\d+|BV[0-9A-Za-z]+)/?', + self.url, + re.IGNORECASE, + ) + if match: + page = int(match1(self.url, r'/p(\d+)') or '1') + self.url = 'https://www.bilibili.com/video/%s?p=%s' % (match.group(1), page) + return + + if re.match(r'https?://(?:www\.)?bilibili\.com/festival/', self.url): + video_id = match1(self.url, r'[?&]bvid=([^&]+)') + if video_id: + self.url = 'https://www.bilibili.com/video/%s' % video_id + + def get_video_info(self): + video_id = self.video_id_from_url(self.url) + if video_id is None: + log.wtf('[Failed] Unable to find a Bilibili video ID in the URL.') + + api_url = self.bilibili_view_api(video_id) + api_content = get_content(api_url, headers=self.bilibili_headers(referer=self.url)) + api_response = json.loads(api_content) + video_info = api_response.get('data') + if api_response.get('code') != 0 or not video_info: + message = api_response.get('message') or 'unknown API error' + log.wtf('[Failed] Unable to fetch Bilibili video metadata: %s' % message) + return video_info + @staticmethod def url_size(url, faker=False, headers={},err_value=0): try: - return url_size(url,faker,headers) + return url_size(url, faker, headers, timeout=5) except: return err_value + def add_dash_audio_streams(self, dash, audio_size_cache, requested_stream_id=None): + known_formats = {stream['id'] for stream in self.stream_types} + for audio in dash.get('audio') or []: + audio_id = int(audio['id']) + format_id = 'dash-audio-%s' % audio_id + if format_id not in known_formats: + continue + if requested_stream_id and format_id != requested_stream_id: + continue + + audio_url = audio.get('baseUrl') or audio.get('base_url') + if not audio_url: + continue + + if audio_id not in audio_size_cache: + audio_size_cache[audio_id] = self.url_size( + audio_url, + headers=self.bilibili_headers(referer=self.url), + ) + + codec = audio.get('codecs', 'unknown codec') + bandwidth = audio.get('bandwidth') + desc = '音频 %s' % codec + if bandwidth: + desc += ' %s kbps' % round(bandwidth / 1000) + self.streams[format_id] = { + 'container': 'm4a', + 'quality': desc, + 'size': audio_size_cache[audio_id], + 'src': [audio_url], + } + def prepare(self, **kwargs): self.stream_qualities = {s['quality']: s for s in self.stream_types} self.streams.clear() self.dash_streams.clear() - try: - html_content = get_content(self.url, headers=self.bilibili_headers(referer=self.url)) - except: - html_content = '' # live always returns 400 (why?) + self.normalize_video_url() + video_info = None + if self.video_id_from_url(self.url): + video_info = self.get_video_info() + redirect_url = video_info.get('redirect_url') + if redirect_url and re.search(r'/bangumi/play/', redirect_url): + self.url = redirect_url + video_info = None + + html_content = '' + if video_info is None: + try: + html_content = get_content(self.url, headers=self.bilibili_headers(referer=self.url)) + except: + html_content = '' # live always returns 400 (why?) #self.title = match1(html_content, # r'

bangumi/play/ep # redirect: bangumi.bilibili.com/anime -> bangumi/play/ep - elif re.match(r'https?://(www\.)?bilibili\.com/bangumi/play/ss(\d+)', self.url) or \ - re.match(r'https?://bangumi\.bilibili\.com/anime/(\d+)/play', self.url): + if re.match(r'https?://(www\.)?bilibili\.com/bangumi/play/ss(\d+)', self.url) or \ + re.match(r'https?://bangumi\.bilibili\.com/anime/(\d+)/play', self.url): initial_state_text = match1(html_content, r'__INITIAL_STATE__=(.*?);\(function\(\)') # FIXME initial_state = json.loads(initial_state_text) ep_id = initial_state['epList'][0]['id'] @@ -184,11 +270,6 @@ def prepare(self, **kwargs): self.url = 'https://www.bilibili.com/%s' % match1(self.url, r'/s/(.+)') html_content = get_content(self.url, headers=self.bilibili_headers()) - # redirect: festival - elif re.match(r'https?://(www\.)?bilibili\.com/festival/(.+)', self.url): - self.url = 'https://www.bilibili.com/video/%s' % match1(self.url, r'bvid=([^&]+)') - html_content = get_content(self.url, headers=self.bilibili_headers()) - # sort it out if re.match(r'https?://(www\.)?bilibili\.com/audio/au(\d+)', self.url): sort = 'audio' @@ -210,52 +291,59 @@ def prepare(self, **kwargs): # regular video if sort == 'video': - initial_state_text = match1(html_content, r'__INITIAL_STATE__=(.*?);\(function\(\)') # FIXME - initial_state = json.loads(initial_state_text) - - playinfo_text = match1(html_content, r'__playinfo__=(.*?)