72 lines
3.5 KiB
Python
72 lines
3.5 KiB
Python
|
"""Pricing catalog CRUD operations."""
|
||
|
|
import json, sys, os
|
||
|
|
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 list_catalogs():
|
||
|
|
conn = get_db()
|
||
|
|
rows = conn.execute("SELECT * FROM pricing_catalogs ORDER BY created_at DESC").fetchall()
|
||
|
|
conn.close()
|
||
|
|
result = rows_to_list(rows)
|
||
|
|
for r in result:
|
||
|
|
r["labor_rates"] = json.loads(r["labor_rates"]) if r["labor_rates"] else {}
|
||
|
|
r["material_markups"] = json.loads(r["material_markups"]) if r["material_markups"] else {}
|
||
|
|
return {"catalogs": result}
|
||
|
|
|
||
|
|
def create_catalog(name, labor_rates, material_markups=None):
|
||
|
|
cid = new_id()
|
||
|
|
conn = get_db()
|
||
|
|
conn.execute("INSERT INTO pricing_catalogs (id, name, labor_rates, material_markups, created_at) VALUES (?,?,?,?,?)",
|
||
|
|
(cid, name, json.dumps(labor_rates), json.dumps(material_markups or {}), now()))
|
||
|
|
conn.commit(); conn.close()
|
||
|
|
return {"id": cid, "name": name}
|
||
|
|
|
||
|
|
def get_catalog(catalog_id):
|
||
|
|
conn = get_db()
|
||
|
|
row = conn.execute("SELECT * FROM pricing_catalogs WHERE id=?", (catalog_id,)).fetchone()
|
||
|
|
conn.close()
|
||
|
|
if not row: return {"error": "not found"}
|
||
|
|
r = row_to_dict(row)
|
||
|
|
r["labor_rates"] = json.loads(r["labor_rates"]) if r["labor_rates"] else {}
|
||
|
|
r["material_markups"] = json.loads(r["material_markups"]) if r["material_markups"] else {}
|
||
|
|
return r
|
||
|
|
|
||
|
|
def update_catalog(catalog_id, name=None, labor_rates=None, material_markups=None):
|
||
|
|
conn = get_db()
|
||
|
|
if name: conn.execute("UPDATE pricing_catalogs SET name=? WHERE id=?", (name, catalog_id))
|
||
|
|
if labor_rates: conn.execute("UPDATE pricing_catalogs SET labor_rates=? WHERE id=?", (json.dumps(labor_rates), catalog_id))
|
||
|
|
if material_markups: conn.execute("UPDATE pricing_catalogs SET material_markups=? WHERE id=?", (json.dumps(material_markups), catalog_id))
|
||
|
|
conn.commit(); conn.close()
|
||
|
|
return {"ok": True, "id": catalog_id}
|
||
|
|
|
||
|
|
def delete_catalog(catalog_id):
|
||
|
|
conn = get_db()
|
||
|
|
conn.execute("DELETE FROM pricing_catalogs WHERE id=?", (catalog_id,))
|
||
|
|
conn.commit(); conn.close()
|
||
|
|
return {"ok": True, "deleted": catalog_id}
|
||
|
|
|
||
|
|
def seed_default():
|
||
|
|
"""Seed default Kellow Construction pricing catalog."""
|
||
|
|
existing = list_catalogs()
|
||
|
|
if existing["catalogs"]: return {"ok": True, "msg": "already seeded"}
|
||
|
|
return create_catalog("Kellow Default", {
|
||
|
|
"general_labor": 55, "framing": 85, "electrical": 95,
|
||
|
|
"plumbing": 90, "painting": 65, "drywall": 75,
|
||
|
|
"tile": 80, "flooring": 70, "roofing": 85,
|
||
|
|
"concrete": 75, "demolition": 60, "finish_carpentry": 90,
|
||
|
|
"project_management": 95, "design": 110
|
||
|
|
}, {"lumber": 1.15, "drywall": 1.20, "tile": 1.25,
|
||
|
|
"plumbing_fixtures": 1.20, "electrical_fixtures": 1.15,
|
||
|
|
"paint": 1.10, "hardware": 1.15, "appliances": 1.10})
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
from lib.db import init_db; init_db()
|
||
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "list"
|
||
|
|
if cmd == "list": print(json.dumps(list_catalogs(), indent=2))
|
||
|
|
elif cmd == "create": print(json.dumps(create_catalog(sys.argv[2], json.loads(sys.argv[3]), json.loads(sys.argv[4]) if len(sys.argv) > 4 else None)))
|
||
|
|
elif cmd == "get": print(json.dumps(get_catalog(sys.argv[2]), indent=2))
|
||
|
|
elif cmd == "update": print(json.dumps(update_catalog(sys.argv[2], labor_rates=json.loads(sys.argv[3]) if len(sys.argv) > 3 else None)))
|
||
|
|
elif cmd == "delete": print(json.dumps(delete_catalog(sys.argv[2])))
|
||
|
|
elif cmd == "seed": print(json.dumps(seed_default()))
|
||
|
|
else: print(json.dumps({"error": f"unknown command: {cmd}"}))
|