import hashlib import re from datetime import datetime from typing import Optional from fastapi import APIRouter, Depends from pydantic import BaseModel, Field from sqlmodel import Session, select from app.auth import get_current_user from app.db import get_session from app.models.models import ( Bank, Currency, Transaction, TransactionSource, TransactionType, ) from app.api.v1.endpoints.transactions import auto_categorize router = APIRouter(prefix="/import", tags=["import"]) class PasteImportRequest(BaseModel): # ~1 MB cap: a full statement paste is a few KB; anything larger is abuse. text: str = Field(max_length=1_000_000) bank: Bank = Bank.BAC source: TransactionSource = TransactionSource.CREDIT_CARD # Override for false-positive dedup (legitimate same-day, same-amount # repeat purchases hash identically) — review UX-20/BE-08. allow_duplicates: bool = False class SkippedDuplicate(BaseModel): line: int merchant: str date: str amount: float currency: str existing_id: int existing_merchant: str existing_date: str existing_amount: float existing_source: str class PasteImportResult(BaseModel): imported: int duplicates: int errors: list[str] skipped: list[SkippedDuplicate] = [] def make_reference_hash(date: datetime, merchant: str, amount: float, currency: str) -> str: raw = f"{date.isoformat()}|{merchant.strip().upper()}|{amount}|{currency}" return hashlib.sha256(raw.encode()).hexdigest()[:16] def parse_bac_line(line: str) -> Optional[dict]: """Parse a single BAC statement line. Format: DD/MM/YYYY\tMERCHANT\\CITY\\COUNTRY\tAMOUNT CURRENCY """ line = line.strip() if not line: return None parts = re.split(r"\t+", line) if len(parts) < 3: # Try multiple spaces as delimiter parts = re.split(r"\s{2,}", line) if len(parts) < 3: return None # Parse date date_str = parts[0].strip() try: date = datetime.strptime(date_str, "%d/%m/%Y") except ValueError: return None # Parse merchant\city\country merchant_raw = parts[1].strip() merchant_parts = merchant_raw.split("\\") merchant = merchant_parts[0].strip() city = merchant_parts[1].strip() if len(merchant_parts) > 1 else None # Parse amount + currency amount_str = parts[2].strip() # Extract currency (last 3 chars) match = re.match(r"^([\d,.-]+)\s*(CRC|USD)$", amount_str, re.IGNORECASE) if not match: return None amount_raw = match.group(1).replace(",", "") currency = match.group(2).upper() amount = float(amount_raw) # Determine transaction type is_refund = amount < 0 or "PAGO RECIBIDO" in merchant.upper() tx_type = TransactionType.DEVOLUCION if is_refund else TransactionType.COMPRA amount = abs(amount) return { "date": date, "merchant": merchant, "city": city, "amount": amount, "currency": currency, "transaction_type": tx_type, } @router.post("/paste", response_model=PasteImportResult) def paste_import( req: PasteImportRequest, session: Session = Depends(get_session), _user: str = Depends(get_current_user), ): imported = 0 duplicates = 0 errors: list[str] = [] skipped: list[SkippedDuplicate] = [] lines = req.text.strip().splitlines() for i, line in enumerate(lines, 1): if not line.strip(): continue parsed = parse_bac_line(line) if parsed is None: errors.append(f"Line {i}: could not parse") continue # Generate reference hash for duplicate detection ref_hash = make_reference_hash( parsed["date"], parsed["merchant"], parsed["amount"], parsed["currency"] ) # Check for duplicates — and SHOW the user what matched instead of # silently dropping the line (UX-20). if not req.allow_duplicates: existing = session.exec( select(Transaction).where(Transaction.reference == ref_hash) ).first() if existing: duplicates += 1 skipped.append( SkippedDuplicate( line=i, merchant=parsed["merchant"], date=parsed["date"].isoformat(), amount=parsed["amount"], currency=parsed["currency"], existing_id=existing.id or 0, existing_merchant=existing.merchant, existing_date=existing.date.isoformat(), existing_amount=float(existing.amount), existing_source=existing.source.value, ) ) continue tx = Transaction( amount=parsed["amount"], currency=Currency(parsed["currency"]), merchant=parsed["merchant"], city=parsed["city"], date=parsed["date"], transaction_type=parsed["transaction_type"], source=req.source, bank=req.bank, reference=ref_hash, ) tx.category_id = auto_categorize(tx.merchant, session) session.add(tx) imported += 1 if imported > 0: session.commit() return PasteImportResult( imported=imported, duplicates=duplicates, errors=errors, skipped=skipped )