popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

capture more capcha info

Browse Source
This commit is contained in:
Dobromir Popov
2025-07-03 02:20:21 +03:00
parent e4a611a0cc
commit 09ed86c8ae
4 changed files with 15742 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
core/mexc_webclient/auto_browser.py
View File

@ -66,86 +66,74 @@ class MEXCRequestInterceptor:
self.requests_file = f"mexc_requests_{self.timestamp}.json" self.requests_file = f"mexc_requests_{self.timestamp}.json"
self.cookies_file = f"mexc_cookies_{self.timestamp}.json" self.cookies_file = f"mexc_cookies_{self.timestamp}.json"
def setup_chrome_with_logging(self) -> webdriver.Chrome: def setup_browser(self):
"""Setup Brave browser with performance logging enabled""" """Setup Chrome browser with necessary options"""
logger.info("Setting up ChromeDriver with request interception for Brave browser...") chrome_options = webdriver.ChromeOptions()
# Enable headless mode if needed
# Chrome options (used for Brave as it's Chromium-based)
chrome_options = Options()
# Set up Chrome options with a user data directory to persist session
user_data_dir = os.path.join(os.getcwd(), 'chrome_user_data')
os.makedirs(user_data_dir, exist_ok=True)
chrome_options.add_argument(f'--user-data-dir={user_data_dir}')
chrome_options.add_argument('--disable-extensions')
chrome_options.add_argument('--disable-gpu')
chrome_options.add_argument('--window-size=1920,1080')
if self.headless: if self.headless:
chrome_options.add_argument('--headless') chrome_options.add_argument('--headless')
chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--disable-gpu')
chrome_options.add_argument('--window-size=1920,1080')
chrome_options.add_argument('--disable-extensions')
# Set the binary location for Brave # Set up Chrome options with a user data directory to persist session
# brave_binary_path = "C:\\Program Files\\BraveSoftware\\Brave-Browser\\Application\\brave.exe" user_data_base_dir = os.path.join(os.getcwd(), 'chrome_user_data')
# chrome_options.binary_location = brave_binary_path os.makedirs(user_data_base_dir, exist_ok=True)
# logger.info(f"Using Brave binary at: {brave_binary_path}")
# Essential options for automation # Check for existing session directories
chrome_options.add_argument("--no-sandbox") session_dirs = [d for d in os.listdir(user_data_base_dir) if d.startswith('session_')]
chrome_options.add_argument("--disable-dev-shm-usage") session_dirs.sort(reverse=True) # Sort descending to get the most recent first
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
chrome_options.add_argument("--disable-web-security")
chrome_options.add_argument("--allow-running-insecure-content")
chrome_options.add_argument("--disable-features=VizDisplayCompositor")
# User agent to avoid detection user_data_dir = None
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" if session_dirs:
chrome_options.add_argument(f"--user-agent={user_agent}") use_existing = input(f"Found {len(session_dirs)} existing sessions. Use an existing session? (y/n): ").lower().strip() == 'y'
if use_existing:
print("Available sessions:")
for i, session in enumerate(session_dirs[:5], 1): # Show up to 5 most recent
print(f"{i}. {session}")
choice = input("Enter session number (default 1) or any other key for most recent: ")
if choice.isdigit() and 1 <= int(choice) <= len(session_dirs):
selected_session = session_dirs[int(choice) - 1]
else:
selected_session = session_dirs[0]
user_data_dir = os.path.join(user_data_base_dir, selected_session)
print(f"Using session: {selected_session}")
# Disable automation flags if user_data_dir is None:
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"]) user_data_dir = os.path.join(user_data_base_dir, f'session_{self.timestamp}')
chrome_options.add_experimental_option('useAutomationExtension', False) os.makedirs(user_data_dir, exist_ok=True)
print(f"Creating new session: session_{self.timestamp}")
# Enable performance logging for network requests chrome_options.add_argument(f'--user-data-dir={user_data_dir}')
chrome_options.add_argument("--enable-logging")
chrome_options.add_argument("--log-level=0")
chrome_options.add_argument("--v=1")
# Set logging preferences for performance data # Enable logging to capture JS console output and network activity
chrome_options.set_capability('goog:loggingPrefs', { chrome_options.set_capability('goog:loggingPrefs', {
'performance': 'ALL', 'browser': 'ALL',
'browser': 'ALL' 'performance': 'ALL'
}) })
# # Add profile directory argument as provided by the user
# chrome_options.add_argument("--profile-directory=Profile 3")
try: try:
# Automatically download and install ChromeDriver self.driver = webdriver.Chrome(options=chrome_options)
logger.info("Downloading/updating ChromeDriver...")
service = Service(ChromeDriverManager().install())
# Create driver
driver = webdriver.Chrome(
service=service,
options=chrome_options
)
# Hide automation indicators
driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
driver.execute_cdp_cmd('Network.setUserAgentOverride', {
"userAgent": user_agent
})
# Enable network domain for CDP
driver.execute_cdp_cmd('Network.enable', {})
driver.execute_cdp_cmd('Runtime.enable', {})
logger.info("ChromeDriver setup complete!")
return driver
except Exception as e: except Exception as e:
logger.error(f"Failed to setup ChromeDriver: {e}") print(f"Failed to start browser with session: {e}")
raise print("Falling back to a new session...")
user_data_dir = os.path.join(user_data_base_dir, f'session_{self.timestamp}_fallback')
os.makedirs(user_data_dir, exist_ok=True)
print(f"Creating fallback session: session_{self.timestamp}_fallback")
chrome_options = webdriver.ChromeOptions()
if self.headless:
chrome_options.add_argument('--headless')
chrome_options.add_argument('--disable-gpu')
chrome_options.add_argument('--window-size=1920,1080')
chrome_options.add_argument('--disable-extensions')
chrome_options.add_argument(f'--user-data-dir={user_data_dir}')
chrome_options.set_capability('goog:loggingPrefs', {
'browser': 'ALL',
'performance': 'ALL'
})
self.driver = webdriver.Chrome(options=chrome_options)
return self.driver
def start_monitoring(self): def start_monitoring(self):
"""Start the browser and begin monitoring""" """Start the browser and begin monitoring"""
@ -153,7 +141,7 @@ class MEXCRequestInterceptor:
try: try:
# Setup ChromeDriver # Setup ChromeDriver
self.driver = self.setup_chrome_with_logging() self.driver = self.setup_browser()
# Navigate to MEXC futures # Navigate to MEXC futures
mexc_url = "https://www.mexc.com/en-GB/futures/ETH_USDT?type=linear_swap" mexc_url = "https://www.mexc.com/en-GB/futures/ETH_USDT?type=linear_swap"

90
core/mexc_webclient/mexc_futures_client.py
View File

@ -22,6 +22,17 @@ from urllib.parse import urlencode
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class MEXCSessionManager:
def __init__(self):
self.captcha_token = None
def get_captcha_token(self) -> str:
return self.captcha_token if self.captcha_token else ""
def save_captcha_token(self, token: str):
self.captcha_token = token
logger.info("MEXC: Captcha token saved in session manager")
class MEXCFuturesWebClient: class MEXCFuturesWebClient:
""" """
MEXC Futures Web Client that mimics browser behavior for futures trading. MEXC Futures Web Client that mimics browser behavior for futures trading.
@ -30,30 +41,27 @@ class MEXCFuturesWebClient:
the exact HTTP requests made by their web interface. the exact HTTP requests made by their web interface.
""" """
def __init__(self, session_cookies: Dict[str, str] = None): def __init__(self, api_key: str, api_secret: str, user_id: str, base_url: str = 'https://www.mexc.com', headless: bool = True):
""" """
Initialize the MEXC Futures Web Client Initialize the MEXC Futures Web Client
Args: Args:
session_cookies: Dictionary of cookies from an authenticated browser session api_key: API key for authentication
api_secret: API secret for authentication
user_id: User ID for authentication
base_url: Base URL for the MEXC website
headless: Whether to run the browser in headless mode
""" """
self.session = requests.Session() self.api_key = api_key
self.api_secret = api_secret
# Base URLs for different endpoints self.user_id = user_id
self.base_url = "https://www.mexc.com" self.base_url = base_url
self.futures_api_url = "https://futures.mexc.com/api/v1"
self.captcha_url = f"{self.base_url}/ucgateway/captcha_api/captcha/robot"
# Session state
self.is_authenticated = False self.is_authenticated = False
self.user_id = None self.headless = headless
self.auth_token = None self.session = requests.Session()
self.fingerprint = None self.session_manager = MEXCSessionManager() # Adding session_manager attribute
self.visitor_id = None self.captcha_url = f'{base_url}/ucgateway/captcha_api'
self.futures_api_url = "https://futures.mexc.com/api/v1"
# Load session cookies if provided
if session_cookies:
self.load_session_cookies(session_cookies)
# Setup default headers that mimic a real browser # Setup default headers that mimic a real browser
self.setup_browser_headers() self.setup_browser_headers()
@ -142,37 +150,59 @@ class MEXCFuturesWebClient:
endpoint = f"robot.future.{side}.{symbol}.{leverage}" endpoint = f"robot.future.{side}.{symbol}.{leverage}"
url = f"{self.captcha_url}/{endpoint}" url = f"{self.captcha_url}/{endpoint}"
# Setup headers for captcha request # Attempt to get captcha token from session manager
captcha_token = self.session_manager.get_captcha_token()
if not captcha_token:
logger.warning("MEXC: No captcha token available, attempting to fetch from browser")
captcha_token = self._extract_captcha_token_from_browser()
if captcha_token:
self.session_manager.save_captcha_token(captcha_token)
else:
logger.error("MEXC: Failed to extract captcha token from browser")
return False
headers = { headers = {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
'Language': 'en-GB', 'Language': 'en-GB',
'Referer': f'{self.base_url}/en-GB/futures/{symbol}?type=linear_swap', 'Referer': f'{self.base_url}/en-GB/futures/{symbol}?type=linear_swap',
'trochilus-uid': self.user_id if self.user_id else '', 'trochilus-uid': self.user_id if self.user_id else '',
'trochilus-trace-id': f"{uuid.uuid4()}-{int(time.time() * 1000) % 10000:04d}" 'trochilus-trace-id': f"{uuid.uuid4()}-{int(time.time() * 1000) % 10000:04d}",
'captcha-token': captcha_token
} }
# Add captcha token if available (this would need to be extracted from browser) logger.info(f"MEXC: Verifying captcha for {endpoint}")
# For now, we'll make the request without it and see what happens
try: try:
response = self.session.get(url, headers=headers, timeout=10) response = self.session.get(url, headers=headers, timeout=10)
if response.status_code == 200: if response.status_code == 200:
data = response.json() data = response.json()
if data.get('success') and data.get('code') == 0: if data.get('success'):
logger.info(f"MEXC: Captcha verification successful for {side} {symbol}") logger.info(f"MEXC: Captcha verified successfully for {endpoint}")
return True return True
else: else:
logger.warning(f"MEXC: Captcha verification failed: {data}") logger.error(f"MEXC: Captcha verification failed for {endpoint}: {data}")
return False return False
else: else:
logger.error(f"MEXC: Captcha request failed with status {response.status_code}") logger.error(f"MEXC: Captcha verification request failed with status {response.status_code}: {response.text}")
return False return False
except Exception as e: except Exception as e:
logger.error(f"MEXC: Captcha verification error: {e}") logger.error(f"MEXC: Captcha verification error for {endpoint}: {str(e)}")
return False return False
def _extract_captcha_token_from_browser(self) -> str:
"""
Extract captcha token from browser session using stored cookies or requests.
This is a placeholder for actual implementation which would interact with browser data.
"""
try:
# Placeholder for extracting token from browser session or stored data
# In a real scenario, this would parse the mexc_requests JSON file or interact with Selenium
logger.info("MEXC: Attempting to extract captcha token from browser data")
# For now, return empty string as placeholder
return ""
except Exception as e:
logger.error(f"MEXC: Error extracting captcha token from browser: {str(e)}")
return ""
def generate_signature(self, method: str, path: str, params: Dict[str, Any], def generate_signature(self, method: str, path: str, params: Dict[str, Any],
timestamp: int, nonce: int) -> str: timestamp: int, nonce: int) -> str:
""" """

15618
mexc_requests_20250703_021049.json Normal file
View File

File diff suppressed because it is too large Load Diff

10
tests/test_mexc_futures_webclient.py
View File

@ -82,7 +82,10 @@ def test_basic_connection():
logger.warning("Extracted cookies may be incomplete") logger.warning("Extracted cookies may be incomplete")
# Initialize the web client # Initialize the web client
client = MEXCFuturesWebClient(session_cookies=cookies) client = MEXCFuturesWebClient(api_key='', api_secret='', user_id='', base_url='https://www.mexc.com', headless=True)
# Load cookies into the client's session
for name, value in cookies.items():
client.session.cookies.set(name, value)
# Update headers to include additional parameters from captured requests # Update headers to include additional parameters from captured requests
client.session.headers.update({ client.session.headers.update({
@ -246,7 +249,10 @@ def main():
logger.error("No valid session available") logger.error("No valid session available")
return return
client = MEXCFuturesWebClient(session_cookies=cookies) client = MEXCFuturesWebClient(api_key='', api_secret='', user_id='', base_url='https://www.mexc.com', headless=True)
# Load cookies into the client's session
for name, value in cookies.items():
client.session.cookies.set(name, value)
if not client.is_authenticated: if not client.is_authenticated:
logger.error("Failed to authenticate with MEXC") logger.error("Failed to authenticate with MEXC")