|
""" |
|
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 |
|
|
|
|
|
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) |
|
|
|
|
|
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*") |
|
|
|
|
|
gr.HTML('<div class="info-box">📄 <strong>Real PDF Processing:</strong> Upload your actual bank statement PDFs for comprehensive financial analysis.</div>') |
|
|
|
with gr.Tabs(): |
|
|
|
with gr.TabItem("📄 PDF Upload & Analysis"): |
|
self._create_pdf_processing_tab() |
|
|
|
|
|
with gr.TabItem("📊 Analysis Dashboard"): |
|
self._create_dashboard_tab() |
|
|
|
|
|
with gr.TabItem("🤖 AI Financial Advisor"): |
|
self._create_chat_tab() |
|
|
|
|
|
with gr.TabItem("📋 Transaction Management"): |
|
self._create_transaction_tab() |
|
|
|
|
|
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): |
|
|
|
gr.Markdown("### 📁 File Upload") |
|
pdf_upload = gr.File( |
|
label="Upload Bank Statement PDFs", |
|
file_count="multiple", |
|
file_types=[".pdf"], |
|
height=150 |
|
) |
|
|
|
|
|
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 |
|
) |
|
|
|
|
|
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_pdf_btn = gr.Button("🚀 Process PDFs", variant="primary", size="lg") |
|
|
|
with gr.Column(scale=1): |
|
|
|
processing_status = gr.HTML() |
|
|
|
|
|
gr.Markdown("### 📊 Processing Results") |
|
processing_results = gr.JSON( |
|
label="Detailed Results", |
|
visible=False |
|
) |
|
|
|
|
|
quick_stats = gr.HTML() |
|
|
|
|
|
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") |
|
|
|
|
|
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) |
|
|
|
|
|
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") |
|
|
|
|
|
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") |
|
|
|
|
|
with gr.Accordion("📋 Detailed Transaction Data", open=False): |
|
transaction_table = gr.Dataframe( |
|
headers=["Date", "Description", "Amount", "Category", "Account"], |
|
interactive=True, |
|
label="All Transactions" |
|
) |
|
|
|
|
|
clear_status = gr.HTML() |
|
clear_info = gr.HTML() |
|
|
|
|
|
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): |
|
|
|
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) |
|
|
|
|
|
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() |
|
|
|
|
|
gr.Markdown("### 📊 Analysis Context") |
|
context_info = gr.JSON( |
|
label="Available Data", |
|
value={"status": "Upload PDFs to start analysis"} |
|
) |
|
|
|
|
|
gr.Markdown("### ⚙️ Chat Settings") |
|
response_style = gr.Radio( |
|
choices=["Detailed", "Concise", "Technical"], |
|
label="Response Style", |
|
value="Detailed" |
|
) |
|
|
|
|
|
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] |
|
) |
|
|
|
|
|
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): |
|
|
|
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") |
|
|
|
|
|
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): |
|
|
|
gr.Markdown("### 📊 Transaction Statistics") |
|
transaction_stats = gr.HTML() |
|
|
|
|
|
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 = gr.Dataframe( |
|
headers=["ID", "Date", "Description", "Amount", "Category", "Account"], |
|
interactive=False, |
|
label="Filtered Transactions" |
|
) |
|
|
|
|
|
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() |
|
|
|
|
|
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] |
|
) |
|
|
|
|
|
def _process_real_pdfs(self, files, passwords_json, auto_categorize, detect_duplicates): |
|
"""Process real PDF files""" |
|
try: |
|
if not files: |
|
return ('<div class="status-box error-box"> No files uploaded</div>', |
|
gr.update(visible=False), "") |
|
|
|
|
|
status_html = '<div class="status-box warning-box"> Processing PDF files...</div>' |
|
|
|
|
|
passwords = {} |
|
if isinstance(passwords_json, dict): |
|
passwords = passwords_json |
|
elif passwords_json.strip(): |
|
try: |
|
passwords = json.loads(passwords_json) |
|
except json.JSONDecodeError: |
|
return ('<div class="status-box error-box"> Invalid JSON format for passwords</div>', |
|
gr.update(visible=False), "") |
|
|
|
all_transactions = [] |
|
processed_files = [] |
|
|
|
|
|
for file in files: |
|
try: |
|
|
|
with open(file.name, 'rb') as f: |
|
pdf_content = f.read() |
|
|
|
|
|
file_password = passwords.get(os.path.basename(file.name)) |
|
|
|
|
|
statement_info = asyncio.run( |
|
self.pdf_processor.process_pdf(pdf_content, file_password) |
|
) |
|
|
|
|
|
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 ('<div class="status-box warning-box"> No transactions found in uploaded files</div>', |
|
gr.update(value={"processed_files": processed_files}, visible=True), "") |
|
|
|
|
|
self.spend_analyzer.load_transactions(all_transactions) |
|
|
|
|
|
self.current_analysis = self.spend_analyzer.export_analysis_data() |
|
|
|
|
|
status_html = f'<div class="status-box success-box"> Successfully processed {len(processed_files)} files with {len(all_transactions)} transactions</div>' |
|
|
|
|
|
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''' |
|
<div class="status-box info-box"> |
|
<h4> Quick Statistics</h4> |
|
<ul> |
|
<li><strong>Total Income:</strong> ${total_income:,.2f}</li> |
|
<li><strong>Total Expenses:</strong> ${total_expenses:,.2f}</li> |
|
<li><strong>Net Cash Flow:</strong> ${total_income - total_expenses:,.2f}</li> |
|
<li><strong>Transaction Count:</strong> {len(all_transactions)}</li> |
|
</ul> |
|
</div> |
|
''' |
|
|
|
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'<div class="status-box error-box"> Processing error: {str(e)}</div>' |
|
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, |
|
'<div class="status-box warning-box"> No analysis data available</div>', |
|
'<div class="status-box warning-box"> Process PDFs first</div>', |
|
'<div class="status-box warning-box"> No recommendations available</div>', |
|
'<div class="status-box warning-box"> No unusual transactions detected</div>', |
|
pd.DataFrame()) |
|
return empty_return |
|
|
|
try: |
|
summary = self.current_analysis.get('financial_summary', {}) |
|
insights = self.current_analysis.get('spending_insights', []) |
|
|
|
|
|
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) |
|
|
|
|
|
charts = self._create_charts(insights, summary) |
|
|
|
|
|
insights_html = self._create_insights_html() |
|
|
|
|
|
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'<div class="status-box error-box"> Dashboard error: {str(e)}</div>' |
|
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 = {} |
|
|
|
|
|
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 |
|
|
|
|
|
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: |
|
|
|
insights['budget_alerts'] = '<div class="status-box warning-box"> No analysis data available</div>' |
|
insights['spending_insights'] = '<div class="status-box warning-box"> No analysis data available</div>' |
|
insights['recommendations'] = '<div class="status-box warning-box"> No analysis data available</div>' |
|
insights['unusual_transactions'] = '<div class="status-box warning-box"> No analysis data available</div>' |
|
return insights |
|
|
|
|
|
budget_alerts = self.current_analysis.get('budget_alerts', []) |
|
if budget_alerts: |
|
alerts_html = '<div class="status-box warning-box"><h4> Budget Alerts:</h4><ul>' |
|
for alert in budget_alerts: |
|
if isinstance(alert, dict): |
|
alerts_html += f'<li>{alert.get("category", "Unknown")}: {alert.get("percentage_used", 0):.1f}% used</li>' |
|
alerts_html += '</ul></div>' |
|
else: |
|
alerts_html = '<div class="status-box success-box"> All budgets on track</div>' |
|
|
|
insights['budget_alerts'] = alerts_html |
|
|
|
|
|
spending_insights = self.current_analysis.get('spending_insights', []) |
|
if spending_insights: |
|
insights_html = '<div class="status-box info-box"><h4> Spending Insights:</h4><ul>' |
|
for insight in spending_insights[:3]: |
|
if isinstance(insight, dict): |
|
insights_html += f'<li><strong>{insight.get("category", "Unknown")}:</strong> ${insight.get("total_amount", 0):.2f} ({insight.get("percentage_of_total", 0):.1f}%)</li>' |
|
insights_html += '</ul></div>' |
|
else: |
|
insights_html = '<div class="status-box">No spending insights available</div>' |
|
|
|
insights['spending_insights'] = insights_html |
|
|
|
|
|
recommendations = self.current_analysis.get('recommendations', []) |
|
if recommendations: |
|
rec_html = '<div class="status-box info-box"><h4> Recommendations:</h4><ul>' |
|
for rec in recommendations[:3]: |
|
if rec: |
|
rec_html += f'<li>{rec}</li>' |
|
rec_html += '</ul></div>' |
|
else: |
|
rec_html = '<div class="status-box">No specific recommendations available</div>' |
|
|
|
insights['recommendations'] = rec_html |
|
|
|
|
|
financial_summary = self.current_analysis.get('financial_summary', {}) |
|
unusual = financial_summary.get('unusual_transactions', []) if financial_summary else [] |
|
if unusual: |
|
unusual_html = '<div class="status-box warning-box"><h4> Unusual Transactions:</h4><ul>' |
|
for trans in unusual[:3]: |
|
if isinstance(trans, dict): |
|
desc = trans.get("description", "Unknown") |
|
amount = trans.get("amount", 0) |
|
unusual_html += f'<li>{desc}: ${amount:.2f}</li>' |
|
unusual_html += '</ul></div>' |
|
else: |
|
unusual_html = '<div class="status-box success-box"> No unusual transactions detected</div>' |
|
|
|
insights['unusual_transactions'] = unusual_html |
|
|
|
return insights |
|
|
|
def _create_transaction_dataframe(self): |
|
"""Create transaction dataframe for display""" |
|
|
|
|
|
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, "", '<div class="status-box warning-box"> Please enter a message</div>' |
|
|
|
|
|
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 = '<div class="status-box success-box"> Response generated</div>' |
|
else: |
|
response = "Please upload and process your PDF statements first to get personalized financial insights." |
|
status_html = '<div class="status-box warning-box"> No data available</div>' |
|
|
|
|
|
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""" |
|
|
|
return pd.DataFrame(), '<div class="status-box info-box">Filtering functionality would be implemented here</div>' |
|
|
|
def _update_transaction(self, transaction_id, new_category): |
|
"""Update transaction category""" |
|
return '<div class="status-box success-box"> Transaction updated</div>', pd.DataFrame() |
|
|
|
def _add_category(self, new_category): |
|
"""Add new transaction category""" |
|
return '<div class="status-box success-box"> Category added</div>', 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 |
|
|
|
|
|
self.spend_analyzer.set_budgets(budget_settings) |
|
|
|
status_html = '<div class="status-box success-box"> Budget settings saved and applied</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 _export_data(self, export_format, export_options, date_range): |
|
"""Export analysis data""" |
|
if not self.current_analysis: |
|
return '<div class="status-box error-box"> No data to export</div>', {}, None |
|
|
|
try: |
|
|
|
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'] = [] |
|
|
|
|
|
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": |
|
|
|
f.write("Export functionality would create CSV here") |
|
|
|
file_path = f.name |
|
|
|
status_html = '<div class="status-box success-box"> Data exported successfully</div>' |
|
return status_html, export_data, file_path |
|
|
|
except Exception as e: |
|
error_html = f'<div class="status-box error-box"> Export error: {str(e)}</div>' |
|
return error_html, {}, None |
|
|
|
def _save_processing_settings(self, settings): |
|
"""Save processing settings""" |
|
try: |
|
self.user_sessions['processing_settings'] = settings |
|
return '<div class="status-box success-box"> Processing settings saved</div>' |
|
except Exception as e: |
|
return f'<div class="status-box error-box"> Error saving settings: {str(e)}</div>' |
|
|
|
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() |
|
|
|
return ('<div class="status-box success-box"> All data cleared</div>', |
|
'<div class="status-box info-box"> Ready for new PDF upload</div>') |
|
|
|
|
|
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() |
|
|