"""
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:
'
for alert in alerts:
alert_html += f'- {alert["category"]}: {alert["percentage_used"]:.1f}% used
'
alert_html += '
'
else:
alert_html = 'โ
All budgets on track
'
# Recommendations
recommendations = self.current_analysis.get('recommendations', [])
if recommendations:
rec_html = 'Recommendations:
'
for rec in recommendations[:3]: # Show top 3
rec_html += f'- {rec}
'
rec_html += '
'
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()