Spend-Analyzer-MCP / gradio_interface.py
Balamurugan Thayalan
Initial Commit
499796e
raw
history blame
33.3 kB
"""
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="[email protected]"
)
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 = '<div class="status-box warning-box">🔄 Processing email statements...</div>'
# 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'<div class="status-box success-box">✅ Processed {result["total_transactions"]} transactions</div>'
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'<div class="status-box error-box">❌ Error: {str(e)}</div>'
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 = '<div class="status-box warning-box">🔄 Retrying with passwords...</div>'
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'<div class="status-box success-box">✅ Processed {result["total_transactions"]} transactions</div>'
self.current_analysis = result.get('analysis', {})
return status_html, result
except Exception as e:
error_html = f'<div class="status-box error-box">❌ Error: {str(e)}</div>'
return error_html, {}
def _analyze_pdf_files(self, files, passwords):
"""Analyze uploaded PDF files"""
try:
if not files:
return '<div class="status-box error-box">❌ No files uploaded</div>', {}
status_html = '<div class="status-box warning-box">🔄 Analyzing PDF files...</div>'
# 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'<div class="status-box success-box">✅ Analyzed {result["total_transactions"]} transactions</div>'
self.current_analysis = result.get('analysis', {})
else:
status_html = '<div class="status-box warning-box">⚠️ No transactions found in uploaded files</div>'
return status_html, result
except Exception as e:
error_html = f'<div class="status-box error-box">❌ Error: {str(e)}</div>'
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 = '<div class="status-box warning-box">⚠️ Using local processing (Modal unavailable)</div>'
# 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,
'<div class="status-box warning-box">⚠️ No analysis data available</div>',
'<div class="status-box warning-box">⚠️ Process statements first</div>',
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 = '<div class="status-box warning-box"><h4>Budget Alerts:</h4><ul>'
for alert in alerts:
alert_html += f'<li>{alert["category"]}: {alert["percentage_used"]:.1f}% used</li>'
alert_html += '</ul></div>'
else:
alert_html = '<div class="status-box success-box">✅ All budgets on track</div>'
# Recommendations
recommendations = self.current_analysis.get('recommendations', [])
if recommendations:
rec_html = '<div class="status-box"><h4>Recommendations:</h4><ul>'
for rec in recommendations[:3]: # Show top 3
rec_html += f'<li>{rec}</li>'
rec_html += '</ul></div>'
else:
rec_html = '<div class="status-box">No specific recommendations at this time.</div>'
# 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'<div class="status-box error-box">❌ Dashboard error: {str(e)}</div>'
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, "", '<div class="status-box warning-box">⚠️ Please enter a message</div>'
try:
# Add user message to chat
chat_history = chat_history or []
chat_history.append([message, None])
status_html = '<div class="status-box warning-box">🤖 AI is thinking...</div>'
# 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 = '<div class="status-box success-box">✅ Response generated</div>'
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 = '<div class="status-box warning-box">⚠️ No data available</div>'
# 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, "", '<div class="status-box error-box">❌ Chat Error</div>'
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 = '<div class="status-box success-box">✅ Budget settings saved</div>'
return status_html, budget_settings
except Exception as e:
error_html = f'<div class="status-box error-box">❌ Error saving budgets: {str(e)}</div>'
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 '<div class="status-box success-box">✅ Email settings saved</div>'
except Exception as e:
return f'<div class="status-box error-box">❌ Error saving settings: {str(e)}</div>'
def _test_email_connection(self, provider, server, port):
"""Test email connection"""
try:
# This would test the actual connection in a real implementation
return '<div class="status-box success-box">✅ Email connection test successful</div>'
except Exception as e:
return f'<div class="status-box error-box">❌ Connection test failed: {str(e)}</div>'
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()