Spaces:
Sleeping
Sleeping
File size: 5,733 Bytes
17a78b5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | import json
import logging
from datetime import datetime
from langchain_core.tools import tool
from langgraph.types import interrupt
from src.db.connection import get_connection
logger = logging.getLogger("cashy.tools")
@tool
def create_transaction(
transaction_type: str,
transaction_description: str,
amount: float,
account_name: str,
category_name: str = "",
date: str = "",
notes: str = "",
) -> str:
"""Create a new financial transaction (expense, income, or transfer).
transaction_type must be 'expense', 'income', or 'transfer'.
amount must be a positive number.
date format: YYYY-MM-DD (optional, defaults to today).
The user will be asked to confirm before the transaction is created."""
logger.info("[create_transaction] type=%s amount=%.2f account=%s category=%s",
transaction_type, amount, account_name, category_name or "none")
# --- Validation ---
if amount <= 0:
return json.dumps({"success": False, "error": "Amount must be positive"})
if transaction_type not in ("expense", "income", "transfer"):
return json.dumps(
{"success": False, "error": "transaction_type must be 'expense', 'income', or 'transfer'"}
)
if date and date.strip():
try:
transaction_date = datetime.strptime(date.strip(), "%Y-%m-%d").date()
except ValueError:
return json.dumps({"success": False, "error": "Invalid date format. Use YYYY-MM-DD"})
else:
transaction_date = datetime.now().date()
# --- Resolve names to IDs ---
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, name FROM accounts WHERE name ILIKE %s AND is_active = true",
(f"%{account_name}%",),
)
account = cur.fetchone()
if not account:
return json.dumps(
{"success": False, "error": f"Account '{account_name}' not found"}
)
account_id, account_full_name = account
category_id = None
category_full_name = "Uncategorized"
if category_name and category_name.strip():
cur.execute(
"SELECT id, name FROM categories WHERE name ILIKE %s AND is_active = true",
(f"%{category_name}%",),
)
category = cur.fetchone()
if category:
category_id, category_full_name = category
except Exception as e:
logger.error("[create_transaction] Lookup error: %s", e)
return json.dumps({"success": False, "error": str(e)})
entry_type = "credit" if transaction_type == "income" else "debit"
desc = transaction_description.strip()
note = notes.strip() if notes and notes.strip() else None
# --- Confirmation gate ---
confirmation = {
"action": "create_transaction",
"message": f"Create {transaction_type} of ${amount:.2f} on {account_full_name}?",
"details": {
"type": transaction_type,
"amount": float(amount),
"account": account_full_name,
"category": category_full_name,
"date": str(transaction_date),
"description": desc,
},
}
response = interrupt(confirmation)
if not response.get("approved"):
logger.info("[create_transaction] Cancelled by user")
return json.dumps({"success": False, "message": "Transaction cancelled by user"})
# --- Execute DB write ---
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO transactions
(transaction_date, description, transaction_type, total_amount, notes)
VALUES (%s, %s, %s, %s, %s)
RETURNING id
""",
(transaction_date, desc, transaction_type, amount, note),
)
transaction_id = cur.fetchone()[0]
cur.execute(
"""
INSERT INTO transaction_entries
(transaction_id, account_id, category_id, amount, entry_type, description)
VALUES (%s, %s, %s, %s, %s, %s)
RETURNING id
""",
(transaction_id, account_id, category_id, amount, entry_type, desc),
)
entry_id = cur.fetchone()[0]
logger.info("[create_transaction] Created txn_id=%d entry_id=%d", transaction_id, entry_id)
return json.dumps(
{
"success": True,
"transaction_id": transaction_id,
"entry_id": entry_id,
"message": f"{transaction_type.title()} of ${amount:.2f} recorded on {transaction_date}",
"details": {
"type": transaction_type,
"amount": float(amount),
"account": account_full_name,
"category": category_full_name,
"date": str(transaction_date),
"description": desc,
},
},
default=str,
)
except Exception as e:
logger.error("[create_transaction] Error: %s", e)
return json.dumps({"success": False, "error": str(e)})
|