import json import time import re import requests import urllib.parse import struct from playwright.sync_api import sync_playwright, Page # from playwright_stealth.stealth import Stealth from bs4 import BeautifulSoup from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime import csv from selenium.webdriver import Chrome, ChromeOptions from selenium.webdriver.common.action_chains import ActionChains class MtopSignatureGenerator: def __init__(self, token, app_key="12574478", api_name="mtop.relationrecommend.WirelessRecommend.recommend", api_version="2.0"): self.token = token self.app_key = app_key self.api_name = api_name self.api_version = api_version def _md5(self, input_string): """ MD5 implementation that matches the JavaScript version in the HTML """ # Convert string to UTF-8 bytes (equivalent to str2rstrUTF8 in JS) input_bytes = input_string.encode('utf-8') # Initialize variables (same as JavaScript) a = 1732584193 b = -271733879 c = -1732584194 d = 271733878 # Process the message in successive 512-bit chunks original_length = len(input_bytes) input_bytes += b'\x80' # Pad with zeros until message length in bits ≡ 448 (mod 512) while (len(input_bytes) * 8) % 512 != 448: input_bytes += b'\x00' # Append original length in bits mod 2^64 (little-endian) input_bytes += struct.pack('> (32 - cnt))) & 0xFFFFFFFF def _ff(self, a, b, c, d, x, s, t): """FF transformation function""" a = self._safe_add(a, self._safe_add(self._safe_add(self._f(b, c, d), x), t)) return self._safe_add(self._bit_rotate_left(a, s), b) def _gg(self, a, b, c, d, x, s, t): """GG transformation function""" a = self._safe_add(a, self._safe_add(self._safe_add(self._g(b, c, d), x), t)) return self._safe_add(self._bit_rotate_left(a, s), b) def _hh(self, a, b, c, d, x, s, t): """HH transformation function""" a = self._safe_add(a, self._safe_add(self._safe_add(self._h(b, c, d), x), t)) return self._safe_add(self._bit_rotate_left(a, s), b) def _ii(self, a, b, c, d, x, s, t): """II transformation function""" a = self._safe_add(a, self._safe_add(self._safe_add(self._i(b, c, d), x), t)) return self._safe_add(self._bit_rotate_left(a, s), b) def _f(self, x, y, z): """F auxiliary function""" return (x & y) | ((~x) & z) def _g(self, x, y, z): """G auxiliary function""" return (x & z) | (y & (~z)) def _h(self, x, y, z): """H auxiliary function""" return x ^ y ^ z def _i(self, x, y, z): """I auxiliary function""" return y ^ (x | (~z)) def generate_signature(self, data): """Generate MTOP signature""" timestamp = str(int(time.time() * 1000)) sign_content = f"{self.token}&{timestamp}&{self.app_key}&{json.dumps(data, ensure_ascii=False, separators=(',', ':'))}" return self._md5(sign_content), timestamp def generate_url(self, data): """Generate the full signed URL""" signature, timestamp = self.generate_signature(data) # Encode data for URL encoded_data = urllib.parse.quote(json.dumps(data, ensure_ascii=False, separators=(',', ':'))) # Construct the URL url = f"https://h5api.m.alibaba.com/h5/{self.api_name}/{self.api_version}/" # url = f"https://h5api.m.1688.com/h5/{self.api_name}/{self.api_version}/" params = { "jsv": "2.7.4", "appKey": self.app_key, "t": timestamp, "sign": signature, "api": self.api_name, "v": self.api_version, "dataType": "json", "type": "originaljson", "data": encoded_data } # Build query string query_string = "&".join([f"{k}={v}" for k, v in params.items()]) return f"{url}?{query_string}", signature, timestamp class NetworkMonitor: def __init__(self): self.captured_requests = [] self.captured_responses = [] def capture_network_activity(self, route, request): print("Trying to capture Request",request.url) """Capture all network requests""" request_data = { 'timestamp': datetime.now().isoformat(), 'url': request.url, 'method': request.method, 'headers': dict(request.headers), 'post_data': request.post_data, 'resource_type': request.resource_type } self.captured_requests.append(request_data) route.continue_() def capture_api_responses(self, response): """Capture API responses, especially captcha-related ones""" if any(keyword in response.url for keyword in ['captcha', 'verify', 'validate', 'security', 'punish', 'mtop']): try: response_data = { 'timestamp': datetime.now().isoformat(), 'url': response.url, 'status': response.status, 'headers': dict(response.headers), 'request_headers': dict(response.request.headers) if response.request else {}, 'body': None, 'json': None } # Try to get response body as text try: response_data['body'] = response.text() except: pass # Try to parse as JSON try: response_data['json'] = response.json() except: pass self.captured_responses.append(response_data) print(f"🔍 Captured API: {response.url} - Status: {response.status}") except Exception as e: print(f"❌ Error capturing response: {e}") def save_to_files(self): """Save captured data to files""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Save requests with open(f'temp/req/network_requests_{timestamp}.json', 'w', encoding='utf-8') as f: json.dump(self.captured_requests, f, indent=2, ensure_ascii=False) # Save responses with open(f'temp/resp/api_responses_{timestamp}.json', 'w', encoding='utf-8') as f: json.dump(self.captured_responses, f, indent=2, ensure_ascii=False) # Save to CSV for easy analysis self.save_to_csv(timestamp) print(f"💾 Saved {len(self.captured_requests)} requests and {len(self.captured_responses)} responses") def save_to_csv(self, timestamp): """Save important data to CSV""" csv_file = f'captcha_api_calls_{timestamp}.csv' with open(csv_file, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(['Timestamp', 'URL', 'Method', 'Status', 'API_Type', 'Has_Body']) for req in self.captured_requests: api_type = 'Unknown' if 'captcha' in req['url']: api_type = 'Captcha' elif 'verify' in req['url']: api_type = 'Verification' elif 'validate' in req['url']: api_type = 'Validation' writer.writerow([ req['timestamp'], req['url'], req['method'], 'N/A', # Status not available in requests api_type, 'Yes' if req['post_data'] else 'No' ]) for resp in self.captured_responses: api_type = 'Unknown' url = resp['url'] if 'captcha' in url: api_type = 'Captcha' elif 'verify' in url: api_type = 'Verification' elif 'validate' in url: api_type = 'Validation' elif 'mtop' in url: api_type = 'MTOP_API' elif 'punish' in url: api_type = 'Punishment' writer.writerow([ resp['timestamp'], url, 'N/A', # Method not directly in response resp['status'], api_type, 'Yes' if resp['body'] else 'No' ]) def scrape_proxynova_proxies(): """Scrape proxies from ProxyNova using Playwright""" proxies = [] with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context( user_agent=get_random_user_agent(), viewport={"width": 1920, "height": 1080} ) page = context.new_page() try: print("Navigating to ProxyNova...") page.goto("https://www.proxynova.com/proxy-server-list/", timeout=60000) # Wait for the proxy table to load page.wait_for_selector('#tbl_proxy_list', timeout=30000) # Get the page content content = page.content() soup = BeautifulSoup(content, 'html.parser') proxy_table = soup.find('table', {'id': 'tbl_proxy_list'}) if not proxy_table: print("Proxy table not found on ProxyNova") return [] rows = proxy_table.find_all('tr')[1:101] # First 100 rows for row in rows: try: cells = row.find_all('td') if len(cells) >= 2: # Extract IP address (handling JavaScript obfuscation) ip_script = cells[0].find('script') if ip_script and ip_script.string: ip_text = ip_script.string ip_match = re.search(r"document\.write\('([^']+)'\.substr\((\d+)\) \+ '([^']+)'\)", ip_text) if ip_match: part1 = ip_match.group(1) start_idx = int(ip_match.group(2)) part2 = ip_match.group(3) ip = part1[start_idx:] + part2 else: # Alternative pattern ip_match = re.search(r"document\.write\('([^']+)' \+ '([^']+)'\)", ip_text) if ip_match: ip = ip_match.group(1) + ip_match.group(2) else: continue else: ip_link = cells[0].find('a') ip = ip_link.text.strip() if ip_link else cells[0].text.strip() # Clean up IP ip = re.sub(r'[^0-9.]', '', ip) # Extract port port = cells[1].text.strip() if ip and port and re.match(r'^\d+\.\d+\.\d+\.\d+$', ip) and re.match(r'^\d+$', port): proxies.append(f"http://{ip}:{port}") except Exception as e: print(f"Error parsing proxy row: {e}") continue print(f"Scraped {len(proxies)} proxies from ProxyNova") except Exception as e: print(f"Error scraping ProxyNova: {e}") finally: browser.close() return proxies # Function to validate proxies def validate_proxy(proxy, timeout=10): """Test if a proxy is working""" try: # Extract protocol, ip and port from proxy string match = re.match(r'(\w+)://([^:]+):(\d+)', proxy) if not match: return False protocol, ip, port = match.groups() test_url = "http://httpbin.org/ip" response = requests.get(test_url, proxies={ "http": f"{protocol}://{ip}:{port}", "https": f"{protocol}://{ip}:{port}" }, timeout=timeout) if response.status_code == 200: print(f"✅ Working proxy: {proxy}") return True except Exception as e: pass print(f"❌ Dead proxy: {proxy}") return False # Function to get validated proxies def get_validated_proxies(): """Get and validate proxies from ProxyNova""" proxies = [ {"ip": "160.16.105.145", "port": 8080, "country": "Japan"}, {"ip": "8.210.17.35", "port": 8445, "country": "Hong Kong"}, {"ip": "47.76.144.139", "port": 9080, "country": "Hong Kong"}, {"ip": "47.250.51.110", "port": 5007, "country": "Malaysia"}, {"ip": "8.220.141.8", "port": 4145, "country": "Philippines"}, {"ip": "39.102.213.187", "port": 5060, "country": "China"}, {"ip": "120.26.52.35", "port": 3128, "country": "China"}, {"ip": "182.155.254.159", "port": 80, "country": "Taiwan"}, {"ip": "120.237.232.172", "port": 8060, "country": "China"}, {"ip": "8.220.204.92", "port": 12000, "country": "Korea"}, {"ip": "8.220.136.174", "port": 8081, "country": "Philippines"}, {"ip": "8.213.128.90", "port": 3, "country": "Singapore"}, {"ip": "8.213.128.9", "port": 8181, "country": "Korea"}, {"ip": "8.213.128.90", "port": 8081, "country": "Korea"} ] raw_proxies = proxies if not raw_proxies: print("Failed to scrape proxies from ProxyNova") return [] print(f"Testing {len(raw_proxies)} proxies...") valid_proxies = [] # Test proxies in parallel for speed with ThreadPoolExecutor(max_workers=20) as executor: future_to_proxy = {executor.submit(validate_proxy, proxy): proxy for proxy in raw_proxies} for future in as_completed(future_to_proxy): proxy = future_to_proxy[future] try: if future.result(): valid_proxies.append(proxy) except Exception as e: print(f"Error testing proxy {proxy}: {e}") print(f"Found {len(valid_proxies)} valid proxies") return valid_proxies # Function to handle 1688 captcha API calls def handle_1688_captcha_api(context, captcha_data): """Handle 1688 captcha API calls based on the captured curl commands""" # Extract parameters from the captcha data slidedata = captcha_data.get("slidedata", "") x5secdata = captcha_data.get("x5secdata", "") uuid = captcha_data.get("uuid", "") # Make the slide API call slide_url = f"https://h5api.m.1688.com/h5/mtop.relationrecommend.WirelessRecommend.recommend/2.0/_____tmd_____/slide" slide_params = { "slidedata": slidedata, "x5secdata": x5secdata, "ppt": "0", "_rand": f"{random.random()}", "landscape": "1", "ts": str(int(time.time() * 1000)), "v": "043772012545879013" } # Make the report API call report_url = f"https://h5api.m.1688.com/h5/mtop.relationrecommend.WirelessRecommend.recommend/2.0/_____tmd_____/report" report_params = { "x5secdata": x5secdata, "type": "verifyFail", "msg": "ZhRBp", "uuid": uuid, "v": "07735536319465004" } # Execute the API calls using Playwright page = context.new_page() try: # Set headers similar to the curl command headers = { 'accept': '*/*', 'accept-language': 'zh-CN', 'priority': 'u=1, i', 'sec-ch-ua': '"Chromium";v="139", "Not;A=Brand";v="99"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"', 'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors', 'sec-fetch-site': 'same-origin', 'user-agent': get_random_user_agent() } # Make the slide API call slide_response = page.goto( f"{slide_url}?{urllib.parse.urlencode(slide_params)}", wait_until="networkidle", headers=headers ) if slide_response and slide_response.status == 200: slide_result = slide_response.json() print("Slide API response:", slide_result) # If slide was successful, make the report API call if slide_result.get("success", False): report_response = page.goto( f"{report_url}?{urllib.parse.urlencode(report_params)}", wait_until="networkidle", headers=headers ) if report_response and report_response.status == 200: report_result = report_response.json() print("Report API response:", report_result) return report_result.get("success", False) return False except Exception as e: print(f"Error handling captcha API: {e}") return False finally: page.close() # Function to solve slider captcha for 1688 def solve_1688_slider_captcha(page): """Solve 1688 slider captcha by simulating the API calls""" print("🔍 Attempting to solve 1688 slider captcha...") try: # Wait for captcha iframe to load page.wait_for_timeout(3000) # Check if we're on a captcha page if "punish" in page.url or "captcha" in page.url: print("🛡️ Captcha page detected") # Extract captcha data from the page captcha_data = page.evaluate(""" () => { // Try to extract captcha data from the page const urlParams = new URLSearchParams(window.location.search); const x5secdata = urlParams.get('x5secdata') || ''; const x5step = urlParams.get('x5step') || ''; // Look for captcha data in the page content let slidedata = ''; let uuid = ''; // Try to find captcha data in script tags const scripts = document.querySelectorAll('script'); for (const script of scripts) { const text = script.textContent; if (text.includes('slidedata')) { const match = text.match(/slidedata["']?\\s*[:=]\\s*["']([^"']+)["']/); if (match) slidedata = match[1]; const uuidMatch = text.match(/uuid["']?\\s*[:=]\\s*["']([^"']+)["']/); if (uuidMatch) uuid = uuidMatch[1]; } } return {slidedata, x5secdata, uuid, x5step}; } """) print(f"Captcha data extracted: {captcha_data}") # If we have the required data, make the API calls if captcha_data.get('slidedata') and captcha_data.get('x5secdata'): success = handle_1688_captcha_api(page.context, captcha_data) if success: print("✅ Captcha solved via API") return True # Fallback to manual slider solving return solve_slider_captcha_advanced(page) return False except Exception as e: print(f"❌ Error solving captcha: {e}") return False def solve_slider_captcha_advanced(page: Page): """Advanced slider captcha solving with extremely human-like behavior""" print("🔍 Attempting to solve slider captcha with advanced simulation...") time.sleep(3) # Wait for captcha to fully load try: # Wait for captcha to fully load page.wait_for_selector(".nc-container", timeout=15000) print("✅ Captcha container found") # Take screenshot for debugging page.screenshot(path="captcha_before.png") # Get slider elements with different selectors slider_selectors = [ ".nc_scale", ".slider", ".btn_slide", ".nc_iconfont.btn_slide", "[class*='slide']", "[class*='btn']" ] slider_track = None slider_handle = None for selector in slider_selectors: try: if not slider_track: slider_track = page.query_selector(selector) if not slider_handle: slider_handle = page.query_selector(f"{selector}.btn_slide, {selector}[class*='btn']") except: pass if not slider_track or not slider_handle: print("❌ Could not find slider elements") return False # Get bounding boxes track_box = slider_track.bounding_box() handle_box = slider_handle.bounding_box() if not track_box or not handle_box: print("❌ Could not get element dimensions") return False print(f"📏 Track dimensions: {track_box}") print(f"📏 Handle dimensions: {handle_box}") # Calculate slide distance (usually full track width minus handle width) slide_distance = track_box['width'] - handle_box['width'] - 5 # 5px margin # Start position (center of handle) start_x = handle_box['x'] + handle_box['width'] / 2 start_y = handle_box['y'] + handle_box['height'] / 2 # End position end_x = start_x + slide_distance end_y = start_y # Same Y level print(f"📍 Moving from ({start_x:.1f}, {start_y:.1f}) to ({end_x:.1f}, {end_y:.1f})") # Approach the slider like a human would approach_slider_realistically(page, start_x, start_y) # Click and hold with slight pressure simulation page.mouse.down() time.sleep(random.uniform(0.15, 0.25)) # Slight pause after clicking # Move along a human-like path with natural variations human_like_slide(page, start_x, start_y, end_x, end_y, slide_distance) # Add a small overshoot and correction (like humans often do) overshoot_x = end_x + random.randint(3, 8) page.mouse.move(overshoot_x, end_y + random.randint(-1, 1)) time.sleep(random.uniform(0.05, 0.1)) # Correct back to the exact position page.mouse.move(end_x, end_y) time.sleep(random.uniform(0.1, 0.2)) # Release the mouse page.mouse.up() print("✅ Slider movement completed") # Wait for verification time.sleep(2) # Check for success return check_captcha_result(page) except Exception as e: print(f"❌ Error in captcha solving: {str(e)}") page.screenshot(path="captcha_error.png") return False def approach_slider_realistically(page: Page, target_x: float, target_y: float): """Approach the slider with human-like movement""" # Get current mouse position (if any) current_x, current_y = 0, 0 # Generate a curved approach path points = generate_approach_path(current_x, current_y, target_x, target_y) # Move along the approach path for x, y in points: page.mouse.move(x, y) time.sleep(random.uniform(0.01, 0.03)) # Small hesitation before clicking time.sleep(random.uniform(0.2, 0.4)) # Final adjustment to exact position page.mouse.move(target_x, target_y) time.sleep(random.uniform(0.1, 0.2)) def human_like_slide(page: Page, start_x: float, start_y: float, end_x: float, end_y: float, distance: float): """Simulate a human-like sliding motion with natural variations""" # Number of movement segments based on distance num_segments = max(20, min(60, int(distance / 5))) # Generate a path with natural human variations base_path = generate_human_like_path(start_x, start_y, end_x, end_y, num_segments) # Move along the path with variable speed for i, (x, y) in enumerate(base_path): # Vary the speed - slower at start and end, faster in middle if i < len(base_path) * 0.2 or i > len(base_path) * 0.8: delay = random.uniform(0.03, 0.05) # Slower at beginning and end else: delay = random.uniform(0.01, 0.02) # Faster in middle page.mouse.move(x, y) time.sleep(delay) # Occasionally add tiny pauses (like human hesitation) if random.random() < 0.05: # 5% chance of micro-pause time.sleep(random.uniform(0.05, 0.1)) def generate_approach_path(start_x: float, start_y: float, end_x: float, end_y: float, num_points: int = 15): """Generate a curved approach path""" points = [] # Create a gentle curve for the approach for i in range(num_points): t = i / (num_points - 1) # Basic linear interpolation x = start_x + (end_x - start_x) * t y = start_y + (end_y - start_y) * t # Add a gentle curve to the path curve_factor = math.sin(t * math.pi) * 10 x += curve_factor * (random.random() - 0.5) y += curve_factor * (random.random() - 0.5) # Add some random jitter (but less as we get closer) jitter_factor = (1 - t) * 3 x += random.uniform(-jitter_factor, jitter_factor) y += random.uniform(-jitter_factor, jitter_factor) points.append((x, y)) return points def generate_human_like_path(start_x: float, start_y: float, end_x: float, end_y: float, num_points: int): """Generate a path with human-like imperfections""" points = [] for i in range(num_points): t = i / (num_points - 1) # Basic linear interpolation x = start_x + (end_x - start_x) * t y = start_y + (end_y - start_y) * t # Add human-like imperfections: # 1. Slight vertical wandering y_wander = math.sin(t * math.pi * 2) * 1.5 y += y_wander # 2. Speed variations (slight hesitation at certain points) if 0.3 < t < 0.4 or 0.7 < t < 0.8: x += random.uniform(-1.5, 1.5) y += random.uniform(-1, 1) # 3. Small random jitter x += random.uniform(-0.5, 0.5) y += random.uniform(-0.3, 0.3) points.append((x, y)) return points def check_captcha_result(page: Page): """Check if captcha was solved successfully""" # Wait a moment for any success animation time.sleep(2) # Check for success indicators success_indicators = [ ".nc_iconfont.btn_ok", ".success", ".verify-success", "[class*='success']", "[class*='ok']" ] for selector in success_indicators: if page.query_selector(selector): print("✅ Captcha verification successful!") page.screenshot(path="captcha_success.png") return True # Check for error messages error_indicators = [ ".error", ".fail", ".nc_iconfont.btn_error", "[class*='error']", "[class*='fail']" ] for selector in error_indicators: element = page.query_selector(selector) if element: try: error_text = element.inner_text() print(f"❌ Captcha error: {error_text}") except: print("❌ Captcha error detected") return False print("⚠️ Captcha status unknown") page.screenshot(path="captcha_unknown.png") return False import math def human_like_mouse_move(page, start_x, start_y, end_x, end_y, duration=1.0): """Simulate human-like mouse movement with natural curves""" steps = int(duration * 100) points = [] # Create a curved path (bezier-like) for i in range(steps + 1): t = i / steps # Add some randomness to the path offset_x = random.randint(-3, 3) offset_y = random.randint(-2, 2) x = start_x + (end_x - start_x) * t + offset_x y = start_y + (end_y - start_y) * t + offset_y # Add some non-linear movement if 0.3 < t < 0.7: y += math.sin(t * math.pi) * 5 points.append((x, y)) return points import random def get_random_user_agent(): user_agents = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0" ] return random.choice(user_agents) from playwright.sync_api import sync_playwright import random import time def login_1688_realistic(username, password, page): """ Login to 1688.com with realistic human-like typing behavior Args: username (str): Your 1688.com username/ID password (str): Your 1688.com password Returns: bool: True if login successful, False otherwise """ def human_type(element, text, field_name): """Simulate human typing with random delays and occasional mistakes""" print(f"Typing {field_name}...") # Click on the field first element.click() time.sleep(random.uniform(0.3, 0.8)) # Type character by character with random delays for i, char in enumerate(text): # Small chance to make a typo and correct it if random.random() < 0.05 and i > 2: # 5% chance after first few chars wrong_char = random.choice('abcdefghijklmnopqrstuvwxyz1234567890') element.press(wrong_char) time.sleep(random.uniform(0.1, 0.3)) element.press('Backspace') time.sleep(random.uniform(0.1, 0.3)) element.press(char) # Vary typing speed delay = random.uniform(0.05, 0.2) # 50-200ms between keystrokes time.sleep(delay) # Final pause after finishing typing time.sleep(random.uniform(0.5, 1.2)) print(f"Finished typing {field_name}") try: time.sleep(random.uniform(2, 4)) # Wait for page to settle # Wait for login form to load print("Waiting for login form...") page.wait_for_selector('#fm-login-id', timeout=15000, state='visible') # Add some random mouse movement before typing username_field = page.query_selector('#fm-login-id') password_field = page.query_selector('#fm-login-password') # Move mouse to username field with human-like path username_box = username_field.bounding_box() page.mouse.move( username_box['x'] + random.randint(0, int(username_box['width'])), username_box['y'] + random.randint(0, int(username_box['height'])), steps=random.randint(5, 15) ) time.sleep(random.uniform(0.3, 0.8)) # Type username with human-like behavior human_type(username_field, username, "username") # Move to password field with realistic delay time.sleep(random.uniform(0.5, 1.5)) password_box = password_field.bounding_box() page.mouse.move( password_box['x'] + random.randint(0, int(password_box['width'])), password_box['y'] + random.randint(0, int(password_box['height'])), steps=random.randint(5, 15) ) time.sleep(random.uniform(0.3, 0.8)) # Type password with human-like behavior human_type(password_field, password, "password") # Small pause before clicking login time.sleep(random.uniform(0.8, 1.5)) # Move mouse to login button with slight randomness # Try multiple possible selectors for login button print("Submitting form...") print("Submitting form directly...") page.evaluate('document.querySelector("form").submit()') time.sleep(random.uniform(0.2, 0.5)) # Wait for login to complete with realistic timeout print("Waiting for login to complete...") try: # Wait for navigation or page change page.wait_for_load_state('domcontentloaded', timeout=20000) time.sleep(random.uniform(3, 6)) # Wait for post-login actions # Check multiple indicators of successful login success_indicators = [ page.url.startswith('https://www.1688.com/'), page.url.startswith('https://login.1688.com/') and 'success' in page.url.lower(), page.query_selector('.user-info') is not None, page.query_selector('.member-name') is not None ] if any(success_indicators): print("✅ Login successful!") # Take a screenshot for verification page.screenshot(path='login_success.png') return True else: # Check for error messages error_selectors = ['.fm-error', '.error-msg', '.message-error', '.ant-message-error'] for selector in error_selectors: error_element = page.query_selector(selector) if error_element: error_text = error_element.text_content().strip() if error_text: print(f"❌ Login failed: {error_text}") return False print("❌ Login failed: Unknown reason") page.screenshot(path='login_failed.png') return False except Exception as e: print(f"⚠️ Login timeout or error: {e}") page.screenshot(path='login_error.png') return False except Exception as e: import traceback traceback.print_exc() print(f"🚨 Error during login process: {e}") return False def get_driver(): options = ChromeOptions() driver = Chrome(options=options) return driver PROXIES="http://gyfamaif:i2l62zihf63e@23.95.150.145:6114,http://gyfamaif:i2l62zihf63e@198.23.239.134:6540,http://gyfamaif:i2l62zihf63e@45.38.107.97:6014,http://gyfamaif:i2l62zihf63e@207.244.217.165:6712,http://gyfamaif:i2l62zihf63e@107.172.163.27:6543,http://gyfamaif:i2l62zihf63e@104.222.161.211:6343,http://gyfamaif:i2l62zihf63e@64.137.96.74:6641,http://gyfamaif:i2l62zihf63e@216.10.27.159:6837,http://gyfamaif:i2l62zihf63e@136.0.207.84:6661,http://gyfamaif:i2l62zihf63e@142.147.128.93:6593" def main(): # API parameters api_name = "mtop.relationrecommend.WirelessRecommend.recommend" api_version = "2.0" app_key = "12574478" data = { "appId": 32517, "params": "{\"verticalProductFlag\":\"pcmarket\",\"searchScene\":\"pcOfferSearch\",\"charset\":\"GBK\",\"beginPage\":1, \"pageSize\":60,\"keywords\":\"%C6%BB%B9%FB16max\",\"spm\":\"a260k.home2025.searchbox.0\",\"method\":\"getOfferList\"}" } # monitor = NetworkMonitor() # Get validated proxies # valid_proxies = get_validated_proxies() # if not valid_proxies: # print("No valid proxies found. Using direct connection.") # proxy_config = None # else: # # Select a random proxy from the validated list # selected_proxy = random.choice(valid_proxies) # print(f"Using proxy: {selected_proxy}") # # Parse proxy URL # proxy_match = re.match(r'(\w+)://([^:]+):(\d+)', selected_proxy) # if proxy_match: # protocol, server, port = proxy_match.groups() # proxy_config = { # "server": f"{protocol}://{server}:{port}", # } # else: # proxy_config = None # with sync_playwright() as p: driver = get_driver() # browser = p.chromium.launch( # headless=False, # # proxy={ # # "server": "http://47.109.31.148:8888", # # # "username": "avzvffai", # # # "password": "2dcyhf4owz3e" # # }, # proxy=proxy_config, # args=[ # '--disable-blink-features=AutomationControlled', # '--disable-web-security', # '--disable-features=VizDisplayCompositor', # '--no-sandbox', # '--disable-setuid-sandbox', # '--disable-dev-shm-usage', # '--disable-accelerated-2d-canvas', # '--no-first-run', # '--no-zygote', # '--disable-gpu', # '--disable-extensions', # '--disable-default-apps', # '--disable-features=IsolateOrigins,site-per-process', # '--disable-site-isolation-trials', # '--disable-background-timer-throttling', # '--disable-backgrounding-occluded-windows', # '--disable-renderer-backgrounding', # '--disable-back-forward-cache', # '--disable-ipc-flooding-protection', # '--disable-hang-monitor', # '--disable-prompt-on-repost', # '--disable-client-side-phishing-detection', # '--disable-component-update', # '--disable-sync', # '--metrics-recording-only', # '--mute-audio', # '--no-default-browser-check', # '--autoplay-policy=user-gesture-required', # '--disable-domain-reliability', # '--disable-partial-raster', # '--disable-speech-api', # '--disable-features=AudioServiceOutOfProcess', # '--disable-features=SharedArrayBuffer', # '--window-size=1920,1080', # '--start-maximized', # '--lang=zh-CN', # '--user-agent=' + get_random_user_agent(), # '--font-render-hinting=none' # ] # ) # Create context with realistic settings # Enable request interception # punish_url_pattern = "**/*punish*" # context.route(punish_url_pattern, monitor.capture_network_activity) # context.route("**/*", monitor.capture_network_activity) # Set realistic headers # page.set_extra_http_headers({ # 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8', # 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', # 'Accept-Encoding': 'gzip, deflate, br', # 'Connection': 'keep-alive', # 'Upgrade-Insecure-Requests': '1', # 'Sec-Fetch-Dest': 'document', # 'Sec-Fetch-Mode': 'navigate', # 'Sec-Fetch-Site': 'none', # 'Sec-Fetch-User': '?1', # 'Cache-Control': 'max-age=0', # 'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"', # 'sec-ch-ua-mobile': '?0', # 'sec-ch-ua-platform': '"Windows"' # }) driver.get("https://www.alibaba.com") time.sleep(random.randint(6,12)) actions = ActionChains(driver) # Move mouse randomly for _ in range(3): x = random.randint(50, 300) y = random.randint(60, 400) actions.move_by_offset(x, y).perform() time.sleep(random.uniform(1.5, 2)) time.sleep(3) # Extract cookies cookies = driver.get_cookies() token = None for cookie in cookies: if cookie["name"] == "_m_h5_tk": token = cookie["value"].split('_')[0] break if "login" in driver.current_url.lower(): print("Detection occurred - redirected to login") # Take screenshot for debugging driver.save_screenshot(filename="detection_screenshot.png") time.sleep(random.randint(0,4)) # login_1688_realistic('tb312129638026','Adnan@1234', driver) if not token: print("Error: Could not find _m_h5_tk cookie") print("To get your token:") print("1. Visit 1688.com in your browser") print("2. Open Developer Tools (F12)") print("3. Go to Application/Storage tab → Cookies") print("4. Find the _m_h5_tk cookie and copy the part before the underscore") driver.quit() return print(f"Extracted token: {token}") # Initialize the generator generator = MtopSignatureGenerator(token, app_key, api_name, api_version) # Generate the URL url, signature, timestamp = generator.generate_url(data) print(f"\nGenerated URL: {url}") print(f"\nSignature: {signature}") print(f"\nTimestamp: {timestamp}") # new_page = context.new_page() driver.execute_script(f"window.open('{url}', '_blank')") driver.switch_to.window(driver.window_handles[-1]) time.sleep(3) # Navigate to the generated URL in the same context to maintain cookies # response = new_page.goto(url, wait_until="domcontentloaded") # Get the content (JSON response) content = driver.page_source print("\nAPI Response:") print(content) # Check if we got the captcha response captcha_url = None response_data = None # Listen for API responses that might contain captcha def check_response(response): nonlocal captcha_url, response_data print("Checking Response") if "punish" in response.url: try: print("inside") response_data = response.json() print("Respo",response_data) if "FAIL_SYS_USER_VALIDATE" in str(response_data): if "url" in response_data.get("data", {}): captcha_url = response_data["data"]["url"] print(f"🎯 Captcha URL detected: {captcha_url}") except Exception as e: print("Error ",str(e)) pass soup = BeautifulSoup(driver.page_source, 'html.parser') pre = soup.find('pre') response = json.loads(pre.get_text()) if 'data' in response and 'url' in response['data'] and 'punish' in response['data']['url']: print("🔄 Redirecting to captcha verification page...") # Clean the URL (remove unnecessary parameters) # clean_url = captcha_url.split('?')[0] if '?' in captcha_url else captcha_url clean_url = response['data']['url'] print(f"🔗 Cleaned URL: {clean_url}") # Navigate to the captcha page in the same context driver.execute_script(f"window.open('{clean_url}','_blank')") driver.switch_to.window(driver.window_handles[-1]) # new_page.goto(clean_url, wait_until="domcontentloaded", timeout=30000) # Try to solve the captcha if True: print("✅ Successfully solved captcha! Returning to main page...") # page.wait_for_timeout(random.randint(5000, 6000)) time.sleep(3) # solve_slider_captcha_advanced(new_page) # Wait a bit after solving time.sleep(10) # monitor.save_to_files() generator = MtopSignatureGenerator(token, app_key, api_name, api_version) # Generate the URL url, signature, timestamp = generator.generate_url(data) print(f"\nGenerated URL: {url}") print(f"\nSignature: {signature}") print(f"\nTimestamp: {timestamp}") # new_page = context.new_page() # Navigate to the generated URL in the same context to maintain cookies # response = new_page.goto(url, wait_until="domcontentloaded") driver.execute_script(f"window.open('{url}','_blank')") driver.switch_to.window(driver.window_handles[-1]) # Get the content (JSON response) content = driver.page_source print("\nAPI Response:") # Optionally, save to file with open("api_response.json", "w", encoding="utf-8") as f: f.write(content) # Display debug info # print("\nDebug Information:") debug_info = { "apiName": api_name, "apiVersion": api_version, "appKey": app_key, "token": token, "timestamp": timestamp, "signContent": f"{token}&{timestamp}&{app_key}&{json.dumps(data, ensure_ascii=False, separators=(',', ':'))}", "generatedSign": signature } # print(json.dumps(debug_info, indent=2, ensure_ascii=False)) # Keep the browser open for inspection input("Press Enter to close the browser...") driver.quit() if __name__ == "__main__": main()