82 lines
4.3 KiB
Python
82 lines
4.3 KiB
Python
"""Invoice management: create, track, sync to JobTread + QBO."""
|
|
import json, sys, os
|
|
from datetime import datetime, timedelta
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
from lib.db import get_db, new_id, now, row_to_dict, rows_to_list
|
|
|
|
def create(estimate_id, due_date=None):
|
|
conn = get_db()
|
|
est = row_to_dict(conn.execute("SELECT * FROM estimates WHERE id=?", (estimate_id,)).fetchone())
|
|
if not est: conn.close(); return {"error": "estimate not found"}
|
|
# Generate invoice number
|
|
count = conn.execute("SELECT COUNT(*) as c FROM invoices").fetchone()["c"]
|
|
inv_num = f"KC-{count + 1001}"
|
|
if not due_date:
|
|
due_date = (datetime.now() + timedelta(days=30)).strftime("%Y-%m-%d")
|
|
inv_id = new_id()
|
|
conn.execute("""INSERT INTO invoices (id, estimate_id, job_id, invoice_number, amount_due, due_date, status, created_at)
|
|
VALUES (?,?,?,?,?,?,?,?)""",
|
|
(inv_id, estimate_id, est["job_id"], inv_num, est["total_cost"], due_date, "draft", now()))
|
|
conn.commit(); conn.close()
|
|
return {"id": inv_id, "invoice_number": inv_num, "amount_due": est["total_cost"], "due_date": due_date, "status": "draft"}
|
|
|
|
def list_invoices(job_id=None):
|
|
conn = get_db()
|
|
if job_id:
|
|
rows = conn.execute("SELECT * FROM invoices WHERE job_id=? ORDER BY created_at DESC", (job_id,)).fetchall()
|
|
else:
|
|
rows = conn.execute("SELECT * FROM invoices ORDER BY created_at DESC").fetchall()
|
|
conn.close()
|
|
return {"invoices": rows_to_list(rows)}
|
|
|
|
def get(invoice_id):
|
|
conn = get_db()
|
|
inv = row_to_dict(conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone())
|
|
if not inv: conn.close(); return {"error": "invoice not found"}
|
|
est = row_to_dict(conn.execute("SELECT * FROM estimates WHERE id=?", (inv["estimate_id"],)).fetchone())
|
|
if est:
|
|
inv["line_items"] = rows_to_list(conn.execute("SELECT * FROM line_items WHERE estimate_id=?", (inv["estimate_id"],)).fetchall())
|
|
job = row_to_dict(conn.execute("SELECT * FROM jobs WHERE id=?", (inv["job_id"],)).fetchone())
|
|
inv["job"] = job
|
|
conn.close()
|
|
return inv
|
|
|
|
def mark_paid(invoice_id, amount_paid=None):
|
|
conn = get_db()
|
|
inv = row_to_dict(conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone())
|
|
if not inv: conn.close(); return {"error": "invoice not found"}
|
|
paid = amount_paid if amount_paid else inv["amount_due"]
|
|
conn.execute("UPDATE invoices SET amount_paid=?, status='paid' WHERE id=?", (paid, invoice_id))
|
|
conn.commit(); conn.close()
|
|
return {"ok": True, "id": invoice_id, "amount_paid": paid, "status": "paid"}
|
|
|
|
def sync_to_qbo(invoice_id):
|
|
"""Sync invoice to QuickBooks Online."""
|
|
conn = get_db()
|
|
inv = row_to_dict(conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone())
|
|
if not inv: conn.close(); return {"error": "invoice not found"}
|
|
items = rows_to_list(conn.execute("SELECT * FROM line_items WHERE estimate_id=?", (inv["estimate_id"],)).fetchall())
|
|
conn.close()
|
|
try:
|
|
from lib.qbo import create_invoice as qbo_create
|
|
qbo_lines = [{"description": i["description"], "amount": i["total"], "qty": i["quantity"]} for i in items]
|
|
# Note: customer_id mapping needed — for now use placeholder
|
|
result = qbo_create("1", qbo_lines, inv["due_date"])
|
|
qbo_id = result.get("Invoice", {}).get("Id")
|
|
conn = get_db()
|
|
conn.execute("UPDATE invoices SET qbo_invoice_id=?, synced_to_qbo_at=? WHERE id=?", (qbo_id, now(), invoice_id))
|
|
conn.commit(); conn.close()
|
|
return {"ok": True, "qbo_invoice_id": qbo_id}
|
|
except Exception as e:
|
|
return {"error": str(e), "msg": "QBO sync failed — check token status"}
|
|
|
|
if __name__ == "__main__":
|
|
from lib.db import init_db; init_db()
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "list"
|
|
if cmd == "create": print(json.dumps(create(sys.argv[2], sys.argv[3] if len(sys.argv)>3 else None)))
|
|
elif cmd == "list": print(json.dumps(list_invoices(sys.argv[2] if len(sys.argv)>2 else None), indent=2))
|
|
elif cmd == "get": print(json.dumps(get(sys.argv[2]), indent=2))
|
|
elif cmd == "mark-paid": print(json.dumps(mark_paid(sys.argv[2], float(sys.argv[3]) if len(sys.argv)>3 else None)))
|
|
elif cmd == "sync-qbo": print(json.dumps(sync_to_qbo(sys.argv[2])))
|
|
else: print(json.dumps({"error": f"unknown: {cmd}"}))
|