107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
import json
|
|
import logging
|
|
from flask import Blueprint, render_template, request, jsonify
|
|
from flask_login import login_required
|
|
from app.models.account import Account
|
|
from app.models.category import Category
|
|
from app.services.bank_import_service import parse_file, do_import
|
|
|
|
bank_import_bp = Blueprint('bank_import', __name__, url_prefix='/bank-import')
|
|
log = logging.getLogger(__name__)
|
|
|
|
ALLOWED_EXTENSIONS = {'csv', 'ofx', 'qfx', 'txt', 'pdf'}
|
|
|
|
|
|
@bank_import_bp.route('/')
|
|
@login_required
|
|
def index():
|
|
accounts = Account.query.filter_by(is_active=True).order_by(Account.name).all()
|
|
categories = Category.query.filter_by(is_active=True).order_by(Category.name).all()
|
|
return render_template('bank_import/index.html',
|
|
accounts=accounts,
|
|
categories=categories)
|
|
|
|
|
|
@bank_import_bp.route('/parse', methods=['POST'])
|
|
@login_required
|
|
def parse():
|
|
"""
|
|
Accepts a multipart/form-data POST with:
|
|
file — the bank statement file
|
|
col_map — JSON string (optional); user-supplied column mapping
|
|
|
|
Returns JSON with parsed rows, detected format, errors, or needs_mapping flag.
|
|
"""
|
|
f = request.files.get('file')
|
|
if not f or not f.filename:
|
|
return jsonify({'error': 'No file uploaded'}), 400
|
|
|
|
ext = f.filename.rsplit('.', 1)[-1].lower() if '.' in f.filename else ''
|
|
if ext not in ALLOWED_EXTENSIONS:
|
|
return jsonify({'error': f'Unsupported file type: .{ext}. Use CSV, OFX, or QFX.'}), 400
|
|
|
|
col_map = None
|
|
raw_map = request.form.get('col_map', '')
|
|
if raw_map:
|
|
try:
|
|
col_map = json.loads(raw_map)
|
|
except (ValueError, TypeError):
|
|
return jsonify({'error': 'Invalid col_map JSON'}), 400
|
|
|
|
file_bytes = f.read()
|
|
log.info('[bank_import] parse request: filename=%s size=%d bytes', f.filename, len(file_bytes))
|
|
|
|
try:
|
|
result = parse_file(file_bytes, f.filename, col_map=col_map)
|
|
except Exception as e:
|
|
log.error('[bank_import] parse failed: %s', e, exc_info=True)
|
|
return jsonify({'error': f'Failed to parse file: {e}'}), 500
|
|
|
|
return jsonify(result)
|
|
|
|
|
|
@bank_import_bp.route('/import', methods=['POST'])
|
|
@login_required
|
|
def confirm_import():
|
|
"""
|
|
Accepts JSON:
|
|
{
|
|
"account_id": int,
|
|
"skip_dupes": bool,
|
|
"rows": [
|
|
{"date": "YYYY-MM-DD", "transaction_type": "expense|income",
|
|
"amount": float, "description": str, "category_id": int|null,
|
|
"notes": str, "source_id": str|null},
|
|
...
|
|
]
|
|
}
|
|
Returns {"imported": N, "skipped": N}.
|
|
"""
|
|
data = request.get_json()
|
|
if not data:
|
|
return jsonify({'error': 'No JSON payload'}), 400
|
|
|
|
rows = data.get('rows', [])
|
|
account_id = data.get('account_id')
|
|
skip_dupes = data.get('skip_dupes', True)
|
|
|
|
if not rows:
|
|
return jsonify({'error': 'No rows provided'}), 400
|
|
if not account_id:
|
|
return jsonify({'error': 'account_id is required'}), 400
|
|
|
|
# Validate account exists
|
|
acct = Account.query.get(int(account_id))
|
|
if not acct:
|
|
return jsonify({'error': f'Account {account_id} not found'}), 400
|
|
|
|
log.info('[bank_import] importing %d rows into account_id=%s skip_dupes=%s',
|
|
len(rows), account_id, skip_dupes)
|
|
try:
|
|
imported, skipped = do_import(rows, int(account_id), skip_dupes=skip_dupes)
|
|
except Exception as e:
|
|
log.error('[bank_import] import failed: %s', e, exc_info=True)
|
|
return jsonify({'error': f'Import failed: {e}'}), 500
|
|
|
|
return jsonify({'imported': imported, 'skipped': skipped})
|