github/DouyinLiveRecorder
1
0
Fork 0
mirror of https://github.com/ihmily/DouyinLiveRecorder.git synced 2026-03-22 15:39:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: migrate to Python 3.10+ type hints and syntax

Browse Source
This commit is contained in:
ihmily
2024-10-29 16:06:55 +08:00
parent f0007db68e
commit 56add9e5b8
13 changed files with 180 additions and 596 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
README.md
View File

@@ -43,7 +43,6 @@
- [x] 流星直播
- [x] ShowRoom
- [x] Acfun
- [x] 时光直播
- [x] 映客直播
- [x] 音播直播
- [x] 知乎直播
@@ -60,8 +59,8 @@
├── /config -> (config record)
├── /logs -> (save runing log file)
├── /backup_config -> (backup file)
├── /libs -> (dll file)
├── /douyinliverecorder -> (package)
├── initializer.py-> (check and install nodejs)
├── spider.py-> (get live data)
├── stream.py-> (get live stream address)
├── utils.py -> (contains utility functions)
@@ -109,7 +108,7 @@
https://live.douyin.com/745964462470
https://v.douyin.com/iQFeBnt/
https://live.douyin.com/yall1102 （链接+抖音号）
https://v.douyin.com/CeiU5cbX （作者主页地址）
https://v.douyin.com/CeiU5cbX （主播主页地址）
TikTok：
https://www.tiktok.com/@pearlgaga88/live
@@ -176,7 +175,6 @@ https://twitcasting.tv/c:uonq
https://live.baidu.com/m/media/pclive/pchome/live.html?room_id=9175031377&tab_category
微博直播:
https://weibo.com/u/7676267963 （主页地址）
https://weibo.com/l/wblive/p/show/1022:2321325026370190442592
酷狗直播:
@@ -189,20 +187,17 @@ LiveMe:
https://www.liveme.com/zh/v/17141543493018047815/index.html
花椒直播:
https://www.huajiao.com/user/223184650 （主页地址）
https://www.huajiao.com/l/345096174
流星直播:
https://www.7u66.com/100960
ShowRoom:
https://www.showroom-live.com/room/profile?room_id=480206 （主页地址）
https://www.showroom-live.com/room/profile?room_id=480206 （主播主页地址）
Acfun:
https://live.acfun.cn/live/179922
时光直播：
https://www.rengzu.com/180778
映客直播：
https://www.inke.cn/liveroom/index.html?uid=22954469&id=1720860391070904
@@ -210,7 +205,7 @@ https://www.inke.cn/liveroom/index.html?uid=22954469&id=1720860391070904
https://live.ybw1666.com/800002949
知乎直播:
https://www.zhihu.com/theater/114453
https://www.zhihu.com/people/ac3a467005c5d20381a82230101308e9 (主播主页地址)
CHZZK：
https://chzzk.naver.com/live/458f6ec20b034f49e0fc6d03921646d2
@@ -223,7 +218,7 @@ https://chzzk.naver.com/live/458f6ec20b034f49e0fc6d03921646d2
 
## 🎃源码运行
使用源码运行，前提要有Python环境，如果没有请先安装Python，再执行下面步骤。
使用源码运行，前提要有**Python>=3.10**环境，如果没有请先自行安装Python，再执行下面步骤。
1.首先拉取或手动下载本仓库项目代码

121
api/convert.php
View File

@@ -1,121 +0,0 @@
<?php
/**
* Author: Hmily
* Github:https://github.com/ihmily
* Date: 2023-07-20 21:06:20
* Update: 2023-09-07 22:34:57
* Copyright (c) 2023 by Hmily, All Rights Reserved.
* Function:convert short url to long url
* Address:https://github.com/ihmily/DouyinLiveRecorder
*/
// 该代码主要是用来转换地址，将短app端直播分享链接转为PC网页端长链接
header('Content-type: application/json; charset=utf-8');
if(empty($_GET['url'])){
exit(json_encode(['code'=>-2,'msg'=>'请输入app端直播间分享地址'],448));
}
$share_url=$_GET['url'];
$get_id=get_redirect_url($share_url);
preg_match('/reflow\/(.*?)\?/', $get_id, $room_id);
preg_match('/sec_user_id=([\w\d_\-]+)&/', $get_id, $sec_user_id);
$room_data=get_live_web_rid($room_id[1],$sec_user_id[1]);
$title=$room_data[0];
$web_rid=$room_data[1];
if(empty($web_rid)){
exit(json_encode(['code'=>-1,'status'=>'解析失败','msg'=>'请检测链接是否正确，多次失败请联系作者修复！https://github.com/ihmily/DouyinLiveRecorder'],448));
}
$long_url="https://live.douyin.com/".$web_rid;
$return=
[
'code'=>0,
'status'=>'解析成功',
'title'=>$title,
'room_id'=>$room_id[1],
'share_url'=>$share_url,
'long_url'=>$long_url,
'source'=>'源码地址：https://github.com/ihmily/DouyinLiveRecorder'
];
exit(json_encode($return,448));
// 抖音X-bogus算法，直接调用我封装的接口
function get_xbogus($url) {
$query = parse_url($url, PHP_URL_QUERY);
$url = "http://43.138.133.177:8890/xbogus";
$data = array(
'url' => $query,
'ua' => "Mozilla/5.0 (Linux; Android 11; SAMSUNG SM-G973U) AppleWebKit/537.36 (KHTML, like Gecko) SamsungBrowser/14.2 Chrome/87.0.4280.141 Mobile Safari/537.36",
);
$params = http_build_query($data);
$url = $url . '?' . $params;
$response = file_get_contents($url);
$response_json = json_decode($response, true);
$xbogus = $response_json['result'];
// echo "生成的X-Bogus签名为: " . $xbogus;
return $xbogus;
}
function get_live_web_rid($room_id, $sec_user_id) {
$url = 'https://webcast.amemv.com/webcast/room/reflow/info/?verifyFp=verify_lk07kv74_QZYCUApD_xhiB_405x_Ax51_GYO9bUIyZQVf&type_id=0&live_id=1&room_id='.$room_id.'&sec_user_id='.$sec_user_id.'&app_id=1128&msToken=wrqzbEaTlsxt52-vxyZo_mIoL0RjNi1ZdDe7gzEGMUTVh_HvmbLLkQrA_1HKVOa2C6gkxb6IiY6TY2z8enAkPEwGq--gM-me3Yudck2ailla5Q4osnYIHxd9dI4WtQ==';
$xbogus = get_xbogus($url); // 获取X-Bogus算法
$url = $url . "&X-Bogus=" . $xbogus;
$headers = array(
'User-Agent: Mozilla/5.0 (Linux; Android 11; SAMSUNG SM-G973U) AppleWebKit/537.36 (KHTML, like Gecko) SamsungBrowser/14.2 Chrome/87.0.4280.141 Mobile Safari/537.36',
'Accept:application/json, text/plain, */*',
'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',
);
$cookies='s_v_web_id=verify_lk07kv74_QZYCUApD_xhiB_405x_Ax51_GYO9bUIyZQVf';
$json_data = get_curl($url,$headers,$cookies);
$json_data = json_decode($json_data, true);
$web_rid = $json_data['data']['room']['owner']['web_rid'];
$title=$json_data["data"]['room']['title'];
return [$title,$web_rid];
}
// 封装的CURL函数
function get_curl($url,$headers=array(),$cookies=''){
$curl=curl_init((string)$url);
curl_setopt($curl,CURLOPT_HEADER,false);
curl_setopt($curl, CURLOPT_FOLLOWLOCATION, true);
curl_setopt($curl,CURLOPT_SSL_VERIFYPEER,false);
curl_setopt($curl, CURLOPT_ENCODING, "");
curl_setopt($curl,CURLOPT_RETURNTRANSFER,true);
curl_setopt($curl,CURLOPT_HTTPHEADER,$headers);
curl_setopt($curl, CURLOPT_COOKIE, $cookies);
curl_setopt($curl,CURLOPT_TIMEOUT,20);
$data = curl_exec($curl);
// var_dump($data);
curl_close($curl);
return $data;
}
function get_redirect_url($url) {
$curl = curl_init();
curl_setopt($curl, CURLOPT_URL, $url);
curl_setopt($curl, CURLOPT_SSL_VERIFYPEER, false);
curl_setopt($curl, CURLOPT_SSL_VERIFYHOST, false);
curl_setopt($curl, CURLOPT_HTTPHEADER, array( "User-Agent:Mozilla/5.0 (iPhone; CPU iPhone OS 11_0 like Mac OS X) AppleWebKit/604.1.38 (KHTML, like Gecko) Version/11.0 Mobile/15A372 Safari/604.1"));
curl_setopt($curl, CURLOPT_HEADER, true);
curl_setopt($curl, CURLOPT_NOBODY, 1);
curl_setopt($curl, CURLOPT_FOLLOWLOCATION, 1);
curl_setopt($curl, CURLOPT_RETURNTRANSFER, 1);
curl_setopt($curl,CURLOPT_TIMEOUT,20);
$ret = curl_exec($curl);
curl_close($curl);
preg_match("/Location: (.*?)\r\n/iU",$ret,$location);
return $location[1];
}

252
api/index.php
View File

@@ -1,252 +0,0 @@
<?php
/**
* Author: Hmily
* Github:https://github.com/ihmily
* Date: 2023-07-20 21:06:20
* Update: 2023-09-17 20:23:00
* Copyright (c) 2023 by Hmily, All Rights Reserved.
* Function:Spider the live stream url
* Address:https://github.com/ihmily/DouyinLiveRecorder
*/
// 本API代码只有解析抖音、快手和虎牙的，有需要其他的可自己根据源码增加
// 注意：抖音和快手的 要添加上自己的cookie才能用
header('Content-type: application/json; charset=utf-8');
if(empty($_GET['url'])){
exit(json_encode(['code'=>-2,'msg'=>'请输入抖音/快手/虎牙等平台的直播间地址'],448));
}
$live_url=$_GET['url'];
if (strpos($live_url, 'douyin') !== false) {
$pattern = "/^https:\/\/v\.douyin\.com\/\w+\/$/";
if (preg_match($pattern, $live_url)) {
// 判断是否是app端分享链接，如果是则转为PC网页端地址，否则无法解析
// 示例链接：
// $live_url="https://live.douyin.com/187615265444";
$json_str=get_curl("https://hmily.vip/api/jx/live/convert.php?url=".$live_url);
$json_data=json_decode($json_str,true);
$live_url = $json_data['long_url'];
}
$json_data2=get_douyin_json_data($live_url);
$return=get_douyin_stream_url2($json_data2,$live_url); // 选用第二种方式
} else if(strpos($live_url, 'kuaishou') !== false) {
$return=get_kuaishou_stream_url($live_url); // 选用第二种方式
}else if(strpos($live_url, 'huya') !== false) {
$return=get_huya_stream_url($live_url); // 选用第二种方式
}else{
$return=['code'=>-1,'msg'=>'暂不支持该平台，请检查链接是否正确'];
}
exit(json_encode($return,448));
function get_douyin_json_data($url) {
$headers = array(
'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/',
);
$cookies='your cookie'; # 任意抖音直播间页面的Cookie
$html_str = get_curl($url,$headers,$cookies);
$pattern = '/\{\\\"state(.*?)\"\]\)\<\/script\>\<div hidden/';
preg_match($pattern, $html_str, $matches);
$json_string = '{\"state' . explode(']\n', $matches[1])[0];
$cleaned_string = str_replace("\\", "", $json_string);
$cleaned_string = preg_replace('/bdp_log=(.*?)&bdpsum=/', '', $cleaned_string);
$replacements = array(
'"[' => '[',
']"' => ']',
'"{' => '{',
'}"' => '}',
'u0026' => '&'
);
$cleaned_string = strtr($cleaned_string, $replacements);
$json_data = json_decode($cleaned_string, true);
return $json_data;
}
# 第一种数据
function get_douyin_stream_url($json_data) {
$initialState = $json_data['state'];
$streamStore = $initialState['streamStore'];
$roomStore = $initialState['roomStore'];
$streamData = $streamStore['streamData']['H265_streamData']['options'];
$stream = $streamStore['streamData']['H265_streamData']['stream'];
$stream2 = $streamStore['streamData']['H264_streamData']['stream'];
$anchor_name = $roomStore['roomInfo']['anchor']['nickname'];
$data=array();
if ($stream === null) {
$data=["live_status"=>'主播未开播或者直播已经结束！'];
} else {
$m3u8_url = $stream['origin']['main']['hls'];
$m3u8_url2 = $stream2['origin']['main']['hls'];
$data=["title"=>$live_title,'stream'=>['m3u8_url_265'=>$m3u8_url,'m3u8_url_264'=>$m3u8_url2]];
}
$return=[
'code'=>0,
'status'=>'解析成功',
'anchor_name'=>$anchor_name,
'live_url'=>$live_url,
'data'=>$data,
'source'=>'源码地址：https://github.com/ihmily/DouyinLiveRecorder'
];
return $return;
}
# 第二种数据（更好）
function get_douyin_stream_url2($json_data,$live_url) {
$roomStore = $json_data['state']['roomStore'];
$roomInfo = $roomStore['roomInfo'];
$anchor_name = $roomInfo['anchor']['nickname'];
$live_title = $roomInfo['room']['title'] ;
// 获取直播间状态
$status = $roomInfo["room"]["status"]; // 直播状态2是正在直播.4是未开播
$data=array();
if ($status == 4) {
$data=["live_status"=>'主播未开播或者直播已经结束！'];
} else {
$stream_url = $roomInfo['room']['stream_url'];
// flv视频流链接
$flv_url_list = $stream_url['flv_pull_url'];
// m3u8视频流链接
$m3u8_url_list = $stream_url['hls_pull_url_map'];
$data=["title"=>$live_title,'stream'=>['flv_url_list'=>$flv_url_list,'m3u8_url_list'=>$m3u8_url_list]];
}
$return=[
'code'=>0,
'status'=>'解析成功',
'platform'=>'抖音直播',
'anchor_name'=>$anchor_name,
'live_url'=>$live_url,
'data'=>$data,
'source'=>'源码地址：https://github.com/ihmily/DouyinLiveRecorder'
];
return $return;
}
function get_kuaishou_stream_url($live_url){
$headers = array(
'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',
);
$cookies = 'your cookie'; # 任意快手直播间页面的cookie
$html_str = get_curl($live_url,$headers,$cookies);
preg_match('/__INITIAL_STATE__=(.*?);\(function/', $html_str, $matches);
$json_data = json_decode($matches[1], true);
$play_list= $json_data['liveroom']['playList'][0];
$live_title=$play_list['liveStream']['caption'];
$anchor_name = $play_list['author']['name'];
# 获取直播间状态
$status = $play_list['isLiving']; # 直播状态True是正在直播.False是未开播
if (!$status) {
$data=["live_status"=>'主播未开播或者直播已经结束！'];
}else{
$stream_data=$play_list['liveStream']['playUrls'][0]['adaptationSet']['representation'];
$data=["title"=>$live_title,'stream'=>$stream_data];
}
$return=[
'code'=>0,
'platform'=>'快手直播',
'status'=>'解析成功',
'anchor_name'=>$anchor_name,
'live_url'=>$live_url,
'data'=>$data,
'source'=>'源码地址：https://github.com/ihmily/DouyinLiveRecorder'
];
return $return;
}
function get_huya_stream_url($live_url){
$headers = array(
'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'
);
$html_str = get_curl($live_url,$headers);
preg_match('/stream: (\{"data".*?),"iWebDefaultBitRate"/', $html_str, $matches);
$json_data = json_decode($matches[1].'}', true);
$gameLiveInfo = $json_data['data'][0]['gameLiveInfo'];
$live_title=$gameLiveInfo['introduction'];
$gameStreamInfoList = $json_data['data'][0]['gameStreamInfoList'];
$anchor_name = $gameLiveInfo['nick'];
if (count($gameStreamInfoList)==0) {
$data=["live_status"=>'主播未开播或者直播已经结束！'];
}else{
# gameStreamInfoList 索引从小到大 分别是'al', 'tx', 'hw', 'hs'四种cdn线路
# 默认使用第二种 即host链接开头为tx的cdn
$sFlvUrl = $gameStreamInfoList[1]['sFlvUrl'];
$sStreamName = $gameStreamInfoList[1]['sStreamName'];
$sFlvUrlSuffix = $gameStreamInfoList[1]['sFlvUrlSuffix'];
$sHlsUrl = $gameStreamInfoList[1]['sHlsUrl'];
$sHlsUrlSuffix = $gameStreamInfoList[1]['sHlsUrlSuffix'];
$sFlvAntiCode = $gameStreamInfoList[1]['sFlvAntiCode'];
$quality_list = explode('&exsphd=', $sFlvAntiCode)[1];
$pattern = "/(?<=264_)\d+/";
$matches = [];
preg_match_all($pattern, $quality_list, $matches);
$quality_list = $matches[0];
$quality_list = array_reverse($quality_list);
$m3u8_list=[];
$flv_list=[];
foreach ($quality_list as $quality){
$flv_url = "{$sFlvUrl}/{$sStreamName}.{$sFlvUrlSuffix}?{$sFlvAntiCode}&ratio={$quality}";
$m3u8_url = "{$sHlsUrl}/{$sStreamName}.{$sHlsUrlSuffix}?{$sFlvAntiCode}&ratio={$quality}";
array_push($m3u8_list,$m3u8_url);
array_push($flv_list,$flv_url);
}
$data=["title"=>$live_title,'stream'=>['flv_url'=>$flv_list,'m3u8_url'=>$m3u8_list]];
}
$return=[
'code'=>0,
'platform'=>'虎牙直播',
'status'=>'解析成功',
'anchor_name'=>$anchor_name,
'live_url'=>$live_url,
'data'=>$data,
'source'=>'源码地址：https://github.com/ihmily/DouyinLiveRecorder'
];
return $return;
}
// 封装的CURL函数
function get_curl($url,$headers=array(),$cookies=''){
$curl=curl_init((string)$url);
curl_setopt($curl,CURLOPT_HEADER,false);
curl_setopt($curl, CURLOPT_FOLLOWLOCATION, true);
curl_setopt($curl,CURLOPT_SSL_VERIFYPEER,false);
curl_setopt($curl, CURLOPT_ENCODING, "");
curl_setopt($curl,CURLOPT_RETURNTRANSFER,true);
curl_setopt($curl,CURLOPT_HTTPHEADER,$headers);
curl_setopt($curl, CURLOPT_COOKIE, $cookies);
curl_setopt($curl,CURLOPT_TIMEOUT,20);
$data = curl_exec($curl);
// var_dump($data);
curl_close($curl);
return $data;
}

2
demo.py
View File

@@ -133,7 +133,7 @@ LIVE_STREAM_CONFIG = {
"func": spider.get_yinbo_stream_url,
},
"zhihu": {
"url": "https://www.zhihu.com/theater/114453",
"url": "https://www.zhihu.com/people/ac3a467005c5d20381a82230101308e9",
"func": spider.get_zhihu_stream_url,
},
"chzzk": {

11
douyinliverecorder/initializer.py
View File

@@ -23,7 +23,7 @@ execute_dir = os.path.split(os.path.realpath(sys.argv[0]))[0]
current_env_path = os.environ.get('PATH')
def unzip_file(zip_path, extract_to):
def unzip_file(zip_path: str | Path, extract_to: str | Path) -> None:
if not os.path.exists(extract_to):
os.makedirs(extract_to)
@@ -148,7 +148,7 @@ def install_nodejs_mac():
logger.error(f"An unexpected error occurred: {e}")
def install_nodejs():
def install_nodejs() -> bool:
if current_platform == "Windows":
return install_nodejs_windows()
elif current_platform == "Linux":
@@ -162,6 +162,7 @@ def install_nodejs():
else:
logger.debug(f"Node.js auto installation is not supported on this platform: {current_platform}. "
f"Please install Node.js manually.")
return False
def ensure_nodejs_installed(func):
@@ -192,7 +193,7 @@ def ensure_nodejs_installed(func):
return wrapped_func
def check_nodejs_installed():
def check_nodejs_installed() -> bool:
try:
result = subprocess.run(['node', '-v'], capture_output=True)
version = result.stdout.strip()
@@ -203,6 +204,6 @@ def check_nodejs_installed():
return False
def check_node():
def check_node() -> bool:
if not check_nodejs_installed():
return install_nodejs()
return install_nodejs()

16
douyinliverecorder/room.py
View File

@@ -10,7 +10,6 @@ Copyright (c) 2023 by Hmily, All Rights Reserved.
import json
import re
import urllib.parse
from typing import Union
import execjs
import requests
import urllib.request
@@ -36,18 +35,16 @@ HEADERS_PC = {
# X-bogus算法
def get_xbogus(url: str, headers: Union[dict, None] = None) -> str:
def get_xbogus(url: str, headers: dict | None = None) -> str:
if not headers or 'user-agent' not in (k.lower() for k in headers):
headers = HEADERS
query = urllib.parse.urlparse(url).query
xbogus = execjs.compile(open(f'{JS_SCRIPT_PATH}/x-bogus.js').read()).call('sign', query, headers.get("User-Agent", "user-agent"))
# print(xbogus)
return xbogus
# 获取房间ID和用户secID
def get_sec_user_id(url: str, proxy_addr: Union[str, None] = None,
headers: Union[dict, None] = None) -> Union[tuple, None]:
def get_sec_user_id(url: str, proxy_addr: str | None = None, headers: dict | None = None) -> tuple | None:
if not headers or all(k.lower() not in ['user-agent', 'cookie'] for k in headers):
headers = HEADERS
@@ -67,8 +64,7 @@ def get_sec_user_id(url: str, proxy_addr: Union[str, None] = None,
# 获取抖音号
def get_unique_id(url: str, proxy_addr: Union[str, None] = None,
headers: Union[dict, None] = None) -> Union[str, None]:
def get_unique_id(url: str, proxy_addr: str | None = None, headers: dict | None = None) -> str:
if not headers or all(k.lower() not in ['user-agent', 'cookie'] for k in headers):
headers = HEADERS_PC
@@ -88,8 +84,8 @@ def get_unique_id(url: str, proxy_addr: Union[str, None] = None,
# 获取直播间webID
def get_live_room_id(room_id: str, sec_user_id: str, proxy_addr: Union[str, None] = None,
params: Union[dict, None] = None, headers: Union[dict, None] = None) -> str:
def get_live_room_id(room_id: str, sec_user_id: str, proxy_addr: str | None = None,
params: dict | None = None, headers: dict | None = None) -> str:
if not headers or all(k.lower() not in ['user-agent', 'cookie'] for k in headers):
headers = HEADERS
@@ -127,4 +123,4 @@ if __name__ == '__main__':
room_url = "https://v.douyin.com/iQLgKSj/"
_room_id, sec_uid = get_sec_user_id(room_url)
web_rid = get_live_room_id(_room_id, sec_uid)
print("return web_rid:", web_rid)
print("return web_rid:", web_rid)

281
douyinliverecorder/spider.py
View File

@@ -17,7 +17,7 @@ from operator import itemgetter
import urllib.parse
import urllib.error
from urllib.request import Request
from typing import Union, Dict, Any, Tuple, List
from typing import List
import requests
import ssl
import re
@@ -38,19 +38,21 @@ opener = urllib.request.build_opener(no_proxy_handler)
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
OptionalStr = str | None
OptionalDict = dict | None
def get_req(
url: str,
proxy_addr: Union[str, None] = None,
headers: Union[dict, None] = None,
data: Union[dict, bytes, None] = None,
json_data: Union[dict, list, None] = None,
proxy_addr: OptionalStr = None,
headers: OptionalDict = None,
data: dict | bytes | None = None,
json_data: dict | list | None = None,
timeout: int = 20,
abroad: bool = False,
content_conding: str = 'utf-8',
redirect_url: bool = False,
) -> Union[str, Any]:
) -> str:
if headers is None:
headers = {}
try:
@@ -60,8 +62,9 @@ def get_req(
'https': proxy_addr
}
if data or json_data:
response = requests.post(url, data=data, json=json_data, headers=headers, proxies=proxies,
timeout=timeout)
response = requests.post(
url, data=data, json=json_data, headers=headers, proxies=proxies, timeout=timeout
)
else:
response = requests.get(url, headers=headers, proxies=proxies, timeout=timeout)
if redirect_url:
@@ -110,7 +113,7 @@ def get_req(
return resp_str
def get_params(url: str, params: str) -> Union[str, None]:
def get_params(url: str, params: str) -> OptionalStr:
parsed_url = urllib.parse.urlparse(url)
query_params = urllib.parse.parse_qs(parsed_url.query)
@@ -118,13 +121,13 @@ def get_params(url: str, params: str) -> Union[str, None]:
return query_params[params][0]
def generate_random_string(length):
def generate_random_string(length: int) -> str:
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]:
def jsonp_to_json(jsonp_str: str) -> OptionalDict:
pattern = r'(\w+)\((.*)\);?$'
match = re.search(pattern, jsonp_str)
@@ -144,7 +147,7 @@ def replace_url(file_path: str, old: str, new: str) -> None:
f.write(content.replace(old, new))
def get_play_url_list(m3u8: str, proxy: Union[str, None] = None, header: Union[dict, None] = None,
def get_play_url_list(m3u8: str, proxy: OptionalStr = None, header: OptionalDict = None,
abroad: bool = False) -> List[str]:
resp = get_req(url=m3u8, proxy_addr=proxy, headers=header, abroad=abroad)
play_url_list = []
@@ -163,8 +166,7 @@ def get_play_url_list(m3u8: str, proxy: Union[str, None] = None, header: Union[d
@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]:
def get_douyin_app_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
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',
@@ -174,7 +176,7 @@ def get_douyin_app_stream_data(url: str, proxy_addr: Union[str, None] = None, co
if cookies:
headers['Cookie'] = cookies
def get_app_data(room_id, sec_uid):
def get_app_data(room_id: str, sec_uid: str) -> dict:
app_params = {
"verifyFp": "verify_lxj5zv70_7szNlAB7_pxNY_48Vh_ALKF_GA1Uf3yteoOY",
"type_id": "0",
@@ -250,8 +252,7 @@ def get_douyin_app_stream_data(url: str, proxy_addr: Union[str, None] = None, co
@trace_error_decorator
def get_douyin_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_douyin_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
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',
@@ -306,8 +307,7 @@ def get_douyin_stream_data(url: str, proxy_addr: Union[str, None] = None, cookie
@trace_error_decorator
def get_tiktok_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_tiktok_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict | None:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:124.0) Gecko/20100101 Firefox/124.0',
'Cookie': 'ttwid=1%7CM-rF193sJugKuNz2RGNt-rh6pAAR9IMceUSzlDnPCNI%7C1683274418%7Cf726d4947f2fc37fecc7aeb0cdaee52892244d04efde6f8a8edd2bb168263269; tiktok_webapp_theme=light; tt_chain_token=VWkygAWDlm1cFg/k8whmOg==; passport_csrf_token=6e422c5a7991f8cec7033a8082921510; passport_csrf_token_default=6e422c5a7991f8cec7033a8082921510; d_ticket=f8c267d4af4523c97be1ccb355e9991e2ae06; odin_tt=320b5f386cdc23f347be018e588873db7f7aea4ea5d1813681c3fbc018ea025dde957b94f74146dbc0e3612426b865ccb95ec8abe4ee36cca65f15dbffec0deff7b0e69e8ea536d46e0f82a4fc37d211; cmpl_token=AgQQAPNSF-RO0rT04baWtZ0T_jUjl4fVP4PZYM2QPw; uid_tt=319b558dbba684bb1557206c92089cd113a875526a89aee30595925d804b81c7; uid_tt_ss=319b558dbba684bb1557206c92089cd113a875526a89aee30595925d804b81c7; sid_tt=ad5e736f4bedb2f6d42ccd849e706b1d; sessionid=ad5e736f4bedb2f6d42ccd849e706b1d; sessionid_ss=ad5e736f4bedb2f6d42ccd849e706b1d; store-idc=useast5; store-country-code=us; store-country-code-src=uid; tt-target-idc=useast5; tt-target-idc-sign=qXNk0bb1pDQ0FbCNF120Pl9WWMLZg9Edv5PkfyCbS4lIk5ieW5tfLP7XWROnN0mEaSlc5hg6Oji1pF-yz_3ZXnUiNMrA9wNMPvI6D9IFKKVmq555aQzwPIGHv0aQC5dNRgKo5Z5LBkgxUMWEojTKclq2_L8lBciw0IGdhFm_XyVJtbqbBKKgybGDLzK8ZyxF4Jl_cYRXaDlshZjc38JdS6wruDueRSHe7YvNbjxCnApEFUv-OwJANSPU_4rvcqpVhq3JI2VCCfw-cs_4MFIPCDOKisk5EhAo2JlHh3VF7_CLuv80FXg_7ZqQ2pJeMOog294rqxwbbQhl3ATvjQV_JsWyUsMd9zwqecpylrPvtySI2u1qfoggx1owLrrUynee1R48QlanLQnTNW_z1WpmZBgVJqgEGLwFoVOmRzJuFFNj8vIqdjM2nDSdWqX8_wX3wplohkzkPSFPfZgjzGnQX28krhgTytLt7BXYty5dpfGtsdb11WOFHM6MZ9R9uLVB; sid_guard=ad5e736f4bedb2f6d42ccd849e706b1d%7C1690990657%7C15525213%7CMon%2C+29-Jan-2024+08%3A11%3A10+GMT; sid_ucp_v1=1.0.0-KGM3YzgwYjZhODgyYWI1NjIwNTA0NjBmOWUxMGRhMjIzYTI2YjMxNDUKGAiqiJ30keKD5WQQwfCppgYYsws4AkDsBxAEGgd1c2Vhc3Q1IiBhZDVlNzM2ZjRiZWRiMmY2ZDQyY2NkODQ5ZTcwNmIxZA; ssid_ucp_v1=1.0.0-KGM3YzgwYjZhODgyYWI1NjIwNTA0NjBmOWUxMGRhMjIzYTI2YjMxNDUKGAiqiJ30keKD5WQQwfCppgYYsws4AkDsBxAEGgd1c2Vhc3Q1IiBhZDVlNzM2ZjRiZWRiMmY2ZDQyY2NkODQ5ZTcwNmIxZA; tt_csrf_token=dD0EIH8q-pe3qDQsCyyD1jLN6KizJDRjOEyk; __tea_cache_tokens_1988={%22_type_%22:%22default%22%2C%22user_unique_id%22:%227229608516049831425%22%2C%22timestamp%22:1683274422659}; ttwid=1%7CM-rF193sJugKuNz2RGNt-rh6pAAR9IMceUSzlDnPCNI%7C1694002151%7Cd89b77afc809b1a610661a9d1c2784d80ebef9efdd166f06de0d28e27f7e4efe; msToken=KfJAVZ7r9D_QVeQlYAUZzDFbc1Yx-nZz6GF33eOxgd8KlqvTg1lF9bMXW7gFV-qW4MCgUwnBIhbiwU9kdaSpgHJCk-PABsHCtTO5J3qC4oCTsrXQ1_E0XtbqiE4OVLZ_jdF1EYWgKNPT2SnwGkQ=; msToken=KfJAVZ7r9D_QVeQlYAUZzDFbc1Yx-nZz6GF33eOxgd8KlqvTg1lF9bMXW7gFV-qW4MCgUwnBIhbiwU9kdaSpgHJCk-PABsHCtTO5J3qC4oCTsrXQ1_E0XtbqiE4OVLZ_jdF1EYWgKNPT2SnwGkQ='
@@ -333,8 +333,7 @@ def get_tiktok_stream_data(url: str, proxy_addr: Union[str, None] = None, cookie
@trace_error_decorator
def get_kuaishou_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_kuaishou_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
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',
@@ -377,8 +376,7 @@ def get_kuaishou_stream_data(url: str, proxy_addr: Union[str, None] = None, cook
@trace_error_decorator
def get_kuaishou_stream_data2(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_kuaishou_stream_data2(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict | None:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'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',
@@ -421,8 +419,7 @@ def get_kuaishou_stream_data2(url: str, proxy_addr: Union[str, None] = None, coo
@trace_error_decorator
def get_huya_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_huya_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
@@ -439,8 +436,7 @@ def get_huya_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies:
@trace_error_decorator
def get_huya_app_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_huya_app_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'xweb_xhr': '1',
@@ -509,11 +505,11 @@ def get_huya_app_stream_url(url: str, proxy_addr: Union[str, None] = None, cooki
}
def md5(data):
def md5(data) -> str:
return hashlib.md5(data.encode('utf-8')).hexdigest()
def get_token_js(rid: str, did: str, proxy_addr: Union[str, None] = None) -> Union[list, Dict[str, Any]]:
def get_token_js(rid: str, did: str, proxy_addr: OptionalStr = None) -> List[str]:
url = f'https://www.douyu.com/{rid}'
html_str = get_req(url=url, proxy_addr=proxy_addr)
@@ -537,8 +533,7 @@ def get_token_js(rid: str, did: str, proxy_addr: Union[str, None] = None) -> Uni
@trace_error_decorator
def get_douyu_info_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> Dict[
str, Any]:
def get_douyu_info_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'Referer': 'https://m.douyu.com/3125893?rid=3125893&dyshid=0-96003918aa5365bc6dcb4933000316p1&dyshci=181',
@@ -574,8 +569,8 @@ def get_douyu_info_data(url: str, proxy_addr: Union[str, None] = None, cookies:
@trace_error_decorator
def get_douyu_stream_data(rid: str, rate: str = '-1', proxy_addr: Union[str, None] = None,
cookies: Union[str, None] = None) -> Dict[str, Any]:
def get_douyu_stream_data(rid: str, rate: str = '-1', proxy_addr: OptionalStr = None,
cookies: OptionalStr = None) -> dict:
did = '10000000000000000000000000003306'
params_list = get_token_js(rid, did, proxy_addr=proxy_addr)
headers = {
@@ -604,8 +599,7 @@ def get_douyu_stream_data(rid: str, rate: str = '-1', proxy_addr: Union[str, Non
@trace_error_decorator
def get_yy_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_yy_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
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',
@@ -648,8 +642,7 @@ def get_yy_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: U
@trace_error_decorator
def get_bilibili_room_info_h5(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
str:
def get_bilibili_room_info_h5(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> str:
headers = {
'origin': 'https://live.bilibili.com',
'referer': 'https://live.bilibili.com/',
@@ -667,8 +660,7 @@ def get_bilibili_room_info_h5(url: str, proxy_addr: Union[str, None] = None, coo
@trace_error_decorator
def get_bilibili_room_info(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_bilibili_room_info(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
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',
@@ -698,8 +690,8 @@ def get_bilibili_room_info(url: str, proxy_addr: Union[str, None] = None, cookie
@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) -> Union[str, None]:
def get_bilibili_stream_data(url: str, qn: str = '10000', platform: str = 'web', proxy_addr: OptionalStr = None,
cookies: OptionalStr = None) -> OptionalStr:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.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',
@@ -761,8 +753,7 @@ def get_bilibili_stream_data(url: str, qn: str = '10000', platform: str = 'web',
@trace_error_decorator
def get_xhs_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_xhs_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'xy-common-params': 'platform=iOS&sid=session.1722166379345546829388',
@@ -824,8 +815,7 @@ def get_xhs_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: U
@trace_error_decorator
def get_bigo_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_bigo_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/119.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',
@@ -873,8 +863,7 @@ def get_bigo_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
@trace_error_decorator
def get_blued_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_blued_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
@@ -903,7 +892,7 @@ def get_blued_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
@trace_error_decorator
def login_afreecatv(username: str, password: str, proxy_addr: Union[str, None] = None) -> Union[str, None]:
def login_afreecatv(username: str, password: str, proxy_addr: OptionalStr = None) -> OptionalStr:
if len(username) < 6 or len(password) < 10:
raise RuntimeError('SOOP[AfreecaTV]登录失败！请在config.ini配置文件中填写正确的SOOP[AfreecaTV]平台的账号和密码')
@@ -951,8 +940,7 @@ def login_afreecatv(username: str, password: str, proxy_addr: Union[str, None] =
@trace_error_decorator
def get_afreecatv_cdn_url(broad_no: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_afreecatv_cdn_url(broad_no: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/119.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',
@@ -979,8 +967,7 @@ def get_afreecatv_cdn_url(broad_no: str, proxy_addr: Union[str, None] = None, co
@trace_error_decorator
def get_afreecatv_tk(url: str, rtype: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Union[str, tuple, None]:
def get_afreecatv_tk(url: str, rtype: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> str | tuple:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0',
'Origin': 'https://play.sooplive.co.kr',
@@ -1024,9 +1011,9 @@ def get_afreecatv_tk(url: str, rtype: str, proxy_addr: Union[str, None] = None,
@trace_error_decorator
def get_afreecatv_stream_data(
url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None,
username: Union[str, None] = None, password: Union[str, None] = None
) -> Dict[str, Any]:
url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None,
username: OptionalStr = None, password: OptionalStr = None
) -> dict:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/119.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',
@@ -1078,13 +1065,13 @@ def get_afreecatv_stream_data(
return play_url_list
if not anchor_name:
def handle_login():
def handle_login() -> OptionalStr:
cookie = login_afreecatv(username, password, proxy_addr=proxy_addr)
if 'AuthTicket=' in cookie:
print('SOOP[AfreecaTV]平台登录成功！开始获取直播数据...')
return cookie
def fetch_data(cookie):
def fetch_data(cookie) -> dict:
aid_token = get_afreecatv_tk(url, rtype='aid', proxy_addr=proxy_addr, cookies=cookie)
_anchor_name, _broad_no = get_afreecatv_tk(url, rtype='info', proxy_addr=proxy_addr, cookies=cookie)
_view_url = get_afreecatv_cdn_url(_broad_no, proxy_addr=proxy_addr)['view_url']
@@ -1130,8 +1117,7 @@ def get_afreecatv_stream_data(
@trace_error_decorator
def get_netease_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_netease_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1160,8 +1146,7 @@ def get_netease_stream_data(url: str, proxy_addr: Union[str, None] = None, cooki
@trace_error_decorator
def get_qiandurebo_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_qiandurebo_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,'
'application/signed-exchange;v=b3;q=0.7',
@@ -1190,8 +1175,7 @@ def get_qiandurebo_stream_data(url: str, proxy_addr: Union[str, None] = None, co
@trace_error_decorator
def get_pandatv_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_pandatv_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'origin': 'https://www.pandalive.co.kr',
'referer': 'https://www.pandalive.co.kr/',
@@ -1242,8 +1226,7 @@ def get_pandatv_stream_data(url: str, proxy_addr: Union[str, None] = None, cooki
@trace_error_decorator
def get_maoerfm_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_maoerfm_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1281,8 +1264,7 @@ def get_maoerfm_stream_url(url: str, proxy_addr: Union[str, None] = None, cookie
@trace_error_decorator
def get_winktv_bj_info(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Tuple[str, Any]:
def get_winktv_bj_info(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> tuple:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1308,8 +1290,7 @@ def get_winktv_bj_info(url: str, proxy_addr: Union[str, None] = None, cookies: U
@trace_error_decorator
def get_winktv_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_winktv_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1353,7 +1334,7 @@ def get_winktv_stream_data(url: str, proxy_addr: Union[str, None] = None, cookie
@trace_error_decorator
def login_flextv(username: str, password: str, proxy_addr: Union[str, None] = None) -> Union[str, int, None]:
def login_flextv(username: str, password: str, proxy_addr: OptionalStr = None) -> OptionalStr:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1401,10 +1382,10 @@ def login_flextv(username: str, password: str, proxy_addr: Union[str, None] = No
def get_flextv_stream_url(
url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None,
username: Union[str, None] = None, password: Union[str, None] = None
) -> Any:
def fetch_data(cookie):
url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None,
username: OptionalStr = None, password: OptionalStr = None
) -> tuple | None:
def fetch_data(cookie) -> dict:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1442,9 +1423,9 @@ def get_flextv_stream_url(
@trace_error_decorator
def get_flextv_stream_data(
url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None,
username: Union[str, None] = None, password: Union[str, None] = None
) -> Dict[str, Any]:
url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None,
username: OptionalStr = None, password: OptionalStr = None
) -> dict:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1480,7 +1461,7 @@ def get_flextv_stream_data(
return result
def get_looklive_secret_data(text):
def get_looklive_secret_data(text) -> tuple:
# 本算法参考项目：https://github.com/785415581/MusicBox/blob/b8f716d43d/doc/analysis/analyze_captured_data.md
modulus = '00e0b509f6259df8642dbc35662901477df22677ec152b5ff68ace615bb7b725152b3ab17a876aea8a5aa76d2e417629ec4ee' \
@@ -1498,7 +1479,7 @@ def get_looklive_secret_data(text):
charset = '1234567890abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ!@#$%^&*()_+-=[]{}|;:,.<>?'
return ''.join(secrets.choice(charset) for _ in range(size)).encode('utf-8')
def aes_encrypt(_text: Union[str, bytes], _sec_key: Union[str, bytes]) -> bytes:
def aes_encrypt(_text: str | bytes, _sec_key: str | bytes) -> bytes:
if isinstance(_text, str):
_text = _text.encode('utf-8')
if isinstance(_sec_key, str):
@@ -1511,7 +1492,7 @@ def get_looklive_secret_data(text):
encoded_ciphertext = base64.b64encode(ciphertext)
return encoded_ciphertext
def rsa_encrypt(_text: Union[str, bytes], pub_key: str, mod: str) -> str:
def rsa_encrypt(_text: str | bytes, pub_key: str, mod: str) -> str:
if isinstance(_text, str):
_text = _text.encode('utf-8')
text_reversed = _text[::-1]
@@ -1525,9 +1506,7 @@ def get_looklive_secret_data(text):
return enc_text.decode(), enc_sec_key
def get_looklive_stream_url(
url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None
) -> Dict[str, Any]:
def get_looklive_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
"""
通过PC网页端的接口获取完整直播源，只有params和encSecKey这两个加密请求参数。
params: 由两次AES加密完成
@@ -1570,8 +1549,8 @@ def get_looklive_stream_url(
@trace_error_decorator
def login_popkontv(
username: str, password: str, proxy_addr: Union[str, None] = None, code: Union[str, None] = 'P-00001'
) -> Union[tuple, None]:
username: str, password: str, proxy_addr: OptionalStr = None, code: OptionalStr = 'P-00001'
) -> tuple:
headers = {
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1618,9 +1597,9 @@ def login_popkontv(
@trace_error_decorator
def get_popkontv_stream_data(
url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None,
username: Union[str, None] = None, code: Union[str, None] = 'P-00001'
) -> Union[tuple, Any]:
url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None,
username: OptionalStr = None, code: OptionalStr = 'P-00001'
) -> tuple:
headers = {
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1684,12 +1663,12 @@ def get_popkontv_stream_data(
@trace_error_decorator
def get_popkontv_stream_url(
url: str,
proxy_addr: Union[str, None] = None,
access_token: Union[str, None] = None,
username: Union[str, None] = None,
password: Union[str, None] = None,
partner_code: Union[str, None] = 'P-00001'
) -> Dict[str, Any]:
proxy_addr: OptionalStr = None,
access_token: OptionalStr = None,
username: OptionalStr = None,
password: OptionalStr = None,
partner_code: OptionalStr = 'P-00001'
) -> dict:
headers = {
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1774,9 +1753,9 @@ 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
) -> Union[str, None]:
account_type: str, username: str, password: str, proxy_addr: OptionalStr = None,
cookies: OptionalStr = None
) -> OptionalStr:
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',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1832,12 +1811,12 @@ def login_twitcasting(
@trace_error_decorator
def get_twitcasting_stream_url(
url: str,
proxy_addr: Union[str, None] = None,
cookies: Union[str, None] = None,
account_type: Union[str, None] = None,
username: Union[str, None] = None,
password: Union[str, None] = None,
) -> Dict[str, Any]:
proxy_addr: OptionalStr = None,
cookies: OptionalStr = None,
account_type: OptionalStr = None,
username: OptionalStr = None,
password: OptionalStr = None,
) -> dict:
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',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -1850,7 +1829,7 @@ def get_twitcasting_stream_url(
if cookies:
headers['Cookie'] = cookies
def get_data(header):
def get_data(header) -> tuple:
html_str = get_req(url, proxy_addr=proxy_addr, headers=header)
anchor = re.search("<title>(.*?) \\(@(.*?)\\) 的直播 - Twit", html_str)
title = re.search('<meta name="twitter:title" content="(.*?)">\n\\s+<meta', html_str)
@@ -1893,8 +1872,7 @@ def get_twitcasting_stream_url(
@trace_error_decorator
def get_baidu_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_baidu_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'Connection': 'keep-alive',
@@ -1959,8 +1937,7 @@ def get_baidu_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_weibo_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_weibo_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
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',
@@ -2012,8 +1989,7 @@ def get_weibo_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_kugou_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_kugou_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:124.0) Gecko/20100101 Firefox/124.0',
'Accept': 'application/json',
@@ -2066,8 +2042,7 @@ def get_kugou_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
return result
def get_twitchtv_room_info(url: str, token: str, proxy_addr: Union[str, None] = None,
cookies: Union[str, None] = None):
def get_twitchtv_room_info(url: str, token: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> tuple:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:124.0) Gecko/20100101 Firefox/124.0',
'Accept-Language': 'zh-CN',
@@ -2105,8 +2080,7 @@ def get_twitchtv_room_info(url: str, token: str, proxy_addr: Union[str, None] =
@trace_error_decorator
def get_twitchtv_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_twitchtv_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0',
'Accept-Language': 'en-US',
@@ -2174,8 +2148,7 @@ def get_twitchtv_stream_data(url: str, proxy_addr: Union[str, None] = None, cook
@trace_error_decorator
def get_liveme_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_liveme_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'origin': 'https://www.liveme.com',
'referer': 'https://www.liveme.com',
@@ -2218,7 +2191,7 @@ def get_liveme_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies
return result
def get_huajiao_sn(url: str, cookies: Union[str, None] = None, proxy_addr: Union[str, None] = None):
def get_huajiao_sn(url: str, cookies: OptionalStr = None, proxy_addr: OptionalStr = None) -> tuple | None:
headers = {
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'referer': 'https://www.huajiao.com/',
@@ -2244,7 +2217,7 @@ def get_huajiao_sn(url: str, cookies: Union[str, None] = None, proxy_addr: Union
raise RuntimeError('获取直播间数据失败，花椒直播间地址非固定，请使用主播主页地址进行录制')
def get_huajiao_user_info(url: str, cookies: Union[str, None] = None, proxy_addr: Union[str, None] = None):
def get_huajiao_user_info(url: str, cookies: OptionalStr = None, proxy_addr: OptionalStr = None) -> tuple:
headers = {
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'referer': 'https://www.huajiao.com/',
@@ -2283,8 +2256,7 @@ def get_huajiao_user_info(url: str, cookies: Union[str, None] = None, proxy_addr
return '未知直播间', None
def get_huajiao_stream_url_app(url: str, proxy_addr: Union[str, None] = None) -> \
Dict[str, Any]:
def get_huajiao_stream_url_app(url: str, proxy_addr: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'living/9.4.0 (com.huajiao.seeding; build:2410231746; iOS 17.0.0) Alamofire/9.4.0',
'accept-language': 'zh-Hans-US;q=1.0',
@@ -2308,8 +2280,7 @@ def get_huajiao_stream_url_app(url: str, proxy_addr: Union[str, None] = None) ->
@trace_error_decorator
def get_huajiao_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_huajiao_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'referer': 'https://www.huajiao.com/',
@@ -2344,8 +2315,7 @@ def get_huajiao_stream_url(url: str, proxy_addr: Union[str, None] = None, cookie
@trace_error_decorator
def get_liuxing_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_liuxing_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -2380,8 +2350,7 @@ def get_liuxing_stream_url(url: str, proxy_addr: Union[str, None] = None, cookie
@trace_error_decorator
def get_showroom_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_showroom_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
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',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -2426,8 +2395,7 @@ def get_showroom_stream_data(url: str, proxy_addr: Union[str, None] = None, cook
@trace_error_decorator
def get_acfun_sign_params(proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Tuple[Any, str, Any]:
def get_acfun_sign_params(proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> tuple:
did = f'web_{generate_random_string(16)}'
headers = {
'referer': 'https://live.acfun.cn/',
@@ -2448,8 +2416,7 @@ def get_acfun_sign_params(proxy_addr: Union[str, None] = None, cookies: Union[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]:
def get_acfun_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'referer': 'https://live.acfun.cn/live/17912421',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0',
@@ -2494,8 +2461,7 @@ def get_acfun_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_changliao_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_changliao_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'Accept': 'application/json, text/plain, */*',
@@ -2539,8 +2505,7 @@ def get_changliao_stream_url(url: str, proxy_addr: Union[str, None] = None, cook
@trace_error_decorator
def get_yingke_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_yingke_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'Referer': 'https://www.inke.cn/',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0',
@@ -2576,8 +2541,7 @@ def get_yingke_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_yinbo_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_yinbo_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'Accept': 'application/json, text/plain, */*',
@@ -2622,8 +2586,7 @@ def get_yinbo_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
@trace_error_decorator
def get_zhihu_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_zhihu_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/115.0',
@@ -2631,11 +2594,20 @@ def get_zhihu_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
if cookies:
headers['Cookie'] = cookies
web_id = url.split('?')[0].rsplit('/', maxsplit=1)[-1]
html_str = get_req(url, proxy_addr=proxy_addr, headers=headers)
json_str = re.findall('<script id="js-initialData" type="text/json">(.*?)</script>', html_str)[0]
json_data = json.loads(json_str)
live_data = json_data['initialState']['theater']['theaters'][web_id]
if 'people/' in url:
user_id = url.split('people/')[1]
api = f'https://api.zhihu.com/people/{user_id}/profile?profile_new_version='
json_str = get_req(api, proxy_addr=proxy_addr, headers=headers)
json_data = json.loads(json_str)
live_page_url = json_data['drama']['living_theater']['theater_url']
else:
live_page_url = url
web_id = live_page_url.split('?')[0].rsplit('/', maxsplit=1)[-1]
html_str = get_req(live_page_url, proxy_addr=proxy_addr, headers=headers)
json_str2 = re.findall('<script id="js-initialData" type="text/json">(.*?)</script>', html_str)[0]
json_data2 = json.loads(json_str2)
live_data = json_data2['initialState']['theater']['theaters'][web_id]
anchor_name = live_data['actor']['name']
live_status = live_data['drama']['status']
result = {"anchor_name": anchor_name, "is_live": False}
@@ -2651,8 +2623,7 @@ def get_zhihu_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
@trace_error_decorator
def get_chzzk_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_chzzk_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
@@ -2685,8 +2656,7 @@ def get_chzzk_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_haixiu_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_haixiu_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'origin': 'https://www.haixiutv.com',
'referer': 'https://www.haixiutv.com/',
@@ -2740,8 +2710,7 @@ def get_haixiu_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_vvxqiu_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_vvxqiu_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'Access-Control-Request-Method': 'GET',
@@ -2772,8 +2741,7 @@ def get_vvxqiu_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_17live_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_17live_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'origin': 'https://17.live',
'referer': 'https://17.live/',
@@ -2808,8 +2776,7 @@ def get_17live_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_langlive_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_langlive_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'origin': 'https://www.lang.live',
'referer': 'https://www.lang.live/',
@@ -2841,8 +2808,7 @@ def get_langlive_stream_url(url: str, proxy_addr: Union[str, None] = None, cooki
@trace_error_decorator
def get_pplive_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_pplive_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'Content-Type': 'application/json',
'Origin': 'https://m.pp.weimipopo.com',
@@ -2882,8 +2848,7 @@ def get_pplive_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies
@trace_error_decorator
def get_6room_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \
Dict[str, Any]:
def get_6room_stream_url(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict:
headers = {
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'referer': 'https://ios.6.cn/?ver=8.0.3&build=4',
@@ -2920,4 +2885,4 @@ def get_6room_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies:
flv_url = f'https://wlive.6rooms.com/httpflv/{flv_title}.flv'
result['flv_url'] = flv_url
result['record_url'] = get_req(flv_url, proxy_addr=proxy_addr, headers=headers, redirect_url=True)
return result
return result

27
douyinliverecorder/stream.py
View File

@@ -8,7 +8,6 @@ Update: 2024-10-27 17:15:00
Copyright (c) 2023-2024 by Hmily, All Rights Reserved.
Function: Get live stream data.
"""
from typing import Dict, Any, Union
import base64
import hashlib
import json
@@ -25,7 +24,7 @@ from .spider import (
@trace_error_decorator
def get_douyin_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
def get_douyin_stream_url(json_data: dict, video_quality: str) -> dict:
anchor_name = json_data.get('anchor_name')
result = {
@@ -38,9 +37,9 @@ def get_douyin_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]
if status == 2:
stream_url = json_data['stream_url']
flv_url_dict = stream_url['flv_pull_url']
flv_url_list = list(flv_url_dict.values())
flv_url_list: list = list(flv_url_dict.values())
m3u8_url_dict = stream_url['hls_pull_url_map']
m3u8_url_list = list(m3u8_url_dict.values())
m3u8_url_list: list = list(m3u8_url_dict.values())
while len(flv_url_list) < 5:
flv_url_list.append(flv_url_list[-1])
@@ -59,11 +58,11 @@ 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]:
def get_tiktok_stream_url(json_data: dict, video_quality: str) -> dict:
if not json_data:
return {"anchor_name": None, "is_live": False}
def get_video_quality_url(stream, q_key) -> list[dict[str, int | Any]]:
def get_video_quality_url(stream, q_key) -> list:
play_list = []
for key in stream:
url_info = stream[key]['main']
@@ -111,7 +110,7 @@ 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]:
def get_kuaishou_stream_url(json_data: dict, video_quality: str) -> dict:
if json_data['type'] == 1 and not json_data["is_live"]:
return json_data
live_status = json_data['is_live']
@@ -148,7 +147,7 @@ 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]:
def get_huya_stream_url(json_data: dict, video_quality: str) -> dict:
game_live_info = json_data['data'][0]['gameLiveInfo']
stream_info_list = json_data['data'][0]['gameStreamInfoList']
anchor_name = game_live_info.get('nick', '')
@@ -237,7 +236,7 @@ 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, video_quality: str, cookies: str, proxy_addr: str) -> Dict[str, Any]:
def get_douyu_stream_url(json_data: dict, video_quality: str, cookies: str, proxy_addr: str) -> dict:
if not json_data["is_live"]:
return json_data
@@ -264,7 +263,7 @@ def get_douyu_stream_url(json_data: dict, video_quality: str, cookies: str, prox
@trace_error_decorator
def get_yy_stream_url(json_data: dict) -> Dict[str, Any]:
def get_yy_stream_url(json_data: dict) -> dict:
anchor_name = json_data.get('anchor_name', '')
result = {
"anchor_name": anchor_name,
@@ -282,7 +281,7 @@ 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, proxy_addr: str, cookies: str) -> Dict[str, Any]:
def get_bilibili_stream_url(json_data: dict, video_quality: str, proxy_addr: str, cookies: str) -> dict:
anchor_name = json_data["anchor_name"]
if not json_data["live_status"]:
return {
@@ -313,7 +312,7 @@ def get_bilibili_stream_url(json_data: dict, video_quality: str, proxy_addr: str
@trace_error_decorator
def get_netease_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any]:
def get_netease_stream_url(json_data: dict, video_quality: str) -> dict:
if not json_data['is_live']:
return json_data
stream_list = json_data['stream_list']['resolution']
@@ -336,7 +335,7 @@ def get_netease_stream_url(json_data: dict, video_quality: str) -> Dict[str, Any
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]:
extra_key: str | int = None) -> dict:
if not json_data['is_live']:
return json_data
@@ -359,4 +358,4 @@ def get_stream_url(json_data: dict, video_quality: str, url_type: str = 'm3u8',
data["flv_url"] = flv
data["record_url"] = flv
data['title'] = json_data.get('title')
return data
return data

12
douyinliverecorder/utils.py
View File

@@ -5,7 +5,7 @@ import functools
import hashlib
import re
import traceback
from typing import Union, Any
from typing import Any
import execjs
from .logger import logger
import configparser
@@ -33,12 +33,12 @@ def check_md5(file_path: str) -> str:
return file_md5
def dict_to_cookie_str(cookies_dict) -> str:
def dict_to_cookie_str(cookies_dict: dict) -> str:
cookie_str = '; '.join([f"{key}={value}" for key, value in cookies_dict.items()])
return cookie_str
def read_config_value(file_path, section, key) -> Union[str, None]:
def read_config_value(file_path: str, section: str, key: str) -> str | None:
config = configparser.ConfigParser()
try:
@@ -58,7 +58,7 @@ def read_config_value(file_path, section, key) -> Union[str, None]:
return None
def update_config(file_path, section, key, new_value) -> None:
def update_config(file_path: str, section: str, key: str, new_value: str) -> None:
config = configparser.ConfigParser()
try:
@@ -83,7 +83,7 @@ def update_config(file_path, section, key, new_value) -> None:
print(f"写入配置文件时出错: {e}")
def get_file_paths(directory) -> list:
def get_file_paths(directory: str) -> list:
file_paths = []
for root, dirs, files in os.walk(directory):
for file in files:
@@ -91,7 +91,7 @@ def get_file_paths(directory) -> list:
return file_paths
def remove_emojis(text, replace_text=r''):
def remove_emojis(text: str, replace_text=r''):
emoji_pattern = re.compile(
"["
"\U0001F1E0-\U0001F1FF" # flags (iOS)

BIN
libs/api-ms-win-core-path-l1-1-0.dll
View File

Binary file not shown.

34
main.py
View File

@@ -23,7 +23,7 @@ from pathlib import Path
import urllib.parse
import urllib.request
from urllib.error import URLError, HTTPError
from typing import Any, Union
from typing import Any
import configparser
from douyinliverecorder import spider
from douyinliverecorder import stream
@@ -81,7 +81,7 @@ def signal_handler(_signal, _frame):
signal.signal(signal.SIGTERM, signal_handler)
def display_info():
def display_info() -> None:
global start_display_time
time.sleep(5)
while True:
@@ -124,7 +124,7 @@ def display_info():
logger.error(f"错误信息: {e} 发生错误的行数: {e.__traceback__.tb_lineno}")
def update_file(file_path: str, old_str: str, new_str: str, start_str: str = None) -> Union[str, None]:
def update_file(file_path: str, old_str: str, new_str: str, start_str: str = None) -> str | None:
if old_str == new_str and start_str is None:
return old_str
with file_update_lock:
@@ -149,7 +149,7 @@ def update_file(file_path: str, old_str: str, new_str: str, start_str: str = Non
return new_str
def delete_line(file_path: str, del_line: str):
def delete_line(file_path: str, del_line: str) -> None:
with file_update_lock:
with open(file_path, 'r+', encoding=text_encoding) as f:
lines = f.readlines()
@@ -169,7 +169,7 @@ def get_startup_info(system_type: str):
return startup_info
def converts_mp4(address: str, is_original_delete: bool = True):
def converts_mp4(address: str, is_original_delete: bool = True) -> None:
_output = subprocess.check_output([
"ffmpeg", "-i", address,
"-c:v", "copy",
@@ -182,7 +182,7 @@ def converts_mp4(address: str, is_original_delete: bool = True):
os.remove(address)
def converts_m4a(address: str, is_original_delete: bool = True):
def converts_m4a(address: str, is_original_delete: bool = True) -> None:
_output = subprocess.check_output([
"ffmpeg", "-i", address,
"-n", "-vn",
@@ -195,7 +195,7 @@ def converts_m4a(address: str, is_original_delete: bool = True):
os.remove(address)
def generate_subtitles(record_name: str, ass_filename: str, sub_format: str = 'srt'):
def generate_subtitles(record_name: str, ass_filename: str, sub_format: str = 'srt') -> None:
index_time = 0
today = datetime.datetime.now()
re_datatime = today.strftime('%Y-%m-%d %H:%M:%S')
@@ -220,7 +220,7 @@ def generate_subtitles(record_name: str, ass_filename: str, sub_format: str = 's
re_datatime = today.strftime('%Y-%m-%d %H:%M:%S')
def adjust_max_request():
def adjust_max_request() -> None:
global max_request, error_count, pre_max_request, error_window
preset = max_request
@@ -277,7 +277,7 @@ def push_message(record_name: str, live_url: str, content: str) -> None:
print(f"直播消息推送到{platform}失败: {e}")
def run_bash(command: list):
def run_bash(command: list) -> None:
process = subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, startupinfo=get_startup_info(os_type)
)
@@ -288,7 +288,7 @@ def run_bash(command: list):
print(stderr_decoded)
def clear_record_info(record_name: str, record_url: str):
def clear_record_info(record_name: str, record_url: str) -> None:
global monitoring
recording.discard(record_name)
if record_url in url_comments and record_url in running_list:
@@ -298,7 +298,7 @@ def clear_record_info(record_name: str, record_url: str):
def check_subprocess(record_name: str, record_url: str, ffmpeg_command: list, save_type: str,
bash_file_path: Union[str, None] = None) -> bool:
bash_file_path: str | None = None) -> bool:
save_file_path = ffmpeg_command[-1]
process = subprocess.Popen(
ffmpeg_command, stderr=subprocess.STDOUT, startupinfo=get_startup_info(os_type)
@@ -352,7 +352,7 @@ def check_subprocess(record_name: str, record_url: str, ffmpeg_command: list, sa
return False
def start_record(url_data: tuple, count_variable: int = -1):
def start_record(url_data: tuple, count_variable: int = -1) -> None:
global error_count
while True:
@@ -1249,7 +1249,7 @@ def start_record(url_data: tuple, count_variable: int = -1):
time.sleep(2)
def backup_file(file_path: str, backup_dir_path: str):
def backup_file(file_path: str, backup_dir_path: str) -> None:
try:
if not os.path.exists(backup_dir_path):
os.makedirs(backup_dir_path)
@@ -1282,7 +1282,7 @@ def backup_file(file_path: str, backup_dir_path: str):
logger.error(f'\r备份配置文件 {file_path} 失败：{str(e)}')
def backup_file_start():
def backup_file_start() -> None:
config_md5 = ''
url_config_md5 = ''
@@ -1305,7 +1305,7 @@ def backup_file_start():
# --------------------------检查是否存在ffmpeg-------------------------------------
def check_ffmpeg_existence():
def check_ffmpeg_existence() -> bool:
dev_null = open(os.devnull, 'wb')
try:
subprocess.run(['ffmpeg', '--help'], stdout=dev_null, stderr=dev_null, check=True)
@@ -1346,7 +1346,7 @@ t3.start()
def read_config_value(config_parser: configparser.RawConfigParser, section: str, option: str, default_value: Any) \
-> Union[str, int, bool]:
-> Any:
try:
config_parser.read(config_file, encoding=text_encoding)
@@ -1704,4 +1704,4 @@ while True:
t2.start()
first_run = False
time.sleep(3)
time.sleep(3)

2
pyproject.toml
View File

@@ -1,5 +1,5 @@
[project]
requires-python = ">=3.10,<3.13"
requires-python = ">=3.10"
[build-system]
requires = ["poetry-core>=1.0.0"]

1
setup.py
View File

@@ -26,5 +26,6 @@ setup(
'Programming Language :: Python :: 3.10',
'Programming Language :: Python :: 3.11',
'Programming Language :: Python :: 3.12',
'Programming Language :: Python :: 3.13',
]
)