""" Gradio Web Interface for Spend Analyzer MCP """ import gradio as gr import pandas as pd import plotly.express as px import plotly.graph_objects as go from plotly.subplots import make_subplots import json import os from typing import Dict, List, Optional, Tuple import asyncio from datetime import datetime, timedelta import modal import logging # Import our Modal functions from modal_deployment import ( process_bank_statements, analyze_uploaded_statements, get_claude_analysis, save_user_data, load_user_data ) class SpendAnalyzerInterface: def __init__(self): self.current_analysis = None self.user_sessions = {} self.logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) def create_interface(self): """Create the main Gradio interface""" with gr.Blocks( title="Spend Analyzer MCP", theme=gr.themes.Soft(), css=""" .main-header { text-align: center; margin: 20px 0; } .status-box { padding: 10px; border-radius: 5px; margin: 10px 0; } .success-box { background-color: #d4edda; border: 1px solid #c3e6cb; } .error-box { background-color: #f8d7da; border: 1px solid #f5c6cb; } .warning-box { background-color: #fff3cd; border: 1px solid #ffeaa7; } """ ) as interface: gr.Markdown("# ๐Ÿ’ฐ Spend Analyzer MCP", elem_classes=["main-header"]) gr.Markdown("*Analyze your bank statements with AI-powered insights*") with gr.Tabs(): # Tab 1: Email Processing with gr.TabItem("๐Ÿ“ง Email Processing"): self._create_email_tab() # Tab 2: PDF Upload with gr.TabItem("๐Ÿ“„ PDF Upload"): self._create_pdf_tab() # Tab 3: Analysis Dashboard with gr.TabItem("๐Ÿ“Š Analysis Dashboard"): self._create_dashboard_tab() # Tab 4: AI Chat with gr.TabItem("๐Ÿค– AI Financial Advisor"): self._create_chat_tab() # Tab 5: Settings with gr.TabItem("โš™๏ธ Settings"): self._create_settings_tab() return interface def _create_email_tab(self): """Create email processing tab""" gr.Markdown("## Connect Your Email to Analyze Bank Statements") gr.Markdown("*Securely connect to your email to automatically process bank statements*") with gr.Row(): with gr.Column(scale=1): email_provider = gr.Dropdown( choices=["Gmail", "Outlook", "Yahoo", "Other"], label="Email Provider", value="Gmail" ) email_address = gr.Textbox( label="Email Address", placeholder="your.email@gmail.com" ) email_password = gr.Textbox( label="Password/App Password", type="password", placeholder="App-specific password recommended" ) days_back = gr.Slider( minimum=7, maximum=90, value=30, step=1, label="Days to Look Back" ) process_email_btn = gr.Button("๐Ÿ” Process Email Statements", variant="primary") with gr.Column(scale=1): email_status = gr.HTML() password_inputs = gr.Column(visible=False) with password_inputs: gr.Markdown("### Password-Protected PDFs Found") pdf_passwords = gr.JSON( label="Enter passwords for protected files", value={} ) retry_with_passwords = gr.Button("๐Ÿ” Retry with Passwords") email_results = gr.JSON(label="Processing Results", visible=False) # Event handlers process_email_btn.click( fn=self._process_email_statements, inputs=[email_provider, email_address, email_password, days_back], outputs=[email_status, email_results, password_inputs] ) retry_with_passwords.click( fn=self._retry_with_passwords, inputs=[email_provider, email_address, email_password, days_back, pdf_passwords], outputs=[email_status, email_results] ) def _create_pdf_tab(self): """Create PDF upload tab""" gr.Markdown("## Upload Bank Statement PDFs") gr.Markdown("*Upload your bank statement PDFs directly for analysis*") with gr.Row(): with gr.Column(): pdf_upload = gr.File( label="Upload Bank Statement PDFs", file_count="multiple", file_types=[".pdf"] ) pdf_passwords_input = gr.JSON( label="PDF Passwords (if needed)", placeholder='{"statement1.pdf": "password123"}', value={} ) analyze_pdf_btn = gr.Button("๐Ÿ“Š Analyze PDFs", variant="primary") with gr.Column(): pdf_status = gr.HTML() pdf_results = gr.JSON(label="Analysis Results", visible=False) # Event handler analyze_pdf_btn.click( fn=self._analyze_pdf_files, inputs=[pdf_upload, pdf_passwords_input], outputs=[pdf_status, pdf_results] ) def _create_dashboard_tab(self): """Create analysis dashboard tab""" gr.Markdown("## ๐Ÿ“Š Financial Analysis Dashboard") with gr.Row(): refresh_btn = gr.Button("๐Ÿ”„ Refresh Dashboard") export_btn = gr.Button("๐Ÿ“ค Export Analysis") # Summary cards with gr.Row(): total_income = gr.Number(label="Total Income", interactive=False) total_expenses = gr.Number(label="Total Expenses", interactive=False) net_cashflow = gr.Number(label="Net Cash Flow", interactive=False) transaction_count = gr.Number(label="Total Transactions", interactive=False) # Charts with gr.Row(): with gr.Column(): spending_by_category = gr.Plot(label="Spending by Category") monthly_trends = gr.Plot(label="Monthly Trends") with gr.Column(): budget_alerts = gr.HTML(label="Budget Alerts") recommendations = gr.HTML(label="Recommendations") # Detailed data with gr.Accordion("Detailed Transaction Data", open=False): transaction_table = gr.Dataframe( headers=["Date", "Description", "Amount", "Category"], interactive=False, label="Recent Transactions" ) # Event handlers refresh_btn.click( fn=self._refresh_dashboard, outputs=[total_income, total_expenses, net_cashflow, transaction_count, spending_by_category, monthly_trends, budget_alerts, recommendations, transaction_table] ) export_btn.click( fn=self._export_analysis, outputs=[gr.File(label="Analysis Export")] ) def _create_chat_tab(self): """Create AI chat tab""" gr.Markdown("## ๐Ÿค– AI Financial Advisor") gr.Markdown("*Ask questions about your spending patterns and get personalized advice*") with gr.Row(): with gr.Column(scale=3): chatbot = gr.Chatbot( label="Financial Advisor Chat", height=400, show_label=True ) with gr.Row(): msg_input = gr.Textbox( placeholder="Ask about your spending patterns, budgets, or financial goals...", label="Your Question", scale=4 ) send_btn = gr.Button("Send", variant="primary", scale=1) # Quick question buttons with gr.Row(): gr.Button("๐Ÿ’ฐ Budget Analysis", size="sm").click( lambda: "How am I doing with my budget this month?", outputs=[msg_input] ) gr.Button("๐Ÿ“ˆ Spending Trends", size="sm").click( lambda: "What are my spending trends over the last few months?", outputs=[msg_input] ) gr.Button("๐Ÿ’ก Save Money Tips", size="sm").click( lambda: "What are some specific ways I can save money based on my spending?", outputs=[msg_input] ) gr.Button("๐Ÿšจ Unusual Activity", size="sm").click( lambda: "Are there any unusual transactions I should be aware of?", outputs=[msg_input] ) with gr.Column(scale=1): chat_status = gr.HTML() # Analysis context gr.Markdown("### Current Analysis Context") context_info = gr.JSON( label="Available Data", value={"status": "No analysis loaded"} ) # Event handlers send_btn.click( fn=self._handle_chat_message, inputs=[msg_input, chatbot], outputs=[chatbot, msg_input, chat_status] ) msg_input.submit( fn=self._handle_chat_message, inputs=[msg_input, chatbot], outputs=[chatbot, msg_input, chat_status] ) def _create_settings_tab(self): """Create settings tab""" gr.Markdown("## โš™๏ธ Settings & Configuration") with gr.Tabs(): with gr.TabItem("Budget Settings"): gr.Markdown("### Set Monthly Budget Limits") with gr.Row(): with gr.Column(): budget_categories = gr.CheckboxGroup( choices=["Food & Dining", "Shopping", "Gas & Transport", "Utilities", "Entertainment", "Healthcare", "Other"], label="Categories to Budget", value=["Food & Dining", "Shopping", "Gas & Transport"] ) budget_amounts = gr.JSON( label="Budget Amounts ($)", value={ "Food & Dining": 500, "Shopping": 300, "Gas & Transport": 200, "Utilities": 150, "Entertainment": 100, "Healthcare": 200, "Other": 100 } ) save_budgets_btn = gr.Button("๐Ÿ’พ Save Budget Settings", variant="primary") with gr.Column(): budget_status = gr.HTML() current_budgets = gr.JSON(label="Current Budget Settings") with gr.TabItem("Email Settings"): gr.Markdown("### Email Configuration") with gr.Row(): with gr.Column(): email_provider_setting = gr.Dropdown( choices=["Gmail", "Outlook", "Yahoo", "Custom"], label="Email Provider", value="Gmail" ) imap_server = gr.Textbox( label="IMAP Server", value="imap.gmail.com", placeholder="imap.gmail.com" ) imap_port = gr.Number( label="IMAP Port", value=993, precision=0 ) auto_process = gr.Checkbox( label="Auto-process new statements", value=False ) save_email_btn = gr.Button("๐Ÿ’พ Save Email Settings", variant="primary") with gr.Column(): email_test_btn = gr.Button("๐Ÿงช Test Email Connection") email_test_status = gr.HTML() with gr.TabItem("Export Settings"): gr.Markdown("### Data Export Options") export_format = gr.Radio( choices=["JSON", "CSV", "Excel"], label="Export Format", value="JSON" ) include_raw_data = gr.Checkbox( label="Include raw transaction data", value=True ) include_analysis = gr.Checkbox( label="Include analysis results", value=True ) export_settings_btn = gr.Button("๐Ÿ“ค Export Current Analysis") # Event handlers save_budgets_btn.click( fn=self._save_budget_settings, inputs=[budget_categories, budget_amounts], outputs=[budget_status, current_budgets] ) save_email_btn.click( fn=self._save_email_settings, inputs=[email_provider_setting, imap_server, imap_port, auto_process], outputs=[email_test_status] ) email_test_btn.click( fn=self._test_email_connection, inputs=[email_provider_setting, imap_server, imap_port], outputs=[email_test_status] ) # Implementation methods def _process_email_statements(self, provider, email, password, days_back): """Process bank statements from email""" try: # Update status status_html = '
๐Ÿ”„ Processing email statements...
' # Configure email settings email_config = { 'email': email, 'password': password, 'imap_server': self._get_imap_server(provider) } # For now, simulate the Modal function call # In production, this would call the actual Modal function try: # Simulate processing import time time.sleep(1) # Simulate processing time # Mock result for demonstration result = { 'processed_statements': [ { 'filename': 'statement1.pdf', 'bank': 'Chase', 'account': '****1234', 'transaction_count': 25, 'status': 'success' } ], 'total_transactions': 25, 'analysis': { 'financial_summary': { 'total_income': 3000.0, 'total_expenses': 1500.0, 'net_cash_flow': 1500.0 }, 'spending_insights': [ { 'category': 'Food & Dining', 'total_amount': 400.0, 'transaction_count': 12, 'percentage_of_total': 26.7 } ], 'recommendations': ['Consider reducing dining out expenses'], 'transaction_count': 25 } } status_html = f'
โœ… Processed {result["total_transactions"]} transactions
' password_inputs_visible = gr.update(visible=False) # Store analysis for dashboard self.current_analysis = result.get('analysis', {}) return status_html, result, password_inputs_visible except Exception as modal_error: # Fallback to local processing if Modal is not available self.logger.warning(f"Modal processing failed, using local fallback: {modal_error}") return self._process_email_local(email_config, days_back) except Exception as e: error_html = f'
โŒ Error: {str(e)}
' return error_html, {}, gr.update(visible=False) def _retry_with_passwords(self, provider, email, password, days_back, pdf_passwords): """Retry processing with PDF passwords""" try: status_html = '
๐Ÿ”„ Retrying with passwords...
' email_config = { 'email': email, 'password': password, 'imap_server': self._get_imap_server(provider) } # Mock retry with passwords result = { 'processed_statements': [ { 'filename': 'protected_statement.pdf', 'bank': 'Bank of America', 'account': '****5678', 'transaction_count': 30, 'status': 'success' } ], 'total_transactions': 30, 'analysis': { 'financial_summary': { 'total_income': 3500.0, 'total_expenses': 1800.0, 'net_cash_flow': 1700.0 }, 'spending_insights': [], 'recommendations': [], 'transaction_count': 30 } } status_html = f'
โœ… Processed {result["total_transactions"]} transactions
' self.current_analysis = result.get('analysis', {}) return status_html, result except Exception as e: error_html = f'
โŒ Error: {str(e)}
' return error_html, {} def _analyze_pdf_files(self, files, passwords): """Analyze uploaded PDF files""" try: if not files: return '
โŒ No files uploaded
', {} status_html = '
๐Ÿ”„ Analyzing PDF files...
' # Mock PDF analysis result = { 'processed_files': [], 'total_transactions': 0, 'analysis': { 'financial_summary': { 'total_income': 0, 'total_expenses': 0, 'net_cash_flow': 0 }, 'spending_insights': [], 'recommendations': [], 'transaction_count': 0 } } # Process each file for file in files: try: # Mock processing file_result = { 'filename': file.name, 'bank': 'Unknown Bank', 'transaction_count': 15, 'status': 'success' } result['processed_files'].append(file_result) result['total_transactions'] += 15 except Exception as file_error: result['processed_files'].append({ 'filename': file.name, 'status': 'error', 'error': str(file_error) }) if result['total_transactions'] > 0: status_html = f'
โœ… Analyzed {result["total_transactions"]} transactions
' self.current_analysis = result.get('analysis', {}) else: status_html = '
โš ๏ธ No transactions found in uploaded files
' return status_html, result except Exception as e: error_html = f'
โŒ Error: {str(e)}
' return error_html, {} def _process_email_local(self, email_config, days_back): """Local fallback for email processing""" # This would use the local email_processor module status_html = '
โš ๏ธ Using local processing (Modal unavailable)
' # Mock local processing result result = { 'processed_statements': [], 'total_transactions': 0, 'analysis': { 'financial_summary': { 'total_income': 0, 'total_expenses': 0, 'net_cash_flow': 0 }, 'spending_insights': [], 'recommendations': ['Please configure Modal deployment for full functionality'], 'transaction_count': 0 } } return status_html, result, gr.update(visible=False) def _refresh_dashboard(self): """Refresh dashboard with current analysis""" if not self.current_analysis: return (0, 0, 0, 0, None, None, '
โš ๏ธ No analysis data available
', '
โš ๏ธ Process statements first
', pd.DataFrame()) try: summary = self.current_analysis.get('financial_summary', {}) insights = self.current_analysis.get('spending_insights', []) # Summary metrics total_income = summary.get('total_income', 0) total_expenses = summary.get('total_expenses', 0) net_cashflow = summary.get('net_cash_flow', 0) transaction_count = self.current_analysis.get('transaction_count', 0) # Create spending by category chart if insights: categories = [insight['category'] for insight in insights] amounts = [insight['total_amount'] for insight in insights] spending_chart = px.pie( values=amounts, names=categories, title="Spending by Category" ) else: spending_chart = None # Create monthly trends chart monthly_trends = summary.get('monthly_trends', {}) if monthly_trends: trends_chart = px.line( x=list(monthly_trends.keys()), y=list(monthly_trends.values()), title="Monthly Spending Trends" ) else: trends_chart = None # Budget alerts alerts = self.current_analysis.get('budget_alerts', []) if alerts: alert_html = '

Budget Alerts:

' else: alert_html = '
โœ… All budgets on track
' # Recommendations recommendations = self.current_analysis.get('recommendations', []) if recommendations: rec_html = '

Recommendations:

' else: rec_html = '
No specific recommendations at this time.
' # Transaction table (sample recent transactions) transaction_df = pd.DataFrame() # Would populate with actual transaction data return (total_income, total_expenses, net_cashflow, transaction_count, spending_chart, trends_chart, alert_html, rec_html, transaction_df) except Exception as e: error_msg = f'
โŒ Dashboard error: {str(e)}
' return (0, 0, 0, 0, None, None, error_msg, error_msg, pd.DataFrame()) def _export_analysis(self): """Export current analysis data""" if not self.current_analysis: return None try: import tempfile import json # Create temporary file with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: json.dump(self.current_analysis, f, indent=2, default=str) return f.name except Exception as e: self.logger.error(f"Export error: {e}") return None def _handle_chat_message(self, message, chat_history): """Handle chat messages with AI advisor""" if not message.strip(): return chat_history, "", '
โš ๏ธ Please enter a message
' try: # Add user message to chat chat_history = chat_history or [] chat_history.append([message, None]) status_html = '
๐Ÿค– AI is thinking...
' # Mock AI response for now (would use Claude API in production) if self.current_analysis: # Generate a contextual response based on the analysis summary = self.current_analysis.get('financial_summary', {}) insights = self.current_analysis.get('spending_insights', []) recommendations = self.current_analysis.get('recommendations', []) if 'budget' in message.lower(): ai_response = f"Based on your current spending analysis, you have a net cash flow of ${summary.get('net_cash_flow', 0):.2f}. Your total expenses are ${summary.get('total_expenses', 0):.2f} against an income of ${summary.get('total_income', 0):.2f}." elif 'trend' in message.lower(): if insights: top_category = insights[0] ai_response = f"Your top spending category is {top_category['category']} at ${top_category['total_amount']:.2f} ({top_category['percentage_of_total']:.1f}% of total spending). This represents {top_category['transaction_count']} transactions." else: ai_response = "I need more transaction data to analyze your spending trends effectively." elif 'save' in message.lower() or 'tip' in message.lower(): if recommendations: ai_response = f"Here are some personalized recommendations: {'. '.join(recommendations[:2])}" else: ai_response = "Based on your spending patterns, consider tracking your largest expense categories and setting monthly budgets for better financial control." elif 'unusual' in message.lower() or 'activity' in message.lower(): ai_response = "I've analyzed your transactions for unusual patterns. Currently, your spending appears consistent with normal patterns. I'll alert you if I detect any anomalies." else: ai_response = f"I can see you have {self.current_analysis.get('transaction_count', 0)} transactions analyzed. Feel free to ask about your budget, spending trends, saving tips, or unusual activity. What specific aspect of your finances would you like to explore?" status_html = '
โœ… Response generated
' else: ai_response = "I don't have any financial data to analyze yet. Please process your bank statements first using the Email Processing or PDF Upload tabs." status_html = '
โš ๏ธ No data available
' # Update chat history with AI response chat_history[-1][1] = ai_response return chat_history, "", status_html except Exception as e: error_response = f"I'm sorry, I encountered an error: {str(e)}" if chat_history: chat_history[-1][1] = error_response return chat_history, "", '
โŒ Chat Error
' def _save_budget_settings(self, categories, amounts): """Save budget settings""" try: # Filter amounts for selected categories budget_settings = {cat: amounts.get(cat, 0) for cat in categories} # Store in user session (in real app, would save to database) self.user_sessions['budgets'] = budget_settings status_html = '
โœ… Budget settings saved
' return status_html, budget_settings except Exception as e: error_html = f'
โŒ Error saving budgets: {str(e)}
' return error_html, {} def _save_email_settings(self, provider, server, port, auto_process): """Save email settings""" try: email_settings = { 'provider': provider, 'imap_server': server, 'imap_port': port, 'auto_process': auto_process } self.user_sessions['email_settings'] = email_settings return '
โœ… Email settings saved
' except Exception as e: return f'
โŒ Error saving settings: {str(e)}
' def _test_email_connection(self, provider, server, port): """Test email connection""" try: # This would test the actual connection in a real implementation return '
โœ… Email connection test successful
' except Exception as e: return f'
โŒ Connection test failed: {str(e)}
' def _get_imap_server(self, provider): """Get IMAP server for email provider""" servers = { 'Gmail': 'imap.gmail.com', 'Outlook': 'outlook.office365.com', 'Yahoo': 'imap.mail.yahoo.com', 'Other': 'imap.gmail.com' # Default } return servers.get(provider, 'imap.gmail.com') # Launch the interface def launch_interface(): """Launch the Gradio interface""" interface = SpendAnalyzerInterface() app = interface.create_interface() app.launch( server_name="0.0.0.0", server_port=7860, share=False, debug=True, show_error=True ) if __name__ == "__main__": launch_interface()