""" Direct Integration watcher: polls MySQL for unprocessed uploads, downloads the Excel file from S3, extracts data, and marks the row processed. """ import threading import time import logging import os from typing import Any, Dict, List import boto3 import subprocess import openpyxl import json import db from .constants import AUTO_EXTRACTOR_CHECK_INTERVAL, CHROME_DOWNLOADS_FOLDER from mysql.mysql_client import ( get_unprocessed_direct_uploads, mark_direct_upload_processed, set_direct_upload_status, get_application_by_id, get_playlist_by_id, log_direct_integration_table_snapshot, get_latest_unprocessed_direct_upload, get_direct_upload_by_id, get_mysql_connection, ) from mysql.config import refresh_mysql_config import pymysql class DirectIntegrationWatcher: """Background watcher for direct integration uploads.""" _instance = None _instance_lock = threading.Lock() def __new__(cls, app, *, poll_interval: float | None = None): with cls._instance_lock: if cls._instance is None: cls._instance = super(DirectIntegrationWatcher, cls).__new__(cls) cls._instance.initialized = False return cls._instance def __init__(self, app, *, poll_interval: float | None = None) -> None: if self.initialized: return self.app = app self.poll_interval = poll_interval or AUTO_EXTRACTOR_CHECK_INTERVAL self._stop = False self.s3 = None self._thread = None self.initialized = True # Background monitor to detect remote cancellations (processed=5) self._cancel_monitor_thread = None try: import threading as _th self._cancel_monitor_stop = _th.Event() except Exception: self._cancel_monitor_stop = None self._cancel_monitor_upload_id = None # Check if we're in local environment env = os.getenv("ENV", "").lower() self.is_local = env == "local" if self.is_local: logging.info("[DirectIntegration] Running in local environment - S3 operations disabled") self.s3 = None return # Initialize S3 client with error handling try: region = os.getenv("AWS_REGION", "ap-southeast-2") self.s3 = boto3.client("s3", region_name=region) # Test AWS credentials by listing a bucket (but don't fail if no buckets exist) try: self.s3.list_buckets() logging.info("[DirectIntegration] AWS S3 client initialized successfully") except Exception as cred_test_error: logging.warning(f"[DirectIntegration] AWS credentials may be invalid: {cred_test_error}") except Exception as init_error: logging.error(f"[DirectIntegration] Failed to initialize S3 client: {init_error}") self.s3 = None def start(self) -> None: if self._thread and self._thread.is_alive(): logging.info("[DirectIntegration] Watcher already running") return self._stop = False self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() logging.info("[DirectIntegration] Started watcher thread") def stop(self) -> None: logging.info("[DirectIntegration] Stopping watcher") self._stop = True if self._thread: self._thread.join(timeout=1.0) self._thread = None # Also stop any active cancel monitor try: self._stop_cancel_monitor() except Exception: pass def restart_polling(self) -> None: """Restart polling for new unprocessed uploads.""" logging.info("[DirectIntegration] Restarting watcher") self.stop() self.start() def _run(self) -> None: """Poll for new unprocessed uploads every 30 seconds until one is found and processed.""" logging.info("[DirectIntegration] Starting watcher thread") while not self._stop: try: # Heartbeat during idle polling loop try: from os import getenv as _getenv hb_interval = float(_getenv("HB_INTERVAL_SEC", "60")) if hasattr(self, 'app') and self.app: self.app.send_heartbeat_throttled(min_interval_sec=hb_interval) except Exception: pass if not hasattr(self.app, 'initialized') or not self.app.initialized: logging.info("[DirectIntegration] App not initialized yet, waiting...") time.sleep(5) continue # Get the latest unprocessed upload with connection recovery upload = None try: # Test connection first to catch auth errors early test_conn = get_mysql_connection() test_conn.close() upload = get_latest_unprocessed_direct_upload() except (pymysql.Error, Exception) as db_error: # Check if this is a connection/auth error error_str = str(db_error).lower() error_code = getattr(db_error, 'args', [None])[0] if hasattr(db_error, 'args') and db_error.args else None is_auth_error = ( 'access denied' in error_str or error_code == 1045 or # MySQL error code for access denied '1045' in str(db_error) or 'authentication' in error_str or 'connection' in error_str or 'timeout' in error_str or 'network' in error_str or 'lost connection' in error_str ) if is_auth_error: logging.warning(f"[DirectIntegration] Database connection/auth error detected: {db_error}") logging.info("[DirectIntegration] Attempting to refresh MySQL credentials and reconnect...") # Refresh credentials from AWS/local config if refresh_mysql_config(): logging.info("[DirectIntegration] MySQL credentials refreshed successfully, retrying...") # Retry the database call with fresh credentials try: test_conn = get_mysql_connection() test_conn.close() upload = get_latest_unprocessed_direct_upload() logging.info("[DirectIntegration] Successfully reconnected after credential refresh") except Exception as retry_error: logging.error(f"[DirectIntegration] Retry after refresh failed: {retry_error}") upload = None else: logging.error("[DirectIntegration] Failed to refresh MySQL credentials") upload = None else: # Not a connection error, log and continue logging.error(f"[DirectIntegration] Unexpected database error: {db_error}") upload = None if not upload: # logging.info("[DirectIntegration] No unprocessed uploads found, waiting...") # Heartbeat while sleeping try: from os import getenv as _getenv hb_interval = float(_getenv("HB_INTERVAL_SEC", "60")) if hasattr(self, 'app') and self.app: self.app.send_heartbeat_throttled(min_interval_sec=hb_interval) except Exception: pass time.sleep(30) # Wait 30 seconds before checking again continue # Always mark as needing user interaction (4) when we first get a record upload_id = upload.get('id') set_direct_upload_status(upload_id, 4) # Capture admin flag for UI try: if hasattr(self, 'app') and self.app: _val = upload.get('is_admin') is_admin_flag = False try: if _val is not None: is_admin_flag = bool(int(_val)) except Exception: is_admin_flag = bool(_val) self.app.current_is_admin = is_admin_flag except Exception: pass # Get playlist name based on mode playlist_mode = upload.get('playlist_mode', '').strip().lower() try: # Stash mode onto app for UI rendering if hasattr(self, 'app') and self.app: self.app.current_playlist_mode = playlist_mode except Exception: pass if playlist_mode == 'existing' and upload.get('playlist_id'): # For existing playlists, always get name from playlists table from mysql.mysql_client import get_playlist_by_id pl_row = get_playlist_by_id(int(upload.get('playlist_id'))) playlist_name = pl_row.get('name') if pl_row else 'Unknown' else: # For new playlists, use name from direct_integration_uploads playlist_name = upload.get('playlist_name', 'Unknown') logging.info( f"[DirectIntegration] Processing upload id={upload_id} " f"mode={playlist_mode} name={playlist_name} " f"s3={bool(upload.get('s3_bucket') and upload.get('s3_key'))} " f"app_id={upload.get('application_id')} local={self.is_local}" ) # Process the upload and stop polling after processing (success or error) self._process_upload(upload) logging.info(f"[DirectIntegration] Processed upload {upload_id}, stopping watcher") self._stop = True # Stop polling after processing any record break except Exception as e: logging.error(f"DirectIntegration error: {e}") time.sleep(30) # Wait before retrying even on error continue def _process_upload(self, upload: Dict[str, Any]) -> None: upload_id = upload["id"] bucket = upload.get("s3_bucket") key = upload.get("s3_key") application_id = upload.get("application_id") playlist_id = upload.get("playlist_id") # Capture admin flag early for UI rendering try: if hasattr(self, 'app') and self.app: _val = upload.get('is_admin') is_admin_flag = False try: if _val is not None: is_admin_flag = bool(int(_val)) except Exception: is_admin_flag = bool(_val) self.app.current_is_admin = is_admin_flag except Exception: pass # Clear local SQLite session to avoid mixing with previous sessions try: logging.info("[DirectIntegration] Clearing local SQLite session tables before processing upload") db.clear_session_db() except Exception: logging.exception("[DirectIntegration] Failed to clear local SQLite session tables (continuing)") # Get playlist mode playlist_mode = (upload.get("playlist_mode") or "").strip().lower() try: if hasattr(self, 'app') and self.app: self.app.current_playlist_mode = playlist_mode except Exception: pass # For new playlists, we don't need to check S3 since there won't be any files if playlist_mode == "new": logging.info(f"[DirectIntegration] New playlist mode - skipping S3 operations for upload {upload_id}") has_s3_info = False else: # Handle both S3 uploads and local uploads for existing playlists has_s3_info = bucket and key # Default: require user interaction for all playlists (Done check) ui_wait_required = True # Only attempt S3 operations for existing playlists in non-local mode if has_s3_info and not self.is_local and playlist_mode != "new": # S3 upload processing (only in non-local mode) if not self.s3: error_msg = "S3 client not available - cannot process S3 upload" logging.error(f"[DirectIntegration] {error_msg}") set_direct_upload_status(upload_id, 3, error_message=error_msg) return elif has_s3_info and self.is_local: # S3 upload in local mode - skip S3 operations, just setup UI logging.info(f"[DirectIntegration] Local mode: Skipping S3 operations for upload {upload_id}") else: # Local upload - no S3 file, just setup UI for manual recording logging.info(f"[DirectIntegration] Processing local upload {upload_id} - no S3 file needed") tmp_path = None try: # Stash user/company context for downstream save try: self.app.current_user_id = upload.get("user_id") self.app.current_company_id = upload.get("company_id") except Exception: pass # Reflect application/playlist into UI for visibility if application_id: try: # Log current state current_app_name = self.app.selected_application.get() if hasattr(self.app, 'selected_application') else None logging.info(f"[DirectIntegration] Current application name: '{current_app_name}'") # Fetch application details app_row = get_application_by_id(int(application_id)) logging.info(f"[DirectIntegration] Fetched application row: {app_row}") if app_row and app_row.get("name"): # Store application info for UI to use (on main thread) def _set_app_info(): try: self.app.current_application_id = int(application_id) self.app.current_application_name = f"{app_row['name']} ({app_row.get('platform','')})" # Capture instance_id for heartbeat usage try: instance_id = app_row.get('instance_id') if instance_id: self.app.current_instance_id = str(instance_id) logging.info(f"[DirectIntegration] Captured instance_id for heartbeat: {self.app.current_instance_id}") # Ensure instance_usage row exists early (non-destructive) try: from mysql.heartbeat_functions import ensure_instance_row ensure_instance_row(self.app.current_instance_id) except Exception: pass except Exception: pass logging.info(f"[DirectIntegration] Stored application name: {self.app.current_application_name}") except Exception: pass try: self.app.after(0, _set_app_info) except Exception: _set_app_info() # Launch the target application immediately using start_point try: # Stash upload id early for per-upload Chrome profile try: self.app.current_direct_upload_id = upload_id except Exception: pass self._launch_application_from_row(app_row) except Exception as e_launch: logging.error(f"[DirectIntegration] Failed to launch application: {e_launch}") else: logging.warning(f"[DirectIntegration] Application ID {application_id} not found or has no name") self.app.current_application_name = f"App ID {application_id}" except Exception as e: logging.error(f"[DirectIntegration] Error fetching application {application_id}: {e}") self.app.current_application_name = f"Error loading app {application_id}" # Playlist selection logic based on playlist_mode playlist_mode = (upload.get("playlist_mode") or "").strip().lower() try: if hasattr(self, 'app') and self.app: self.app.current_playlist_mode = playlist_mode except Exception: pass playlist_name_from_upload = upload.get("playlist_name") if playlist_mode == "new" and playlist_name_from_upload: try: # New playlist flow: ensure we don't append to a previous MySQL playlist id try: self.app.current_mysql_playlist_id = None except Exception: pass # UI updates must occur on the Tk thread def _set_new_playlist_ui(): try: # Ensure DI UI flags are set BEFORE rebuilding widgets try: self.app.user_interaction_upload_id = upload_id self.app.current_direct_upload_id = upload_id self.app.user_interaction_mode = True except Exception: pass self.app.selected_playlist.set(str(playlist_name_from_upload)) self.app.playlists = [] try: self.app.recording_name_var.set(str(playlist_name_from_upload)) self.app.hide_recording_name = True except Exception: pass self.app.create_widgets() except Exception: pass try: self.app.after(0, _set_new_playlist_ui) except Exception: _set_new_playlist_ui() except Exception: pass elif playlist_mode == "existing" and playlist_id: pl_row = get_playlist_by_id(int(playlist_id)) if pl_row and pl_row.get("name"): try: playlist_name = pl_row["name"] def _set_existing_playlist_ui(): try: # Ensure DI UI flags are set BEFORE rebuilding widgets try: self.app.user_interaction_upload_id = upload_id self.app.current_direct_upload_id = upload_id self.app.user_interaction_mode = True except Exception: pass # Set playlist and trigger selection event self.app.selected_playlist.set(playlist_name) # Store MySQL playlist ID self.app.current_mysql_playlist_id = int(playlist_id) # Hide dropdown similarly by clearing list self.app.playlists = [] # Set recording name to playlist name self.app.recording_name_var.set(playlist_name) # Keep recording name visible but read-only for existing playlists self.app.hide_recording_name = False self.app.create_widgets() # Trigger playlist selection to load actions try: self.app.after(0, self.app.playlist_manager.on_playlist_selected) except Exception: self.app.playlist_manager.on_playlist_selected() except Exception: pass try: self.app.after(0, _set_existing_playlist_ui) except Exception: _set_existing_playlist_ui() except Exception: pass # Handle S3 vs Local uploads differently data = None # Default to no data if has_s3_info: if self.is_local: # In local mode, just log that we're skipping S3 logging.info(f"[DirectIntegration] Local mode: Skipping S3 operations for upload {upload_id}") else: # Only try S3 operations in non-local mode filename = key.rsplit("/", 1)[-1] tmp_path = os.path.join("/tmp", filename) os.makedirs(os.path.dirname(tmp_path), exist_ok=True) try: self.s3.download_file(bucket, key, tmp_path) logging.info(f"[DirectIntegration] Successfully downloaded s3://{bucket}/{key}") # Extract Excel data data = self._extract_excel(tmp_path) logging.info(f"[DirectIntegration] Extracted {len(json.loads(data)) if data else 0} rows from Excel") except Exception as download_error: error_msg = f"Failed to download S3 file s3://{bucket}/{key}: {download_error}" logging.error(f"[DirectIntegration] {error_msg}") if not self.is_local and playlist_mode != "new": # Only mark as error in non-local mode and for existing playlists logging.info(f"[DirectIntegration] Upload {upload_id} failed to download - marking as error") set_direct_upload_status(upload_id, 3, error_message=error_msg) return elif playlist_mode == "new": # For new playlists, log but continue without error logging.info(f"[DirectIntegration] Ignoring S3 error for new playlist upload {upload_id}") pass else: # Local upload - no S3 file needed logging.info(f"[DirectIntegration] Local upload {upload_id} - no S3 file needed") # Preview: write a sample to UI (only for S3 uploads with data in non-local mode) if has_s3_info and data and not self.is_local: try: rows = json.loads(data) preview = rows[:5] self._display_extracted_preview( source=f"s3://{bucket}/{key}", filename=upload.get("original_filename") or upload.get("filename") or filename, total=len(rows), sample=preview, ) except Exception: pass # Stash DI context for downstream (screenshots, manage file processing) try: # Bucket and base prefix (folder without filename) if bucket: self.app.current_di_bucket = bucket if key: base_prefix = key.rsplit('/', 1)[0] if '/' in key else '' self.app.current_di_base_prefix = base_prefix # Step details payload step_details = upload.get('step_details') if isinstance(step_details, str): import json as _json try: step_details = _json.loads(step_details) except Exception: pass if isinstance(step_details, dict): self.app.current_di_step_details = step_details # Also capture userName if available for folder suffix try: self.app.current_di_username = step_details.get('userName') or step_details.get('user') except Exception: pass # Allow stepDetails to force/skip UI wait (optional override) try: if isinstance(step_details.get('requireUserInteraction'), bool): ui_wait_required = bool(step_details.get('requireUserInteraction')) except Exception: pass except Exception: pass # Update upload record based on whether we require user interaction if ui_wait_required: # Mark as waiting for user interaction (4) try: set_direct_upload_status(upload_id, 4) except Exception: pass # Immediately flip UI into interaction mode (synchronous, as in previous working version) try: logging.info(f"[DirectIntegration] Entering user interaction mode for upload {upload_id}") self.app.user_interaction_upload_id = upload_id self.app.current_direct_upload_id = upload_id self.app.user_interaction_mode = True self.app.create_widgets() except Exception as e: logging.error(f"[DirectIntegration] Failed to setup user interaction mode: {e}") pass # Start background cancellation monitor for this upload try: self._start_cancel_monitor(upload_id) except Exception: pass else: # Keep as needing user interaction unless explicitly overridden try: set_direct_upload_status(upload_id, 4) except Exception: pass # Start background cancellation monitor even if UI wait is overridden try: self._start_cancel_monitor(upload_id) except Exception: pass # Skip an extra immediate rebuild; _enter_ui_wait already scheduled UI creation except Exception as e: error_msg = f"Failed to process upload {upload_id}: {e}" logging.error(f"[DirectIntegration] {error_msg}") # Check playlist mode to determine retry behavior playlist_mode = (upload.get("playlist_mode") or "").strip().lower() # Always mark as error to prevent reprocessing logging.info(f"[DirectIntegration] Upload {upload_id} failed to process - marking as error") set_direct_upload_status(upload_id, 3, error_message=error_msg) # Close browsers on error try: self.app._close_browser_processes() except Exception: pass # Close desktop apps on error try: self.app._close_desktop_app_processes() except Exception: pass finally: if tmp_path and os.path.exists(tmp_path): try: os.remove(tmp_path) except Exception: pass def _start_cancel_monitor(self, upload_id: int) -> None: """Start or restart a background thread that polls for processed=5 (canceled). Poll interval is controlled by env CANCEL_CHECK_INTERVAL_SEC (default 60 seconds). When cancellation is detected, schedule app.on_cancel_playlist() on the UI thread. """ # Stop any existing monitor first try: self._stop_cancel_monitor() except Exception: pass try: import threading as _th if self._cancel_monitor_stop is None: self._cancel_monitor_stop = _th.Event() else: self._cancel_monitor_stop.clear() self._cancel_monitor_upload_id = int(upload_id) self._cancel_monitor_thread = _th.Thread( target=self._cancel_monitor_loop, args=(int(upload_id),), daemon=True, ) self._cancel_monitor_thread.start() logging.info(f"[DirectIntegration] Started cancel monitor for upload {upload_id}") except Exception as e: logging.warning(f"[DirectIntegration] Failed to start cancel monitor: {e}") def _stop_cancel_monitor(self) -> None: """Signal and join the cancel monitor thread if running.""" try: if self._cancel_monitor_thread and self._cancel_monitor_thread.is_alive(): if self._cancel_monitor_stop is not None: self._cancel_monitor_stop.set() self._cancel_monitor_thread.join(timeout=1.0) finally: self._cancel_monitor_thread = None self._cancel_monitor_upload_id = None try: if self._cancel_monitor_stop is not None: # Create a fresh Event for next run import threading as _th self._cancel_monitor_stop = _th.Event() except Exception: pass def _cancel_monitor_loop(self, upload_id: int) -> None: """Poll MySQL periodically; if processed becomes 5, trigger cancel flow.""" try: from os import getenv as _getenv interval = float(_getenv("CANCEL_CHECK_INTERVAL_SEC", "60")) except Exception: interval = 60.0 # Lightweight loop; exit when upload context ends or stop is requested while True: try: # Stop conditions if self._cancel_monitor_stop is not None and self._cancel_monitor_stop.is_set(): break try: current_id = getattr(self.app, 'current_direct_upload_id', None) if current_id is None or int(current_id) != int(upload_id): break # Upload context ended or changed except Exception: # If we cannot verify, continue polling cautiously pass # Heartbeat during long waits try: hb_interval = float(os.getenv("HB_INTERVAL_SEC", "60")) if hasattr(self, 'app') and self.app: self.app.send_heartbeat_throttled(min_interval_sec=hb_interval) except Exception: pass # Poll the database for status with connection recovery try: # Test connection first to catch auth errors early test_conn = get_mysql_connection() test_conn.close() row = get_direct_upload_by_id(int(upload_id)) or {} status_val = row.get('processed') try: status_int = int(status_val) if status_val is not None else None except Exception: status_int = None if status_int == 5: logging.info(f"[DirectIntegration] Upload {upload_id} marked canceled remotely (processed=5); stopping playlist") # Schedule cancel on UI thread to ensure safe UI updates try: self.app.after(0, self.app.on_cancel_playlist) except Exception: try: self.app.on_cancel_playlist() except Exception: pass break except (pymysql.Error, Exception) as db_error: # Check if this is a connection/auth error error_str = str(db_error).lower() error_code = getattr(db_error, 'args', [None])[0] if hasattr(db_error, 'args') and db_error.args else None is_auth_error = ( 'access denied' in error_str or error_code == 1045 or # MySQL error code for access denied '1045' in str(db_error) or 'authentication' in error_str or 'connection' in error_str or 'timeout' in error_str or 'network' in error_str or 'lost connection' in error_str ) if is_auth_error: logging.warning(f"[DirectIntegration] Cancel monitor: Database connection/auth error detected: {db_error}") logging.info("[DirectIntegration] Cancel monitor: Attempting to refresh MySQL credentials and reconnect...") # Refresh credentials from AWS/local config if refresh_mysql_config(): logging.info("[DirectIntegration] Cancel monitor: MySQL credentials refreshed successfully, retrying...") # Retry the database call with fresh credentials try: test_conn = get_mysql_connection() test_conn.close() row = get_direct_upload_by_id(int(upload_id)) or {} status_val = row.get('processed') try: status_int = int(status_val) if status_val is not None else None except Exception: status_int = None if status_int == 5: logging.info(f"[DirectIntegration] Upload {upload_id} marked canceled remotely (processed=5); stopping playlist") try: self.app.after(0, self.app.on_cancel_playlist) except Exception: try: self.app.on_cancel_playlist() except Exception: pass break except Exception as retry_error: logging.error(f"[DirectIntegration] Cancel monitor: Retry after refresh failed: {retry_error}") else: logging.error("[DirectIntegration] Cancel monitor: Failed to refresh MySQL credentials") # Continue; a transient DB error shouldn't stop monitoring pass # Sleep until next poll time.sleep(max(5.0, float(interval))) except Exception: # Ensure monitor remains resilient try: time.sleep(max(5.0, float(interval))) except Exception: # As a last resort, break out break def _launch_application_from_row(self, app_row: Dict[str, Any]) -> None: """Launch application based on platform and start_point from application row. - Web platform: launch Chrome in kiosk mode; fallback to regular mode - Desktop platform: execute start_point as a shell command """ try: start_point = app_row.get('start_point') or app_row.get('start_POINT') platform = (app_row.get('platform') or '').strip().lower() if not start_point: logging.warning("[DirectIntegration] No start_point provided; skipping launch") return # Prevent duplicate launches for the same upload id try: current_id = getattr(self.app, 'current_direct_upload_id', None) if getattr(self.app, 'browser_launch_upload_id', None) == current_id and current_id is not None: logging.info(f"[DirectIntegration] Browser already launched for upload {current_id} - skipping re-launch") return except Exception: pass if platform == 'web': try: # Harden Chrome for headless/server GPU-less environments upload_id = getattr(self.app, 'current_direct_upload_id', None) user_data_dir = None try: if upload_id: user_data_dir = f"/tmp/chrome-profile-{int(upload_id)}" else: # Create a temporary profile directory if no upload_id import tempfile user_data_dir = tempfile.mkdtemp(prefix="chrome-profile-") os.makedirs(user_data_dir, exist_ok=True) except Exception: user_data_dir = None # Enable DevTools for full-page screenshots (loopback only by default) try: rd_host = os.getenv('CHROME_REMOTE_DEBUGGING_HOST', '127.0.0.1') rd_port = str(int(os.getenv('CHROME_REMOTE_DEBUGGING_PORT', '9222'))) except Exception: rd_host, rd_port = '127.0.0.1', '9222' rd_flags = [ f"--remote-debugging-port={rd_port}", f"--remote-debugging-address={rd_host}", ] chrome_cmd = [ "google-chrome", "--kiosk", "--disable-gpu", "--no-sandbox", "--disable-dev-shm-usage", "--no-first-run", "--no-default-browser-check", "--disable-background-networking", "--disable-crash-reporter", "--disable-logging", "--log-level=3", "--use-gl=swiftshader", "--enable-unsafe-swiftshader", # "--incognito", "--new-window", "--disable-extensions", "--disable-component-update", "--disable-session-crashed-bubble", "--disable-features=DownloadBubbleV2", ] # Insert remote debugging flags if not already present for f in rd_flags: if f not in chrome_cmd: chrome_cmd.append(f) if user_data_dir: chrome_cmd += [f"--user-data-dir={user_data_dir}"] # Disable save dialog and password prompts by setting Chrome preferences try: prefs_dir = os.path.join(user_data_dir, "Default") os.makedirs(prefs_dir, exist_ok=True) prefs_file = os.path.join(prefs_dir, "Preferences") prefs = {} if os.path.exists(prefs_file): with open(prefs_file, 'r', encoding='utf-8') as f: prefs = json.load(f) # Downloads: no prompt, fixed downloads folder downloads_path = os.path.abspath(os.path.expanduser(CHROME_DOWNLOADS_FOLDER)) os.makedirs(downloads_path, exist_ok=True) dl = prefs.setdefault("download", {}) dl["prompt_for_download"] = False dl["default_directory"] = downloads_path.replace("\\", "/") dl["directory_upgrade"] = True # Password manager: completely disabled prefs["credentials_enable_service"] = False profile_prefs = prefs.setdefault("profile", {}) profile_prefs["password_manager_enabled"] = False with open(prefs_file, 'w', encoding='utf-8') as f: json.dump(prefs, f, indent=2) # Also create Master Preferences file (used when profile is first created) master_prefs_file = os.path.join(user_data_dir, "Master Preferences") with open(master_prefs_file, 'w', encoding='utf-8') as f: json.dump(prefs, f, indent=2) logging.info( "[DirectIntegration] Set Chrome prefs: downloads -> '%s', password manager disabled", downloads_path, ) except Exception as e: logging.warning(f"[DirectIntegration] Failed to set Chrome preferences: {e}") chrome_cmd += [start_point] # Launch Chrome detached so our thread does not block try: subprocess.Popen( chrome_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) except Exception: # Fallback to run if Popen fails, but this may block subprocess.run(chrome_cmd, check=False) logging.info(f"[DirectIntegration] Launched web application in kiosk mode: {start_point}") try: if upload_id is not None: setattr(self.app, 'browser_launch_upload_id', upload_id) except Exception: pass except Exception as e: logging.error(f"[DirectIntegration] Chrome kiosk failed: {e}") try: chrome_cmd = [ "google-chrome", "--disable-gpu", "--no-sandbox", "--disable-dev-shm-usage", "--no-first-run", "--no-default-browser-check", "--disable-background-networking", "--disable-crash-reporter", "--disable-logging", "--log-level=3", "--use-gl=swiftshader", "--enable-unsafe-swiftshader", # "--incognito", "--new-window", "--disable-extensions", "--disable-component-update", "--disable-session-crashed-bubble", "--disable-features=DownloadBubbleV2", ] # Insert remote debugging flags if not already present for f in rd_flags: if f not in chrome_cmd: chrome_cmd.append(f) if user_data_dir: chrome_cmd += [f"--user-data-dir={user_data_dir}"] # Disable save dialog by setting Chrome preference try: prefs_dir = os.path.join(user_data_dir, "Default") os.makedirs(prefs_dir, exist_ok=True) prefs_file = os.path.join(prefs_dir, "Preferences") prefs = {} if os.path.exists(prefs_file): with open(prefs_file, 'r', encoding='utf-8') as f: prefs = json.load(f) prefs.setdefault("download", {})["prompt_for_download"] = False downloads_path = os.path.abspath(os.path.expanduser(CHROME_DOWNLOADS_FOLDER)) os.makedirs(downloads_path, exist_ok=True) prefs.setdefault("download", {})["default_directory"] = downloads_path.replace("\\", "/") with open(prefs_file, 'w', encoding='utf-8') as f: json.dump(prefs, f, indent=2) # Also create Master Preferences file (works even in incognito) master_prefs_file = os.path.join(user_data_dir, "Master Preferences") with open(master_prefs_file, 'w', encoding='utf-8') as f: json.dump(prefs, f, indent=2) logging.info(f"[DirectIntegration] Set Chrome download preferences: prompt_for_download=False, directory={downloads_path}") except Exception as e: logging.warning(f"[DirectIntegration] Failed to set Chrome download preferences: {e}") chrome_cmd += [start_point] try: subprocess.Popen( chrome_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) except Exception: subprocess.run(chrome_cmd, check=False) logging.info(f"[DirectIntegration] Launched web application (regular): {start_point}") try: if upload_id is not None: setattr(self.app, 'browser_launch_upload_id', upload_id) except Exception: pass except Exception as e2: logging.error(f"[DirectIntegration] Chrome launch failed: {e2}") elif platform == 'desktop': try: if os.name == 'nt': # Use native Windows launcher first try: # Prefer PowerShell Start-Process -PassThru to capture the TARGET app PID ps_cmd = [ "powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", f"$p=Start-Process -FilePath \"{start_point}\" -PassThru; $p.Id" ] res = subprocess.run(ps_cmd, capture_output=True, text=True, check=False) pid_val = None if res.stdout: try: pid_val = int(res.stdout.strip().splitlines()[-1]) except Exception: pid_val = None if pid_val: try: if hasattr(self.app, 'launched_desktop_pids'): self.app.launched_desktop_pids.add(pid_val) except Exception: pass logging.info(f"[DirectIntegration] Launched desktop application: {start_point} (pid={pid_val})") # Also capture descendant/GUI PIDs under the same install folder (Sage spawns children) try: dir_path = os.path.dirname(start_point).replace('"', '\"') ps_children = [ "powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ( "$dir=\"" + dir_path + "\"; " "$procs=Get-Process | Where-Object { $_.Path -like (\"$dir\\*\") -and $_.MainWindowHandle -ne 0 }; " "$procs | ForEach-Object { $_.Id }" ) ] res2 = subprocess.run(ps_children, capture_output=True, text=True, check=False) if res2.stdout: for line in res2.stdout.splitlines(): try: cid = int(line.strip()) if hasattr(self.app, 'launched_desktop_pids'): self.app.launched_desktop_pids.add(cid) except Exception: pass except Exception: pass else: os.startfile(start_point) # type: ignore[attr-defined] logging.info(f"[DirectIntegration] Launched desktop application (no pid captured): {start_point}") try: img_name = os.path.splitext(os.path.basename(start_point))[0] if hasattr(self.app, 'launched_desktop_process_names'): self.app.launched_desktop_process_names.add(img_name) if hasattr(self.app, 'launched_desktop_executable_paths'): self.app.launched_desktop_executable_paths.add(start_point) try: if hasattr(self.app, 'launched_desktop_name_keywords'): # Use 'sage' and app name as keywords for substring match cleanup self.app.launched_desktop_name_keywords.add('sage') name_kw = str(app_row.get('name') or '').strip().lower() if name_kw: self.app.launched_desktop_name_keywords.add(name_kw) except Exception: pass except Exception: pass except Exception as startfile_err: logging.warning(f"[DirectIntegration] os.startfile failed ({startfile_err}); falling back to 'start'") subprocess.Popen( ['cmd', '/c', 'start', '', start_point], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) logging.info(f"[DirectIntegration] Launched desktop application via cmd start: {start_point}") try: img_name = os.path.splitext(os.path.basename(start_point))[0] if hasattr(self.app, 'launched_desktop_process_names'): self.app.launched_desktop_process_names.add(img_name) if hasattr(self.app, 'launched_desktop_executable_paths'): self.app.launched_desktop_executable_paths.add(start_point) try: if hasattr(self.app, 'launched_desktop_name_keywords'): self.app.launched_desktop_name_keywords.add('sage') name_kw = str(app_row.get('name') or '').strip().lower() if name_kw: self.app.launched_desktop_name_keywords.add(name_kw) except Exception: pass except Exception: pass else: subprocess.Popen( [start_point], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) logging.info(f"[DirectIntegration] Launched desktop application: {start_point}") except Exception as e: logging.error(f"[DirectIntegration] Desktop app launch failed: {e}") else: logging.warning(f"[DirectIntegration] Unknown platform '{platform}' for start_point {start_point}") except Exception: logging.exception("[DirectIntegration] Unexpected error while launching application") def _extract_excel(self, file_path: str) -> str: """Extracts rows from the first sheet and returns JSON string of rows. The resulting structure is a list of dicts using the first row as headers. """ wb = openpyxl.load_workbook(file_path, data_only=True) sheet = wb.active rows = list(sheet.iter_rows(values_only=True)) if not rows: return "[]" headers = [str(h).strip() if h is not None else f"col_{i}" for i, h in enumerate(rows[0])] result: List[Dict[str, Any]] = [] for row in rows[1:]: item = {headers[i]: row[i] for i in range(len(headers))} result.append(item) try: import json as _json return _json.dumps(result, default=str) except Exception: # Fallback string repr return str(result) def _display_extracted_preview(self, *, source: str, filename: str, total: int, sample: List[Dict[str, Any]]) -> None: """Write a small preview of the extracted data to the live clicks pane.""" if not hasattr(self.app, 'live_clicks_text') or not self.app.live_clicks_text: return try: self.app.live_clicks_text.config(state='normal') self.app.live_clicks_text.delete('1.0', 'end') self.app.live_clicks_text.insert('end', 'Direct Integration Upload Processed\n') self.app.live_clicks_text.insert('end', f'Source: {source}\n') self.app.live_clicks_text.insert('end', f'File: {filename}\n') self.app.live_clicks_text.insert('end', f'Total rows: {total}\n') if sample: self.app.live_clicks_text.insert('end', 'Sample (first 5 rows):\n') for i, row in enumerate(sample, 1): self.app.live_clicks_text.insert('end', f' {i}. {row}\n') self.app.live_clicks_text.see('end') finally: self.app.live_clicks_text.config(state='disabled')