Spaces:
Sleeping
Sleeping
| """ | |
| environments/trace_env/tools/transaction_parser.py | |
| Universal Transaction Parser for Trace. | |
| Detects and parses ALL types of transactional Gmail: | |
| - Ride receipts (Uber, Ola, Rapido) | |
| - Food orders (Swiggy, Zomato, Dunzo) | |
| - Shopping (Amazon, Flipkart, Meesho, Myntra) | |
| - Payments (PhonePe, GPay, Paytm, NEFT/IMPS) | |
| - Subscriptions (Netflix, Spotify, YouTube, SaaS) | |
| - Travel (IRCTC, MakeMyTrip, Air India, IndiGo) | |
| - Banking (Bank alerts, EMI, credit card statements) | |
| - Utility bills (Electricity, gas, water, telecom) | |
| - Education (Udemy, Coursera, college fees) | |
| - Healthcare (Pharmacy, hospital, insurance) | |
| Works on: | |
| - Gmail body_text (plaintext extracted by gmail_tool) | |
| - image_analyses (from image_tool Ollama pipeline) | |
| - doc_analyses (from doc_tool Docling pipeline) | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from datetime import datetime | |
| from typing import Optional | |
| # ── Vendor → Category mapping ───────────────────────────────────────────────── | |
| VENDOR_MAP = { | |
| # Rides | |
| "uber": ("ride", "Uber"), | |
| "ola": ("ride", "Ola"), | |
| "rapido": ("ride", "Rapido"), | |
| "namma yatri": ("ride", "Namma Yatri"), | |
| # Food | |
| "swiggy": ("food", "Swiggy"), | |
| "zomato": ("food", "Zomato"), | |
| "dunzo": ("food", "Dunzo"), | |
| "blinkit": ("food", "Blinkit"), | |
| "zepto": ("food", "Zepto"), | |
| # Shopping | |
| "amazon": ("shopping", "Amazon"), | |
| "flipkart": ("shopping", "Flipkart"), | |
| "myntra": ("shopping", "Myntra"), | |
| "meesho": ("shopping", "Meesho"), | |
| "ajio": ("shopping", "AJIO"), | |
| "nykaa": ("shopping", "Nykaa"), | |
| "snapdeal": ("shopping", "Snapdeal"), | |
| # Payments | |
| "phonepe": ("payment", "PhonePe"), | |
| "gpay": ("payment", "Google Pay"), | |
| "google pay": ("payment", "Google Pay"), | |
| "paytm": ("payment", "Paytm"), | |
| "bhim": ("payment", "BHIM UPI"), | |
| "neft": ("payment", "NEFT Transfer"), | |
| "imps": ("payment", "IMPS Transfer"), | |
| "upi": ("payment", "UPI Transfer"), | |
| # Subscriptions | |
| "netflix": ("subscription", "Netflix"), | |
| "spotify": ("subscription", "Spotify"), | |
| "youtube": ("subscription", "YouTube Premium"), | |
| "hotstar": ("subscription", "Disney+ Hotstar"), | |
| "prime": ("subscription", "Amazon Prime"), | |
| "notion": ("subscription", "Notion"), | |
| "github": ("subscription", "GitHub"), | |
| "openai": ("subscription", "OpenAI"), | |
| "anthropic": ("subscription", "Anthropic"), | |
| # Travel | |
| "irctc": ("travel", "IRCTC"), | |
| "makemytrip": ("travel", "MakeMyTrip"), | |
| "indigo": ("travel", "IndiGo"), | |
| "air india": ("travel", "Air India"), | |
| "goibibo": ("travel", "Goibibo"), | |
| "cleartrip": ("travel", "Cleartrip"), | |
| "agoda": ("travel", "Agoda"), | |
| "booking.com": ("travel", "Booking.com"), | |
| "airbnb": ("travel", "Airbnb"), | |
| "oyo": ("travel", "OYO Rooms"), | |
| "fabhotels": ("travel", "FabHotels"), | |
| "taj": ("travel", "Taj Hotels"), | |
| "marriott": ("travel", "Marriott"), | |
| "vistara": ("travel", "Vistara"), | |
| "spicejet": ("travel", "SpiceJet"), | |
| "akasa": ("travel", "Akasa Air"), | |
| "emirates": ("travel", "Emirates"), | |
| # Banking | |
| "hdfc": ("banking", "HDFC Bank"), | |
| "icici": ("banking", "ICICI Bank"), | |
| "sbi": ("banking", "SBI"), | |
| "axis": ("banking", "Axis Bank"), | |
| "kotak": ("banking", "Kotak Bank"), | |
| "idfc": ("banking", "IDFC Bank"), | |
| # Utilities | |
| "bescom": ("utility", "BESCOM"), | |
| "airtel": ("utility", "Airtel"), | |
| "jio": ("utility", "Jio"), | |
| "vi ": ("utility", "Vi"), | |
| "bsnl": ("utility", "BSNL"), | |
| # Education | |
| "udemy": ("education", "Udemy"), | |
| "coursera": ("education", "Coursera"), | |
| "unacademy": ("education", "Unacademy"), | |
| "byju": ("education", "BYJU'S"), | |
| # Healthcare | |
| "apollo": ("healthcare", "Apollo Pharmacy"), | |
| "1mg": ("healthcare", "1mg"), | |
| "practo": ("healthcare", "Practo"), | |
| } | |
| # ── Category display config ─────────────────────────────────────────────────── | |
| CATEGORY_CONFIG = { | |
| "ride": {"icon": "🚗", "color": "#1a1a1a", "label": "Ride"}, | |
| "food": {"icon": "🍔", "color": "#e85d04", "label": "Food & Delivery"}, | |
| "shopping": {"icon": "🛍️", "color": "#7c6fcd", "label": "Shopping"}, | |
| "payment": {"icon": "💸", "color": "#1a7a4a", "label": "Payment"}, | |
| "subscription": {"icon": "🔄", "color": "#0077b6", "label": "Subscription"}, | |
| "travel": {"icon": "✈️", "color": "#6d4c41", "label": "Travel"}, | |
| "banking": {"icon": "🏦", "color": "#b5179e", "label": "Banking"}, | |
| "utility": {"icon": "⚡", "color": "#4a6741", "label": "Utility Bill"}, | |
| "education": {"icon": "📚", "color": "#e63946", "label": "Education"}, | |
| "healthcare": {"icon": "💊", "color": "#d62828", "label": "Healthcare"}, | |
| "unknown": {"icon": "📧", "color": "#6b6560", "label": "Transactional"}, | |
| } | |
| def detect_category(text: str) -> tuple[str, str]: | |
| """Detect transaction category and vendor from email text.""" | |
| text_lower = text.lower() | |
| for keyword, (category, vendor) in VENDOR_MAP.items(): | |
| if keyword in text_lower: | |
| return category, vendor | |
| return "unknown", "Unknown" | |
| def extract_amounts(text: str) -> list[str]: | |
| """Extract all monetary amounts from text.""" | |
| patterns = [ | |
| r'(?:₹|Rs\.?|INR|USD|\$|€|£)\s?[-\u2010\u2212]?\s*[\d,]+(?:\.\d{1,2})?', | |
| r'[-\u2010\u2212]?\s*[\d,]+(?:\.\d{1,2})?\s*(?:₹|Rs\.?|INR)', | |
| r'(?:Total(?: Amount)?|Grand Total|Amount Paid|You paid|Net Amount|Final Amount|Total Fare|Amount|Total Payable)\b.{0,100}?[-\u2010\u2212]?\s*([\d,]+(?:\.\d{1,2})?)\b', | |
| ] | |
| amounts = [] | |
| for pattern in patterns: | |
| found = re.findall(pattern, text, re.IGNORECASE | re.DOTALL) | |
| amounts.extend(found) | |
| return list(dict.fromkeys(amounts))[:10] # deduplicate, max 10 | |
| def extract_total(text: str, structured_total: Optional[str] = None) -> Optional[str]: | |
| """ | |
| Extract the final total amount. Priority: | |
| 1. Explicit 'Grand Total' / 'Total Payable' label (highest confidence) | |
| 2. Structured tool output (from VLM/Docling JSON) | |
| 3. Amounts inside Docling Markdown table pipes | |
| 4. Largest Rs/INR amount in text (fallback) | |
| """ | |
| # Priority 1: Labelled totals - find label then grab number within 150 chars | |
| labelled_patterns = [ | |
| r'(?:Grand\s+Total|Total\s+Payable|Total\s+Amount\s+Paid|Amount\s+Paid|Net\s+Amount|You\s+Paid|Paid|Sent|Payment\s+of|Transaction\s+of|Final\s+Amount|Total\s+Fare|Total\s+Due|Amount\s+Due|Invoice\s+Total)\b.{0,150}?(?:[\u20b9]|INR|Rs\.?|₹)\s*([\d,]+(?:\.\d{1,2})?)', | |
| r'(?:Grand\s+Total|Total\s+Payable|Total\s+Amount\s+Paid|Amount\s+Paid|Net\s+Amount|You\s+Paid|Paid|Sent|Payment\s+of|Transaction\s+of|Final\s+Amount|Total\s+Fare|Total\s+Due|Amount\s+Due|Invoice\s+Total)\b.{0,150}?([\d,]+(?:\.\d{1,2})?)', | |
| ] | |
| for pattern in labelled_patterns: | |
| match = re.search(pattern, text, re.IGNORECASE | re.DOTALL) | |
| if match: | |
| val = match.group(1).replace(',', '') | |
| try: | |
| if float(val) > 0: | |
| return f"\u20b9{match.group(1)}" | |
| except ValueError: | |
| pass | |
| if structured_total: | |
| return structured_total | |
| # Priority 2: Docling Markdown table cells with currency | |
| pipe_match = re.search( | |
| r'\|\s*(?:[\u20b9]|INR|Rs\.?)\s*([\d,]+(?:\.\d{1,2})?)\s*\|', | |
| text, re.IGNORECASE | |
| ) | |
| if pipe_match: | |
| return f"\u20b9{pipe_match.group(1)}" | |
| # Priority 3: Largest currency amount in text (avoids picking up tax sub-totals) | |
| all_amounts = re.findall( | |
| r'(?:[\u20b9]|INR|Rs\.?)\s*([\d,]+(?:\.\d{1,2})?)', | |
| text, re.IGNORECASE | |
| ) | |
| if all_amounts: | |
| try: | |
| largest = max(all_amounts, key=lambda x: float(x.replace(',', ''))) | |
| if float(largest.replace(',', '')) > 0: | |
| return f"\u20b9{largest}" | |
| except ValueError: | |
| pass | |
| return None | |
| def extract_dates(text: str) -> list[str]: | |
| """Extract dates from text.""" | |
| patterns = [ | |
| r'\d{1,2}\s+(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]*\s+\d{4}', | |
| r'(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]*\s+\d{1,2},?\s+\d{4}', | |
| r'\d{1,2}[/-]\d{1,2}[/-]\d{2,4}', | |
| r'\d{4}-\d{2}-\d{2}', | |
| ] | |
| dates = [] | |
| for p in patterns: | |
| dates.extend(re.findall(p, text, re.IGNORECASE)) | |
| return list(dict.fromkeys(dates))[:5] | |
| def extract_order_id(text: str) -> Optional[str]: | |
| """Extract order/transaction ID.""" | |
| patterns = [ | |
| r'Order\s*(?:ID|#|No\.?)\s*:?\s*([A-Z0-9\-]{6,20})', | |
| r'Transaction\s*(?:ID|#|No\.?)\s*:?\s*([A-Z0-9\-]{6,20})', | |
| r'Booking\s*(?:ID|#|No\.?)\s*:?\s*([A-Z0-9\-]{6,20})', | |
| r'(?:UPI|Ref)\s*(?:Ref|ID|No\.?)\s*:?\s*([A-Z0-9]{8,20})', | |
| r'#([A-Z0-9\-]{8,20})', | |
| ] | |
| for p in patterns: | |
| match = re.search(p, text, re.IGNORECASE) | |
| if match: | |
| return match.group(1) | |
| return None | |
| def parse_ride_details(text: str) -> dict: | |
| """Extract ride-specific details.""" | |
| details = {} | |
| km = re.search(r'([\d.]+)\s*(?:km|kilometres?|kms)', text, re.IGNORECASE) | |
| if km: | |
| details["distance"] = f"{km.group(1)} km" | |
| mins = re.search(r'(\d+)\s*(?:minutes?|mins?)', text, re.IGNORECASE) | |
| if mins: | |
| details["duration"] = f"{mins.group(1)} min" | |
| plate = re.search(r'\b([A-Z]{2}\d{2}[A-Z]{1,2}\d{4})\b', text) | |
| if plate: | |
| details["license_plate"] = plate.group(1) | |
| # pickup/drop from common patterns | |
| from_match = re.search(r'(?:From|Pickup|Start)\s*:?\s*(.{10,60}?)(?:\n|To:|Drop)', text, re.IGNORECASE) | |
| to_match = re.search(r'(?:To|Drop|End|Destination)\s*:?\s*(.{10,60}?)(?:\n|$)', text, re.IGNORECASE) | |
| if from_match: | |
| details["from"] = from_match.group(1).strip()[:60] | |
| if to_match: | |
| details["to"] = to_match.group(1).strip()[:60] | |
| return details | |
| def parse_food_details(text: str) -> dict: | |
| """Extract food order details.""" | |
| details = {} | |
| restaurant = re.search(r'(?:from|restaurant|ordered from)\s*:?\s*([A-Za-z\s&\']{3,40})', text, re.IGNORECASE) | |
| if restaurant: | |
| details["restaurant"] = restaurant.group(1).strip() | |
| items = re.findall(r'^\s*[\d]+\s*[xX×]\s*(.{5,50}?)(?:\s+₹|\s+Rs)', text, re.MULTILINE) | |
| if items: | |
| details["items"] = items[:5] | |
| delivery = re.search(r'Delivery\s*(?:fee|charge)\s*:?\s*(?:₹|Rs\.?)?\s*([\d,]+(?:\.\d{2})?)', text, re.IGNORECASE) | |
| if delivery: | |
| details["delivery_fee"] = f"₹{delivery.group(1)}" | |
| return details | |
| def parse_payment_details(text: str) -> dict: | |
| """Extract payment/UPI details.""" | |
| details = {} | |
| to_match = re.search(r'(?:To|Paid to|Sent to|Recipient)\s*:?\s*(.{3,50}?)(?:\n|$)', text, re.IGNORECASE) | |
| from_match = re.search(r'(?:From|Paid from|Account)\s*:?\s*(.{3,50}?)(?:\n|$)', text, re.IGNORECASE) | |
| if to_match: | |
| details["to"] = to_match.group(1).strip() | |
| if from_match: | |
| details["from"] = from_match.group(1).strip() | |
| upi = re.search(r'UPI\s*(?:ID|Ref)\s*:?\s*([\w@.]+)', text, re.IGNORECASE) | |
| if upi: | |
| details["upi_ref"] = upi.group(1) | |
| status = re.search(r'\b(Success(?:ful)?|Failed|Pending|Declined|Completed)\b', text, re.IGNORECASE) | |
| if status: | |
| details["status"] = status.group(1) | |
| return details | |
| def parse_shopping_details(text: str) -> dict: | |
| """Extract shopping order details.""" | |
| details = {} | |
| items = re.findall(r'^\s*[-•]\s*(.{5,80}?)(?:\s+₹|\s+Rs|\n)', text, re.MULTILINE) | |
| if not items: | |
| items = re.findall(r'Item\s*:?\s*(.{5,60}?)(?:\n|$)', text, re.IGNORECASE) | |
| if items: | |
| details["items"] = [i.strip() for i in items[:5]] | |
| delivery = re.search(r'(?:Expected|Delivery|Estimated)\s*(?:by|date|on)?\s*:?\s*(.{5,40}?)(?:\n|$)', text, re.IGNORECASE) | |
| if delivery: | |
| details["delivery_date"] = delivery.group(1).strip() | |
| return details | |
| def parse_travel_details(text: str) -> dict: | |
| """Extract travel (flight/hotel/train) details.""" | |
| details = {} | |
| # Flight details | |
| flight_pnr = re.search(r'(?:PNR|Booking Ref|Booking Reference)\s*:?\s*([A-Z0-9]{5,8})', text, re.IGNORECASE) | |
| if flight_pnr: | |
| details["pnr"] = flight_pnr.group(1) | |
| flight_no = re.search(r'(?:Flight|Flight No)\s*:?\s*([A-Z0-9]{2,6}\s*\d{1,4})', text, re.IGNORECASE) | |
| if flight_no: | |
| details["flight_no"] = flight_no.group(1).strip() | |
| # Hotel details | |
| check_in = re.search(r'(?:Check-in|Check in|Arrival)\s*:?\s*([\w\s,-]{5,20})(?:\n|$)', text, re.IGNORECASE) | |
| if check_in: | |
| details["check_in"] = check_in.group(1).strip() | |
| check_out = re.search(r'(?:Check-out|Check out|Departure)\s*:?\s*([\w\s,-]{5,20})(?:\n|$)', text, re.IGNORECASE) | |
| if check_out: | |
| details["check_out"] = check_out.group(1).strip() | |
| # Generic Travel | |
| passenger = re.search(r'(?:Passenger|Guest|Name)\s*:?\s*([A-Za-z\s]{3,40})(?:\n|$)', text, re.IGNORECASE) | |
| if passenger: | |
| details["passenger"] = passenger.group(1).strip() | |
| return details | |
| def parse_transaction(email: dict) -> dict: | |
| """ | |
| Main entry point. Parse a Gmail email dict into a structured transaction. | |
| Input: email dict from gmail_tool.search_gmail_with_attachments() | |
| Output: enriched transaction dict for the dashboard renderer. | |
| """ | |
| body = email.get("body_text", "") or email.get("snippet", "") | |
| # Check Structured Tool Outputs FIRST to avoid hallucinations | |
| structured_total = None | |
| for analysis in email.get("image_analyses", []): | |
| if isinstance(analysis, dict) and analysis.get("totals", {}).get("total"): | |
| structured_total = analysis["totals"]["total"] | |
| break | |
| if not structured_total: | |
| for doc in email.get("doc_analyses", []): | |
| if isinstance(doc, dict): | |
| amounts = doc.get("entities", {}).get("amounts", []) | |
| if amounts: # Docling identifies the largest amounts as totals | |
| structured_total = amounts[0] | |
| break | |
| # Append text from image attachments | |
| for analysis in email.get("image_analyses", []): | |
| if isinstance(analysis, dict): | |
| body += "\n" + analysis.get("extracted_text", "") | |
| body += "\n" + analysis.get("summary", "") | |
| # Append text from document attachments | |
| for doc in email.get("doc_analyses", []): | |
| if isinstance(doc, dict): | |
| body += "\n" + doc.get("extracted_text", "") | |
| # Doc analyses might have embedded image analyses too | |
| for img_analysis in doc.get("image_analyses", []): | |
| if isinstance(img_analysis, dict): | |
| body += "\n" + img_analysis.get("extracted_text", "") | |
| subject = email.get("subject", "") | |
| from_email = email.get("from", "") | |
| full_text = subject + " " + from_email + " " + body | |
| # Detect category | |
| category, vendor = detect_category(full_text) | |
| # Extract universals | |
| total = extract_total(full_text, structured_total) | |
| amounts = extract_amounts(full_text) | |
| dates = extract_dates(full_text) | |
| order_id = extract_order_id(full_text) | |
| # Category-specific parsing | |
| details = {} | |
| if category == "ride": | |
| details = parse_ride_details(body) | |
| elif category == "food": | |
| details = parse_food_details(body) | |
| elif category == "payment": | |
| details = parse_payment_details(body) | |
| elif category == "shopping": | |
| details = parse_shopping_details(body) | |
| elif category == "travel": | |
| details = parse_travel_details(body) | |
| # Build transaction object | |
| transaction = { | |
| "id": email.get("id", ""), | |
| "category": category, | |
| "vendor": vendor, | |
| "category_config": CATEGORY_CONFIG.get(category, CATEGORY_CONFIG["unknown"]), | |
| "subject": subject, | |
| "from_email": email.get("from", ""), | |
| "date": email.get("date", ""), | |
| "total": total, | |
| "amounts": amounts, | |
| "dates": dates, | |
| "order_id": order_id, | |
| "details": details, | |
| "snippet": email.get("snippet", "")[:200], | |
| "body_preview": body[:800], | |
| "image_analyses": email.get("image_analyses", []), | |
| "doc_analyses": email.get("doc_analyses", []), | |
| "attachment_count": email.get("attachment_count", 0), | |
| "reimbursable": _is_reimbursable(body, category), | |
| "payment_method": _detect_payment_method(full_text), | |
| } | |
| return transaction | |
| def _is_reimbursable(text: str, category: str) -> bool: | |
| """Check if receipt is reimbursable.""" | |
| if category in ("ride", "food", "travel", "shopping"): | |
| keywords = ["reimburs", "official", "business", "expense", "tax invoice", "gst"] | |
| return any(k in text.lower() for k in keywords) | |
| return False | |
| def _detect_payment_method(text: str) -> str: | |
| """Detect payment method used.""" | |
| text_lower = text.lower() | |
| if "cash" in text_lower: | |
| return "Cash" | |
| if "upi" in text_lower: | |
| return "UPI" | |
| if "credit card" in text_lower or "credit" in text_lower: | |
| return "Credit Card" | |
| if "debit card" in text_lower or "debit" in text_lower: | |
| return "Debit Card" | |
| if "wallet" in text_lower: | |
| return "Wallet" | |
| if "net banking" in text_lower or "netbanking" in text_lower: | |
| return "Net Banking" | |
| return "Unknown" | |
| def parse_transactions_bulk(emails: list[dict]) -> dict: | |
| """ | |
| Parse a list of emails into transactions and compute summary stats. | |
| Returns: | |
| { | |
| "transactions": [...], | |
| "summary": { | |
| "total_spend": float, | |
| "count": int, | |
| "by_category": {...}, | |
| "by_vendor": {...}, | |
| } | |
| } | |
| """ | |
| # ── Step 1: Parse and Deduplicate ─────────────────────────────────────── | |
| merged = {} | |
| for item in emails: | |
| # Check if this is already a parsed transaction (e.g. from Google Sheets) | |
| if item.get("_source") == "sheets": | |
| t = item | |
| else: | |
| t = parse_transaction(item) | |
| tid = t.get("id") | |
| if not tid: | |
| if t.get("_source") == "sheets": | |
| import uuid | |
| tid = f"sheets_legacy_{uuid.uuid4().hex[:8]}" | |
| t["id"] = tid | |
| else: | |
| continue | |
| # If we already have this ID, MERGE the data | |
| if tid in merged: | |
| existing = merged[tid] | |
| # Identify which version is which | |
| new_is_sheets = t.get("_source") == "sheets" | |
| existing_is_sheets = existing.get("_source") == "sheets" | |
| # If we have one Gmail and one Sheets version, merge them | |
| if (new_is_sheets and not existing_is_sheets) or (not new_is_sheets and existing_is_sheets): | |
| gmail_v = existing if not existing_is_sheets else t | |
| sheets_v = existing if existing_is_sheets else t | |
| # Use Sheets version for editable fields (may contain user overrides) | |
| merged_t = gmail_v.copy() | |
| for field in ["category", "vendor", "total", "payment_method", "notes"]: | |
| if sheets_v.get(field) and str(sheets_v[field]).lower() not in ("", "unknown"): | |
| merged_t[field] = sheets_v[field] | |
| # Special handling for notes: combine Sheets notes with Gmail details | |
| sheet_notes = sheets_v.get("notes", "") | |
| if sheet_notes: | |
| if "details" not in merged_t: | |
| merged_t["details"] = {} | |
| merged_t["details"]["sheet_notes"] = sheet_notes | |
| merged_t["_source"] = "merged" | |
| merged[tid] = merged_t | |
| else: | |
| # Both are same source, use previous heuristic (keep one with more info) | |
| existing_has_total = bool(existing.get("total") and re.sub(r'[^\d.]', '', str(existing["total"]))) | |
| new_has_total = bool(t.get("total") and re.sub(r'[^\d.]', '', str(t["total"]))) | |
| if new_has_total and not existing_has_total: | |
| merged[tid] = t | |
| elif new_has_total and existing_has_total: | |
| if len(t.get("details", {})) > len(existing.get("details", {})): | |
| merged[tid] = t | |
| else: | |
| merged[tid] = t | |
| transactions = list(merged.values()) | |
| # ── Step 2: Compute summary stats ────────────────────────────────────── | |
| total_spend = 0.0 | |
| by_category = {} | |
| by_vendor = {} | |
| for t in transactions: | |
| if t["total"]: | |
| amount_str = re.sub(r'[^\d.]', '', t["total"]) | |
| try: | |
| amount = float(amount_str) | |
| total_spend += amount | |
| cat = t["category"] | |
| by_category[cat] = by_category.get(cat, 0) + amount | |
| vendor = t["vendor"] | |
| by_vendor[vendor] = by_vendor.get(vendor, 0) + amount | |
| except ValueError: | |
| pass | |
| return { | |
| "transactions": transactions, | |
| "summary": { | |
| "total_spend": round(total_spend, 2), | |
| "count": len(transactions), | |
| "by_category": {k: round(v, 2) for k, v in sorted(by_category.items(), key=lambda x: -x[1])}, | |
| "by_vendor": {k: round(v, 2) for k, v in sorted(by_vendor.items(), key=lambda x: -x[1])}, | |
| } | |
| } | |