"""Expense tools — now with smart category detection & graceful handling."""
import datetime
from typing import Optional, List, Dict, Any
from utils.expenses_core import (
add_expense_to_db,
update_expense_in_db,
delete_expense_from_db,
list_expenses_from_db,
summarize_expenses
)
from utils.date_utils import parse_flexible_date, get_today
from utils.budget_core import budget_exists_for_month, calculate_remaining_budget
from utils.affordability_core import check_affordability_logic
from utils.category_detection import detect_category # NEW ✔
from config import CATEGORIES, CATEGORIES_STRUCTURE
def register_expense_tools(mcp):
# =====================================================================
# CATEGORY INFORMATION
# =====================================================================
@mcp.tool()
def get_category_info(category: str = '') -> dict:
if category:
if category not in CATEGORIES:
# Suggest similar categories
det = detect_category(category)
suggestion = det["detected_category"]
return {
"status": "needs_clarification",
"message": f"Category '{category}' not recognized.",
"suggested_category": suggestion,
"alternatives": det["alternatives"],
"next_steps": [
f"Use '{suggestion}' instead",
"Try retyping the category",
"Use 'misc' if unsure"
]
}
subcats = CATEGORIES_STRUCTURE.get(category, [])
return {
"status": "ok",
"category": category,
"subcategories": subcats,
"total_subcategories": len(subcats),
"description": f"Information about {category}."
}
# Return all categories
return {
"status": "ok",
"categories": [
{
"name": cat,
"subcategories": CATEGORIES_STRUCTURE.get(cat, []),
"count": len(CATEGORIES_STRUCTURE.get(cat, []))
}
for cat in CATEGORIES
]
}
# =====================================================================
# ADD EXPENSE (SMART CATEGORY DETECTION)
# =====================================================================
@mcp.tool()
def add_expense(
amount: float,
description: str,
category: str = '',
subcategory: str = '',
date: str = '',
skip_affordability_check: bool = False
) -> dict:
"""Add a new expense with smart detection and conversational flow."""
# -----------------------------------------------------------------
# 1) Smart category auto-detection
# -----------------------------------------------------------------
detected_meta = None
if not category:
detection = detect_category(description, amount)
if detection["auto_assign"]:
category = detection["detected_category"]
detected_meta = {
"auto_detected": True,
"category": category,
"confidence": detection["confidence"],
"matched_keywords": detection["extracted_keywords"]
}
else:
# Ask user which category to choose
return {
"status": "category_suggestion",
"message": "I wasn't fully confident about the category. Please choose one.",
"detected_top": detection["detected_category"],
"confidence": detection["confidence"],
"alternatives": detection["alternatives"],
"next_steps": [
"Pick one of the suggestions",
"Specify a category manually",
"Say 'use misc'"
]
}
# If user provided an invalid category:
if category not in CATEGORIES:
guess = detect_category(category)
return {
"status": "invalid_category",
"message": f"'{category}' is not a valid category.",
"closest_match": guess["detected_category"],
"alternatives": guess["alternatives"],
"next_steps": [
f"Use '{guess['detected_category']}'",
"Pick from available categories",
"Use 'misc' if unsure"
]
}
# -----------------------------------------------------------------
# 2) Subcategory flow (conversational)
# -----------------------------------------------------------------
available_subcats = CATEGORIES_STRUCTURE.get(category, [])
if available_subcats and not subcategory:
return {
"status": "subcategory_selection_needed",
"message": f"Please pick a subcategory for '{category}'.",
"available_subcategories": available_subcats,
"instruction": (
"Call add_expense again with the chosen subcategory "
f"(e.g., subcategory='{available_subcats[0]}')."
)
}
# Validate subcategory
if subcategory and available_subcats and subcategory not in available_subcats:
return {
"status": "needs_clarification",
"message": f"'{subcategory}' is not a subcategory of '{category}'.",
"alternatives": available_subcats,
"suggestion": "Choose one from the list or use 'other'."
}
# -----------------------------------------------------------------
# 3) Date parsing (graceful fallback)
# -----------------------------------------------------------------
try:
parsed_date = parse_flexible_date(date) if date else get_today()
except ValueError:
parsed_date = get_today() # graceful fallback
date_error = True
else:
date_error = False
# -----------------------------------------------------------------
# 4) Optional affordability check
# -----------------------------------------------------------------
affordability = None
month = parsed_date[:7]
if not skip_affordability_check and budget_exists_for_month(month):
success, b = calculate_remaining_budget(month, category)
if success:
if b.get("budget_mode") == "total_only":
cat_rem = None
cat_budget = None
total_rem = b["remaining_total"]
total_budget = b["total_budget"]
else:
cat_info = b.get("category_breakdown", {}).get(category, {})
cat_rem = cat_info.get("remaining", 0)
cat_budget = cat_info.get("budget", 0)
total_rem = b["remaining_total"]
total_budget = b["total_budget"]
affordability = check_affordability_logic(
amount,
cat_rem or 0,
total_rem or 0,
cat_budget or 0,
total_budget or 0,
category,
parsed_date
)
# -----------------------------------------------------------------
# 5) Insert the expense
# -----------------------------------------------------------------
try:
expense_id = add_expense_to_db(amount, category, subcategory, parsed_date, description)
except Exception as e:
return {
"status": "soft_error",
"message": f"Couldn't add expense: {e}",
"suggestion": "Try again or adjust your data."
}
# -----------------------------------------------------------------
# 6) Build response
# -----------------------------------------------------------------
response = {
"status": "ok",
"expense": {
"id": expense_id,
"amount": amount,
"description": description,
"category": category,
"subcategory": subcategory or None,
"date": parsed_date,
}
}
if detected_meta:
response["auto_category_detection"] = detected_meta
if date_error:
response["date_warning"] = "The date you provided was unclear; I used today's date instead."
if affordability:
response["affordability"] = affordability
return response
# =====================================================================
# BULK ADD (SMART, VALIDATING, NO LEARNING MODE)
# =====================================================================
@mcp.tool()
def bulk_add_expenses(
expenses: List[Dict[str, Any]],
strict: bool = False,
auto_detect: bool = True
) -> dict:
"""
Add multiple expenses in one call.
- strict = True: stop on first failure
- auto_detect = True: guess category/subcategory when missing
"""
results = {
"added": [],
"failed": [],
"warnings": [],
}
for idx, exp in enumerate(expenses):
row_context = {"index": idx}
# ------------------------------------------------------------
# 1) Validate required fields
# ------------------------------------------------------------
amount = exp.get("amount")
description = exp.get("description", "")
if amount is None:
results["failed"].append({
**row_context,
"error": "Missing 'amount'"
})
if strict:
break
continue
if not description:
results["failed"].append({
**row_context,
"error": "Missing 'description'"
})
if strict:
break
continue
category = exp.get("category", "")
subcategory = exp.get("subcategory", "")
date = exp.get("date", "")
skip_afford = exp.get("skip_affordability_check", False)
# ------------------------------------------------------------
# 2) AUTO-DETECT CATEGORY IF MISSING
# ------------------------------------------------------------
detection_meta = None
if auto_detect and not category:
detection = detect_category(description, amount)
if detection["auto_assign"]:
category = detection["detected_category"]
detection_meta = {
"auto_detected": True,
"category": category,
"confidence": detection["confidence"],
"keywords": detection["extracted_keywords"],
}
else:
# Low confidence → soft failure
results["failed"].append({
**row_context,
"error": "Category unclear",
"suggested": detection["detected_category"],
"alternatives": detection["alternatives"],
})
if strict:
break
continue
# ------------------------------------------------------------
# 3) VALIDATE CATEGORY
# ------------------------------------------------------------
if category not in CATEGORIES:
guess = detect_category(category)
results["failed"].append({
**row_context,
"error": f"Invalid category '{category}'",
"closest_match": guess["detected_category"],
"alternatives": guess["alternatives"],
})
if strict:
break
continue
# ------------------------------------------------------------
# 4) VALIDATE OR AUTOFILL SUBCATEGORY
# ------------------------------------------------------------
subcats = CATEGORIES_STRUCTURE.get(category, [])
if subcats and not subcategory:
# Auto-assign "other" if valid
if "other" in subcats:
subcategory = "other"
else:
# Use first available
subcategory = subcats[0]
results["warnings"].append({
**row_context,
"warning": f"Missing subcategory, auto-assigned '{subcategory}'"
})
# Subcategory invalid?
if subcategory and subcats and subcategory not in subcats:
results["failed"].append({
**row_context,
"error": f"Invalid subcategory '{subcategory}' for category '{category}'",
"allowed": subcats
})
if strict:
break
continue
# ------------------------------------------------------------
# 5) DATE PARSING (graceful)
# ------------------------------------------------------------
try:
parsed_date = parse_flexible_date(date) if date else get_today()
except Exception:
parsed_date = get_today()
results["warnings"].append({
**row_context,
"warning": f"Invalid date '{date}', replaced with today."
})
# ------------------------------------------------------------
# 6) DUPLICATE CHECK ON SAME DAY
# ------------------------------------------------------------
existing = list_expenses_from_db(
category=category,
start_date=parsed_date,
end_date=parsed_date
)
if any(
e["description"] == description and abs(e["amount"] - amount) < 1e-6
for e in existing
):
results["warnings"].append({
**row_context,
"warning": "Duplicate detected; skipped"
})
continue
# ------------------------------------------------------------
# 7) ADD TO DATABASE
# ------------------------------------------------------------
try:
exp_id = add_expense_to_db(amount, category, subcategory, parsed_date, description)
except Exception as e:
results["failed"].append({
**row_context,
"error": f"Database insert failed: {e}"
})
if strict:
break
continue
# Build added row
added_entry = {
"id": exp_id,
"amount": amount,
"description": description,
"category": category,
"subcategory": subcategory,
"date": parsed_date,
}
if detection_meta:
added_entry["category_detection"] = detection_meta
results["added"].append({
**row_context,
"expense": added_entry
})
# ------------------------------------------------------------
# Final Summary
# ------------------------------------------------------------
return {
"status": (
"ok" if not results["failed"]
else "partial_success" if results["added"]
else "failed"
),
**results,
"summary": {
"total_rows": len(expenses),
"added": len(results["added"]),
"failed": len(results["failed"]),
"warnings": len(results["warnings"]),
}
}
# ---------------------------------------------------------------------
# Quick wrappers for completeness
# ---------------------------------------------------------------------
@mcp.tool()
def quick_add_expense(amount: float, description: str, date: str = '') -> dict:
return add_expense(amount, description, '', '', date)
# =====================================================================
# UPDATE EXPENSE (smart, validated, MCP-safe)
# =====================================================================
@mcp.tool()
def update_expense(
expense_id: int,
amount: Optional[float] = None,
description: Optional[str] = None,
category: Optional[str] = None,
subcategory: Optional[str] = None,
date: Optional[str] = None,
auto_detect_category: bool = True,
) -> dict:
"""
Update an existing expense with safe validation and smart detection.
"""
# 1) Fetch existing
existing_list = list_expenses_from_db()
existing = next((e for e in existing_list if e["id"] == expense_id), None)
if not existing:
return {
"status": "not_found",
"message": f"Expense ID {expense_id} does not exist."
}
original = existing.copy()
updates = {}
# ------------------------------------------------------------
# 2) Update simple fields
# ------------------------------------------------------------
if amount is not None:
updates["amount"] = amount
if description is not None:
updates["description"] = description
# ------------------------------------------------------------
# 3) CATEGORY HANDLING
# ------------------------------------------------------------
if category is None and auto_detect_category and description:
auto = detect_category(description, amount or original["amount"])
if auto["auto_assign"]:
category = auto["detected_category"]
if category is not None:
if category not in CATEGORIES:
guess = detect_category(category)
return {
"status": "invalid_category",
"message": f"'{category}' is not a valid category.",
"closest_match": guess["detected_category"],
"alternatives": guess["alternatives"]
}
updates["category"] = category
else:
category = original["category"] # fallback for subcategory validation
# ------------------------------------------------------------
# 4) SUBCATEGORY VALIDATION
# ------------------------------------------------------------
if subcategory is not None:
allowed = CATEGORIES_STRUCTURE.get(category, [])
if allowed and subcategory not in allowed:
return {
"status": "invalid_subcategory",
"message": f"'{subcategory}' is not a valid subcategory of '{category}'.",
"allowed": allowed
}
updates["subcategory"] = subcategory
# ------------------------------------------------------------
# 5) DATE VALIDATION
# ------------------------------------------------------------
if date is not None:
try:
updates["date"] = parse_flexible_date(date)
except Exception:
return {
"status": "invalid_date",
"message": f"Could not parse date '{date}'."
}
# ------------------------------------------------------------
# 6) No changes?
# ------------------------------------------------------------
if not updates:
return {
"status": "no_changes",
"message": "No valid fields to update."
}
# ------------------------------------------------------------
# 7) Apply update
# ------------------------------------------------------------
try:
update_expense_in_db(expense_id, **updates)
except Exception as e:
return {"status": "soft_error", "message": str(e)}
# ------------------------------------------------------------
# 8) Build diff result
# ------------------------------------------------------------
new_data = {**original, **updates}
diff = {
k: {"from": original[k], "to": updates[k]}
for k in updates
if original.get(k) != updates[k]
}
return {
"status": "ok",
"updated_id": expense_id,
"changes": diff,
"new_expense": new_data
}
# =====================================================================
# DELETE EXPENSE (safe, validated)
# =====================================================================
@mcp.tool()
def delete_expense(
expense_id: int,
require_confirmation: bool = False,
confirmed: bool = False
) -> dict:
"""
Delete an expense safely.
- If require_confirmation=True, user must call again with confirmed=True.
"""
# Fetch to verify existence
all_exp = list_expenses_from_db()
exp = next((e for e in all_exp if e["id"] == expense_id), None)
if not exp:
return {
"status": "not_found",
"message": f"Expense ID {expense_id} does not exist."
}
# Confirmation step to avoid mistakes
if require_confirmation and not confirmed:
return {
"status": "confirmation_needed",
"message": f"Are you sure you want to delete expense {expense_id}?",
"expense": exp,
"next_step": {
"call_again_with": {
"expense_id": expense_id,
"confirmed": True,
}
}
}
# Perform deletion
try:
delete_expense_from_db(expense_id)
except Exception as e:
return {"status": "soft_error", "message": str(e)}
return {
"status": "ok",
"deleted_id": expense_id,
"deleted_expense": exp
}
