import json import re from curl_cffi import requests import time import random import secrets import base64 import global_encryption_generate def generate_p(text):# p=b+r+q+z+b+d+v """ 生成函数，用于处理输入文本并返回处理后的结果 参数: text (str): 需要处理的文本字符串 返回: 处理后的结果，通过unit_generate.right3方法对x1和0进行处理得到 """ x1 = 0 # 初始化变量x1为0，用于在循环过程中累积处理结果 for i in text: # 遍历输入文本的每个字符 x1 = global_encryption_generate.encodechat(i, x1) # 对每个字符进行处理，并将结果累积到x1中 return global_encryption_generate.right3(x1, 0) # 使用unit_generate.right3方法处理最终的x1值，传入0作为第二个参数 def generate_nonce(d,b): # d,b res=[] for i in range(4): str=f"0{d}{i}{b}" strbytes=str.encode('utf-8') res_num = global_encryption_generate.fnv1a_32(strbytes) res=res+global_encryption_generate.hex_to_byte_list(global_encryption_generate.in32(res_num)) return res def generate_X_net_sync_term(dicts,SST,header=[2, 188, 0, 4]): # header=[3, 12, 0, 4] x_header = header + list(base64.b64decode(SST)) nonce = generate_nonce(dicts['d'], dicts['b']) # nonce =[11, 238, 53, 123, 42, 202, 7, 80, 124, 206, 214, 77, 64, 223, 81, 190] ciphertext = list(global_encryption_generate.aes_ctr_encrypt(global_encryption_generate.encode_dict_to_bytearray(dicts),nonce)) return base64.b64encode(bytes(x_header+nonce+ciphertext)).decode() class GetSyncTerm: def __init__(self, session=None, home_url="https://www.bet365.com", useragent=''): self.session = session or requests.Session() self.useragent = useragent self.headers = {'User-Agent': self.useragent} self.home_url = home_url self.configuration_path = {} self.res_configs = {} def generate_similar_ipv6(self, base_address): """ 生成与给定 IPv6 地址格式相似的随机地址 参数: base_address: 基础 IPv6 地址，用于提取模式 返回: 格式相似但值不同的 IPv6 地址 """ parts = base_address.split(":") new_parts = [] for part in parts: if random.random() < 0.7: # 70% 的概率改变这部分 new_part = secrets.token_hex(2) # 保持第一个字符相同（增加相似度） if random.random() < 0.5: new_part = part[0] + new_part[1:] new_parts.append(new_part) else: new_parts.append(part) # 保留原部分 return ":".join(new_parts) # 示例使用 # base_ip = "f054:747d:68cb:a835:f30e:747f:cba0:4861" # similar_ips = [generate_similar_ipv6(base_ip) for _ in range(5)] # # print("原始地址:", base_ip) # print("相似地址:") # for ip in similar_ips: # print(ip) def advanced_graphics_info(self, ua_string): """ 更高级的图形信息生成函数，包含更多硬件检测逻辑 """ # 硬件数据库映射 gpu_mappings = {'AMD': {'patterns': [r'Renoir', r'Vega', r'Navi', r'Radeon.*Graphics'], 'models': ['renoir', 'vega', 'navi', 'radeon']}, 'Intel': {'patterns': [r'Iris.*Xe', r'UHD', r'HD Graphics'], 'models': ['xe', 'uhd', 'hd']}, 'NVIDIA': {'patterns': [r'RTX', r'GTX', r'GeForce'], 'models': ['rtx', 'gtx', 'geforce']}} # 检测硬件 detected_vendor = "unknown" detected_model = "unknown" for vendor, data in gpu_mappings.items(): for pattern, model in zip(data['patterns'], data['models']): if re.search(pattern, ua_string, re.IGNORECASE): detected_vendor = vendor.lower() detected_model = model break if detected_vendor != "unknown": break # 确定渲染器 if re.search(r'Windows', ua_string, re.IGNORECASE): renderer = "angle" elif re.search(r'Mac|iOS', ua_string, re.IGNORECASE): renderer = "metal" else: renderer = "opengl" # 版本信息（这里使用默认值，实际中可能需要更复杂的检测） opengl_version = "4.6" llvm_version = "15.0.7" # 构建结果 if detected_model != "unknown": return f"{renderer} ({detected_vendor}, {detected_model} ({detected_model} llvm {llvm_version}), opengl {opengl_version})" else: return f"{renderer} ({detected_vendor}, opengl {opengl_version})" import random def generate_graphics_parameter(self, ua_string=None): """ 综合方法生成图形供应商参数 参数: ua_string (str): 可选的User-Agent字符串 返回: str: 图形供应商信息 """ # 常见组合 common_combinations = ["Google Inc. (Intel)", "Google Inc. (AMD)", "Google Inc. (NVIDIA Corporation)", "Mozilla Foundation (Intel)", "Mozilla Foundation (AMD)", "Apple Inc. (Apple)", "Microsoft (Intel)"] # 如果有UA字符串，尝试基于它生成 if ua_string: if "Chrome" in ua_string and "Windows" in ua_string: if "Intel" in ua_string: return "Google Inc. (Intel)" elif "AMD" in ua_string: return "Google Inc. (AMD)" else: return random.choice(["Google Inc. (Intel)", "Google Inc. (AMD)", "Google Inc. (NVIDIA Corporation)"]) elif "Firefox" in ua_string: return "Mozilla Foundation (Intel)" elif "Safari" in ua_string and "Mac" in ua_string: return "Apple Inc. (Apple)" elif "Edge" in ua_string: return "Microsoft (Intel)" # 如果没有UA或无法确定，返回随机常见组合 return random.choice(common_combinations) def generate_guid(self, ): guid = [] for i in range(32): # 生成0-15的随机数并转换为十六进制字符串 guid.append(random.randint(0, 15).__format__('x')) # 在特定位置添加连字符 if i in (7, 11, 15, 19): guid.append('-') return ''.join(guid) # 生成并打印GUID # guid = generate_guid() def generate_dict(self, dict, ramdom_float_num, ua,frist_time,sessionID,apipath,webpath ): nowtime = int(time.time() * 1000) dict['d'] = nowtime dict['e'] = ua dict['s'] = sessionID dict['u'] = apipath dict['c'] = webpath dict['f']['i_e'] = frist_time dict['f']['ua'] = ua # dict['b'] = int(ramdom_float_num*2147483647) # p=b+r+q+z+b+d+v textp=f"{dict['b']}{dict['r']}{dict['q']}{dict['z']}{dict['b']}{dict['d']}{dict['v']}" dict['p'] = self.generate_p(textp) return dict def generate_p(self, text):# p=b+r+q+z+b+d+v x1 = 0 for i in text: x1 = global_encryption_generate.encodechat(i, x1) return global_encryption_generate.right3(x1,0) def generate_nonce(self,d,b): # d,b res=[] for i in range(4): str=f"0{d}{i}{b}" strbytes=str.encode('utf-8') res_num = global_encryption_generate.fnv1a_32(strbytes) res=res+global_encryption_generate.hex_to_byte_list(global_encryption_generate.in32(res_num)) return res def generate_X_net_sync_term(self,dicts,SST): x_header = [2, 188, 0, 4] + list(base64.b64decode(SST)) nonce = self.generate_nonce(dicts['d'], dicts['b']) ciphertext = list(global_encryption_generate.aes_ctr_encrypt(global_encryption_generate.encode_dict_to_byte(dicts),nonce)) return base64.b64encode(bytes(x_header+nonce+ciphertext)).decode() def extract_ns_weblib_util(self, html_content): # 正则表达式模式：匹配var ns_weblib_util = 后面的字典数据 # 考虑可能的空格和不同的闭合方式 pattern = r'var ns_weblib_util\s*=\s*({.*?});' # 使用非贪婪匹配来捕获字典内容 match = re.search(pattern, html_content, re.DOTALL) if match: # 提取并返回匹配到的字典字符串 return match.group(1) else: return None def return_configuration_path(self, response_text): result = self.extract_ns_weblib_util(response_text) configuration_path = json.loads(result.replace("'", "\"")) return configuration_path def request_main_txt(self, session=None, origin_name='https://www.bet365.com', ua = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36'): session = session or requests.Session(headers=self.headers, impersonate='chrome133a') headerss = {'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,ja;q=0.8', 'cache-control': 'no-cache', 'pragma': 'no-cache', 'priority': 'u=0, i', 'sec-ch-ua': '"Google Chrome";v="125", "Chromium";v="125", "Not.A/Brand";v="24"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Linux"', 'sec-fetch-dest': 'document', 'sec-fetch-mode': 'navigate', 'sec-fetch-site': 'none', 'sec-fetch-user': '?1', 'upgrade-insecure-requests': '1', 'user-agent':ua , } response = session.get(origin_name, headers=headerss) response_text = response.text self.configuration_path = self.return_configuration_path(response_text) return self.configuration_path def get_WebsiteConfig(self, session): configuration_path = self.request_main_txt(session) res_configs = session.get(self.home_url + configuration_path["WebsiteConfig"]["SITE_CONFIG_LOCATION"]).json() return res_configs def return_res_configs(self, res_configs): SST = res_configs['ns_weblib_util']['WebsiteConfig']['SST'] SESSION_ID = res_configs['flashvars']['SESSION_ID'] randomnum = random.random() fisttime = int(time.time() * 1000) return SESSION_ID, SST, randomnum, fisttime def inits(self, configuration_path, res_configs, session=None): SST = res_configs['ns_weblib_util']['WebsiteConfig']['SST'] SESSION_ID = res_configs['flashvars']['SESSION_ID'] randomnum = random.random() fisttime = int(time.time() * 1000) return SESSION_ID, SST, randomnum, fisttime def leagua_data_dict(self,mian_url,u,type_, ctid='88', g='',h="I", origin='www.z1.bet365.com', api_str="", request_id='f047:7468:68c6:8134:f301:746a:cba0:5044'): leagua_data = {} if type_=="web": leagua_data = {'d': 171231231, 'a': '1', 'b': 171231231, 'c': mian_url, 'e': u, 'p': 171231231, 'g': g, 'h': h, 'n': 'null', 'o': '88', 'q': 0, 'r': 0, 's': '59B4061EF0AC8D8BB18C821B14E167E8000003', 't': '-480', 'u': '/splashcontentapi/soccertab?lid=10&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', 'f': { 'i_c': 1, 'i_e': 171231231, 'i_n': 0, 'i_u': 15, 'i_r': 1, 'i_cl': 0, 'i_ev': 0, 'i_ps': 764416712, 'uqid': '', 'fw': '110', 'ua': u, 'rf': 'Direct', # 'd': origin, 'i_l': 0, 'z': 't2', 'x': 's', 'i_z': 3, 'pi': '2' }} elif type_ == "api": # request_id = self.generate_similar_ipv6('f09d:74c6:68cc:c881:f357:74c8:cba0:48aa') # 'f09d:74c6:68cc:f836:f357:74c8:cba0:48aa' # request_id = 'f09d:74c6:68cc:f836:f357:74c8:cba0:48aa' # request_id = 'f09d:74c6:68cd:2445:f357:74c8:cba0:48aa' # leagua_data = {'d': 171231231, 'a': '1', 'b': 171231231, 'c': mian_url, 'e': u, 'p': 171231231, 'g': g, "i": request_id, 'h': h, 'k': origin, 'n': 'null', 'o': ctid,'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 0, 'r': 0, 's': '59B4061EF0AC8D8BB18C821B14E167E8000003', 't': '-480', 'u': '/splashcontentapi/soccertab?lid=10&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', 'f': {'i_c': 1, 'i_e': 171231231, 'i_n': 0, 'i_u': 15, 'i_r': 1, 'i_cl': 0, 'i_ev': 0, 'i_ps': 764416712, 'uqid': '', 'fw': '170', 'ua': u, 'rf': 'Direct', 'd': origin, # 'i_l': 0, 'z': 't2', 'x': 's', 'i_z': 3, 'pi': '2'}} leagua_data = { 'a': '1', 'b': 743047743, 'c': mian_url, 'd': 1758105079538, 'e': u, 'g': g, 'h': h, 'i': request_id, 'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'k': origin, 'n': 'null', 'o': ctid, 'p': 305921974, 'q': 0, 'r': 0, 's': 'AED2D8ACA55CCC5590ED5237BA651134000003', 't': '-480', 'u': api_str, 'v': 2, 'x': '', 'z': 'Asia/Shanghai', 'f': {'d': origin, 'db': '1', 'db2': '2', 'dm': '8', 'fw': '171', 'h': 'I', 'i_abta': 0, 'i_ams': 0, 'i_bsc': 0, 'i_bt': 0, 'i_c': 1, 'i_cl': 0, 'i_ct': 0, 'i_e': 1758105077652, 'i_ev': 0, 'i_fd': 0, 'i_kp': 0, 'i_ks': 0, 'i_l': 0, 'i_n': 0, 'i_ps': 907367449, 'i_r': 4, 'i_rc': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, 'i_smm': 0, 'i_tf': 0, 'i_tm': 0, 'i_u': 185, 'i_z': 2, 'il': 'false', 'pi': '1', 'pr': '', 'rf': 'Direct', 'ua': u, 'ul': api_str, 'uqid': '', 'ur': self.advanced_graphics_info(u), 'uv': self.generate_graphics_parameter(u), 'x': 's', 'z': 'm2'}, } # leagua_data = {'d': 1757668996686, 'a': '1', 'b': 1344743547, 'c': mian_url, 'e': u, 'p': 1344808334, 'g': '', "i": request_id,'h': 'I', 'k': origin, 'n': 'null', 'o': ctid, 'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 272629826, 'r': 16778240, 's': '22D991EFACB84D449DF91AAEBCD8C284000003', 't': '-480', 'u': '/gamingservices/gam-client-gos-svcv1/gamingconfiguration?countryid=212&stateid=0&loggedin=0&platformtypeid=1&cid=212&languageid=2&statuscheck=1&cgid=2&ctid=212', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', # 'f': {'i_c': 2, 'i_e': 1757669000687, 'i_n': 0, 'i_u': 516, 'i_r': 2, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 1262838774, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua':u, 'rf': 'Direct', 'd': origin, 'db': '1', 'db2': '1', 'i_l': 272629826, 'h': 'K', 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}} # leagua_data = {'d': 1757839669191, 'a': '1', 'b': 1708426658, 'c': mian_url, 'e': u, 'p': 171203974, 'g': '', 'h': h, # leagua_data = {'d': 171231231, 'a': '1', 'b': 171231231, 'c': mian_url, 'e': u, 'p': 171231231, 'g': '', 'h': h, # 'i': request_id, # 'k': origin, 'n': 'null', 'o': ctid, 'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 0, 'r': 0, 's': 'B1AE76BFE6B58A40B77EC669AEF27C3C000003', 't': '-480', 'u': '/splashcontentapi/soccertab?lid=2&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', # 'f': { # "q":f' ~ https://{origin}/sports-assets/sportsoc/FirebaseWebSDKModule/assets/firebase_iframe.html', # 'i_c': 1, 'i_e': 1757839666948, 'i_n': 0, 'i_u': 87, 'i_r': 4, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 907367449, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua': u, 'rf': 'Direct', 'd': origin, 'db': '1', 'db2': '2', 'i_l': 0, 'h': h, 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}} # leagua_data = {'d': 1758031985075, 'a': '1', 'b': 255293688, 'c': '#/AS/B1/K^5/', 'e': u, 'p': 505589040, 'g': '', 'h': h, 'i': 'f03e:7467:68c9:7072:f2f8:7469:cba0:484b', 'k': origin, 'n': 'null', 'o': ctid, 'j':self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'q': 0, 'r': 0, 's': '2A59F77145006C3F9D3F6AAE5E75873E000003', 't': '-480', 'u': '/splashcontentapi/soccertab?lid=2&zid=0&pd=%23AS%23B1%23K%5E5%23&cid=88&cgid=1&ctid=88', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', # 'f': {'i_c': 2, 'i_e': 1758031986314, 'i_n': 0, 'i_u': 148, 'i_r': 3, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 907367449, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua': u, 'rf': 'Direct', 'd': origin, 'db': '1', 'db2': '2', 'i_l': 0, 'h': 'I', 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, # 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}} elif type_ == "app": leagua_data = {'a': '1', 'b': 1656656128, 'c': mian_url, 'd': 1760775263554, 'e': u, 'p': 133468807, 'g': g, 'h':h, 'i': request_id, 'j': self.res_configs.get('flashvars', {}).get('CURRENCY_EXCHANGE_RATE'), 'k': origin, 'n': 'null', 'o': ctid, 'q': 0, 'r': 0, 's': '4037F87FA5604F8496B50F4AA92C102C000003', 't': '-480', 'u': api_str, 'v': 2, 'x': '', 'z': 'Asia/Shanghai', 'f': {'i_u': 162, 'q': f' ~ https://{origin}/sports-assets/sportsoc/FirebaseWebSDKModule/assets/firebase_iframe.html', 'i_r': 14, 'i_cl': 1, 'i_ps': 50339904, 'uqid': '', 'uv': self.generate_graphics_parameter(u), 'ur': self.advanced_graphics_info(u), 'ua': u, 'il': 'false', 'ul': api_str, 'd': 'www.z1.bet365.com', 'x': 's', 'i_z': 2, 'pi': '1'}} return leagua_data def get_SyncTerm(self, webpath, res_configs, ctid="88", api_str='', connectionID='', h="I", origin='www.z1.bet365.com', type_="web", header=None, request_id=""): header = header or [2, 188, 0, 4] self.res_configs = res_configs # data = {'d': 1757841227203, 'a': '1', 'b': 1479666406, 'c': '#/AS/B1/K^5/', 'e': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0', 'p': 778743229, 'g': 'I3MRW36-AzWIbxNeAAMo', 'h': 'K', 'i': 'f047:7468:68c6:874c:f301:746a:cba0:5044', 'k': 'www.z1.bet365.com', 'n': 'null', 'o': '99', 'j': '199.5756', 'q': 0, 'r': 0, 's': '25D412EA6E443C9080A71824BA63A630000003', 't': '-480', 'u': '/zap/?uid=9162611288618789', 'v': 2, 'x': '', 'z': 'Asia/Shanghai', 'f': {'i_c': 1, 'i_e': 1757841226911, 'i_n': 0, 'i_u': 89, 'q': ' ~ https://www.z1.bet365.com/sports-assets/sportsoc/FirebaseWebSDKModule/assets/firebase_iframe.html', 'i_r': 2, 'i_tf': 0, 'i_bt': 0, 'i_cl': 0, 'i_ev': 0, 'i_ps': 954625300, 'uqid': '', 'fw': '171', 'uv': 'Google Inc. (Intel)', 'ur': 'angle (intel, intel(r) iris(r) xe graphics (0x00009a49) direct3d11 vs_5_0 ps_5_0, d3d11)', 'pr': '', 'ua': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0', 'rf': 'Direct', 'd': 'www.z1.bet365.com', 'db': '1', 'db2': '2', 'i_l': 0, 'h': 'K', 'z': 'm2', 'x': 's', 'i_z': 2, 'i_fd': 0, 'i_kp': 0, 'i_sc': 0, 'i_sd': 0, 'i_sf': 0, 'i_ks': 0, 'i_ams': 0, 'i_tm': 0, 'i_smm': 0, 'i_abta': 0, 'i_bsc': 0, 'i_ct': 0, 'i_rc': 0, 'pi': '1', 'dm': '8'}} leagua_data = self.leagua_data_dict(webpath, self.useragent, type_, ctid=ctid, g=connectionID,h=h, origin=origin, api_str=api_str, request_id=request_id) SESSION_ID, SST, randomnum, fisttime = self.inits(self.configuration_path, res_configs) dicts = self.generate_dict(leagua_data, randomnum, self.useragent, fisttime, SESSION_ID, api_str,f'#{webpath.replace("#", "/")}') sync_term = generate_X_net_sync_term(dicts, SST, header=header) # print(dicts) return sync_term async def webrtc_via_socks_proxy(self, urls="s3.btsfapi.com"): # 配置 ICE 服务器 ice_servers = [RTCIceServer(urls=f"stun:{urls}")] configuration = RTCConfiguration(iceServers=ice_servers) pc = RTCPeerConnection(configuration) # 设置 ICE 候选收集回调 @pc.on("icecandidate") def on_ice_candidate(candidate): if candidate: print(f"ICE 候选: {candidate.candidate}") # 解析候选信息获取 IP # 创建数据通道 dc = pc.createDataChannel("proxy-channel") @dc.on("open") def on_open(): print("通过 SOCKS5 代理的数据通道已打开") @dc.on("message") def on_message(message): print(f"通过代理收到消息: {message}") # 创建 offer offer = await pc.createOffer() offer = await pc.createOffer() await pc.setLocalDescription(offer) sdp_text = pc.localDescription.sdp parsed_sdp = self.parse_sdp(sdp_text) # ip_ = self.extract_srflx_ipv6_from_sdp(sdp_text)[0] port = parsed_sdp.get('media', {}).get('port') # priority = [i.get('priority') for i in parsed_sdp.get('attributes', {}).get('candidates', []) if i.get('related_port') == port][0] # numbers = [str(random.randint(1, 10)) for _ in range(9)] # foundation = "".join(numbers) # foundation = [i.get('foundation') for i in parsed_sdp.get('attributes', {}).get('candidates', []) if i.get('related_port') == port][0] # 模拟添加 ICE 候选 # ice_candidate = RTCIceCandidate(component=1, foundation=foundation, ip=ip_, port=port, priority=priority, protocol="udp", type="host", sdpMid="1", relatedAddress=None, relatedPort=None, sdpMLineIndex=0) # await pc.addIceCandidate(ice_candidate) # print(f"Local SDP (通过 SOCKS5 代理):\n{pc.localDescription.sdp}") # 模拟等待 # await asyncio.sleep(3) sdp_text = pc.localDescription.sdp await pc.close() return sdp_text, self.extract_srflx_ipv6_from_sdp(pc.localDescription.sdp) # def get_index_leaguas(): # webpath = "#AS#B1#K^5#" # api = f"/splashcontentapi/soccertab?lid={langid}&zid=3&pd={urllib.parse.quote(webpath)}&cid=88&cgid=1&ctid=88" # return req(webpath, api) # def get_leaguas_by_country(self, PD_country): # webpath = "#AS#B1#K^5#" # api = f"/splashcontentapi/soccerpartial?lid={langid}&zid=3&pd=%23AS%23B1%23C1%23D1002%23{PD_country}%23I40%23O2%23&cid=88&cgid=1&ctid=88" # return req(webpath, api) # # def get_matchs_by_leagua(PD_leaguas): # webpath = f"#AC#B1#C1#D1002#{PD_leaguas}#G40#" # api = f"/matchmarketscontentapi/markets?lid={langid}&zid=3&pd=%23AC%23B1%23C1%23D1002%23{PD_leaguas}%23G40%23&cid=88&cgid=1&ctid=88" # return req(webpath, api) # # def req(self, webpath, api): # dicts = generate_dict(leagua_data, randomnum, useragent, fisttime, SESSION_ID, api, # f'#{webpath.replace("#", "/")}') # X_net_sync_term = self.generate_X_net_sync_term(dicts, self.SST) # session.headers.update({"X-Net-Sync-Term": X_net_sync_term}) # res = session.get(home_url + api) # print(res.text) # return res.text # '#/AS/B1/K^5/' # def get_index_leaguas(): # webpath = "#AS#B1#K^5#" # api = f"/splashcontentapi/soccertab?lid={langid}&zid=3&pd={urllib.parse.quote(webpath)}&cid=88&cgid=1&ctid=88" # return req(webpath, api) # # def get_leaguas_by_country(PD_country): # webpath = "#AS#B1#K^5#" # api = f"/splashcontentapi/soccerpartial?lid={langid}&zid=3&pd=%23AS%23B1%23C1%23D1002%23{PD_country}%23I40%23O2%23&cid=88&cgid=1&ctid=88" # return req(webpath, api) # # # def get_matchs_by_leagua(PD_leaguas): # webpath = f"#AC#B1#C1#D1002#{PD_leaguas}#G40#" # api = f"/matchmarketscontentapi/markets?lid={langid}&zid=3&pd=%23AC%23B1%23C1%23D1002%23{PD_leaguas}%23G40%23&cid=88&cgid=1&ctid=88" # return req(webpath, api) # # if __name__ == '__main__': # from parse_res import convert_to_dict # langid = 10 # aaa=get_index_leaguas() # print(aaa) # print(convert_to_dict(aaa)) # aaa = get_leaguas_by_country("D1002") # print(aaa) # print(convert_to_dict(aaa)) # from hbmqtt.mqtt.packet import MQTTFixedHeader, PUBLISH, SUBSCRIBE, UNSUBSCRIBE, CONNECT # from hbmqtt.mqtt import PublishPacket # from hbmqtt.adapters import BufferReader # async def parse_mqtt_packet(self, data): # buffer = BufferReader(data) # # 解析固定头部 # fixed_header = await MQTTFixedHeader.from_stream(buffer) # # 根据包类型进行处理 # if fixed_header.packet_type == PUBLISH: # packet = await PublishPacket.from_stream(buffer, fixed_header=fixed_header) # return packet.payload.data # elif fixed_header.packet_type == SUBSCRIBE: # print("\033[1;34m This is a SUBSCRIBE packet. \033[0m") # elif fixed_header.packet_type == UNSUBSCRIBE: # print("\033[1;34m This is an UNSUBSCRIBE packet. \033[0m") # elif fixed_header.packet_type == CONNECT: # print("\033[1;34m This is a CONNECT packet. \033[0m") # else: # print("\033[1;34m This is an unknown packet type. \033[0m")