435 lines
16 KiB
Python
435 lines
16 KiB
Python
import os
|
|
import uuid
|
|
from flask import (Blueprint, render_template, redirect, url_for, flash,
|
|
request, current_app, send_from_directory)
|
|
from flask_login import login_required, current_user
|
|
from flask_wtf import FlaskForm
|
|
from flask_wtf.file import FileField, FileAllowed
|
|
from wtforms import (StringField, SelectField, PasswordField, SubmitField,
|
|
DecimalField, DateField, BooleanField)
|
|
from wtforms.validators import DataRequired, Optional, Length, EqualTo, NumberRange
|
|
from app.extensions import db
|
|
from app.models.recurring_rule import RecurringRule
|
|
from app.models.account import Account
|
|
from app.models.category import Category
|
|
from app.models.receipt import Receipt
|
|
from app.models.transaction import Transaction
|
|
from app.services.recurring_service import get_upcoming, process_due_rules
|
|
from app.services.import_service import parse_csv, import_rows
|
|
from datetime import date
|
|
from werkzeug.utils import secure_filename
|
|
|
|
settings_bp = Blueprint('settings', __name__, url_prefix='/settings')
|
|
|
|
GROQ_MODELS = [
|
|
('llama-3.3-70b-versatile', 'Llama 3.3 70B — Best quality'),
|
|
('llama-3.1-8b-instant', 'Llama 3.1 8B — Fastest'),
|
|
('mixtral-8x7b-32768', 'Mixtral 8x7B — Balanced'),
|
|
]
|
|
|
|
CURRENCIES = [
|
|
('USD', 'USD — US Dollar ($)'),
|
|
('VND', 'VND — Vietnamese Dong (₫)'),
|
|
('EUR', 'EUR — Euro (€)'),
|
|
('GBP', 'GBP — British Pound (£)'),
|
|
('JPY', 'JPY — Japanese Yen (¥)'),
|
|
('AUD', 'AUD — Australian Dollar (A$)'),
|
|
('CAD', 'CAD — Canadian Dollar (C$)'),
|
|
('SGD', 'SGD — Singapore Dollar (S$)'),
|
|
]
|
|
|
|
CURRENCY_SYMBOLS = {
|
|
'USD': '$', 'VND': '₫', 'EUR': '€', 'GBP': '£',
|
|
'JPY': '¥', 'AUD': 'A$', 'CAD': 'C$', 'SGD': 'S$',
|
|
}
|
|
|
|
FREQ_CHOICES = [
|
|
('daily', 'Daily'),
|
|
('weekly', 'Weekly'),
|
|
('biweekly', 'Bi-weekly'),
|
|
('monthly', 'Monthly'),
|
|
('quarterly', 'Quarterly'),
|
|
('yearly', 'Yearly'),
|
|
]
|
|
|
|
|
|
class ProfileForm(FlaskForm):
|
|
display_name = StringField('Display Name', validators=[Optional(), Length(max=100)])
|
|
email = StringField('Email', validators=[Optional(), Length(max=120)])
|
|
timezone = SelectField('Timezone', choices=[
|
|
('Asia/Ho_Chi_Minh', 'Asia/Ho_Chi_Minh (Vietnam)'),
|
|
('America/New_York', 'America/New_York (EST)'),
|
|
('America/Los_Angeles', 'America/Los_Angeles (PST)'),
|
|
('Europe/London', 'Europe/London (GMT)'),
|
|
('Asia/Tokyo', 'Asia/Tokyo (JST)'),
|
|
('Asia/Singapore', 'Asia/Singapore (SGT)'),
|
|
('UTC', 'UTC'),
|
|
])
|
|
currency = SelectField('Currency', choices=CURRENCIES)
|
|
groq_model = SelectField('AI Model', choices=GROQ_MODELS)
|
|
submit = SubmitField('Save Profile')
|
|
|
|
|
|
class PasswordForm(FlaskForm):
|
|
current_password = PasswordField('Current Password', validators=[DataRequired()])
|
|
new_password = PasswordField('New Password', validators=[DataRequired(), Length(min=6)])
|
|
confirm_password = PasswordField('Confirm Password',
|
|
validators=[DataRequired(), EqualTo('new_password')])
|
|
submit = SubmitField('Change Password')
|
|
|
|
|
|
class RecurringRuleForm(FlaskForm):
|
|
name = StringField('Name', validators=[DataRequired(), Length(1, 100)])
|
|
transaction_type = SelectField('Type', choices=[('income', 'Income'), ('expense', '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()])
|
|
frequency = SelectField('Frequency', choices=FREQ_CHOICES)
|
|
start_date = DateField('Start Date', validators=[DataRequired()], default=date.today)
|
|
end_date = DateField('End Date', validators=[Optional()])
|
|
submit = SubmitField('Save Rule')
|
|
|
|
|
|
class ImportForm(FlaskForm):
|
|
csv_file = FileField('CSV File', validators=[
|
|
DataRequired(),
|
|
FileAllowed(['csv'], 'CSV files only'),
|
|
])
|
|
default_account_id = SelectField('Default Account (if not in CSV)', validators=[Optional()])
|
|
skip_duplicates = BooleanField('Skip duplicate transactions', default=True)
|
|
submit = SubmitField('Preview Import')
|
|
|
|
|
|
def _account_choices():
|
|
return [('', '— None —')] + [
|
|
(str(a.id), a.name)
|
|
for a in Account.query.filter_by(is_active=True).order_by(Account.name).all()
|
|
]
|
|
|
|
|
|
def _category_choices(txn_type='expense'):
|
|
cats = Category.query.filter(
|
|
Category.category_type.in_([txn_type, 'both']),
|
|
Category.is_active == True,
|
|
).order_by(Category.name).all()
|
|
return [('', '— None —')] + [(str(c.id), c.name) for c in cats]
|
|
|
|
|
|
@settings_bp.route('/')
|
|
@login_required
|
|
def index():
|
|
upcoming = get_upcoming(days=30)
|
|
rules = RecurringRule.query.filter_by(is_active=True).order_by(RecurringRule.name).all()
|
|
return render_template('settings/index.html',
|
|
upcoming=upcoming,
|
|
rules=rules)
|
|
|
|
|
|
@settings_bp.route('/profile', methods=['GET', 'POST'])
|
|
@login_required
|
|
def profile():
|
|
form = ProfileForm(obj=current_user)
|
|
if form.validate_on_submit():
|
|
current_user.display_name = form.display_name.data
|
|
current_user.email = form.email.data
|
|
current_user.timezone = form.timezone.data
|
|
current_user.currency = form.currency.data
|
|
current_user.currency_symbol = CURRENCY_SYMBOLS.get(form.currency.data, '$')
|
|
current_user.groq_model = form.groq_model.data
|
|
db.session.commit()
|
|
flash('Profile updated.', 'success')
|
|
return redirect(url_for('settings.profile'))
|
|
return render_template('settings/profile.html', form=form)
|
|
|
|
|
|
@settings_bp.route('/password', methods=['GET', 'POST'])
|
|
@login_required
|
|
def password():
|
|
form = PasswordForm()
|
|
if form.validate_on_submit():
|
|
if not current_user.check_password(form.current_password.data):
|
|
flash('Current password is incorrect.', 'danger')
|
|
else:
|
|
current_user.set_password(form.new_password.data)
|
|
db.session.commit()
|
|
flash('Password changed successfully.', 'success')
|
|
return redirect(url_for('settings.profile'))
|
|
return render_template('settings/password.html', form=form)
|
|
|
|
|
|
# ── Recurring rules ───────────────────────────────────────────────────────────
|
|
|
|
@settings_bp.route('/recurring')
|
|
@login_required
|
|
def recurring():
|
|
rules = RecurringRule.query.order_by(RecurringRule.is_active.desc(),
|
|
RecurringRule.name).all()
|
|
upcoming = get_upcoming(30)
|
|
return render_template('settings/recurring.html', rules=rules, upcoming=upcoming)
|
|
|
|
|
|
@settings_bp.route('/recurring/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
def recurring_new():
|
|
form = RecurringRuleForm()
|
|
form.account_id.choices = [(str(a.id), a.name)
|
|
for a in Account.query.filter_by(is_active=True).all()]
|
|
form.category_id.choices = _category_choices(form.transaction_type.data or 'expense')
|
|
|
|
if form.validate_on_submit():
|
|
from app.services.recurring_service import next_occurrence
|
|
start = form.start_date.data
|
|
rule = RecurringRule(
|
|
name=form.name.data.strip(),
|
|
transaction_type=form.transaction_type.data,
|
|
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(),
|
|
frequency=form.frequency.data,
|
|
start_date=start,
|
|
end_date=form.end_date.data,
|
|
next_run=start,
|
|
)
|
|
db.session.add(rule)
|
|
db.session.commit()
|
|
flash(f'Recurring rule "{rule.name}" created.', 'success')
|
|
return redirect(url_for('settings.recurring'))
|
|
|
|
return render_template('settings/recurring_form.html', form=form, title='New Recurring Rule')
|
|
|
|
|
|
@settings_bp.route('/recurring/<int:id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def recurring_edit(id):
|
|
rule = db.get_or_404(RecurringRule, id)
|
|
form = RecurringRuleForm(obj=rule)
|
|
form.account_id.choices = [(str(a.id), a.name)
|
|
for a in Account.query.filter_by(is_active=True).all()]
|
|
form.category_id.choices = _category_choices(rule.transaction_type)
|
|
|
|
if request.method == 'GET':
|
|
form.account_id.data = str(rule.account_id)
|
|
form.category_id.data = str(rule.category_id) if rule.category_id else ''
|
|
|
|
if form.validate_on_submit():
|
|
rule.name = form.name.data.strip()
|
|
rule.transaction_type = form.transaction_type.data
|
|
rule.account_id = int(form.account_id.data)
|
|
rule.category_id = int(form.category_id.data) if form.category_id.data else None
|
|
rule.amount = form.amount.data
|
|
rule.description = form.description.data.strip()
|
|
rule.frequency = form.frequency.data
|
|
rule.start_date = form.start_date.data
|
|
rule.end_date = form.end_date.data
|
|
db.session.commit()
|
|
flash('Rule updated.', 'success')
|
|
return redirect(url_for('settings.recurring'))
|
|
|
|
return render_template('settings/recurring_form.html', form=form,
|
|
title='Edit Recurring Rule', rule=rule)
|
|
|
|
|
|
@settings_bp.route('/recurring/<int:id>/toggle', methods=['POST'])
|
|
@login_required
|
|
def recurring_toggle(id):
|
|
rule = db.get_or_404(RecurringRule, id)
|
|
rule.is_active = not rule.is_active
|
|
db.session.commit()
|
|
flash(f'Rule {"enabled" if rule.is_active else "disabled"}.', 'info')
|
|
return redirect(url_for('settings.recurring'))
|
|
|
|
|
|
@settings_bp.route('/recurring/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def recurring_delete(id):
|
|
rule = db.get_or_404(RecurringRule, id)
|
|
db.session.delete(rule)
|
|
db.session.commit()
|
|
flash('Rule deleted.', 'info')
|
|
return redirect(url_for('settings.recurring'))
|
|
|
|
|
|
@settings_bp.route('/recurring/run', methods=['POST'])
|
|
@login_required
|
|
def recurring_run():
|
|
created = process_due_rules()
|
|
if created:
|
|
flash(f'Processed {len(created)} recurring transaction(s).', 'success')
|
|
else:
|
|
flash('No recurring transactions due.', 'info')
|
|
return redirect(url_for('settings.recurring'))
|
|
|
|
|
|
# ── CSV Import ────────────────────────────────────────────────────────────────
|
|
|
|
@settings_bp.route('/import', methods=['GET', 'POST'])
|
|
@login_required
|
|
def import_csv():
|
|
form = ImportForm()
|
|
form.default_account_id.choices = _account_choices()
|
|
preview = None
|
|
parse_errors = []
|
|
|
|
if form.validate_on_submit():
|
|
f = form.csv_file.data
|
|
content = f.read()
|
|
default_acc = int(form.default_account_id.data) if form.default_account_id.data else None
|
|
rows, parse_errors = parse_csv(content, default_account_id=default_acc)
|
|
if rows and not parse_errors:
|
|
# Store in session for confirm step
|
|
import json
|
|
from flask import session
|
|
session['import_rows'] = [
|
|
{**r, 'date': r['date'].isoformat()}
|
|
for r in rows
|
|
]
|
|
session['import_skip_dupes'] = form.skip_duplicates.data
|
|
preview = rows
|
|
|
|
return render_template('settings/import.html',
|
|
form=form,
|
|
preview=preview,
|
|
parse_errors=parse_errors)
|
|
|
|
|
|
@settings_bp.route('/import/confirm', methods=['POST'])
|
|
@login_required
|
|
def import_confirm():
|
|
from flask import session
|
|
from datetime import date as date_cls
|
|
import json
|
|
|
|
raw_rows = session.pop('import_rows', [])
|
|
skip = session.pop('import_skip_dupes', True)
|
|
|
|
if not raw_rows:
|
|
flash('No import data found. Please upload again.', 'warning')
|
|
return redirect(url_for('settings.import_csv'))
|
|
|
|
rows = []
|
|
for r in raw_rows:
|
|
r['date'] = date_cls.fromisoformat(r['date'])
|
|
rows.append(r)
|
|
|
|
imported, skipped = import_rows(rows, skip_duplicates=skip)
|
|
flash(f'Imported {imported} transaction(s). Skipped {skipped} duplicate(s).', 'success')
|
|
return redirect(url_for('transactions.index'))
|
|
|
|
|
|
# ── Receipt upload ────────────────────────────────────────────────────────────
|
|
|
|
ALLOWED_RECEIPT_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf'}
|
|
|
|
# Magic bytes → (canonical_extension, mime_type)
|
|
_MAGIC = [
|
|
(b'\xff\xd8\xff', 'jpg', 'image/jpeg'),
|
|
(b'\x89PNG\r\n\x1a\n', 'png', 'image/png'),
|
|
(b'GIF87a', 'gif', 'image/gif'),
|
|
(b'GIF89a', 'gif', 'image/gif'),
|
|
(b'%PDF-', 'pdf', 'application/pdf'),
|
|
# WEBP: RIFF????WEBP (bytes 0-3 = RIFF, bytes 8-11 = WEBP)
|
|
]
|
|
|
|
|
|
def _allowed_receipt(filename):
|
|
return '.' in filename and \
|
|
filename.rsplit('.', 1)[1].lower() in ALLOWED_RECEIPT_EXTENSIONS
|
|
|
|
|
|
def _check_magic(header_bytes):
|
|
"""
|
|
Inspect the first 12 bytes of a file and return (ext, mime) if recognised,
|
|
or (None, None) if the bytes don't match any allowed type.
|
|
"""
|
|
for magic, ext, mime in _MAGIC:
|
|
if header_bytes[:len(magic)] == magic:
|
|
return ext, mime
|
|
# WEBP special case: RIFF at 0, WEBP at 8
|
|
if header_bytes[:4] == b'RIFF' and header_bytes[8:12] == b'WEBP':
|
|
return 'webp', 'image/webp'
|
|
return None, None
|
|
|
|
|
|
@settings_bp.route('/receipt/upload/<int:txn_id>', methods=['POST'])
|
|
@login_required
|
|
def upload_receipt(txn_id):
|
|
txn = db.get_or_404(Transaction, txn_id)
|
|
|
|
if 'receipt' not in request.files:
|
|
flash('No file selected.', 'warning')
|
|
return redirect(request.referrer or url_for('transactions.index'))
|
|
|
|
f = request.files['receipt']
|
|
if not f.filename or not _allowed_receipt(f.filename):
|
|
flash('Invalid file type. Allowed: PNG, JPG, GIF, PDF', 'warning')
|
|
return redirect(request.referrer or url_for('transactions.index'))
|
|
|
|
max_size = current_app.config.get('MAX_CONTENT_LENGTH', 10 * 1024 * 1024)
|
|
f.seek(0, 2)
|
|
file_size = f.tell()
|
|
f.seek(0)
|
|
if file_size > max_size:
|
|
flash('File too large. Maximum 10MB.', 'warning')
|
|
return redirect(request.referrer or url_for('transactions.index'))
|
|
|
|
# Validate actual file content via magic bytes (prevents renamed-file attacks)
|
|
header = f.read(12)
|
|
f.seek(0)
|
|
magic_ext, magic_mime = _check_magic(header)
|
|
if magic_ext is None:
|
|
flash('File content does not match an allowed type (PNG, JPG, GIF, PDF, WEBP).', 'warning')
|
|
return redirect(request.referrer or url_for('transactions.index'))
|
|
|
|
ext = magic_ext # use the extension we detected, not what the user claimed
|
|
unique_name = f'{uuid.uuid4().hex}.{ext}'
|
|
upload_dir = current_app.config.get('UPLOAD_FOLDER', '/home/pfm/app/uploads')
|
|
os.makedirs(upload_dir, exist_ok=True)
|
|
f.save(os.path.join(upload_dir, unique_name))
|
|
|
|
# Remove old receipt if exists
|
|
if txn.receipt:
|
|
old_path = os.path.join(upload_dir, txn.receipt.filename)
|
|
if os.path.exists(old_path):
|
|
os.remove(old_path)
|
|
db.session.delete(txn.receipt)
|
|
|
|
receipt = Receipt(
|
|
filename=unique_name,
|
|
original_filename=secure_filename(f.filename),
|
|
file_size=file_size,
|
|
mime_type=magic_mime, # use validated type, not the browser-supplied header
|
|
)
|
|
db.session.add(receipt)
|
|
db.session.flush()
|
|
txn.receipt_id = receipt.id
|
|
db.session.commit()
|
|
|
|
flash('Receipt uploaded.', 'success')
|
|
return redirect(request.referrer or url_for('transactions.index'))
|
|
|
|
|
|
@settings_bp.route('/receipt/<int:txn_id>/delete', methods=['POST'])
|
|
@login_required
|
|
def delete_receipt(txn_id):
|
|
txn = db.get_or_404(Transaction, txn_id)
|
|
if txn.receipt:
|
|
upload_dir = current_app.config.get('UPLOAD_FOLDER', '/home/pfm/app/uploads')
|
|
path = os.path.join(upload_dir, txn.receipt.filename)
|
|
if os.path.exists(path):
|
|
os.remove(path)
|
|
db.session.delete(txn.receipt)
|
|
txn.receipt_id = None
|
|
db.session.commit()
|
|
flash('Receipt deleted.', 'info')
|
|
return redirect(request.referrer or url_for('transactions.index'))
|
|
|
|
|
|
@settings_bp.route('/receipt/view/<filename>')
|
|
@login_required
|
|
def view_receipt(filename):
|
|
upload_dir = current_app.config.get('UPLOAD_FOLDER', '/home/pfm/app/uploads')
|
|
# Strip any path components to prevent directory traversal
|
|
return send_from_directory(upload_dir, os.path.basename(filename))
|