import csv import hashlib import io from datetime import date as date_type, timedelta from decimal import Decimal, InvalidOperation from flask import make_response, render_template, redirect, url_for, flash, request from app.blueprints.transactions import transactions_bp from app.blueprints.transactions.forms import TransactionForm from app.extensions import db from app.models.account import Account from app.models.category import Category from app.models.transaction import Transaction from app.services.duplicate_detector import flag_duplicates from app.utils.validators import validate_amount def _compute_dedup_hash(merchant: str, amount: Decimal, date_str: str) -> str: normalized_amt = str(amount.quantize(Decimal('0.01'))) raw = merchant + normalized_amt + date_str return hashlib.sha256(raw.encode()).hexdigest() def _load_form_choices(form): form.category_id.choices = ( [(0, '— Select Category —')] + [(c.id, c.name) for c in Category.query.filter_by(is_active=True) .order_by(Category.is_system.desc(), Category.name).all()] ) form.account_id.choices = [ (a.id, a.name) for a in Account.query.filter_by(is_active=True).order_by(Account.name).all() ] PER_PAGE = 50 def _parse_filter_args(args): """Parse and return filter/sort parameters from a request args dict.""" return { 'sort': args.get('sort', 'date'), 'direction': args.get('dir', 'desc'), 'q': (args.get('q') or '').strip(), 'category_id': args.get('category_id', type=int), 'account_id': args.get('account_id', type=int), 'date_from': args.get('date_from', '').strip(), 'date_to': args.get('date_to', '').strip(), 'amount_min': args.get('amount_min', '').strip(), 'amount_max': args.get('amount_max', '').strip(), } def _build_transaction_query(params): """Build a filtered, sorted SQLAlchemy query from parsed filter params.""" q = params['q'] category_id = params['category_id'] account_id = params['account_id'] date_from = params['date_from'] date_to = params['date_to'] amount_min = params['amount_min'] amount_max = params['amount_max'] sort = params['sort'] direction = params['direction'] query = Transaction.query if q: like = f'%{q}%' query = query.filter( db.or_( Transaction.merchant_normalized.ilike(like), Transaction.notes.ilike(like), ) ) if category_id: query = query.filter(Transaction.category_id == category_id) if account_id: query = query.filter(Transaction.account_id == account_id) if date_from: query = query.filter(Transaction.date >= date_from) if date_to: query = query.filter(Transaction.date <= date_to) if amount_min: try: query = query.filter(Transaction.amount >= Decimal(amount_min)) except InvalidOperation: pass if amount_max: try: query = query.filter(Transaction.amount <= Decimal(amount_max)) except InvalidOperation: pass sort_col = { 'date': Transaction.date, 'amount': Transaction.amount, 'merchant': Transaction.merchant_normalized, }.get(sort, Transaction.date) order = sort_col.asc() if direction == 'asc' else sort_col.desc() return query.order_by(order) @transactions_bp.route('/') def index(): page = max(1, request.args.get('page', 1, type=int)) params = _parse_filter_args(request.args) query = _build_transaction_query(params) total = query.count() transactions = query.offset((page - 1) * PER_PAGE).limit(PER_PAGE).all() filter_params = {k: v for k, v in request.args.items() if v and k != 'page'} def url_for_page(p): return url_for('transactions.index', **filter_params, page=p) def sort_url(col): new_dir = 'asc' if (params['sort'] == col and params['direction'] == 'desc') else 'desc' return url_for('transactions.index', **{**filter_params, 'sort': col, 'dir': new_dir}) no_data = (total == 0 and not any([ params['q'], params['category_id'], params['account_id'], params['date_from'], params['date_to'], params['amount_min'], params['amount_max'], ])) no_match = (total == 0 and not no_data) categories = Category.query.filter_by(is_active=True).order_by(Category.name).all() accounts = Account.query.filter_by(is_active=True).order_by(Account.name).all() csv_url = url_for('transactions.export_csv', **filter_params) return render_template( 'transactions/index.html', active_page='transactions', transactions=transactions, total=total, page=page, per_page=PER_PAGE, sort=params['sort'], direction=params['direction'], url_for_page=url_for_page, sort_url=sort_url, no_data=no_data, no_match=no_match, categories=categories, accounts=accounts, csv_url=csv_url, q=params['q'], category_id=params['category_id'], account_id=params['account_id'], date_from=params['date_from'], date_to=params['date_to'], amount_min=params['amount_min'], amount_max=params['amount_max'], ) @transactions_bp.route('/export.csv') def export_csv(): params = _parse_filter_args(request.args) transactions = _build_transaction_query(params).all() si = io.StringIO() writer = csv.writer(si) writer.writerow(['Date', 'Merchant', 'Amount', 'Category', 'Account', 'Notes']) for txn in transactions: writer.writerow([ txn.date, txn.merchant_normalized, str(txn.amount), txn.category.name if txn.category else '', txn.account.name if txn.account else '', txn.notes or '', ]) output = make_response(si.getvalue()) output.headers['Content-Disposition'] = 'attachment; filename="transactions.csv"' output.headers['Content-Type'] = 'text/csv' return output @transactions_bp.route('/create', methods=['GET', 'POST']) def create(): form = TransactionForm() _load_form_choices(form) # Auto-select single account if request.method == 'GET' and len(form.account_id.choices) == 1: form.account_id.data = form.account_id.choices[0][0] duplicate_detected = False if form.validate_on_submit(): # Validate amount with domain validator try: amount = validate_amount(form.amount.data) except ValueError as e: form.amount.errors.append(str(e)) return render_template('transactions/create.html', form=form, duplicate_detected=False, active_page='transactions') merchant = form.merchant.data.strip() date_str = form.date.data.isoformat() new_hash = _compute_dedup_hash(merchant, amount, date_str) force_save = request.form.get('force_save', '0') == '1' if not force_save: window_start = (form.date.data - timedelta(days=3)).isoformat() window_end = (form.date.data + timedelta(days=3)).isoformat() recent_hashes = [ t.dedup_hash for t in Transaction.query .filter(Transaction.date >= window_start, Transaction.date <= window_end, Transaction.dedup_hash.isnot(None)) .all() ] if flag_duplicates([new_hash], recent_hashes)[0]: flash('Possible duplicate transaction detected.', 'warning') return render_template('transactions/create.html', form=form, duplicate_detected=True, active_page='transactions') cat_id = form.category_id.data if form.category_id.data != 0 else None txn = Transaction( date=date_str, merchant_normalized=merchant, merchant_raw=merchant, amount=amount, category_id=cat_id, account_id=form.account_id.data, notes=form.notes.data or None, is_manual=True, dedup_hash=new_hash, ) db.session.add(txn) db.session.commit() flash('Transaction added.', 'success') return redirect(url_for('transactions.index')) return render_template('transactions/create.html', form=form, duplicate_detected=False, active_page='transactions') @transactions_bp.route('//edit', methods=['GET', 'POST']) def edit(txn_id): txn = Transaction.query.get_or_404(txn_id) form = TransactionForm() _load_form_choices(form) if request.method == 'GET': form.date.data = date_type.fromisoformat(txn.date) form.merchant.data = txn.merchant_normalized form.amount.data = str(txn.amount) form.category_id.data = txn.category_id or 0 form.account_id.data = txn.account_id form.notes.data = txn.notes or '' return render_template('transactions/edit.html', form=form, txn=txn, duplicate_detected=False, active_page='transactions') if form.validate_on_submit(): try: amount = validate_amount(form.amount.data) except ValueError as e: form.amount.errors.append(str(e)) return render_template('transactions/edit.html', form=form, txn=txn, duplicate_detected=False, active_page='transactions') merchant = form.merchant.data.strip() date_str = form.date.data.isoformat() new_hash = _compute_dedup_hash(merchant, amount, date_str) force_save = request.form.get('force_save', '0') == '1' if not force_save: window_start = (form.date.data - timedelta(days=3)).isoformat() window_end = (form.date.data + timedelta(days=3)).isoformat() recent_hashes = [ t.dedup_hash for t in Transaction.query .filter(Transaction.id != txn.id, Transaction.date >= window_start, Transaction.date <= window_end, Transaction.dedup_hash.isnot(None)) .all() ] if flag_duplicates([new_hash], recent_hashes)[0]: flash('Possible duplicate transaction detected.', 'warning') return render_template('transactions/edit.html', form=form, txn=txn, duplicate_detected=True, active_page='transactions') txn.date = date_str txn.merchant_normalized = merchant txn.merchant_raw = merchant txn.amount = amount txn.category_id = form.category_id.data if form.category_id.data != 0 else None txn.account_id = form.account_id.data txn.notes = form.notes.data or None txn.dedup_hash = new_hash db.session.commit() flash('Transaction updated.', 'success') return redirect(url_for('transactions.index')) return render_template('transactions/edit.html', form=form, txn=txn, duplicate_detected=False, active_page='transactions') @transactions_bp.route('//delete', methods=['POST']) def delete(txn_id): txn = Transaction.query.get_or_404(txn_id) db.session.delete(txn) db.session.commit() flash('Transaction deleted.', 'success') return redirect(url_for('transactions.index'))