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: add acfun live record and fix douyin and weibo

Browse Source
This commit is contained in:
ihmily 2024-06-18 06:25:45 +08:00
parent d296540ae9
commit c82f31b785
3 changed files with 202 additions and 204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
config/config.ini
View File

@ -58,6 +58,9 @@ kugou_cookie =
twitch_cookie =
liveme_cookie =
huajiao_cookie =
liuxing_cookie =
showroom_cookie =
acfun_cookie =
[Authorization]
popkontv_token =

227
main.py
View File

@ -4,7 +4,7 @@
Author: Hmily
GitHub: https://github.com/ihmily
Date: 2023-07-17 23:52:05
Update: 2024-06-13 21:33:10
Update: 2024-06-18 06:18:29
Copyright (c) 2023-2024 by Hmily, All Rights Reserved.
Function: Record live stream video.
"""
@ -30,6 +30,7 @@ import configparser
from spider import (
get_douyin_stream_data,
get_douyin_app_stream_data,
get_tiktok_stream_data,
get_kuaishou_stream_data,
get_huya_stream_data,
@ -51,27 +52,24 @@ from spider import (
get_popkontv_stream_url,
get_twitcasting_stream_url,
get_baidu_stream_data,
get_weibo_stream_url,
get_weibo_stream_data,
get_kugou_stream_url,
get_twitchtv_stream_data,
get_liveme_stream_url,
get_huajiao_stream_url,
get_liuxing_stream_url,
get_showroom_stream_data
get_showroom_stream_data,
get_acfun_stream_data
)
from web_rid import (
get_live_room_id,
get_sec_user_id
)
from utils import (
logger, check_md5,
trace_error_decorator
)
from msg_push import dingtalk, xizhi, tg_bot
version = "v3.0.5"
platforms = "\n国内站点：抖音|快手|虎牙|斗鱼|YY|B站|小红书|bigo|blued|网易CC|千度热播|猫耳FM|Look|TwitCasting|百度|微博|酷狗|LiveMe|花椒|流星|ShowRoom" \
version = "v3.0.6"
platforms = "\n国内站点：抖音|快手|虎牙|斗鱼|YY|B站|小红书|bigo|blued|网易CC|千度热播|猫耳FM|Look|TwitCasting|百度|微博|酷狗|LiveMe|花椒|流星|ShowRoom|Acfun" \
"\n海外站点：TikTok|AfreecaTV|PandaTV|WinkTV|FlexTV|PopkonTV|TwitchTV"
recording = set()
@ -111,7 +109,6 @@ signal.signal(signal.SIGTERM, signal_handler)
def display_info():
# TODO: 显示当前录制配置信息
global start_display_time
global recording_time_list
time.sleep(5)
@ -207,7 +204,6 @@ def converts_m4a(address: str):
def create_ass_file(filegruop: list):
# TODO: 录制时生成ass格式的字幕文件
anchor_name = filegruop[0]
ass_filename = filegruop[1]
index_time = -1
@ -226,7 +222,6 @@ def create_ass_file(filegruop: list):
if anchor_name not in recording:
finish += 1
offset = datetime.timedelta(seconds=1)
# 获取修改后的时间并格式化
re_datatime = (today + offset).strftime('%Y-%m-%d %H:%M:%S')
today = today + offset
else:
@ -275,8 +270,6 @@ def change_max_connect():
@trace_error_decorator
def get_douyin_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取抖音直播源地址
anchor_name = json_data.get('anchor_name', None)
result = {
@ -310,8 +303,6 @@ def get_douyin_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]
@trace_error_decorator
def get_tiktok_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取tiktok直播源地址
if not json_data:
return {"anchor_name": None, "is_live": False}
@ -347,8 +338,6 @@ def get_tiktok_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]
stream_data = json.loads(stream_data).get('data', {})
flv_url_list = get_video_quality_url(stream_data, 'flv')
m3u8_url_list = get_video_quality_url(stream_data, 'hls')
# for item in flv_url_list:
# print(f"FLV URL: {item['url']}, VBitrate: {item['vbitrate']} Resolution: {item['resolution']}")
while len(flv_url_list) < 5:
flv_url_list.append(flv_url_list[-1])
@ -365,8 +354,6 @@ def get_tiktok_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]
@trace_error_decorator
def get_kuaishou_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取快手直播源地址
if json_data['type'] == 1 and not json_data["is_live"]:
return json_data
live_status = json_data['is_live']
@ -404,8 +391,6 @@ def get_kuaishou_stream_url(json_data: dict, video_quality: str) -> Dict[str, An
@trace_error_decorator
def get_huya_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取虎牙直播源地址
game_live_info = json_data.get('data', [])[0].get('gameLiveInfo', {})
stream_info_list = json_data.get('data', [])[0].get('gameStreamInfoList', [])
anchor_name = game_live_info.get('nick', '')
@ -494,7 +479,6 @@ def get_huya_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
@trace_error_decorator
def get_douyu_stream_url(json_data: dict, cookies: str, video_quality: str, proxy_address: str) -> Dict[str, Any]:
# TODO: 获取斗鱼直播源地址
if not json_data["is_live"]:
return json_data
@ -522,7 +506,6 @@ def get_douyu_stream_url(json_data: dict, cookies: str, video_quality: str, prox
@trace_error_decorator
def get_yy_stream_url(json_data: dict) -> Dict[str, Any]:
# TODO: 获取YY直播源地址
anchor_name = json_data.get('anchor_name', '')
result = {
"anchor_name": anchor_name,
@ -540,7 +523,6 @@ 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]:
# TODO: 获取B站直播源地址
if "is_live" in json_data and not json_data['anchor_name']:
return json_data
@ -603,31 +585,8 @@ def get_bilibili_stream_url(json_data: dict, video_quality: str) -> Dict[str, An
return result
@trace_error_decorator
def get_afreecatv_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取afreecatv直播源地址
if not json_data['is_live']:
return json_data
play_url_list = json_data['play_url_list']
quality_list = {'原画': 0, '蓝光': 0, '超清': 1, '高清': 2, '标清': 3, '流畅': 4}
while len(play_url_list) < 5:
play_url_list.append(play_url_list[-1])
selected_quality = quality_list[video_quality]
m3u8_url = play_url_list[selected_quality]
return {
"anchor_name": json_data['anchor_name'],
"is_live": True,
"m3u8_url": json_data['m3u8_url'],
"record_url": m3u8_url
}
@trace_error_decorator
def get_netease_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取netease直播源地址
if not json_data['is_live']:
return json_data
stream_list = json_data['stream_list']['resolution']
@ -648,9 +607,8 @@ def get_netease_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any
}
@trace_error_decorator
def get_pandatv_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取pandatv直播源地址
def get_stream_url(json_data: dict, video_quality: str, url_type: str = 'm3u8', spec: bool = False,
extra_key: Union[str, int] = None) -> Dict[str, Any]:
if not json_data['is_live']:
return json_data
@ -660,101 +618,20 @@ def get_pandatv_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any
play_url_list.append(play_url_list[-1])
selected_quality = quality_list[video_quality]
m3u8_url = play_url_list[selected_quality]
return {
data = {
"anchor_name": json_data['anchor_name'],
"is_live": True,
"m3u8_url": json_data['m3u8_url'],
"record_url": m3u8_url
"is_live": True
}
if url_type == 'm3u8':
m3u8_url = play_url_list[selected_quality][extra_key] if extra_key else play_url_list[selected_quality]
data["m3u8_url"] = m3u8_url
data["record_url"] = json_data['m3u8_url'] if spec else m3u8_url
else:
flv = play_url_list[selected_quality][extra_key] if extra_key else play_url_list[selected_quality]
data["m3u8_url"] = flv
data["record_url"] = flv
@trace_error_decorator
def get_winktv_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取winktv直播源地址
if not json_data['is_live']:
return json_data
play_url_list = json_data['play_url_list']
quality_list = {'原画': 0, '蓝光': 0, '超清': 1, '高清': 2, '标清': 3, '流畅': 4}
while len(play_url_list) < 5:
play_url_list.append(play_url_list[-1])
selected_quality = quality_list[video_quality]
m3u8_url = play_url_list[selected_quality]
return {
"anchor_name": json_data['anchor_name'],
"is_live": True,
"m3u8_url": json_data['m3u8_url'],
"record_url": m3u8_url
}
@trace_error_decorator
def get_flextv_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取flextv直播源地址
if not json_data['is_live']:
return json_data
play_url_list = json_data['play_url_list']
quality_list = {'原画': 0, '蓝光': 0, '超清': 1, '高清': 2, '标清': 3, '流畅': 4}
while len(play_url_list) < 5:
play_url_list.append(play_url_list[-1])
selected_quality = quality_list[video_quality]
m3u8_url = play_url_list[selected_quality]
return {
"anchor_name": json_data['anchor_name'],
"is_live": True,
"m3u8_url": json_data['m3u8_url'],
"record_url": m3u8_url
}
@trace_error_decorator
def get_baidu_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取百度直播源地址
if not json_data['is_live']:
return json_data
play_url_list = json_data['play_url_list']
quality_list = {'原画': 0, '蓝光': 0, '超清': 1, '高清': 2, '标清': 3, '流畅': 4}
while len(play_url_list) < 5:
play_url_list.append(play_url_list[-1])
selected_quality = quality_list[video_quality]
m3u8_url = play_url_list[selected_quality]
return {
"anchor_name": json_data['anchor_name'],
"is_live": True,
"m3u8_url": m3u8_url,
"record_url": m3u8_url
}
def get_twitchtv_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
# TODO: 获取twitchtv直播源地址
if not json_data['is_live']:
return json_data
play_url_list = json_data['play_url_list']
quality_list = {'原画': 0, '蓝光': 0, '超清': 1, '高清': 2, '标清': 3, '流畅': 4}
while len(play_url_list) < 5:
play_url_list.append(play_url_list[-1])
selected_quality = quality_list[video_quality]
m3u8_url = play_url_list[selected_quality]
return {
"anchor_name": json_data['anchor_name'],
"is_live": True,
"m3u8_url": json_data['m3u8_url'],
"record_url": m3u8_url
}
return data
def push_message(content: str) -> Union[str, list]:
@ -811,36 +688,25 @@ def start_record(url_data: tuple, count_variable: int = -1):
while True:
try:
port_info = []
if record_url.find("https://live.douyin.com/") > -1:
if record_url.find("douyin.com/") > -1:
platform = '抖音直播'
# 判断如果是浏览器长链接
with semaphore:
json_data = get_douyin_stream_data(
url=record_url,
proxy_addr=proxy_address,
cookies=dy_cookie)
port_info = get_douyin_stream_url(json_data, record_quality)
elif record_url.find("https://v.douyin.com/") > -1:
platform = '抖音直播'
# 判断如果是app分享链接
is_long_url = True
room_id, sec_user_id = get_sec_user_id(url=record_url, proxy_addr=proxy_address)
web_rid = get_live_room_id(room_id, sec_user_id, proxy_addr=proxy_address)
if len(web_rid) == 0:
print('web_rid 获取失败，若多次失败请联系作者修复或者使用浏览器打开后的长链接')
new_record_url = "https://live.douyin.com/" + str(web_rid)
not_record_list.append(new_record_url)
with semaphore:
json_data = get_douyin_stream_data(
url=new_record_url,
proxy_addr=proxy_address,
cookies=dy_cookie)
if 'live.douyin.com' in record_url:
json_data = get_douyin_stream_data(
url=record_url,
proxy_addr=proxy_address,
cookies=dy_cookie)
else:
json_data = get_douyin_app_stream_data(
url=record_url,
proxy_addr=proxy_address,
cookies=dy_cookie)
port_info = get_douyin_stream_url(json_data, record_quality)
elif record_url.find("https://www.tiktok.com/") > -1:
platform = 'TikTok直播'
with semaphore:
if global_proxy or proxy_address:
json_data = get_tiktok_stream_data(
url=record_url,
@ -922,7 +788,7 @@ def start_record(url_data: tuple, count_variable: int = -1):
username=afreecatv_username,
password=afreecatv_password
)
port_info = get_afreecatv_stream_url(json_data, record_quality)
port_info = get_stream_url(json_data, record_quality, spec=True)
else:
logger.error(f"错误信息: 网络异常，请检查本网络是否能正常访问AfreecaTV平台")
@ -947,7 +813,7 @@ def start_record(url_data: tuple, count_variable: int = -1):
proxy_addr=proxy_address,
cookies=pandatv_cookie
)
port_info = get_pandatv_stream_url(json_data, record_quality)
port_info = get_stream_url(json_data, record_quality, spec=True)
else:
logger.error(f"错误信息: 网络异常，请检查本网络是否能正常访问PandaTV直播平台")
@ -965,7 +831,7 @@ def start_record(url_data: tuple, count_variable: int = -1):
url=record_url,
proxy_addr=proxy_address,
cookies=winktv_cookie)
port_info = get_winktv_stream_url(json_data, record_quality)
port_info = get_stream_url(json_data, record_quality, spec=True)
else:
logger.error(f"错误信息: 网络异常，请检查本网络是否能正常访问WinkTV直播平台")
@ -980,7 +846,7 @@ def start_record(url_data: tuple, count_variable: int = -1):
username=flextv_username,
password=flextv_password
)
port_info = get_flextv_stream_url(json_data, record_quality)
port_info = get_stream_url(json_data, record_quality, spec=True)
else:
logger.error(f"错误信息: 网络异常，请检查本网络是否能正常访问FlexTV直播平台")
@ -1025,13 +891,14 @@ def start_record(url_data: tuple, count_variable: int = -1):
url=record_url,
proxy_addr=proxy_address,
cookies=baidu_cookie)
port_info = get_baidu_stream_url(json_data, record_quality)
port_info = get_stream_url(json_data, record_quality)
elif record_url.find("weibo.com/") > -1:
platform = '微博直播'
with semaphore:
port_info = get_weibo_stream_url(
json_data = get_weibo_stream_data(
url=record_url, proxy_addr=proxy_address, cookies=weibo_cookie)
port_info = get_stream_url(json_data, record_quality, extra_key='m3u8_url')
elif record_url.find("kugou.com/") > -1:
platform = '酷狗直播'
@ -1048,7 +915,7 @@ def start_record(url_data: tuple, count_variable: int = -1):
proxy_addr=proxy_address,
cookies=twitch_cookie
)
port_info = get_twitchtv_stream_url(json_data, record_quality)
port_info = get_stream_url(json_data, record_quality, spec=True)
else:
logger.error(f"错误信息: 网络异常，请检查本网络是否能正常访问TwitchTV直播平台")
@ -1075,7 +942,15 @@ def start_record(url_data: tuple, count_variable: int = -1):
with semaphore:
json_data = get_showroom_stream_data(
url=record_url, proxy_addr=proxy_address, cookies=showroom_cookie)
port_info = get_twitchtv_stream_url(json_data, record_quality)
port_info = get_stream_url(json_data, record_quality, spec=True)
elif record_url.find("live.acfun.cn/") > -1 or record_url.find("m.acfun.cn/") > -1:
platform = 'Acfun'
with semaphore:
json_data = get_acfun_stream_data(
url=record_url, proxy_addr=proxy_address, cookies=acfun_cookie)
port_info = get_stream_url(json_data, record_quality, url_type='flv', extra_key='url')
else:
logger.error(f'{record_url} 未知直播地址')
return
@ -1574,7 +1449,6 @@ def backup_file(file_path: str, backup_dir_path: str):
shutil.copy2(file_path, backup_file_path)
# print(f'\r已备份配置文件 {file_path} 到 {backup_file_path}')
# 删除多余的备份文件
files = os.listdir(backup_dir_path)
url_files = [f for f in files if f.startswith('URL_config.ini')]
config_files = [f for f in files if f.startswith('config.ini')]
@ -1789,6 +1663,7 @@ while True:
huajiao_cookie = read_config_value(config, 'Cookie', 'huajiao_cookie', '')
liuxing_cookie = read_config_value(config, 'Cookie', 'liuxing_cookie', '')
showroom_cookie = read_config_value(config, 'Cookie', 'showroom_cookie', '')
acfun_cookie = read_config_value(config, 'Cookie', 'acfun_cookie', '')
if len(video_save_type) > 0:
if video_save_type.upper().lower() == "FLV".lower():
@ -1884,6 +1759,8 @@ while True:
'www.7u66.com',
'wap.7u66.com',
'www.showroom-live.com',
'live.acfun.cn',
'm.acfun.cn'
]
overseas_platform_host = [
'www.tiktok.com',
@ -1958,4 +1835,4 @@ while True:
first_run = False
time.sleep(3)
time.sleep(3)

176
spider.py
View File

@ -4,13 +4,14 @@
Author: Hmily
GitHub: https://github.com/ihmily
Date: 2023-07-15 23:15:00
Update: 2024-06-13 21:19:48
Update: 2024-06-18 05:47:10
Copyright (c) 2023 by Hmily, All Rights Reserved.
Function: Get live stream data.
"""
import gzip
import hashlib
import random
import string
import time
import urllib.parse
import urllib.error
@ -28,6 +29,7 @@ from utils import (
)
from logger import script_path
import http.cookiejar
from web_rid import get_sec_user_id
no_proxy_handler = urllib.request.ProxyHandler({})
opener = urllib.request.build_opener(no_proxy_handler)
@ -115,6 +117,12 @@ def get_params(url: str, params: str) -> Union[str, None]:
return query_params[params][0]
def generate_random_string(length):
characters = string.ascii_uppercase + string.digits
random_string = ''.join(random.choices(characters, k=length))
return random_string
def jsonp_to_json(jsonp_str: str) -> Union[dict, None]:
pattern = r'(\w+)\((.*)\);?$'
match = re.search(pattern, jsonp_str)
@ -153,6 +161,61 @@ def get_play_url_list(m3u8: str, proxy: Union[str, None] = None, header: Union[d
return play_url_list
@trace_error_decorator
def get_douyin_app_stream_data(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:109.0) Gecko/20100101 Firefox/115.0',
'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',
'Referer': 'https://live.douyin.com/',
'Cookie': 'ttwid=1%7CB1qls3GdnZhUov9o2NxOMxxYS2ff6OSvEWbv0ytbES4%7C1680522049%7C280d802d6d478e3e78d0c807f7c487e7ffec0ae4e5fdd6a0fe74c3c6af149511; my_rd=1; passport_csrf_token=3ab34460fa656183fccfb904b16ff742; passport_csrf_token_default=3ab34460fa656183fccfb904b16ff742; d_ticket=9f562383ac0547d0b561904513229d76c9c21; n_mh=hvnJEQ4Q5eiH74-84kTFUyv4VK8xtSrpRZG1AhCeFNI; store-region=cn-fj; store-region-src=uid; LOGIN_STATUS=1; __security_server_data_status=1; FORCE_LOGIN=%7B%22videoConsumedRemainSeconds%22%3A180%7D; pwa2=%223%7C0%7C3%7C0%22; download_guide=%223%2F20230729%2F0%22; volume_info=%7B%22isUserMute%22%3Afalse%2C%22isMute%22%3Afalse%2C%22volume%22%3A0.6%7D; strategyABtestKey=%221690824679.923%22; stream_recommend_feed_params=%22%7B%5C%22cookie_enabled%5C%22%3Atrue%2C%5C%22screen_width%5C%22%3A1536%2C%5C%22screen_height%5C%22%3A864%2C%5C%22browser_online%5C%22%3Atrue%2C%5C%22cpu_core_num%5C%22%3A8%2C%5C%22device_memory%5C%22%3A8%2C%5C%22downlink%5C%22%3A10%2C%5C%22effective_type%5C%22%3A%5C%224g%5C%22%2C%5C%22round_trip_time%5C%22%3A150%7D%22; VIDEO_FILTER_MEMO_SELECT=%7B%22expireTime%22%3A1691443863751%2C%22type%22%3Anull%7D; home_can_add_dy_2_desktop=%221%22; __live_version__=%221.1.1.2169%22; device_web_cpu_core=8; device_web_memory_size=8; xgplayer_user_id=346045893336; csrf_session_id=2e00356b5cd8544d17a0e66484946f28; odin_tt=724eb4dd23bc6ffaed9a1571ac4c757ef597768a70c75fef695b95845b7ffcd8b1524278c2ac31c2587996d058e03414595f0a4e856c53bd0d5e5f56dc6d82e24004dc77773e6b83ced6f80f1bb70627; __ac_nonce=064caded4009deafd8b89; __ac_signature=_02B4Z6wo00f01HLUuwwAAIDBh6tRkVLvBQBy9L-AAHiHf7; ttcid=2e9619ebbb8449eaa3d5a42d8ce88ec835; webcast_leading_last_show_time=1691016922379; webcast_leading_total_show_times=1; webcast_local_quality=sd; live_can_add_dy_2_desktop=%221%22; msToken=1JDHnVPw_9yTvzIrwb7cQj8dCMNOoesXbA_IooV8cezcOdpe4pzusZE7NB7tZn9TBXPr0ylxmv-KMs5rqbNUBHP4P7VBFUu0ZAht_BEylqrLpzgt3y5ne_38hXDOX8o=; msToken=jV_yeN1IQKUd9PlNtpL7k5vthGKcHo0dEh_QPUQhr8G3cuYv-Jbb4NnIxGDmhVOkZOCSihNpA2kvYtHiTW25XNNX_yrsv5FN8O6zm3qmCIXcEe0LywLn7oBO2gITEeg=; tt_scid=mYfqpfbDjqXrIGJuQ7q-DlQJfUSG51qG.KUdzztuGP83OjuVLXnQHjsz-BRHRJu4e986'
}
if cookies:
headers['Cookie'] = cookies
def get_app_data():
room_id, sec_uid = get_sec_user_id(url=url, proxy_addr=proxy_addr)
api2 = f'https://webcast.amemv.com/webcast/room/reflow/info/?verifyFp=verify_lxj5zv70_7szNlAB7_pxNY_48Vh_ALKF_GA1Uf3yteoOY&type_id=0&live_id=1&room_id={room_id}&sec_user_id={sec_uid}&version_code=99.99.99&app_id=1128'
json_str2 = get_req(url=api2, proxy_addr=proxy_addr, headers=headers)
json_data2 = json.loads(json_str2)['data']
room_data2 = json_data2['room']
room_data2['anchor_name'] = room_data2['owner']['nickname']
return room_data2
try:
web_rid = url.split('?')[0].split('live.douyin.com/')
if len(web_rid) > 1:
web_rid = web_rid[1]
api = f'https://live.douyin.com/webcast/room/web/enter/?aid=6383&app_name=douyin_web&live_id=1&device_platform=web&language=zh-CN&browser_language=zh-CN&browser_platform=Win32&browser_name=Chrome&browser_version=116.0.0.0&web_rid={web_rid}'
json_str = get_req(url=api, proxy_addr=proxy_addr, headers=headers)
json_data = json.loads(json_str)['data']
room_data = json_data['data'][0]
room_data['anchor_name'] = json_data['user']['nickname']
else:
room_data = get_app_data()
if 'stream_url' not in room_data:
raise RuntimeError('该直播类型或玩法电脑端暂未支持，请使用app端分享链接进行录制')
live_core_sdk_data = room_data['stream_url']['live_core_sdk_data']
if room_data['status'] == 2:
if live_core_sdk_data:
json_str = live_core_sdk_data['pull_data']['stream_data']
json_data = json.loads(json_str)
if 'origin' in json_data['data']:
origin_url_list = json_data['data']['origin']['main']
origin_m3u8 = {'ORIGIN': origin_url_list["hls"]}
origin_flv = {'ORIGIN': origin_url_list["flv"]}
hls_pull_url_map = room_data['stream_url']['hls_pull_url_map']
flv_pull_url = room_data['stream_url']['flv_pull_url']
room_data['stream_url']['hls_pull_url_map'] = {**origin_m3u8, **hls_pull_url_map}
room_data['stream_url']['flv_pull_url'] = {**origin_flv, **flv_pull_url}
except Exception as e:
print(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}")
room_data = {'anchor_name': ""}
return room_data
@trace_error_decorator
def get_douyin_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
@ -203,26 +266,8 @@ def get_douyin_stream_data(url: str, proxy_addr: Union[str, None] = None, cookie
return json_data
except Exception as e:
print(f'失败地址：{url} 准备切换解析方法{e}')
web_rid = re.match('https://live.douyin.com/(\d+)', url).group(1)
url2 = f'https://live.douyin.com/webcast/room/web/enter/?aid=6383&app_name=douyin_web&live_id=1&device_platform=web&language=zh-CN&browser_language=zh-CN&browser_platform=Win32&browser_name=Chrome&browser_version=116.0.0.0&web_rid={web_rid}'
json_str = get_req(url=url2, proxy_addr=proxy_addr, headers=headers)
json_data = json.loads(json_str)['data']
room_data = json_data['data'][0]
room_data['anchor_name'] = json_data['user']['nickname']
live_core_sdk_data = room_data['stream_url']['live_core_sdk_data']
if live_core_sdk_data:
json_str = live_core_sdk_data['pull_data']['stream_data']
json_data = json.loads(json_str)
if 'origin' in json_data['data']:
origin_url_list = json_data['data']['origin']['main']
origin_m3u8 = {'ORIGIN': origin_url_list["hls"]}
origin_flv = {'ORIGIN': origin_url_list["flv"]}
hls_pull_url_map = room_data['stream_url']['hls_pull_url_map']
flv_pull_url = room_data['stream_url']['flv_pull_url']
room_data['stream_url']['hls_pull_url_map'] = {**origin_m3u8, **hls_pull_url_map}
room_data['stream_url']['flv_pull_url'] = {**origin_flv, **flv_pull_url}
return room_data
print(f'第一次获取数据失败：{url} 准备切换解析方法{e}')
return get_douyin_app_stream_data(url=url, proxy_addr=proxy_addr, cookies=cookies)
@trace_error_decorator
@ -1469,7 +1514,8 @@ def get_popkontv_stream_url(
@trace_error_decorator
def login_twitcasting(
account_type: str, username: str, password: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None
account_type: str, username: str, password: str, proxy_addr: Union[str, None] = None,
cookies: Union[str, None] = None
) -> Union[str, None]:
headers = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
@ -1557,7 +1603,8 @@ def get_twitcasting_stream_url(
anchor_name, live_status = get_data(headers)
except AttributeError:
print('获取TwitCasting数据失败，正在尝试登录...')
new_cookie = login_twitcasting(account_type=account_type, username=username, password=password, proxy_addr=proxy_addr, cookies=cookies)
new_cookie = login_twitcasting(account_type=account_type, username=username, password=password,
proxy_addr=proxy_addr, cookies=cookies)
if not new_cookie:
raise RuntimeError('TwitCasting登录失败,请检查配置文件中的账号密码是否正确')
print('TwitCasting 登录成功！开始获取数据...')
@ -1636,9 +1683,8 @@ def get_baidu_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_weibo_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
def get_weibo_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
headers = {
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'Cookie': 'XSRF-TOKEN=qAP-pIY5V4tO6blNOhA4IIOD; SUB=_2AkMRNMCwf8NxqwFRmfwWymPrbI9-zgzEieKnaDFrJRMxHRl-yT9kqmkhtRB6OrTuX5z9N_7qk9C3xxEmNR-8WLcyo2PM; SUBP=0033WrSXqPxfM72-Ws9jqgMF55529P9D9WWemwcqkukCduUO11o9sBqA; WBPSESS=Wk6CxkYDejV3DDBcnx2LOXN9V1LjdSTNQPMbBDWe4lO2HbPmXG_coMffJ30T-Avn_ccQWtEYFcq9fab1p5RR6PEI6w661JcW7-56BszujMlaiAhLX-9vT4Zjboy1yf2l',
@ -1678,9 +1724,12 @@ def get_weibo_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
if live_status == 1:
result["is_live"] = True
play_url_list = json_data['data']['item']['stream_info']['pull']
result['m3u8_url'] = play_url_list['live_origin_hls_url']
result['flv_url'] = play_url_list['live_origin_flv_url']
result['record_url'] = play_url_list['live_origin_hls_url']
m3u8_url = play_url_list['live_origin_hls_url']
flv_url = play_url_list['live_origin_flv_url']
result['play_url_list'] = [
{"m3u8_url": m3u8_url, "flv_url": flv_url},
{"m3u8_url": m3u8_url.split('_')[0] + '.m3u8', "flv_url": flv_url.split('_')[0] + '.flv'}
]
return result
@ -2039,6 +2088,72 @@ def get_showroom_stream_data(url: str, proxy_addr: Union[str, None] = None, cook
return result
@trace_error_decorator
def get_acfun_sign_params(proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Tuple[Any, str, Any]:
did = f'web_{generate_random_string(16)}'
headers = {
'referer': 'https://live.acfun.cn/',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0',
'cookie': f'_did={did};',
}
if cookies:
headers['Cookie'] = cookies
data = {
'sid': 'acfun.api.visitor',
}
api = 'https://id.app.acfun.cn/rest/app/visitor/login'
json_str = get_req(api, data=data, proxy_addr=proxy_addr, headers=headers)
json_data = json.loads(json_str)
user_id = json_data["userId"]
visitor_st = json_data["acfun.api.visitor_st"]
return user_id, did, visitor_st
@trace_error_decorator
def get_acfun_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
headers = {
'referer': 'https://live.acfun.cn/live/17912421',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0',
}
if cookies:
headers['Cookie'] = cookies
author_id = url.split('?')[0].rsplit('/', maxsplit=1)[1]
user_info_api = f'https://live.acfun.cn/rest/pc-direct/user/userInfo?userId={author_id}'
json_str = get_req(user_info_api, proxy_addr=proxy_addr, headers=headers)
json_data = json.loads(json_str)
anchor_name = json_data['profile']['name']
status = 'liveId' in json_data['profile']
result = {"anchor_name": anchor_name, "is_live": False}
if status:
result["is_live"] = True
user_id, did, visitor_st = get_acfun_sign_params(proxy_addr=proxy_addr, cookies=cookies)
params = {
'subBiz': 'mainApp',
'kpn': 'ACFUN_APP',
'kpf': 'PC_WEB',
'userId': user_id,
'did': did,
'acfun.api.visitor_st': visitor_st,
}
data = {
'authorId': author_id,
'pullStreamType': 'FLV',
}
play_api = f'https://api.kuaishouzt.com/rest/zt/live/web/startPlay?{urllib.parse.urlencode(params)}'
json_str = get_req(play_api, data=data, proxy_addr=proxy_addr, headers=headers)
json_data = json.loads(json_str)
videoPlayRes = json_data['data']['videoPlayRes']
play_url_list = json.loads(videoPlayRes)['liveAdaptiveManifest'][0]['adaptationSet']['representation']
play_url_list = sorted(play_url_list, key=lambda x: x['bitrate'], reverse=True)
result['play_url_list'] = play_url_list
return result
if __name__ == '__main__':
# 尽量用自己的cookie，以避免默认的不可用导致无法获取数据
# 以下示例链接不保证时效性，请自行查看链接是否能正常访问
@ -2080,8 +2195,10 @@ if __name__ == '__main__':
# room_url = 'https://www.7u66.com/100960' # 流星直播
# room_url = 'https://www.showroom-live.com/room/profile?room_id=511033' # showroom
# room_url = 'https://www.showroom-live.com/r/TPS0728' # showroom
# room_url = 'https://live.acfun.cn/live/17912421' # Acfun
print(get_douyin_stream_data(room_url, proxy_addr=''))
# print(get_douyin_app_stream_data(room_url, proxy_addr=''))
# print(get_tiktok_stream_data(room_url, proxy_addr=''))
# print(get_kuaishou_stream_data2(room_url, proxy_addr=''))
# print(get_huya_stream_data(room_url, proxy_addr=''))
@ -2109,4 +2226,5 @@ if __name__ == '__main__':
# print(get_liveme_stream_url(room_url, proxy_addr=''))
# print(get_huajiao_stream_url(room_url, proxy_addr=''))
# print(get_liuxing_stream_url(room_url, proxy_addr=''))
# print(get_showroom_stream_data(room_url, proxy_addr=''))
# print(get_showroom_stream_data(room_url, proxy_addr=''))
# print(get_acfun_stream_data(room_url, proxy_addr=''))