# -*- coding: utf-8 -*- """ ================================================================================ 千戟快手刷分享 · 三合一至尊整合版 ================================================================================ 【版本选择】 输入 1 → 五彩稳定版 (Python 2/3 通用，低并发，自适应) 输入 2 → 极限极速版 (极速调度，45 线程，风控修复) 输入 3 → 智能自适应版 (高并发，智能调速，150+ 请求/秒) 【新手引导】 如果你是第一次使用，建议从「五彩稳定版」开始，它最稳定、兼容性最好。 熟悉后再尝试「极限极速版」或「智能自适应版」获得更高速度。 【运行环境】 Python 3.6+，安装 requests 库：pip install requests ================================================================================ """ import sys import os import time import re import json import random import hashlib import secrets import threading import ssl from collections import deque, defaultdict # ---------- 尝试导入 requests ---------- try: import requests except ImportError: print("\n❌ 缺少 requests 库！请安装：pip install requests") sys.exit(1) # ============================================================================ # 颜色与UI工具 # ============================================================================ def supports_color(): if not sys.stdout.isatty(): return False if os.name == 'nt': return 'ANSICON' in os.environ or 'WT_SESSION' in os.environ or 'TERM' in os.environ return True USE_COLOR = supports_color() if USE_COLOR: RED = '\033[91m' GREEN = '\033[92m' YELLOW = '\033[93m' BLUE = '\033[94m' MAGENTA = '\033[95m' CYAN = '\033[96m' WHITE = '\033[97m' GRAY = '\033[90m' BOLD = '\033[1m' RESET = '\033[0m' CLEAR = '\033[2J' HOME = '\033[H' CHECK = '✓' CROSS = '✗' ARROW = '▶' PROGRESS_FILL = '█' PROGRESS_EMPTY = '░' SPINNER = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'] RAINBOW = [RED, YELLOW, GREEN, CYAN, BLUE, MAGENTA] else: RED = GREEN = YELLOW = BLUE = MAGENTA = CYAN = WHITE = GRAY = '' BOLD = RESET = CLEAR = HOME = '' CHECK = '[OK]' CROSS = '[FAIL]' ARROW = '>' PROGRESS_FILL = '#' PROGRESS_EMPTY = '-' SPINNER = ['-', '\\', '|', '/'] RAINBOW = [''] * 6 def clear_screen(): if USE_COLOR: sys.stdout.write(CLEAR + HOME) sys.stdout.flush() else: os.system('cls' if os.name == 'nt' else 'clear') def rainbow_text(text): if not USE_COLOR: return text chars = [] for i, ch in enumerate(text): chars.append(f'{RAINBOW[i % len(RAINBOW)]}{ch}{RESET}') return ''.join(chars) def print_header(title, subtitle=""): """打印漂亮的标题头""" print(f'\n{MAGENTA}{BOLD}{"=" * 70}{RESET}') print(f'{MAGENTA}{BOLD}|{RESET} {rainbow_text(title)}{RESET}') if subtitle: print(f'{MAGENTA}{BOLD}|{RESET} {CYAN}{subtitle}{RESET}') print(f'{MAGENTA}{BOLD}{"=" * 70}{RESET}\n') def print_section(title, color=CYAN): """打印章节标题""" print(f'{color}{BOLD}─── {title} ───{RESET}') # ============================================================================ # 版本信息 # ============================================================================ VERSION_INFO = { 1: { 'name': '五彩稳定版', 'color': GREEN, 'icon': '1️⃣', 'desc': 'Python 2/3 通用 · 低并发 2~25 · 自适应调速', 'detail': [ '✅ 兼容 Python 2 和 Python 3，无需额外配置', '✅ 起始并发低，对老设备和网络不友好环境极其友好', '✅ 自适应调速，根据成功率自动调整并发数', '✅ 内置 200+ 伪装头池，防封效果优秀', '✅ 超低 CPU 和内存占用，可长时间运行', '✅ 适合：老设备、网络不稳定、兼容性优先的场景' ], 'auth': '千戟正版 · 通用稳定' }, 2: { 'name': '极限极速版', 'color': YELLOW, 'icon': '2️⃣', 'desc': '极速调度 · 45 线程峰值 · 风控修复 · 2400组伪装', 'detail': [ '✅ 预生成 2400 组真实设备伪装，高度防封', '✅ 极速调度策略：低速起步 → 快速冲至 45 线程峰值', '✅ 失败缓慢降速，风控解除后自动爬坡提速', '✅ 修复所有已知崩溃、死循环、线程死锁问题', '✅ 批量 12 任务、0.02s 极低休眠、2.2s 智能超时', '✅ 适合：追求极致速度、已熟练使用、可接受少量失败' ], 'auth': '千戟正版 · 极限修复' }, 3: { 'name': '智能自适应版', 'color': BLUE, 'icon': '3️⃣', 'desc': '高并发 10~120 · 智能调速 · 150~250 请求/秒', 'detail': [ '✅ 自动探测最佳并发数（10 → 120 动态调整）', '✅ 成功率优先，失败时自动降速，成功后逐步提速', '✅ 预生成 5000 组伪装头，轮询使用，防封效果顶级', '✅ 自适应冷却机制，遇到 429/403 自动暂停等待', '✅ 实时监控面板，显示进度、速度、峰值、线程数、冷却状态', '✅ 详细执行报告：成功/失败数、延迟分布、失败原因分类', '✅ 适合：高速稳定运行、优先成功率、长期批量执行' ], 'auth': '千戟正版 · 智能自适应' } } # ============================================================================ # 显示菜单 # ============================================================================ def show_menu(): """显示主菜单""" clear_screen() print(f'\n{MAGENTA}{BOLD}{"█" * 70}{RESET}') print(f'{MAGENTA}{BOLD}█{RESET} {rainbow_text("千戟快手刷分享 · 三合一至尊整合版")}{RESET}') print(f'{MAGENTA}{BOLD}█{RESET} {CYAN}版本切换 · 一键启动 · 智能防封{RESET}') print(f'{MAGENTA}{BOLD}{"█" * 70}{RESET}\n') print(f' {WHITE}━━━ 请选择要执行的版本 ━━━{RESET}\n') for ver_id, info in VERSION_INFO.items(): color = info['color'] print(f' {color}{BOLD}{info["icon"]} {info["name"]}{RESET}') print(f' {GRAY}└─ {info["desc"]}{RESET}') print(f' {GRAY} └─ 防伪：{info["auth"]}{RESET}\n') print(f' {RED}{BOLD}0️⃣ 退出程序{RESET}') print(f'\n{MAGENTA}{BOLD}{"─" * 70}{RESET}') print(f'{YELLOW}💡 新手建议：先选 {GREEN}1{RESET}{YELLOW} 五彩稳定版，兼容性最好！{RESET}') print(f'{MAGENTA}{BOLD}{"─" * 70}{RESET}') def show_version_detail(ver_id): """显示版本详细说明""" info = VERSION_INFO.get(ver_id) if not info: return clear_screen() color = info['color'] print(f'\n{color}{BOLD}{"=" * 70}{RESET}') print(f'{color}{BOLD}|{RESET} {info["icon"]} {info["name"]}{RESET}') print(f'{color}{BOLD}|{RESET} {GRAY}{info["desc"]}{RESET}') print(f'{color}{BOLD}|{RESET} {GRAY}防伪码：{info["auth"]}{RESET}') print(f'{color}{BOLD}{"=" * 70}{RESET}\n') print(f'{CYAN}{BOLD}【版本特性】{RESET}') for item in info['detail']: print(f' {GREEN}•{RESET} {item}') print(f'\n{color}{BOLD}{"=" * 70}{RESET}') print(f'{YELLOW}按 Enter 键启动此版本，或输入 0 返回主菜单{RESET}') return input(f'{CYAN}👉 选择 (Enter 启动 / 0 返回)：{RESET}').strip() # ============================================================================ # 导入并运行版本（动态执行） # ============================================================================ def run_version(ver_id): """执行对应版本的主函数""" info = VERSION_INFO.get(ver_id) if not info: print(f'{RED}❌ 无效版本{RESET}') return print(f'\n{GREEN}🚀 正在启动 {info["name"]}...{RESET}\n') time.sleep(0.5) try: if ver_id == 1: # 直接调用 v1 的主函数 v1_main() elif ver_id == 2: v2_main() elif ver_id == 3: v3_main() except KeyboardInterrupt: print(f'\n{YELLOW}用户中断，返回主菜单{RESET}') except Exception as e: print(f'\n{RED}❌ 运行出错：{e}{RESET}') import traceback traceback.print_exc() # ============================================================================ # 版本1：五彩稳定版（来自 千戟.py） # ============================================================================ # ---------- 兼容 Python 2/3 ---------- try: from urllib.parse import urlparse, parse_qs except ImportError: from urlparse import urlparse, parse_qs # 版本1的全局配置 V1_API_URL = 'https://www.kuaishou.com/rest/zt/share/w/any' V1_FIX_PARAMS = { 'kpn': 'KUAISHOU_VISION', 'kpf': 'PC_WEB', 'subBiz': 'SINGLE_ROW_WEB', 'sdkVersion': '1.1.2.4.0', 'shareChannel': 'WECHAT', 'shareMethod': 'LINK' } V1_BRANDS = ['Apple','Samsung','Xiaomi','Huawei','OPPO','vivo','OnePlus','Google'] V1_IPHONE_MODELS = ['iPhone15,2','iPhone14,3','iPhone13,4'] V1_ANDROID_MODELS = { 'Samsung':['SM-S918B','SM-S911B'], 'Xiaomi':['2211133C','M2101K7AG'], 'Huawei':['LNA-AL00','RNA-AL00'], 'OPPO':['CPH2357','CPH2325'], 'vivo':['V2144','V2133'], 'OnePlus':['LE2123','LE2113'], 'Google':['Pixel 7 Pro','Pixel 7'] } V1_PC_MODELS = { 'Dell':['XPS 15','Latitude 7420'], 'HP':['Spectre x360','Envy 15'], 'Lenovo':['ThinkPad X1','Yoga 9i'], 'Apple':['MacBookPro18,3','MacBookAir10,1'] } V1_IOS_VERSIONS = ['17.1','17.0','16.7'] V1_ANDROID_VERSIONS = ['14','13','12'] V1_WINDOWS_VERSIONS = ['10.0','11.0'] V1_MACOS_VERSIONS = ['10_15_7','11_0','12_0','13_0','14_0'] V1_ACCEPT_VALUES = ['application/json, text/plain, */*'] V1_ACCEPT_LANGUAGES = ['zh-CN,zh;q=0.9,en;q=0.8'] V1_ACCEPT_ENCODINGS = ['gzip, deflate, br'] V1_SEC_CH_UA_VARIANTS = [ lambda: f'"Google Chrome";v="{random.randint(120,131)}", "Chromium";v="{random.randint(120,131)}", "Not?A_Brand";v="99"' ] V1_ARCHITECTURES = ['"x64"','"arm64"'] V1_BITNESS = ['"64"'] V1_PC_RESOLUTIONS = ['1920x1080','1366x768','2560x1440'] V1_MOBILE_RESOLUTIONS = ['390x844','375x812','414x896'] V1_COLOR_DEPTHS = [24,30] V1_PIXEL_RATIOS = [1.0,2.0,3.0] V1_FAKE_IP_PREFIXES = ['221.219','113.132','119.147','124.95','112.94','183.247','39.144'] V1_TIMEZONES = ['Asia/Shanghai','Asia/Chongqing'] V1_FONTS = ['Microsoft YaHei','SimHei','Arial','PingFang SC'] def v1_generate_chrome_ua(): os_choices = [ f'Windows NT {random.choice(V1_WINDOWS_VERSIONS)}; Win64; x64', f'Macintosh; Intel Mac OS X {random.choice(V1_MACOS_VERSIONS)}', 'X11; Linux x86_64' ] os = random.choice(os_choices) v = random.randint(120,131) return f'Mozilla/5.0 ({os}) AppleWebKit/537.36 Chrome/{v}.0.0.0 Safari/537.36' def v1_generate_mobile_ua(): brand = random.choice(V1_BRANDS) if brand == 'Apple': model = random.choice(V1_IPHONE_MODELS) ver = random.choice(V1_IOS_VERSIONS).replace('.','_') return f'Mozilla/5.0 (iPhone; CPU iPhone OS {ver} like Mac OS X) AppleWebKit/605.1.15 Mobile/15E148' else: model = random.choice(V1_ANDROID_MODELS.get(brand, ['SM-G991B'])) ver = random.choice(V1_ANDROID_VERSIONS) v = random.randint(120,131) return f'Mozilla/5.0 (Linux; Android {ver}; {model}) AppleWebKit/537.36 Chrome/{v}.0.0.0 Mobile Safari/537.36' V1_USER_AGENT_GENERATORS = [v1_generate_chrome_ua, v1_generate_mobile_ua] def v1_get_fake_ip(): prefix = random.choice(V1_FAKE_IP_PREFIXES) return f'{prefix}.{random.randint(1,255)}.{random.randint(1,255)}' def v1_generate_device_info(ua): is_mobile = 'Mobile' in ua or 'iPhone' in ua is_ios = 'iPhone' in ua is_android = 'Android' in ua is_windows = 'Windows' in ua is_mac = 'Macintosh' in ua if is_ios: brand='Apple'; model=random.choice(V1_IPHONE_MODELS); os_ver=random.choice(V1_IOS_VERSIONS); os_name='iOS' elif is_android: brand=random.choice(['Samsung','Xiaomi','Huawei','OPPO','vivo','OnePlus','Google']) model=random.choice(V1_ANDROID_MODELS.get(brand,['SM-G991B'])) os_ver=random.choice(V1_ANDROID_VERSIONS); os_name='Android' elif is_windows: brand=random.choice(['Dell','HP','Lenovo']); model=random.choice(V1_PC_MODELS.get(brand,['Laptop'])) os_ver=random.choice(V1_WINDOWS_VERSIONS); os_name='Windows' elif is_mac: brand='Apple'; model=random.choice(V1_PC_MODELS['Apple']) os_ver=random.choice(V1_MACOS_VERSIONS).replace('_','.'); os_name='macOS' else: brand='Dell'; model='Desktop'; os_ver='Unknown'; os_name='Linux' return { 'is_mobile':is_mobile, 'brand':brand, 'model':model, 'os_name':os_name, 'os_version':os_ver, 'ram':random.choice([4,8,16]), 'cpu_cores':random.choice([4,8]) } def v1_generate_random_cookie(): items = [ f'did=web_{secrets.token_hex(16)}', f'clientid={secrets.token_hex(8)}', f'sid={secrets.token_hex(32)}', f'uid={secrets.token_hex(16)}', f'userId={random.randint(1000000,99999999)}', 'kpn=KUAISHOU_VISION', f'_ga=GA1.2.{random.randint(100000000,999999999)}.{int(time.time())}' ] random.shuffle(items) return '; '.join(items[:random.randint(6,8)]) def v1_generate_headers(): ua = random.choice(V1_USER_AGENT_GENERATORS)() device = v1_generate_device_info(ua) is_mobile = device['is_mobile'] resolution = random.choice(V1_MOBILE_RESOLUTIONS if is_mobile else V1_PC_RESOLUTIONS) w, h = resolution.split('x') fake_ip = v1_get_fake_ip() headers = { 'User-Agent': ua, 'Accept': random.choice(V1_ACCEPT_VALUES), 'Accept-Language': random.choice(V1_ACCEPT_LANGUAGES), 'Accept-Encoding': random.choice(V1_ACCEPT_ENCODINGS), 'Referer': 'https://www.kuaishou.com/', 'Origin': 'https://www.kuaishou.com', 'Connection': 'keep-alive', 'Cache-Control': 'no-cache', 'X-Forwarded-For': fake_ip, 'X-Real-IP': fake_ip, 'Sec-Ch-Ua': random.choice(V1_SEC_CH_UA_VARIANTS)(), 'Sec-Ch-Ua-Mobile': '?1' if is_mobile else '?0', 'Sec-Ch-Ua-Platform': f'"{device["brand"]}"', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'Device-Memory': str(device['ram']), 'Hardware-Concurrency': str(device['cpu_cores']), 'Cookie': v1_generate_random_cookie() } return headers V1_HEADER_POOL = [v1_generate_headers() for _ in range(200)] def v1_get_cached_header(): return random.choice(V1_HEADER_POOL) def v1_get_ks_url(text): m = re.search(r'https://v\.kuaishou\.com/[A-Za-z0-9]+', text) return m.group(0) if m else None def v1_parse_id(url): try: resp = requests.get(url, headers=v1_generate_headers(), allow_redirects=True, timeout=8) final = resp.url m = re.search(r'/photo/([A-Za-z0-9]+)', final) if m: return m.group(1) parsed = urlparse(final) qs = parse_qs(parsed.query) if 'photoId' in qs: return qs['photoId'][0] m2 = re.search(r'v\.kuaishou\.com/([A-Za-z0-9]+)', url) if m2: return m2.group(1) return None except: return None def v1_generate_signature(pid): ts = int(time.time() * 1000) nonce = secrets.token_hex(16) raw = f'shareObjectId={pid}&kpn=KUAISHOU_VISION&nonce={nonce}×tamp={ts}' sig = hashlib.md5(raw.encode()).hexdigest()[:16] return {'_nonce': nonce, '_timestamp': ts, '_signature': sig} V1_lock = threading.Lock() V1_success_count = 0 V1_total_sent = 0 V1_success_times = deque(maxlen=30) V1_current_threads = 3 V1_active_threads = [] V1_thread_lock = threading.Lock() V1_MIN_THREAD = 2 V1_MAX_THREAD = 25 V1_ADJUST_INTERVAL = 8 V1_consecutive_failures = 0 V1_cool_down_until = 0 def v1_task_share(pid, session): global V1_success_count, V1_total_sent, V1_success_times, V1_consecutive_failures, V1_cool_down_until time.sleep(random.uniform(0.3, 1.0)) now = time.time() if now < V1_cool_down_until: time.sleep(V1_cool_down_until - now + random.uniform(0.2, 0.5)) headers = v1_get_cached_header() payload = dict(V1_FIX_PARAMS) payload['shareObjectId'] = pid payload.update(v1_generate_signature(pid)) success = False for attempt in range(2): start = time.time() try: res = session.post(V1_API_URL, json=payload, headers=headers, timeout=4) if res.status_code == 200: data = res.json() if data.get('result') == 1: success = True elapsed = time.time() - start with V1_lock: V1_success_count += 1 V1_success_times.append(elapsed) V1_consecutive_failures = max(0, V1_consecutive_failures - 1) break else: with V1_lock: V1_consecutive_failures += 1 if V1_consecutive_failures > 4: V1_cool_down_until = time.time() + random.uniform(3, 5) time.sleep(1 + attempt) else: time.sleep(0.5 + attempt * 0.5) except: time.sleep(0.5 + attempt * 0.5) if not success: with V1_lock: V1_consecutive_failures += 1 def v1_adjust_threads(): global V1_current_threads with V1_lock: if V1_total_sent < 5: return rate = V1_success_count / float(V1_total_sent) if V1_total_sent > 0 else 0 fail = V1_consecutive_failures avg = sum(V1_success_times) / float(len(V1_success_times)) if V1_success_times else 1.0 speed = 1.0 / avg if avg > 0 else 0 if fail > 6: with V1_thread_lock: V1_current_threads = max(V1_MIN_THREAD, int(V1_current_threads * 0.6)) V1_cool_down_until = time.time() + 4 return if rate < 0.2: with V1_thread_lock: V1_current_threads = max(V1_MIN_THREAD, V1_current_threads - 1) elif rate > 0.7 and speed > 0.5: with V1_thread_lock: V1_current_threads = min(V1_MAX_THREAD, V1_current_threads + 1) def v1_progress_bar(percent, width=20): filled = int(width * percent / 100) return PROGRESS_FILL * filled + PROGRESS_EMPTY * (width - filled) def v1_run_adaptive(pid, total): global V1_success_count, V1_total_sent, V1_consecutive_failures, V1_cool_down_until, V1_current_threads, V1_active_threads V1_success_count = 0 V1_total_sent = 0 V1_consecutive_failures = 0 V1_cool_down_until = 0 V1_current_threads = V1_MIN_THREAD start_time = time.time() print(f'{CYAN}[*] 初始化Session池...{RESET}') session_pool = [requests.Session() for _ in range(V1_MAX_THREAD)] for s in session_pool: s.headers.update({'Connection': 'keep-alive'}) print(f'{GREEN}[+] Session池就绪 ({V1_MAX_THREAD}个){RESET}') completed = 0 adjust_counter = 0 stop_flag = False spin_idx = 0 def progress_printer(): nonlocal spin_idx while not stop_flag: time.sleep(0.7) now = time.time() with V1_lock: succ = V1_success_count sent = V1_total_sent fail = V1_consecutive_failures cd = V1_cool_down_until rate = (succ / float(sent) * 100) if sent > 0 else 0 speed = sent / (now - start_time) if now > start_time else 0 bar = v1_progress_bar(rate) sp = SPINNER[spin_idx % len(SPINNER)] spin_idx += 1 cd_status = f' [CD:{cd-now:.0f}s]' if cd > now else '' if USE_COLOR: rate_color = GREEN if rate > 70 else YELLOW if rate > 40 else RED line = (f'\r{BLUE}{ARROW}{RESET} {sp} {bar} ' f'{succ}/{total} {YELLOW}c:{V1_current_threads}{RESET} ' f'{CYAN}sp:{speed:.1f}/s{RESET} ' f'{rate_color}rt:{rate:.1f}%{RESET} ' f'fail:{fail}{cd_status}') else: line = (f'\r{ARROW} {sp} {bar} ' f'{succ}/{total} c:{V1_current_threads} ' f'sp:{speed:.1f}/s rt:{rate:.1f}% ' f'fail:{fail}{cd_status}') sys.stdout.write(line) sys.stdout.flush() printer = threading.Thread(target=progress_printer, daemon=True) printer.start() print(f'{GREEN}[+] 开始发送请求...{RESET}\n') V1_active_threads = [] while completed < total: now = time.time() if now < V1_cool_down_until: time.sleep(0.2) continue with V1_thread_lock: V1_active_threads = [t for t in V1_active_threads if t.is_alive()] running = len(V1_active_threads) available = V1_current_threads - running batch = min(available, total - completed, 2) for _ in range(batch): idx = completed % len(session_pool) t = threading.Thread(target=v1_task_share, args=(pid, session_pool[idx])) t.start() V1_active_threads.append(t) completed += 1 with V1_lock: V1_total_sent += 1 adjust_counter += batch if adjust_counter >= V1_ADJUST_INTERVAL: adjust_counter = 0 v1_adjust_threads() if batch == 0 and completed < total: time.sleep(0.1) for t in V1_active_threads: t.join() stop_flag = True printer.join(timeout=2) elapsed = time.time() - start_time speed = total / elapsed if elapsed > 0 else 0 rate = (V1_success_count / float(total) * 100) if total > 0 else 0 print(f'\n\n{GREEN}{BOLD}═══ 执行完成报告 ═══{RESET}') print(f' 总发送次数: {total}') print(f' 成功次数: {V1_success_count}') print(f' 成功率: {rate:.1f}%') print(f' 总耗时: {elapsed:.1f}s') print(f' 平均速度: {speed:.1f} 次/秒') print(f' 最终并发: {V1_current_threads}') print(f' 防伪状态: 千戟正版 · 五彩稳定\n') def v1_main(): print(f'\n{CYAN}{BOLD}━━━ 千戟 · 五彩稳定版 ━━━{RESET}') print(f'{GRAY}适用：老设备 / 网络不稳定 / 兼容性优先{RESET}\n') print(f'{CYAN}请输入作品链接或短链接 (可直接粘贴):{RESET}') link = input().strip() if not link: print(f'{RED}[!] 输入为空，退出{RESET}') return print(f'{YELLOW}[*] 正在解析链接...{RESET}') ks_url = v1_get_ks_url(link) if not ks_url: print(f'{RED}[!] 无法提取有效链接{RESET}') return print(f'{GREEN}[+] 提取到短链接: {ks_url}{RESET}') photo_id = v1_parse_id(ks_url) if not photo_id: print(f'{YELLOW}[!] 自动解析失败，请手动输入作品ID:{RESET}') photo_id = input().strip() if not photo_id: print(f'{RED}[!] 未输入ID，退出{RESET}') return print(f'{GREEN}[+] 作品ID: {photo_id}{RESET}') try: print(f'\n{CYAN}请输入要执行的分享次数 (建议50~200次/轮):{RESET}') total = int(input().strip()) if total <= 0: print(f'{RED}[!] 次数必须大于0{RESET}') return except: print(f'{RED}[!] 请输入有效数字{RESET}') return print(f'{MAGENTA}[>] 目标次数: {total} 次{RESET}') print(f'{MAGENTA}[>] 自适应并发范围: {V1_MIN_THREAD} ~ {V1_MAX_THREAD}{RESET}') print(f'{YELLOW}[>] 按 Ctrl+C 可随时中断{RESET}') time.sleep(1.5) v1_run_adaptive(photo_id, total) # ============================================================================ # 版本2：极限极速版（来自 千戟v2.py） # ============================================================================ V2_MIN_THREAD = 2 V2_MAX_THREAD = 45 V2_FAST_STEP_UP = 3 V2_SMALL_FAIL_STEP_DOWN = 1 V2_BIG_FAIL_STEP_DOWN = 2 V2_ADJUST_INTERVAL = 3 V2_SAFE_SUCC_RATE = 0.60 V2_RECOVER_SUCC_RATE = 0.50 V2_DANGER_SUCC_RATE = 0.35 V2_COOL_TINY = 1.0 V2_COOL_MID = 2.2 V2_COOL_LONG = 6.0 V2_COOL_CRASH = 8.5 V2_REQUEST_TIMEOUT = 2.2 V2_MIN_TASK_SLEEP = 0.02 V2_MAX_TASK_SLEEP = 0.12 V2_LOOP_MIN_DELAY = 0.005 V2_BATCH_MAX_TASK = 12 V2_API_URL = 'https://www.kuaishou.com/rest/zt/share/w/any' V2_FIX_PARAMS = { 'kpn': 'KUAISHOU_VISION', 'kpf': 'PC_WEB', 'subBiz': 'SINGLE_ROW_WEB', 'sdkVersion': '1.1.2.4.0', 'shareChannel': 'WECHAT', 'shareMethod': 'LINK' } V2_BRANDS = ['Apple','Samsung','Xiaomi','Huawei','OPPO','vivo','OnePlus','Google','Meizu','realme','iQOO','Honor','Nubia','BlackShark','ROG'] V2_IPHONE_MODELS = ['iPhone15,2','iPhone14,3','iPhone13,4','iPhone16,1','iPhone15,3','iPhone14,6','iPhone13,1','iPhone16,2','iPhone16,3'] V2_ANDROID_MODELS = { 'Samsung':['SM-S918B','SM-S911B','SM-G998B','SM-F946B','SM-A546E','SM-A346E','SM-S928B'], 'Xiaomi':['2211133C','M2101K7AG','2304FPN6DC','2203121C','24041696G','23127PN5BC','24072PN5AC'], 'Huawei':['LNA-AL00','RNA-AL00','ELS-AN00','VOG-AL10','TET-AN00','NOP-AN00','MNA-AL00'], 'OPPO':['CPH2357','CPH2325','PGEM10','PFZM10','PFFM20','PJC110','PGW110'], 'vivo':['V2144','V2133','V2304A','V2242A','V2409','V2319A','V2418A'], 'OnePlus':['LE2123','LE2113','NE2211','PHB110','KF110','NE2411'], 'Google':['Pixel 7 Pro','Pixel 7','Pixel 8','Pixel 8 Pro','Pixel 9','Pixel 9 Pro'], 'Meizu':['M1931','M1872','M2392','M2281','M2452'], 'realme':['RMX3700','RMX3610','RMX3850','RMX3900','RMX4010'], 'iQOO':['V2243A','I2211','V2301A','V2401A','V2412A'], 'Honor':['ELZ-AN00','HPB-AN00','BMH-AN10','FNE-AN00','GIA-AN00'], 'Nubia':['NX729J','NX709J','NX699J','NX732J'], 'BlackShark':['SKW-A0','KSR-A0'], 'ROG':['AI2201','AS_AI2301'] } V2_PC_MODELS = { 'Dell':['XPS 15','Latitude 7420','G15 5530','Inspiron 14','OptiPlex 7010','XPS 17'], 'HP':['Spectre x360','Envy 15','Victus 16','Pavilion 14','ZBook','暗影精灵10'], 'Lenovo':['ThinkPad X1','Yoga 9i','拯救者Y9000P','小新Pro16','ThinkBook 14','R9000X'], 'Apple':['MacBookPro18,3','MacBookAir10,1','MacBookPro19,1','iMac21,1','MacMini9,1','MacBookPro20,1'], 'Asus':['ZenBook Pro','ROG幻16','无畏Pro14','天选4','ProArt'] } V2_IOS_VERSIONS = ['17.1','17.0','16.7','18.0','17.4','16.6','18.1','18.2'] V2_ANDROID_VERSIONS = ['14','13','12','11','10','9'] V2_WINDOWS_VERSIONS = ['10.0','11.0'] V2_MACOS_VERSIONS = ['10_15_7','11_0','12_0','13_0','14_0','15_0'] V2_CHROME_VER_RANGE = list(range(120, 140)) V2_ACCEPT_VALUES = ['application/json, text/plain, */*'] V2_ACCEPT_LANGUAGES = ['zh-CN,zh;q=0.9','zh-CN,zh;q=0.95','zh-CN,zh-TW;q=0.9','zh-CN'] V2_ACCEPT_ENCODINGS = ['gzip, deflate, br'] V2_SEC_CH_UA_VARIANTS = [lambda: f'"Google Chrome";v="{random.choice(V2_CHROME_VER_RANGE)}", "Chromium";v="{random.choice(V2_CHROME_VER_RANGE)}", "Not=A?Brand";v="99"'] V2_ARCHITECTURES = ['"x64"','"arm64"'] V2_BITNESS = ['"64"'] V2_PC_RESOLUTIONS = ['1920x1080','1366x768','2560x1440','3840x2160','1600x900','1280x720','2880x1800'] V2_MOBILE_RESOLUTIONS = ['390x844','375x812','414x896','430x932','360x800','412x915','480x1080'] V2_COLOR_DEPTHS = [24,30,32] V2_PIXEL_RATIOS = [1.0,2.0,2.5,3.0,3.5] V2_FAKE_IP_PREFIXES = [ '221.219','113.132','119.147','124.95','183.247','39.144','61.184','106.121','122.193','115.48','116.224', '112.94','111.175','27.188','180.113','223.72','120.92','183.191','182.130','218.201', '110.242','125.39','123.125','119.4','114.246','118.248','42.231' ] V2_TIMEZONES = ['Asia/Shanghai','Asia/Chongqing','Asia/Hong_Kong'] V2_FONTS = ['Microsoft YaHei','SimHei','Arial','PingFang SC','Hiragino Sans GB','Noto Sans SC','SimSun','Microsoft YaHei UI'] V2_RAM_LIST = [4,6,8,12,16,32,64] V2_CPU_CORE_LIST = [4,6,8,10,12,16,20] def v2_generate_chrome_ua(): os_choices = [ f'Windows NT {random.choice(V2_WINDOWS_VERSIONS)}; Win64; x64', f'Macintosh; Intel Mac OS X {random.choice(V2_MACOS_VERSIONS)}', 'X11; Linux x86_64' ] os_sys = random.choice(os_choices) ver = random.choice(V2_CHROME_VER_RANGE) return f'Mozilla/5.0 ({os_sys}) AppleWebKit/537.36 Chrome/{ver}.0.0.0 Safari/537.36' def v2_generate_mobile_ua(): brand = random.choice(V2_BRANDS) if brand == 'Apple': model = random.choice(V2_IPHONE_MODELS) ios_ver = random.choice(V2_IOS_VERSIONS).replace('.','_') return f'Mozilla/5.0 (iPhone; CPU iPhone OS {ios_ver} like Mac OS X) AppleWebKit/605.1.15 Mobile/15E148' else: model = random.choice(list(V2_ANDROID_MODELS.keys())) dev_model = random.choice(V2_ANDROID_MODELS[model]) android_ver = random.choice(V2_ANDROID_VERSIONS) chrome_ver = random.choice(V2_CHROME_VER_RANGE) return f'Mozilla/5.0 (Linux; Android {android_ver}; {dev_model}) AppleWebKit/537.36 Chrome/{chrome_ver}.0.0.0 Mobile Safari/537.36' V2_USER_AGENT_GENERATORS = [v2_generate_chrome_ua, v2_generate_mobile_ua] def v2_get_fake_ip(): prefix = random.choice(V2_FAKE_IP_PREFIXES) return f'{prefix}.{random.randint(1,255)}.{random.randint(1,255)}' def v2_get_ip_operator(ip): seg = ip.split('.')[0] telecom_seg = ['221','113','119','124','183','39','61','106','122','115','116'] mobile_seg = ['112','111','27','180','223','120','183','182','218'] unicom_seg = ['110','125','123','119','114','118','42'] if seg in telecom_seg: return "电信" elif seg in mobile_seg: return "移动" elif seg in unicom_seg: return "联通" return "未知运营商" def v2_generate_device_info(ua): is_mobile = 'Mobile' in ua or 'iPhone' in ua is_ios = 'iPhone' in ua is_android = 'Android' in ua is_windows = 'Windows' in ua is_mac = 'Macintosh' in ua brand = model = os_ver = os_name = "" if is_ios: brand='Apple' model=random.choice(V2_IPHONE_MODELS) os_ver=random.choice(V2_IOS_VERSIONS) os_name='iOS' elif is_android: brand=random.choice(list(V2_ANDROID_MODELS.keys())) model=random.choice(V2_ANDROID_MODELS[brand]) os_ver=random.choice(V2_ANDROID_VERSIONS) os_name='Android' elif is_windows: brand=random.choice(['Dell','HP','Lenovo','Asus']) model=random.choice(V2_PC_MODELS[brand]) os_ver=random.choice(V2_WINDOWS_VERSIONS) os_name='Windows' elif is_mac: brand='Apple' model=random.choice(V2_PC_MODELS['Apple']) os_ver=random.choice(V2_MACOS_VERSIONS).replace('_','.') os_name='macOS' else: brand='Dell' model='Desktop' os_ver='Unknown' os_name='Linux' return { 'is_mobile':is_mobile, 'brand':brand, 'model':model, 'os_name':os_name, 'os_version':os_ver, 'ram':random.choice(V2_RAM_LIST), 'cpu_cores':random.choice(V2_CPU_CORE_LIST), 'pixel_ratio':random.choice(V2_PIXEL_RATIOS), 'color_depth':random.choice(V2_COLOR_DEPTHS) } def v2_generate_full_cookie(): cookie_items = [ f'did=web_{secrets.token_hex(16)}', f'clientid={secrets.token_hex(8)}', f'sid={secrets.token_hex(32)}', f'uid={secrets.token_hex(16)}', f'userId={random.randint(1000000,999999999)}', 'kpn=KUAISHOU_VISION', f'_ga=GA1.2.{random.randint(100000000,999999999)}.{int(time.time())}', f'_gid=GA1.2.{random.randint(100000000,999999999)}.{int(time.time())}', f'_gat_gtag_UA_{random.randint(100000,999999)}=1', f'visitor_id={secrets.token_hex(20)}', f'cache_token={secrets.token_hex(12)}', f'user_trace={secrets.token_hex(24)}', f'page_mark={secrets.token_hex(10)}' ] random.shuffle(cookie_items) return '; '.join(cookie_items[:random.randint(10,15)]) def v2_generate_headers_full(): ua = random.choice(V2_USER_AGENT_GENERATORS)() device = v2_generate_device_info(ua) is_mobile = device['is_mobile'] resolution = random.choice(V2_MOBILE_RESOLUTIONS if is_mobile else V2_PC_RESOLUTIONS) fake_ip = v2_get_fake_ip() headers = { 'User-Agent': ua, 'Accept': random.choice(V2_ACCEPT_VALUES), 'Accept-Language': random.choice(V2_ACCEPT_LANGUAGES), 'Accept-Encoding': random.choice(V2_ACCEPT_ENCODINGS), 'Referer': 'https://www.kuaishou.com/', 'Origin': 'https://www.kuaishou.com', 'Connection': 'keep-alive', 'Cache-Control': 'no-cache', 'X-Forwarded-For': fake_ip, 'X-Real-IP': fake_ip, 'Sec-Ch-Ua': random.choice(V2_SEC_CH_UA_VARIANTS)(), 'Sec-Ch-Ua-Mobile': '?1' if is_mobile else '?0', 'Sec-Ch-Ua-Platform': f'"{device["brand"]}"', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'Device-Memory': str(device['ram']), 'Hardware-Concurrency': str(device['cpu_cores']), 'Cookie': v2_generate_full_cookie() } return headers, fake_ip, device V2_HEADER_POOL = [] V2_DEVICE_POOL = [] V2_IP_POOL = [] V2_OPERATOR_POOL = [] print(f'{CYAN}[V2系统初始化] 预生成2400组伪装缓存池...{RESET}') for _ in range(2400): try: h, ip, dev = v2_generate_headers_full() op = v2_get_ip_operator(ip) V2_HEADER_POOL.append(h) V2_DEVICE_POOL.append(dev) V2_IP_POOL.append(ip) V2_OPERATOR_POOL.append(op) except Exception: continue print(f'{GREEN}[V2初始化完成] 2400组伪装数据加载完毕{RESET}') def v2_get_cached_full_info(): idx = random.randint(0, max(0, len(V2_HEADER_POOL)-1)) return V2_HEADER_POOL[idx], V2_IP_POOL[idx], V2_DEVICE_POOL[idx], V2_OPERATOR_POOL[idx] def v2_extract_ks_link(raw_text): pat1 = r'https://v\.kuaishou\.com/[A-Za-z0-9]+' pat2 = r'https://www\.kuaishou\.com/[^\s]+' m1 = re.search(pat1, raw_text) m2 = re.search(pat2, raw_text) if m1: return m1.group(0), "短链 v.kuaishou.com" if m2: return m2.group(0), "官网完整链接" return None, "未识别快手链接" def v2_parse_work_pid(url): print(f'\n{CYAN}{ARROW} V2链接解析启动{RESET}') print(f'{WHITE}原始链接：{url}{RESET}') try: hd,_,_,_ = v2_get_cached_full_info() resp = requests.get(url, headers=hd, allow_redirects=True, timeout=8) final_url = resp.url print(f'{WHITE}最终地址：{final_url}{RESET}') path_match = re.search(r'/photo/([A-Za-z0-9]+)', final_url) if path_match: pid = path_match.group(1) print(f'{GREEN}{CHECK}V2路径匹配ID:{pid}{RESET}') return pid parse_res = urlparse(final_url) qs = parse_qs(parse_res.query) if 'photoId' in qs: pid = qs['photoId'][0] print(f'{GREEN}{CHECK}V2参数匹配ID:{pid}{RESET}') return pid short_m = re.search(r'v\.kuaishou\.com/([A-Za-z0-9]+)', url) if short_m: code = short_m.group(1) print(f'{YELLOW}{ARROW}V2短链二次解析 {code}{RESET}') sub_resp = requests.get(f'https://v.kuaishou.com/{code}', headers=hd, allow_redirects=True, timeout=6) sub_final = sub_resp.url sub_pat = re.search(r'/photo/([A-Za-z0-9]+)', sub_final) if sub_pat: pid = sub_pat.group(1) print(f'{GREEN}{CHECK}V2短链解析成功ID:{pid}{RESET}') return pid print(f'{RED}{CROSS}V2全部规则解析失败{RESET}') return None except Exception as e: print(f'{RED}{CROSS}V2链接访问异常:{str(e)}{RESET}') return None def v2_build_sign(pid): ts = int(time.time() * 1000) nonce = secrets.token_hex(16) raw = f'shareObjectId={pid}&kpn=KUAISHOU_VISION&nonce={nonce}×tamp={ts}' sign = hashlib.md5(raw.encode('utf-8')).hexdigest()[:16] return {'_nonce': nonce, '_timestamp': ts, '_signature': sign} V2_lock = threading.Lock() V2_success_count = 0 V2_total_sent = 0 V2_fail_total = 0 V2_target_task_num = 0 V2_complete_task = 0 V2_success_delay_list = deque(maxlen=200) V2_fail_detail = defaultdict(int) V2_current_thread = V2_MIN_THREAD V2_active_thread_list = [] V2_thread_mux = threading.Lock() V2_consecutive_fail = 0 V2_global_cool_end = 0 V2_cool_trigger_records = [] V2_speed_adjust_records = [] V2_peak_per_second = 0.0 V2_min_per_second = 9999.0 V2_global_start_time = 0 def v2_single_share_task(pid, session, task_id): global V2_success_count, V2_total_sent, V2_fail_total, V2_consecutive_fail, V2_global_cool_end, V2_complete_task global V2_fail_detail, V2_cool_trigger_records, V2_success_delay_list try: time.sleep(random.uniform(V2_MIN_TASK_SLEEP, V2_MAX_TASK_SLEEP)) now = time.time() if now < V2_global_cool_end: wait_sec = round(V2_global_cool_end - now, 2) print(f'{MAGENTA}[V2任务{task_id}] 冷却等待 {wait_sec}s{RESET}') time.sleep(wait_sec + 0.03) hd, fake_ip, dev, op = v2_get_cached_full_info() payload = dict(V2_FIX_PARAMS) payload['shareObjectId'] = pid payload.update(v2_build_sign(pid)) task_ok = False req_start = time.time() for retry in range(1): try: res = session.post(V2_API_URL, json=payload, headers=hd, timeout=V2_REQUEST_TIMEOUT) cost = round(time.time() - req_start, 3) dev_text = f"{dev['brand']} {dev['model']} 内存{dev['ram']}G CPU{dev['cpu_cores']}" ip_text = f"IP:{fake_ip} {op}" print(f'{WHITE}[V2任务{task_id}] {ip_text} | {dev_text} 延迟{cost}s 并发{V2_current_thread}{RESET}') if res.status_code == 200: data = res.json() ret_code = data.get('result', -999) if ret_code == 1: task_ok = True with V2_lock: V2_success_count += 1 V2_success_delay_list.append(cost) V2_consecutive_fail = max(0, V2_consecutive_fail - 2) print(f'{GREEN}[V2任务{task_id}] {CHECK}分享成功{RESET}') break else: err_msg = data.get('errorMsg', '无提示') err_key = f'业务码{ret_code}:{err_msg}' with V2_lock: V2_fail_total += 1 V2_fail_detail[err_key] += 1 V2_consecutive_fail += 1 print(f'{YELLOW}[V2任务{task_id}] {CROSS}{err_key}{RESET}') time.sleep(V2_COOL_TINY) elif res.status_code == 429: err_key = "429请求限流" with V2_lock: V2_fail_total +=1 V2_fail_detail[err_key] +=1 V2_consecutive_fail +=4 V2_global_cool_end = time.time() + V2_COOL_LONG V2_cool_trigger_records.append(f"429限流长冷却{V2_COOL_LONG}s") print(f'{RED}[V2任务{task_id}] {CROSS}触发限流{RESET}') time.sleep(V2_COOL_LONG) elif res.status_code == 403: err_key = "403访问封禁" with V2_lock: V2_fail_total +=1 V2_fail_detail[err_key] +=1 V2_consecutive_fail +=5 V2_global_cool_end = time.time() + V2_COOL_CRASH V2_cool_trigger_records.append(f"403超长冷却{V2_COOL_CRASH}s") print(f'{RED}[V2任务{task_id}] {CROSS}封禁拦截{RESET}') time.sleep(V2_COOL_CRASH) elif 500 <= res.status_code < 600: err_key = f"服务器{res.status_code}" with V2_lock: V2_fail_total +=1 V2_fail_detail[err_key] +=1 V2_consecutive_fail +=2 print(f'{RED}[V2任务{task_id}] {CROSS}服务器异常{RESET}') time.sleep(V2_COOL_MID) else: err_key = f"HTTP{res.status_code}" with V2_lock: V2_fail_total +=1 V2_fail_detail[err_key] +=1 V2_consecutive_fail +=1 print(f'{RED}[V2任务{task_id}] {CROSS}网络异常{RESET}') time.sleep(V2_COOL_TINY) except requests.exceptions.Timeout: err_key = "请求超时" with V2_lock: V2_fail_total +=1 V2_fail_detail[err_key] +=1 V2_consecutive_fail +=1 print(f'{RED}[V2任务{task_id}] {CROSS}接口超时{RESET}') time.sleep(V2_COOL_TINY) except Exception as e: err_key = f"内部异常:{str(e)[:40]}" with V2_lock: V2_fail_total +=1 V2_fail_detail[err_key] +=1 V2_consecutive_fail +=1 print(f'{RED}[V2任务{task_id}] {CROSS}{str(e)}{RESET}') with V2_lock: V2_total_sent += 1 V2_complete_task += 1 except Exception as e: with V2_lock: V2_total_sent +=1 V2_complete_task +=1 V2_fail_total +=1 print(f'{GRAY}[V2任务{task_id}] 线程内部捕获异常跳过:{str(e)}{RESET}') def v2_auto_adjust_thread(): global V2_current_thread, V2_global_cool_end with V2_lock: sent = V2_total_sent succ = V2_success_count fail_cnt = V2_consecutive_fail if sent < 12: log = f"预热样本{sent}，维持{V2_current_thread}" print(f'{CYAN}{ARROW}{log}{RESET}') V2_speed_adjust_records.append(log) return succ_rate = succ / float(sent) if sent > 0 else 0 avg_delay = sum(V2_success_delay_list)/len(V2_success_delay_list) if V2_success_delay_list else 1.0 req_speed = 1 / avg_delay if avg_delay > 0 else 0 if fail_cnt >= 9: with V2_thread_mux: old = V2_current_thread V2_current_thread = max(V2_MIN_THREAD, V2_current_thread - V2_BIG_FAIL_STEP_DOWN) V2_global_cool_end = time.time() + V2_COOL_LONG log = f"连续失败{fail_cnt}，{old}→{V2_current_thread}长冷却" V2_cool_trigger_records.append(log) V2_speed_adjust_records.append(log) print(f'{RED}{ARROW}{log}{RESET}') return if succ_rate < V2_DANGER_SUCC_RATE: with V2_thread_mux: old = V2_current_thread new_t = max(V2_MIN_THREAD, V2_current_thread - V2_SMALL_FAIL_STEP_DOWN) if new_t != V2_current_thread: V2_current_thread = new_t log = f"低成功率{succ_rate:.1%} {old}→{V2_current_thread}" V2_speed_adjust_records.append(log) print(f'{YELLOW}{ARROW}{log}{RESET}') elif V2_RECOVER_SUCC_RATE <= succ_rate < V2_SAFE_SUCC_RATE: with V2_thread_mux: old = V2_current_thread new_t = min(V2_MAX_THREAD, V2_current_thread + 1) if new_t != V2_current_thread: V2_current_thread = new_t log = f"风控回升爬坡 {old}→{V2_current_thread}" V2_speed_adjust_records.append(log) print(f'{BLUE}{ARROW}{log}{RESET}') elif succ_rate > V2_SAFE_SUCC_RATE and req_speed > 0.4: with V2_thread_mux: old = V2_current_thread new_t = min(V2_MAX_THREAD, V2_current_thread + V2_FAST_STEP_UP) if new_t != V2_current_thread: V2_current_thread = new_t log = f"状态优秀极速+3 {old}→{V2_current_thread}" V2_speed_adjust_records.append(log) print(f'{GREEN}{ARROW}{log}{RESET}') else: log = f"平稳运行 并发{V2_current_thread} 成功率{succ_rate:.1%}" print(f'{BLUE}{ARROW}{log}{RESET}') V2_speed_adjust_records.append(log) def v2_rainbow_text(text): if not USE_COLOR: return text res = [] for i, c in enumerate(text): res.append(f'{RAINBOW[i%len(RAINBOW)]}{c}{RESET}') return ''.join(res) def v2_progress_bar(percent, width=34): fill = int(width * percent / 100) return PROGRESS_FILL * fill + PROGRESS_EMPTY * (width - fill) def v2_monitor_panel(): spin_idx = 0 stop_flag = False global V2_peak_per_second, V2_min_per_second while not stop_flag: try: time.sleep(0.2) now = time.time() with V2_lock: succ = V2_success_count sent = V2_total_sent fail = V2_fail_total cons_fail = V2_consecutive_fail cd_end = V2_global_cool_end comp = V2_complete_task target = V2_target_task_num run_sec = now - V2_global_start_time if sent == 0: succ_rate = 0.0 per_sec = 0.0 avg_delay = 0.0 else: succ_rate = (succ / sent) * 100 per_sec = sent / run_sec if run_sec > 0 else 0 avg_delay = sum(V2_success_delay_list)/len(V2_success_delay_list) if V2_success_delay_list else 0 finish_pct = (comp / target)*100 if target>0 else 0 bar = v2_progress_bar(finish_pct, 34) sp = SPINNER[spin_idx % len(SPINNER)] spin_idx += 1 if per_sec > V2_peak_per_second: V2_peak_per_second = per_sec if sent > 15 and per_sec < V2_min_per_second: V2_min_per_second = per_sec cd_text = f"冷却剩余:{round(cd_end-now,1)}s" if cd_end>now else "无冷却" rate_color = GREEN if succ_rate>=70 else YELLOW if succ_rate>=45 else RED panel = ( f"\r{WHITE}====================【V2极限极速监控】===================={RESET}\n" f"{CYAN}进度 [{bar}] {finish_pct:.2f}% 状态:{sp}{RESET}\n" f"{BLUE}目标:{target} 已分配:{comp} 请求:{sent} 成功:{succ} 失败:{fail}{RESET}\n" f"{rate_color}成功率:{succ_rate:.2f}% 平均延迟:{avg_delay:.3f}s{RESET}\n" f"{MAGENTA}当前:{per_sec:.2f}/s 峰值:{V2_peak_per_second:.2f}/s 最低:{V2_min_per_second:.2f}/s{RESET}\n" f"{YELLOW}线程:{V2_current_thread} 连续失败:{cons_fail} | {cd_text}{RESET}\n" f"{WHITE}====================================================================={RESET}" ) sys.stdout.write(panel) sys.stdout.flush() except Exception: continue def v2_run_core(pid, total_target): global V2_success_count, V2_total_sent, V2_fail_total, V2_consecutive_fail, V2_global_cool_end, V2_current_thread global V2_complete_task, V2_target_task_num, V2_global_start_time, V2_cool_trigger_records, V2_success_delay_list global V2_peak_per_second, V2_min_per_second, V2_fail_detail, V2_active_thread_list, V2_speed_adjust_records V2_success_count = V2_total_sent = V2_fail_total = V2_consecutive_fail = V2_complete_task = 0 V2_global_cool_end = 0 V2_current_thread = V2_MIN_THREAD V2_target_task_num = total_target V2_global_start_time = time.time() V2_cool_trigger_records.clear() V2_speed_adjust_records.clear() V2_fail_detail.clear() V2_success_delay_list.clear() V2_peak_per_second = 0.0 V2_min_per_second = 9999.0 print(f'\n{CYAN}{ARROW}初始化{V2_MAX_THREAD}长连接池{RESET}') session_pool = [] for _ in range(V2_MAX_THREAD): s = requests.Session() s.headers['Connection'] = 'keep-alive' s.keep_alive = True adapter = requests.adapters.HTTPAdapter(pool_connections=V2_MAX_THREAD, pool_maxsize=V2_MAX_THREAD, max_retries=0) s.mount("https://", adapter) session_pool.append(s) print(f'{GREEN}{CHECK}V2连接池+2400伪装池就绪{RESET}') monitor_t = threading.Thread(target=v2_monitor_panel, daemon=True) monitor_t.start() print(f'\n{MAGENTA}{v2_rainbow_text("极限极速稳定模式启动")}{RESET}\n') V2_active_thread_list = [] adjust_counter = 0 task_id = 1 while V2_complete_task < total_target: now = time.time() if now < V2_global_cool_end: time.sleep(V2_LOOP_MIN_DELAY) continue with V2_thread_mux: V2_active_thread_list = [t for t in V2_active_thread_list if t.is_alive()] running_num = len(V2_active_thread_list) idle = V2_current_thread - running_num batch = min(idle, total_target - V2_complete_task, V2_BATCH_MAX_TASK) for _ in range(batch): sess_idx = task_id % len(session_pool) t = threading.Thread(target=v2_single_share_task, args=(pid, session_pool[sess_idx], task_id), daemon=True) t.start() V2_active_thread_list.append(t) task_id += 1 adjust_counter += batch if adjust_counter >= V2_ADJUST_INTERVAL: adjust_counter = 0 v2_auto_adjust_thread() if batch == 0 and V2_complete_task < total_target: time.sleep(V2_LOOP_MIN_DELAY) print(f'\n{YELLOW}{ARROW}V2任务分配完成，等待线程收尾{RESET}') for t in V2_active_thread_list: try: t.join(timeout=3) except Exception: continue time.sleep(1.0) total_run = round(time.time() - V2_global_start_time, 2) avg_throughput = total_target / total_run if total_run > 0 else 0 final_succ_rate = (V2_success_count / V2_total_sent * 100) if V2_total_sent > 0 else 0 print(f'\n\n{RAINBOW[0]}+========================================================================================+{RESET}') print(f'{RAINBOW[0]}|{RESET}{v2_rainbow_text("极限极速稳定修复版 完整执行报告")}{RAINBOW[0]}|') print(f'{RAINBOW[0]}+========================================================================================+{RESET}') print(f'{CYAN}【一、基础统计】{RESET}') print(f'目标总次数：{BOLD}{total_target}{RESET}') print(f'实际请求：{BOLD}{V2_total_sent}{RESET}') print(f'成功分享：{GREEN}{BOLD}{V2_success_count}{RESET}') print(f'失败请求：{RED}{BOLD}{V2_fail_total}{RESET}') print(f'综合成功率：{GREEN if final_succ_rate>60 else YELLOW}{BOLD}{final_succ_rate:.2f}%{RESET}') print(f'\n{CYAN}【二、速度性能】{RESET}') print(f'总运行时长：{BOLD}{total_run}s{RESET}') print(f'平均速度：{BOLD}{avg_throughput:.2f}次/秒{RESET}') print(f'峰值速度：{BOLD}{V2_peak_per_second:.2f}次/秒{RESET}') print(f'最低速度：{BOLD}{V2_min_per_second:.2f}次/秒{RESET}') if len(V2_success_delay_list) > 0: avg_delay_all = sum(V2_success_delay_list)/len(V2_success_delay_list) print(f'平均接口延迟：{BOLD}{round(avg_delay_all,3)}s{RESET}') print(f'\n{CYAN}【三、并发调速记录】{RESET}') print(f'线程区间 {V2_MIN_THREAD} ~ {V2_MAX_THREAD}，启动{V2_MIN_THREAD}，结束{V2_current_thread}') print(f'调速记录共{len(V2_speed_adjust_records)}条') for rec in V2_speed_adjust_records[-10:]: print(f' {BLUE}-{rec}{RESET}') print(f'\n{CYAN}【四、风控冷却记录】{RESET}') if len(V2_cool_trigger_records) == 0: print(f' {GREEN}全程无风控{RESET}') else: for r in V2_cool_trigger_records[-10:]: print(f' {RED}-{r}{RESET}') print(f'\n{CYAN}【五、失败分类】{RESET}') if len(V2_fail_detail) == 0: print(f' {GREEN}无任何失败{RESET}') else: for k,v in V2_fail_detail.items(): rate = (v/V2_fail_total)*100 if V2_fail_total>0 else 0 print(f' {RED}{k} 次数{v} 占比{rate:.1f}%{RESET}') print(f'\n{CYAN}【六、运行机制说明】{RESET}') print('低速预热→稳定单次+3极速冲峰值→失败小幅降→风控解除自动爬坡提速') print('底层优化：批量12、2.2s超时、极低休眠、2400组伪装、单任务异常隔离') print(f'{RAINBOW[0]}+========================================================================================+{RESET}') print(f'{BOLD}{MAGENTA}V2极限极速稳定版运行完毕{RESET}\n') def v2_main(): print(f'\n{CYAN}{BOLD}━━━ 千戟 · 极限极速版 ━━━{RESET}') print(f'{GRAY}适用：追求极致速度 / 已熟练使用 / 可接受少量失败{RESET}\n') print(f'{CYAN}粘贴快手作品链接：{RESET}') input_link = input().strip() if not input_link: print(f'{RED}{CROSS}链接为空退出{RESET}') return print(f'{YELLOW}{ARROW}识别链接中{RESET}') ks_url, link_type = v2_extract_ks_link(input_link) if not ks_url: print(f'{RED}{CROSS}识别失败{link_type}{RESET}') pid_manual = input(f'{CYAN}手动输入作品ID：{RESET}').strip() if not pid_manual: print(f'{RED}{CROSS}无ID退出{RESET}') return work_id = pid_manual else: print(f'{GREEN}{CHECK}提取链接{ks_url}{RESET}') work_id = v2_parse_work_pid(ks_url) if not work_id: pid_manual = input(f'{CYAN}解析失败，请输入ID：{RESET}').strip() if not pid_manual: print(f'{RED}{CROSS}退出{RESET}') return work_id = pid_manual print(f'\n{CYAN}{ARROW}输入目标分享次数：{RESET}') try: target_total = int(input().strip()) if target_total <= 0: print(f'{RED}{CROSS}次数必须大于0{RESET}') return except ValueError: print(f'{RED}{CROSS}输入数字{RESET}') return print(f'\n{MAGENTA}V2任务确认{RESET}') print(f'作品ID：{work_id}') print(f'目标次数：{target_total}') print(f'并发区间 {V2_MIN_THREAD} ~ {V2_MAX_THREAD}') print(f'极速调速，失败缓慢降速，无断崖卡顿') print(f'Ctrl+C 随时终止') print(f'{GREEN}3秒后启动{RESET}') time.sleep(3) v2_run_core(work_id, target_total) # ============================================================================ # 版本3：智能自适应版（来自 千戟v3.py） # ============================================================================ V3_MAX_THREAD = 120 V3_START_THREAD = 10 V3_STEP_UP = 5 V3_STEP_DOWN = 2 V3_SUCC_RATE_HIGH = 0.75 V3_SUCC_RATE_LOW = 0.60 V3_ADJUST_INTERVAL = 10 V3_REQUEST_TIMEOUT = 1.5 V3_COOL_SHORT = 0.5 V3_COOL_MID = 1.5 V3_COOL_LONG = 4.0 V3_COOL_CRASH = 6.0 V3_LOOP_DELAY = 0.001 V3_BATCH_SIZE = 12 V3_API_URL = 'https://www.kuaishou.com/rest/zt/share/w/any' V3_FIX_PARAMS = { 'kpn': 'KUAISHOU_VISION', 'kpf': 'PC_WEB', 'subBiz': 'SINGLE_ROW_WEB', 'sdkVersion': '1.1.2.4.0', 'shareChannel': 'WECHAT', 'shareMethod': 'LINK' } V3_BRANDS = ['Apple','Samsung','Xiaomi','Huawei','OPPO','vivo','OnePlus','Google','Meizu','realme','iQOO','Honor','Nubia','BlackShark','ROG'] V3_IPHONE_MODELS = ['iPhone15,2','iPhone14,3','iPhone13,4','iPhone16,1','iPhone15,3','iPhone14,6','iPhone13,1','iPhone16,2','iPhone16,3'] V3_ANDROID_MODELS = { 'Samsung':['SM-S918B','SM-S911B','SM-G998B','SM-F946B','SM-A546E','SM-A346E','SM-S928B'], 'Xiaomi':['2211133C','M2101K7AG','2304FPN6DC','2203121C','24041696G','23127PN5BC','24072PN5AC'], 'Huawei':['LNA-AL00','RNA-AL00','ELS-AN00','VOG-AL10','TET-AN00','NOP-AN00','MNA-AL00'], 'OPPO':['CPH2357','CPH2325','PGEM10','PFZM10','PFFM20','PJC110','PGW110'], 'vivo':['V2144','V2133','V2304A','V2242A','V2409','V2319A','V2418A'], 'OnePlus':['LE2123','LE2113','NE2211','PHB110','KF110','NE2411'], 'Google':['Pixel 7 Pro','Pixel 7','Pixel 8','Pixel 8 Pro','Pixel 9','Pixel 9 Pro'], 'Meizu':['M1931','M1872','M2392','M2281','M2452'], 'realme':['RMX3700','RMX3610','RMX3850','RMX3900','RMX4010'], 'iQOO':['V2243A','I2211','V2301A','V2401A','V2412A'], 'Honor':['ELZ-AN00','HPB-AN00','BMH-AN10','FNE-AN00','GIA-AN00'], 'Nubia':['NX729J','NX709J','NX699J','NX732J'], 'BlackShark':['SKW-A0','KSR-A0'], 'ROG':['AI2201','AS_AI2301'] } V3_PC_MODELS = { 'Dell':['XPS 15','Latitude 7420','G15 5530','Inspiron 14','OptiPlex 7010','XPS 17'], 'HP':['Spectre x360','Envy 15','Victus 16','Pavilion 14','ZBook','暗影精灵10'], 'Lenovo':['ThinkPad X1','Yoga 9i','拯救者Y9000P','小新Pro16','ThinkBook 14','R9000X'], 'Apple':['MacBookPro18,3','MacBookAir10,1','MacBookPro19,1','iMac21,1','MacMini9,1','MacBookPro20,1'], 'Asus':['ZenBook Pro','ROG幻16','无畏Pro14','天选4','ProArt'] } V3_IOS_VERSIONS = ['17.1','17.0','16.7','18.0','17.4','16.6','18.1','18.2'] V3_ANDROID_VERSIONS = ['14','13','12','11','10','9'] V3_WINDOWS_VERSIONS = ['10.0','11.0'] V3_MACOS_VERSIONS = ['10_15_7','11_0','12_0','13_0','14_0','15_0'] V3_CHROME_VER_RANGE = list(range(120, 140)) V3_ACCEPT_VALUES = ['application/json, text/plain, */*'] V3_ACCEPT_LANGUAGES = ['zh-CN,zh;q=0.9','zh-CN,zh;q=0.95','zh-CN,zh-TW;q=0.9','zh-CN'] V3_ACCEPT_ENCODINGS = ['gzip, deflate, br'] V3_SEC_CH_UA_VARIANTS = [lambda: f'"Google Chrome";v="{random.choice(V3_CHROME_VER_RANGE)}", "Chromium";v="{random.choice(V3_CHROME_VER_RANGE)}", "Not=A?Brand";v="99"'] V3_PC_RESOLUTIONS = ['1920x1080','1366x768','2560x1440','3840x2160','1600x900','1280x720','2880x1800'] V3_MOBILE_RESOLUTIONS = ['390x844','375x812','414x896','430x932','360x800','412x915','480x1080'] V3_COLOR_DEPTHS = [24,30,32] V3_PIXEL_RATIOS = [1.0,2.0,2.5,3.0,3.5] V3_FAKE_IP_PREFIXES = [ '221.219','113.132','119.147','124.95','183.247','39.144','61.184','106.121','122.193','115.48','116.224', '112.94','111.175','27.188','180.113','223.72','120.92','183.191','182.130','218.201', '110.242','125.39','123.125','119.4','114.246','118.248','42.231' ] V3_RAM_LIST = [4,6,8,12,16,32,64] V3_CPU_CORE_LIST = [4,6,8,10,12,16,20] def v3_generate_chrome_ua(): os_choices = [ f'Windows NT {random.choice(V3_WINDOWS_VERSIONS)}; Win64; x64', f'Macintosh; Intel Mac OS X {random.choice(V3_MACOS_VERSIONS)}', 'X11; Linux x86_64' ] os_sys = random.choice(os_choices) ver = random.choice(V3_CHROME_VER_RANGE) return f'Mozilla/5.0 ({os_sys}) AppleWebKit/537.36 Chrome/{ver}.0.0.0 Safari/537.36' def v3_generate_mobile_ua(): brand = random.choice(V3_BRANDS) if brand == 'Apple': ios_ver = random.choice(V3_IOS_VERSIONS).replace('.','_') return f'Mozilla/5.0 (iPhone; CPU iPhone OS {ios_ver} like Mac OS X) AppleWebKit/605.1.15 Mobile/15E148' else: model = random.choice(list(V3_ANDROID_MODELS.keys())) dev_model = random.choice(V3_ANDROID_MODELS[model]) android_ver = random.choice(V3_ANDROID_VERSIONS) chrome_ver = random.choice(V3_CHROME_VER_RANGE) return f'Mozilla/5.0 (Linux; Android {android_ver}; {dev_model}) AppleWebKit/537.36 Chrome/{chrome_ver}.0.0.0 Mobile Safari/537.36' V3_USER_AGENT_GENERATORS = [v3_generate_chrome_ua, v3_generate_mobile_ua] def v3_get_fake_ip(): prefix = random.choice(V3_FAKE_IP_PREFIXES) return f'{prefix}.{random.randint(1,255)}.{random.randint(1,255)}' def v3_get_ip_operator(ip): seg = ip.split('.')[0] telecom = ['221','113','119','124','183','39','61','106','122','115','116'] mobile = ['112','111','27','180','223','120','183','182','218'] unicom = ['110','125','123','119','114','118','42'] if seg in telecom: return "电信" elif seg in mobile: return "移动" elif seg in unicom: return "联通" return "未知" def v3_generate_device_info(ua): is_mobile = 'Mobile' in ua or 'iPhone' in ua is_ios = 'iPhone' in ua is_android = 'Android' in ua is_windows = 'Windows' in ua is_mac = 'Macintosh' in ua if is_ios: return {'is_mobile':True,'brand':'Apple','model':random.choice(V3_IPHONE_MODELS),'os_name':'iOS','os_version':random.choice(V3_IOS_VERSIONS), 'ram':random.choice(V3_RAM_LIST),'cpu_cores':random.choice(V3_CPU_CORE_LIST),'pixel_ratio':random.choice(V3_PIXEL_RATIOS),'color_depth':random.choice(V3_COLOR_DEPTHS)} elif is_android: brand=random.choice(list(V3_ANDROID_MODELS.keys())) return {'is_mobile':True,'brand':brand,'model':random.choice(V3_ANDROID_MODELS[brand]),'os_name':'Android','os_version':random.choice(V3_ANDROID_VERSIONS), 'ram':random.choice(V3_RAM_LIST),'cpu_cores':random.choice(V3_CPU_CORE_LIST),'pixel_ratio':random.choice(V3_PIXEL_RATIOS),'color_depth':random.choice(V3_COLOR_DEPTHS)} elif is_windows: brand=random.choice(['Dell','HP','Lenovo','Asus']) return {'is_mobile':False,'brand':brand,'model':random.choice(V3_PC_MODELS[brand]),'os_name':'Windows','os_version':random.choice(V3_WINDOWS_VERSIONS), 'ram':random.choice(V3_RAM_LIST),'cpu_cores':random.choice(V3_CPU_CORE_LIST),'pixel_ratio':random.choice(V3_PIXEL_RATIOS),'color_depth':random.choice(V3_COLOR_DEPTHS)} elif is_mac: return {'is_mobile':False,'brand':'Apple','model':random.choice(V3_PC_MODELS['Apple']),'os_name':'macOS','os_version':random.choice(V3_MACOS_VERSIONS).replace('_','.'), 'ram':random.choice(V3_RAM_LIST),'cpu_cores':random.choice(V3_CPU_CORE_LIST),'pixel_ratio':random.choice(V3_PIXEL_RATIOS),'color_depth':random.choice(V3_COLOR_DEPTHS)} else: return {'is_mobile':False,'brand':'Dell','model':'Desktop','os_name':'Linux','os_version':'Unknown', 'ram':random.choice(V3_RAM_LIST),'cpu_cores':random.choice(V3_CPU_CORE_LIST),'pixel_ratio':random.choice(V3_PIXEL_RATIOS),'color_depth':random.choice(V3_COLOR_DEPTHS)} def v3_generate_full_cookie(): items = [ f'did=web_{secrets.token_hex(16)}', f'clientid={secrets.token_hex(8)}', f'sid={secrets.token_hex(32)}', f'uid={secrets.token_hex(16)}', f'userId={random.randint(1000000,999999999)}', 'kpn=KUAISHOU_VISION', f'_ga=GA1.2.{random.randint(100000000,999999999)}.{int(time.time())}', f'_gid=GA1.2.{random.randint(100000000,999999999)}.{int(time.time())}', f'_gat_gtag_UA_{random.randint(100000,999999)}=1', f'visitor_id={secrets.token_hex(20)}', f'cache_token={secrets.token_hex(12)}', f'user_trace={secrets.token_hex(24)}', f'page_mark={secrets.token_hex(10)}' ] random.shuffle(items) return '; '.join(items[:random.randint(10,15)]) def v3_generate_headers_full(): ua = random.choice(V3_USER_AGENT_GENERATORS)() device = v3_generate_device_info(ua) is_mobile = device['is_mobile'] fake_ip = v3_get_fake_ip() headers = { 'User-Agent': ua, 'Accept': random.choice(V3_ACCEPT_VALUES), 'Accept-Language': random.choice(V3_ACCEPT_LANGUAGES), 'Accept-Encoding': random.choice(V3_ACCEPT_ENCODINGS), 'Referer': 'https://www.kuaishou.com/', 'Origin': 'https://www.kuaishou.com', 'Connection': 'keep-alive', 'Cache-Control': 'no-cache', 'X-Forwarded-For': fake_ip, 'X-Real-IP': fake_ip, 'Sec-Ch-Ua': random.choice(V3_SEC_CH_UA_VARIANTS)(), 'Sec-Ch-Ua-Mobile': '?1' if is_mobile else '?0', 'Sec-Ch-Ua-Platform': f'"{device["brand"]}"', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'Device-Memory': str(device['ram']), 'Hardware-Concurrency': str(device['cpu_cores']), 'Cookie': v3_generate_full_cookie() } return headers, fake_ip, device V3_HEADER_POOL, V3_DEVICE_POOL, V3_IP_POOL, V3_OPERATOR_POOL = [], [], [], [] print(f'{CYAN}[V3初始化] 生成5000组伪装...{RESET}') for _ in range(5000): try: h, ip, dev = v3_generate_headers_full() op = v3_get_ip_operator(ip) V3_HEADER_POOL.append(h); V3_DEVICE_POOL.append(dev); V3_IP_POOL.append(ip); V3_OPERATOR_POOL.append(op) except: continue print(f'{GREEN}[V3就绪] 伪装池大小 {len(V3_HEADER_POOL)}{RESET}') V3_pool_idx = 0 V3_pool_lock = threading.Lock() def v3_get_cached_full_info(): global V3_pool_idx with V3_pool_lock: idx = V3_pool_idx V3_pool_idx = (V3_pool_idx + 1) % len(V3_HEADER_POOL) return V3_HEADER_POOL[idx], V3_IP_POOL[idx], V3_DEVICE_POOL[idx], V3_OPERATOR_POOL[idx] def v3_extract_ks_link(raw): m = re.search(r'https://v\.kuaishou\.com/[A-Za-z0-9]+', raw) or re.search(r'https://www\.kuaishou\.com/[^\s]+', raw) if m: return m.group(0), "链接" return None, "未识别" def v3_parse_work_pid(url): print(f'{CYAN}{ARROW} V3解析 {url}{RESET}') try: hd,_,_,_ = v3_get_cached_full_info() resp = requests.get(url, headers=hd, allow_redirects=True, timeout=8) final = resp.url m = re.search(r'/photo/([A-Za-z0-9]+)', final) if m: pid = m.group(1); print(f'{GREEN}{CHECK} V3 ID: {pid}{RESET}'); return pid qs = parse_qs(urlparse(final).query) if 'photoId' in qs: pid = qs['photoId'][0]; print(f'{GREEN}{CHECK} V3 ID: {pid}{RESET}'); return pid short = re.search(r'v\.kuaishou\.com/([A-Za-z0-9]+)', url) if short: code = short.group(1) r2 = requests.get(f'https://v.kuaishou.com/{code}', headers=hd, allow_redirects=True, timeout=6) m2 = re.search(r'/photo/([A-Za-z0-9]+)', r2.url) if m2: pid = m2.group(1); print(f'{GREEN}{CHECK} V3 ID: {pid}{RESET}'); return pid print(f'{RED}{CROSS} V3解析失败{RESET}') return None except Exception as e: print(f'{RED}{CROSS} V3异常: {e}{RESET}') return None def v3_build_sign(pid): ts = int(time.time()*1000) nonce = secrets.token_hex(16) raw = f'shareObjectId={pid}&kpn=KUAISHOU_VISION&nonce={nonce}×tamp={ts}' sign = hashlib.md5(raw.encode()).hexdigest()[:16] return {'_nonce':nonce,'_timestamp':ts,'_signature':sign} V3_lock = threading.Lock() V3_success_count = 0 V3_total_sent = 0 V3_fail_total = 0 V3_target_task = 0 V3_completed = 0 V3_success_delay = deque(maxlen=200) V3_fail_detail = defaultdict(int) V3_current_thread = V3_START_THREAD V3_active_threads = [] V3_thread_lock = threading.Lock() V3_consecutive_fail = 0 V3_global_cool_until = 0 V3_cool_records = [] V3_adjust_records = [] V3_peak_speed = 0.0 V3_min_speed = 9999.0 V3_start_time = 0 V3_adjust_counter = 0 def v3_single_share(pid, session, task_id): global V3_success_count, V3_total_sent, V3_fail_total, V3_consecutive_fail, V3_global_cool_until, V3_completed global V3_fail_detail, V3_cool_records, V3_success_delay try: now = time.time() if now < V3_global_cool_until: time.sleep(V3_global_cool_until - now + 0.02) hd, fake_ip, dev, op = v3_get_cached_full_info() payload = dict(V3_FIX_PARAMS) payload['shareObjectId'] = pid payload.update(v3_build_sign(pid)) req_start = time.time() try: res = session.post(V3_API_URL, json=payload, headers=hd, timeout=V3_REQUEST_TIMEOUT) cost = round(time.time()-req_start, 3) if res.status_code == 200: data = res.json() if data.get('result') == 1: with V3_lock: V3_success_count += 1 V3_success_delay.append(cost) V3_consecutive_fail = max(0, V3_consecutive_fail-2) if V3_success_count % 50 == 0: print(f'{GREEN}[V3成功] {V3_success_count}{RESET}') else: err = data.get('errorMsg','') with V3_lock: V3_fail_total +=1; V3_fail_detail[f'业务码{data.get("result")}'] +=1; V3_consecutive_fail +=1 if V3_fail_total % 20 == 0: print(f'{YELLOW}[V3失败] {err}{RESET}') time.sleep(V3_COOL_SHORT) elif res.status_code == 429: with V3_lock: V3_fail_total +=1; V3_fail_detail['429'] +=1; V3_consecutive_fail +=4 V3_global_cool_until = time.time() + V3_COOL_LONG V3_cool_records.append('429') print(f'{RED}[V3 429限流]{RESET}') time.sleep(V3_COOL_LONG) elif res.status_code == 403: with V3_lock: V3_fail_total +=1; V3_fail_detail['403'] +=1; V3_consecutive_fail +=5 V3_global_cool_until = time.time() + V3_COOL_CRASH V3_cool_records.append('403') print(f'{RED}[V3 403封禁]{RESET}') time.sleep(V3_COOL_CRASH) elif 500 <= res.status_code < 600: with V3_lock: V3_fail_total +=1; V3_fail_detail['5xx'] +=1; V3_consecutive_fail +=2 time.sleep(V3_COOL_MID) else: with V3_lock: V3_fail_total +=1; V3_fail_detail[f'HTTP{res.status_code}'] +=1; V3_consecutive_fail +=1 time.sleep(V3_COOL_SHORT) except requests.exceptions.Timeout: with V3_lock: V3_fail_total +=1; V3_fail_detail['超时'] +=1; V3_consecutive_fail +=1 except Exception as e: with V3_lock: V3_fail_total +=1; V3_fail_detail['异常'] +=1; V3_consecutive_fail +=1 finally: with V3_lock: V3_total_sent +=1 V3_completed +=1 except Exception: with V3_lock: V3_total_sent +=1 V3_completed +=1 V3_fail_total +=1 def v3_adjust_threads(): global V3_current_thread, V3_global_cool_until with V3_lock: sent = V3_total_sent succ = V3_success_count if sent < 20: return succ_rate = succ / sent fail_cnt = V3_consecutive_fail if fail_cnt >= 15: with V3_thread_lock: old = V3_current_thread V3_current_thread = max(2, V3_current_thread - 5) V3_global_cool_until = time.time() + V3_COOL_LONG log = f"V3连续失败{fail_cnt} → {V3_current_thread}" V3_cool_records.append(log); V3_adjust_records.append(log) print(f'{RED}{log}{RESET}') return if succ_rate < V3_SUCC_RATE_LOW: with V3_thread_lock: old = V3_current_thread new_t = max(2, V3_current_thread - V3_STEP_DOWN) if new_t != V3_current_thread: V3_current_thread = new_t log = f"V3成功率{succ_rate:.0%} ↓{V3_current_thread}" V3_adjust_records.append(log) print(f'{YELLOW}{log}{RESET}') elif succ_rate >= V3_SUCC_RATE_HIGH: with V3_thread_lock: old = V3_current_thread new_t = min(V3_MAX_THREAD, V3_current_thread + V3_STEP_UP) if new_t != V3_current_thread: V3_current_thread = new_t log = f"V3成功率{succ_rate:.0%} ↑{V3_current_thread}" V3_adjust_records.append(log) print(f'{GREEN}{log}{RESET}') else: with V3_thread_lock: if succ_rate > 0.68: new_t = min(V3_MAX_THREAD, V3_current_thread + 1) if new_t != V3_current_thread: V3_current_thread = new_t log = f"V3稳升+1 →{V3_current_thread}" V3_adjust_records.append(log) print(f'{BLUE}{log}{RESET}') elif succ_rate < 0.65: new_t = max(2, V3_current_thread - 1) if new_t != V3_current_thread: V3_current_thread = new_t log = f"V3稳降-1 →{V3_current_thread}" V3_adjust_records.append(log) print(f'{BLUE}{log}{RESET}') def v3_monitor(): spin=0 global V3_peak_speed, V3_min_speed while True: try: time.sleep(0.25) now = time.time() with V3_lock: succ=V3_success_count; sent=V3_total_sent; fail=V3_fail_total comp=V3_completed; target=V3_target_task cd=V3_global_cool_until run = now - V3_start_time if sent == 0: speed=0; rate=0; avg=0 else: speed = sent/run rate = succ/sent*100 avg = sum(V3_success_delay)/len(V3_success_delay) if V3_success_delay else 0 pct = (comp/target*100) if target else 0 if speed > V3_peak_speed: V3_peak_speed = speed if sent > 10 and speed < V3_min_speed: V3_min_speed = speed cd_text = f"冷却{round(cd-now,1)}s" if cd>now else "无冷却" bar = PROGRESS_FILL*int(34*pct/100) + PROGRESS_EMPTY*(34-int(34*pct/100)) sys.stdout.write( f"\r{WHITE}V3进度[{bar}] {pct:.1f}% {SPINNER[spin%len(SPINNER)]} " f"成功{succ} 失败{fail} 成功率{rate:.1f}% " f"速度{speed:.1f}/s 峰值{V3_peak_speed:.1f}/s " f"线程{V3_current_thread} {cd_text}{RESET}" ) sys.stdout.flush() spin+=1 except: continue def v3_run_core(pid, total): global V3_success_count, V3_total_sent, V3_fail_total, V3_consecutive_fail, V3_global_cool_until, V3_current_thread global V3_completed, V3_target_task, V3_start_time, V3_cool_records, V3_adjust_records, V3_fail_detail global V3_success_delay, V3_peak_speed, V3_min_speed, V3_adjust_counter V3_success_count = V3_total_sent = V3_fail_total = V3_consecutive_fail = V3_completed = 0 V3_global_cool_until = 0 V3_current_thread = V3_START_THREAD V3_target_task = total V3_start_time = time.time() V3_cool_records.clear(); V3_adjust_records.clear(); V3_fail_detail.clear(); V3_success_delay.clear() V3_peak_speed = 0.0; V3_min_speed = 9999.0 V3_adjust_counter = 0 print(f'\n{CYAN}[V3初始化] 建立 {V3_MAX_THREAD} 个长连接...{RESET}') session_pool = [] for _ in range(V3_MAX_THREAD): s = requests.Session() s.headers['Connection'] = 'keep-alive' s.keep_alive = True adapter = requests.adapters.HTTPAdapter(pool_connections=V3_MAX_THREAD, pool_maxsize=V3_MAX_THREAD*2, max_retries=0) s.mount("https://", adapter) session_pool.append(s) print(f'{GREEN}[V3就绪] 连接池与伪装池已预热{RESET}') monitor_t = threading.Thread(target=v3_monitor, daemon=True) monitor_t.start() print(f'\n{MAGENTA}{BOLD}V3自适应极限模式启动 (起始{V3_START_THREAD}，上限{V3_MAX_THREAD}){RESET}\n') V3_active_threads.clear() task_id = 1 while V3_completed < total: now = time.time() if now < V3_global_cool_until: time.sleep(V3_LOOP_DELAY) continue with V3_thread_lock: V3_active_threads[:] = [t for t in V3_active_threads if t.is_alive()] running = len(V3_active_threads) idle = V3_current_thread - running batch = min(idle, total - V3_completed, V3_BATCH_SIZE) for _ in range(batch): sess_idx = task_id % len(session_pool) t = threading.Thread(target=v3_single_share, args=(pid, session_pool[sess_idx], task_id), daemon=True) t.start() V3_active_threads.append(t) task_id += 1 V3_adjust_counter += batch if V3_adjust_counter >= V3_ADJUST_INTERVAL: V3_adjust_counter = 0 v3_adjust_threads() if batch == 0 and V3_completed < total: time.sleep(V3_LOOP_DELAY) print(f'\n{YELLOW}[V3收尾] 等待剩余线程...{RESET}') for t in V3_active_threads: try: t.join(timeout=2) except: pass time.sleep(0.5) run_time = round(time.time()-V3_start_time, 2) avg_speed = total/run_time if run_time else 0 succ_rate = (V3_success_count/V3_total_sent*100) if V3_total_sent else 0 print(f'\n\n{RAINBOW[0]}+================================================================+{RESET}') print(f'{RAINBOW[0]}|{RESET}{BOLD}V3自适应极限版 执行报告{RAINBOW[0]}|') print(f'{RAINBOW[0]}+================================================================+{RESET}') print(f'目标：{total} 请求：{V3_total_sent} 成功：{GREEN}{V3_success_count}{RESET} 失败：{RED}{V3_fail_total}{RESET}') print(f'成功率：{GREEN if succ_rate>60 else YELLOW}{succ_rate:.2f}%{RESET}') print(f'总耗时：{run_time}s 平均速度：{BOLD}{avg_speed:.2f}{RESET} 次/秒') print(f'峰值速度：{V3_peak_speed:.2f} 最低速度：{V3_min_speed:.2f}') avg_delay = sum(V3_success_delay)/len(V3_success_delay) if V3_success_delay else 0 print(f'平均延迟：{avg_delay:.3f}s') print(f'并发区间 {V3_START_THREAD}~{V3_MAX_THREAD} 结束并发 {V3_current_thread}') if V3_adjust_records: print('调速记录（最近5条）：') for r in V3_adjust_records[-5:]: print(f' {r}') if V3_fail_detail: print('失败分布：') for k,v in V3_fail_detail.items(): print(f' {k}: {v}') print(f'{RAINBOW[0]}+================================================================+{RESET}\n') def v3_main(): print(f'\n{CYAN}{BOLD}━━━ 千戟 · 智能自适应版 ━━━{RESET}') print(f'{GRAY}适用：高速稳定运行 / 优先成功率 / 长期批量执行{RESET}\n') print(f'{CYAN}{ARROW} 粘贴快手链接：{RESET}') link = input().strip() if not link: print(f'{RED}链接为空{RESET}'); return ks_url, _ = v3_extract_ks_link(link) if not ks_url: pid = input(f'{CYAN}手动输入作品ID：{RESET}').strip() if not pid: return work_id = pid else: work_id = v3_parse_work_pid(ks_url) if not work_id: pid = input(f'{CYAN}解析失败，输入ID：{RESET}').strip() if not pid: return work_id = pid print(f'{CYAN}{ARROW} 目标次数：{RESET}') try: total = int(input().strip()) if total <= 0: return except: return print(f'\n{MAGENTA}确认：作品 {work_id}，次数 {total}，V3自适应极限模式{RESET}') print(f'{GREEN}3秒后启动...{RESET}') time.sleep(3) v3_run_core(work_id, total) # ============================================================================ # 主启动器 # ============================================================================ def main(): while True: clear_screen() show_menu() choice = input(f'{CYAN}请输入选择 (0-3)：{RESET}').strip() if choice == '0': print(f'\n{GREEN}👋 感谢使用千戟快手刷分享工具！{RESET}') print(f'{GRAY}官方千戟 快手号：tyknb888{RESET}') break elif choice in ('1', '2', '3'): ver_id = int(choice) # 显示详细说明 detail_choice = show_version_detail(ver_id) if detail_choice == '0': continue # 启动对应版本 run_version(ver_id) input(f'\n{YELLOW}按 Enter 键返回主菜单...{RESET}') else: print(f'{RED}❌ 无效输入，请输入 0~3{RESET}') time.sleep(1) if __name__ == '__main__': try: main() except KeyboardInterrupt: print(f'\n{YELLOW}用户中断，退出程序。{RESET}') except Exception as e: print(f'{RED}发生未知错误：{e}{RESET}') import traceback traceback.print_exc()