#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ CTG Port Authority Tracking Automation (Electron WebContents) Author: Izaz Ahamed ------------------------------------------------------------- ✅ Runs headless with Electron WebContents ✅ Auto PDF export per container ✅ Handles popups, alerts, and summary reports ✅ Works on Linux VPS (Ubuntu) with Chrome installed """ import os, sys, time, logging, base64, tempfile, shutil import pandas as pd from datetime import datetime from PyPDF2 import PdfMerger # 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 try: import requests except ImportError: requests = None print("⚠️ Warning: 'requests' library not found. Server communication will be disabled.") print(" Install with: pip install requests") 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: # Get file metadata file_size_bytes = os.path.getsize(log_file_path) file_size_kb = file_size_bytes / 1024 file_name = os.path.basename(log_file_path) self.logger.info("📤 Starting log file upload...") self.logger.info(f"📊 File metadata:") self.logger.info(f" - File path: {log_file_path}") self.logger.info(f" - File name: {file_name}") self.logger.info(f" - File size: {file_size_bytes} bytes ({file_size_kb:.2f} KB)") url = f"{self.api_base_url}/api/local-exe/jobs/{self.job_id}/upload-log" self.logger.info(f"🌐 Upload URL: {url}") # Log job data being sent self.logger.info("📊 Job data being sent:") self.logger.info(f" - User ID: {job_data.get('userId', self.user_id)}") self.logger.info(f" - Service ID: {job_data.get('serviceId', 'ctg-port-tracking')}") self.logger.info(f" - Service Name: {job_data.get('serviceName', 'CTG Port Tracking')}") self.logger.info(f" - Input File Name: {job_data.get('inputFileName', 'input.csv')}") self.logger.info(f" - Credits Used: {job_data.get('creditsUsed', 0)}") self.logger.info("⚡ Opening file for reading...") with open(log_file_path, 'rb') as f: files = {'logFile': (file_name, f, 'text/plain')} data = { 'userId': job_data.get('userId', self.user_id), 'serviceId': job_data.get('serviceId', 'ctg-port-tracking'), 'serviceName': job_data.get('serviceName', 'CTG Port Tracking'), 'inputFileName': job_data.get('inputFileName', 'input.csv'), 'inputFilePath': job_data.get('inputFilePath', ''), 'creditsUsed': str(job_data.get('creditsUsed', 0)) } self.logger.info("⚡ Sending POST request to server...") request_start_time = time.time() response = requests.post(url, files=files, data=data, timeout=30) request_duration = time.time() - request_start_time self.logger.info(f"⏱️ Request completed in {request_duration:.2f} seconds") self.logger.info(f"📊 HTTP Status Code: {response.status_code}") response.raise_for_status() result = response.json() self.logger.info("📊 Response received:") self.logger.info(f" - Success: {result.get('success', False)}") if result.get('logFilePath'): self.logger.info(f" - Server file path: {result.get('logFilePath')}") if result.get('error'): self.logger.warning(f" - Error: {result.get('error')}") if result.get('success') and result.get('logFilePath'): self.logger.info(f"✅ Log file uploaded successfully: {result['logFilePath']}") self.logger.info(f"⏱️ Total upload time: {request_duration:.2f} seconds") 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.error(f"❌ Failed to upload log file: {e}") self.logger.error(f" Error type: {type(e).__name__}") return None except Exception as e: self.logger.error(f"❌ Unexpected error uploading log file: {e}") self.logger.error(f" Error type: {type(e).__name__}") import traceback self.logger.debug(traceback.format_exc()) 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', 'ctg-port-tracking'), 'serviceName': job_data.get('serviceName', 'CTG Port Tracking'), '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_buffer(self, csv_bytes: bytes, file_name: str, job_data: dict) -> str: if not self.enabled: self.logger.warning("⚠️ Server communication disabled. Skipping CSV upload.") return None try: url = f"{self.api_base_url}/api/local-exe/jobs/{self.job_id}/upload-log" files = { 'logFile': (file_name, csv_bytes, 'text/csv') } data = { 'userId': job_data.get('userId', self.user_id), 'serviceId': job_data.get('serviceId', 'ctg-port-tracking'), 'serviceName': job_data.get('serviceName', 'CTG Port Tracking'), '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'): return result['logFilePath'] return None except Exception as e: self.logger.error(f"❌ Failed to upload CSV buffer: {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: self.logger.info("📤 Starting job result reporting...") url = f"{self.api_base_url}/api/local-exe/jobs/{self.job_id}/complete" self.logger.info(f"🌐 Report URL: {url}") payload = { 'success': True, 'resultFiles': result_files, 'outputFiles': [], 'outputDirectory': output_directory } if successful_count is not None: payload['successfulCount'] = successful_count self.logger.info("📊 Payload being sent:") self.logger.info(f" - Success: {payload.get('success')}") self.logger.info(f" - Result files: {len(payload.get('resultFiles', []))} file(s)") for i, file_path in enumerate(payload.get('resultFiles', []), 1): self.logger.info(f" {i}. {file_path}") self.logger.info(f" - Output directory: {payload.get('outputDirectory')}") if successful_count is not None: self.logger.info(f" - Successful count: {payload.get('successfulCount')}") self.logger.info("⚡ Sending POST request to server...") request_start_time = time.time() response = requests.post(url, json=payload, timeout=30) request_duration = time.time() - request_start_time self.logger.info(f"⏱️ Request completed in {request_duration:.2f} seconds") self.logger.info(f"📊 HTTP Status Code: {response.status_code}") response.raise_for_status() result = response.json() self.logger.info("📊 Response received:") self.logger.info(f" - Success: {result.get('success', False)}") if result.get('error'): self.logger.warning(f" - Error: {result.get('error')}") if result.get('success'): self.logger.info("✅ Job result reported successfully to server") self.logger.info(f"⏱️ Total reporting time: {request_duration:.2f} seconds") 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.error(f"❌ Failed to report job result: {e}") self.logger.error(f" Error type: {type(e).__name__}") return False except Exception as e: self.logger.error(f"❌ Unexpected error reporting job result: {e}") self.logger.error(f" Error type: {type(e).__name__}") import traceback self.logger.debug(traceback.format_exc()) return False class CtgPortTrackingAutomation: def __init__(self, headless=True, output_dir='results', job_id=None, server_client=None): self.headless = headless # Use isolated service-specific output directory provided by the server self.output_dir = os.path.normpath(os.path.abspath(output_dir)) os.makedirs(self.output_dir, exist_ok=True) self.job_id = job_id self.driver = None self.wait = None self.results = [] self.result_files = [] self.base_url = "https://cpatos.gov.bd/pcs/" self.temp_profile = None self.server_client = server_client self.input_file_path = None self.credits_used = 0.0 self.setup_logging(job_id) # Log the resolved absolute output directory path print(f"📁 Output directory (absolute path): {self.output_dir}") # ------------------------------------------------------------- def setup_logging(self, job_id=None): from io import StringIO self.log_buffer = StringIO() buffer_handler = logging.StreamHandler(self.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) logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[console_handler, buffer_handler], force=True ) self.logger = logging.getLogger(f"CtgPortTracking-{job_id}") self.log_file_path = '' # ------------------------------------------------------------- def setup_driver(self): self.logger.info("🔧 Launching Electron browser (headless mode)...") self.temp_profile = None if not self.headless: self.logger.info("🗂️ Visible mode requested (window will be shown)") else: self.logger.info("🕶️ Headless mode active - browser window will be hidden") try: # Get job_id from args (should be 3rd argument) job_id = self.job_id if hasattr(self, 'job_id') else 'unknown' self.driver = ElectronBrowser.create(job_id=job_id, headless=self.headless, logger=self.logger) self.wait = ElectronWait(self.driver, 20) # Ensure output directory exists using absolute path output_dir_abs = os.path.normpath(os.path.abspath(self.output_dir)) os.makedirs(output_dir_abs, exist_ok=True) self.logger.info(f"📁 Output directory created/verified (absolute path): {output_dir_abs}") # Use output_dir directly, no pdfs subfolder needed self.logger.info("✅ Electron browser started successfully") return True except Exception as e: self.logger.error(f"❌ Failed to launch Electron browser: {e}") return False # ------------------------------------------------------------- def _log_page_state(self, context=""): """Helper method to log current page state (URL, title, ready state). Args: context: Optional context string to include in log messages """ try: url = self.driver.current_url self.logger.info(f"🌐 {context}Current URL: {url}" if context else f"🌐 Current URL: {url}") except Exception as e: self.logger.warning(f"⚠️ Could not get URL: {e}") try: title = self.driver.title self.logger.info(f"📄 {context}Page title: {title}" if context else f"📄 Page title: {title}") except Exception as e: self.logger.warning(f"⚠️ Could not get page title: {e}") try: ready_state = self.driver.execute_script("return document.readyState") self.logger.info(f"📊 {context}Page ready state: {ready_state}" if context else f"📊 Page ready state: {ready_state}") except Exception as e: self.logger.warning(f"⚠️ Could not get page ready state: {e}") # ------------------------------------------------------------- def navigate_to_portal(self): try: # Log current URL before navigation try: current_url = self.driver.current_url self.logger.info(f"🌐 Current URL before navigation: {current_url}") except Exception: self.logger.info("🌐 No current URL (browser just started)") # Log target URL self.logger.info(f"🌐 Navigating to CTG Port Authority portal: {self.base_url}") nav_start_time = time.time() # Navigate to portal self.driver.get(self.base_url) # Wait for page load self.logger.info("⏱️ Waiting for page to load...") self.wait.until(EC.presence_of_element_located((By.TAG_NAME, "body"))) # Calculate navigation time nav_duration = time.time() - nav_start_time # Get final URL (in case of redirects) try: final_url = self.driver.current_url self.logger.info(f"🌐 Final URL after navigation: {final_url}") if final_url != self.base_url: self.logger.info(f"🔄 Redirect detected: {self.base_url} -> {final_url}") except Exception as e: self.logger.warning(f"⚠️ Could not get final URL: {e}") # Get page title try: page_title = self.driver.title self.logger.info(f"📄 Page title: {page_title}") except Exception as e: self.logger.warning(f"⚠️ Could not get page title: {e}") # Get page ready state try: ready_state = self.driver.execute_script("return document.readyState") self.logger.info(f"📊 Page ready state: {ready_state}") except Exception as e: self.logger.warning(f"⚠️ Could not get page ready state: {e}") self.logger.info(f"📍 Portal loaded successfully in {nav_duration:.2f} seconds") return True except Exception as e: self.logger.error(f"❌ Navigation failed: {e}") return False # ------------------------------------------------------------- def process_container(self, container_number, invoice_no, index): """Process a single container using direct POST request (optimized approach)""" temp_html_file = None try: self.logger.info(f"🔍 Processing container {index}: {container_number} (Invoice: {invoice_no})") # Check if requests library is available if requests is None: raise ImportError("requests library is required for direct POST requests. Install with: pip install requests") # Direct POST request to the endpoint post_url = "https://cpatos.gov.bd/pcs/index.php/Report/mySearchContainerLocation" self.logger.info(f"📤 Making direct POST request to: {post_url}") self.logger.info(f"📝 Container number: {container_number}") post_start = time.time() # Create a session to maintain cookies if needed session = requests.Session() session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9', 'Content-Type': 'application/x-www-form-urlencoded', 'Referer': 'https://cpatos.gov.bd/pcs/', 'Origin': 'https://cpatos.gov.bd' }) # Make POST request with form data form_data = { 'containerLocation': container_number } try: response = session.post(post_url, data=form_data, timeout=30, allow_redirects=True) post_duration = time.time() - post_start self.logger.info(f"⏱️ POST request completed in {post_duration:.2f} seconds") self.logger.info(f"📊 HTTP Status Code: {response.status_code}") # Check if request was successful response.raise_for_status() # Get HTML content html_content = response.text html_size = len(html_content) html_size_kb = html_size / 1024 self.logger.info(f"📊 HTML response size: {html_size} bytes ({html_size_kb:.2f} KB)") # Check if response contains error or empty content if html_size < 100: raise ValueError(f"Response too small ({html_size} bytes), likely an error or empty response") # Check for common error indicators in HTML if "error" in html_content.lower() and "container" in html_content.lower(): if "not found" in html_content.lower() or "invalid" in html_content.lower(): self.logger.warning(f"⚠️ Container may not be found or invalid: {container_number}") except requests.exceptions.RequestException as e: self.logger.error(f"❌ POST request failed: {e}") raise Exception(f"Failed to fetch container data: {e}") # Save HTML to temporary file temp_dir = tempfile.gettempdir() temp_html_file = os.path.join(temp_dir, f"ctg_port_{self.job_id}_{index}_{container_number}.html") temp_html_file_abs = os.path.normpath(os.path.abspath(temp_html_file)) self.logger.info(f"💾 Saving HTML to temporary file: {temp_html_file_abs}") try: # Write HTML content to file with UTF-8 encoding with open(temp_html_file_abs, 'w', encoding='utf-8') as f: f.write(html_content) self.logger.info("✅ HTML file saved successfully") except Exception as e: raise Exception(f"Failed to save HTML to temp file: {e}") # Load HTML file in browser using file:// URL # Convert path to file:// URL format (works on both Windows and Unix) # Windows: file:///C:/path/to/file.html # Unix: file:///path/to/file.html normalized_path = temp_html_file_abs.replace(os.sep, '/') if os.name == 'nt' and normalized_path[1] == ':': # Windows absolute path: C:/path -> file:///C:/path file_url = f"file:///{normalized_path}" else: # Unix absolute path: /path -> file:///path file_url = f"file://{normalized_path}" self.logger.info(f"🌐 Loading HTML in browser: {file_url}") load_start = time.time() try: self.driver.get(file_url) load_duration = time.time() - load_start self.logger.info(f"⏱️ HTML loaded in browser in {load_duration:.2f} seconds") except Exception as e: raise Exception(f"Failed to load HTML in browser: {e}") # Wait for page to be ready self.logger.info("⏱️ Waiting for page to be ready...") try: self.wait.until(EC.presence_of_element_located((By.TAG_NAME, "body"))) self.logger.info("✅ Page is ready") except Exception as e: self.logger.warning(f"⚠️ Timeout waiting for body element: {e}") # Continue anyway, page might still be usable # Small delay to ensure page is fully rendered time.sleep(1) # PDF generation pdf_filename = f"{index:03d} {invoice_no} {container_number}.pdf" pdf_path = os.path.join(self.output_dir, pdf_filename) pdf_path_abs = os.path.normpath(os.path.abspath(pdf_path)) self.logger.info(f"💾 Generating PDF: {pdf_filename}") self.logger.info(f"📁 PDF path (absolute): {pdf_path_abs}") pdf_start = time.time() try: pdf_data = self.driver.execute_cdp_cmd("Page.printToPDF", { "format": "A4", "printBackground": True, "marginTop": 0.4, "marginBottom": 0.4, "marginLeft": 0.4, "marginRight": 0.4 }) pdf_duration = time.time() - pdf_start self.logger.info(f"⏱️ PDF generation completed in {pdf_duration:.2f} seconds") except Exception as e: raise Exception(f"PDF generation failed: {e}") # Get PDF data size try: pdf_size_bytes = len(base64.b64decode(pdf_data["data"])) pdf_size_kb = pdf_size_bytes / 1024 self.logger.info(f"📊 PDF data size: {pdf_size_bytes} bytes ({pdf_size_kb:.2f} KB)") except Exception as e: self.logger.warning(f"⚠️ Could not calculate PDF size: {e}") # Write PDF to disk self.logger.info("⚡ Writing PDF file to disk...") try: with open(pdf_path_abs, "wb") as f: f.write(base64.b64decode(pdf_data["data"])) self.logger.info("✅ PDF file written successfully") except Exception as e: raise Exception(f"Failed to write PDF file: {e}") self.logger.info(f"✅ PDF saved: {pdf_filename}") self.logger.info(f"✅ PDF saved to (absolute path): {pdf_path_abs}") # Record success self.results.append({ "container_number": container_number, "invoice_no": invoice_no, "status": "success", "pdf_file": pdf_filename, "timestamp": datetime.now().isoformat() }) return pdf_filename except Exception as e: self.logger.error(f"❌ Error for {container_number} (Invoice: {invoice_no}): {e}") import traceback self.logger.debug(traceback.format_exc()) self.results.append({ "container_number": container_number, "invoice_no": invoice_no, "status": "error", "error": str(e), "timestamp": datetime.now().isoformat() }) return None finally: # Clean up temporary HTML file if temp_html_file and os.path.exists(temp_html_file): try: os.remove(temp_html_file) self.logger.debug(f"🗑️ Cleaned up temporary HTML file: {temp_html_file}") except Exception as e: self.logger.warning(f"⚠️ Failed to clean up temp file {temp_html_file}: {e}") # ------------------------------------------------------------- def read_container_file(self, file_path): try: ext = os.path.splitext(file_path)[1].lower() if ext != ".csv": error_msg = ( f"⚠️ WARNING: Invalid File Type\n" f"📁 File provided: {file_path}\n" f"📄 File extension: {ext}\n\n" f"ℹ️ This automation only supports CSV (Comma-Separated Values) files.\n" f" Please ensure your input file has a .csv extension.\n\n" ) raise ValueError(error_msg) df = pd.read_csv(file_path) # Check for required columns required_columns = ['container_number', 'invoice_no'] missing_columns = [col for col in required_columns if col not in df.columns] if missing_columns: error_msg = ( f"❌ Missing required columns: {', '.join(missing_columns)}\n" f"📋 Required columns: container_number, invoice_no\n" f"📋 Found columns: {', '.join(df.columns.tolist())}\n" ) raise ValueError(error_msg) # Read both columns and create list of dictionaries containers = [] for _, row in df.iterrows(): container_number = str(row['container_number']).strip() invoice_no = str(row['invoice_no']).strip() if container_number and invoice_no: containers.append({ "container_number": container_number, "invoice_no": invoice_no }) return containers except Exception as e: self.logger.error(f"❌ Failed to read file: {e}") return [] # ------------------------------------------------------------- def generate_results_csv(self): """Generate CSV bytes (no local file) with processing results""" if not self.results: self.logger.warning("No results to generate CSV") return None, None try: rows = [] for r in self.results: rows.append([ r.get("container_number", ""), r.get("invoice_no", ""), r.get("status", "error"), r.get("error", ""), r.get("timestamp", "") ]) import io, csv as _csv buf = io.StringIO() w = _csv.writer(buf) w.writerow(["container_number","invoice_no","status","error","timestamp"]) w.writerows(rows) file_name = f"ctg_port_tracking_results_{self.job_id}.csv" return file_name, buf.getvalue().encode('utf-8') except Exception as e: self.logger.error(f"❌ Failed to generate results CSV: {e}") return None, None # ------------------------------------------------------------- def process_all(self, containers): success, fail = [], [] total_containers = len(containers) success_count = 0 failed_count = 0 self.logger.info(f"\n{'='*60}") self.logger.info(f"Found {total_containers} containers to process") self.logger.info(f"{'='*60}\n") for i, container_data in enumerate(containers, 1): # Check for stop signal stop_file = os.path.join(self.output_dir, f"stop_{self.job_id}.txt") if os.path.exists(stop_file): self.logger.info("Stop signal received, stopping automation...") self.cleanup() sys.exit(130) # Log progress: items remaining items_remaining = total_containers - i + 1 self.logger.info(f"📊 Progress: {i}/{total_containers} (Remaining: {items_remaining})") container_number = container_data['container_number'] invoice_no = container_data['invoice_no'] pdf = self.process_container(container_number, invoice_no, i) if pdf: success.append(container_data) success_count += 1 self.logger.info(f"✓ Successfully processed container {i}/{total_containers}: {container_number} (Invoice: {invoice_no})") else: fail.append(container_data) failed_count += 1 self.logger.error(f"✗ Failed to process container {i}/{total_containers}: {container_number} (Invoice: {invoice_no})") self.logger.info(f"📥 Downloads complete: {success_count} successful, {failed_count} failed") time.sleep(2) file_name, csv_bytes = self.generate_results_csv() uploaded_log_path = None uploaded_csv_path = None if self.server_client: try: job_data = { 'userId': self.server_client.user_id, 'serviceId': 'ctg-port-tracking', 'serviceName': 'CTG Port Tracking', 'inputFileName': os.path.basename(self.input_file_path) if self.input_file_path else 'input.csv', 'inputFilePath': self.input_file_path if self.input_file_path else '', 'creditsUsed': self.credits_used } if csv_bytes and file_name: uploaded_csv_path = self.server_client.upload_results_buffer(csv_bytes, file_name, job_data) uploaded_log_path = None if hasattr(self, 'log_buffer'): log_text = self.log_buffer.getvalue() if log_text: uploaded_log_path = self.server_client.upload_log_buffer(log_text, f'ctg_port_tracking_{self.job_id}.log', job_data) result_files = [] if uploaded_csv_path: result_files.append(uploaded_csv_path) elif uploaded_log_path: result_files.append(uploaded_log_path) self.server_client.report_job_result( result_files, self.output_dir, successful_count=success_count ) except Exception as e: self.logger.error(f"❌ Error uploading files to server: {e}") # Print summary self.logger.info(f"\n{'='*60}") self.logger.info("PROCESSING SUMMARY") self.logger.info(f"{'='*60}") self.logger.info(f"Total containers: {total_containers}") self.logger.info(f"✓ Successful downloads: {success_count}") self.logger.info(f"✗ Failed downloads: {failed_count}") self.logger.info(f"📥 Total downloads completed: {success_count + failed_count}/{total_containers}") self.logger.info(f"{'='*60}\n") return success, fail, success_count # ------------------------------------------------------------- def generate_combined_report(self, pdfs): if not pdfs: return None combined_name = f"ctg_port_tracking_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf" combined_path = os.path.join(self.output_dir, combined_name) merger = PdfMerger() for pdf in pdfs: path = os.path.join(self.output_dir, pdf) if os.path.exists(path): merger.append(path) merger.write(combined_path) merger.close() self.logger.info(f"✅ Combined PDF saved: {combined_name}") return combined_name # ------------------------------------------------------------- def cleanup(self): try: if self.driver: self.driver.quit() self.logger.info("🔒 Browser closed.") except Exception as e: self.logger.warning(f"Cleanup error: {e}") if self.temp_profile and os.path.exists(self.temp_profile): try: shutil.rmtree(self.temp_profile) self.logger.info(f"🗑️ Cleaned profile: {self.temp_profile}") except Exception as e: self.logger.warning(f"Temp cleanup error: {e}") # ------------------------------------------------------------- def run(self, file_path): try: # Setup browser (needed for PDF generation) if not self.setup_driver(): return False # Note: No need to navigate to portal anymore - we use direct POST requests containers = self.read_container_file(file_path) if not containers: self.logger.error("No container numbers found.") return False success, fail, success_count = self.process_all(containers) self.logger.info(f"🎉 Done. Success={len(success)} Fail={len(fail)} SuccessCount={success_count}") return True finally: self.cleanup() def main(): # Early execution logging print("🔵 [CTG_PORT_TRACKING] Script execution started", flush=True) print(f"🔵 [CTG_PORT_TRACKING] sys.argv: {sys.argv}", flush=True) print(f"🔵 [CTG_PORT_TRACKING] __name__: {__name__}", flush=True) print(f"🔵 [CTG_PORT_TRACKING] Python version: {sys.version}", flush=True) if len(sys.argv) < 4: error_msg = "Usage: python ctg_port_tracking.py [api_base_url] [user_id] [credits_used]" print(f"❌ [CTG_PORT_TRACKING] {error_msg}", file=sys.stderr, flush=True) print(error_msg, file=sys.stderr) sys.exit(1) file_path, output_dir, job_id = sys.argv[1], sys.argv[2], sys.argv[3] print(f"🔵 [CTG_PORT_TRACKING] Parsed arguments: file_path={file_path}, output_dir={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 print(f"🔵 [CTG_PORT_TRACKING] Server params: api_base_url={api_base_url}, user_id={user_id}, credits_used={credits_used}", flush=True) # File path check file_exists = os.path.exists(file_path) print(f"🔵 [CTG_PORT_TRACKING] File path check: {file_path} exists: {file_exists}", flush=True) if not file_exists: error_msg = f"❌ File not found: {file_path}" print(f"❌ [CTG_PORT_TRACKING] {error_msg}", file=sys.stderr, flush=True) print(error_msg, file=sys.stderr) sys.exit(1) # Create server client if parameters provided print("🔵 [CTG_PORT_TRACKING] Importing modules...", flush=True) server_client = None if api_base_url and user_id: try: server_client = ServerClient(api_base_url, user_id, job_id) if server_client.enabled: print(f"✅ [CTG_PORT_TRACKING] Server communication enabled: {api_base_url}", flush=True) print(f"✅ Server communication enabled: {api_base_url}") else: print(f"⚠️ [CTG_PORT_TRACKING] Server communication disabled (requests library not available)", flush=True) print(f"⚠️ Server communication disabled (requests library not available)") except Exception as e: print(f"❌ [CTG_PORT_TRACKING] Failed to create ServerClient: {e}", file=sys.stderr, flush=True) import traceback traceback.print_exc(file=sys.stderr) print("🔵 [CTG_PORT_TRACKING] Creating CtgPortTrackingAutomation instance...", flush=True) try: app = CtgPortTrackingAutomation( headless=True, output_dir=output_dir, job_id=job_id, server_client=server_client ) # Store input file path and credits for server uploads app.input_file_path = file_path app.credits_used = credits_used print("🔵 [CTG_PORT_TRACKING] Starting automation.run()...", flush=True) ok = app.run(file_path) print(f"🔵 [CTG_PORT_TRACKING] Automation completed: success={ok}", flush=True) sys.exit(0 if ok else 1) except Exception as e: print(f"❌ [CTG_PORT_TRACKING] Error in main execution: {e}", file=sys.stderr, flush=True) import traceback traceback.print_exc(file=sys.stderr) sys.exit(1) if __name__ == "__main__": try: main() except Exception as e: print(f"❌ [CTG_PORT_TRACKING] Fatal error: {e}", file=sys.stderr, flush=True) import traceback traceback.print_exc(file=sys.stderr) sys.exit(1)