# -*- encoding: utf-8 -*- """ Author: Hmily GitHub: https://github.com/ihmily Date: 2023-07-15 23:15:00 Update: 2024-07-01 23:04:00 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 from urllib.request import Request from typing import Union, Dict, Any, Tuple, List import requests import ssl import re import json import execjs import urllib.request from utils import ( trace_error_decorator, update_config, dict_to_cookie_str ) 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) ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_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, timeout: int = 20, abroad: bool = False, content_conding: str = 'utf-8', redirect_url: bool = False, ) -> Union[str, Any]: if headers is None: headers = {} try: if proxy_addr: proxies = { 'http': proxy_addr, 'https': proxy_addr } if data or json_data: 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: return response.url resp_str = response.text else: if data and not isinstance(data, bytes): data = urllib.parse.urlencode(data).encode(content_conding) if json_data and isinstance(json_data, (dict, list)): data = json.dumps(json_data).encode(content_conding) req = urllib.request.Request(url, data=data, headers=headers) try: if abroad: response = urllib.request.urlopen(req, timeout=timeout) else: response = opener.open(req, timeout=timeout) if redirect_url: return response.url content_encoding = response.info().get('Content-Encoding') try: if content_encoding == 'gzip': with gzip.open(response, 'rt', encoding=content_conding) as gzipped: resp_str = gzipped.read() else: resp_str = response.read().decode(content_conding) finally: response.close() except urllib.error.HTTPError as e: if e.code == 400: resp_str = e.read().decode(content_conding) else: raise except urllib.error.URLError as e: print("URL Error:", e) raise except Exception as e: print("An error occurred:", e) raise except Exception as e: resp_str = str(e) return resp_str def get_params(url: str, params: str) -> Union[str, None]: parsed_url = urllib.parse.urlparse(url) query_params = urllib.parse.parse_qs(parsed_url.query) if params in query_params: 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) if match: _, json_str = match.groups() json_obj = json.loads(json_str) return json_obj else: raise Exception("No JSON data found in JSONP response.") def replace_url(file_path: str, old: str, new: str) -> None: with open(file_path, 'r', encoding='utf-8-sig') as f: content = f.read() if old in content: with open(file_path, 'w', encoding='utf-8-sig') as f: f.write(content.replace(old, new)) def get_play_url_list(m3u8: str, proxy: Union[str, None] = None, header: Union[dict, None] = None, abroad: bool = False) -> List[str]: resp = get_req(url=m3u8, proxy_addr=proxy, headers=header, abroad=abroad) play_url_list = [] for i in resp.split('\n'): if i.startswith('https://'): play_url_list.append(i.strip()) if not play_url_list: for i in resp.split('\n'): if i.strip().endswith('m3u8'): play_url_list.append(i.strip()) bandwidth_pattern = re.compile(r'BANDWIDTH=(\d+)') bandwidth_list = bandwidth_pattern.findall(resp) url_to_bandwidth = {url: int(bandwidth) for bandwidth, url in zip(bandwidth_list, play_url_list)} play_url_list = sorted(play_url_list, key=lambda url: url_to_bandwidth[url], reverse=True) 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]: 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 try: origin_url_list = None html_str = get_req(url=url, proxy_addr=proxy_addr, headers=headers) match_json_str = re.search(r'(\{\\"state\\":.*?)]\\n"]\)', html_str) if not match_json_str: match_json_str = re.search(r'(\{\\"common\\":.*?)]\\n"]\)
', html_str, re.S)[0] except Exception: raise ConnectionError("请检查你的网络是否可以正常访问TikTok网站") json_data = json.loads(json_str) return json_data @trace_error_decorator def get_kuaishou_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', } if cookies: headers['Cookie'] = cookies try: html_str = get_req(url=url, proxy_addr=proxy_addr, headers=headers) except Exception as e: print(f"Failed to fetch data from {url}.{e}") return {"type": 1, "is_live": False} try: json_str = re.search('', html_str, re.S).group(1) json_data = json.loads(json_str) room_data = json_data['props']['pageProps']['roomInfoInitData'] live_data = room_data['live'] result = {"is_live": False} if 'quickplay' not in live_data: result["anchor_name"] = room_data['nickname'] else: result["anchor_name"] = live_data['nickname'] result["stream_list"] = live_data['quickplay'] result["is_live"] = True return result @trace_error_decorator def get_qiandurebo_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \ Dict[str, Any]: 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', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6', 'referer': 'https://qiandurebo.com/web/index.php', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.58', } if cookies: headers['Cookie'] = cookies html_str = get_req(url=url, proxy_addr=proxy_addr, headers=headers) data = re.search('var user = (.*?)\r\n\s+user\.play_url', html_str, re.S).group(1) anchor_name = re.findall('"zb_nickname": "(.*?)",\r\n', data) result = {"anchor_name": "", "is_live": False} if len(anchor_name) > 0: result['anchor_name'] = anchor_name[0] play_url = re.findall('"play_url": "(.*?)",\r\n', data) if len(play_url) > 0 and 'common-text-center" style="display:block' not in html_str: result['anchor_name'] = anchor_name[0] result['flv_url'] = play_url[0] result['is_live'] = True result['record_url'] = play_url[0] return result @trace_error_decorator def get_pandatv_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \ Dict[str, Any]: headers = { 'referer': 'https://www.pandalive.co.kr/', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.58', } if cookies: headers['Cookie'] = cookies user_id = url.split('?')[0].rsplit('/', maxsplit=1)[1] url2 = 'https://api.pandalive.co.kr/v1/live/play' data = { 'userId': user_id, 'info': 'media fanGrade', } room_password = get_params(url, "pwd") if not room_password: room_password = '' data2 = { 'action': 'watch', 'userId': user_id, 'password': room_password, 'shareLinkType': '', } result = {"anchor_name": "", "is_live": False} json_str = get_req('https://api.pandalive.co.kr/v1/member/bj', proxy_addr=proxy_addr, headers=headers, data=data, abroad=True) json_data = json.loads(json_str) anchor_id = json_data['bjInfo']['id'] anchor_name = f"{json_data['bjInfo']['nick']}-{anchor_id}" result['anchor_name'] = anchor_name live_status = 'media' in json_data if live_status: json_str = get_req(url2, proxy_addr=proxy_addr, headers=headers, data=data2, abroad=True) json_data = json.loads(json_str) if 'errorData' in json_data: if json_data['errorData']['code'] == 'needAdult': raise RuntimeError(f'{url} 直播间需要登录后成人才可观看，请你在配置文件中正确填写登录后的cookie') else: raise RuntimeError(json_data['errorData']['code'], json_data['message']) play_url = json_data['PlayList']['hls'][0]['url'] result['m3u8_url'] = play_url result['is_live'] = True result['play_url_list'] = get_play_url_list(m3u8=play_url, proxy=proxy_addr, header=headers, abroad=True) return result @trace_error_decorator def get_maoerfm_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \ Dict[str, Any]: 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', 'referer': 'https://fm.missevan.com/live/868895007', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.58', } if cookies: headers['Cookie'] = cookies room_id = url.split('?')[0].rsplit('/', maxsplit=1)[1] url2 = f'https://fm.missevan.com/api/v2/live/{room_id}' json_str = get_req(url=url2, proxy_addr=proxy_addr, headers=headers) json_data = json.loads(json_str) anchor_name = json_data['info']['creator']['username'] live_status = False if 'room' in json_data['info']: live_status = json_data['info']['room']['status']['broadcasting'] result = { "anchor_name": anchor_name, "is_live": live_status, } if live_status: stream_list = json_data['info']['room']['channel'] m3u8_url = stream_list['hls_pull_url'] flv_url = stream_list['flv_pull_url'] result['m3u8_url'] = m3u8_url result['flv_url'] = flv_url result['is_live'] = True result['record_url'] = m3u8_url return result @trace_error_decorator def get_winktv_bj_info(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \ Tuple[str, Any]: 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', 'content-type': 'application/x-www-form-urlencoded', 'referer': 'https://www.winktv.co.kr/', '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 user_id = url.split('?')[0].rsplit('/', maxsplit=1)[-1] data = { 'userId': user_id, 'info': 'media', } info_api = 'https://api.winktv.co.kr/v1/member/bj' json_str = get_req(url=info_api, proxy_addr=proxy_addr, headers=headers, data=data, abroad=True) json_data = json.loads(json_str) live_status = 'media' in json_data anchor_id = json_data['bjInfo']['id'] anchor_name = f"{json_data['bjInfo']['nick']}-{anchor_id}" return anchor_name, live_status @trace_error_decorator def get_winktv_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \ Dict[str, Any]: 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', 'content-type': 'application/x-www-form-urlencoded', 'referer': 'https://www.winktv.co.kr/', '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 user_id = url.split('?')[0].rsplit('/', maxsplit=1)[-1] room_password = get_params(url, "pwd") if not room_password: room_password = '' data = { 'action': 'watch', 'userId': user_id, 'password': room_password, 'shareLinkType': '', } anchor_name, live_status = get_winktv_bj_info(url=url, proxy_addr=proxy_addr, cookies=cookies) result = {"anchor_name": anchor_name, "is_live": live_status} if live_status: play_api = 'https://api.winktv.co.kr/v1/live/play' json_str = get_req(url=play_api, proxy_addr=proxy_addr, headers=headers, data=data, abroad=True) json_data = json.loads(json_str) if 'errorData' in json_data: if json_data['errorData']['code'] == 'needAdult': raise RuntimeError(f'{url} 直播间需要登录后成人才可观看，请你在配置文件中正确填写登录后的cookie') else: raise RuntimeError(json_data['errorData']['code'], json_data['message']) m3u8_url = json_data['PlayList']['hls'][0]['url'] result['m3u8_url'] = m3u8_url result['play_url_list'] = get_play_url_list(m3u8=m3u8_url, proxy=proxy_addr, header=headers, abroad=True) return result @trace_error_decorator def login_flextv(username: str, password: str, proxy_addr: Union[str, None] = None) -> Union[str, int, None]: 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', 'content-type': 'application/json;charset=UTF-8', 'referer': 'https://www.flextv.co.kr/', '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', } data = { 'loginId': username, 'password': password, 'loginKeep': True, 'saveId': True, 'device': 'PCWEB', } url = 'https://api.flextv.co.kr/v2/api/auth/signin' try: if proxy_addr: proxies = { 'http': proxy_addr, 'https': proxy_addr } response = requests.post(url, json=data, headers=headers, proxies=proxies, timeout=20) json_data = response.json() cookie_dict = response.cookies.get_dict() else: req_json_data = json.dumps(data).encode('utf-8') cookie_jar = http.cookiejar.CookieJar() login_opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookie_jar)) req = Request(url, data=req_json_data, headers=headers) response = login_opener.open(req, timeout=20) resp_str = response.read().decode('utf-8') json_data = json.loads(resp_str) cookie_dict = {cookie.name: cookie.value for cookie in cookie_jar} if "error" not in json_data: cookie = dict_to_cookie_str(cookie_dict) return cookie print('请检查配置文件中的FlexTV账号和密码是否正确') except Exception as e: print('FlexTV登录请求异常', e) 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 ) -> Union[str, Any]: def fetch_data(cookie): 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', 'referer': 'https://www.flextv.co.kr/', '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', } user_id = url.split('/live')[0].rsplit('/', maxsplit=1)[-1] if cookie: headers['Cookie'] = cookie play_api = f'https://api.flextv.co.kr/api/channels/{user_id}/stream?option=all' json_str = get_req(play_api, proxy_addr=proxy_addr, headers=headers, abroad=True) if 'HTTP Error 400: Bad Request' in json_str: raise ConnectionError('获取FlexTV直播数据失败，请切换代理重试') return json.loads(json_str) json_data = fetch_data(cookies) if "message" in json_data and json_data["message"] == "로그인후 이용이 가능합니다.": print("FlexTV直播获取失败[未登录]: 19+直播需要登录后是成人才可观看") print("正在尝试登录FlexTV直播平台，请确保已在配置文件中填写好您的账号和密码") if len(username) < 6 or len(password) < 8: raise RuntimeError('FlexTV登录失败！请在config.ini配置文件中填写正确的FlexTV平台的账号和密码') print('FlexTV平台登录中...') new_cookie = login_flextv(username, password, proxy_addr=proxy_addr) if new_cookie: print('FlexTV平台登录成功！开始获取直播数据...') json_data = fetch_data(new_cookie) update_config('./config/config.ini', 'Cookie', 'flextv_cookie', new_cookie) else: raise RuntimeError('FlexTV登录失败') if 'sources' in json_data and len(json_data['sources']) > 0: play_url = json_data['sources'][0]['url'] return play_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]: 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', 'referer': 'https://www.flextv.co.kr/', '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 user_id = url.split('/live')[0].rsplit('/', maxsplit=1)[-1] result = {"anchor_name": '', "is_live": False} try: url2 = f'https://www.flextv.co.kr/channels/{user_id}' html_str = get_req(url2, proxy_addr=proxy_addr, headers=headers, abroad=True) json_str = re.search('', html_str).group(1) json_data = json.loads(json_str) channel_data = json_data['props']['pageProps']['channel'] live_status = channel_data['isInLive'] anchor_id = channel_data['owner']['loginId'] anchor_name = f"{channel_data['owner']['nickname']}-{anchor_id}" result["anchor_name"] = anchor_name if live_status: result['is_live'] = True play_url = get_flextv_stream_url( url=url, proxy_addr=proxy_addr, cookies=cookies, username=username, password=password) if play_url: result['m3u8_url'] = play_url result['play_url_list'] = get_play_url_list(m3u8=play_url, proxy=proxy_addr, header=headers, abroad=True) except Exception as e: print('FlexTV直播间数据获取失败', e) return result def get_looklive_secret_data(text): # 本算法参考项目：https://github.com/785415581/MusicBox/blob/b8f716d43d/doc/analysis/analyze_captured_data.md modulus = '00e0b509f6259df8642dbc35662901477df22677ec152b5ff68ace615bb7b725152b3ab17a876aea8a5aa76d2e417629ec4ee' \ '341f56135fccf695280104e0312ecbda92557c93870114af6c9d05c4f7f0c3685b7a46bee255932575cce10b424d813cfe487' \ '5d3e82047b97ddef52741d546b8e289dc6935b3ece0462db0a22b8e7' nonce = b'0CoJUm6Qyw8W8jud' public_key = '010001' from Crypto.Cipher import AES from Crypto.Util.Padding import pad import base64 import binascii import secrets def create_secret_key(size: int) -> bytes: charset = '1234567890abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ!@#$%^&*()_+-=[]{}|;:,.<>?' return ''.join(secrets.choice(charset) for _ in range(size)).encode('utf-8') def aes_encrypt(text: Union[str, bytes], seckey: Union[str, bytes]) -> bytes: if isinstance(text, str): text = text.encode('utf-8') if isinstance(seckey, str): seckey = seckey.encode('utf-8') seckey = seckey[:16] # 16 (AES-128), 24 (AES-192), or 32 (AES-256) bytes iv = bytes('0102030405060708', 'utf-8') encryptor = AES.new(seckey, AES.MODE_CBC, iv) padded_text = pad(text, AES.block_size) ciphertext = encryptor.encrypt(padded_text) encoded_ciphertext = base64.b64encode(ciphertext) return encoded_ciphertext def rsa_encrypt(text: Union[str, bytes], pub_key: str, mod: str) -> str: if isinstance(text, str): text = text.encode('utf-8') text_reversed = text[::-1] text_int = int(binascii.hexlify(text_reversed), 16) encrypted_int = pow(text_int, int(pub_key, 16), int(mod, 16)) return format(encrypted_int, 'x').zfill(256) sec_key = create_secret_key(16) enc_text = aes_encrypt(aes_encrypt(json.dumps(text), nonce), sec_key) enc_sec_key = rsa_encrypt(sec_key, public_key, modulus) 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]: """ 通过PC网页端的接口获取完整直播源，只有params和encSecKey这两个加密请求参数。 params: 由两次AES加密完成 ncSecKey: 由一次自写的加密函数完成，值可固定 """ headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0', 'Accept': 'application/json, text/javascript', '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', 'Content-Type': 'application/x-www-form-urlencoded', 'Referer': 'https://look.163.com/', } if cookies: headers['Cookie'] = cookies room_id = re.search('live\?id=(.*?)&', url).group(1) params, secretkey = get_looklive_secret_data({"liveRoomNo": room_id}) request_data = {'params': params, 'encSecKey': secretkey} api = 'https://api.look.163.com/weapi/livestream/room/get/v3' json_str = get_req(api, proxy_addr=proxy_addr, headers=headers, data=request_data) json_data = json.loads(json_str) anchor_name = json_data['data']['anchor']['nickName'] live_status = json_data['data']['liveStatus'] result = {"anchor_name": anchor_name, "is_live": False} if live_status == 1: result["is_live"] = True if json_data['data']['roomInfo']['liveType'] == 1: print('Look直播暂时只支持音频直播，不支持Look视频直播!') else: play_url_list = json_data['data']['roomInfo']['liveUrl'] result["flv_url"] = play_url_list['httpPullUrl'] result["m3u8_url"] = play_url_list['hlsPullUrl'] result["record_url"] = play_url_list['hlsPullUrl'] return result @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]: 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', 'Authorization': 'Basic FpAhe6mh8Qtz116OENBmRddbYVirNKasktdXQiuHfm88zRaFydTsFy63tzkdZY0u', 'Content-Type': 'application/json', 'Origin': 'https://www.popkontv.com', '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', } data = { 'partnerCode': code, 'signId': username, 'signPwd': password, } url = 'https://www.popkontv.com/api/proxy/member/v1/login' if proxy_addr: proxies = { 'http': proxy_addr, 'https': proxy_addr } response = requests.post(url, json=data, headers=headers, proxies=proxies, timeout=20) json_data = response.json() else: req_json_data = json.dumps(data).encode('utf-8') cookie_jar = http.cookiejar.CookieJar() login_opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookie_jar)) req = Request(url, data=req_json_data, headers=headers) response = login_opener.open(req, timeout=20) resp_str = response.read().decode('utf-8') json_data = json.loads(resp_str) login_status_code = json_data["statusCd"] if login_status_code == 'E4010': raise Exception('popkontv登录失败，请重新配置正确的登录账号或者密码！') elif json_data["statusCd"] == 'S2000': token = json_data['data']["token"] partner_code = json_data['data']["partnerCode"] return token, partner_code else: raise Exception(f'popkontv登录失败，{json_data["statusMsg"]}') @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]: 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', 'Content-Type': 'application/json', 'Origin': 'https://www.popkontv.com', '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 if 'mcid' in url: anchor_id = re.search('mcid=(.*?)&', url).group(1) else: anchor_id = re.search('castId=(.*?)(?=&|$)', url).group(1) data = { 'partnerCode': code, 'searchKeyword': anchor_id, 'signId': username, } api = 'https://www.popkontv.com/api/proxy/broadcast/v1/search/all' json_str = get_req(api, proxy_addr=proxy_addr, headers=headers, json_data=data, abroad=True) json_data = json.loads(json_str) partner_code = '' anchor_name = 'Unknown' for item in json_data['data']['broadCastList']: if item['mcSignId'] == anchor_id: mc_name = item['nickName'] anchor_name = f"{mc_name}-{anchor_id}" partner_code = item['mcPartnerCode'] break if not partner_code: if 'mcPartnerCode' in url: regex_result = re.search('mcPartnerCode=(P-\d+)', url) else: regex_result = re.search('partnerCode=(P-\d+)', url) partner_code = regex_result.group(1) if regex_result else code notices_url = f'https://www.popkontv.com/channel/notices?mcid={anchor_id}&mcPartnerCode={partner_code}' notices_response = get_req(notices_url, proxy_addr=proxy_addr, headers=headers, abroad=True) mc_name_match = re.search(r'"mcNickName":"([^"]+)"', notices_response) mc_name = mc_name_match.group(1) if mc_name_match else 'Unknown' anchor_name = f"{anchor_id}-{mc_name}" live_url = f"https://www.popkontv.com/live/view?castId={anchor_id}&partnerCode={partner_code}" html_str2 = get_req(live_url, proxy_addr=proxy_addr, headers=headers, abroad=True) json_str2 = re.search('', html_str2).group(1) json_data2 = json.loads(json_str2) if 'mcData' in json_data2['props']['pageProps']: room_data = json_data2['props']['pageProps']['mcData']['data'] is_private = room_data['mc_isPrivate'] cast_start_date_code = room_data['mc_castStartDate'] mc_sign_id = room_data['mc_signId'] cast_type = room_data['castType'] return anchor_name, [cast_start_date_code, partner_code, mc_sign_id, cast_type, is_private] else: return anchor_name, None @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]: 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', 'ClientKey': 'Client FpAhe6mh8Qtz116OENBmRddbYVirNKasktdXQiuHfm88zRaFydTsFy63tzkdZY0u', 'Content-Type': 'application/json', 'Origin': 'https://www.popkontv.com', '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 access_token: headers['Authorization'] = f'Bearer {access_token}' anchor_name, room_info = get_popkontv_stream_data( url, proxy_addr=proxy_addr, code=partner_code, username=username) result = {"anchor_name": anchor_name, "is_live": False} if room_info: cast_start_date_code, cast_partner_code, mc_sign_id, cast_type, is_private = room_info result["is_live"] = True room_password = get_params(url, "pwd") if int(is_private) != 0 and not room_password: raise RuntimeError(f"直播间数据获取失败，因为{anchor_name}直播间为私密房间，请配置房间密码后重试") def fetch_data(header: dict = None, code: str = None) -> str: data = { 'androidStore': 0, 'castCode': f'{mc_sign_id}-{cast_start_date_code}', 'castPartnerCode': cast_partner_code, 'castSignId': mc_sign_id, 'castType': cast_type, 'commandType': 0, 'exePath': 5, 'isSecret': is_private, 'partnerCode': code, 'password': room_password, 'signId': username, 'version': '4.6.2', } play_api = 'https://www.popkontv.com/api/proxy/broadcast/v1/castwatchonoff' return get_req(play_api, proxy_addr=proxy_addr, json_data=data, headers=header, abroad=True) json_str = fetch_data(headers, partner_code) if 'HTTP Error 400' in json_str or 'statusCd":"E5000' in json_str: print("popkontv直播获取失败[token不存在或者已过期]: 请登录后观看") print("正在尝试登录popkontv直播平台，请确保已在配置文件中填写好您的账号和密码") if len(username) < 4 or len(password) < 10: raise RuntimeError('popkontv登录失败！请在config.ini配置文件中填写正确的popkontv平台的账号和密码') print('popkontv平台登录中...') new_access_token, new_partner_code = login_popkontv( username=username, password=password, proxy_addr=proxy_addr, code=partner_code ) if new_access_token and len(new_access_token) == 640: print('popkontv平台登录成功！开始获取直播数据...') headers['Authorization'] = f'Bearer {new_access_token}' json_str = fetch_data(headers, new_partner_code) update_config('./config/config.ini', 'Authorization', 'popkontv_token', new_access_token) else: raise RuntimeError('popkontv登录失败，请检查账号和密码是否正确') json_data = json.loads(json_str) status_msg = json_data["statusMsg"] if json_data['statusCd'] == "L000A": print('获取直播源失败,', status_msg) raise RuntimeError( '你是未认证会员。登录popkontv官方网站后，在“我的页面”>“修改我的信息”底部进行手机认证后可用') elif json_data['statusCd'] == "L0001": cast_start_date_code = int(cast_start_date_code) - 1 json_str = fetch_data(headers, partner_code) json_data = json.loads(json_str) m3u8_url = json_data['data']['castHlsUrl'] result["m3u8_url"] = m3u8_url result["record_url"] = m3u8_url elif json_data['statusCd'] == "L0000": m3u8_url = json_data['data']['castHlsUrl'] result["m3u8_url"] = m3u8_url result["record_url"] = m3u8_url else: raise RuntimeError('获取直播源失败,', status_msg) return result @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]: 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', 'Content-Type': 'application/x-www-form-urlencoded', 'Referer': 'https://twitcasting.tv/indexcaslogin.php?redir=%2Findexloginwindow.php%3Fnext%3D%252F&keep=1', 'Cookie': 'hl=zh; did=04fb08f1b15d248644f1dfa82816d323; _ga=GA1.1.1021187740.1709706998; keep=1; mfadid=yrQiEB26ruRg7mlMavABMBZWdOddzojW; _ga_X8R46Y30YM=GS1.1.1709706998.1.1.1709712274.0.0.0', 'User-Agent': 'Mozilla/5.0 (Linux; Android 8.0.0; SM-G955U Build/R16NW) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Mobile Safari/537.36 Edg/121.0.0.0', } if cookies: headers['Cookie'] = cookies if account_type == "normal": login_url = 'https://twitcasting.tv/indexcaslogin.php?redir=%2F&keep=1' login_api = 'https://twitcasting.tv/indexcaslogin.php?redir=/indexloginwindow.php?next=%2F&keep=1' else: login_url = 'https://twitcasting.tv/indexpasswordlogin.php' login_api = 'https://twitcasting.tv/indexpasswordlogin.php?redir=/indexloginwindow.php?next=%2F&keep=1' html_str = get_req(login_url, proxy_addr=proxy_addr, headers=headers) cs_session_id = re.search('', html_str).group(1) data = { 'username': username, 'password': password, 'action': 'login', 'cs_session_id': cs_session_id, } try: if proxy_addr: proxies = { 'http': proxy_addr, 'https': proxy_addr } response = requests.post(login_api, data=data, headers=headers, proxies=proxies, timeout=20) cookie_dict = response.cookies.get_dict() else: req_json_data = urllib.parse.urlencode(data).encode('utf-8') cookie_jar = http.cookiejar.CookieJar() login_opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookie_jar)) req = Request(login_api, data=req_json_data, headers=headers) _ = login_opener.open(req, timeout=20) cookie_dict = {cookie.name: cookie.value for cookie in cookie_jar} if 'tc_ss' in cookie_dict: cookie = dict_to_cookie_str(cookie_dict) return cookie except Exception as e: print('TwitCasting登录出错,', e) @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]: 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', 'Referer': 'https://twitcasting.tv/?ch0', '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', } anchor_id = url.split('/')[3] if cookies: headers['Cookie'] = cookies def get_data(header): html_str = get_req(url, proxy_addr=proxy_addr, headers=header) anchor = re.search("(.*?) \(@(.*?)\) 的直播 - Twit", html_str) status = re.search('data-is-onlive="(.*?)"\n\s+data-view-mode', html_str) movie_id = re.search('data-movie-id="(.*?)" data-audience-id', html_str) return f'{anchor.group(1).strip()}-{anchor.group(2)}-{movie_id.group(1)}', status.group(1) result = {"anchor_name": '', "is_live": False} try: 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) if not new_cookie: raise RuntimeError('TwitCasting登录失败,请检查配置文件中的账号密码是否正确') print('TwitCasting 登录成功！开始获取数据...') headers['Cookie'] = new_cookie update_config('./config/config.ini', 'Cookie', 'twitcasting_cookie', new_cookie) anchor_name, live_status = get_data(headers) result["anchor_name"] = anchor_name if live_status == 'true': play_url = f'https://twitcasting.tv/{anchor_id}/metastream.m3u8/?video=1&mode=source' result['m3u8_url'] = play_url result['record_url'] = play_url result['is_live'] = True return result @trace_error_decorator def get_baidu_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', 'Connection': 'keep-alive', 'Referer': 'https://live.baidu.com/', 'User-Agent': 'Mozilla/5.0 (Linux; Android 8.0.0; SM-G955U Build/R16NW) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Mobile Safari/537.36 Edg/121.0.0.0', } if cookies: headers['Cookie'] = cookies uid = random.choice([ 'h5-683e85bdf741bf2492586f7ca39bf465', 'h5-c7c6dc14064a136be4215b452fab9eea', 'h5-4581281f80bb8968bd9a9dfba6050d3a' ]) room_id = re.search('room_id=(.*?)&', url).group(1) params = { 'cmd': '371', 'action': 'star', 'service': 'bdbox', 'osname': 'baiduboxapp', 'data': '{"data":{"room_id":"' + room_id + '","device_id":"h5-683e85bdf741bf2492586f7ca39bf465","source_type":0,"osname":"baiduboxapp"},"replay_slice":0,"nid":"","schemeParams":{"src_pre":"pc","src_suf":"other","bd_vid":"","share_uid":"","share_cuk":"","share_ecid":"","zb_tag":"","shareTaskInfo":"{\\"room_id\\":\\"9175031377\\"}","share_from":"","ext_params":"","nid":""}}', 'ua': '360_740_ANDROID_0', 'bd_vid': '', 'uid': uid, '_': str(int(time.time() * 1000)), 'callback': '__jsonp_callback_1__', } app_api = f'https://mbd.baidu.com/searchbox?{urllib.parse.urlencode(params)}' jsonp_str = get_req(url=app_api, proxy_addr=proxy_addr, headers=headers) json_data = jsonp_to_json(jsonp_str) key = list(json_data['data'].keys())[0] data = json_data['data'][key] anchor_name = data['host']['name'] result = { "anchor_name": anchor_name, "is_live": False, } live_status = data['video']['stream'] if live_status == 1: result["is_live"] = True play_url_list = data['video']['url_clarity_list'] url_list = [] prefix = 'https://hls.liveshow.bdstatic.com/live/' if play_url_list: for i in play_url_list: url_list.append( prefix + i['urls']['flv'].rsplit('.', maxsplit=1)[0].rsplit('/', maxsplit=1)[1] + '.m3u8') else: play_url_list = data['video']['url_list'] for i in play_url_list: url_list.append(prefix + i['urls'][0]['hls'].rsplit('?', maxsplit=1)[0].rsplit('/', maxsplit=1)[1]) if url_list: result['play_url_list'] = url_list result['is_live'] = True return result @trace_error_decorator 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', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36', } if cookies: headers['Cookie'] = cookies room_id = '' if 'show/' in url: room_id = url.split('?')[0].split('show/')[1] else: uid = url.split('?')[0].rsplit('/u/', maxsplit=1)[1] web_api = f'https://weibo.com/ajax/statuses/mymblog?uid={uid}&page=1&feature=0' json_str = get_req(web_api, proxy_addr=proxy_addr, headers=headers) json_data = json.loads(json_str) for i in json_data['data']['list']: if 'page_info' in i and i['page_info']['object_type'] == 'live': room_id = i['page_info']['object_id'] break result = { "anchor_name": '', "is_live": False, } if room_id: app_api = f'https://weibo.com/l/pc/anchor/live?live_id={room_id}' # app_api = f'https://weibo.com/l/!/2/wblive/room/show_pc_live.json?live_id={room_id}' json_str = get_req(url=app_api, proxy_addr=proxy_addr, headers=headers) json_data = json.loads(json_str) anchor_name = json_data['data']['user_info']['name'] result["anchor_name"] = anchor_name live_status = json_data['data']['item']['status'] if live_status == 1: result["is_live"] = True play_url_list = json_data['data']['item']['stream_info']['pull'] 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 @trace_error_decorator def get_kugou_stream_url(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:124.0) Gecko/20100101 Firefox/124.0', 'Accept': 'application/json', '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://fanxing2.kugou.com/', } if cookies: headers['Cookie'] = cookies if 'roomId' in url: room_id = re.search('roomId=(\d+)', url).group(1) else: room_id = url.split('?')[0].rsplit('/', maxsplit=1)[1] app_api = f'https://service2.fanxing.kugou.com/roomcen/room/web/cdn/getEnterRoomInfo?roomId={room_id}' json_str = get_req(url=app_api, proxy_addr=proxy_addr, headers=headers) json_data = json.loads(json_str) anchor_name = json_data['data']['normalRoomInfo']['nickName'] result = { "anchor_name": anchor_name, "is_live": False, } if not anchor_name: raise RuntimeError('不支持音乐频道直播间录制，请切换直播间录制') live_status = json_data['data']['liveType'] if live_status == 0: params = { 'std_rid': room_id, 'std_plat': '7', 'std_kid': '0', 'streamType': '1-2-4-5-8', 'ua': 'fx-flash', 'targetLiveTypes': '1-5-6', 'version': '1000', 'supportEncryptMode': '1', 'appid': '1010', '_': str(int(time.time() * 1000)), } stream_api = f'https://fx1.service.kugou.com/video/pc/live/pull/mutiline/streamaddr?{urllib.parse.urlencode(params)}' json_str2 = get_req(url=stream_api, proxy_addr=proxy_addr, headers=headers) json_data2 = json.loads(json_str2) stream_data = json_data2['data']['lines'] if stream_data: result["is_live"] = True result['flv_url'] = stream_data[-1]['streamProfiles'][0]['httpsFlv'][0] result['record_url'] = result['flv_url'] return result def get_twitchtv_room_info(url: str, token: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:124.0) Gecko/20100101 Firefox/124.0', 'Accept-Language': 'zh-CN', 'Referer': 'https://www.twitch.tv/', 'Client-Id': 'kimne78kx3ncx6brgo4mv6wki5h1ko', 'Client-Integrity': token, 'Content-Type': 'text/plain;charset=UTF-8', } if cookies: headers['Cookie'] = cookies uid = url.split('?')[0].rsplit('/', maxsplit=1)[-1] data = [ { "operationName": "ChannelShell", "variables": { "login": uid }, "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "580ab410bcd0c1ad194224957ae2241e5d252b2c5173d8e0cce9d32d5bb14efe" } } }, ] json_str = get_req('https://gql.twitch.tv/gql', proxy_addr=proxy_addr, headers=headers, json_data=data, abroad=True) json_data = json.loads(json_str) user_data = json_data[0]['data']['userOrError'] login_name = user_data["login"] nickname = f"{user_data['displayName']}-{login_name}" status = True if user_data['stream'] else False return nickname, status @trace_error_decorator def get_twitchtv_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) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36', 'Accept-Language': 'en-US', 'Referer': 'https://www.twitch.tv/', 'Client-ID': 'kimne78kx3ncx6brgo4mv6wki5h1ko', } if cookies: headers['Cookie'] = cookies uid = url.split('?')[0].rsplit('/', maxsplit=1)[-1] data = { "operationName": "PlaybackAccessToken_Template", "query": "query PlaybackAccessToken_Template($login: String!, $isLive: Boolean!, $vodID: ID!, $isVod: Boolean!, $playerType: String!) { streamPlaybackAccessToken(channelName: $login, params: {platform: \"web\", playerBackend: \"mediaplayer\", playerType: $playerType}) @include(if: $isLive) { value signature authorization { isForbidden forbiddenReasonCode } __typename } videoPlaybackAccessToken(id: $vodID, params: {platform: \"web\", playerBackend: \"mediaplayer\", playerType: $playerType}) @include(if: $isVod) { value signature __typename }}", "variables": { "isLive": True, "login": uid, "isVod": False, "vodID": "", "playerType": "site" } } json_str = get_req('https://gql.twitch.tv/gql', proxy_addr=proxy_addr, headers=headers, json_data=data, abroad=True) json_data = json.loads(json_str) token = json_data['data']['streamPlaybackAccessToken']['value'] sign = json_data['data']['streamPlaybackAccessToken']['signature'] anchor_name, live_status = get_twitchtv_room_info(url=url, token=token, proxy_addr=proxy_addr, cookies=cookies) result = {"anchor_name": anchor_name, "is_live": live_status} if live_status: play_session_id = random.choice(["bdd22331a986c7f1073628f2fc5b19da", "064bc3ff1722b6f53b0b5b8c01e46ca5"]) params = { "acmb": "e30=", "allow_sourc": "true", "browser_family": "firefox", "browser_version": "124.0", "cdm": "wv", "fast_bread": "true", "os_name": "Windows", "os_version": "NT%2010.0", "p": "3553732", "platform": "web", "play_session_id": play_session_id, "player_backend": "mediaplayer", "player_version": "1.28.0-rc.1", "playlist_include_framerate": "true", "reassignments_supported": "true", "sig": sign, "token": token, "transcode_mode": "cbr_v1" } access_key = urllib.parse.urlencode(params) m3u8_url = f'https://usher.ttvnw.net/api/channel/hls/{uid}.m3u8?{access_key}' play_url_list = get_play_url_list(m3u8=m3u8_url, proxy=proxy_addr, header=headers, abroad=True) result['m3u8_url'] = m3u8_url result['play_url_list'] = play_url_list return result @trace_error_decorator def get_liveme_stream_url(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', 'if-none-match': '"5056-h/Mk04yXaLRQmg+4iqhJH8Sa8l0"', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0', } if cookies: headers['Cookie'] = cookies html_str = get_req(url=url, proxy_addr=proxy_addr, headers=headers) video_url = re.search('hlsvideosource:"(.*?)",sdkstreamid', html_str) anchor_name = re.search('<title>(.*?) ', html_str).group(1) result = { "anchor_name": anchor_name, "is_live": False, } if video_url: play_url = video_url.group(1).replace('\\u002F', '/') result["is_live"] = True result['m3u8_url'] = play_url result['record_url'] = play_url return result def get_huajiao_sn(url: str, cookies: Union[str, None] = None, proxy_addr: Union[str, None] = 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/', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0', } if cookies: headers['Cookie'] = cookies live_id = url.split('?')[0].rsplit('/', maxsplit=1)[1] api = f'https://www.huajiao.com/l/{live_id}' try: html_str = get_req(url=api, proxy_addr=proxy_addr, headers=headers) json_str = re.search('var feed = (.*?});', html_str).group(1) json_data = json.loads(json_str) sn = json_data['feed']['sn'] uid = json_data['author']['uid'] nickname = json_data['author']['nickname'] live_id = url.split('?')[0].rsplit('/', maxsplit=1)[1] return nickname, sn, uid, live_id except Exception: replace_url(f'{script_path}/config/URL_config.ini', old=url, new='#' + url) raise RuntimeError('获取直播间数据失败，花椒直播间地址非固定，请使用主播主页地址进行录制') def get_huajiao_user_info(url: str, cookies: Union[str, None] = None, proxy_addr: Union[str, None] = 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/', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0', } if cookies: headers['Cookie'] = cookies if 'user' in url: uid = url.split('?')[0].split('user/')[1] params = { 'uid': uid, 'fmt': 'json', '_': str(int(time.time() * 1000)), } api = f'https://webh.huajiao.com/User/getUserFeeds?{urllib.parse.urlencode(params)}' json_str = get_req(url=api, proxy_addr=proxy_addr, headers=headers) json_data = json.loads(json_str) html_str = get_req(url=f'https://www.huajiao.com/user/{uid}', proxy_addr=proxy_addr, headers=headers) anchor_name = re.search('<title>(.*?)的主页.*', html_str).group(1) if json_data['data'] and 'rtop' in json_data['data']['feeds'][0]['feed']: live_id = json_data['data']['feeds'][0]['feed']['relateid'] sn = json_data['data']['feeds'][0]['feed']['sn'] return anchor_name, [sn, uid, live_id] else: return anchor_name, None else: data = get_huajiao_sn(url) if data: return data[0], data[1:] else: return '未知直播间', 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]: 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/', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0', } if cookies: headers['Cookie'] = cookies anchor_name, room_info = get_huajiao_user_info(url) result = {"anchor_name": anchor_name, "is_live": False} if room_info: sn, uid, live_id = room_info params = { "time": int(time.time() * 1000), "version": "1.0.0", "sn": sn, "uid": uid, "liveid": live_id, "encode": "h265" # 可选 h264 } api = f'https://live.huajiao.com/live/substream?{urllib.parse.urlencode(params)}' json_str = get_req(url=api, proxy_addr=proxy_addr, headers=headers) json_data = json.loads(json_str) result["is_live"] = True result['flv_url'] = json_data['data']['h264_url'] result['record_url'] = json_data['data']['h264_url'] return result @trace_error_decorator def get_liuxing_stream_url(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \ Dict[str, Any]: 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', 'Referer': 'https://wap.7u66.com/198189?promoters=0', 'User-Agent': 'Mozilla/5.0 (Linux; Android 8.0.0; SM-G955U Build/R16NW) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Mobile Safari/537.36 Edg/121.0.0.0', } if cookies: headers['Cookie'] = cookies room_id = url.split('?')[0].rsplit('/', maxsplit=1)[1] api = f'https://wap.7u66.com/api/ui/room/v1.0.0/live.ashx?promoters=0&roomidx={room_id}¤tUrl=https://www.7u66.com/{room_id}?promoters=0' json_str = get_req(url=api, proxy_addr=proxy_addr, headers=headers) json_data = json.loads(json_str) room_info = json_data['data']['roomInfo'] anchor_name = room_info['nickname'] live_status = room_info["live_stat"] result = {"anchor_name": anchor_name, "is_live": False} if live_status == 1: result["is_live"] = True idx = room_info['idx'] live_id = room_info['liveId1'] flv_url = f'https://txpull1.5see.com/live/{idx}/{live_id}.flv' result['flv_url'] = flv_url result['record_url'] = flv_url return result @trace_error_decorator def get_showroom_stream_data(url: str, proxy_addr: Union[str, None] = None, cookies: Union[str, None] = None) -> \ Dict[str, Any]: 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', '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 if '/room/profile' in url: room_id = url.split('room_id=')[-1] else: html_str = get_req(url, proxy_addr=proxy_addr, headers=headers, abroad=True) room_id = re.search('href="/room/profile\?room_id=(.*?)"', html_str).group(1) info_api = f'https://www.showroom-live.com/api/live/live_info?room_id={room_id}' json_str = get_req(info_api, proxy_addr=proxy_addr, headers=headers, abroad=True) json_data = json.loads(json_str) anchor_name = json_data['room_name'] result = {"anchor_name": anchor_name, "is_live": False} live_status = json_data['live_status'] if live_status == 2: result["is_live"] = True web_api = f'https://www.showroom-live.com/api/live/streaming_url?room_id={room_id}&abr_available=1' json_str = get_req(web_api, proxy_addr=proxy_addr, headers=headers, abroad=True) if json_str: json_data = json.loads(json_str) streaming_url_list = json_data['streaming_url_list'] for i in streaming_url_list: if i['type'] == 'hls_all': m3u8_url = i['url'] result['m3u8_url'] = m3u8_url if m3u8_url: m3u8_url_list = get_play_url_list(m3u8_url, proxy=proxy_addr, header=headers, abroad=True) if m3u8_url_list: result['play_url_list'] = [f"{m3u8_url.rsplit('/', maxsplit=1)[0]}/{i}" for i in m3u8_url_list] else: result['play_url_list'] = [m3u8_url] result['play_url_list'] = [i.replace('https://', 'http://') for i in result['play_url_list']] break 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，以避免默认的不可用导致无法获取数据 # 以下示例链接不保证时效性，请自行查看链接是否能正常访问 room_url = "https://live.douyin.com/745964462470" # 抖音直播 # room_url = "https://www.tiktok.com/@pearlgaga88/live" # Tiktok直播 # room_url = "https://live.kuaishou.com/u/yall1102" # 快手直播 # room_url = 'https://www.huya.com/116' # 虎牙直播 # room_url = 'https://www.douyu.com/topic/wzDBLS6?rid=4921614&dyshid=' # 斗鱼直播 # room_url = 'https://www.douyu.com/3637778?dyshid' # room_url = 'https://www.yy.com/22490906/22490906' # YY直播 # room_url = 'https://live.bilibili.com/21593109' # b站直播 # room_url = 'https://live.bilibili.com/23448867' # b站直播 # 小红书直播 # room_url = 'http://xhslink.com/O9f9fM' # room_url = 'https://www.redelight.cn/hina/livestream/569077534207413574?appuid=5f3f478a00000000010005b3&' # room_url = 'https://www.xiaohongshu.com/hina/livestream/569098486282043893?appuid=5f3f478a00000000010005b3&' # room_url = 'https://www.bigo.tv/cn/716418802' # bigo直播 # room_url = 'https://app.blued.cn/live?id=Mp6G2R' # blued直播 # room_url = 'https://play.afreecatv.com/sw7love' # afreecatv直播 # room_url = 'https://m.afreecatv.com/#/player/hl6260' # afreecatv直播 # room_url = 'https://play.afreecatv.com/secretx' # afreecatv直播 # room_url = 'https://cc.163.com/583946984' # 网易cc直播 # room_url = 'https://qiandurebo.com/web/video.php?roomnumber=33333' # 千度热播 # room_url = 'https://www.pandalive.co.kr/live/play/bara0109' # PandaTV # room_url = 'https://fm.missevan.com/live/868895007' # 猫耳FM直播 # room_url = 'https://www.winktv.co.kr/live/play/anjer1004' # WinkTV # room_url = 'https://www.flextv.co.kr/channels/593127/live' # FlexTV # room_url = 'https://look.163.com/live?id=65108820&position=3' # Look直播 # room_url = 'https://www.popkontv.com/live/view?castId=wjfal007&partnerCode=P-00117' # popkontv # room_url = 'https://www.popkontv.com/channel/notices?mcid=wjfal007&mcPartnerCode=P-00117' # popkontv # room_url = 'https://twitcasting.tv/c:uonq' # TwitCasting # room_url = 'https://live.baidu.com/m/media/pclive/pchome/live.html?room_id=9175031377&tab_category' # 百度直播 # room_url = 'https://weibo.com/u/7849520225' # 微博直播 # room_url = 'https://fanxing2.kugou.com/50428671?refer=2177&sourceFrom=' # 酷狗直播 # room_url = 'https://www.twitch.tv/gamerbee' # TwitchTV # room_url = 'https://www.liveme.com/zh/v/17141937295821012854/index.html' # LiveMe # room_url = 'https://www.huajiao.com/user/223184650' # 花椒直播 # 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='')) # print(get_huya_app_stream_url(room_url, proxy_addr='')) # print(get_douyu_info_data(room_url, proxy_addr='')) # print(get_douyu_stream_data("4921614", proxy_addr='')) # print(get_yy_stream_data(room_url, proxy_addr='')) # print(get_bilibili_stream_data(room_url, proxy_addr='')) # print(get_xhs_stream_url(room_url, proxy_addr='')) # print(get_bigo_stream_url(room_url, proxy_addr='')) # print(get_blued_stream_url(room_url, proxy_addr='')) # print(get_afreecatv_stream_data(room_url, proxy_addr='', username='', password='')) # print(get_netease_stream_data(room_url, proxy_addr='')) # print(get_qiandurebo_stream_data(room_url, proxy_addr='')) # print(get_pandatv_stream_data(room_url, proxy_addr='')) # print(get_maoerfm_stream_url(room_url, proxy_addr='')) # print(get_winktv_stream_data(room_url, proxy_addr='')) # print(get_flextv_stream_data(room_url,proxy_addr='', username='', password='')) # print(get_looklive_stream_url(room_url, proxy_addr='')) # print(get_popkontv_stream_url(room_url, proxy_addr='', username='', password='')) # print(get_twitcasting_stream_url(room_url, proxy_addr='', username='', password='')) # print(get_baidu_stream_data(room_url, proxy_addr='')) # print(get_weibo_stream_data(room_url, proxy_addr='')) # print(get_kugou_stream_url(room_url, proxy_addr='')) # print(get_twitchtv_stream_data(room_url, proxy_addr='')) # 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_acfun_stream_data(room_url, proxy_addr=''))