"""Interactive Streamlit Interface for Feed Generation and SFTP Transfer. This application provides a comprehensive interface to: - Generate feed files from Snowflake data (Run tab) - Transfer files via SFTP with monitoring (Deliver tab) - Manage configurations with save/load - Real-time console output with streaming logs Run with: streamlit run streamlit_sftp_interface_v2.py """ __version__ = "2.0.0" import json import os import sys import subprocess import threading import queue import time from datetime import datetime from pathlib import Path from typing import Optional from dotenv import load_dotenv import streamlit as st from streamlit.logger import get_logger import streamlit.components.v1 as components # Session and operation managers (v2 architecture) import operation_manager import session_manager_v2 as session_mgr # Load environment variables load_dotenv() # Add project root to path sys.path.insert(0, str(Path(__file__).parent)) from sftp_transfer import ( # noqa: E402 SFTPConfig, SFTPTransferManager, ) # Import local SFTP test utilities sys.path.insert(0, str(Path(__file__).parent / "tests")) try: from utils_local_sftp import enable_local_sftp_monkeypatch LOCAL_SFTP_AVAILABLE = True except ImportError: LOCAL_SFTP_AVAILABLE = False logger = get_logger(__name__) # ============================================================================ # SESSION STATE INITIALIZATION # ============================================================================ def init_session_state() -> None: """Initialize Streamlit session state variables.""" if "run_config" not in st.session_state: st.session_state.run_config = { "snowflake_user": os.getenv("SNOWFLAKE_USER", ""), "snowflake_password": os.getenv("SNOWFLAKE_PASSWORD", ""), "snowflake_database": os.getenv("SNOWFLAKE_DATABASE", "INTEGRATION"), "snowflake_schema": os.getenv("SNOWFLAKE_SCHEMA", "dev"), "snowflake_account": os.getenv("SNOWFLAKE_ACCOUNT", ""), "snowflake_role": os.getenv("SNOWFLAKE_ROLE", "DEV_ENGINEERING"), "snowflake_warehouse": os.getenv( "SNOWFLAKE_WAREHOUSE", "DEV_OWS_WAREHOUSE" ), "aws_access_key_id": os.getenv("AWS_ACCESS_KEY_ID", ""), "aws_secret_access_key": os.getenv("AWS_SECRET_ACCESS_KEY", ""), "snowflake_source_table": os.getenv( "SNOWFLAKE_SOURCE_TABLE", "SONY_STARS_MONTHLY_EARLY_SPOTIFY_FEED_VIEW" ), "process_mode": os.getenv("PROCESS_MODE", "US"), "period_id": os.getenv("PERIOD_ID", "322"), # Additional configuration variables "file_output_path": os.getenv( "FILE_OUTPUT_PATH", "output/{date}" ), "copy_table": os.getenv("COPY_TABLE", "True") == "True", "delete_copy": os.getenv("DELETE_COPY", "True") == "True", "booking_affiliate": os.getenv("BOOKING_AFFILIATE", "US, GB, NONE"), "environment": os.getenv("ENVIRONMENT", "DEV"), "logger_level": os.getenv("LOGGER_LEVEL", "INFO"), "console_log_level": os.getenv("CONSOLE_LOG_LEVEL", "INFO"), "zip_generation_mode": os.getenv( "ZIP_GENERATION_MODE", "Both" ), "debug_row_processing": os.getenv( "DEBUG_ROW_PROCESSING", "False" ) == "True", } if "deliver_config" not in st.session_state: st.session_state.deliver_config = { "sftp_enable": os.getenv("SFTP_ENABLE", "True") == "True", "sftp_host": os.getenv("SFTP_HOST", "ftp.sme-dsr.com"), "sftp_user": os.getenv("SFTP_USER", "FTP_Orchard"), "sftp_folder": os.getenv("SFTP_FOLDER", "test"), "sftp_password": os.getenv("SFTP_PASSWORD", ""), "sftp_private_key_path": os.getenv("SFTP_PRIVATE_KEY_PATH", ""), "sftp_max_retries": int(os.getenv("SFTP_MAX_RETRIES", "3")), "sftp_retry_backoff": float(os.getenv("SFTP_RETRY_BACKOFF", "2.0")), "sftp_partial_failure_mode": os.getenv("SFTP_PARTIAL_FAILURE_MODE", "FAIL"), "sftp_timeout": int(os.getenv("SFTP_TIMEOUT", "7200")), "sftp_dev_mode": os.getenv("SFTP_DEV_MODE", "True") == "True", "source_directory": "./output", "test_mode": False, # New configuration variables "transfer_mode": os.getenv("TRANSFER_MODE", "Individual Files"), "sftp_port": int(os.getenv("SFTP_PORT", "22")), "sftp_strict_host_key": os.getenv("SFTP_STRICT_HOST_KEY", "True") == "True", "sftp_continue_on_error": os.getenv("SFTP_CONTINUE_ON_ERROR", "False") == "True", "sftp_mock": os.getenv("SFTP_MOCK", "False") == "True", "sftp_prompt_password": os.getenv("SFTP_PROMPT_PASSWORD", "False") == "True", "preserve_files_after_upload": os.getenv("PRESERVE_FILES_AFTER_UPLOAD", "True") == "True", "upload_selection": os.getenv("UPLOAD_SELECTION", "Individual Files"), } if "console_output" not in st.session_state: st.session_state.console_output = [] if "show_sftp_confirm" not in st.session_state: st.session_state.show_sftp_confirm = False if "selected_files" not in st.session_state: st.session_state.selected_files = [] if "auto_scroll_console" not in st.session_state: st.session_state.auto_scroll_console = True if "sessions_list" not in st.session_state: st.session_state.sessions_list = session_mgr.load_sessions() if "selected_session_id" not in st.session_state: st.session_state.selected_session_id = None if "current_session" not in st.session_state: # Try to restore last active session from browser storage sessions = st.session_state.sessions_list if sessions and len(sessions) > 0: # Get the most recently created session latest_session = max(sessions, key=lambda s: s.get('created_at', '')) # Check if it's from today (same session should persist across page loads within a day) from datetime import date if latest_session.get('created_at', '').startswith(date.today().isoformat()): st.session_state.current_session = latest_session else: st.session_state.current_session = None else: st.session_state.current_session = None def log_to_console(message: str, level: str = "INFO") -> None: """Log a message to the session console output.""" timestamp = datetime.now().strftime("%H:%M:%S") st.session_state.console_output.append(f"[{timestamp}] {level}: {message}") def get_console_text() -> str: """Return all console output as a single string.""" return "\n".join(st.session_state.console_output) def clear_console() -> None: """Clear console output.""" st.session_state.console_output = [] def execute_single_operation(operation_index: int) -> None: """ Execute a single operation from the session queue. Args: operation_index: Index of the operation in session's operations list """ if not st.session_state.current_session: st.error("โŒ No active session") return session = st.session_state.current_session operations = session.get('operations', []) if operation_index >= len(operations): st.error(f"โŒ Invalid operation index: {operation_index}") return operation = operations[operation_index] if operation.get('status') != 'queued': st.warning( f"โš ๏ธ Operation already {operation.get('status')}" ) return clear_console() log_to_console( f"๐Ÿš€ Executing operation {operation_index + 1}..." ) # Update status to running operation['status'] = 'running' operation['execution_start'] = datetime.now().isoformat() op_type = operation.get('operation_type', 'unknown') log_to_console(f"{'='*60}") log_to_console(f"โ–ถ๏ธ Operation Type: {op_type}") log_to_console(f"{'='*60}") try: if op_type == 'generation': config = operation.get('run_config', {}) success = execute_generation_operation(config, operation) if success: operation['status'] = 'completed' log_to_console("โœ… Operation completed successfully") else: operation['status'] = 'failed' operation['error'] = "Generation failed" log_to_console("โŒ Operation failed", "ERROR") elif op_type == 'delivery': config = operation.get('run_config', {}) files = operation.get('files', []) success = execute_delivery_operation(config, files, operation) if success: operation['status'] = 'completed' log_to_console("โœ… Operation completed successfully") else: operation['status'] = 'failed' operation['error'] = "Delivery failed" log_to_console("โŒ Operation failed", "ERROR") elif op_type == 'generation+delivery': config = operation.get('run_config', {}) # Generation first log_to_console("๐Ÿ“ Phase 1: Generation...") gen_success = execute_generation_operation(config, operation) if gen_success: log_to_console("โœ… Generation phase completed") # Delivery second log_to_console("๐Ÿ“ค Phase 2: Delivery...") files = operation.get('files', []) del_success = execute_delivery_operation( config, files, operation ) if del_success: operation['status'] = 'completed' log_to_console("โœ… Operation completed successfully") else: operation['status'] = 'failed' operation['error'] = "Delivery phase failed" log_to_console( "โŒ Delivery phase failed", "ERROR" ) else: operation['status'] = 'failed' operation['error'] = "Generation phase failed" log_to_console("โŒ Generation phase failed", "ERROR") else: operation['status'] = 'failed' operation['error'] = f"Unknown operation type: {op_type}" log_to_console( f"โŒ Unknown operation type: {op_type}", "ERROR" ) except Exception as e: operation['status'] = 'failed' operation['error'] = str(e) log_to_console( f"โŒ Operation failed with exception: {e}", "ERROR" ) # Update execution end time operation['execution_end'] = datetime.now().isoformat() # Update session session['operations'][operation_index] = operation st.session_state.current_session = session log_to_console(f"{'='*60}") log_to_console("๐ŸŽ‰ Operation execution complete!") log_to_console(f"{'='*60}") def execute_session_operations() -> None: """Execute all queued operations in the current session.""" if not st.session_state.current_session: st.error("โŒ No active session") return session = st.session_state.current_session operations = session.get('operations', []) if not operations: st.warning("โš ๏ธ No operations to execute") return clear_console() log_to_console(f"๐Ÿš€ Executing {len(operations)} operation(s)...") for i, operation in enumerate(operations): if operation.get('status') != 'queued': log_to_console(f"โญ๏ธ Skipping operation {i+1} (already {operation.get('status')})") continue # Update status to running operation['status'] = 'running' operation['execution_start'] = datetime.now().isoformat() op_type = operation.get('operation_type', 'unknown') log_to_console(f"\n{'='*60}") log_to_console(f"โ–ถ๏ธ Executing Operation {i+1}/{len(operations)}: {op_type}") log_to_console(f"{'='*60}") try: if op_type == 'generation': # Execute generation operation config = operation.get('run_config', {}) success = execute_generation_operation(config, operation) if success: operation['status'] = 'completed' log_to_console(f"โœ… Operation {i+1} completed successfully") else: operation['status'] = 'failed' operation['error'] = "Generation failed" log_to_console(f"โŒ Operation {i+1} failed", "ERROR") elif op_type == 'delivery': # Execute delivery operation config = operation.get('run_config', {}) files = operation.get('files', []) success = execute_delivery_operation(config, files, operation) if success: operation['status'] = 'completed' log_to_console(f"โœ… Operation {i+1} completed successfully") else: operation['status'] = 'failed' operation['error'] = "Delivery failed" log_to_console(f"โŒ Operation {i+1} failed", "ERROR") elif op_type == 'generation+delivery': # Execute both in sequence config = operation.get('run_config', {}) # Generation first log_to_console("๐Ÿ“ Phase 1: Generation...") gen_success = execute_generation_operation(config, operation) if gen_success: log_to_console("โœ… Generation phase completed") # Delivery second log_to_console("๐Ÿ“ค Phase 2: Delivery...") files = operation.get('files', []) del_success = execute_delivery_operation(config, files, operation) if del_success: operation['status'] = 'completed' log_to_console(f"โœ… Operation {i+1} completed successfully") else: operation['status'] = 'failed' operation['error'] = "Delivery phase failed" log_to_console("โŒ Delivery phase failed", "ERROR") else: operation['status'] = 'failed' operation['error'] = "Generation phase failed" log_to_console("โŒ Generation phase failed", "ERROR") else: operation['status'] = 'failed' operation['error'] = f"Unknown operation type: {op_type}" log_to_console(f"โŒ Unknown operation type: {op_type}", "ERROR") except Exception as e: operation['status'] = 'failed' operation['error'] = str(e) log_to_console(f"โŒ Operation {i+1} failed with exception: {e}", "ERROR") # Update execution end time operation['execution_end'] = datetime.now().isoformat() # Update session session['operations'][i] = operation # Update session status all_completed = all( op.get('status') in ['completed', 'failed'] for op in operations ) if all_completed: session['status'] = 'completed' # Save session st.session_state.current_session = session log_to_console(f"\n{'='*60}") log_to_console("๐ŸŽ‰ Session execution complete!") log_to_console(f"{'='*60}") # Show summary completed = sum(1 for op in operations if op.get('status') == 'completed') failed = sum(1 for op in operations if op.get('status') == 'failed') log_to_console(f"โœ… Completed: {completed}") log_to_console(f"โŒ Failed: {failed}") if failed == 0: st.success(f"๐ŸŽ‰ All {completed} operation(s) completed successfully!") else: st.warning(f"โš ๏ธ {completed} completed, {failed} failed") def execute_generation_operation(config: dict, operation: dict) -> bool: """Execute a generation operation using the feed_file_exporter.""" # Use the existing generate_feed_files logic but with operation config # This is a simplified version - you may want to refactor generate_feed_files # to be reusable for operations # For now, just set up environment and call the script try: import subprocess import queue import threading env_vars = os.environ.copy() env_vars.update({ "PYTHONUNBUFFERED": "1", "SNOWFLAKE_USER": config.get("snowflake_user", ""), "SNOWFLAKE_PASSWORD": config.get("snowflake_password", ""), "SNOWFLAKE_ACCOUNT": config.get("snowflake_account", ""), "SNOWFLAKE_ROLE": config.get("snowflake_role", ""), "SNOWFLAKE_WAREHOUSE": config.get("snowflake_warehouse", ""), "SNOWFLAKE_DATABASE": config.get("snowflake_database", ""), "SNOWFLAKE_SCHEMA": config.get("snowflake_schema", ""), "SNOWFLAKE_SOURCE_TABLE": config.get("snowflake_source_table", ""), "PROCESS_MODE": config.get("process_mode", ""), "PERIOD_ID": config.get("period_id", ""), "ZIP_GENERATION_MODE": config.get("zip_generation_mode", "Both"), }) output_queue: queue.Queue = queue.Queue() process_complete = threading.Event() return_code_holder = {"code": None} def run_subprocess(): try: process = subprocess.Popen( ["python", "feed_file_exporter.py"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, env=env_vars, ) for line in iter(process.stdout.readline, ''): if line: output_queue.put(line.rstrip('\n')) process.wait() return_code_holder["code"] = process.returncode finally: process_complete.set() thread = threading.Thread(target=run_subprocess, daemon=True) thread.start() # Process output while not process_complete.is_set() or not output_queue.empty(): try: line = output_queue.get(timeout=0.5) log_to_console(line) except queue.Empty: continue thread.join(timeout=1) return return_code_holder["code"] == 0 except Exception as e: log_to_console(f"โŒ Generation failed: {e}", "ERROR") return False def execute_delivery_operation(config: dict, files: list, operation: dict) -> bool: """Execute a delivery operation.""" # Placeholder for delivery logic log_to_console("๐Ÿ“ค Delivery operation not yet implemented") return True def scrollable_console(text: str, height: int = 300, container_id: str = "console") -> None: """Render scrollable console with auto-scroll to bottom. Args: text: Console text to display height: Height of console in pixels container_id: Unique ID for the console container """ # Escape HTML special characters to prevent injection text_escaped = ( text.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) .replace("'", "'") ) # Auto-scroll only if preference is enabled auto_scroll_script = "" if st.session_state.get("auto_scroll_console", True): auto_scroll_script = f""" """ html = f"""
{text_escaped}
{auto_scroll_script} """ components.html(html, height=height + 30, scrolling=False) def render_session_sidebar(): """Render session/operation sidebar (v2 architecture).""" with st.sidebar: # Global Test Mode Toggle st.markdown("## โš™๏ธ Settings") if LOCAL_SFTP_AVAILABLE: test_mode = st.checkbox( "๐Ÿงช Test Mode (Local SFTP)", value=st.session_state.deliver_config.get("test_mode", False), help="Use local filesystem instead of real SFTP for testing", key="global_test_mode" ) # Update config when changed if test_mode != st.session_state.deliver_config.get("test_mode"): st.session_state.deliver_config["test_mode"] = test_mode if test_mode: # Auto-configure test mode settings st.session_state.deliver_config["sftp_host"] = "localhost" st.session_state.deliver_config["sftp_user"] = "devuser" st.session_state.deliver_config["sftp_password"] = "devpass" st.session_state.deliver_config["sftp_dev_mode"] = True if test_mode: st.success("๐Ÿงช Test Mode Active") st.caption("Files โ†’ `tmp/local_sftp_root/`") st.markdown("---") st.markdown("## ๐Ÿ“ฆ Sessions & Operations") # Load sessions st.session_state.sessions_list = session_mgr.load_sessions() # Current active session if st.session_state.current_session: session = st.session_state.current_session st.markdown(f"### ๐ŸŸข {session['name']}") # Show operation queue operations = session.get('operations', []) if operations: st.markdown(f"**{len(operations)} operation(s)**") for i, op in enumerate(operations): status_emoji = operation_manager.get_operation_status_emoji( op.get('status', 'queued') ) op_type_short = { 'generation': 'Gen', 'delivery': 'Del', 'generation+delivery': 'Gen+Del', }.get(op.get('operation_type', ''), 'Unknown') # Extract key config info for display config = op.get('run_config', {}) period = config.get('period_id', 'N/A') mode = config.get('process_mode', 'N/A') # Create compact title title = f"{i+1}. {status_emoji} {op_type_short} | {mode} | P{period}" with st.expander(title, expanded=False): st.caption(f"**Status:** {op.get('status', 'queued')}") st.caption(f"**Type:** {op.get('operation_type', 'unknown')}") st.caption(f"**Files:** {len(op.get('files', []))}") # Show output directory if available if op.get('output_directory'): st.caption( f"**Location:** `{op['output_directory']}`" ) # Show detailed configuration if config: st.markdown("**Configuration:**") if 'snowflake_source_table' in config: st.caption( f"โ€ข Table: " f"`{config['snowflake_source_table']}`" ) if 'snowflake_database' in config: st.caption( f"โ€ข Database: " f"`{config['snowflake_database']}`" ) if 'snowflake_schema' in config: st.caption( f"โ€ข Schema: " f"`{config['snowflake_schema']}`" ) if 'period_id' in config: st.caption(f"โ€ข Period: {config['period_id']}") if 'process_mode' in config: st.caption(f"โ€ข Mode: {config['process_mode']}") # Show file list if available files = op.get('files', []) if files: with st.expander( f"๐Ÿ“ Generated Files ({len(files)})", expanded=False ): for file_path in files[:10]: # First 10 st.caption(f"โ€ข {file_path}") if len(files) > 10: st.caption( f"... and {len(files) - 10} " f"more files" ) if op.get('error'): st.error(f"โŒ {op['error']}") # Add Run Now button for queued operations if op.get('status') == 'queued': if st.button( "โ–ถ๏ธ Run Now", key=f"run_single_op_{i}", use_container_width=True, help="Execute this operation immediately " "without affecting queue" ): execute_single_operation(i) st.rerun() else: st.info("No operations queued") # Session actions col1, col2 = st.columns(2) with col1: if operations and st.button( "โ–ถ๏ธ Execute", key="exec_all", use_container_width=True, help="Execute all queued operations in sequence" ): execute_session_operations() # DEBUGGING # Don't rerun - let user see console output # st.rerun() with col2: if st.button( "โœ“ Close", key="close_session", use_container_width=True, help="Save and close current session" ): # Save session before closing st.session_state.sessions_list = session_mgr.add_session( st.session_state.current_session, st.session_state.sessions_list ) st.session_state.current_session = None # DEBUGGING # Don't rerun - let user see console output # st.rerun() else: # Show recent sessions if st.session_state.sessions_list: st.markdown( f"**{len(st.session_state.sessions_list)} session(s)**" ) for idx, sess in enumerate( st.session_state.sessions_list[:5] ): summary = session_mgr.get_session_summary(sess) display = ( f"{sess['name']} " f"({summary['total_operations']} ops)" ) with st.expander(display, expanded=False): st.caption(f"Created: {sess['created_at'][:10]}") st.caption( f"Status: {sess.get('status', 'unknown')}" ) st.caption( f"Operations: {summary['total_operations']}" ) # Show operations summary ops = sess.get('operations', []) if ops: st.caption("**Operations:**") for i, op in enumerate(ops[:3]): status_emoji = ( operation_manager .get_operation_status_emoji( op.get('status', 'queued') ) ) op_type = op.get( 'operation_type', 'unknown' ) st.caption( f" {i+1}. {status_emoji} {op_type}" ) if len(ops) > 3: st.caption( f" ... and {len(ops) - 3} more" ) # Reopen button with unique key if st.button( "๐Ÿ“‚ Open Session", key=f"open_{idx}_{sess['id']}", use_container_width=True, help="Reopen this saved session" ): st.session_state.current_session = sess st.success(f"โœ… Opened: {sess['name']}") st.rerun() else: st.info("๐Ÿ’ก Create a session to queue operations") # Session management buttons col1, col2, col3 = st.columns(3) with col1: if st.button( "โž• New", key="new_session_btn", use_container_width=True, help="Create a new empty session" ): session = session_mgr.create_session( name=f"Session {session_mgr.create_session_id()}", description="New session" ) st.session_state.current_session = session st.success( "โœ… Session created! Queue operations from Run or " "Deliver tabs." ) st.rerun() with col2: # Export current sessions to JSON if st.button( "๐Ÿ’พ Export", key="export_sessions_btn", use_container_width=True, help="Download all sessions as JSON" ): import json sessions_json = json.dumps( st.session_state.sessions_list, indent=2 ) st.download_button( label="๐Ÿ“ฅ Download Sessions", data=sessions_json, file_name=( f"sessions_" f"{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" ), mime="application/json", key="download_sessions", use_container_width=True ) with col3: # Import sessions from JSON uploaded = st.file_uploader( "๐Ÿ“‚ Import", type="json", key="import_sessions_file", label_visibility="collapsed", help="Upload sessions JSON file" ) if uploaded: import json try: imported_sessions = json.load(uploaded) if isinstance(imported_sessions, list): st.session_state.sessions_list = imported_sessions session_mgr.save_sessions(imported_sessions) st.success( f"โœ… Imported " f"{len(imported_sessions)} session(s)" ) st.rerun() else: st.error( "โŒ Invalid format: expected list of sessions" ) except Exception as e: st.error(f"โŒ Import failed: {e}") # State Management Controls (bottom of sidebar) st.markdown("---") st.markdown("### โš™๏ธ State Management") col1, col2 = st.columns(2) with col1: if st.button( "๐Ÿ—‘๏ธ Clear All", key="clear_all_state", use_container_width=True, help="Reset all sessions and start fresh" ): st.session_state.current_session = None st.session_state.sessions_list = [] session_mgr.save_sessions([]) st.success("โœ… All state cleared!") st.rerun() with col2: if st.button( "๐Ÿ”„ Reload", key="reload_sessions", use_container_width=True, help="Reload sessions from disk" ): st.session_state.sessions_list = session_mgr.load_sessions() st.success("โœ… Sessions reloaded!") st.rerun() # ============================================================================ # CONFIGURATION MANAGEMENT # ============================================================================ def save_run_config(output_dir: Optional[str] = None) -> str: """Save run configuration with dynamic filename and versioning. Args: output_dir: Optional output directory. If None, uses 'configs' folder. Returns: Path to saved configuration file. """ config = st.session_state.run_config # Filter out sensitive fields before saving SENSITIVE_KEYS = { "snowflake_password", "aws_secret_access_key", } safe_config = {k: v for k, v in config.items() if k not in SENSITIVE_KEYS} table = safe_config.get("snowflake_source_table", "UNKNOWN").replace(".", "_") mode = safe_config.get("process_mode", "UNKNOWN") period = safe_config.get("period_id", "UNKNOWN") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") base_filename = f"{table}-{mode}-{period}-Settings-{timestamp}" # Determine output directory if output_dir: filepath = Path(output_dir) / f"{base_filename}.json" else: filepath = Path("configs") / f"{base_filename}.json" # Handle collisions with versioning if filepath.exists(): version = 1 while filepath.exists(): if output_dir: filepath = Path(output_dir) / f"{base_filename}_{version:02d}.json" else: filepath = Path("configs") / f"{base_filename}_{version:02d}.json" version += 1 filepath.parent.mkdir(parents=True, exist_ok=True) with open(filepath, "w") as f: json.dump(safe_config, f, indent=2) log_to_console(f"๐Ÿ’พ Configuration saved to {filepath} (secrets excluded)") return str(filepath) def load_run_config(filepath: str) -> bool: """Load run configuration from file.""" try: with open(filepath, "r") as f: config = json.load(f) st.session_state.run_config.update(config) log_to_console(f"๐Ÿ“‚ Configuration loaded from {filepath}") return True except Exception as e: st.error(f"โŒ Error loading config: {str(e)}") return False # ============================================================================ # RUN TAB - FEED GENERATION # ============================================================================ def render_run_tab() -> None: """Render the Run tab for feed generation.""" st.markdown("## ๐Ÿ”ง Feed Generation Configuration") st.markdown("Configure and execute feed file generation from Snowflake.") # Snowflake Configuration with st.expander("๐Ÿ—„๏ธ Snowflake Configuration", expanded=True): col1, col2 = st.columns(2) with col1: st.session_state.run_config["snowflake_user"] = st.text_input( "Snowflake User ๐Ÿ‘ค", value=st.session_state.run_config["snowflake_user"], help="Snowflake username for authentication", key="run_sf_user", ) st.session_state.run_config["snowflake_database"] = st.text_input( "Database ๐Ÿ—„๏ธ", value=st.session_state.run_config["snowflake_database"], help="Database containing source tables", key="run_sf_db", ) with col2: st.session_state.run_config["snowflake_password"] = st.text_input( "Snowflake Password ๐Ÿ”", value=st.session_state.run_config.get("snowflake_password", ""), type="password", help="Snowflake password (stored in session only)", key="run_sf_pass", ) st.session_state.run_config["snowflake_schema"] = st.text_input( "Schema ๐Ÿ“‚", value=st.session_state.run_config["snowflake_schema"], help="Schema containing source tables", key="run_sf_schema", ) col1, col2, col3 = st.columns(3) with col1: st.session_state.run_config["snowflake_account"] = st.text_input( "Snowflake Account ๐Ÿข", value=st.session_state.run_config["snowflake_account"], help="Snowflake account identifier", key="run_sf_account", ) with col2: st.session_state.run_config["snowflake_role"] = st.text_input( "Snowflake Role ๐Ÿ‘”", value=st.session_state.run_config["snowflake_role"], help="Snowflake role for execution", key="run_sf_role", ) with col3: st.session_state.run_config["snowflake_warehouse"] = st.text_input( "Snowflake Warehouse ๐Ÿญ", value=st.session_state.run_config["snowflake_warehouse"], help="Snowflake warehouse for compute", key="run_sf_warehouse", ) # AWS Configuration with st.expander("โ˜๏ธ AWS Configuration (Optional)", expanded=False): col1, col2 = st.columns(2) with col1: st.session_state.run_config["aws_access_key_id"] = st.text_input( "AWS Access Key ID ๐Ÿ”‘", value=st.session_state.run_config.get("aws_access_key_id", ""), type="password", help="AWS access key for S3 operations (if needed)", key="run_aws_key", ) with col2: st.session_state.run_config["aws_secret_access_key"] = st.text_input( "AWS Secret Access Key ๐Ÿ”’", value=st.session_state.run_config.get("aws_secret_access_key", ""), type="password", help="AWS secret key for S3 operations (if needed)", key="run_aws_secret", ) # Feed Configuration with st.expander("๐Ÿ“Š Feed Configuration", expanded=True): st.session_state.run_config["snowflake_source_table"] = st.text_input( "Source Table ๐Ÿ“‹", value=st.session_state.run_config["snowflake_source_table"], help="Fully qualified table name (e.g., VIEW_NAME)", key="run_source_table", ) col1, col2 = st.columns(2) with col1: st.session_state.run_config["process_mode"] = st.selectbox( "Processing Mode โš™๏ธ", options=["US", "GB", "AGGREGATE", "EXUS"], index=["US", "GB", "AGGREGATE", "EXUS"].index( st.session_state.run_config.get("process_mode", "US") ), help="Feed processing mode", key="run_mode", ) with col2: st.session_state.run_config["period_id"] = st.text_input( "Period ID ๐Ÿ“…", value=st.session_state.run_config["period_id"], help="Period identifier (e.g., 322, 2025-10)", key="run_period", ) # Advanced Configuration with st.expander("๐Ÿ”ง Advanced Configuration", expanded=False): st.markdown("**File Output & Processing Settings**") # Zip Generation Mode st.session_state.run_config["zip_generation_mode"] = st.selectbox( "Zip Generation Mode ๐Ÿ“ฆ", options=["Individual Files Only", "Zip File Only", "Both"], index=["Individual Files Only", "Zip File Only", "Both"].index( st.session_state.run_config.get( "zip_generation_mode", "Both" ) ), help="Individual Files Only: Keep files unzipped | Zip File Only: Create zip archive and delete source files | Both: Keep individual files AND create zip", key="run_zip_mode", ) st.session_state.run_config["file_output_path"] = st.text_input( "File Output Path ๐Ÿ“‚", value=st.session_state.run_config.get( "file_output_path", "tmp/sony_stars_monthly_feed/{date}" ), help="Output directory path (use {date} placeholder for date)", key="run_output_path", ) col1, col2 = st.columns(2) with col1: st.session_state.run_config["copy_table"] = st.checkbox( "Copy Table ๐Ÿ“‹", value=st.session_state.run_config.get("copy_table", True), help="Create temporary copy of source table", key="run_copy_table", ) env_options = ["DEV", "QA", "PROD"] current_env = st.session_state.run_config.get( "environment", "DEV" ) # Handle unexpected environment values if current_env not in env_options: env_options.append(current_env) st.session_state.run_config["environment"] = st.selectbox( "Environment ๐ŸŒ", options=env_options, index=env_options.index(current_env), help="Runtime environment", key="run_environment", ) with col2: st.session_state.run_config["delete_copy"] = st.checkbox( "Delete Copy After Run ๐Ÿ—‘๏ธ", value=st.session_state.run_config.get("delete_copy", True), help="Delete temporary table after processing", key="run_delete_copy", ) st.session_state.run_config["logger_level"] = st.selectbox( "Logger Level ๐Ÿ“Š", options=["DEBUG", "INFO", "WARNING", "ERROR"], index=["DEBUG", "INFO", "WARNING", "ERROR"].index( st.session_state.run_config.get("logger_level", "INFO") ), help="Logging level for file logs", key="run_logger_level", ) st.session_state.run_config["booking_affiliate"] = st.text_input( "Booking Affiliate ๐Ÿข", value=st.session_state.run_config.get( "booking_affiliate", "US, GB, NONE" ), help="Comma-separated list of booking affiliates", key="run_booking_affiliate", ) st.session_state.run_config["console_log_level"] = st.selectbox( "Console Log Level ๐Ÿ–ฅ๏ธ", options=["DEBUG", "INFO", "WARNING", "ERROR"], index=["DEBUG", "INFO", "WARNING", "ERROR"].index( st.session_state.run_config.get("console_log_level", "INFO") ), help="Logging level for console output", key="run_console_log_level", ) # Debug Mode Toggle st.markdown("---") st.markdown("**๐Ÿ› Debug Options**") st.session_state.run_config["debug_row_processing"] = st.checkbox( "Enable Row Processing Debug ๐Ÿ”", value=st.session_state.run_config.get("debug_row_processing", False), help="โš ๏ธ DEBUG: Show detailed row-by-row processing (very verbose!)", key="run_debug_row_processing", ) if st.session_state.run_config.get("debug_row_processing"): st.warning("โš ๏ธ Debug mode will generate extensive logs. Use only for troubleshooting.") # Configuration Management st.markdown("---") st.markdown("**๐Ÿ’พ Configuration Management**") col1, col2 = st.columns(2) with col1: if st.button("๐Ÿ’พ Save Configuration", use_container_width=True, key="save_config_advanced"): filepath = save_run_config() st.success(f"โœ… Saved to {filepath}") with col2: uploaded_file = st.file_uploader( "๐Ÿ“‚ Load Configuration", type="json", key="run_upload_config_advanced", label_visibility="collapsed", ) if uploaded_file: config_data = json.loads(uploaded_file.read().decode("utf-8")) st.session_state.run_config.update(config_data) st.success("โœ… Configuration loaded") st.rerun() # Show last saved config location config_dir = Path("configs") if config_dir.exists(): configs = sorted(config_dir.glob("*.json"), key=lambda x: x.stat().st_mtime, reverse=True) if configs: st.caption(f"๐Ÿ“ Last saved: {configs[0].name}") st.markdown("---") st.caption( "โš ๏ธ Advanced settings affect feed generation behavior. " "Modify with caution." ) st.markdown("---") # Action Buttons - Session-aware if st.session_state.current_session: # Active session mode - queue operations st.info(f"๐Ÿ“ฆ Active Session: {st.session_state.current_session['name']}") col1, col2 = st.columns([2, 1]) with col1: if st.button( "โž• Queue Generation Operation", use_container_width=True, type="primary", key="queue_gen_op", help="Add a feed generation operation to the session queue" ): # Create operation and add to session operation = operation_manager.create_operation( run_config=st.session_state.run_config, output_directory="", # Will be set after generation files=[], operation_type="generation" ) st.session_state.current_session = session_mgr.add_operation_to_session( st.session_state.current_session, operation ) st.success("โœ… Generation operation queued!") st.rerun() with col2: if st.button( "๐Ÿ’พ Quick Save", use_container_width=True, key="quick_save_btn", help="Save current configuration without running" ): filepath = save_run_config() st.toast(f"โœ… Saved: {filepath}") else: # No active session - direct execution col1, col2 = st.columns([2, 1]) with col1: if st.button( "๐Ÿš€ Generate Feed Files", use_container_width=True, type="primary", key="generate_feeds_btn", help="Generate feed files immediately (creates auto-session)" ): generate_feed_files() with col2: if st.button( "๐Ÿ’พ Quick Save", use_container_width=True, key="quick_save_nosession", help="Save current configuration without running" ): filepath = save_run_config() st.toast(f"โœ… Saved: {filepath}") # Queue Run & Deliver Button (if SFTP enabled and active session exists) if st.session_state.deliver_config.get("sftp_enable", False) and \ st.session_state.current_session: st.markdown("---") col1, col2, col3 = st.columns([1, 2, 1]) with col2: if st.button( "โž•๐Ÿš€๐Ÿ“ค Queue Run & Deliver", use_container_width=True, type="secondary", help="Queue generation and SFTP delivery as a single " "combined operation", key="queue_run_deliver_button" ): # Create combined generation+delivery operation operation = operation_manager.create_operation( run_config=st.session_state.run_config, output_directory="", # Will be set after generation files=[], operation_type="generation+delivery" ) st.session_state.current_session = \ session_mgr.add_operation_to_session( st.session_state.current_session, operation ) st.success("โœ… Generation+Delivery operation queued!") st.rerun() # Execute run and deliver if flag is set if st.session_state.get("run_and_deliver", False): st.session_state.run_and_deliver = False # Generate without spinner - console will show progress generate_feed_files() # Check if generation succeeded and session was created if st.session_state.selected_session_id: st.info( "โœ… Generation complete! Switching to Deliver tab..." ) # Switch to Deliver tab st.session_state.active_tab = "Deliver" time.sleep(0.5) # Brief pause st.rerun() else: st.error("โŒ Generation failed - no transfer performed") st.rerun() # Console output section header and controls st.markdown("---") col1, col2, col3 = st.columns([5, 2, 1]) with col1: st.markdown("### ๐Ÿ“‹ Console Output") with col2: st.session_state.auto_scroll_console = st.checkbox( "๐Ÿ”ฝ Auto-scroll", value=st.session_state.get("auto_scroll_console", True), help="Automatically scroll to latest log entries", key="run_auto_scroll_toggle" ) with col3: if st.button("๐Ÿงน Clear", key="run_clear_console"): clear_console() st.rerun() # Placeholder for dynamic console updates (full width) if "run_console_placeholder" not in st.session_state: st.session_state.run_console_placeholder = st.empty() # Render console - either static text_area or live HTML console_text = get_console_text() if console_text: st.session_state.run_console_placeholder.text_area( "Run Output Log", value=console_text, height=300, disabled=True, key="run_console_display", label_visibility="collapsed", ) # Show refresh button after execution if st.session_state.get("run_execution_completed"): if st.button("๐Ÿ”„ Update Session and Show Log", key="refresh_run_session", use_container_width=True, type="primary"): # Reload sessions st.session_state.sessions_list = session_mgr.load_sessions() # Clear completion flag st.session_state.run_execution_completed = False st.success("โœ… Sessions updated!") st.rerun() def generate_feed_files() -> None: """Execute feed file generation with real-time output streaming. Uses threading to run subprocess without blocking Streamlit's UI loop, with a queue-based approach to capture and display output in real-time. """ import time # Used for timing and sleep operations clear_console() log_to_console("๐Ÿš€ Starting feed file generation...") config = st.session_state.run_config # Validate required fields before proceeding required_fields = { "snowflake_user": "Snowflake User", "snowflake_account": "Snowflake Account", "snowflake_role": "Snowflake Role", "snowflake_warehouse": "Snowflake Warehouse", "snowflake_database": "Snowflake Database", "snowflake_source_table": "Source Table", "period_id": "Period ID", } missing = [ label for key, label in required_fields.items() if not config.get(key, "").strip() ] if missing: error_msg = f"โŒ Missing required fields: {', '.join(missing)}" log_to_console(error_msg, "ERROR") st.error(error_msg) return # Build environment variables (include PYTHONUNBUFFERED for real-time output) env_vars = os.environ.copy() env_vars.update({ "PYTHONUNBUFFERED": "1", # Critical for real-time output "SNOWFLAKE_USER": config["snowflake_user"], "SNOWFLAKE_PASSWORD": config.get("snowflake_password", ""), "SNOWFLAKE_ACCOUNT": config["snowflake_account"], "SNOWFLAKE_ROLE": config["snowflake_role"], "SNOWFLAKE_WAREHOUSE": config["snowflake_warehouse"], "SNOWFLAKE_DATABASE": config["snowflake_database"], "SNOWFLAKE_SCHEMA": config["snowflake_schema"], "SNOWFLAKE_SOURCE_TABLE": config["snowflake_source_table"], "PROCESS_MODE": config["process_mode"], "PERIOD_ID": config["period_id"], "ZIP_GENERATION_MODE": config.get("zip_generation_mode", "Both"), "DEBUG_ROW_PROCESSING": "True" if config.get("debug_row_processing", False) else "False", }) if config.get("aws_access_key_id"): env_vars["AWS_ACCESS_KEY_ID"] = config["aws_access_key_id"] if config.get("aws_secret_access_key"): env_vars["AWS_SECRET_ACCESS_KEY"] = config["aws_secret_access_key"] log_to_console(f"๐Ÿ“Š Source: {config['snowflake_source_table']}") log_to_console(f"โš™๏ธ Mode: {config['process_mode']}") log_to_console(f"๐Ÿ“… Period: {config['period_id']}") log_to_console(f"๐Ÿข Account: {config['snowflake_account']}") # Queue for thread-safe output capture output_queue: queue.Queue = queue.Queue() process_complete = threading.Event() return_code_holder = {"code": None, "error": None} def run_subprocess(): """Run the feed generation subprocess in a background thread.""" try: process = subprocess.Popen( ["python", "-u", "feed_file_exporter.py"], # -u for unbuffered env=env_vars, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) # Read output line by line for line in iter(process.stdout.readline, ''): if line: output_queue.put(line.rstrip()) process.stdout.close() return_code_holder["code"] = process.wait(timeout=3600) except subprocess.TimeoutExpired: try: process.kill() except Exception: pass return_code_holder["error"] = "Process timed out after 1 hour" except Exception as e: return_code_holder["error"] = str(e) finally: process_complete.set() # Start subprocess in background thread thread = threading.Thread(target=run_subprocess, daemon=True) thread.start() # Show initial console state immediately (non-blocking) status_placeholder = st.empty() console_placeholder = st.empty() with console_placeholder.container(): scrollable_console( get_console_text(), height=300, container_id="run_console_initial" ) # Wait for thread to complete while draining output queue # This is still blocking but doesn't use st.spinner while not process_complete.is_set(): # Drain the queue without UI updates (faster) try: line = output_queue.get(timeout=0.1) if line.strip(): log_to_console(line) except queue.Empty: pass # Final drain of queue while True: try: line = output_queue.get_nowait() if line.strip(): log_to_console(line) except queue.Empty: break # Report final status (log first, then show in UI) if return_code_holder["error"]: log_to_console(f"โŒ Error: {return_code_holder['error']}", "ERROR") elif return_code_holder["code"] == 0: log_to_console("โœ… Feed generation completed successfully") log_to_console("๐Ÿ“ Check output directory for generated files") # Create session from generated files try: # The feed_file_exporter creates paths under tmp/sony_stars_monthly_feed/ # with date (YYYYMMDD) and then subdirectories based on timestamp from datetime import datetime as dt import config as app_config # Build the expected base path with current date proc_mode = st.session_state.run_config.get("process_mode", "US") base_path = app_config.FILE_OUTPUT_PATH.replace( '{date}', dt.now().strftime('%Y%m%d') ) proc_mode_lower = proc_mode.lower().replace('-', '_') expected_base = Path(base_path) / proc_mode_lower generated_files = [] output_dir = None log_to_console("๐Ÿ” Discovering generated files...") # Search for most recent files in the expected path if expected_base.exists(): log_to_console(f"๐Ÿ“‚ Scanning: {expected_base}") # Find all files recursively, sorted by modification time all_files = [] for file in expected_base.rglob("*"): if file.is_file() and not file.name.startswith('.'): all_files.append(file) # Sort by modification time, newest first all_files.sort(key=lambda f: f.stat().st_mtime, reverse=True) # Take files modified in the last 10 minutes cutoff_time = time.time() - 600 # 10 minutes ago recent_files = [ f for f in all_files if f.stat().st_mtime > cutoff_time ] if recent_files: # Use the directory of the first file as output_dir output_dir = recent_files[0].parent # Store absolute paths to ensure they can be found later generated_files = [str(f.absolute()) for f in recent_files] log_to_console(f" Found: {len(recent_files)} file(s)") for f in recent_files[:5]: # Log first 5 log_to_console(f" โœ“ {f.name}") if len(recent_files) > 5: log_to_console(f" ... and {len(recent_files) - 5} more") # Fallback to ./output if nothing found if not generated_files: fallback_dir = Path("./output") if fallback_dir.exists(): log_to_console(f"๐Ÿ“‚ Scanning fallback: {fallback_dir}") output_dir = fallback_dir for file in fallback_dir.rglob("*"): if file.is_file() and not file.name.startswith('.'): generated_files.append(str(file.absolute())) if not output_dir: output_dir = expected_base if expected_base.exists() else Path("./output") log_to_console(f"๐Ÿ“Š Total files found: {len(generated_files)}") # Create session even if no files (for tracking) session = session_mgr.create_session( name=f"Feed Gen {st.session_state.run_config.get('process_mode', 'US')} P{st.session_state.run_config.get('period_id', 'N/A')}", description=f"Generated from {st.session_state.run_config.get('snowflake_source_table', 'N/A')}" ) # Add generation operation to session operation = operation_manager.create_operation( run_config=st.session_state.run_config.copy(), output_directory=str(output_dir.absolute()), files=generated_files, operation_type="generation" ) operation['status'] = 'completed' session = session_mgr.add_operation_to_session(session, operation) # Save session st.session_state.sessions_list = session_mgr.add_session( session, st.session_state.sessions_list ) st.session_state.current_session = session log_to_console(f"๐Ÿ“ Session created: {session['name']}") # Auto-save configuration to output directory try: _config_saved_path = save_run_config( # noqa: F841 output_dir=str(output_dir.absolute()) ) log_to_console("๐Ÿ’พ Configuration auto-saved to output") except Exception as e: log_to_console(f"โš ๏ธ Config save error: {e}", "WARNING") if not generated_files: log_to_console("โš ๏ธ No recent files found") log_to_console("๐Ÿ’ก Check console output for generation errors") except Exception as e: log_to_console(f"โš ๏ธ Session creation error: {e}", "ERROR") log_to_console("๐Ÿ’ก Continuing without session...") else: msg = f"โŒ Feed generation failed with exit code {return_code_holder['code']}" log_to_console(msg, "ERROR") # Clear spinner and show final status in the status placeholder if return_code_holder["error"]: status_placeholder.error(f"โŒ {return_code_holder['error']}") elif return_code_holder["code"] == 0: status_placeholder.success("โœ… Feed files generated successfully!") else: status_placeholder.error("โŒ Generation failed. Check console for details.") # Replace HTML console with simple text_area showing final output console_text = get_console_text() console_placeholder.empty() # Clear the HTML component console_placeholder.text_area( "Run Output Log", value=console_text, height=300, disabled=True, key=f"run_final_{int(time.time() * 1000)}", label_visibility="collapsed" ) # Set flag to show refresh button st.session_state.run_execution_completed = True # ============================================================================ # DELIVER TAB - SFTP TRANSFER # ============================================================================ def render_deliver_tab() -> None: """Render the Deliver tab for SFTP transfer.""" st.markdown("## ๐Ÿ“ค SFTP File Delivery") st.markdown("Configure and execute SFTP transfers to remote servers.") # Show test mode indicator if active if st.session_state.deliver_config.get("test_mode"): st.info("๐Ÿงช **Test Mode Active**: Files will be saved to `tmp/local_sftp_root/`") # SFTP Enable Toggle st.session_state.deliver_config["sftp_enable"] = st.toggle( "Enable SFTP Transfer ๐Ÿ”„", value=st.session_state.deliver_config["sftp_enable"], help="Enable SFTP file transfer functionality", key="deliver_sftp_enable", ) if not st.session_state.deliver_config["sftp_enable"]: st.info("โ„น๏ธ SFTP transfer is disabled. Enable it to configure and transfer files.") return # Server Configuration with st.expander("๐ŸŒ Server Configuration", expanded=True): col1, col2 = st.columns(2) with col1: st.session_state.deliver_config["sftp_host"] = st.text_input( "SFTP Host ๐Ÿ–ฅ๏ธ", value=st.session_state.deliver_config["sftp_host"], help="Remote SFTP server hostname or IP", key="deliver_host", ) st.session_state.deliver_config["sftp_folder"] = st.text_input( "Remote Folder ๐Ÿ“‚", value=st.session_state.deliver_config["sftp_folder"], help="Target directory on remote server", key="deliver_folder", ) with col2: st.session_state.deliver_config["sftp_user"] = st.text_input( "Username ๐Ÿ‘ค", value=st.session_state.deliver_config["sftp_user"], help="SFTP username for authentication", key="deliver_user", ) st.session_state.deliver_config["sftp_timeout"] = st.number_input( "Timeout (seconds) โฑ๏ธ", value=st.session_state.deliver_config["sftp_timeout"], min_value=30, max_value=14400, step=60, help="Connection and transfer timeout", key="deliver_timeout", ) # Authentication with st.expander("๐Ÿ” Authentication", expanded=True): auth_method = st.radio( "Authentication Method", ["Password", "Private Key"], help="Choose authentication method", key="deliver_auth_method", ) if auth_method == "Password": st.session_state.deliver_config["sftp_password"] = st.text_input( "Password ๐Ÿ”’", value=st.session_state.deliver_config.get("sftp_password", ""), type="password", help="SFTP password (stored in session only)", key="deliver_password", ) else: # Use st.text_input with a button to browse col1, col2 = st.columns([3, 1]) with col1: st.session_state.deliver_config["sftp_private_key_path"] = st.text_input( "Private Key Path ๐Ÿ”‘", value=st.session_state.deliver_config.get("sftp_private_key_path", ""), help="Path to private key file", key="deliver_privkey", ) with col2: st.markdown("
", unsafe_allow_html=True) if st.button("๐Ÿ“ Browse", key="browse_privkey"): st.info("๐Ÿ’ก Enter the full path to your private key file") # Transfer Settings with st.expander("โš™๏ธ Transfer Settings", expanded=False): col1, col2 = st.columns(2) with col1: st.session_state.deliver_config["sftp_max_retries"] = st.number_input( "Max Retries ๐Ÿ”„", value=st.session_state.deliver_config["sftp_max_retries"], min_value=0, max_value=10, help="Number of retry attempts after initial failure", key="deliver_retries", ) failure_mode = st.session_state.deliver_config.get( "sftp_partial_failure_mode", "FAIL" ) st.session_state.deliver_config["sftp_partial_failure_mode"] = st.radio( "File Upload Failure Handling โš ๏ธ", options=["FAIL", "CONTINUE"], index=0 if failure_mode == "FAIL" else 1, help="FAIL: Stop on first error | CONTINUE: Skip failed files", key="deliver_failure_mode", horizontal=True, ) with col2: st.session_state.deliver_config["sftp_retry_backoff"] = st.number_input( "Retry Backoff (sec) โณ", value=st.session_state.deliver_config["sftp_retry_backoff"], min_value=0.5, max_value=10.0, step=0.5, help="Base seconds for exponential backoff", key="deliver_backoff", ) st.session_state.deliver_config["sftp_dev_mode"] = st.checkbox( "Dev Mode (Skip Host Key Check) ๐Ÿ”“", value=st.session_state.deliver_config["sftp_dev_mode"], help="โš ๏ธ WARNING: Only use in development! Disables host key verification", key="deliver_dev_mode", ) # Advanced Configuration with st.expander("๐Ÿ”ง Advanced Configuration", expanded=False): st.markdown("**Transfer Mode & Additional Settings**") # Transfer Mode Selection st.session_state.deliver_config["transfer_mode"] = st.selectbox( "Transfer Mode ๐Ÿ“ฆ", options=["Individual Files", "Zip File", "Both"], index=["Individual Files", "Zip File", "Both"].index( st.session_state.deliver_config.get("transfer_mode", "Individual Files") ), help="Individual Files: Transfer non-zipped files | Zip File: Transfer zip archive only | Both: Transfer both types", key="deliver_transfer_mode", ) # Upload Selection (when both zip and individual files exist) if st.session_state.deliver_config.get("transfer_mode") == "Both": st.session_state.deliver_config["upload_selection"] = st.selectbox( "Upload Selection (When Both Exist) ๐ŸŽฏ", options=["Individual Files", "Zip File", "Both"], index=["Individual Files", "Zip File", "Both"].index( st.session_state.deliver_config.get( "upload_selection", "Individual Files" ) ), help="Choose what to upload when both individual files and zip exist", key="deliver_upload_selection", ) # Preserve Files After Upload st.session_state.deliver_config["preserve_files_after_upload"] = st.checkbox( "Preserve Files After Upload ๐Ÿ’พ", value=st.session_state.deliver_config.get( "preserve_files_after_upload", True ), help="Keep files after successful upload (unchecked = delete)", key="deliver_preserve_files", ) col1, col2 = st.columns(2) with col1: st.session_state.deliver_config["sftp_port"] = st.number_input( "SFTP Port ๐Ÿ”Œ", value=st.session_state.deliver_config.get("sftp_port", 22), min_value=1, max_value=65535, help="SFTP server port (default: 22)", key="deliver_port", ) st.session_state.deliver_config["sftp_strict_host_key"] = st.checkbox( "Strict Host Key Verification ๐Ÿ”", value=st.session_state.deliver_config.get("sftp_strict_host_key", True), help="Enforce SSH host key verification (recommended for production)", key="deliver_strict_host_key", ) st.session_state.deliver_config["sftp_mock"] = st.checkbox( "Mock Mode (No Network) ๐ŸŽญ", value=st.session_state.deliver_config.get("sftp_mock", False), help="Simulate SFTP transfers without real network connection", key="deliver_mock", ) with col2: st.session_state.deliver_config["sftp_prompt_password"] = st.checkbox( "Prompt for Password ๐Ÿ’ฌ", value=st.session_state.deliver_config.get("sftp_prompt_password", False), help="Prompt interactively for password if missing (DEV only)", key="deliver_prompt_password", ) st.markdown("---") st.caption("โš ๏ธ Advanced settings affect SFTP transfer behavior. Modify with caution.") # Operation Selection for Delivery st.markdown("---") st.markdown("### ๐Ÿ“‹ Select Operations to Deliver") # Session-aware operation selection if st.session_state.current_session: session = st.session_state.current_session operations = session.get('operations', []) # Filter to show only generation operations with completed status deliverable_ops = [ op for op in operations if op.get('operation_type') in ['generation', 'generation+delivery'] and op.get('status') == 'completed' and op.get('files') ] if not deliverable_ops: st.warning("โš ๏ธ No completed generation operations with files available in this session.") st.info("๐Ÿ’ก Run feed generation first in the Run tab, then return here to deliver.") return st.info(f"๐Ÿ“ฆ Active Session: {session['name']} - {len(deliverable_ops)} deliverable operation(s)") # Operation selection interface st.markdown("**Select operations to deliver:**") # Initialize selection state if "selected_operations" not in st.session_state: st.session_state.selected_operations = [] # Select all / None buttons col1, col2 = st.columns(2) with col1: if st.button( "โœ… Select All", use_container_width=True, key="select_all_ops", help="Select all deliverable operations" ): st.session_state.selected_operations = [i for i in range(len(deliverable_ops))] st.rerun() with col2: if st.button( "โŒ Clear Selection", use_container_width=True, key="clear_all_ops", help="Deselect all operations" ): st.session_state.selected_operations = [] st.rerun() # Display each operation with checkbox for i, op in enumerate(deliverable_ops): config = op.get('run_config', {}) mode = config.get('process_mode', 'N/A') period = config.get('period_id', 'N/A') table = config.get('snowflake_source_table', 'N/A') file_count = len(op.get('files', [])) is_selected = i in st.session_state.selected_operations col_check, col_info = st.columns([1, 9]) with col_check: if st.checkbox( "", value=is_selected, key=f"op_checkbox_{i}", label_visibility="collapsed" ): if i not in st.session_state.selected_operations: st.session_state.selected_operations.append(i) else: if i in st.session_state.selected_operations: st.session_state.selected_operations.remove(i) with col_info: with st.expander(f"Operation {i+1}: {mode} | P{period} ({file_count} files)", expanded=False): st.caption(f"**Table:** {table}") st.caption(f"**Mode:** {mode}") st.caption(f"**Period:** {period}") st.caption(f"**Files:** {file_count}") # Show output directory if op.get('output_directory'): st.caption(f"**Location:** `{op['output_directory']}`") # Show file paths in expandable section files = op.get('files', []) if files: with st.expander(f"๐Ÿ“ File Paths ({len(files)} files)", expanded=False): # Get absolute paths abs_files = operation_manager.get_operation_files_absolute(op) for f in abs_files[:10]: # Show first 10 with full paths st.caption(f"โ€ข {f}") if len(abs_files) > 10: st.caption(f"... and {len(abs_files) - 10} more files") # Collect all files from selected operations all_selected_files = [] for i in st.session_state.selected_operations: if i < len(deliverable_ops): files = deliverable_ops[i].get('files', []) all_selected_files.extend(files) st.session_state.selected_files = all_selected_files # Show summary if st.session_state.selected_operations: st.success(f"โœ… {len(st.session_state.selected_operations)} operation(s) selected โ†’ {len(all_selected_files)} total files") else: st.info("๐Ÿ’ก Select operations above to deliver their files") else: # No active session - show message st.info("๐Ÿ’ก Create a session in the sidebar and run feed generation first.") st.info("๐Ÿ“ Generated files from operations will appear here for delivery.") return st.markdown("---") # Transfer Execution if st.session_state.selected_files and st.button( "๐Ÿš€ Transfer Files via SFTP", use_container_width=True, type="primary", disabled=len(st.session_state.get("selected_files", [])) == 0, help="Initiate SFTP transfer of selected files" ): st.session_state.show_sftp_confirm = True # Confirmation Modal if st.session_state.get("show_sftp_confirm", False): with st.container(): st.warning("โš ๏ธ **Confirm SFTP Transfer**") st.write(f"Transfer {len(st.session_state.selected_files)} files to:") st.write(f"- Host: {st.session_state.deliver_config['sftp_host']}") st.write(f"- User: {st.session_state.deliver_config['sftp_user']}") st.write(f"- Folder: {st.session_state.deliver_config['sftp_folder']}") col1, col2, col3 = st.columns(3) with col1: if st.button( "โœ… Confirm Transfer", use_container_width=True, type="primary", help="Proceed with SFTP file transfer" ): st.session_state.show_sftp_confirm = False # Clear console before starting transfer clear_console() # Set transfer in progress flag st.session_state.transfer_in_progress = True st.rerun() with col2: if st.button( "โŒ Cancel", use_container_width=True, help="Cancel SFTP transfer" ): st.session_state.show_sftp_confirm = False st.rerun() # Execute transfer if flag is set if st.session_state.get("transfer_in_progress", False): st.session_state.transfer_in_progress = False # Execute without spinner - console will show progress execute_sftp_transfer() # Rerun to show final console output st.rerun() # Console output section header and controls st.markdown("---") col1, col2, col3 = st.columns([5, 2, 1]) with col1: st.markdown("### ๐Ÿ“‹ Console Output") with col2: st.session_state.auto_scroll_console = st.checkbox( "๐Ÿ”ฝ Auto-scroll", value=st.session_state.get("auto_scroll_console", True), help="Automatically scroll to latest log entries", key="deliver_auto_scroll_toggle" ) with col3: if st.button( "๐Ÿงน Clear", key="deliver_clear_console", help="Clear console output" ): clear_console() st.rerun() # Placeholder for dynamic console updates (full width) if "deliver_console_placeholder" not in st.session_state: st.session_state.deliver_console_placeholder = st.empty() # Render console - static text_area showing current content console_text = get_console_text() if console_text: st.session_state.deliver_console_placeholder.text_area( "Delivery Output Log", value=console_text, height=300, disabled=True, key="deliver_console_display", label_visibility="collapsed", ) # Show refresh button after execution if st.session_state.get("deliver_execution_completed"): if st.button( "๐Ÿ”„ Update Session and Show Log", key="refresh_deliver_session", use_container_width=True, type="primary", help="Reload session data and refresh display" ): # Reload sessions st.session_state.sessions_list = session_mgr.load_sessions() # Clear completion flag st.session_state.deliver_execution_completed = False st.success("โœ… Sessions updated!") st.rerun() def execute_sftp_transfer() -> None: """Execute SFTP file transfer with real-time console updates.""" # Don't clear console here - it should be cleared when button is pressed # clear_console() # Removed to preserve console during rerun config = st.session_state.deliver_config files = st.session_state.selected_files # Enable local SFTP monkeypatch if in test mode undo_monkeypatch = None if config.get("test_mode") and LOCAL_SFTP_AVAILABLE: log_to_console("๐Ÿงช Test Mode: Enabling local SFTP server") tmp_root = Path("tmp/local_sftp_root") tmp_root.mkdir(parents=True, exist_ok=True) undo_monkeypatch = enable_local_sftp_monkeypatch( tmp_root, config["sftp_user"], config["sftp_password"] ) log_to_console(f"๐Ÿ“‚ Files will be saved to: {tmp_root.absolute()}") log_to_console("๐Ÿš€ Initiating SFTP transfer...") # Get placeholder for real-time updates stored_placeholder = st.session_state.get("deliver_console_placeholder") console_placeholder = stored_placeholder or st.empty() status_placeholder = st.empty() def update_console(): """Update console placeholder with current output.""" with console_placeholder.container(): scrollable_console( get_console_text(), height=300, container_id=f"deliver_console_{int(time.time() * 1000)}" ) log_to_console(f"๐Ÿ“ฆ Files to transfer: {len(files)}") msg = f"๐Ÿ”Œ Connecting to {config['sftp_host']} " msg += f"as {config['sftp_user']}" log_to_console(msg) update_console() try: sftp_config = SFTPConfig( host=config["sftp_host"], port=config.get("sftp_port", 22), username=config["sftp_user"], # TODO: DEPRECATE - Remove password param when key-only auth password=config.get("sftp_password") or None, remote_dir=config["sftp_folder"], retries=config["sftp_max_retries"], backoff_sec=config["sftp_retry_backoff"], timeout=float(config["sftp_timeout"]), strict_host_key=config.get("sftp_strict_host_key", True), continue_on_error=( config.get("sftp_partial_failure_mode", "FAIL") == "CONTINUE" ), auth_mode=config.get("sftp_auth_mode", "key"), private_key_path=config.get( "sftp_private_key_path", "~/.ssh/id_rsa" ), ) mgr = SFTPTransferManager(sftp_config) log_to_console("๐Ÿ”— Establishing connection...") update_console() mgr.connect() log_to_console("โœ… Connected successfully") update_console() log_to_console(f"๐Ÿ“ค Uploading {len(files)} file(s)...") update_console() result = mgr.upload_files([Path(f) for f in files]) log_to_console("โœ… Transfer complete") msg = f"๐Ÿ“Š Results: {result.uploaded}/{result.total} uploaded" log_to_console(msg) update_console() if result.failed: failed_str = ', '.join(str(f) for f in result.failed) log_to_console(f"โš ๏ธ Failed files: {failed_str}") update_console() mgr.close() log_to_console("๐Ÿ”’ Connection closed") # Test mode success message if config.get("test_mode"): tmp_root = Path("tmp/local_sftp_root") log_to_console("") log_to_console("๐Ÿงช Test Mode: Files saved locally") log_to_console(f"๐Ÿ“‚ Location: {tmp_root.absolute()}") remote = config['sftp_folder'] log_to_console(f"๐Ÿ’ก Run: ls -la {tmp_root}/{remote}") update_console() success_msg = ( f"โœ… Transfer successful! " f"{result.uploaded}/{result.total} files uploaded" ) error_msg = ( f"โŒ Transfer completed with errors. " f"{result.uploaded}/{result.total} files uploaded" ) if result.success(): status_placeholder.success(success_msg) # Update session status to delivered if st.session_state.selected_session_id: st.session_state.sessions = session_mgr.update_session_status( st.session_state.selected_session_id, 'delivered', st.session_state.sessions, delivery_info={ 'delivered_at': datetime.now().isoformat(), 'host': config['sftp_host'], 'files_uploaded': result.uploaded, 'files_total': result.total, } ) log_to_console(f"๐Ÿ“ Session {st.session_state.selected_session_id} marked as delivered") else: status_placeholder.error(error_msg) # Update session status to failed if st.session_state.selected_session_id: st.session_state.sessions = session_mgr.update_session_status( st.session_state.selected_session_id, 'failed', st.session_state.sessions, delivery_info={ 'failed_at': datetime.now().isoformat(), 'error': 'Partial failure', 'files_uploaded': result.uploaded, 'files_failed': len(result.failed), } ) # Replace HTML console with simple text_area showing final output console_text = get_console_text() console_placeholder.empty() # Clear the HTML component console_placeholder.text_area( "Delivery Output Log", value=console_text, height=300, disabled=True, key=f"deliver_final_{int(time.time() * 1000)}", label_visibility="collapsed" ) # Set flag to show refresh button st.session_state.deliver_execution_completed = True except Exception as e: log_to_console(f"โŒ Error: {str(e)}", "ERROR") status_placeholder.error(f"โŒ Transfer failed: {str(e)}") # Replace HTML console with simple text_area showing final output console_text = get_console_text() console_placeholder.empty() # Clear the HTML component console_placeholder.text_area( "Delivery Output Log", value=console_text, height=300, disabled=True, key=f"deliver_error_{int(time.time() * 1000)}", label_visibility="collapsed" ) # Set flag to show refresh button even on error st.session_state.deliver_execution_completed = True finally: # Restore original paramiko if monkeypatched if undo_monkeypatch: undo_monkeypatch() # ============================================================================ # MAIN APP # ============================================================================ def main() -> None: """Main Streamlit application.""" st.set_page_config( page_title="SME Feed File Exporter", page_icon="๐Ÿ“ค", layout="wide", initial_sidebar_state="collapsed", ) init_session_state() # Render session sidebar render_session_sidebar() # Header st.markdown("# ๐Ÿ“ค SME Feed File Exporter") st.caption(f"v{__version__}") st.markdown( "Generate feed files from Snowflake and deliver them via SFTP " "with real-time monitoring." ) st.markdown("---") # Initialize active tab state if "active_tab" not in st.session_state: st.session_state.active_tab = "Run" # Tabs with programmatic control tab_index = 0 if st.session_state.active_tab == "Run" else 1 # Use radio buttons for tab selection (hidden, controlled by state) selected_tab = st.radio( "Navigation", ["๐Ÿš€ Run", "๐Ÿ“ค Deliver"], index=tab_index, key="tab_selector", horizontal=True, label_visibility="collapsed" ) # Update active tab based on selection if selected_tab == "๐Ÿš€ Run": st.session_state.active_tab = "Run" render_run_tab() else: st.session_state.active_tab = "Deliver" render_deliver_tab() # Footer st.markdown("---") st.caption( "๐Ÿ” Passwords and sensitive data are stored in session memory only. " "Configurations can be saved locally as JSON (no secrets included)." ) if __name__ == "__main__": main()