368 lines
14 KiB
Python
368 lines
14 KiB
Python
from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify
|
|
from flask_login import login_required
|
|
from flask_wtf import FlaskForm
|
|
from wtforms import StringField, SelectField, TextAreaField, SubmitField, DecimalField, DateField, HiddenField
|
|
from wtforms.validators import DataRequired, Optional, NumberRange
|
|
from app.extensions import db
|
|
from app.models.transaction import Transaction
|
|
from app.models.account import Account
|
|
from app.models.category import Category
|
|
from app.services.account_service import calc_balance
|
|
from datetime import date, datetime
|
|
import os
|
|
from sqlalchemy import or_
|
|
|
|
transactions_bp = Blueprint('transactions', __name__, url_prefix='/transactions')
|
|
|
|
|
|
def _account_choices():
|
|
return [(str(a.id), a.name)
|
|
for a in Account.query.filter_by(is_active=True).order_by(Account.name).all()]
|
|
|
|
|
|
def _category_choices(cat_type):
|
|
cats = Category.query.filter(
|
|
Category.category_type.in_([cat_type, 'both']),
|
|
Category.is_active == True,
|
|
Category.parent_id == None
|
|
).order_by(Category.name).all()
|
|
return [('', '— None —')] + [(str(c.id), c.name) for c in cats]
|
|
|
|
|
|
def _category_choices_json(cat_type):
|
|
cats = Category.query.filter(
|
|
Category.category_type.in_([cat_type, 'both']),
|
|
Category.is_active == True,
|
|
Category.parent_id == None
|
|
).order_by(Category.name).all()
|
|
return [{'id': str(c.id), 'name': c.name} for c in cats]
|
|
|
|
|
|
class TransactionForm(FlaskForm):
|
|
transaction_type = HiddenField(default='expense')
|
|
account_id = SelectField('Account', validators=[DataRequired()])
|
|
category_id = SelectField('Category', validators=[Optional()])
|
|
amount = DecimalField('Amount', validators=[DataRequired(), NumberRange(min=0.01)],
|
|
places=2)
|
|
description = StringField('Description', validators=[DataRequired()])
|
|
date = DateField('Date', validators=[DataRequired()], default=date.today)
|
|
notes = TextAreaField('Notes', validators=[Optional()])
|
|
submit = SubmitField('Save')
|
|
|
|
|
|
class TransferForm(FlaskForm):
|
|
from_account_id = SelectField('From Account', validators=[DataRequired()])
|
|
to_account_id = SelectField('To Account', validators=[DataRequired()])
|
|
amount = DecimalField('Amount', validators=[DataRequired(), NumberRange(min=0.01)],
|
|
places=2)
|
|
description = StringField('Description', default='Transfer')
|
|
date = DateField('Date', validators=[DataRequired()], default=date.today)
|
|
notes = TextAreaField('Notes', validators=[Optional()])
|
|
submit = SubmitField('Transfer')
|
|
|
|
|
|
@transactions_bp.route('/')
|
|
@login_required
|
|
def index():
|
|
tab = request.args.get('tab', 'expense') # 'income' | 'expense'
|
|
page = request.args.get('page', 1, type=int)
|
|
search = request.args.get('q', '').strip()
|
|
category_id = request.args.get('category_id', '', type=str)
|
|
account_id = request.args.get('account_id', '', type=str)
|
|
date_from = request.args.get('date_from', '')
|
|
date_to = request.args.get('date_to', '')
|
|
|
|
query = Transaction.query.filter(
|
|
Transaction.transaction_type == tab
|
|
).order_by(Transaction.date.desc(), Transaction.id.desc())
|
|
|
|
if search:
|
|
query = query.filter(Transaction.description.ilike(f'%{search}%'))
|
|
try:
|
|
if category_id:
|
|
query = query.filter(Transaction.category_id == int(category_id))
|
|
if account_id:
|
|
query = query.filter(Transaction.account_id == int(account_id))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
if date_from:
|
|
try:
|
|
query = query.filter(Transaction.date >= datetime.strptime(date_from, '%Y-%m-%d').date())
|
|
except ValueError:
|
|
pass
|
|
if date_to:
|
|
try:
|
|
query = query.filter(Transaction.date <= datetime.strptime(date_to, '%Y-%m-%d').date())
|
|
except ValueError:
|
|
pass
|
|
|
|
pagination = query.paginate(page=page, per_page=30, error_out=False)
|
|
|
|
# Counts for tabs
|
|
income_count = Transaction.query.filter_by(transaction_type='income').count()
|
|
expense_count = Transaction.query.filter_by(transaction_type='expense').count()
|
|
|
|
accounts = Account.query.filter_by(is_active=True).order_by(Account.name).all()
|
|
categories = Category.query.filter(
|
|
Category.category_type.in_([tab, 'both']),
|
|
Category.is_active == True
|
|
).order_by(Category.name).all()
|
|
|
|
return render_template('transactions/index.html',
|
|
pagination=pagination,
|
|
transactions=pagination.items,
|
|
tab=tab,
|
|
income_count=income_count,
|
|
expense_count=expense_count,
|
|
accounts=accounts,
|
|
categories=categories,
|
|
search=search,
|
|
category_id=category_id,
|
|
account_id=account_id,
|
|
date_from=date_from,
|
|
date_to=date_to)
|
|
|
|
|
|
@transactions_bp.route('/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
def new():
|
|
txn_type = request.args.get('type', 'expense')
|
|
if txn_type not in ('income', 'expense'):
|
|
txn_type = 'expense'
|
|
|
|
form = TransactionForm()
|
|
# On POST, honour the type the user selected in the toggle
|
|
if request.method == 'POST':
|
|
submitted = request.form.get('transaction_type', txn_type)
|
|
if submitted in ('income', 'expense'):
|
|
txn_type = submitted
|
|
form.account_id.choices = _account_choices()
|
|
form.category_id.choices = _category_choices(txn_type)
|
|
|
|
if form.validate_on_submit():
|
|
saved_type = form.transaction_type.data
|
|
txn = Transaction(
|
|
transaction_type=saved_type,
|
|
account_id=int(form.account_id.data),
|
|
category_id=int(form.category_id.data) if form.category_id.data else None,
|
|
amount=form.amount.data,
|
|
description=form.description.data.strip(),
|
|
date=form.date.data,
|
|
notes=form.notes.data,
|
|
)
|
|
db.session.add(txn)
|
|
db.session.commit()
|
|
calc_balance(txn.account_id)
|
|
flash(f'{"Income" if saved_type == "income" else "Expense"} added.', 'success')
|
|
if saved_type == 'expense':
|
|
from app.services.alert_service import check_and_flash_budget_alerts
|
|
check_and_flash_budget_alerts(flash)
|
|
return redirect(url_for('transactions.index', tab=saved_type))
|
|
|
|
return render_template('transactions/form.html',
|
|
form=form,
|
|
txn_type=txn_type,
|
|
income_cats=_category_choices_json('income'),
|
|
expense_cats=_category_choices_json('expense'),
|
|
title=f'New {"Income" if txn_type == "income" else "Expense"}')
|
|
|
|
|
|
@transactions_bp.route('/<int:id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def edit(id):
|
|
txn = db.get_or_404(Transaction, id)
|
|
|
|
# Determine active type: from POST toggle or existing record
|
|
if request.method == 'POST':
|
|
submitted_type = request.form.get('transaction_type', txn.transaction_type)
|
|
active_type = submitted_type if submitted_type in ('income', 'expense') else txn.transaction_type
|
|
else:
|
|
active_type = txn.transaction_type
|
|
|
|
form = TransactionForm(obj=txn)
|
|
form.account_id.choices = _account_choices()
|
|
form.category_id.choices = _category_choices(active_type)
|
|
|
|
if request.method == 'GET':
|
|
form.transaction_type.data = txn.transaction_type
|
|
form.account_id.data = str(txn.account_id)
|
|
form.category_id.data = str(txn.category_id) if txn.category_id else ''
|
|
|
|
if form.validate_on_submit():
|
|
old_account_id = txn.account_id
|
|
txn.transaction_type = active_type
|
|
txn.account_id = int(form.account_id.data)
|
|
txn.category_id = int(form.category_id.data) if form.category_id.data else None
|
|
txn.amount = form.amount.data
|
|
txn.description = form.description.data.strip()
|
|
txn.date = form.date.data
|
|
txn.notes = form.notes.data
|
|
db.session.commit()
|
|
calc_balance(old_account_id)
|
|
calc_balance(txn.account_id)
|
|
flash('Transaction updated.', 'success')
|
|
if txn.transaction_type == 'expense':
|
|
from app.services.alert_service import check_and_flash_budget_alerts
|
|
check_and_flash_budget_alerts(flash)
|
|
return redirect(url_for('transactions.index', tab=txn.transaction_type))
|
|
|
|
return render_template('transactions/form.html',
|
|
form=form,
|
|
txn=txn,
|
|
txn_type=active_type,
|
|
income_cats=_category_choices_json('income'),
|
|
expense_cats=_category_choices_json('expense'),
|
|
title='Edit Transaction')
|
|
|
|
|
|
@transactions_bp.route('/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def delete(id):
|
|
txn = db.get_or_404(Transaction, id)
|
|
account_id = txn.account_id
|
|
txn_type = txn.transaction_type
|
|
db.session.delete(txn)
|
|
db.session.commit()
|
|
calc_balance(account_id)
|
|
flash('Transaction deleted.', 'info')
|
|
return redirect(url_for('transactions.index', tab=txn_type))
|
|
|
|
|
|
@transactions_bp.route('/transfer', methods=['GET', 'POST'])
|
|
@login_required
|
|
def transfer():
|
|
form = TransferForm()
|
|
form.from_account_id.choices = _account_choices()
|
|
form.to_account_id.choices = _account_choices()
|
|
|
|
if form.validate_on_submit():
|
|
if form.from_account_id.data == form.to_account_id.data:
|
|
flash('Source and destination accounts must be different.', 'warning')
|
|
else:
|
|
txn = Transaction(
|
|
transaction_type='transfer',
|
|
account_id=int(form.from_account_id.data),
|
|
to_account_id=int(form.to_account_id.data),
|
|
amount=form.amount.data,
|
|
description=form.description.data.strip() or 'Transfer',
|
|
date=form.date.data,
|
|
notes=form.notes.data,
|
|
)
|
|
db.session.add(txn)
|
|
db.session.commit()
|
|
calc_balance(txn.account_id)
|
|
calc_balance(txn.to_account_id)
|
|
flash('Transfer recorded.', 'success')
|
|
return redirect(url_for('transactions.index'))
|
|
|
|
return render_template('transactions/transfer.html', form=form)
|
|
|
|
|
|
@transactions_bp.route('/ocr', methods=['POST'])
|
|
@login_required
|
|
def ocr_receipt():
|
|
"""
|
|
POST a receipt image, get back extracted transaction data as JSON.
|
|
Used by both new expense form and edit form.
|
|
"""
|
|
from app.services.ocr_service import extract_from_bytes
|
|
from app.models.category import Category
|
|
|
|
if 'receipt' not in request.files:
|
|
return jsonify({'error': 'No file uploaded'}), 400
|
|
|
|
f = request.files['receipt']
|
|
if not f.filename:
|
|
return jsonify({'error': 'Empty filename'}), 400
|
|
|
|
ext = os.path.splitext(f.filename)[1].lower()
|
|
allowed = {'.jpg', '.jpeg', '.png', '.gif', '.webp'}
|
|
if ext not in allowed:
|
|
return jsonify({'error': f'Unsupported type: {ext}. Use JPG, PNG, GIF, WEBP'}), 400
|
|
|
|
# Read bytes — limit 10MB
|
|
f.seek(0, 2)
|
|
size = f.tell()
|
|
f.seek(0)
|
|
if size > 10 * 1024 * 1024:
|
|
return jsonify({'error': 'File too large (max 10MB)'}), 400
|
|
|
|
# Validate actual file content via magic bytes
|
|
from app.routes.settings import _check_magic
|
|
header = f.read(12)
|
|
f.seek(0)
|
|
magic_ext, magic_mime = _check_magic(header)
|
|
if magic_ext is None or magic_ext not in ('jpg', 'jpeg', 'png', 'gif', 'webp'):
|
|
return jsonify({'error': 'File content does not match an allowed image type'}), 400
|
|
|
|
mime_type = magic_mime
|
|
image_bytes = f.read()
|
|
|
|
result = extract_from_bytes(image_bytes, mime_type)
|
|
|
|
if result['error']:
|
|
return jsonify({'error': result['error']}), 422
|
|
|
|
# Look up category ID from suggestion
|
|
category_id = None
|
|
if result['category_suggestion']:
|
|
cat = Category.query.filter(
|
|
Category.name.ilike(result['category_suggestion']),
|
|
Category.is_active == True,
|
|
).first()
|
|
if cat:
|
|
category_id = cat.id
|
|
|
|
return jsonify({
|
|
'amount': result['amount'],
|
|
'date': result['date'],
|
|
'description': result['merchant'] or result['notes'] or '',
|
|
'notes': result['notes'],
|
|
'category_suggestion': result['category_suggestion'],
|
|
'category_id': category_id,
|
|
})
|
|
|
|
|
|
@transactions_bp.route('/ocr-file', methods=['POST'])
|
|
@login_required
|
|
def ocr_receipt_file():
|
|
"""
|
|
Re-extract from an already-uploaded receipt file stored on disk.
|
|
Body: { "filename": "abc123.jpg" }
|
|
"""
|
|
from app.services.ocr_service import extract_from_file
|
|
from app.models.category import Category
|
|
from flask import current_app
|
|
|
|
data = request.get_json()
|
|
if not data or not data.get('filename'):
|
|
return jsonify({'error': 'No filename provided'}), 400
|
|
|
|
# Security: only allow basenames, no path traversal
|
|
filename = os.path.basename(data['filename'])
|
|
upload_dir = current_app.config.get('UPLOAD_FOLDER', '/home/pfm/app/uploads')
|
|
file_path = os.path.join(upload_dir, filename)
|
|
|
|
result = extract_from_file(file_path)
|
|
|
|
if result['error']:
|
|
return jsonify({'error': result['error']}), 422
|
|
|
|
category_id = None
|
|
if result['category_suggestion']:
|
|
cat = Category.query.filter(
|
|
Category.name.ilike(result['category_suggestion']),
|
|
Category.is_active == True,
|
|
).first()
|
|
if cat:
|
|
category_id = cat.id
|
|
|
|
return jsonify({
|
|
'amount': result['amount'],
|
|
'date': result['date'],
|
|
'description': result['merchant'] or result['notes'] or '',
|
|
'notes': result['notes'],
|
|
'category_suggestion': result['category_suggestion'],
|
|
'category_id': category_id,
|
|
})
|