358 lines
13 KiB
Python
358 lines
13 KiB
Python
"""
|
|
MEXC Browser Automation for Cookie Extraction and Request Monitoring
|
|
|
|
This module uses Selenium to automate browser interactions and extract
|
|
session cookies and request data for MEXC futures trading.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
import json
|
|
from typing import Dict, List, Optional, Any
|
|
from selenium import webdriver
|
|
from selenium.webdriver.chrome.options import Options
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.common.exceptions import TimeoutException, WebDriverException
|
|
from selenium.webdriver.chrome.service import Service
|
|
from webdriver_manager.chrome import ChromeDriverManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class MEXCBrowserAutomation:
|
|
"""
|
|
Browser automation for MEXC futures trading session management
|
|
"""
|
|
|
|
def __init__(self, headless: bool = False, proxy: Optional[str] = None):
|
|
"""
|
|
Initialize browser automation
|
|
|
|
Args:
|
|
headless: Run browser in headless mode
|
|
proxy: HTTP proxy to use (format: host:port)
|
|
"""
|
|
self.driver = None
|
|
self.headless = headless
|
|
self.proxy = proxy
|
|
self.logged_in = False
|
|
|
|
def setup_chrome_driver(self) -> webdriver.Chrome:
|
|
"""Setup Chrome driver with appropriate options"""
|
|
chrome_options = Options()
|
|
|
|
if self.headless:
|
|
chrome_options.add_argument("--headless")
|
|
|
|
# Basic Chrome options for automation
|
|
chrome_options.add_argument("--no-sandbox")
|
|
chrome_options.add_argument("--disable-dev-shm-usage")
|
|
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
|
|
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
|
chrome_options.add_experimental_option('useAutomationExtension', False)
|
|
|
|
# Set user agent to avoid detection
|
|
chrome_options.add_argument("--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36")
|
|
|
|
# Proxy setup if provided
|
|
if self.proxy:
|
|
chrome_options.add_argument(f"--proxy-server=http://{self.proxy}")
|
|
|
|
# Enable network logging
|
|
chrome_options.add_argument("--enable-logging")
|
|
chrome_options.add_argument("--log-level=0")
|
|
chrome_options.set_capability("goog:loggingPrefs", {"performance": "ALL"})
|
|
|
|
# Automatically download and setup ChromeDriver
|
|
service = Service(ChromeDriverManager().install())
|
|
|
|
try:
|
|
driver = webdriver.Chrome(service=service, options=chrome_options)
|
|
|
|
# Execute script to avoid detection
|
|
driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
|
|
|
|
return driver
|
|
except WebDriverException as e:
|
|
logger.error(f"Failed to setup Chrome driver: {e}")
|
|
raise
|
|
|
|
def start_browser(self):
|
|
"""Start the browser session"""
|
|
if self.driver is None:
|
|
logger.info("Starting Chrome browser for MEXC automation")
|
|
self.driver = self.setup_chrome_driver()
|
|
logger.info("Browser started successfully")
|
|
|
|
def stop_browser(self):
|
|
"""Stop the browser session"""
|
|
if self.driver:
|
|
logger.info("Stopping browser")
|
|
self.driver.quit()
|
|
self.driver = None
|
|
|
|
def navigate_to_mexc_futures(self, symbol: str = "ETH_USDT"):
|
|
"""
|
|
Navigate to MEXC futures trading page
|
|
|
|
Args:
|
|
symbol: Trading symbol to navigate to
|
|
"""
|
|
if not self.driver:
|
|
self.start_browser()
|
|
|
|
url = f"https://www.mexc.com/en-GB/futures/{symbol}?type=linear_swap"
|
|
logger.info(f"Navigating to MEXC futures: {url}")
|
|
|
|
self.driver.get(url)
|
|
|
|
# Wait for page to load
|
|
try:
|
|
WebDriverWait(self.driver, 10).until(
|
|
EC.presence_of_element_located((By.TAG_NAME, "body"))
|
|
)
|
|
logger.info("MEXC futures page loaded")
|
|
except TimeoutException:
|
|
logger.error("Timeout waiting for MEXC page to load")
|
|
|
|
def wait_for_login(self, timeout: int = 300) -> bool:
|
|
"""
|
|
Wait for user to manually log in to MEXC
|
|
|
|
Args:
|
|
timeout: Maximum time to wait for login (seconds)
|
|
|
|
Returns:
|
|
bool: True if login detected, False if timeout
|
|
"""
|
|
logger.info("Please log in to MEXC manually in the browser window")
|
|
logger.info("Waiting for login completion...")
|
|
|
|
start_time = time.time()
|
|
|
|
while time.time() - start_time < timeout:
|
|
# Check if we can find elements that indicate logged in state
|
|
try:
|
|
# Look for user-specific elements that appear after login
|
|
cookies = self.driver.get_cookies()
|
|
|
|
# Check for authentication cookies
|
|
auth_cookies = ['uc_token', 'u_id']
|
|
logged_in_indicators = 0
|
|
|
|
for cookie in cookies:
|
|
if cookie['name'] in auth_cookies and cookie['value']:
|
|
logged_in_indicators += 1
|
|
|
|
if logged_in_indicators >= 2:
|
|
logger.info("Login detected!")
|
|
self.logged_in = True
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Error checking login status: {e}")
|
|
|
|
time.sleep(2) # Check every 2 seconds
|
|
|
|
logger.error(f"Login timeout after {timeout} seconds")
|
|
return False
|
|
|
|
def extract_session_cookies(self) -> Dict[str, str]:
|
|
"""
|
|
Extract all cookies from current browser session
|
|
|
|
Returns:
|
|
Dictionary of cookie name-value pairs
|
|
"""
|
|
if not self.driver:
|
|
logger.error("Browser not started")
|
|
return {}
|
|
|
|
cookies = {}
|
|
|
|
try:
|
|
browser_cookies = self.driver.get_cookies()
|
|
|
|
for cookie in browser_cookies:
|
|
cookies[cookie['name']] = cookie['value']
|
|
|
|
logger.info(f"Extracted {len(cookies)} cookies from browser session")
|
|
|
|
# Log important cookies (without values for security)
|
|
important_cookies = ['uc_token', 'u_id', 'x-mxc-fingerprint', 'mexc_fingerprint_visitorId']
|
|
for cookie_name in important_cookies:
|
|
if cookie_name in cookies:
|
|
logger.info(f"Found important cookie: {cookie_name}")
|
|
else:
|
|
logger.warning(f"Missing important cookie: {cookie_name}")
|
|
|
|
return cookies
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to extract cookies: {e}")
|
|
return {}
|
|
|
|
def monitor_network_requests(self, duration: int = 60) -> List[Dict[str, Any]]:
|
|
"""
|
|
Monitor network requests for the specified duration
|
|
|
|
Args:
|
|
duration: How long to monitor requests (seconds)
|
|
|
|
Returns:
|
|
List of captured network requests
|
|
"""
|
|
if not self.driver:
|
|
logger.error("Browser not started")
|
|
return []
|
|
|
|
logger.info(f"Starting network monitoring for {duration} seconds")
|
|
logger.info("Please perform trading actions in the browser (open/close positions)")
|
|
|
|
start_time = time.time()
|
|
captured_requests = []
|
|
|
|
while time.time() - start_time < duration:
|
|
try:
|
|
# Get performance logs (network requests)
|
|
logs = self.driver.get_log('performance')
|
|
|
|
for log in logs:
|
|
message = json.loads(log['message'])
|
|
|
|
# Filter for relevant MEXC API requests
|
|
if (message.get('message', {}).get('method') == 'Network.responseReceived'):
|
|
response = message['message']['params']['response']
|
|
url = response.get('url', '')
|
|
|
|
# Look for futures API calls
|
|
if ('futures.mexc.com' in url or
|
|
'ucgateway/captcha_api' in url or
|
|
'api/v1/private' in url):
|
|
|
|
request_data = {
|
|
'url': url,
|
|
'method': response.get('mimeType', ''),
|
|
'status': response.get('status'),
|
|
'headers': response.get('headers', {}),
|
|
'timestamp': log['timestamp']
|
|
}
|
|
|
|
captured_requests.append(request_data)
|
|
logger.info(f"Captured request: {url}")
|
|
|
|
except Exception as e:
|
|
logger.debug(f"Error in network monitoring: {e}")
|
|
|
|
time.sleep(1)
|
|
|
|
logger.info(f"Network monitoring complete. Captured {len(captured_requests)} requests")
|
|
return captured_requests
|
|
|
|
def perform_test_trade(self, symbol: str = "ETH_USDT", volume: float = 1.0, leverage: int = 200):
|
|
"""
|
|
Attempt to perform a test trade to capture the complete request flow
|
|
|
|
Args:
|
|
symbol: Trading symbol
|
|
volume: Position size
|
|
leverage: Leverage multiplier
|
|
"""
|
|
if not self.logged_in:
|
|
logger.error("Not logged in - cannot perform test trade")
|
|
return
|
|
|
|
logger.info(f"Attempting test trade: {symbol}, Volume: {volume}, Leverage: {leverage}x")
|
|
logger.info("This will attempt to click trading interface elements")
|
|
|
|
try:
|
|
# This would need to be implemented based on MEXC's specific UI elements
|
|
# For now, just wait and let user perform manual actions
|
|
logger.info("Please manually place a small test trade while monitoring is active")
|
|
time.sleep(30)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error during test trade: {e}")
|
|
|
|
def full_session_capture(self, symbol: str = "ETH_USDT") -> Dict[str, Any]:
|
|
"""
|
|
Complete session capture workflow
|
|
|
|
Args:
|
|
symbol: Trading symbol to use
|
|
|
|
Returns:
|
|
Dictionary containing cookies and captured requests
|
|
"""
|
|
logger.info("Starting full MEXC session capture")
|
|
|
|
try:
|
|
# Start browser and navigate to MEXC
|
|
self.navigate_to_mexc_futures(symbol)
|
|
|
|
# Wait for manual login
|
|
if not self.wait_for_login():
|
|
return {'success': False, 'error': 'Login timeout'}
|
|
|
|
# Extract session cookies
|
|
cookies = self.extract_session_cookies()
|
|
|
|
if not cookies:
|
|
return {'success': False, 'error': 'Failed to extract cookies'}
|
|
|
|
# Monitor network requests while user performs actions
|
|
logger.info("Starting network monitoring - please perform trading actions now")
|
|
requests = self.monitor_network_requests(duration=120) # 2 minutes
|
|
|
|
return {
|
|
'success': True,
|
|
'cookies': cookies,
|
|
'network_requests': requests,
|
|
'timestamp': int(time.time())
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in session capture: {e}")
|
|
return {'success': False, 'error': str(e)}
|
|
|
|
finally:
|
|
self.stop_browser()
|
|
|
|
def main():
|
|
"""Main function for standalone execution"""
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
print("MEXC Browser Automation - Session Capture")
|
|
print("This will open a browser window for you to log into MEXC")
|
|
print("Make sure you have Chrome browser installed")
|
|
|
|
automation = MEXCBrowserAutomation(headless=False)
|
|
|
|
try:
|
|
result = automation.full_session_capture()
|
|
|
|
if result['success']:
|
|
print(f"\nSession capture successful!")
|
|
print(f"Extracted {len(result['cookies'])} cookies")
|
|
print(f"Captured {len(result['network_requests'])} network requests")
|
|
|
|
# Save results to file
|
|
output_file = f"mexc_session_capture_{int(time.time())}.json"
|
|
with open(output_file, 'w') as f:
|
|
json.dump(result, f, indent=2)
|
|
|
|
print(f"Results saved to: {output_file}")
|
|
|
|
else:
|
|
print(f"Session capture failed: {result['error']}")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\nSession capture interrupted by user")
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
finally:
|
|
automation.stop_browser()
|
|
|
|
if __name__ == "__main__":
|
|
main() |