ags / app.py
Markishime's picture
Claude Opus 5.5
Update dependencies and fix runtime bugs found in audit
3f62288
Raw History Blame Contribute Delete
24.9 kB
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("""
<script>
(function() {
// Fix form field accessibility issues
function fixFormAccessibility() {
// Find all input, select, and textarea elements without id or name
const formFields = document.querySelectorAll('input, select, textarea');
formFields.forEach((field, index) => {
// Add id if missing
if (!field.id) {
// Try to get id from Streamlit's key attribute or generate one
const streamlitKey = field.getAttribute('data-testid') ||
field.getAttribute('aria-label') ||
field.getAttribute('placeholder') ||
field.type || 'field';
const baseId = streamlitKey.replace(/[^a-zA-Z0-9]/g, '_').toLowerCase();
field.id = baseId + '_' + index + '_' + Date.now();
}
// Add name if missing (use id as name)
if (!field.name && field.id) {
field.name = field.id;
}
// Ensure label association
if (field.id) {
// Check if label exists for this field
const label = document.querySelector(`label[for="${field.id}"]`);
if (!label) {
// Try to find parent label
const parentLabel = field.closest('label');
if (parentLabel && !parentLabel.getAttribute('for')) {
parentLabel.setAttribute('for', field.id);
} else if (!parentLabel) {
// Create label if field has aria-label or placeholder
const labelText = field.getAttribute('aria-label') ||
field.getAttribute('placeholder') ||
field.getAttribute('title') ||
'Form field';
const newLabel = document.createElement('label');
newLabel.setAttribute('for', field.id);
newLabel.textContent = labelText;
newLabel.style.visuallyHidden = 'true';
newLabel.style.position = 'absolute';
newLabel.style.left = '-9999px';
field.parentNode.insertBefore(newLabel, field);
}
}
}
});
}
// Run on page load
if (document.readyState === 'loading') {
document.addEventListener('DOMContentLoaded', fixFormAccessibility);
} else {
fixFormAccessibility();
}
// Also run after Streamlit updates (using MutationObserver)
const observer = new MutationObserver(function(mutations) {
fixFormAccessibility();
});
observer.observe(document.body, {
childList: true,
subtree: true
});
})();
</script>
""", unsafe_allow_html=True)
# Hide Streamlit branding and user profile elements
st.markdown(
"""
<style>
/* Hide selected Streamlit branding elements, but keep header/menu visible */
/* Hide Streamlit Cloud user avatar/profile button */
button[title*="Manage app"] {display: none !important;}
button[title*="Profile"] {display: none !important;}
/* Hide Streamlit logo */
div[data-testid="stHeader"] img {display: none !important;}
a[href*="streamlit.io"] img {display: none !important;}
/* Hide user profile menu and avatar - but keep deploy button */
[data-testid="stHeader"] [data-testid="stDecoration"] {display: none !important;}
/* Hide Streamlit Cloud profile link */
a[href*="streamlit.io/author"] {display: none !important;}
/* Community Cloud "Hosted with Streamlit" badge */
a[class*="viewerBadge_link__"] {display: none !important;}
div[class*="viewerBadge_container__"] {display: none !important;}
/* Fallback: any Streamlit-hosting anchor in bottom container */
[data-testid="stBottomBlockContainer"] a[href*="streamlit.io"] {display: none !important;}
/* Hide bottom right elements - creator avatar and Streamlit logo */
footer {display: none !important; visibility: hidden !important;}
footer[data-testid="stFooter"] {display: none !important; visibility: hidden !important;}
div[data-testid="stFooter"] {display: none !important; visibility: hidden !important;}
/* Hide Streamlit logo and badges in bottom right */
.stApp footer {display: none !important;}
.stApp > footer {display: none !important;}
a[href*="streamlit.io"][target="_blank"]:not([href*="share.streamlit.io/deploy"]) {display: none !important;}
/* Hide creator avatar/profile in bottom right */
footer img {display: none !important;}
footer a img {display: none !important;}
[data-testid="stFooter"] img {display: none !important;}
[data-testid="stFooter"] a {display: none !important;}
/* Hide "Made with Streamlit" and similar badges */
footer a[href*="streamlit.io"] {display: none !important;}
footer a[href*="share.streamlit.io"] {display: none !important;}
/* Show deploy button - ensure it's visible */
button[data-testid="baseButton-secondary"][kind="secondary"] {display: block !important;}
.stDeployButton {display: block !important; visibility: visible !important;}
button[kind="secondary"]:has-text("Deploy") {display: block !important; visibility: visible !important;}
/* Keep header visible but hide specific elements */
header[data-testid="stHeader"] {display: flex !important; visibility: visible !important;}
div[data-testid="stHeader"] {display: flex !important; visibility: visible !important;}
</style>
""",
unsafe_allow_html=True,
)
# Custom CSS for better styling
st.markdown("""
<style>
.main-header {
background: linear-gradient(90deg, #2E8B57 0%, #228B22 100%);
padding: 1.5rem;
border-radius: 10px;
margin-bottom: 2rem;
margin-top: 1rem;
text-align: center;
color: white;
box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
position: relative;
z-index: 1;
}
.main-header h1 {
margin: 0;
font-size: 2.5rem;
font-weight: 700;
}
.main-header p {
margin: 0.5rem 0 0 0;
font-size: 1.1rem;
opacity: 0.95;
}
.auth-container {
max-width: 400px;
margin: 0 auto;
padding: 2rem;
border-radius: 10px;
box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
background-color: #f8f9fa;
}
.sidebar-logo {
text-align: center;
padding: 1rem;
margin-bottom: 2rem;
}
.nav-button {
width: 100%;
margin-bottom: 0.5rem;
border-radius: 5px;
border: none;
padding: 0.5rem 1rem;
background-color: #2E8B57;
color: white;
cursor: pointer;
}
.nav-button:hover {
background-color: #228B22;
}
.footer {
text-align: center;
padding: 2rem;
margin-top: 3rem;
border-top: 1px solid #e0e0e0;
color: #666;
}
</style>
""", 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"""
<div class="main-header">
<h1>🌴 {t('app_title')}</h1>
<p>{t('app_subtitle')}</p>
</div>
""", unsafe_allow_html=True)
def show_sidebar():
"""Display sidebar navigation"""
with st.sidebar:
# Logo and title
st.markdown(f"""
<div class="sidebar-logo">
<h2>🌴 CropDrive™ AI</h2>
<p>{t('app_subtitle')}</p>
</div>
""", 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"""
<div style="background-color: #f8f9fa; padding: 2rem; border-radius: 10px; margin: 1rem 0; border-left: 4px solid #2E8B57;">
<h3 style="color: #2E8B57; margin-top: 0;">{t('home_what_title')}</h3>
<ul style="color: #333; line-height: 1.8;">
<li>{t('home_what_1')}</li>
<li>{t('home_what_2')}</li>
<li>{t('home_what_3')}</li>
<li>{t('home_what_4')}</li>
</ul>
</div>
""", unsafe_allow_html=True)
with col2:
st.markdown(f"""
<div style="background-color: #f8f9fa; padding: 2rem; border-radius: 10px; margin: 1rem 0; border-left: 4px solid #2E8B57;">
<h3 style="color: #2E8B57; margin-top: 0;">{t('home_how_title')}</h3>
<ol style="color: #333; line-height: 1.8;">
<li>{t('home_how_1')}</li>
<li>{t('home_how_2')}</li>
<li>{t('home_how_3')}</li>
<li>{t('home_how_4')}</li>
</ol>
</div>
""", unsafe_allow_html=True)
# Simple call to action
st.markdown(f"""
<div style="text-align: center; padding: 2rem; background: #f0f8ff; border-radius: 10px; margin: 2rem 0; border: 2px solid #2E8B57;">
<h3 style="color: #2E8B57; margin-bottom: 1rem;">{t('home_ready')}</h3>
<p style="color: #333; margin-bottom: 2rem;">{t('home_ready_desc')}</p>
</div>
""", 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"""
<div class="footer">
<p>{t('footer_copyright')}</p>
</div>
""", unsafe_allow_html=True)
if __name__ == "__main__":
main()