github/DouyinLiveRecorder
1
0
Fork 0
mirror of https://github.com/ihmily/DouyinLiveRecorder.git synced 2025-12-25 21:36:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add vcodec flag for tiktok and douyin

Browse Source
This commit is contained in:
ihmily 2025-07-19 19:06:27 +08:00
parent 9f499f3fa6
commit 8e4e9b098f
3 changed files with 38 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
main.py
View File

@ -523,7 +523,8 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
url=record_url,
proxy_addr=proxy_address,
cookies=dy_cookie))
port_info = asyncio.run(stream.get_douyin_stream_url(json_data, record_quality))
port_info = asyncio.run(
stream.get_douyin_stream_url(json_data, record_quality, proxy_address))
elif record_url.find("https://www.tiktok.com/") > -1:
platform = 'TikTok直播'
@ -533,7 +534,8 @@ def start_record(url_data: tuple, count_variable: int = -1) -> None:
url=record_url,
proxy_addr=proxy_address,
cookies=tiktok_cookie))
port_info = asyncio.run(stream.get_tiktok_stream_url(json_data, record_quality))
port_info = asyncio.run(
stream.get_tiktok_stream_url(json_data, record_quality, proxy_address))
else:
logger.error("错误信息: 网络异常，请检查网络是否能正常访问TikTok平台")

7
src/spider.py
View File

@ -194,8 +194,9 @@ async def get_douyin_stream_data(url: str, proxy_addr: OptionalStr = None, cooki
origin_url_list = json.loads(match_json_str3.group(1) + '}')
if origin_url_list:
origin_m3u8 = {'ORIGIN': origin_url_list["hls"]}
origin_flv = {'ORIGIN': origin_url_list["flv"]}
origin_hls_codec = origin_url_list['sdk_params'].get('VCodec') or ''
origin_m3u8 = {'ORIGIN': origin_url_list["hls"] + '&codec=' + origin_hls_codec}
origin_flv = {'ORIGIN': origin_url_list["flv"] + '&codec=' + origin_hls_codec}
hls_pull_url_map = json_data['stream_url']['hls_pull_url_map']
flv_pull_url = json_data['stream_url']['flv_pull_url']
json_data['stream_url']['hls_pull_url_map'] = {**origin_m3u8, **hls_pull_url_map}
@ -216,7 +217,7 @@ async def get_tiktok_stream_data(url: str, proxy_addr: OptionalStr = None, cooki
if cookies:
headers['Cookie'] = cookies
for i in range(3):
html_str = await async_req(url=url, proxy_addr=proxy_addr, headers=headers, abroad=True)
html_str = await async_req(url=url, proxy_addr=proxy_addr, headers=headers, abroad=True, http2=False)
time.sleep(1)
if "We regret to inform you that we have discontinued operating TikTok" in html_str:
msg = re.search('<p>\n\\s+(We regret to inform you that we have discontinu.*?)\\.\n\\s+</p>', html_str)

35
src/stream.py
View File

@ -21,6 +21,7 @@ from .utils import trace_error_decorator
from .spider import (
get_douyu_stream_data, get_bilibili_stream_data
)
from .http_clients.async_http import get_response_status
QUALITY_MAPPING = {"OD": 0, "BD": 0, "UHD": 1, "HD": 2, "SD": 3, "LD": 4}
@ -37,7 +38,7 @@ def get_quality_index(quality) -> tuple:
@trace_error_decorator
async def get_douyin_stream_url(json_data: dict, video_quality: str) -> dict:
async def get_douyin_stream_url(json_data: dict, video_quality: str, proxy_addr: str) -> dict:
anchor_name = json_data.get('anchor_name')
result = {
@ -61,6 +62,11 @@ async def get_douyin_stream_url(json_data: dict, video_quality: str) -> dict:
video_quality, quality_index = get_quality_index(video_quality)
m3u8_url = m3u8_url_list[quality_index]
flv_url = flv_url_list[quality_index]
ok = await get_response_status(url=m3u8_url, proxy_addr=proxy_addr)
if not ok:
index = quality_index + 1 if quality_index < 4 else quality_index - 1
m3u8_url = m3u8_url_list[index]
flv_url = flv_url_list[index]
result |= {
'is_live': True,
'title': json_data['title'],
@ -73,7 +79,7 @@ async def get_douyin_stream_url(json_data: dict, video_quality: str) -> dict:
@trace_error_decorator
async def get_tiktok_stream_url(json_data: dict, video_quality: str) -> dict:
async def get_tiktok_stream_url(json_data: dict, video_quality: str, proxy_addr: str) -> dict:
if not json_data:
return {"anchor_name": None, "is_live": False}
@ -81,10 +87,18 @@ async def get_tiktok_stream_url(json_data: dict, video_quality: str) -> dict:
play_list = []
for key in stream:
url_info = stream[key]['main']
play_url = url_info[q_key]
sdk_params = url_info['sdk_params']
sdk_params = json.loads(sdk_params)
vbitrate = int(sdk_params['vbitrate'])
v_codec = sdk_params.get('VCodec', '')
play_url = ''
if url_info.get(q_key):
if url_info[q_key].endswith(".flv") or url_info[q_key].endswith(".m3u8"):
play_url = url_info[q_key] + '?codec=' + v_codec
else:
play_url = url_info[q_key] + '&codec=' + v_codec
resolution = sdk_params['resolution']
if vbitrate != 0 and resolution:
width, height = map(int, resolution.split('x'))
@ -115,8 +129,19 @@ async def get_tiktok_stream_url(json_data: dict, video_quality: str) -> dict:
while len(m3u8_url_list) < 5:
m3u8_url_list.append(m3u8_url_list[-1])
video_quality, quality_index = get_quality_index(video_quality)
flv_url = flv_url_list[quality_index]['url'].replace("https://", "http://")
m3u8_url = m3u8_url_list[quality_index]['url'].replace("https://", "http://")
flv_dict: dict = flv_url_list[quality_index]
m3u8_dict: dict = m3u8_url_list[quality_index]
check_url = m3u8_dict.get('url') or flv_dict.get('url')
ok = await get_response_status(url=check_url, proxy_addr=proxy_addr, http2=False)
if not ok:
index = quality_index + 1 if quality_index < 4 else quality_index - 1
flv_dict: dict = flv_url_list[index]
m3u8_dict: dict = m3u8_url_list[index]
flv_url = flv_dict['url']
m3u8_url = m3u8_dict['url']
result |= {
'is_live': True,
'title': live_room['liveRoom']['title'],