""" Gradio Web Interface for Spend Analyzer MCP - Real PDF Processing """ import gradio as gr import pandas as pd import plotly.express as px import plotly.graph_objects as go import json import os import asyncio from typing import Dict, List, Optional, Tuple from datetime import datetime, timedelta import logging import time import tempfile # Import our local modules from email_processor import PDFProcessor from spend_analyzer import SpendAnalyzer class RealSpendAnalyzerInterface: def __init__(self): self.current_analysis = None self.user_sessions = {} self.logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) # Initialize processors self.pdf_processor = PDFProcessor() self.spend_analyzer = SpendAnalyzer() def create_interface(self): """Create the main Gradio interface""" with gr.Blocks( title="Spend Analyzer MCP - Real PDF Processing", 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; } .info-box { background-color: #e7f3ff; border: 1px solid #b3d9ff; } """ ) as interface: gr.Markdown("# 💰 Spend Analyzer MCP - Real PDF Processing", elem_classes=["main-header"]) gr.Markdown("*Analyze your real bank statement PDFs with AI-powered insights*") # Info notice gr.HTML('
📄 Real PDF Processing: Upload your actual bank statement PDFs for comprehensive financial analysis.
') with gr.Tabs(): # Tab 1: PDF Upload & Processing with gr.TabItem("📄 PDF Upload & Analysis"): self._create_pdf_processing_tab() # Tab 2: Analysis Dashboard with gr.TabItem("📊 Analysis Dashboard"): self._create_dashboard_tab() # Tab 3: AI Financial Advisor with gr.TabItem("🤖 AI Financial Advisor"): self._create_chat_tab() # Tab 4: Transaction Management with gr.TabItem("📋 Transaction Management"): self._create_transaction_tab() # Tab 5: Settings & Export with gr.TabItem("⚙️ Settings & Export"): self._create_settings_tab() return interface def _create_pdf_processing_tab(self): """Create PDF processing tab""" gr.Markdown("## 📄 Upload & Process Bank Statement PDFs") gr.Markdown("*Upload your bank statement PDFs for real financial analysis*") with gr.Row(): with gr.Column(scale=2): # File upload section gr.Markdown("### 📁 File Upload") pdf_upload = gr.File( label="Upload Bank Statement PDFs", file_count="multiple", file_types=[".pdf"], height=150 ) # Password section gr.Markdown("### 🔐 PDF Passwords (if needed)") pdf_passwords_input = gr.Textbox( label="PDF Passwords (JSON format)", placeholder='{"statement1.pdf": "password123", "statement2.pdf": "password456"}', lines=3 ) # Processing options gr.Markdown("### ⚙️ Processing Options") with gr.Row(): auto_categorize = gr.Checkbox( label="Auto-categorize transactions", value=True ) detect_duplicates = gr.Checkbox( label="Detect duplicate transactions", value=True ) # Process button process_pdf_btn = gr.Button("🚀 Process PDFs", variant="primary", size="lg") with gr.Column(scale=1): # Status and results processing_status = gr.HTML() # Processing progress gr.Markdown("### 📊 Processing Results") processing_results = gr.JSON( label="Detailed Results", visible=False ) # Quick stats quick_stats = gr.HTML() # Event handler process_pdf_btn.click( fn=self._process_real_pdfs, inputs=[pdf_upload, pdf_passwords_input, auto_categorize, detect_duplicates], outputs=[processing_status, processing_results, quick_stats] ) 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") clear_btn = gr.Button("🗑️ Clear Data", variant="stop") # Summary cards gr.Markdown("### 💰 Financial Summary") 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 section gr.Markdown("### 📈 Visual Analysis") with gr.Row(): with gr.Column(): spending_by_category = gr.Plot(label="Spending by Category") monthly_trends = gr.Plot(label="Monthly Spending Trends") with gr.Column(): income_vs_expenses = gr.Plot(label="Income vs Expenses") top_merchants = gr.Plot(label="Top Merchants") # Insights section gr.Markdown("### 🎯 Financial Insights") with gr.Row(): with gr.Column(): budget_alerts = gr.HTML(label="Budget Alerts") spending_insights = gr.HTML(label="Spending Insights") with gr.Column(): recommendations = gr.HTML(label="AI Recommendations") unusual_transactions = gr.HTML(label="Unusual Transactions") # Detailed data with gr.Accordion("📋 Detailed Transaction Data", open=False): transaction_table = gr.Dataframe( headers=["Date", "Description", "Amount", "Category", "Account"], interactive=True, label="All Transactions" ) # Status displays for clear function clear_status = gr.HTML() clear_info = gr.HTML() # Event handlers refresh_btn.click( fn=self._refresh_dashboard, outputs=[total_income, total_expenses, net_cashflow, transaction_count, spending_by_category, monthly_trends, income_vs_expenses, top_merchants, budget_alerts, spending_insights, recommendations, unusual_transactions, transaction_table] ) export_btn.click( fn=self._export_analysis, outputs=[gr.File(label="Analysis Export")] ) clear_btn.click( fn=self._clear_data, outputs=[clear_status, clear_info] ) def _create_chat_tab(self): """Create AI chat tab""" gr.Markdown("## 🤖 AI Financial Advisor") gr.Markdown("*Get personalized insights about your spending patterns*") with gr.Row(): with gr.Column(scale=3): # Chat interface chatbot = gr.Chatbot( label="Financial Advisor Chat", height=500, 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 gr.Markdown("### 🎯 Quick Questions") with gr.Row(): budget_btn = gr.Button("💰 Budget Analysis", size="sm") trends_btn = gr.Button("📈 Spending Trends", size="sm") tips_btn = gr.Button("💡 Save Money Tips", size="sm") unusual_btn = gr.Button("🚨 Unusual Activity", size="sm") with gr.Row(): categories_btn = gr.Button("📊 Category Breakdown", size="sm") merchants_btn = gr.Button("🏪 Top Merchants", size="sm") monthly_btn = gr.Button("📅 Monthly Analysis", size="sm") goals_btn = gr.Button("🎯 Financial Goals", size="sm") with gr.Column(scale=1): chat_status = gr.HTML() # Analysis context gr.Markdown("### 📊 Analysis Context") context_info = gr.JSON( label="Available Data", value={"status": "Upload PDFs to start analysis"} ) # Chat settings gr.Markdown("### ⚙️ Chat Settings") response_style = gr.Radio( choices=["Detailed", "Concise", "Technical"], label="Response Style", value="Detailed" ) # Event handlers send_btn.click( fn=self._handle_chat_message, inputs=[msg_input, chatbot, response_style], outputs=[chatbot, msg_input, chat_status] ) msg_input.submit( fn=self._handle_chat_message, inputs=[msg_input, chatbot, response_style], outputs=[chatbot, msg_input, chat_status] ) # Quick question handlers budget_btn.click(lambda: "How am I doing with my budget this month?", outputs=[msg_input]) trends_btn.click(lambda: "What are my spending trends over the last few months?", outputs=[msg_input]) tips_btn.click(lambda: "What are specific ways I can save money based on my spending?", outputs=[msg_input]) unusual_btn.click(lambda: "Are there any unusual transactions I should be aware of?", outputs=[msg_input]) categories_btn.click(lambda: "Break down my spending by category", outputs=[msg_input]) merchants_btn.click(lambda: "Who are my top merchants and how much do I spend with them?", outputs=[msg_input]) monthly_btn.click(lambda: "Analyze my monthly spending patterns", outputs=[msg_input]) goals_btn.click(lambda: "Help me set realistic financial goals based on my spending", outputs=[msg_input]) def _create_transaction_tab(self): """Create transaction management tab""" gr.Markdown("## 📋 Transaction Management") gr.Markdown("*Review, edit, and categorize your transactions*") with gr.Row(): with gr.Column(scale=2): # Transaction filters gr.Markdown("### 🔍 Filter Transactions") with gr.Row(): date_from = gr.Textbox(label="From Date (YYYY-MM-DD)", placeholder="2024-01-01") date_to = gr.Textbox(label="To Date (YYYY-MM-DD)", placeholder="2024-12-31") with gr.Row(): category_filter = gr.Dropdown( choices=["All", "Food & Dining", "Shopping", "Gas & Transport", "Utilities", "Entertainment", "Healthcare", "Other"], label="Category Filter", value="All" ) amount_filter = gr.Radio( choices=["All", "Income Only", "Expenses Only", "> $100", "> $500"], label="Amount Filter", value="All" ) filter_btn = gr.Button("🔍 Apply Filters", variant="secondary") # Transaction editing gr.Markdown("### ✏️ Edit Transaction") with gr.Row(): edit_transaction_id = gr.Number(label="Transaction ID", precision=0) edit_category = gr.Dropdown( choices=["Food & Dining", "Shopping", "Gas & Transport", "Utilities", "Entertainment", "Healthcare", "Other"], label="New Category" ) update_btn = gr.Button("💾 Update Transaction", variant="primary") with gr.Column(scale=1): # Transaction stats gr.Markdown("### 📊 Transaction Statistics") transaction_stats = gr.HTML() # Category management gr.Markdown("### 🏷️ Category Management") add_category = gr.Textbox(label="Add New Category") add_category_btn = gr.Button("➕ Add Category") category_status = gr.HTML() # Filtered transactions table filtered_transactions = gr.Dataframe( headers=["ID", "Date", "Description", "Amount", "Category", "Account"], interactive=False, label="Filtered Transactions" ) # Event handlers filter_btn.click( fn=self._filter_transactions, inputs=[date_from, date_to, category_filter, amount_filter], outputs=[filtered_transactions, transaction_stats] ) update_btn.click( fn=self._update_transaction, inputs=[edit_transaction_id, edit_category], outputs=[category_status, filtered_transactions] ) add_category_btn.click( fn=self._add_category, inputs=[add_category], outputs=[category_status, edit_category, category_filter] ) def _create_settings_tab(self): """Create settings and export tab""" gr.Markdown("## ⚙️ Settings & Export") with gr.Tabs(): with gr.TabItem("Budget Settings"): gr.Markdown("### 💰 Monthly Budget Configuration") 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("Export Options"): gr.Markdown("### 📤 Data Export") with gr.Row(): with gr.Column(): export_format = gr.Radio( choices=["JSON", "CSV", "Excel"], label="Export Format", value="CSV" ) export_options = gr.CheckboxGroup( choices=["Raw Transactions", "Analysis Summary", "Charts Data", "Recommendations"], label="Include in Export", value=["Raw Transactions", "Analysis Summary"] ) date_range_export = gr.CheckboxGroup( choices=["Last 30 days", "Last 90 days", "Last 6 months", "All data"], label="Date Range", value=["All data"] ) export_data_btn = gr.Button("📤 Export Data", variant="primary") with gr.Column(): export_status = gr.HTML() gr.Markdown("### 📊 Export Preview") export_preview = gr.JSON(label="Export Preview") with gr.TabItem("Processing Settings"): gr.Markdown("### ⚙️ PDF Processing Configuration") processing_settings = gr.JSON( label="Processing Settings", value={ "auto_categorize": True, "detect_duplicates": True, "merge_similar_transactions": False, "confidence_threshold": 0.8, "date_format": "auto", "amount_format": "auto" } ) save_processing_btn = gr.Button("💾 Save Processing Settings", variant="primary") processing_status = gr.HTML() # Event handlers save_budgets_btn.click( fn=self._save_budget_settings, inputs=[budget_categories, budget_amounts], outputs=[budget_status, current_budgets] ) export_data_btn.click( fn=self._export_data, inputs=[export_format, export_options, date_range_export], outputs=[export_status, export_preview, gr.File(label="Export File")] ) save_processing_btn.click( fn=self._save_processing_settings, inputs=[processing_settings], outputs=[processing_status] ) # Implementation methods def _process_real_pdfs(self, files, passwords_json, auto_categorize, detect_duplicates): """Process real PDF files""" try: if not files: return ('
No files uploaded
', gr.update(visible=False), "") # Update status status_html = '
Processing PDF files...
' # Parse passwords if provided passwords = {} if isinstance(passwords_json, dict): passwords = passwords_json elif passwords_json.strip(): try: passwords = json.loads(passwords_json) except json.JSONDecodeError: return ('
Invalid JSON format for passwords
', gr.update(visible=False), "") all_transactions = [] processed_files = [] # Process each PDF for file in files: try: # Read file content with open(file.name, 'rb') as f: pdf_content = f.read() # Get password for this file file_password = passwords.get(os.path.basename(file.name)) # Process PDF statement_info = asyncio.run( self.pdf_processor.process_pdf(pdf_content, file_password) ) # Add transactions all_transactions.extend(statement_info.transactions) processed_files.append({ 'filename': os.path.basename(file.name), 'bank': statement_info.bank_name, 'account': statement_info.account_number, 'period': statement_info.statement_period, 'transaction_count': len(statement_info.transactions), 'opening_balance': statement_info.opening_balance, 'closing_balance': statement_info.closing_balance, 'status': 'success' }) except Exception as e: processed_files.append({ 'filename': os.path.basename(file.name), 'status': 'error', 'error': str(e) }) if not all_transactions: return ('
No transactions found in uploaded files
', gr.update(value={"processed_files": processed_files}, visible=True), "") # Load transactions into analyzer self.spend_analyzer.load_transactions(all_transactions) # Generate analysis self.current_analysis = self.spend_analyzer.export_analysis_data() # Create success status status_html = f'
Successfully processed {len(processed_files)} files with {len(all_transactions)} transactions
' # Create quick stats total_income = sum(t.amount for t in all_transactions if t.amount > 0) total_expenses = abs(sum(t.amount for t in all_transactions if t.amount < 0)) quick_stats_html = f'''

Quick Statistics

''' results = { "processed_files": processed_files, "total_transactions": len(all_transactions), "analysis_summary": { "total_income": total_income, "total_expenses": total_expenses, "net_cash_flow": total_income - total_expenses } } return (status_html, gr.update(value=results, visible=True), quick_stats_html) except Exception as e: error_html = f'
Processing error: {str(e)}
' return error_html, gr.update(visible=False), "" def _refresh_dashboard(self): """Refresh dashboard with current analysis""" if not self.current_analysis: empty_return = (0, 0, 0, 0, None, None, None, None, '
No analysis data available
', '
Process PDFs first
', '
No recommendations available
', '
No unusual transactions detected
', pd.DataFrame()) return empty_return 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 charts charts = self._create_charts(insights, summary) # Create insights HTML insights_html = self._create_insights_html() # Create transaction table transaction_df = self._create_transaction_dataframe() return (total_income, total_expenses, net_cashflow, transaction_count, charts['spending_by_category'], charts['monthly_trends'], charts['income_vs_expenses'], charts['top_merchants'], insights_html['budget_alerts'], insights_html['spending_insights'], insights_html['recommendations'], insights_html['unusual_transactions'], transaction_df) except Exception as e: error_msg = f'
Dashboard error: {str(e)}
' empty_return = (0, 0, 0, 0, None, None, None, None, error_msg, error_msg, error_msg, error_msg, pd.DataFrame()) return empty_return def _create_charts(self, insights, summary): """Create visualization charts""" charts = {} # Spending by category chart if insights: categories = [insight['category'] for insight in insights] amounts = [insight['total_amount'] for insight in insights] charts['spending_by_category'] = px.pie( values=amounts, names=categories, title="Spending by Category" ) else: charts['spending_by_category'] = None # Monthly trends (placeholder) charts['monthly_trends'] = None charts['income_vs_expenses'] = None charts['top_merchants'] = None return charts def _create_insights_html(self): """Create insights HTML sections""" insights = {} if not self.current_analysis: # Return empty insights if no analysis available insights['budget_alerts'] = '
No analysis data available
' insights['spending_insights'] = '
No analysis data available
' insights['recommendations'] = '
No analysis data available
' insights['unusual_transactions'] = '
No analysis data available
' return insights # Budget alerts budget_alerts = self.current_analysis.get('budget_alerts', []) if budget_alerts: alerts_html = '

Budget Alerts:

' else: alerts_html = '
All budgets on track
' insights['budget_alerts'] = alerts_html # Spending insights spending_insights = self.current_analysis.get('spending_insights', []) if spending_insights: insights_html = '

Spending Insights:

' else: insights_html = '
No spending insights available
' insights['spending_insights'] = insights_html # Recommendations recommendations = self.current_analysis.get('recommendations', []) if recommendations: rec_html = '

Recommendations:

' else: rec_html = '
No specific recommendations available
' insights['recommendations'] = rec_html # Unusual transactions financial_summary = self.current_analysis.get('financial_summary', {}) unusual = financial_summary.get('unusual_transactions', []) if financial_summary else [] if unusual: unusual_html = '

Unusual Transactions:

' else: unusual_html = '
No unusual transactions detected
' insights['unusual_transactions'] = unusual_html return insights def _create_transaction_dataframe(self): """Create transaction dataframe for display""" # This would create a dataframe from the actual transactions # For now, return empty dataframe return pd.DataFrame(columns=["Date", "Description", "Amount", "Category", "Account"]) def _handle_chat_message(self, message, chat_history, response_style): """Handle chat messages""" if not message.strip(): return chat_history, "", '
Please enter a message
' # Simple response generation based on analysis if self.current_analysis: summary = self.current_analysis.get('financial_summary', {}) response = f"Based on your financial data: Total income ${summary.get('total_income', 0):.2f}, Total expenses ${summary.get('total_expenses', 0):.2f}. Your question: '{message}' - This is a simplified response. Full AI integration would provide detailed insights here." status_html = '
Response generated
' else: response = "Please upload and process your PDF statements first to get personalized financial insights." status_html = '
No data available
' # Add to chat history chat_history = chat_history or [] chat_history.append([message, response]) return chat_history, "", status_html def _filter_transactions(self, date_from, date_to, category_filter, amount_filter): """Filter transactions based on criteria""" # Placeholder implementation return pd.DataFrame(), '
Filtering functionality would be implemented here
' def _update_transaction(self, transaction_id, new_category): """Update transaction category""" return '
Transaction updated
', pd.DataFrame() def _add_category(self, new_category): """Add new transaction category""" return '
Category added
', gr.update(), gr.update() def _save_budget_settings(self, categories, amounts): """Save budget settings""" try: budget_settings = {cat: amounts.get(cat, 0) for cat in categories} self.user_sessions['budgets'] = budget_settings # Apply budgets to analyzer self.spend_analyzer.set_budgets(budget_settings) status_html = '
Budget settings saved and applied
' return status_html, budget_settings except Exception as e: error_html = f'
Error saving budgets: {str(e)}
' return error_html, {} def _export_data(self, export_format, export_options, date_range): """Export analysis data""" if not self.current_analysis: return '
No data to export
', {}, None try: # Create export data export_data = {} if "Analysis Summary" in export_options: export_data['summary'] = self.current_analysis.get('financial_summary', {}) if "Raw Transactions" in export_options: export_data['transactions'] = [] # Would populate with actual transaction data # Create temporary file with tempfile.NamedTemporaryFile(mode='w', suffix=f'.{export_format.lower()}', delete=False) as f: if export_format == "JSON": json.dump(export_data, f, indent=2, default=str) elif export_format == "CSV": # Would create CSV format f.write("Export functionality would create CSV here") file_path = f.name status_html = '
Data exported successfully
' return status_html, export_data, file_path except Exception as e: error_html = f'
Export error: {str(e)}
' return error_html, {}, None def _save_processing_settings(self, settings): """Save processing settings""" try: self.user_sessions['processing_settings'] = settings return '
Processing settings saved
' except Exception as e: return f'
Error saving settings: {str(e)}
' def _export_analysis(self): """Export current analysis""" if not self.current_analysis: return None try: 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 _clear_data(self): """Clear all data""" self.current_analysis = None self.spend_analyzer = SpendAnalyzer() # Reset analyzer return ('
All data cleared
', '
Ready for new PDF upload
') # Launch the interface def launch_interface(): """Launch the Gradio interface""" interface = RealSpendAnalyzerInterface() app = interface.create_interface() print(" Starting Spend Analyzer MCP - Real PDF Processing") print(" Upload your bank statement PDFs for analysis") print(" Opening in browser...") app.launch( server_name="0.0.0.0", server_port=7862, share=False, debug=True, show_error=True, inbrowser=True ) if __name__ == "__main__": launch_interface()