#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Bangladesh Customs Bill Tracking Automation (Headless VPS + Private Mode) Author : Izaz Ahamed ----------------------------------------------------------------------- ✅ Headless Chrome (Linux VPS) ✅ Private (Incognito) mode — no history, no cache, no cookies ✅ Retry + requeue failed rows (never skip) ✅ Screenshot + HTML debug on failure ✅ Audio reCAPTCHA solving supported ✅ 1360×768 consistent screenshot → PDF ✅ Exports failed_rows.csv for Commercial RPA requeue """ import os, sys, time, random, tempfile, logging, urllib.request, shutil, pandas as pd import re import subprocess from PIL import Image import pydub # ----------------------------------------------------------------------------- # OPTIONAL: SpeechRecognition (for audio challenge) # ----------------------------------------------------------------------------- try: import speech_recognition as sr SR_AVAILABLE = True except ImportError: SR_AVAILABLE = False print("⚠️ SpeechRecognition not installed — audio challenge skipped.") try: import requests except ImportError: requests = None print("⚠️ Warning: 'requests' library not found. Server communication will be disabled.") print(" Install with: pip install requests") # Use Electron browser instead of Selenium try: from shared.electron_browser import ElectronBrowser, By, ElectronWait, EC except ImportError: # Fallback for direct execution import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from automation_scripts.shared.electron_browser import ElectronBrowser, By, ElectronWait, EC # Compatibility - Select is not yet implemented in ElectronBrowser class Select: def __init__(self, element): self.element = element def select_by_value(self, value): # Use JavaScript to set select value script = f"arguments[0].value = '{value}'; arguments[0].dispatchEvent(new Event('change', {{bubbles: true}}));" return self.element.browser.execute_script(script) # Windows registry access for Chrome version detection try: import winreg HAS_WINREG = True except ImportError: HAS_WINREG = False # ----------------------------------------------------------------------------- # Browser Setup Functions # ----------------------------------------------------------------------------- def _detect_chrome_binary_paths(): """Detect bundled Chromium binary path from environment variables. Only uses bundled Chromium from CHROMIUM_BINARY_PATH or GOOGLE_CHROME_BIN. System Chrome installations are NOT used as fallback. Returns: list: List containing the bundled Chromium path if found Raises: FileNotFoundError: If bundled Chromium is not found or environment variables are not set """ import logging logger = logging.getLogger(__name__) candidates = [] # Check for bundled Chromium from Electron (required, no fallback) logger.info("🔍 Checking for bundled Chromium from environment variables...") bundled_paths = [ ('CHROMIUM_BINARY_PATH', os.environ.get('CHROMIUM_BINARY_PATH')), ('GOOGLE_CHROME_BIN', os.environ.get('GOOGLE_CHROME_BIN')), ] for env_var_name, path in bundled_paths: if path: logger.info(f" {env_var_name}={path}") if os.path.exists(path): logger.info(f" ✅ Bundled Chromium found at: {path}") candidates.append(path) return candidates else: error_msg = ( f"Bundled Chromium is required but not found.\n" f"{env_var_name} is set to '{path}' but the file does not exist.\n" f"Please ensure Chromium is bundled in resources/chromium/ directory." ) logger.error(f" ❌ {error_msg}") raise FileNotFoundError(error_msg) else: logger.debug(f" {env_var_name} not set") # No bundled Chromium found - raise error error_msg = ( "Bundled Chromium is required but not found.\n" "CHROMIUM_BINARY_PATH and GOOGLE_CHROME_BIN environment variables are not set.\n" "Please ensure Chromium is bundled in resources/chromium/ directory.\n" "System Chrome installations are not used as fallback." ) logger.error(f" ❌ {error_msg}") raise FileNotFoundError(error_msg) def get_installed_chrome_major_version(): """Detect the major version of installed Chrome/Chromium browser. Returns: int: Major version number (e.g., 141) or None if not detected """ # Try detected binary paths first for binary in _detect_chrome_binary_paths(): try: out = subprocess.check_output([binary, '--version'], stderr=subprocess.STDOUT, timeout=5) text = out.decode('utf-8', errors='ignore') # Match patterns like "Chrome 141.0.7390.123" or "Chromium 141.0.7390.123" m = re.search(r'(?:Chrome|Chromium)\s+([0-9]+)\.', text) if m: version = int(m.group(1)) return version except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError, ValueError): continue except Exception: continue # Fallback: try plain `google-chrome --version` commands for cmd in ['google-chrome', 'google-chrome-stable', 'chromium', 'chromium-browser']: try: out = subprocess.check_output([cmd, '--version'], stderr=subprocess.STDOUT, timeout=5) text = out.decode('utf-8', errors='ignore') m = re.search(r'([0-9]+)\.', text) if m: version = int(m.group(1)) return version except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError, ValueError): pass except Exception: pass # Windows-specific: Try reading from registry if HAS_WINREG and os.name == 'nt': try: # Try multiple registry paths where Chrome version might be stored registry_paths = [ (winreg.HKEY_CURRENT_USER, r'Software\Google\Chrome\BLBeacon'), (winreg.HKEY_LOCAL_MACHINE, r'Software\Google\Chrome\BLBeacon'), (winreg.HKEY_CURRENT_USER, r'Software\Google\Update\ClientState\{8A69D345-D564-463C-AFF1-A69D9E530F96}'), (winreg.HKEY_LOCAL_MACHINE, r'Software\Google\Update\ClientState\{8A69D345-D564-463C-AFF1-A69D9E530F96}'), ] for hkey, path in registry_paths: try: key = winreg.OpenKey(hkey, path) try: # Try to get version from 'version' value version_str, _ = winreg.QueryValueEx(key, 'version') m = re.search(r'([0-9]+)\.', str(version_str)) if m: version = int(m.group(1)) winreg.CloseKey(key) return version except FileNotFoundError: pass finally: winreg.CloseKey(key) except (FileNotFoundError, OSError): continue except Exception: pass # Windows-specific: Try reading from Chrome's version file if os.name == 'nt': version_file_paths = [ os.path.expanduser(r'~\AppData\Local\Google\Chrome\Application\version'), r'C:\Program Files\Google\Chrome\Application\version', r'C:\Program Files (x86)\Google\Chrome\Application\version', ] for version_file in version_file_paths: try: if os.path.exists(version_file): with open(version_file, 'r', encoding='utf-8', errors='ignore') as f: content = f.read() m = re.search(r'([0-9]+)\.', content) if m: version = int(m.group(1)) return version except Exception: continue return None def ensure_uc_cache_directory(): """Ensure undetected_chromedriver cache directory exists with proper permissions. Returns: str: Path to the cache directory that was ensured, or None if failed """ cache_dirs = [ os.path.expanduser('~/.local/share/undetected_chromedriver'), # Linux os.path.expanduser('~/AppData/Roaming/undetected_chromedriver'), # Windows os.path.expanduser('~/Library/Application Support/undetected_chromedriver'), # macOS ] for cache_dir in cache_dirs: try: # Create directory if it doesn't exist os.makedirs(cache_dir, mode=0o755, exist_ok=True) # Verify it's actually a directory and we can write to it if os.path.isdir(cache_dir) and os.access(cache_dir, os.W_OK): return cache_dir except Exception: continue return None def clear_uc_cache(): """Clear undetected_chromedriver cache from common locations""" cache_dirs = [ os.path.expanduser('~/.local/share/undetected_chromedriver'), # Linux os.path.expanduser('~/AppData/Roaming/undetected_chromedriver'), # Windows os.path.expanduser('~/Library/Application Support/undetected_chromedriver'), # macOS ] for cache_dir in cache_dirs: try: if os.path.isdir(cache_dir): shutil.rmtree(cache_dir, ignore_errors=True) except Exception: pass def validate_driver_file(cache_dir=None, logger=None): """Validate that ChromeDriver executable exists and is accessible. Args: cache_dir: Optional cache directory path to check logger: Optional logger instance for logging Returns: bool: True if driver file appears to exist, False otherwise """ if not cache_dir: cache_dir = ensure_uc_cache_directory() if not cache_dir: return False # Check common driver file locations driver_paths = [ os.path.join(cache_dir, 'undetected_chromedriver'), os.path.join(cache_dir, 'undetected', 'chromedriver-linux64', 'chromedriver'), os.path.join(cache_dir, 'undetected', 'chromedriver-win64', 'chromedriver.exe'), os.path.join(cache_dir, 'undetected', 'chromedriver-mac-x64', 'chromedriver'), ] for driver_path in driver_paths: if os.path.exists(driver_path): # Check if it's executable (on Unix-like systems) if os.name != 'nt' and not os.access(driver_path, os.X_OK): try: os.chmod(driver_path, 0o755) except Exception: pass if logger: logger.debug(f"✅ Found driver file: {driver_path}") return True return False def validate_driver_session(driver, logger=None): """Validate that the ChromeDriver session is actually working. Args: driver: ChromeDriver instance to validate logger: Optional logger instance for logging Returns: bool: True if session is working, False otherwise """ try: # Try a simple command to verify the session is alive _ = driver.current_url return True except Exception as e: error_msg = str(e) if logger: # Check for connection errors if "Connection refused" in error_msg or "Failed to establish" in error_msg: logger.warning(f"⚠️ Driver session connection error: {error_msg}") else: logger.warning(f"⚠️ Driver session validation failed: {error_msg}") return False def build_default_options(headless=True, user_data_dir=None, window_size='1366,768', incognito=True): """Build ChromeOptions with proper headless configuration. Args: headless: If True, run Chrome in headless mode (no visible window) user_data_dir: Optional user data directory path (if provided, cache will be preserved for session persistence) window_size: Window size string like '1920,1080' incognito: If True, run Chrome in incognito/private mode (default True for backward compatibility) Note: When user_data_dir is provided, incognito is automatically disabled for session persistence Returns: uc.ChromeOptions: Configured ChromeOptions object """ opts = uc.ChromeOptions() # When user_data_dir is provided, we need to preserve cache and disable incognito for session persistence # Session persistence requires non-incognito mode and cache to be enabled if user_data_dir: incognito = False # Override incognito when using user_data_dir for session persistence # Headless mode configuration - must be set first if headless: # Primary headless flag (--headless=new is the modern headless mode) opts.add_argument('--headless=new') opts.add_argument('--disable-gpu') # Required for headless on Windows opts.add_argument('--remote-debugging-port=0') # Disable remote debugging in headless opts.add_argument('--disable-software-rasterizer') # Window positioning to move off-screen if window appears opts.add_argument('--window-position=-2000,-2000') # Additional flags to prevent window visibility opts.add_argument('--hide-scrollbars') opts.add_argument('--mute-audio') opts.add_argument('--disable-background-timer-throttling') opts.add_argument('--disable-backgrounding-occluded-windows') opts.add_argument('--disable-renderer-backgrounding') opts.add_argument('--disable-features=TranslateUI') opts.add_argument('--disable-ipc-flooding-protection') # Additional aggressive flags to prevent any window visibility opts.add_argument('--disable-infobars') opts.add_argument('--disable-notifications') opts.add_argument('--disable-popup-blocking') opts.add_argument('--disable-default-apps') opts.add_argument('--disable-session-crashed-bubble') opts.add_argument('--disable-component-update') opts.add_argument('--disable-sync') opts.add_argument('--no-pings') opts.add_argument('--disable-hang-monitor') opts.add_argument('--disable-prompt-on-repost') opts.add_argument('--disable-domain-reliability') opts.add_argument('--disable-client-side-phishing-detection') opts.add_argument('--disable-component-extensions-with-background-pages') opts.add_argument('--force-color-profile=srgb') opts.add_argument('--metrics-recording-only') opts.add_argument('--disable-breakpad') else: # Even in non-headless mode, disable GPU can help with stability opts.add_argument('--disable-gpu') # Common options for both headless and non-headless opts.add_argument('--no-sandbox') opts.add_argument('--disable-dev-shm-usage') opts.add_argument('--disable-blink-features=AutomationControlled') # Only add --incognito flag when incognito=True and user_data_dir is not provided if incognito and not user_data_dir: opts.add_argument('--incognito') opts.add_argument(f'--window-size={window_size}') # Only disable cache when user_data_dir is NOT provided (to allow session persistence) # When user_data_dir is provided, cache must be enabled for cookies and session storage to persist if not user_data_dir: opts.add_argument('--disable-cache') opts.add_argument('--disable-application-cache') opts.add_argument('--disk-cache-size=0') opts.add_argument('--no-first-run') opts.add_argument('--no-default-browser-check') opts.add_argument('--disable-extensions') opts.add_argument('--disable-plugins') # Prefer an existing Chrome binary if present for path in _detect_chrome_binary_paths(): opts.binary_location = path break if user_data_dir: opts.add_argument('--user-data-dir=' + user_data_dir) return opts def get_uc_driver(headless=True, user_data_dir=None, prefs=None, window_size='1366,768', retries=2, logger=None, incognito=True): """Get undetected Chrome driver with automatic version matching and retry logic. Creates fresh ChromeOptions for each attempt to avoid reuse errors. Automatically matches ChromeDriver version to installed Chrome version. Args: headless: If True, run Chrome in headless mode (no visible browser window) user_data_dir: Optional user data directory path prefs: Dictionary of Chrome preferences (download settings, etc.) window_size: Window size string like '1920,1080' retries: Number of retry attempts (unused, kept for backward compatibility) logger: Logger instance for logging messages incognito: If True, run Chrome in incognito/private mode (default True for backward compatibility) """ # Set environment variable to force headless mode if headless: os.environ['CHROME_HEADLESS'] = '1' os.environ['DISPLAY'] = ':99' # For Linux systems major = get_installed_chrome_major_version() attempt = 0 last_err = None connection_error_detected = False if logger: if major: logger.info(f"🔍 Detected Chrome major version: {major}") logger.info(f"🔧 Headless mode: {'ENABLED' if headless else 'DISABLED'}") if logger and headless: logger.info("📌 Chrome will run in headless mode (no visible browser window)") # Ensure cache directory exists before attempting driver creation cache_dir = ensure_uc_cache_directory() if logger and cache_dir: logger.debug(f"✅ Cache directory ensured: {cache_dir}") elif logger: logger.warning("⚠️ Could not ensure cache directory exists, continuing anyway") # Build version ring: prefer exact match, then try -1, never try +1 ring = [] if major: # Try exact version first (most reliable) ring.append(major) # Try -1 as fallback (older compatible version) if major > 1: ring.append(major - 1) else: # If version detection fails, let uc auto-manage ring.append(None) # Track if we've seen version mismatch errors (driver too new for Chrome) version_mismatch_detected = False detected_chrome_version = None for vm in ring: attempt += 1 try: if logger: logger.info(f"🧩 Starting Chrome (attempt {attempt}) with version_main={vm}") # Create fresh ChromeOptions for each attempt to avoid reuse errors opts = build_default_options(headless=headless, user_data_dir=user_data_dir, window_size=window_size, incognito=incognito) if prefs: opts.add_experimental_option('prefs', prefs) # Launch with specified version if vm: driver = uc.Chrome(options=opts, use_subprocess=True, version_main=vm) else: driver = uc.Chrome(options=opts, use_subprocess=True) # Validate driver file exists (if we can check) if cache_dir and not validate_driver_file(cache_dir, logger): if logger: logger.warning("⚠️ Driver file validation failed, but driver was created. Continuing...") # Validate session is working if not validate_driver_session(driver, logger): connection_error_detected = True try: driver.quit() except Exception: pass raise Exception("Driver session validation failed: connection refused or session lost") if logger: logger.info("✅ Chrome driver created and session validated successfully") return driver except Exception as e: last_err = e error_msg = str(e) # Check for connection errors if "Connection refused" in error_msg or "Failed to establish" in error_msg or "Connection refused" in str(type(e).__name__): connection_error_detected = True if logger: logger.warning(f"⚠️ Connection error detected (attempt {attempt}): {error_msg}") # Check if error indicates version mismatch (driver too new for Chrome) if "only supports Chrome version" in error_msg and "Current browser version" in error_msg: version_mismatch_detected = True # Extract the actual Chrome version from error message # Error format: "This version of ChromeDriver only supports Chrome version 142\nCurrent browser version is 141.0.7390.123" browser_version_match = re.search(r'Current browser version is (\d+)\.', error_msg) if browser_version_match: detected_chrome_version = int(browser_version_match.group(1)) if logger: logger.warning(f"⚠️ Version mismatch detected: Driver too new. Chrome is version {detected_chrome_version}") if logger: logger.warning(f"⚠️ Version mismatch error: {error_msg}") # If we detected the Chrome version from error, use it immediately if detected_chrome_version and detected_chrome_version != vm: if logger: logger.info(f"🔁 Breaking out of version ring to try detected Chrome version: {detected_chrome_version}") break # Exit the loop to try the detected version # Check for driver download/file errors if "No such file or directory" in error_msg or "Unable to obtain driver" in error_msg: if logger: logger.warning(f"⚠️ Driver download/file error (attempt {attempt}): {error_msg}") if logger: logger.warning(f"Chrome launch failed (version_main={vm}): {e}") # Only clear cache if we detected a version mismatch or connection error # Don't clear cache on every retry as it may interrupt downloads if version_mismatch_detected or connection_error_detected: if logger: logger.info("🧹 Clearing cache due to version mismatch or connection error...") clear_uc_cache() time.sleep(5) # Increased wait time to allow cache clearing and downloads to complete else: # Small delay before retry without clearing cache time.sleep(2) # If we detected version mismatch, try the detected Chrome version immediately if version_mismatch_detected and detected_chrome_version: try: if logger: logger.info(f"🔁 Trying detected Chrome version (version_main={detected_chrome_version}) due to version mismatch") # Create fresh ChromeOptions opts = build_default_options(headless=headless, user_data_dir=user_data_dir, window_size=window_size, incognito=incognito) if prefs: opts.add_experimental_option('prefs', prefs) # Clear cache before trying detected version clear_uc_cache() time.sleep(5) # Increased wait time driver = uc.Chrome(options=opts, use_subprocess=True, version_main=detected_chrome_version) # Validate session if not validate_driver_session(driver, logger): connection_error_detected = True try: driver.quit() except Exception: pass raise Exception("Driver session validation failed after version match") if logger: logger.info("✅ Chrome driver created with detected version and session validated") return driver except Exception as e: if logger: logger.warning(f"Chrome launch failed with version_main={detected_chrome_version}: {e}") last_err = e error_msg = str(e) if "Connection refused" in error_msg or "Failed to establish" in error_msg: connection_error_detected = True # If we detected version mismatch but couldn't extract version, try major-1 and major-2 if version_mismatch_detected and major and major > 2: for fallback_version in [major - 1, major - 2]: if fallback_version <= 0: continue try: if logger: logger.info(f"🔁 Trying older ChromeDriver version (version_main={fallback_version}) due to version mismatch") # Create fresh ChromeOptions opts = build_default_options(headless=headless, user_data_dir=user_data_dir, window_size=window_size, incognito=incognito) if prefs: opts.add_experimental_option('prefs', prefs) # Clear cache before fallback attempt clear_uc_cache() time.sleep(5) # Increased wait time driver = uc.Chrome(options=opts, use_subprocess=True, version_main=fallback_version) # Validate session if not validate_driver_session(driver, logger): connection_error_detected = True try: driver.quit() except Exception: pass raise Exception("Driver session validation failed with fallback version") if logger: logger.info("✅ Chrome driver created with fallback version and session validated") return driver except Exception as e: if logger: logger.warning(f"Chrome launch failed with version_main={fallback_version}: {e}") last_err = e error_msg = str(e) if "Connection refused" in error_msg or "Failed to establish" in error_msg: connection_error_detected = True # Final fallback: only use auto-detection if we haven't detected a version mismatch # If version mismatch was detected, we should have tried the correct version already if not version_mismatch_detected: try: if logger: logger.info('🔁 Final fallback: launching uc.Chrome() with auto-version detection') # Create fresh ChromeOptions for final fallback opts = build_default_options(headless=headless, user_data_dir=user_data_dir, window_size=window_size, incognito=incognito) if prefs: opts.add_experimental_option('prefs', prefs) # Clear cache before final attempt if we had connection errors if connection_error_detected: if logger: logger.info("🧹 Clearing cache before final fallback attempt...") clear_uc_cache() time.sleep(5) # Increased wait time # Let undetected_chromedriver auto-detect and download the correct version driver = uc.Chrome(options=opts, use_subprocess=True) # Validate session if not validate_driver_session(driver, logger): connection_error_detected = True try: driver.quit() except Exception: pass raise Exception("Driver session validation failed in final fallback") if logger: logger.info("✅ Chrome driver created with auto-detection and session validated") return driver except Exception as e: if logger: logger.error(f"❌ Final Chrome launch failed: {e}") raise last_err or e else: # If we detected version mismatch but all attempts failed, raise the last error if logger: logger.error(f"❌ All Chrome launch attempts failed after version mismatch detection") raise last_err or Exception("Failed to launch Chrome after version mismatch detection") # ----------------------------------------------------------------------------- # SERVER CLIENT # ----------------------------------------------------------------------------- class ServerClient: """Handles all server communication for the automation script""" def __init__(self, api_base_url: str, user_id: str, job_id: str, logger=None): self.api_base_url = api_base_url.rstrip('/') self.user_id = user_id self.job_id = job_id self.logger = logger or logging.getLogger("ServerClient") if requests is None: self.logger.warning("⚠️ 'requests' library not available. Server communication disabled.") self.enabled = False else: self.enabled = True def upload_log_file(self, log_file_path: str, job_data: dict) -> str: """Upload log file to server via POST /api/local-exe/jobs/:jobId/upload-log""" if not self.enabled: self.logger.warning("⚠️ Server communication disabled. Skipping log file upload.") return None if not os.path.exists(log_file_path): self.logger.warning(f"⚠️ Log file not found: {log_file_path}") return None try: url = f"{self.api_base_url}/api/local-exe/jobs/{self.job_id}/upload-log" with open(log_file_path, 'rb') as f: files = {'logFile': (os.path.basename(log_file_path), f, 'text/plain')} data = { 'userId': job_data.get('userId', self.user_id), 'serviceId': job_data.get('serviceId', 'egm-download'), 'serviceName': job_data.get('serviceName', 'EGM Download'), 'inputFileName': job_data.get('inputFileName', 'input.csv'), 'inputFilePath': job_data.get('inputFilePath', ''), 'creditsUsed': str(job_data.get('creditsUsed', 0)) } response = requests.post(url, files=files, data=data, timeout=30) response.raise_for_status() result = response.json() if result.get('success') and result.get('logFilePath'): self.logger.info(f"✅ Log file uploaded: {result['logFilePath']}") return result['logFilePath'] else: self.logger.warning(f"⚠️ Log upload failed: {result.get('error', 'Unknown error')}") return None except requests.exceptions.RequestException as e: self.logger.warning(f"⚠️ Failed to upload log file: {e}") return None except Exception as e: self.logger.warning(f"⚠️ Unexpected error uploading log file: {e}") return None def upload_log_buffer(self, log_text: str, file_name: str, job_data: dict) -> str: if not self.enabled: self.logger.warning("⚠️ Server communication disabled. Skipping log buffer upload.") return None try: import io url = f"{self.api_base_url}/api/local-exe/jobs/{self.job_id}/upload-log" files = { 'logFile': (file_name, io.BytesIO(log_text.encode('utf-8')), 'text/plain') } data = { 'userId': job_data.get('userId', self.user_id), 'serviceId': job_data.get('serviceId', 'egm-download'), 'serviceName': job_data.get('serviceName', 'EGM Download'), 'inputFileName': job_data.get('inputFileName', 'input.csv'), 'inputFilePath': job_data.get('inputFilePath', ''), 'creditsUsed': str(job_data.get('creditsUsed', 0)) } response = requests.post(url, files=files, data=data, timeout=30) response.raise_for_status() result = response.json() return result.get('logFilePath') if result.get('success') else None except Exception as e: self.logger.error(f"❌ Failed to upload log buffer: {e}") return None def upload_results_csv(self, csv_file_path: str, job_data: dict) -> str: """Upload results CSV file to server via POST /api/local-exe/jobs/:jobId/upload-log""" if not self.enabled: self.logger.warning("⚠️ Server communication disabled. Skipping CSV upload.") return None if not os.path.exists(csv_file_path): self.logger.warning(f"⚠️ CSV file not found: {csv_file_path}") return None try: url = f"{self.api_base_url}/api/local-exe/jobs/{self.job_id}/upload-log" with open(csv_file_path, 'rb') as f: files = {'logFile': (os.path.basename(csv_file_path), f, 'text/csv')} data = { 'userId': job_data.get('userId', self.user_id), 'serviceId': job_data.get('serviceId', 'egm-download'), 'serviceName': job_data.get('serviceName', 'EGM Download'), 'inputFileName': job_data.get('inputFileName', 'input.csv'), 'inputFilePath': job_data.get('inputFilePath', ''), 'creditsUsed': str(job_data.get('creditsUsed', 0)) } response = requests.post(url, files=files, data=data, timeout=30) response.raise_for_status() result = response.json() if result.get('success') and result.get('logFilePath'): self.logger.info(f"✅ Results CSV uploaded: {result['logFilePath']}") return result['logFilePath'] else: self.logger.warning(f"⚠️ CSV upload failed: {result.get('error', 'Unknown error')}") return None except requests.exceptions.RequestException as e: self.logger.warning(f"⚠️ Failed to upload CSV file: {e}") return None except Exception as e: self.logger.warning(f"⚠️ Unexpected error uploading CSV file: {e}") return None def report_job_result(self, result_files: list, output_directory: str, successful_count: int = None) -> bool: """Report job completion to server via POST /api/local-exe/jobs/:jobId/complete""" if not self.enabled: self.logger.warning("⚠️ Server communication disabled. Skipping job result reporting.") return False try: url = f"{self.api_base_url}/api/local-exe/jobs/{self.job_id}/complete" payload = { 'success': True, 'resultFiles': result_files, 'outputFiles': [], 'outputDirectory': output_directory } if successful_count is not None: payload['successfulCount'] = successful_count response = requests.post(url, json=payload, timeout=30) response.raise_for_status() result = response.json() if result.get('success'): self.logger.info("✅ Job result reported successfully to server") return True else: self.logger.warning(f"⚠️ Job result reporting failed: {result.get('error', 'Unknown error')}") return False except requests.exceptions.RequestException as e: self.logger.warning(f"⚠️ Failed to report job result: {e}") return False except Exception as e: self.logger.warning(f"⚠️ Unexpected error reporting job result: {e}") return False # ----------------------------------------------------------------------------- # CONFIGURATION # ----------------------------------------------------------------------------- URL = "https://customs.gov.bd/portal/services/billTracking/billTracking.jsf" WAIT_TIMEOUT = 25 MAX_ATTEMPTS = 3 MAX_ROW_RETRIES = 3 # Early execution logging at module level print("🔵 [EGM_DOWNLOAD] Script execution started (module level)", flush=True) print(f"🔵 [EGM_DOWNLOAD] sys.argv: {sys.argv}", flush=True) print(f"🔵 [EGM_DOWNLOAD] __name__: {__name__}", flush=True) if len(sys.argv) < 3: error_msg = "❌ Usage: python3 egm_download.py [job_id] [api_base_url] [user_id] [credits_used]" print(f"❌ [EGM_DOWNLOAD] {error_msg}", file=sys.stderr, flush=True) print(error_msg, file=sys.stderr) sys.exit(1) INPUT_FILE = sys.argv[1] BASE_OUTPUT_DIR = sys.argv[2] JOB_ID = sys.argv[3] if len(sys.argv) > 3 else "unknown" print(f"🔵 [EGM_DOWNLOAD] Parsed arguments: INPUT_FILE={INPUT_FILE}, BASE_OUTPUT_DIR={BASE_OUTPUT_DIR}, JOB_ID={JOB_ID}", flush=True) # Optional server communication parameters API_BASE_URL = sys.argv[4] if len(sys.argv) > 4 else None USER_ID = sys.argv[5] if len(sys.argv) > 5 else None CREDITS_USED = float(sys.argv[6]) if len(sys.argv) > 6 and sys.argv[6] else 0.0 # Validate CSV file extension if not INPUT_FILE.lower().endswith('.csv'): print("=" * 80) print("⚠️ WARNING: Invalid File Type") print("=" * 80) print(f"📁 File provided: {INPUT_FILE}") print("") print("ℹ️ This automation only supports CSV (Comma-Separated Values) files.") print(" Please ensure your input file has a .csv extension.") print("") print("=" * 80) sys.exit(1) # Use isolated service-specific output directory provided by the server OUTPUT_DIR = os.path.abspath(BASE_OUTPUT_DIR) os.makedirs(OUTPUT_DIR, exist_ok=True) # Use OUTPUT_DIR directly instead of creating pdfs subfolder PDFS_DIR = OUTPUT_DIR SCREENSHOT_DIR = os.path.join(OUTPUT_DIR, "screenshots") FAILED_CSV = os.path.join(OUTPUT_DIR, f"failed_rows_{JOB_ID}.csv") # ----------------------------------------------------------------------------- # LOGGING # ----------------------------------------------------------------------------- from io import StringIO log_file_path = '' log_buffer = StringIO() buffer_handler = logging.StreamHandler(log_buffer) buffer_handler.setLevel(logging.INFO) buffer_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.INFO) console_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") console_handler.setFormatter(console_formatter) logger = logging.getLogger("EgmDownload-" + JOB_ID) logger.setLevel(logging.INFO) logger.addHandler(console_handler) logger.addHandler(buffer_handler) logger.info("📁 Input: %s", INPUT_FILE) logger.info("📂 Output: %s", OUTPUT_DIR) logger.info("🆔 Job ID: %s", JOB_ID) # ----------------------------------------------------------------------------- # ENVIRONMENT CHECK # ----------------------------------------------------------------------------- FFMPEG_PATH = shutil.which("ffmpeg") if FFMPEG_PATH: logger.info("🎧 FFmpeg found at: %s", FFMPEG_PATH) else: logger.warning("⚠️ FFmpeg not found — audio challenge may fail.") # ----------------------------------------------------------------------------- # DEBUG SAVE HELPERS # ----------------------------------------------------------------------------- def save_debug(driver, name): """Save screenshot and HTML for debugging""" os.makedirs(SCREENSHOT_DIR, exist_ok=True) try: png = os.path.join(SCREENSHOT_DIR, f"{name}.png") driver.save_screenshot(png) logger.info("📸 Screenshot saved: %s", png) except Exception as e: logger.error("❌ Screenshot failed: %s", e) try: html = os.path.join(SCREENSHOT_DIR, f"{name}.html") with open(html, "w", encoding="utf-8") as f: f.write(driver.page_source) logger.info("📝 HTML saved: %s", html) except Exception as e: logger.error("❌ HTML save failed: %s", e) # ----------------------------------------------------------------------------- # DRIVER SETUP # ----------------------------------------------------------------------------- def setup_driver(): """Launch Electron browser (headless mode)""" # Use JOB_ID from global scope (set from sys.argv[3]) driver = ElectronBrowser.create(job_id=JOB_ID, headless=True, logger=logger) logger.info("🧩 Electron browser launched") return driver # ----------------------------------------------------------------------------- # RECAPTCHA SOLVER # ----------------------------------------------------------------------------- class RecaptchaSolver: def __init__(self, driver): self.driver = driver def is_solved(self): try: self.driver.switch_to.default_content() iframe = self.driver.find_element(By.CSS_SELECTOR, "iframe[title*='reCAPTCHA']") self.driver.switch_to.frame(iframe) anchor = self.driver.find_element(By.ID, "recaptcha-anchor") checked = anchor.get_attribute("aria-checked") == "true" or \ "recaptcha-checkbox-checked" in anchor.get_attribute("class") self.driver.switch_to.default_content() return checked except Exception: self.driver.switch_to.default_content() return False def _solve_audio_challenge(self, wait): """Handle audio challenge via SpeechRecognition""" attempt = 0 while True: attempt += 1 try: audio_src = wait.until(EC.presence_of_element_located((By.ID, "audio-source"))).get_attribute("src") logger.info("🎵 Audio challenge %d: %s", attempt, audio_src) mp3 = os.path.join(tempfile.gettempdir(), f"captcha_{attempt}.mp3") wav = os.path.join(tempfile.gettempdir(), f"captcha_{attempt}.wav") urllib.request.urlretrieve(audio_src, mp3) pydub.AudioSegment.from_mp3(mp3).export(wav, format="wav") if not SR_AVAILABLE: raise RuntimeError("SpeechRecognition not available") recog = sr.Recognizer() with sr.AudioFile(wav) as s: audio = recog.record(s) text = recog.recognize_google(audio) logger.info("🗣️ Recognized text: %s", text) input_box = wait.until(EC.presence_of_element_located((By.ID, "audio-response"))) input_box.clear() input_box.send_keys(text.lower()) verify = wait.until(EC.element_to_be_clickable((By.ID, "recaptcha-verify-button"))) self.driver.execute_script("arguments[0].click();", verify) time.sleep(3) error_msgs = self.driver.find_elements(By.CSS_SELECTOR, ".rc-audiochallenge-error-message") if any("Multiple correct solutions required" in e.text for e in error_msgs): logger.info("🔁 Multiple audio required — solving next challenge...") continue self.driver.switch_to.default_content() if self.is_solved(): logger.info("✅ reCAPTCHA solved via audio after %d attempt(s)", attempt) return True except Exception as e: logger.error("❌ Audio challenge failed: %s", e) save_debug(self.driver, f"audio_fail_{attempt}") break return False def solveCaptcha(self): wait = WebDriverWait(self.driver, WAIT_TIMEOUT) self.driver.switch_to.default_content() # Click checkbox try: iframe = self.driver.find_element(By.CSS_SELECTOR, "iframe[title*='reCAPTCHA']") self.driver.switch_to.frame(iframe) box = self.driver.find_element(By.CLASS_NAME, "recaptcha-checkbox-border") box.click() time.sleep(2) self.driver.switch_to.default_content() if self.is_solved(): logger.info("✅ reCAPTCHA checkbox solved.") return except Exception: self.driver.switch_to.default_content() # Fallback to audio logger.info("🎧 Fallback to audio challenge...") try: challenge_iframe = wait.until(EC.presence_of_element_located( (By.CSS_SELECTOR, "iframe[title*='recaptcha challenge']") )) self.driver.switch_to.frame(challenge_iframe) audio_btn = wait.until(EC.element_to_be_clickable((By.ID, "recaptcha-audio-button"))) self.driver.execute_script("arguments[0].click();", audio_btn) time.sleep(2) self._solve_audio_challenge(wait) except Exception as e: logger.error("❌ Could not launch audio challenge: %s", e) save_debug(self.driver, "audio_init_fail") # ----------------------------------------------------------------------------- # BILL FETCH # ----------------------------------------------------------------------------- def fetch_bill_status(driver, solver, office, serial, number, year): wait = WebDriverWait(driver, WAIT_TIMEOUT) driver.get(URL) Select(wait.until(EC.presence_of_element_located((By.ID, "formAct:customOfficeCode")))).select_by_value(str(office)) driver.find_element(By.ID, "formAct:billEntrySerial").send_keys(str(serial)) driver.find_element(By.ID, "formAct:billEntryNumber").send_keys(str(number)) Select(driver.find_element(By.ID, "formAct:billEntryYear")).select_by_value(str(year)) for attempt in range(1, MAX_ATTEMPTS + 1): try: logger.info("🔐 Solving reCAPTCHA (Attempt %d)…", attempt) solver.solveCaptcha() break except Exception as e: logger.warning("Attempt %d failed: %s", attempt, e) save_debug(driver, f"captcha_fail_{serial}_{number}_{year}_try{attempt}") if attempt == MAX_ATTEMPTS: raise btn = wait.until(EC.element_to_be_clickable((By.XPATH, "//span[text()='Retrieve B/E Status']"))) driver.execute_script("arguments[0].click();", btn) logger.info("🖱️ Clicked 'Retrieve B/E Status'") time.sleep(3) # OUTPUT_DIR is already created above, no need to create pdfs subfolder pdf_path = os.path.join(PDFS_DIR, f"{office}_{serial}_{number}_{year}.pdf") tmp_png = os.path.join(tempfile.gettempdir(), f"{office}_{serial}_{number}_{year}.png") driver.save_screenshot(tmp_png) Image.open(tmp_png).convert("RGB").save(pdf_path, "PDF", resolution=100.0) logger.info("📄 PDF saved: %s", pdf_path) try: back_btn = wait.until(EC.element_to_be_clickable((By.XPATH, "//span[text()='BACK TO MAIN PAGE']"))) driver.execute_script("arguments[0].click();", back_btn) time.sleep(2) except Exception: pass return pdf_path # ----------------------------------------------------------------------------- # MAIN # ----------------------------------------------------------------------------- def main(): # Early execution logging print("🔵 [EGM_DOWNLOAD] Script execution started", flush=True) print(f"🔵 [EGM_DOWNLOAD] sys.argv: {sys.argv}", flush=True) print(f"🔵 [EGM_DOWNLOAD] __name__: {__name__}", flush=True) print(f"🔵 [EGM_DOWNLOAD] Python version: {sys.version}", flush=True) print(f"🔵 [EGM_DOWNLOAD] INPUT_FILE: {INPUT_FILE}", flush=True) print(f"🔵 [EGM_DOWNLOAD] OUTPUT_DIR: {OUTPUT_DIR}", flush=True) print(f"🔵 [EGM_DOWNLOAD] JOB_ID: {JOB_ID}", flush=True) file_exists = os.path.exists(INPUT_FILE) print(f"🔵 [EGM_DOWNLOAD] File path check: {INPUT_FILE} exists: {file_exists}", flush=True) if not file_exists: error_msg = f"❌ Missing input file: {INPUT_FILE}" print(f"❌ [EGM_DOWNLOAD] {error_msg}", file=sys.stderr, flush=True) logger.error(error_msg) return 1 df = pd.read_csv(INPUT_FILE) generated, failed = [], [] all_results = [] # Track all results for main CSV jobs = [ (row["customOfficeCode"], row["billEntrySerial"], row["billEntryNumber"], row["billEntryYear"]) for _, row in df.iterrows() ] total_jobs = len(jobs) initial_total = total_jobs success_count = 0 failed_count = 0 processed_count = 0 logger.info("\n%s", "="*60) logger.info("🚀 Starting automation for %d entries…", total_jobs) logger.info("%s\n", "="*60) driver = setup_driver() solver = RecaptchaSolver(driver) while jobs: # Check for stop signal stop_file = os.path.join(OUTPUT_DIR, f"stop_{JOB_ID}.txt") if os.path.exists(stop_file): logger.info("Stop signal received, stopping automation...") driver.quit() sys.exit(130) office, serial, number, year = jobs.pop(0) processed_count += 1 items_remaining = len(jobs) items_completed = processed_count - 1 logger.info("📊 Progress: %d/%d processed (Remaining: %d)", processed_count, initial_total, items_remaining) logger.info("\n%s\nProcessing %s-%s-%s (Attempt 1)\n%s", "="*50, serial, number, year, "="*50) attempt = 1 while attempt <= MAX_ROW_RETRIES: if attempt > 1: logger.info("\n%s\nProcessing %s-%s-%s (Attempt %d)\n%s", "="*50, serial, number, year, attempt, "="*50) try: pdf = fetch_bill_status(driver, solver, office, serial, number, year) if pdf: generated.append(pdf) success_count += 1 # Record successful result all_results.append({ "customOfficeCode": office, "billEntrySerial": serial, "billEntryNumber": number, "billEntryYear": year, "status": "success", "error": "", "timestamp": pd.Timestamp.now().isoformat() }) logger.info("✓ Successfully processed entry %d/%d: %s-%s-%s", processed_count, initial_total, serial, number, year) logger.info("📥 Downloads complete: %d successful, %d failed", success_count, failed_count) break except Exception as e: error_msg = str(e) logger.error("❌ Error: %s", error_msg) save_debug(driver, f"fail_{serial}_{number}_{year}_try{attempt}") if attempt == MAX_ROW_RETRIES: failed.append((office, serial, number, year)) failed_count += 1 # Record failed result all_results.append({ "customOfficeCode": office, "billEntrySerial": serial, "billEntryNumber": number, "billEntryYear": year, "status": "error", "error": error_msg, "timestamp": pd.Timestamp.now().isoformat() }) logger.error("✗ Failed to process entry %d/%d: %s-%s-%s", processed_count, initial_total, serial, number, year) logger.info("📥 Downloads complete: %d successful, %d failed", success_count, failed_count) else: logger.info("🔁 Retrying row (%d/%d)...", attempt, MAX_ROW_RETRIES) try: driver.quit() except: pass driver = setup_driver() solver = RecaptchaSolver(driver) attempt += 1 if attempt > MAX_ROW_RETRIES and (office, serial, number, year) in failed: jobs.append((office, serial, number, year)) logger.info("⏳ Row requeued: %s-%s-%s", serial, number, year) driver.quit() # Generate main result CSV with all results (success + failed) result_csv_path = os.path.join(OUTPUT_DIR, f"egm_download_result_{JOB_ID}.csv") logger.info("ℹ️ Results CSV generation disabled") # if all_results: # result_df = pd.DataFrame(all_results) # # logger.info("ℹ️ Results CSV generation disabled") # else: # logger.warning("⚠️ No results to save to CSV file") # Print summary logger.info("\n%s", "="*60) logger.info("PROCESSING SUMMARY") logger.info("%s", "="*60) logger.info("Total entries: %d", initial_total) logger.info("✓ Successful downloads: %d", success_count) logger.info("✗ Failed downloads: %d", failed_count) logger.info("📥 Total downloads completed: %d/%d", success_count + failed_count, initial_total) logger.info("\n📋 Detailed log saved to: %s", log_file_path) logger.info("%s\n", "="*60) # Save failed rows CSV if failed: logger.warning("⚠️ %d rows failed after retries.", len(failed)) # Upload files to server if server parameters provided uploaded_log_path = None uploaded_csv_path = None if API_BASE_URL and USER_ID: try: server_client = ServerClient(API_BASE_URL, USER_ID, JOB_ID, logger) if server_client.enabled: # Prepare job data for uploads job_data = { 'userId': USER_ID, 'serviceId': 'egm-download', 'serviceName': 'EGM Download', 'inputFileName': os.path.basename(INPUT_FILE), 'inputFilePath': INPUT_FILE, 'creditsUsed': CREDITS_USED } uploaded_log_path = None log_text = log_buffer.getvalue() if log_text: uploaded_log_path = server_client.upload_log_buffer(log_text, f'egm_download_{JOB_ID}.log', job_data) # Upload results CSV if os.path.exists(result_csv_path): uploaded_csv_path = server_client.upload_results_csv(result_csv_path, job_data) # Report job result result_files = [] if uploaded_csv_path: result_files.append(uploaded_csv_path) elif uploaded_log_path: result_files.append(uploaded_log_path) if result_files: server_client.report_job_result( result_files, OUTPUT_DIR, successful_count=success_count ) except Exception as e: logger.warning("⚠️ Error uploading files to server: %s", e) # Continue even if upload fails logger.info("🎉 Completed successfully!") return 0 if __name__ == "__main__": try: main() print(f"🔵 [EGM_DOWNLOAD] Script completed successfully", flush=True) except Exception as e: print(f"❌ [EGM_DOWNLOAD] Fatal error: {e}", file=sys.stderr, flush=True) import traceback traceback.print_exc(file=sys.stderr) sys.exit(1)