05/31 Phase 7

app/services/import_service.py

@@ -0,0 +1,169 @@
+"""
+Import Service — parses CSV files for bulk transaction import.
+Expected columns: date, type, description, category, account, amount, notes
+Date formats: YYYY-MM-DD, MM/DD/YYYY, DD/MM/YYYY
+"""
+
+import csv
+import io
+import logging
+from datetime import datetime
+from app.extensions import db
+from app.models.transaction import Transaction
+from app.models.category import Category
+from app.models.account import Account
+from app.services.account_service import calc_balance
+
+log = logging.getLogger(__name__)
+
+REQUIRED_COLS = {'date', 'type', 'description', 'amount'}
+DATE_FORMATS = ['%Y-%m-%d', '%m/%d/%Y', '%d/%m/%Y', '%Y/%m/%d', '%d-%m-%Y']
+
+
+def _parse_date(s):
+    s = s.strip()
+    for fmt in DATE_FORMATS:
+        try:
+            return datetime.strptime(s, fmt).date()
+        except ValueError:
+            continue
+    raise ValueError(f'Unrecognised date format: {s!r}')
+
+
+def _match_category(name, txn_type):
+    if not name:
+        return None
+    cat = Category.query.filter(
+        Category.name.ilike(name.strip()),
+        Category.category_type.in_([txn_type, 'both']),
+        Category.is_active == True,
+    ).first()
+    return cat.id if cat else None
+
+
+def _match_account(name):
+    if not name:
+        return None
+    acct = Account.query.filter(
+        Account.name.ilike(name.strip()),
+        Account.is_active == True,
+    ).first()
+    return acct.id if acct else None
+
+
+def parse_csv(file_content, default_account_id=None):
+    """
+    Parse CSV content (str or bytes).
+    Returns (preview_rows, errors, column_map)
+    preview_rows: list of dicts ready for import
+    errors: list of error strings
+    """
+    if isinstance(file_content, bytes):
+        file_content = file_content.decode('utf-8-sig')  # handle BOM
+
+    reader = csv.DictReader(io.StringIO(file_content))
+    headers = {h.strip().lower() for h in (reader.fieldnames or [])}
+
+    missing = REQUIRED_COLS - headers
+    if missing:
+        return [], [f'Missing required columns: {", ".join(missing)}'], {}
+
+    rows = []
+    errors = []
+
+    for i, row in enumerate(reader, start=2):
+        clean = {k.strip().lower(): v.strip() for k, v in row.items()}
+        row_errors = []
+
+        # Date
+        try:
+            txn_date = _parse_date(clean.get('date', ''))
+        except ValueError as e:
+            row_errors.append(f'Row {i}: {e}')
+            continue
+
+        # Type
+        txn_type = clean.get('type', '').lower()
+        if txn_type not in ('income', 'expense'):
+            row_errors.append(f'Row {i}: type must be "income" or "expense", got {txn_type!r}')
+            continue
+
+        # Amount
+        try:
+            amount = float(clean.get('amount', '0').replace(',', '').replace('$', '').strip())
+            if amount <= 0:
+                raise ValueError('Amount must be > 0')
+        except ValueError as e:
+            row_errors.append(f'Row {i}: invalid amount — {e}')
+            continue
+
+        # Description
+        description = clean.get('description', '').strip()
+        if not description:
+            row_errors.append(f'Row {i}: description is required')
+            continue
+
+        # Optional fields
+        category_id = _match_category(clean.get('category', ''), txn_type)
+        account_id = _match_account(clean.get('account', '')) or default_account_id
+        notes = clean.get('notes', '')
+
+        if row_errors:
+            errors.extend(row_errors)
+        else:
+            rows.append({
+                'date': txn_date,
+                'transaction_type': txn_type,
+                'description': description,
+                'amount': amount,
+                'category_id': category_id,
+                'account_id': account_id,
+                'notes': notes,
+                'category_name': clean.get('category', ''),
+                'account_name': clean.get('account', ''),
+            })
+
+    return rows, errors
+
+
+def import_rows(rows, skip_duplicates=True):
+    """
+    Insert parsed rows into DB.
+    Returns (imported_count, skipped_count)
+    """
+    imported = 0
+    skipped = 0
+    affected_accounts = set()
+
+    for row in rows:
+        if skip_duplicates:
+            existing = Transaction.query.filter_by(
+                date=row['date'],
+                description=row['description'],
+                amount=row['amount'],
+                transaction_type=row['transaction_type'],
+            ).first()
+            if existing:
+                skipped += 1
+                continue
+
+        txn = Transaction(
+            date=row['date'],
+            transaction_type=row['transaction_type'],
+            description=row['description'],
+            amount=row['amount'],
+            category_id=row.get('category_id'),
+            account_id=row.get('account_id'),
+            notes=row.get('notes', ''),
+        )
+        db.session.add(txn)
+        if row.get('account_id'):
+            affected_accounts.add(row['account_id'])
+        imported += 1
+
+    db.session.commit()
+
+    for account_id in affected_accounts:
+        calc_balance(account_id)
+
+    return imported, skipped