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.
134 lines
3.6 KiB
134 lines
3.6 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 |
|
import os |
|
|
|
app = Flask( |
|
__name__, |
|
template_folder="../templates", |
|
static_folder="../static", |
|
) |
|
|
|
LOCAL_TZ = ZoneInfo("America/Toronto") |
|
|
|
# load version |
|
def load_version(): |
|
try: |
|
with open("../VERSION") as f: |
|
return f.read().strip() |
|
except: |
|
return "unknown" |
|
|
|
APP_VERSION = load_version() |
|
|
|
@app.context_processor |
|
def inject_version(): |
|
return dict(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() |
|
|
|
@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>" |
|
|
|
# rest of routes remain unchanged
|
|
|