Spaces:
Sleeping
Sleeping
File size: 8,703 Bytes
35d761c 494f97c 35d761c d6d30f3 35d761c 5bf21f7 c881d7b 35d761c | 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 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | # backend/app/api/process.py
from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.orm import Session
from typing import List
from app.models.schemas import Transaction, ProcessResponse, ProcessRequest
from app.ai.orchestrator import process_transactions
from app.db.database import get_db
from app.db.models import JournalEntry, LineItem, Account
from app.api.auth import get_current_user
from app.db.models import User
router = APIRouter()
import hashlib
from app.api.auth import get_current_user_id
from app.api.rate_limiter import process_rate_limiter
@router.post("/", response_model=ProcessResponse, dependencies=[Depends(process_rate_limiter)])
async def trigger_ai_processing(
req: ProcessRequest,
db: Session = Depends(get_db), # <-- 1. This opens the connection to Supabase
user: User = Depends(get_current_user) # <-- 2. Get the authenticated user
):
"""Takes parsed transactions, runs them through the AI, and saves them to the Vault."""
user_id = user.user_id # Extract the ID string
try:
# 1. Filter out duplicates BEFORE running expensive AI
unique_transactions = []
skipped_count = 0
seen_hashes = set() # Track hashes we've seen in THIS file
for txn in req.transactions:
# Create a stable fingerprint locked to the user
raw_string = f"{user_id}|{txn.date}|{txn.description}|{txn.amount}"
tx_hash = hashlib.sha256(raw_string.encode()).hexdigest()
# Check if we already processed this exact row in THIS file
if tx_hash in seen_hashes:
skipped_count += 1
continue
# Check if hash already exists in DB from a previous upload
exists = db.query(JournalEntry).filter(JournalEntry.tx_hash == tx_hash).first()
if exists:
skipped_count += 1
else:
seen_hashes.add(tx_hash)
txn.tx_hash = tx_hash
unique_transactions.append(txn)
print(f"🔄 Deduplication: Skipped {skipped_count} existing transactions. Processing {len(unique_transactions)} new ones.")
if not unique_transactions:
return {
"message": f"Ignored {skipped_count} duplicates. No new transactions to process.",
"total_processed": 0,
"results": []
}
# 2. Run the 5-Tier Categorization Engine
categorized_results = process_transactions(unique_transactions, user_id, db)
# 3. The Immutable Audit Trail (Double-Entry)
print("Saving AI decisions to Supabase Vault...")
# Grab the user's source account
source_account = db.query(Account).filter(
Account.id == req.source_account_id,
Account.user_id == user_id
).first()
if not source_account:
raise HTTPException(status_code=400, detail="Invalid source account selected.")
# Grab fallback accounts
default_expense_account = db.query(Account).filter(Account.user_id == user_id, Account.account_type == "EXPENSE").first()
default_revenue_account = db.query(Account).filter(Account.user_id == user_id, Account.account_type == "REVENUE").first()
# Find credit card account for transfers
credit_card_account = db.query(Account).filter(Account.user_id == user_id, Account.name == "Credit Card").first()
checking_account = db.query(Account).filter(Account.user_id == user_id, Account.name == "Checking Account").first()
for result in categorized_results:
raw = result.get("raw_transaction", {})
desc_lower = raw.get("description", "").lower()
abs_amount = abs(float(raw.get("amount", 0.0)))
# --- TRANSFER DETECTION LOGIC ---
is_money_leaving = float(raw.get("amount", 0.0)) < 0
is_transfer = any(keyword in desc_lower for keyword in ["credit card payment", "autopay", "brex pay", "amex pay", "chase cc"])
if is_transfer and credit_card_account and checking_account:
print(f"🔄 Transfer Detected: {raw.get('description')} - Bypassing AI Categorizer")
# A payment is a transfer from Checking to Credit Card
# DR Credit Card (Liability decreases)
# CR Checking Account (Asset decreases)
journal_entry = JournalEntry(
user_id=user_id,
tx_hash=raw.get("tx_hash"),
date=raw.get("date"),
description=raw.get("description"),
status="categorized"
)
db.add(journal_entry)
db.flush()
debit_leg = LineItem(
journal_entry_id=journal_entry.id,
account_id=credit_card_account.id,
amount=abs_amount,
entry_type="DEBIT"
)
credit_leg = LineItem(
journal_entry_id=journal_entry.id,
account_id=checking_account.id,
amount=abs_amount,
entry_type="CREDIT"
)
db.add(debit_leg)
db.add(credit_leg)
continue # Skip normal expense/revenue processing
# --- NORMAL EXPENSE/REVENUE LOGIC ---
# Pre-AI guard: amount sign is deterministic truth — don't let AI override it
if not is_money_leaving and source_account.account_type == "ASSET":
# Positive amount into a checking account = Revenue, bypass AI entirely
print(f"💰 Revenue Detected: {raw.get('description')} — routing to SaaS Subscriptions")
target_account = default_revenue_account
else:
predicted_account_name = result.get("account_name", "Software Subscriptions")
target_account = db.query(Account).filter(
Account.user_id == user_id,
Account.name == predicted_account_name
).first()
# Smart fallback: if AI returned an unknown account name
if not target_account:
target_account = default_expense_account if is_money_leaving else default_revenue_account
print(f"⚠️ Account '{predicted_account_name}' not found — falling back to {'EXPENSE' if is_money_leaving else 'REVENUE'} default")
# 1. Create the Receipt Header
journal_entry = JournalEntry(
user_id=user_id,
tx_hash=raw.get("tx_hash"),
date=raw.get("date"),
description=raw.get("description"),
status=result.get("status", "categorized"),
ambiguity_reason=result.get("ambiguity_reason")
)
db.add(journal_entry)
db.flush() # Force PostgreSQL to generate the ID so we can attach the legs!
# 2. Create the Legs (Debits and Credits)
# The direction of the flow is strictly dictated by whether money
# is entering (> 0) or leaving (< 0) the source account.
debit_leg = LineItem(
journal_entry_id=journal_entry.id,
account_id=target_account.id if is_money_leaving else source_account.id,
amount=abs_amount,
entry_type="DEBIT"
)
credit_leg = LineItem(
journal_entry_id=journal_entry.id,
account_id=source_account.id if is_money_leaving else target_account.id,
amount=abs_amount,
entry_type="CREDIT"
)
db.add(debit_leg)
db.add(credit_leg)
# Push all records to Supabase securely
db.commit()
print("✅ Successfully saved to Supabase!")
return {
"message": f"Processing complete. Ignored {skipped_count} duplicates.",
"total_processed": len(categorized_results),
"results": categorized_results
}
except Exception as e:
import traceback
import logging
# Log internally only — never expose stack trace to client
logging.error("AI Processing Error:\n" + traceback.format_exc())
db.rollback()
raise HTTPException(status_code=500, detail="An internal error occurred during processing. Please try again.") |