534 lines
22 KiB
Python
534 lines
22 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, DecimalField, DateField, SelectField, IntegerField,
|
|
TextAreaField, SubmitField)
|
|
from wtforms.validators import DataRequired, Optional, NumberRange, Length
|
|
from sqlalchemy.exc import IntegrityError
|
|
from datetime import date, timedelta
|
|
import logging
|
|
|
|
from app.extensions import db
|
|
from app.models.utility import (UtilityProvider, UtilityBill,
|
|
UTILITY_TYPE_META, UTILITY_TYPES)
|
|
from app.models.account import Account
|
|
from app.models.category import Category
|
|
from app.models.transaction import Transaction
|
|
from app.services.account_service import calc_balance
|
|
from app.services import utility_service
|
|
|
|
utilities_bp = Blueprint('utilities', __name__, url_prefix='/utilities')
|
|
|
|
log = logging.getLogger('app.utilities')
|
|
|
|
UTILITY_COLORS = ['#f59e0b', '#0ea5e9', '#ef4444', '#6366f1', '#8b5cf6', '#10b981', '#ec4899', '#64748b']
|
|
UTILITY_ICONS = [
|
|
'bi-lightning-charge', 'bi-droplet', 'bi-fire', 'bi-wifi', 'bi-phone',
|
|
'bi-trash3', 'bi-plug', 'bi-thermometer-half', 'bi-tv', 'bi-router',
|
|
]
|
|
|
|
|
|
def _account_choices():
|
|
accts = Account.query.filter_by(is_active=True).order_by(Account.name).all()
|
|
return [('', '— None —')] + [(str(a.id), a.name) for a in accts]
|
|
|
|
|
|
def _category_choices():
|
|
cats = (Category.query
|
|
.filter(Category.is_active == True,
|
|
Category.category_type.in_(['expense', 'both']))
|
|
.order_by(Category.name).all())
|
|
return [('', '— None —')] + [(str(c.id), c.name) for c in cats]
|
|
|
|
|
|
def _provider_choices(include_inactive=False):
|
|
q = UtilityProvider.query
|
|
if not include_inactive:
|
|
q = q.filter_by(is_active=True)
|
|
provs = q.order_by(UtilityProvider.name).all()
|
|
return [(str(p.id), f'{p.name} · {p.type_label}') for p in provs]
|
|
|
|
|
|
def _type_defaults():
|
|
"""Per-type unit/icon/color suggestions for the provider form's JS."""
|
|
return {k: {'unit': v[3], 'icon': v[1], 'color': v[2]}
|
|
for k, v in UTILITY_TYPE_META.items()}
|
|
|
|
|
|
def _provider_units():
|
|
"""provider id -> usage unit, so the bill form can label consumption fields."""
|
|
return {str(p.id): (p.usage_unit or '') for p in UtilityProvider.query.all()}
|
|
|
|
|
|
def _safe_int(value, default=None):
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
# ── forms ────────────────────────────────────────────────────────────────────
|
|
|
|
class ProviderForm(FlaskForm):
|
|
name = StringField('Provider Name', validators=[DataRequired(), Length(1, 100)])
|
|
utility_type = SelectField('Utility Type', validators=[DataRequired()],
|
|
choices=[(k, v[0]) for k, v in UTILITY_TYPE_META.items()])
|
|
account_number = StringField('Account Number', validators=[Optional(), Length(max=100)])
|
|
usage_unit = StringField('Usage Unit', validators=[Optional(), Length(max=20)])
|
|
default_account_id = SelectField('Pay From Account', validators=[Optional()])
|
|
category_id = SelectField('Expense Category', validators=[Optional()])
|
|
billing_day = IntegerField('Billing Day', validators=[Optional(), NumberRange(min=1, max=31)])
|
|
color = StringField('Color', default='#8b5cf6')
|
|
icon = StringField('Icon', default='bi-lightning-charge')
|
|
notes = TextAreaField('Notes', validators=[Optional()])
|
|
submit = SubmitField('Save Provider')
|
|
|
|
|
|
class BillForm(FlaskForm):
|
|
provider_id = SelectField('Provider', validators=[DataRequired()])
|
|
period_start = DateField('Period Start', validators=[DataRequired()])
|
|
period_end = DateField('Period End', validators=[DataRequired()])
|
|
amount = DecimalField('Amount', validators=[DataRequired(), NumberRange(min=0)], places=2)
|
|
due_date = DateField('Due Date', validators=[Optional()])
|
|
usage = DecimalField('Usage', validators=[Optional(), NumberRange(min=0)], places=3)
|
|
meter_start = DecimalField('Meter Start', validators=[Optional(), NumberRange(min=0)], places=3)
|
|
meter_end = DecimalField('Meter End', validators=[Optional(), NumberRange(min=0)], places=3)
|
|
notes = TextAreaField('Notes', validators=[Optional()])
|
|
submit = SubmitField('Save Bill')
|
|
|
|
def validate(self, extra_validators=None):
|
|
if not super().validate(extra_validators):
|
|
return False
|
|
ok = True
|
|
if self.period_end.data and self.period_start.data and \
|
|
self.period_end.data < self.period_start.data:
|
|
self.period_end.errors.append('Period end must be on or after period start.')
|
|
ok = False
|
|
if self.meter_start.data is not None and self.meter_end.data is not None and \
|
|
self.meter_end.data < self.meter_start.data:
|
|
self.meter_end.errors.append('Meter end reading is lower than the start reading.')
|
|
ok = False
|
|
return ok
|
|
|
|
|
|
class PayForm(FlaskForm):
|
|
account_id = SelectField('Pay From Account', validators=[DataRequired()])
|
|
category_id = SelectField('Category', validators=[Optional()])
|
|
paid_date = DateField('Payment Date', validators=[DataRequired()], default=date.today)
|
|
submit = SubmitField('Mark Paid')
|
|
|
|
|
|
# ── index ────────────────────────────────────────────────────────────────────
|
|
|
|
@utilities_bp.route('/')
|
|
@login_required
|
|
def index():
|
|
providers = (UtilityProvider.query
|
|
.filter_by(is_active=True)
|
|
.order_by(UtilityProvider.utility_type, UtilityProvider.name)
|
|
.all())
|
|
|
|
summary = utility_service.dashboard_summary()
|
|
summaries = {p.id: utility_service.provider_summary(p) for p in providers}
|
|
|
|
return render_template('utilities/index.html',
|
|
providers=providers,
|
|
summaries=summaries,
|
|
summary=summary,
|
|
chart=utility_service.monthly_series(12),
|
|
type_totals=utility_service.type_totals(12),
|
|
type_meta=UTILITY_TYPE_META)
|
|
|
|
|
|
# ── providers ────────────────────────────────────────────────────────────────
|
|
|
|
@utilities_bp.route('/providers')
|
|
@login_required
|
|
def providers():
|
|
provs = (UtilityProvider.query
|
|
.order_by(UtilityProvider.is_active.desc(),
|
|
UtilityProvider.utility_type, UtilityProvider.name)
|
|
.all())
|
|
counts = {p.id: p.bills.count() for p in provs}
|
|
return render_template('utilities/providers.html', providers=provs, counts=counts)
|
|
|
|
|
|
@utilities_bp.route('/providers/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
def provider_new():
|
|
form = ProviderForm()
|
|
form.default_account_id.choices = _account_choices()
|
|
form.category_id.choices = _category_choices()
|
|
|
|
if request.method == 'GET':
|
|
# Default to the seeded "Utilities" expense category when it exists
|
|
util_cat = Category.query.filter(Category.name == 'Utilities').first()
|
|
if util_cat:
|
|
form.category_id.data = str(util_cat.id)
|
|
|
|
if form.validate_on_submit():
|
|
meta = UTILITY_TYPE_META.get(form.utility_type.data, UTILITY_TYPE_META['other'])
|
|
prov = UtilityProvider(
|
|
name=form.name.data.strip(),
|
|
utility_type=form.utility_type.data,
|
|
account_number=(form.account_number.data or '').strip() or None,
|
|
usage_unit=(form.usage_unit.data or '').strip() or None,
|
|
default_account_id=_safe_int(form.default_account_id.data),
|
|
category_id=_safe_int(form.category_id.data),
|
|
billing_day=form.billing_day.data,
|
|
color=form.color.data or meta[2],
|
|
icon=form.icon.data or meta[1],
|
|
notes=form.notes.data,
|
|
)
|
|
db.session.add(prov)
|
|
db.session.commit()
|
|
flash(f'Provider "{prov.name}" added.', 'success')
|
|
return redirect(url_for('utilities.provider_detail', id=prov.id))
|
|
|
|
return render_template('utilities/provider_form.html', form=form, title='New Provider',
|
|
colors=UTILITY_COLORS, icons=UTILITY_ICONS,
|
|
type_defaults=_type_defaults())
|
|
|
|
|
|
@utilities_bp.route('/providers/<int:id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def provider_edit(id):
|
|
prov = db.get_or_404(UtilityProvider, id)
|
|
form = ProviderForm(obj=prov)
|
|
form.default_account_id.choices = _account_choices()
|
|
form.category_id.choices = _category_choices()
|
|
|
|
if request.method == 'GET':
|
|
form.default_account_id.data = str(prov.default_account_id) if prov.default_account_id else ''
|
|
form.category_id.data = str(prov.category_id) if prov.category_id else ''
|
|
|
|
if form.validate_on_submit():
|
|
prov.name = form.name.data.strip()
|
|
prov.utility_type = form.utility_type.data
|
|
prov.account_number = (form.account_number.data or '').strip() or None
|
|
prov.usage_unit = (form.usage_unit.data or '').strip() or None
|
|
prov.default_account_id = _safe_int(form.default_account_id.data)
|
|
prov.category_id = _safe_int(form.category_id.data)
|
|
prov.billing_day = form.billing_day.data
|
|
prov.color = form.color.data or prov.color
|
|
prov.icon = form.icon.data or prov.icon
|
|
prov.notes = form.notes.data
|
|
db.session.commit()
|
|
flash('Provider updated.', 'success')
|
|
return redirect(url_for('utilities.provider_detail', id=prov.id))
|
|
|
|
return render_template('utilities/provider_form.html', form=form, title='Edit Provider',
|
|
provider=prov, colors=UTILITY_COLORS, icons=UTILITY_ICONS,
|
|
type_defaults=_type_defaults())
|
|
|
|
|
|
@utilities_bp.route('/providers/<int:id>/toggle', methods=['POST'])
|
|
@login_required
|
|
def provider_toggle(id):
|
|
prov = db.get_or_404(UtilityProvider, id)
|
|
prov.is_active = not prov.is_active
|
|
db.session.commit()
|
|
flash(f'Provider "{prov.name}" {"reactivated" if prov.is_active else "archived"}.', 'info')
|
|
return redirect(url_for('utilities.providers'))
|
|
|
|
|
|
@utilities_bp.route('/providers/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def provider_delete(id):
|
|
prov = db.get_or_404(UtilityProvider, id)
|
|
count = prov.bills.count()
|
|
if count and request.form.get('confirm_bills') != 'yes':
|
|
flash(f'"{prov.name}" still has {count} bill(s). Archive it instead, '
|
|
f'or confirm deletion from the provider page.', 'warning')
|
|
return redirect(url_for('utilities.providers'))
|
|
|
|
name = prov.name
|
|
db.session.delete(prov) # cascades to its bills
|
|
db.session.commit()
|
|
flash(f'Provider "{name}" and {count} bill(s) deleted.', 'info')
|
|
return redirect(url_for('utilities.providers'))
|
|
|
|
|
|
@utilities_bp.route('/providers/<int:id>')
|
|
@login_required
|
|
def provider_detail(id):
|
|
prov = db.get_or_404(UtilityProvider, id)
|
|
page = _safe_int(request.args.get('page'), 1) or 1
|
|
pagination = (prov.bills
|
|
.order_by(UtilityBill.period_start.desc())
|
|
.paginate(page=page, per_page=24, error_out=False))
|
|
|
|
return render_template('utilities/detail.html',
|
|
provider=prov,
|
|
bills=pagination.items,
|
|
pagination=pagination,
|
|
stats=utility_service.provider_summary(prov),
|
|
series=utility_service.usage_series(prov, 24))
|
|
|
|
|
|
# ── bills ────────────────────────────────────────────────────────────────────
|
|
|
|
@utilities_bp.route('/bills')
|
|
@login_required
|
|
def bills():
|
|
page = _safe_int(request.args.get('page'), 1) or 1
|
|
provider_id = _safe_int(request.args.get('provider_id'))
|
|
utility_type = request.args.get('utility_type') or ''
|
|
status = request.args.get('status') or ''
|
|
year = _safe_int(request.args.get('year'))
|
|
|
|
q = UtilityBill.query.join(UtilityProvider)
|
|
|
|
if provider_id:
|
|
q = q.filter(UtilityBill.provider_id == provider_id)
|
|
if utility_type in UTILITY_TYPES:
|
|
q = q.filter(UtilityProvider.utility_type == utility_type)
|
|
if year:
|
|
q = q.filter(UtilityBill.period_start >= date(year, 1, 1),
|
|
UtilityBill.period_start <= date(year, 12, 31))
|
|
if status == 'paid':
|
|
q = q.filter(UtilityBill.is_paid == True)
|
|
elif status == 'unpaid':
|
|
q = q.filter(UtilityBill.is_paid == False)
|
|
elif status == 'overdue':
|
|
q = q.filter(UtilityBill.is_paid == False,
|
|
UtilityBill.due_date.isnot(None),
|
|
UtilityBill.due_date < date.today())
|
|
|
|
pagination = (q.order_by(UtilityBill.period_start.desc(), UtilityBill.id.desc())
|
|
.paginate(page=page, per_page=30, error_out=False))
|
|
|
|
years = sorted({row[0].year for row in
|
|
db.session.query(UtilityBill.period_start).all()}, reverse=True)
|
|
|
|
return render_template('utilities/bills.html',
|
|
bills=pagination.items,
|
|
pagination=pagination,
|
|
providers=UtilityProvider.query.order_by(UtilityProvider.name).all(),
|
|
type_meta=UTILITY_TYPE_META,
|
|
provider_id=provider_id, utility_type=utility_type,
|
|
status=status, year=year, years=years)
|
|
|
|
|
|
def _apply_bill_form(bill, form):
|
|
bill.provider_id = int(form.provider_id.data)
|
|
bill.period_start = form.period_start.data
|
|
bill.period_end = form.period_end.data
|
|
bill.amount = form.amount.data
|
|
bill.due_date = form.due_date.data
|
|
bill.meter_start = form.meter_start.data
|
|
bill.meter_end = form.meter_end.data
|
|
bill.usage = form.usage.data
|
|
bill.notes = form.notes.data
|
|
|
|
provider = db.session.get(UtilityProvider, bill.provider_id)
|
|
bill.usage_unit = provider.usage_unit if provider else None
|
|
# Meter readings win over a typed usage figure
|
|
bill.sync_usage_from_meter()
|
|
|
|
|
|
@utilities_bp.route('/bills/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
def bill_new():
|
|
form = BillForm()
|
|
form.provider_id.choices = _provider_choices()
|
|
|
|
if not form.provider_id.choices:
|
|
flash('Add a utility provider before recording bills.', 'warning')
|
|
return redirect(url_for('utilities.provider_new'))
|
|
|
|
preset = _safe_int(request.args.get('provider_id'))
|
|
if request.method == 'GET':
|
|
if preset:
|
|
form.provider_id.data = str(preset)
|
|
# Default to last month's billing period
|
|
today = date.today()
|
|
first_this = today.replace(day=1)
|
|
form.period_end.data = first_this - timedelta(days=1)
|
|
form.period_start.data = form.period_end.data.replace(day=1)
|
|
|
|
if form.validate_on_submit():
|
|
bill = UtilityBill()
|
|
_apply_bill_form(bill, form)
|
|
db.session.add(bill)
|
|
try:
|
|
db.session.commit()
|
|
except IntegrityError:
|
|
db.session.rollback()
|
|
flash('A bill for that provider and period start already exists.', 'danger')
|
|
return render_template('utilities/bill_form.html', form=form, title='New Bill',
|
|
provider_units=_provider_units())
|
|
|
|
flash('Bill recorded.', 'success')
|
|
if request.form.get('pay_now') == 'yes':
|
|
return redirect(url_for('utilities.bill_pay', id=bill.id))
|
|
return redirect(url_for('utilities.provider_detail', id=bill.provider_id))
|
|
|
|
return render_template('utilities/bill_form.html', form=form, title='New Bill',
|
|
provider_units=_provider_units())
|
|
|
|
|
|
@utilities_bp.route('/bills/<int:id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def bill_edit(id):
|
|
bill = db.get_or_404(UtilityBill, id)
|
|
form = BillForm(obj=bill)
|
|
form.provider_id.choices = _provider_choices(include_inactive=True)
|
|
|
|
if request.method == 'GET':
|
|
form.provider_id.data = str(bill.provider_id)
|
|
|
|
if form.validate_on_submit():
|
|
_apply_bill_form(bill, form)
|
|
try:
|
|
db.session.commit()
|
|
except IntegrityError:
|
|
db.session.rollback()
|
|
flash('A bill for that provider and period start already exists.', 'danger')
|
|
return render_template('utilities/bill_form.html', form=form, title='Edit Bill',
|
|
bill=bill, provider_units=_provider_units())
|
|
|
|
flash('Bill updated.', 'success')
|
|
return redirect(url_for('utilities.provider_detail', id=bill.provider_id))
|
|
|
|
return render_template('utilities/bill_form.html', form=form, title='Edit Bill',
|
|
bill=bill, provider_units=_provider_units())
|
|
|
|
|
|
@utilities_bp.route('/bills/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def bill_delete(id):
|
|
bill = db.get_or_404(UtilityBill, id)
|
|
provider_id = bill.provider_id
|
|
|
|
# Only remove the payment transaction if this feature created it
|
|
txn = bill.transaction
|
|
drop_txn = utility_service.is_generated_payment(bill)
|
|
account_id = txn.account_id if txn else None
|
|
|
|
db.session.delete(bill)
|
|
if drop_txn:
|
|
db.session.delete(txn)
|
|
db.session.commit()
|
|
|
|
if drop_txn and account_id:
|
|
calc_balance(account_id)
|
|
flash('Bill and its payment transaction deleted.', 'info')
|
|
else:
|
|
flash('Bill deleted.', 'info')
|
|
return redirect(url_for('utilities.provider_detail', id=provider_id))
|
|
|
|
|
|
# ── payment ──────────────────────────────────────────────────────────────────
|
|
|
|
@utilities_bp.route('/bills/<int:id>/pay', methods=['GET', 'POST'])
|
|
@login_required
|
|
def bill_pay(id):
|
|
bill = db.get_or_404(UtilityBill, id)
|
|
if bill.is_paid:
|
|
flash('That bill is already marked paid.', 'info')
|
|
return redirect(url_for('utilities.provider_detail', id=bill.provider_id))
|
|
|
|
form = PayForm()
|
|
form.account_id.choices = [c for c in _account_choices() if c[0]]
|
|
form.category_id.choices = _category_choices()
|
|
|
|
if not form.account_id.choices:
|
|
flash('Create an active account before paying bills.', 'warning')
|
|
return redirect(url_for('accounts.index'))
|
|
|
|
if request.method == 'GET':
|
|
prov = bill.provider
|
|
if prov.default_account_id:
|
|
form.account_id.data = str(prov.default_account_id)
|
|
if prov.category_id:
|
|
form.category_id.data = str(prov.category_id)
|
|
form.paid_date.data = bill.due_date or date.today()
|
|
|
|
if form.validate_on_submit():
|
|
txn = utility_service.build_payment_transaction(
|
|
bill,
|
|
account_id=int(form.account_id.data),
|
|
paid_date=form.paid_date.data,
|
|
category_id=_safe_int(form.category_id.data),
|
|
)
|
|
db.session.flush() # need txn.id before linking
|
|
|
|
bill.is_paid = True
|
|
bill.paid_date = form.paid_date.data
|
|
bill.transaction_id = txn.id
|
|
db.session.commit()
|
|
|
|
calc_balance(txn.account_id)
|
|
log.info('[utilities] bill %s marked paid — txn %s', bill.id, txn.id)
|
|
|
|
from app.services.alert_service import check_and_flash_budget_alerts
|
|
check_and_flash_budget_alerts(flash)
|
|
|
|
flash(f'Bill paid — expense of {bill.amount} recorded.', 'success')
|
|
return redirect(url_for('utilities.provider_detail', id=bill.provider_id))
|
|
|
|
return render_template('utilities/pay.html', form=form, bill=bill)
|
|
|
|
|
|
@utilities_bp.route('/bills/<int:id>/link', methods=['GET', 'POST'])
|
|
@login_required
|
|
def bill_link(id):
|
|
bill = db.get_or_404(UtilityBill, id)
|
|
|
|
if request.method == 'POST':
|
|
txn_id = _safe_int(request.form.get('transaction_id'))
|
|
txn = db.session.get(Transaction, txn_id) if txn_id else None
|
|
if not txn:
|
|
flash('Select a transaction to link.', 'warning')
|
|
return redirect(url_for('utilities.bill_link', id=bill.id))
|
|
|
|
clash = UtilityBill.query.filter(UtilityBill.transaction_id == txn.id,
|
|
UtilityBill.id != bill.id).first()
|
|
if clash:
|
|
flash('That transaction is already linked to another bill.', 'danger')
|
|
return redirect(url_for('utilities.bill_link', id=bill.id))
|
|
|
|
bill.transaction_id = txn.id
|
|
bill.is_paid = True
|
|
bill.paid_date = txn.date
|
|
db.session.commit()
|
|
flash('Payment linked.', 'success')
|
|
return redirect(url_for('utilities.provider_detail', id=bill.provider_id))
|
|
|
|
return render_template('utilities/link.html', bill=bill,
|
|
candidates=utility_service.candidate_transactions(bill))
|
|
|
|
|
|
@utilities_bp.route('/bills/<int:id>/unpay', methods=['POST'])
|
|
@login_required
|
|
def bill_unpay(id):
|
|
bill = db.get_or_404(UtilityBill, id)
|
|
txn = bill.transaction
|
|
drop_txn = utility_service.is_generated_payment(bill)
|
|
account_id = txn.account_id if txn else None
|
|
|
|
bill.is_paid = False
|
|
bill.paid_date = None
|
|
bill.transaction_id = None
|
|
if drop_txn:
|
|
db.session.delete(txn)
|
|
db.session.commit()
|
|
|
|
if drop_txn and account_id:
|
|
calc_balance(account_id)
|
|
flash('Bill reopened — its payment transaction was removed.', 'info')
|
|
else:
|
|
flash('Bill reopened — the linked transaction was left in place.', 'info')
|
|
return redirect(url_for('utilities.provider_detail', id=bill.provider_id))
|
|
|
|
|
|
# ── api ──────────────────────────────────────────────────────────────────────
|
|
|
|
@utilities_bp.route('/api/usage/<int:provider_id>')
|
|
@login_required
|
|
def api_usage(provider_id):
|
|
prov = db.get_or_404(UtilityProvider, provider_id)
|
|
months = _safe_int(request.args.get('months'), 24) or 24
|
|
return jsonify(utility_service.usage_series(prov, min(max(months, 3), 60)))
|