You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1224 lines
38 KiB
1224 lines
38 KiB
from flask import Flask, render_template, request, redirect |
|
from db import get_db_connection |
|
from utils import generate_client_code, generate_service_code |
|
from datetime import datetime, timezone |
|
from zoneinfo import ZoneInfo |
|
from decimal import Decimal, InvalidOperation |
|
|
|
app = Flask( |
|
__name__, |
|
template_folder="../templates", |
|
static_folder="../static", |
|
) |
|
|
|
LOCAL_TZ = ZoneInfo("America/Toronto") |
|
|
|
def load_version(): |
|
try: |
|
with open("/home/def/otb_billing/VERSION", "r") as f: |
|
return f.read().strip() |
|
except Exception: |
|
return "unknown" |
|
|
|
APP_VERSION = load_version() |
|
|
|
@app.context_processor |
|
def inject_version(): |
|
return {"app_version": APP_VERSION} |
|
|
|
def fmt_local(dt_value): |
|
if not dt_value: |
|
return "" |
|
if isinstance(dt_value, str): |
|
try: |
|
dt_value = datetime.fromisoformat(dt_value) |
|
except ValueError: |
|
return str(dt_value) |
|
if dt_value.tzinfo is None: |
|
dt_value = dt_value.replace(tzinfo=timezone.utc) |
|
return dt_value.astimezone(LOCAL_TZ).strftime("%Y-%m-%d %I:%M:%S %p") |
|
|
|
def to_decimal(value): |
|
if value is None or value == "": |
|
return Decimal("0") |
|
try: |
|
return Decimal(str(value)) |
|
except (InvalidOperation, ValueError): |
|
return Decimal("0") |
|
|
|
def fmt_money(value, currency_code="CAD"): |
|
amount = to_decimal(value) |
|
if currency_code == "CAD": |
|
return f"{amount:.2f}" |
|
return f"{amount:.8f}" |
|
|
|
def refresh_overdue_invoices(): |
|
conn = get_db_connection() |
|
cursor = conn.cursor() |
|
cursor.execute(""" |
|
UPDATE invoices |
|
SET status = 'overdue' |
|
WHERE due_at IS NOT NULL |
|
AND due_at < UTC_TIMESTAMP() |
|
AND status IN ('pending', 'partial') |
|
""") |
|
conn.commit() |
|
conn.close() |
|
|
|
def recalc_invoice_totals(invoice_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute(""" |
|
SELECT id, total_amount, due_at, status |
|
FROM invoices |
|
WHERE id = %s |
|
""", (invoice_id,)) |
|
invoice = cursor.fetchone() |
|
|
|
if not invoice: |
|
conn.close() |
|
return |
|
|
|
cursor.execute(""" |
|
SELECT COALESCE(SUM(payment_amount), 0) AS total_paid |
|
FROM payments |
|
WHERE invoice_id = %s |
|
AND payment_status = 'confirmed' |
|
""", (invoice_id,)) |
|
row = cursor.fetchone() |
|
|
|
total_paid = to_decimal(row["total_paid"]) |
|
total_amount = to_decimal(invoice["total_amount"]) |
|
|
|
if invoice["status"] == "cancelled": |
|
update_cursor = conn.cursor() |
|
update_cursor.execute(""" |
|
UPDATE invoices |
|
SET amount_paid = %s, |
|
paid_at = NULL |
|
WHERE id = %s |
|
""", ( |
|
str(total_paid), |
|
invoice_id |
|
)) |
|
conn.commit() |
|
conn.close() |
|
return |
|
|
|
if total_paid >= total_amount and total_amount > 0: |
|
new_status = "paid" |
|
paid_at_value = "UTC_TIMESTAMP()" |
|
elif total_paid > 0: |
|
new_status = "partial" |
|
paid_at_value = "NULL" |
|
else: |
|
if invoice["due_at"] and invoice["due_at"] < datetime.utcnow(): |
|
new_status = "overdue" |
|
else: |
|
new_status = "pending" |
|
paid_at_value = "NULL" |
|
|
|
update_cursor = conn.cursor() |
|
update_cursor.execute(f""" |
|
UPDATE invoices |
|
SET amount_paid = %s, |
|
status = %s, |
|
paid_at = {paid_at_value} |
|
WHERE id = %s |
|
""", ( |
|
str(total_paid), |
|
new_status, |
|
invoice_id |
|
)) |
|
|
|
conn.commit() |
|
conn.close() |
|
|
|
def get_client_credit_balance(client_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
cursor.execute(""" |
|
SELECT COALESCE(SUM(amount), 0) AS balance |
|
FROM credit_ledger |
|
WHERE client_id = %s |
|
""", (client_id,)) |
|
row = cursor.fetchone() |
|
conn.close() |
|
return to_decimal(row["balance"]) |
|
|
|
@app.template_filter("localtime") |
|
def localtime_filter(value): |
|
return fmt_local(value) |
|
|
|
@app.template_filter("money") |
|
def money_filter(value, currency_code="CAD"): |
|
return fmt_money(value, currency_code) |
|
|
|
@app.route("/") |
|
def index(): |
|
refresh_overdue_invoices() |
|
|
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute("SELECT COUNT(*) AS total_clients FROM clients") |
|
total_clients = cursor.fetchone()["total_clients"] |
|
|
|
cursor.execute("SELECT COUNT(*) AS active_services FROM services WHERE status = 'active'") |
|
active_services = cursor.fetchone()["active_services"] |
|
|
|
cursor.execute(""" |
|
SELECT COUNT(*) AS outstanding_invoices |
|
FROM invoices |
|
WHERE status IN ('pending', 'partial', 'overdue') |
|
""") |
|
outstanding_invoices = cursor.fetchone()["outstanding_invoices"] |
|
|
|
cursor.execute(""" |
|
SELECT COALESCE(SUM(cad_value_at_payment), 0) AS revenue_received |
|
FROM payments |
|
WHERE payment_status = 'confirmed' |
|
""") |
|
revenue_received = cursor.fetchone()["revenue_received"] |
|
|
|
conn.close() |
|
|
|
return render_template( |
|
"dashboard.html", |
|
total_clients=total_clients, |
|
active_services=active_services, |
|
outstanding_invoices=outstanding_invoices, |
|
revenue_received=revenue_received, |
|
) |
|
|
|
@app.route("/dbtest") |
|
def dbtest(): |
|
try: |
|
conn = get_db_connection() |
|
cursor = conn.cursor() |
|
cursor.execute("SELECT NOW()") |
|
result = cursor.fetchone() |
|
conn.close() |
|
return f""" |
|
<h1>OTB Billing v{APP_VERSION}</h1> |
|
<h2>Database OK</h2> |
|
<p><a href="/">Home</a></p> |
|
<p>DB server time (UTC): {result[0]}</p> |
|
<p>Displayed local time: {fmt_local(result[0])}</p> |
|
""" |
|
except Exception as e: |
|
return f"<h1>Database FAILED</h1><pre>{e}</pre>" |
|
|
|
@app.route("/clients") |
|
def clients(): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
cursor.execute("SELECT * FROM clients ORDER BY id DESC") |
|
clients = cursor.fetchall() |
|
conn.close() |
|
|
|
for client in clients: |
|
client["credit_balance"] = get_client_credit_balance(client["id"]) |
|
|
|
return render_template("clients/list.html", clients=clients) |
|
|
|
@app.route("/clients/new", methods=["GET", "POST"]) |
|
def new_client(): |
|
if request.method == "POST": |
|
company_name = request.form["company_name"] |
|
contact_name = request.form["contact_name"] |
|
email = request.form["email"] |
|
phone = request.form["phone"] |
|
|
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
cursor.execute("SELECT MAX(id) AS last_id FROM clients") |
|
result = cursor.fetchone() |
|
last_number = result["last_id"] if result["last_id"] else 0 |
|
|
|
client_code = generate_client_code(company_name, last_number) |
|
|
|
insert_cursor = conn.cursor() |
|
insert_cursor.execute( |
|
""" |
|
INSERT INTO clients |
|
(client_code, company_name, contact_name, email, phone) |
|
VALUES (%s, %s, %s, %s, %s) |
|
""", |
|
(client_code, company_name, contact_name, email, phone) |
|
) |
|
conn.commit() |
|
conn.close() |
|
|
|
return redirect("/clients") |
|
|
|
return render_template("clients/new.html") |
|
|
|
@app.route("/clients/edit/<int:client_id>", methods=["GET", "POST"]) |
|
def edit_client(client_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
if request.method == "POST": |
|
company_name = request.form.get("company_name", "").strip() |
|
contact_name = request.form.get("contact_name", "").strip() |
|
email = request.form.get("email", "").strip() |
|
phone = request.form.get("phone", "").strip() |
|
status = request.form.get("status", "").strip() |
|
notes = request.form.get("notes", "").strip() |
|
|
|
errors = [] |
|
|
|
if not company_name: |
|
errors.append("Company name is required.") |
|
if not status: |
|
errors.append("Status is required.") |
|
|
|
if errors: |
|
cursor.execute("SELECT * FROM clients WHERE id = %s", (client_id,)) |
|
client = cursor.fetchone() |
|
client["credit_balance"] = get_client_credit_balance(client_id) |
|
conn.close() |
|
return render_template("clients/edit.html", client=client, errors=errors) |
|
|
|
update_cursor = conn.cursor() |
|
update_cursor.execute(""" |
|
UPDATE clients |
|
SET company_name = %s, |
|
contact_name = %s, |
|
email = %s, |
|
phone = %s, |
|
status = %s, |
|
notes = %s |
|
WHERE id = %s |
|
""", ( |
|
company_name, |
|
contact_name or None, |
|
email or None, |
|
phone or None, |
|
status, |
|
notes or None, |
|
client_id |
|
)) |
|
conn.commit() |
|
conn.close() |
|
return redirect("/clients") |
|
|
|
cursor.execute("SELECT * FROM clients WHERE id = %s", (client_id,)) |
|
client = cursor.fetchone() |
|
conn.close() |
|
|
|
if not client: |
|
return "Client not found", 404 |
|
|
|
client["credit_balance"] = get_client_credit_balance(client_id) |
|
|
|
return render_template("clients/edit.html", client=client, errors=[]) |
|
|
|
@app.route("/credits/<int:client_id>") |
|
def client_credits(client_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute(""" |
|
SELECT id, client_code, company_name |
|
FROM clients |
|
WHERE id = %s |
|
""", (client_id,)) |
|
client = cursor.fetchone() |
|
|
|
if not client: |
|
conn.close() |
|
return "Client not found", 404 |
|
|
|
cursor.execute(""" |
|
SELECT * |
|
FROM credit_ledger |
|
WHERE client_id = %s |
|
ORDER BY id DESC |
|
""", (client_id,)) |
|
entries = cursor.fetchall() |
|
|
|
conn.close() |
|
|
|
balance = get_client_credit_balance(client_id) |
|
|
|
return render_template( |
|
"credits/list.html", |
|
client=client, |
|
entries=entries, |
|
balance=balance, |
|
) |
|
|
|
@app.route("/credits/add/<int:client_id>", methods=["GET", "POST"]) |
|
def add_credit(client_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute(""" |
|
SELECT id, client_code, company_name |
|
FROM clients |
|
WHERE id = %s |
|
""", (client_id,)) |
|
client = cursor.fetchone() |
|
|
|
if not client: |
|
conn.close() |
|
return "Client not found", 404 |
|
|
|
if request.method == "POST": |
|
entry_type = request.form.get("entry_type", "").strip() |
|
amount = request.form.get("amount", "").strip() |
|
currency_code = request.form.get("currency_code", "").strip() |
|
notes = request.form.get("notes", "").strip() |
|
|
|
errors = [] |
|
|
|
if not entry_type: |
|
errors.append("Entry type is required.") |
|
if not amount: |
|
errors.append("Amount is required.") |
|
if not currency_code: |
|
errors.append("Currency code is required.") |
|
|
|
if not errors: |
|
try: |
|
amount_value = Decimal(str(amount)) |
|
if amount_value == 0: |
|
errors.append("Amount cannot be zero.") |
|
except Exception: |
|
errors.append("Amount must be a valid number.") |
|
|
|
if errors: |
|
conn.close() |
|
return render_template("credits/add.html", client=client, errors=errors) |
|
|
|
insert_cursor = conn.cursor() |
|
insert_cursor.execute(""" |
|
INSERT INTO credit_ledger |
|
( |
|
client_id, |
|
entry_type, |
|
amount, |
|
currency_code, |
|
notes |
|
) |
|
VALUES (%s, %s, %s, %s, %s) |
|
""", ( |
|
client_id, |
|
entry_type, |
|
amount, |
|
currency_code, |
|
notes or None |
|
)) |
|
conn.commit() |
|
conn.close() |
|
|
|
return redirect(f"/credits/{client_id}") |
|
|
|
conn.close() |
|
return render_template("credits/add.html", client=client, errors=[]) |
|
|
|
@app.route("/services") |
|
def services(): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
cursor.execute(""" |
|
SELECT s.*, c.client_code, c.company_name |
|
FROM services s |
|
JOIN clients c ON s.client_id = c.id |
|
ORDER BY s.id DESC |
|
""") |
|
services = cursor.fetchall() |
|
conn.close() |
|
return render_template("services/list.html", services=services) |
|
|
|
@app.route("/services/new", methods=["GET", "POST"]) |
|
def new_service(): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
if request.method == "POST": |
|
client_id = request.form["client_id"] |
|
service_name = request.form["service_name"] |
|
service_type = request.form["service_type"] |
|
billing_cycle = request.form["billing_cycle"] |
|
currency_code = request.form["currency_code"] |
|
recurring_amount = request.form["recurring_amount"] |
|
status = request.form["status"] |
|
start_date = request.form["start_date"] or None |
|
description = request.form["description"] |
|
|
|
cursor.execute("SELECT MAX(id) AS last_id FROM services") |
|
result = cursor.fetchone() |
|
last_number = result["last_id"] if result["last_id"] else 0 |
|
service_code = generate_service_code(service_name, last_number) |
|
|
|
insert_cursor = conn.cursor() |
|
insert_cursor.execute( |
|
""" |
|
INSERT INTO services |
|
( |
|
client_id, |
|
service_code, |
|
service_name, |
|
service_type, |
|
billing_cycle, |
|
status, |
|
currency_code, |
|
recurring_amount, |
|
start_date, |
|
description |
|
) |
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) |
|
""", |
|
( |
|
client_id, |
|
service_code, |
|
service_name, |
|
service_type, |
|
billing_cycle, |
|
status, |
|
currency_code, |
|
recurring_amount, |
|
start_date, |
|
description |
|
) |
|
) |
|
conn.commit() |
|
conn.close() |
|
|
|
return redirect("/services") |
|
|
|
cursor.execute("SELECT id, client_code, company_name FROM clients ORDER BY company_name ASC") |
|
clients = cursor.fetchall() |
|
conn.close() |
|
return render_template("services/new.html", clients=clients) |
|
|
|
@app.route("/services/edit/<int:service_id>", methods=["GET", "POST"]) |
|
def edit_service(service_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
if request.method == "POST": |
|
client_id = request.form.get("client_id", "").strip() |
|
service_name = request.form.get("service_name", "").strip() |
|
service_type = request.form.get("service_type", "").strip() |
|
billing_cycle = request.form.get("billing_cycle", "").strip() |
|
currency_code = request.form.get("currency_code", "").strip() |
|
recurring_amount = request.form.get("recurring_amount", "").strip() |
|
status = request.form.get("status", "").strip() |
|
start_date = request.form.get("start_date", "").strip() |
|
description = request.form.get("description", "").strip() |
|
|
|
errors = [] |
|
|
|
if not client_id: |
|
errors.append("Client is required.") |
|
if not service_name: |
|
errors.append("Service name is required.") |
|
if not service_type: |
|
errors.append("Service type is required.") |
|
if not billing_cycle: |
|
errors.append("Billing cycle is required.") |
|
if not currency_code: |
|
errors.append("Currency code is required.") |
|
if not recurring_amount: |
|
errors.append("Recurring amount is required.") |
|
if not status: |
|
errors.append("Status is required.") |
|
|
|
if not errors: |
|
try: |
|
recurring_amount_value = float(recurring_amount) |
|
if recurring_amount_value < 0: |
|
errors.append("Recurring amount cannot be negative.") |
|
except ValueError: |
|
errors.append("Recurring amount must be a valid number.") |
|
|
|
if errors: |
|
cursor.execute(""" |
|
SELECT s.*, c.company_name |
|
FROM services s |
|
LEFT JOIN clients c ON s.client_id = c.id |
|
WHERE s.id = %s |
|
""", (service_id,)) |
|
service = cursor.fetchone() |
|
|
|
cursor.execute("SELECT id, client_code, company_name FROM clients ORDER BY company_name ASC") |
|
clients = cursor.fetchall() |
|
|
|
conn.close() |
|
return render_template("services/edit.html", service=service, clients=clients, errors=errors) |
|
|
|
update_cursor = conn.cursor() |
|
update_cursor.execute(""" |
|
UPDATE services |
|
SET client_id = %s, |
|
service_name = %s, |
|
service_type = %s, |
|
billing_cycle = %s, |
|
status = %s, |
|
currency_code = %s, |
|
recurring_amount = %s, |
|
start_date = %s, |
|
description = %s |
|
WHERE id = %s |
|
""", ( |
|
client_id, |
|
service_name, |
|
service_type, |
|
billing_cycle, |
|
status, |
|
currency_code, |
|
recurring_amount, |
|
start_date or None, |
|
description or None, |
|
service_id |
|
)) |
|
conn.commit() |
|
conn.close() |
|
return redirect("/services") |
|
|
|
cursor.execute(""" |
|
SELECT s.*, c.company_name |
|
FROM services s |
|
LEFT JOIN clients c ON s.client_id = c.id |
|
WHERE s.id = %s |
|
""", (service_id,)) |
|
service = cursor.fetchone() |
|
|
|
cursor.execute("SELECT id, client_code, company_name FROM clients ORDER BY company_name ASC") |
|
clients = cursor.fetchall() |
|
conn.close() |
|
|
|
if not service: |
|
return "Service not found", 404 |
|
|
|
return render_template("services/edit.html", service=service, clients=clients, errors=[]) |
|
|
|
@app.route("/invoices") |
|
def invoices(): |
|
refresh_overdue_invoices() |
|
|
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
cursor.execute(""" |
|
SELECT |
|
i.*, |
|
c.client_code, |
|
c.company_name, |
|
COALESCE((SELECT COUNT(*) FROM payments p WHERE p.invoice_id = i.id), 0) AS payment_count |
|
FROM invoices i |
|
JOIN clients c ON i.client_id = c.id |
|
ORDER BY i.id DESC |
|
""") |
|
invoices = cursor.fetchall() |
|
conn.close() |
|
return render_template("invoices/list.html", invoices=invoices) |
|
|
|
@app.route("/invoices/new", methods=["GET", "POST"]) |
|
def new_invoice(): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
if request.method == "POST": |
|
client_id = request.form.get("client_id", "").strip() |
|
service_id = request.form.get("service_id", "").strip() |
|
currency_code = request.form.get("currency_code", "").strip() |
|
total_amount = request.form.get("total_amount", "").strip() |
|
due_at = request.form.get("due_at", "").strip() |
|
notes = request.form.get("notes", "").strip() |
|
|
|
errors = [] |
|
|
|
if not client_id: |
|
errors.append("Client is required.") |
|
if not service_id: |
|
errors.append("Service is required.") |
|
if not currency_code: |
|
errors.append("Currency is required.") |
|
if not total_amount: |
|
errors.append("Total amount is required.") |
|
if not due_at: |
|
errors.append("Due date is required.") |
|
|
|
if not errors: |
|
try: |
|
amount_value = float(total_amount) |
|
if amount_value <= 0: |
|
errors.append("Total amount must be greater than zero.") |
|
except ValueError: |
|
errors.append("Total amount must be a valid number.") |
|
|
|
if errors: |
|
cursor.execute("SELECT id, client_code, company_name FROM clients ORDER BY company_name") |
|
clients = cursor.fetchall() |
|
|
|
cursor.execute("SELECT id, service_code, service_name FROM services ORDER BY service_name") |
|
services = cursor.fetchall() |
|
|
|
conn.close() |
|
|
|
form_data = { |
|
"client_id": client_id, |
|
"service_id": service_id, |
|
"currency_code": currency_code, |
|
"total_amount": total_amount, |
|
"due_at": due_at, |
|
"notes": notes, |
|
} |
|
|
|
return render_template( |
|
"invoices/new.html", |
|
clients=clients, |
|
services=services, |
|
errors=errors, |
|
form_data=form_data, |
|
) |
|
|
|
cursor.execute("SELECT MAX(id) AS last_id FROM invoices") |
|
result = cursor.fetchone() |
|
number = (result["last_id"] or 0) + 1 |
|
invoice_number = f"INV-{number:04d}" |
|
|
|
insert_cursor = conn.cursor() |
|
insert_cursor.execute(""" |
|
INSERT INTO invoices |
|
( |
|
client_id, |
|
service_id, |
|
invoice_number, |
|
currency_code, |
|
total_amount, |
|
subtotal_amount, |
|
issued_at, |
|
due_at, |
|
status, |
|
notes |
|
) |
|
VALUES (%s, %s, %s, %s, %s, %s, UTC_TIMESTAMP(), %s, 'pending', %s) |
|
""", ( |
|
client_id, |
|
service_id, |
|
invoice_number, |
|
currency_code, |
|
total_amount, |
|
total_amount, |
|
due_at, |
|
notes |
|
)) |
|
|
|
conn.commit() |
|
conn.close() |
|
|
|
return redirect("/invoices") |
|
|
|
cursor.execute("SELECT id, client_code, company_name FROM clients ORDER BY company_name") |
|
clients = cursor.fetchall() |
|
|
|
cursor.execute("SELECT id, service_code, service_name FROM services ORDER BY service_name") |
|
services = cursor.fetchall() |
|
|
|
conn.close() |
|
|
|
return render_template( |
|
"invoices/new.html", |
|
clients=clients, |
|
services=services, |
|
errors=[], |
|
form_data={}, |
|
) |
|
|
|
@app.route("/invoices/edit/<int:invoice_id>", methods=["GET", "POST"]) |
|
def edit_invoice(invoice_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute(""" |
|
SELECT i.*, |
|
COALESCE((SELECT COUNT(*) FROM payments p WHERE p.invoice_id = i.id), 0) AS payment_count |
|
FROM invoices i |
|
WHERE i.id = %s |
|
""", (invoice_id,)) |
|
invoice = cursor.fetchone() |
|
|
|
if not invoice: |
|
conn.close() |
|
return "Invoice not found", 404 |
|
|
|
locked = invoice["payment_count"] > 0 or float(invoice["amount_paid"]) > 0 |
|
|
|
if request.method == "POST": |
|
due_at = request.form.get("due_at", "").strip() |
|
notes = request.form.get("notes", "").strip() |
|
|
|
if locked: |
|
update_cursor = conn.cursor() |
|
update_cursor.execute(""" |
|
UPDATE invoices |
|
SET due_at = %s, |
|
notes = %s |
|
WHERE id = %s |
|
""", ( |
|
due_at or None, |
|
notes or None, |
|
invoice_id |
|
)) |
|
conn.commit() |
|
conn.close() |
|
return redirect("/invoices") |
|
|
|
client_id = request.form.get("client_id", "").strip() |
|
service_id = request.form.get("service_id", "").strip() |
|
currency_code = request.form.get("currency_code", "").strip() |
|
total_amount = request.form.get("total_amount", "").strip() |
|
status = request.form.get("status", "").strip() |
|
|
|
errors = [] |
|
|
|
if not client_id: |
|
errors.append("Client is required.") |
|
if not service_id: |
|
errors.append("Service is required.") |
|
if not currency_code: |
|
errors.append("Currency is required.") |
|
if not total_amount: |
|
errors.append("Total amount is required.") |
|
if not due_at: |
|
errors.append("Due date is required.") |
|
if not status: |
|
errors.append("Status is required.") |
|
|
|
manual_statuses = {"draft", "pending", "cancelled"} |
|
if status and status not in manual_statuses: |
|
errors.append("Manual invoice status must be draft, pending, or cancelled.") |
|
|
|
if not errors: |
|
try: |
|
amount_value = float(total_amount) |
|
if amount_value < 0: |
|
errors.append("Total amount cannot be negative.") |
|
except ValueError: |
|
errors.append("Total amount must be a valid number.") |
|
|
|
cursor.execute("SELECT id, client_code, company_name FROM clients ORDER BY company_name") |
|
clients = cursor.fetchall() |
|
|
|
cursor.execute("SELECT id, service_code, service_name FROM services ORDER BY service_name") |
|
services = cursor.fetchall() |
|
|
|
if errors: |
|
invoice["client_id"] = int(client_id) if client_id else invoice["client_id"] |
|
invoice["service_id"] = int(service_id) if service_id else invoice["service_id"] |
|
invoice["currency_code"] = currency_code or invoice["currency_code"] |
|
invoice["total_amount"] = total_amount or invoice["total_amount"] |
|
invoice["due_at"] = due_at or invoice["due_at"] |
|
invoice["status"] = status or invoice["status"] |
|
invoice["notes"] = notes |
|
conn.close() |
|
return render_template("invoices/edit.html", invoice=invoice, clients=clients, services=services, errors=errors, locked=locked) |
|
|
|
update_cursor = conn.cursor() |
|
update_cursor.execute(""" |
|
UPDATE invoices |
|
SET client_id = %s, |
|
service_id = %s, |
|
currency_code = %s, |
|
total_amount = %s, |
|
subtotal_amount = %s, |
|
due_at = %s, |
|
status = %s, |
|
notes = %s |
|
WHERE id = %s |
|
""", ( |
|
client_id, |
|
service_id, |
|
currency_code, |
|
total_amount, |
|
total_amount, |
|
due_at, |
|
status, |
|
notes or None, |
|
invoice_id |
|
)) |
|
conn.commit() |
|
conn.close() |
|
return redirect("/invoices") |
|
|
|
clients = [] |
|
services = [] |
|
|
|
if not locked: |
|
cursor.execute("SELECT id, client_code, company_name FROM clients ORDER BY company_name") |
|
clients = cursor.fetchall() |
|
|
|
cursor.execute("SELECT id, service_code, service_name FROM services ORDER BY service_name") |
|
services = cursor.fetchall() |
|
|
|
conn.close() |
|
return render_template("invoices/edit.html", invoice=invoice, clients=clients, services=services, errors=[], locked=locked) |
|
|
|
@app.route("/payments") |
|
def payments(): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute(""" |
|
SELECT |
|
p.*, |
|
i.invoice_number, |
|
c.client_code, |
|
c.company_name |
|
FROM payments p |
|
JOIN invoices i ON p.invoice_id = i.id |
|
JOIN clients c ON p.client_id = c.id |
|
ORDER BY p.id DESC |
|
""") |
|
payments = cursor.fetchall() |
|
|
|
conn.close() |
|
return render_template("payments/list.html", payments=payments) |
|
|
|
@app.route("/payments/new", methods=["GET", "POST"]) |
|
def new_payment(): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
if request.method == "POST": |
|
invoice_id = request.form.get("invoice_id", "").strip() |
|
payment_method = request.form.get("payment_method", "").strip() |
|
payment_currency = request.form.get("payment_currency", "").strip() |
|
payment_amount = request.form.get("payment_amount", "").strip() |
|
cad_value_at_payment = request.form.get("cad_value_at_payment", "").strip() |
|
reference = request.form.get("reference", "").strip() |
|
sender_name = request.form.get("sender_name", "").strip() |
|
txid = request.form.get("txid", "").strip() |
|
wallet_address = request.form.get("wallet_address", "").strip() |
|
notes = request.form.get("notes", "").strip() |
|
|
|
errors = [] |
|
|
|
if not invoice_id: |
|
errors.append("Invoice is required.") |
|
if not payment_method: |
|
errors.append("Payment method is required.") |
|
if not payment_currency: |
|
errors.append("Payment currency is required.") |
|
if not payment_amount: |
|
errors.append("Payment amount is required.") |
|
if not cad_value_at_payment: |
|
errors.append("CAD value at payment is required.") |
|
|
|
if not errors: |
|
try: |
|
payment_amount_value = Decimal(str(payment_amount)) |
|
if payment_amount_value <= Decimal("0"): |
|
errors.append("Payment amount must be greater than zero.") |
|
except Exception: |
|
errors.append("Payment amount must be a valid number.") |
|
|
|
if not errors: |
|
try: |
|
cad_value_value = Decimal(str(cad_value_at_payment)) |
|
if cad_value_value < Decimal("0"): |
|
errors.append("CAD value at payment cannot be negative.") |
|
except Exception: |
|
errors.append("CAD value at payment must be a valid number.") |
|
|
|
invoice_row = None |
|
|
|
if not errors: |
|
cursor.execute(""" |
|
SELECT |
|
i.id, |
|
i.client_id, |
|
i.invoice_number, |
|
i.currency_code, |
|
i.total_amount, |
|
i.amount_paid, |
|
i.status, |
|
c.client_code, |
|
c.company_name |
|
FROM invoices i |
|
JOIN clients c ON i.client_id = c.id |
|
WHERE i.id = %s |
|
""", (invoice_id,)) |
|
invoice_row = cursor.fetchone() |
|
|
|
if not invoice_row: |
|
errors.append("Selected invoice was not found.") |
|
else: |
|
allowed_statuses = {"pending", "partial", "overdue"} |
|
if invoice_row["status"] not in allowed_statuses: |
|
errors.append("Payments can only be recorded against pending, partial, or overdue invoices.") |
|
else: |
|
remaining_balance = to_decimal(invoice_row["total_amount"]) - to_decimal(invoice_row["amount_paid"]) |
|
entered_amount = to_decimal(payment_amount) |
|
|
|
if remaining_balance <= Decimal("0"): |
|
errors.append("This invoice has no remaining balance.") |
|
elif entered_amount > remaining_balance: |
|
errors.append( |
|
f"Payment amount exceeds remaining balance. Remaining balance is {fmt_money(remaining_balance, invoice_row['currency_code'])} {invoice_row['currency_code']}." |
|
) |
|
|
|
if errors: |
|
cursor.execute(""" |
|
SELECT |
|
i.id, |
|
i.invoice_number, |
|
i.currency_code, |
|
i.total_amount, |
|
i.amount_paid, |
|
i.status, |
|
c.client_code, |
|
c.company_name |
|
FROM invoices i |
|
JOIN clients c ON i.client_id = c.id |
|
WHERE i.status IN ('pending', 'partial', 'overdue') |
|
AND (i.total_amount - i.amount_paid) > 0 |
|
ORDER BY i.id DESC |
|
""") |
|
invoices = cursor.fetchall() |
|
conn.close() |
|
|
|
form_data = { |
|
"invoice_id": invoice_id, |
|
"payment_method": payment_method, |
|
"payment_currency": payment_currency, |
|
"payment_amount": payment_amount, |
|
"cad_value_at_payment": cad_value_at_payment, |
|
"reference": reference, |
|
"sender_name": sender_name, |
|
"txid": txid, |
|
"wallet_address": wallet_address, |
|
"notes": notes, |
|
} |
|
|
|
return render_template( |
|
"payments/new.html", |
|
invoices=invoices, |
|
errors=errors, |
|
form_data=form_data, |
|
) |
|
|
|
client_id = invoice_row["client_id"] |
|
|
|
insert_cursor = conn.cursor() |
|
insert_cursor.execute(""" |
|
INSERT INTO payments |
|
( |
|
invoice_id, |
|
client_id, |
|
payment_method, |
|
payment_currency, |
|
payment_amount, |
|
cad_value_at_payment, |
|
reference, |
|
sender_name, |
|
txid, |
|
wallet_address, |
|
payment_status, |
|
received_at, |
|
notes |
|
) |
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'confirmed', UTC_TIMESTAMP(), %s) |
|
""", ( |
|
invoice_id, |
|
client_id, |
|
payment_method, |
|
payment_currency, |
|
payment_amount, |
|
cad_value_at_payment, |
|
reference or None, |
|
sender_name or None, |
|
txid or None, |
|
wallet_address or None, |
|
notes or None |
|
)) |
|
|
|
conn.commit() |
|
conn.close() |
|
|
|
recalc_invoice_totals(invoice_id) |
|
|
|
return redirect("/payments") |
|
|
|
cursor.execute(""" |
|
SELECT |
|
i.id, |
|
i.invoice_number, |
|
i.currency_code, |
|
i.total_amount, |
|
i.amount_paid, |
|
i.status, |
|
c.client_code, |
|
c.company_name |
|
FROM invoices i |
|
JOIN clients c ON i.client_id = c.id |
|
WHERE i.status IN ('pending', 'partial', 'overdue') |
|
AND (i.total_amount - i.amount_paid) > 0 |
|
ORDER BY i.id DESC |
|
""") |
|
invoices = cursor.fetchall() |
|
conn.close() |
|
|
|
return render_template( |
|
"payments/new.html", |
|
invoices=invoices, |
|
errors=[], |
|
form_data={}, |
|
) |
|
|
|
|
|
|
|
@app.route("/payments/void/<int:payment_id>", methods=["POST"]) |
|
def void_payment(payment_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute(""" |
|
SELECT id, invoice_id, payment_status |
|
FROM payments |
|
WHERE id = %s |
|
""", (payment_id,)) |
|
payment = cursor.fetchone() |
|
|
|
if not payment: |
|
conn.close() |
|
return "Payment not found", 404 |
|
|
|
if payment["payment_status"] != "confirmed": |
|
conn.close() |
|
return redirect("/payments") |
|
|
|
update_cursor = conn.cursor() |
|
update_cursor.execute(""" |
|
UPDATE payments |
|
SET payment_status = 'reversed' |
|
WHERE id = %s |
|
""", (payment_id,)) |
|
|
|
conn.commit() |
|
conn.close() |
|
|
|
recalc_invoice_totals(payment["invoice_id"]) |
|
|
|
return redirect("/payments") |
|
|
|
recalc_invoice_totals(payment["invoice_id"]) |
|
|
|
return redirect("/payments") |
|
|
|
@app.route("/payments/edit/<int:payment_id>", methods=["GET", "POST"]) |
|
def edit_payment(payment_id): |
|
conn = get_db_connection() |
|
cursor = conn.cursor(dictionary=True) |
|
|
|
cursor.execute(""" |
|
SELECT |
|
p.*, |
|
i.invoice_number, |
|
c.client_code, |
|
c.company_name |
|
FROM payments p |
|
JOIN invoices i ON p.invoice_id = i.id |
|
JOIN clients c ON p.client_id = c.id |
|
WHERE p.id = %s |
|
""", (payment_id,)) |
|
payment = cursor.fetchone() |
|
|
|
if not payment: |
|
conn.close() |
|
return "Payment not found", 404 |
|
|
|
if request.method == "POST": |
|
payment_method = request.form.get("payment_method", "").strip() |
|
payment_currency = request.form.get("payment_currency", "").strip() |
|
payment_amount = request.form.get("payment_amount", "").strip() |
|
cad_value_at_payment = request.form.get("cad_value_at_payment", "").strip() |
|
reference = request.form.get("reference", "").strip() |
|
sender_name = request.form.get("sender_name", "").strip() |
|
txid = request.form.get("txid", "").strip() |
|
wallet_address = request.form.get("wallet_address", "").strip() |
|
notes = request.form.get("notes", "").strip() |
|
|
|
errors = [] |
|
|
|
if not payment_method: |
|
errors.append("Payment method is required.") |
|
if not payment_currency: |
|
errors.append("Payment currency is required.") |
|
if not payment_amount: |
|
errors.append("Payment amount is required.") |
|
if not cad_value_at_payment: |
|
errors.append("CAD value at payment is required.") |
|
|
|
if not errors: |
|
try: |
|
amount_value = float(payment_amount) |
|
if amount_value <= 0: |
|
errors.append("Payment amount must be greater than zero.") |
|
except ValueError: |
|
errors.append("Payment amount must be a valid number.") |
|
|
|
try: |
|
cad_value = float(cad_value_at_payment) |
|
if cad_value < 0: |
|
errors.append("CAD value at payment cannot be negative.") |
|
except ValueError: |
|
errors.append("CAD value at payment must be a valid number.") |
|
|
|
if errors: |
|
payment["payment_method"] = payment_method or payment["payment_method"] |
|
payment["payment_currency"] = payment_currency or payment["payment_currency"] |
|
payment["payment_amount"] = payment_amount or payment["payment_amount"] |
|
payment["cad_value_at_payment"] = cad_value_at_payment or payment["cad_value_at_payment"] |
|
payment["reference"] = reference |
|
payment["sender_name"] = sender_name |
|
payment["txid"] = txid |
|
payment["wallet_address"] = wallet_address |
|
payment["notes"] = notes |
|
conn.close() |
|
return render_template("payments/edit.html", payment=payment, errors=errors) |
|
|
|
update_cursor = conn.cursor() |
|
update_cursor.execute(""" |
|
UPDATE payments |
|
SET payment_method = %s, |
|
payment_currency = %s, |
|
payment_amount = %s, |
|
cad_value_at_payment = %s, |
|
reference = %s, |
|
sender_name = %s, |
|
txid = %s, |
|
wallet_address = %s, |
|
notes = %s |
|
WHERE id = %s |
|
""", ( |
|
payment_method, |
|
payment_currency, |
|
payment_amount, |
|
cad_value_at_payment, |
|
reference or None, |
|
sender_name or None, |
|
txid or None, |
|
wallet_address or None, |
|
notes or None, |
|
payment_id |
|
)) |
|
conn.commit() |
|
invoice_id = payment["invoice_id"] |
|
conn.close() |
|
|
|
recalc_invoice_totals(invoice_id) |
|
|
|
return redirect("/payments") |
|
|
|
conn.close() |
|
return render_template("payments/edit.html", payment=payment, errors=[]) |
|
|
|
if __name__ == "__main__": |
|
app.run(host="0.0.0.0", port=5050, debug=True)
|
|
|