import streamlit as st import sys import os import time from datetime import datetime # Page configuration - MUST be the first Streamlit command st.set_page_config( page_title="CropDrive™ AI Assistant", page_icon="🌴", layout="wide", initial_sidebar_state="expanded" ) # Ensure project root and utils are on sys.path repo_root = os.path.dirname(__file__) if repo_root not in sys.path: sys.path.insert(0, repo_root) utils_dir = os.path.join(repo_root, 'utils') if utils_dir not in sys.path: sys.path.append(utils_dir) # Import utilities with robust error handling try: from utils.firebase_config import initialize_firebase, initialize_admin_codes except ImportError: try: from firebase_config import initialize_firebase, initialize_admin_codes except ImportError as e: print(f"Warning: Failed to import firebase_config: {e}") initialize_firebase = None initialize_admin_codes = None try: from utils.translations import translate, t, get_language, set_language, toggle_language except ImportError: try: from translations import translate, t, get_language, set_language, toggle_language except ImportError as e: print(f"Warning: Failed to import translations: {e}") # Provide fallback functions def translate(text, lang=None): return text def t(key): return key def get_language(): return "en" def set_language(lang): pass def toggle_language(): return "en" # Import CropDrive integration try: from utils.cropdrive_integration import ( initialize_integration, send_analysis_complete, is_feature_available, send_feature_restricted, get_user_plan, get_user_id, request_config_update ) CROPDRIVE_INTEGRATION_AVAILABLE = True except ImportError: try: from cropdrive_integration import ( initialize_integration, send_analysis_complete, is_feature_available, send_feature_restricted, get_user_plan, get_user_id, request_config_update ) CROPDRIVE_INTEGRATION_AVAILABLE = True except ImportError: CROPDRIVE_INTEGRATION_AVAILABLE = False # Fallback functions def initialize_integration(): return 'en', 'none', [] def send_analysis_complete(*args, **kwargs): pass def is_feature_available(*args, **kwargs): return True def send_feature_restricted(*args, **kwargs): pass def get_user_plan(): return 'none' def get_user_id(): return '' import json from datetime import datetime # Ensure Gemini API key is available to LLM clients (HF Spaces + Streamlit secrets) try: api_key = None for env_name in ( "GOOGLE_AI_API_KEY", "GOOGLE_API_KEY", "GEMINI_API_KEY", "GOOGLE_GEMINI_API_KEY", "GENAI_API_KEY", ): api_key = os.getenv(env_name) if api_key: break # Fallback to Streamlit secrets if env var not set if not api_key: try: from utils.secrets_utils import get_secret_section, has_secret google_ai = get_secret_section("google_ai") if google_ai is not None: api_key = ( google_ai.get("api_key") or google_ai.get("google_api_key") or google_ai.get("gemini_api_key") or google_ai.get("GOOGLE_AI_API_KEY") or google_ai.get("GEMINI_API_KEY") ) if not api_key: import streamlit as st for key in ("GOOGLE_AI_API_KEY", "GOOGLE_API_KEY", "GEMINI_API_KEY"): try: if has_secret(key) or (hasattr(st, "secrets") and key in st.secrets): api_key = st.secrets[key] if api_key: break except Exception: continue except Exception: pass if api_key: api_key = str(api_key).strip() if not os.getenv("GOOGLE_API_KEY"): os.environ["GOOGLE_API_KEY"] = api_key if not os.getenv("GOOGLE_AI_API_KEY"): os.environ["GOOGLE_AI_API_KEY"] = api_key except Exception: pass # Add modules to path modules_dir = os.path.join(repo_root, 'modules') if modules_dir not in sys.path: sys.path.append(modules_dir) # Import pages with robust fallbacks try: from modules.upload import show_upload_page except Exception: from upload import show_upload_page try: from modules.results import show_results_page as results_page_func except Exception as e: print(f"Warning: Could not import results module: {e}") import traceback traceback.print_exc() results_page_func = None try: from modules.admin import show_admin_panel as admin_panel_func except Exception as e: print(f"Warning: Could not import admin module: {e}") import traceback traceback.print_exc() admin_panel_func = None try: from modules.history import show_history_page except Exception as e: print(f"Warning: Could not import history module: {e}") import traceback traceback.print_exc() show_history_page = None # Page configuration is now at the top of the file # Note: Browser console warnings about "Unrecognized feature" for Permissions Policy # (ambient-light-sensor, battery, document-domain, etc.) are harmless and come from # Streamlit's default security headers. They do not affect functionality. # Fix accessibility issues: Ensure all form inputs have id/name attributes and proper labels st.markdown(""" """, unsafe_allow_html=True) # Hide Streamlit branding and user profile elements st.markdown( """ """, unsafe_allow_html=True, ) # Custom CSS for better styling st.markdown(""" """, unsafe_allow_html=True) # Authentication functions removed - no longer needed def initialize_app(): """Initialize the application""" # CRITICAL: Inject LOAD_ANALYSIS listener globally so it's always active # This ensures the listener works even if user is on a different page try: from modules.history import inject_load_analysis_listener inject_load_analysis_listener() except Exception as e: print(f"Warning: Could not inject LOAD_ANALYSIS listener: {e}") # Initialize CropDrive integration first (if available) # This handles language and user config from parent website if CROPDRIVE_INTEGRATION_AVAILABLE: try: current_lang, user_plan, features = initialize_integration() # Language is already set by initialize_integration() from URL params # It always updates st.session_state.language, so no need to check here except Exception as e: print(f"Warning: CropDrive integration failed: {e}") # Fallback: Initialize language from URL params manually try: query_params = st.query_params current_lang = query_params.get('lang', 'en') if current_lang not in ['en', 'ms', 'id']: current_lang = 'en' st.session_state.language = current_lang except Exception: st.session_state.language = 'en' else: # Fallback: Initialize language from URL params manually try: query_params = st.query_params current_lang = query_params.get('lang', 'en') if current_lang not in ['en', 'ms', 'id']: current_lang = 'en' st.session_state.language = current_lang except Exception: st.session_state.language = 'en' # Initialize Firebase (non-fatal: keep Space running if secrets are missing) if not initialize_firebase(): st.session_state.firebase_ready = False if 'firebase_warning_shown' not in st.session_state: st.warning( "⚠️ Firebase is not configured. Database features are unavailable until " "FIREBASE_SERVICE_ACCOUNT_KEY (or [firebase] secrets) is added in Space settings." ) st.session_state.firebase_warning_shown = True else: st.session_state.firebase_ready = True initialize_admin_codes() # Initialize session state if 'current_page' not in st.session_state: st.session_state.current_page = 'home' def show_header(): """Display application header""" # Display main header container (deploy button is handled by Streamlit's built-in button) st.markdown(f"""

🌴 {t('app_title')}

{t('app_subtitle')}

""", unsafe_allow_html=True) def show_sidebar(): """Display sidebar navigation""" with st.sidebar: # Logo and title st.markdown(f""" """, unsafe_allow_html=True) st.divider() # Navigation buttons if st.button(t('nav_home'), width="stretch"): st.session_state.current_page = 'home' st.rerun() # Check upload limit before allowing access to analyze page can_analyze = True try: from utils.cropdrive_integration import can_start_analysis can_analyze = can_start_analysis() except ImportError: # If cropdrive_integration not available, allow (fallback) pass except Exception: # On error, allow (fallback) pass if can_analyze: if st.button(t('nav_analyze'), width="stretch"): st.session_state.current_page = 'upload' st.rerun() else: # Disable button and show tooltip st.button(t('nav_analyze'), width="stretch", disabled=True, help="Upload limit reached. Please view your analysis history instead.") # Admin panel button if st.button(t('nav_admin'), width="stretch"): st.session_state.current_page = 'admin' st.rerun() # Help Us Improve button if st.button(t('nav_help_improve'), width="stretch"): st.session_state.current_page = 'help_improve' st.rerun() def show_home_page(): """Display home/landing page""" # Simple explanation of what the tool does col1, col2 = st.columns(2) with col1: st.markdown(f"""

{t('home_what_title')}

""", unsafe_allow_html=True) with col2: st.markdown(f"""

{t('home_how_title')}

  1. {t('home_how_1')}
  2. {t('home_how_2')}
  3. {t('home_how_3')}
  4. {t('home_how_4')}
""", unsafe_allow_html=True) # Simple call to action st.markdown(f"""

{t('home_ready')}

{t('home_ready_desc')}

""", unsafe_allow_html=True) # Button to start - upload files (check limit first) can_analyze = True try: from utils.cropdrive_integration import can_start_analysis, get_upload_limit_info can_analyze = can_start_analysis() if not can_analyze: limit_info = get_upload_limit_info() uploads_used = limit_info.get('uploads_used', 0) uploads_limit = limit_info.get('uploads_limit', 0) st.warning(f"⚠️ **Upload Limit Reached:** You have used {uploads_used}/{uploads_limit} analyses. Please view your analysis history or upgrade your plan.") except ImportError: # If cropdrive_integration not available, allow (fallback) pass except Exception: # On error, allow (fallback) pass if can_analyze: if st.button(t('home_upload_files'), width="stretch", type="primary"): st.session_state.current_page = 'upload' st.rerun() else: # Show disabled button with message st.button(t('home_upload_files'), width="stretch", type="primary", disabled=True) st.info("💡 **Tip:** View your past analyses from the CropDrive website by clicking 'Open Recent Analysis' on any report.") # Authentication pages removed - no longer needed # Settings page removed - authentication no longer required def show_upload_page(): """Display upload page""" try: from modules.upload import show_upload_page as upload_page upload_page() except ImportError: st.error(t('status_error') + ": " + t('upload_error', default='Upload module not available')) st.info(t('status_info') + ": " + t('upload_error', default='Please contact support if this issue persists.')) def show_results_page(): """Display results page""" if results_page_func is not None: results_page_func() else: st.error(t('status_error') + ": " + t('results_no_results', default='Results module not available')) st.info(t('status_info') + ": " + t('results_no_results', default='Please contact support if this issue persists.')) def show_help_improve_page(): """Display Help Us Improve page""" try: from modules.dashboard import display_help_us_improve_tab display_help_us_improve_tab() except ImportError: st.error(t('status_error') + ": Help Us Improve module not available") st.info(t('status_info') + ": Please contact support if this issue persists.") def show_admin_panel(): """Display admin panel""" if admin_panel_func is not None: admin_panel_func() else: st.error(t('status_error') + ": " + t('admin_restricted', default='Admin module not available')) st.info(t('status_info') + ": " + t('admin_restricted', default='Please contact support if this issue persists.')) def main(): """Main application function""" # Initialize the application initialize_app() # CRITICAL: Check for analysisId in query params FIRST - if present, show history page # This must happen before showing header/sidebar to ensure proper routing query_params = st.query_params analysis_id = query_params.get('analysisId', None) # Also check for analysisData in URL (from postMessage) analysis_data_encoded = query_params.get('analysisData', None) # Debug logging if analysis_id or analysis_data_encoded: print(f"🔍 Routing to history page - analysisId: {analysis_id}, has analysisData: {bool(analysis_data_encoded)}") # If we have analysisId or analysisData, route to history page if (analysis_id or analysis_data_encoded) and show_history_page: print("✅ Routing to history page") # Show header show_header() # Show sidebar show_sidebar() # Show history page (it will handle loading the data) show_history_page() return elif analysis_id or analysis_data_encoded: print(f"⚠️ analysisId/analysisData found but show_history_page is not available") # Show header show_header() # Show sidebar show_sidebar() # Route to appropriate page based on current_page current_page = st.session_state.current_page if current_page == 'home': show_home_page() elif current_page == 'upload': show_upload_page() elif current_page == 'results': if results_page_func is not None: results_page_func() else: st.error(t('status_error') + ": " + t('results_no_results')) st.info(t('status_info') + ": " + t('results_no_results', default='Please contact support if this issue persists.')) elif current_page == 'admin': if admin_panel_func is not None: admin_panel_func() else: st.error(t('status_error') + ": " + t('admin_restricted')) st.info(t('status_info') + ": " + t('admin_restricted', default='Please contact support if this issue persists.')) elif current_page == 'help_improve': show_help_improve_page() else: # Default to home if page not found st.session_state.current_page = 'home' show_home_page() # Footer st.markdown(f"""

{t('footer_copyright')}

""", unsafe_allow_html=True) if __name__ == "__main__": main()