import hashlib import re from datetime import datetime from typing import Optional from fastapi import APIRouter, Depends from pydantic import BaseModel 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): text: str bank: Bank = Bank.BAC source: TransactionSource = TransactionSource.CREDIT_CARD class PasteImportResult(BaseModel): imported: int duplicates: int errors: list[str] 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] = [] 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 existing = session.exec( select(Transaction).where(Transaction.reference == ref_hash) ).first() if existing: duplicates += 1 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)