from datetime import datetime, timedelta import os import secrets from fastapi import FastAPI, HTTPException, UploadFile, File, Response, Query from fastapi.responses import FileResponse from starlette.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, EmailStr from passlib.context import CryptContext import mysql.connector from mysql.connector import pooling from typing import Optional from dotenv import load_dotenv import smtplib from email.message import EmailMessage # Load environment variables from .env file load_dotenv() # Database configuration from environment variables DB_HOST = os.getenv("DB_HOST", "localhost") DB_PORT = int(os.getenv("DB_PORT", "3306")) DB_USER = os.getenv("DB_USER", "bullseye_user") DB_PASSWORD = os.getenv("DB_PASSWORD", "4#zqJOSeownWs86c") DB_NAME = os.getenv("DB_NAME", "SouthernDirectionsql4") pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") app = FastAPI() # File uploads storage UPLOAD_DIR = os.getenv("UPLOAD_DIR", os.path.join(os.getcwd(), "uploads")) os.makedirs(UPLOAD_DIR, exist_ok=True) app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads") app.add_middleware( CORSMiddleware, allow_origins=[ "https://bullseye.mysouth.io", "http://localhost:40773", "http://127.0.0.1:40773", ], allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["*"], ) # Serve uploaded files UPLOAD_DIR = os.getenv("UPLOAD_DIR", os.path.join(os.getcwd(), "uploads")) os.makedirs(UPLOAD_DIR, exist_ok=True) app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads") # Create a connection pool (unique pool name per process to avoid reload conflicts) _pool_suffix = str(os.getpid()) connection_pool = pooling.MySQLConnectionPool( pool_name=f"be_pool_{_pool_suffix}", pool_size=int(os.getenv("DB_POOL_SIZE", "15")), host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, autocommit=True, ) def get_connection(): try: return connection_pool.get_connection() except mysql.connector.errors.PoolError: # Fallback to a direct connection if pool is briefly exhausted return mysql.connector.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, autocommit=True, ) def _blank_to_none(value): if isinstance(value, str): v = value.strip() return v if v != "" else None return value def _get_table_columns(table_name: str) -> set: """Return set of column names for the given table. Empty set if table missing.""" conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute(f"SHOW COLUMNS FROM {table_name}") rows = cur.fetchall() or [] return {r.get('Field') for r in rows if isinstance(r, dict)} except Exception: return set() finally: conn.close() @app.get("/items/line-item-rug-mount") def list_items_line_item_rug_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, name, notes FROM line_item_rug_mount WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-rug-mount") def create_line_item_rug_mount(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_rug_mount (category, name, notes, is_active) VALUES (%s, %s, %s, %s) """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-rug-mount/{record_id}") def update_line_item_rug_mount(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_rug_mount SET category=%s, name=%s, notes=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Rug Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-rug-mount/{record_id}") def delete_line_item_rug_mount(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_rug_mount WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Rug Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-misc-products") def list_items_line_item_misc_products(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, section, item, `option`, internal_note, notes, price_usd FROM line_item_misc_products WHERE is_active = 1 ORDER BY section, item, `option` """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-misc-products") def create_line_item_misc_products(payload: dict): required = ["section", "item"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_misc_products (section, item, `option`, internal_note, notes, price_zar, price_usd, price_eur, price_bwp, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( payload.get("section"), payload.get("item"), _blank_to_none(payload.get("option")), _blank_to_none(payload.get("internal_note")), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-misc-products/{record_id}") def update_line_item_misc_products(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_misc_products SET section=%s, item=%s, `option`=%s, internal_note=%s, notes=%s, price_zar=%s, price_usd=%s, price_eur=%s, price_bwp=%s, is_active=%s WHERE id=%s """, ( payload.get("section"), payload.get("item"), _blank_to_none(payload.get("option")), _blank_to_none(payload.get("internal_note")), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Misc products line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-misc-products/{record_id}") def delete_line_item_misc_products(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_misc_products WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Misc products line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-half-mount") def list_items_line_item_half_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, name, notes FROM line_item_half_mount WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-half-mount") def create_line_item_half_mount(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_half_mount (category, name, notes, is_active) VALUES (%s, %s, %s, %s) """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-half-mount/{record_id}") def update_line_item_half_mount(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_half_mount SET category=%s, name=%s, notes=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Half Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-half-mount/{record_id}") def delete_line_item_half_mount(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_half_mount WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Half Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-african-map-mount") def list_items_line_item_african_map_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, name, notes, price_zar, price_usd, price_eur, price_bwp, is_active FROM line_item_african_map_mount WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-african-map-mount") def create_line_item_african_map_mount(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_african_map_mount (category, name, notes, price_zar, price_usd, price_eur, price_bwp, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-african-map-mount/{record_id}") def update_line_item_african_map_mount(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_african_map_mount SET category=%s, name=%s, notes=%s, price_zar=%s, price_usd=%s, price_eur=%s, price_bwp=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="African Map line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-african-map-mount/{record_id}") def delete_line_item_african_map_mount(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_african_map_mount WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="African Map line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-skulls") def list_items_line_item_skulls(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, name, surcharge_pct, notes FROM line_item_skulls WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-skulls") def create_line_item_skulls(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_skulls (category, name, surcharge_pct, notes, is_active) VALUES (%s, %s, %s, %s, %s) """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("surcharge_pct")), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-skulls/{record_id}") def update_line_item_skulls(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_skulls SET category=%s, name=%s, surcharge_pct=%s, notes=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("surcharge_pct")), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Skulls line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-skulls/{record_id}") def delete_line_item_skulls(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_skulls WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Skulls line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-pedestal-bases") def list_items_line_item_pedestal_bases(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, item, category, internal_note_applicable_to, price_usd FROM line_item_pedestal_bases WHERE is_active = 1 ORDER BY category, item """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-pedestal-bases") def create_line_item_pedestal_bases(payload: dict): required = ["item", "category"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_pedestal_bases (item, category, internal_note_applicable_to, price_usd, price_eur, price_zar, price_bwp, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """, ( payload.get("item"), payload.get("category"), _blank_to_none(payload.get("internal_note_applicable_to")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-pedestal-bases/{record_id}") def update_line_item_pedestal_bases(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_pedestal_bases SET item=%s, category=%s, internal_note_applicable_to=%s, price_usd=%s, price_eur=%s, price_zar=%s, price_bwp=%s, is_active=%s WHERE id=%s """, ( payload.get("item"), payload.get("category"), _blank_to_none(payload.get("internal_note_applicable_to")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Pedestal bases line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-pedestal-bases/{record_id}") def delete_line_item_pedestal_bases(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_pedestal_bases WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Pedestal bases line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-pedestal-shoulder-mount") def list_items_line_item_pedestal_shoulder_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, name, surcharge_pct, notes FROM line_item_pedestal_shoulder_mount WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-pedestal-shoulder-mount") def create_line_item_pedestal_shoulder_mount(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_pedestal_shoulder_mount (category, name, surcharge_pct, notes, is_active) VALUES (%s, %s, %s, %s, %s) """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("surcharge_pct")), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-pedestal-shoulder-mount/{record_id}") def update_line_item_pedestal_shoulder_mount(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_pedestal_shoulder_mount SET category=%s, name=%s, surcharge_pct=%s, notes=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("surcharge_pct")), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Pedestal Shoulder Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-pedestal-shoulder-mount/{record_id}") def delete_line_item_pedestal_shoulder_mount(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_pedestal_shoulder_mount WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Pedestal Shoulder Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-wall-pedestal-shoulder-mount") def list_items_line_item_wall_pedestal_shoulder_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, name, notes FROM line_item_wall_pedestal_shoulder_mount WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-wall-pedestal-shoulder-mount") def create_line_item_wall_pedestal_shoulder_mount(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_wall_pedestal_shoulder_mount (category, name, notes, is_active) VALUES (%s, %s, %s, %s) """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-wall-pedestal-shoulder-mount/{record_id}") def update_line_item_wall_pedestal_shoulder_mount(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_wall_pedestal_shoulder_mount SET category=%s, name=%s, notes=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Wall Pedestal Shoulder Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-wall-pedestal-shoulder-mount/{record_id}") def delete_line_item_wall_pedestal_shoulder_mount(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_wall_pedestal_shoulder_mount WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Wall Pedestal Shoulder Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-shoulder-mount") def list_items_line_item_shoulder_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, name, notes FROM line_item_shoulder_mount WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-shoulder-mount") def create_line_item_shoulder_mount(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_shoulder_mount (category, name, notes, is_active) VALUES (%s, %s, %s, %s) """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-shoulder-mount/{record_id}") def update_line_item_shoulder_mount(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_shoulder_mount SET category=%s, name=%s, notes=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), payload.get("name"), _blank_to_none(payload.get("notes")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Shoulder Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-shoulder-mount/{record_id}") def delete_line_item_shoulder_mount(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_shoulder_mount WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Shoulder Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-full-mount") def list_items_line_item_full_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, base_type, name, surcharge_pct, notes, price_usd FROM line_item_full_mount WHERE is_active = 1 ORDER BY category, base_type, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-full-mount") def create_line_item_full_mount(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_full_mount (category, base_type, name, surcharge_pct, notes, price_zar, price_usd, price_eur, price_bwp, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( payload.get("category"), _blank_to_none(payload.get("base_type")), payload.get("name"), _blank_to_none(payload.get("surcharge_pct")), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-full-mount/{record_id}") def update_line_item_full_mount(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_full_mount SET category=%s, base_type=%s, name=%s, surcharge_pct=%s, notes=%s, price_zar=%s, price_usd=%s, price_eur=%s, price_bwp=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), _blank_to_none(payload.get("base_type")), payload.get("name"), _blank_to_none(payload.get("surcharge_pct")), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Full Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-full-mount/{record_id}") def delete_line_item_full_mount(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_full_mount WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Full Mount line item not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/line-item-skins") def list_items_line_item_skins(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, skin_type, tanning_type, name, notes, add_on_price_usd FROM line_item_skins WHERE is_active = 1 ORDER BY category, name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-skins") def create_line_item_skins(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_skins (category, skin_type, tanning_type, name, notes, add_on_price_zar, add_on_price_usd, add_on_price_eur, add_on_price_bwp, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( payload.get("category"), _blank_to_none(payload.get("skin_type")), _blank_to_none(payload.get("tanning_type")), payload.get("name"), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("add_on_price_zar")), _blank_to_none(payload.get("add_on_price_usd")), _blank_to_none(payload.get("add_on_price_eur")), _blank_to_none(payload.get("add_on_price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-skins/{record_id}") def update_line_item_skins(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_skins SET category=%s, skin_type=%s, tanning_type=%s, name=%s, notes=%s, add_on_price_zar=%s, add_on_price_usd=%s, add_on_price_eur=%s, add_on_price_bwp=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), _blank_to_none(payload.get("skin_type")), _blank_to_none(payload.get("tanning_type")), payload.get("name"), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("add_on_price_zar")), _blank_to_none(payload.get("add_on_price_usd")), _blank_to_none(payload.get("add_on_price_eur")), _blank_to_none(payload.get("add_on_price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Skins line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-skins/{record_id}") def delete_line_item_skins(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_skins WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Skins line item not found") conn.commit() return {"success": True} finally: conn.close() def ensure_schema(): conn = get_connection() try: cur = conn.cursor() # Ensure company_info table exists try: cur.execute("SHOW TABLES LIKE 'company_info'") if not cur.fetchone(): cur.execute( """ CREATE TABLE company_info ( id INT AUTO_INCREMENT PRIMARY KEY, company_name VARCHAR(160) NOT NULL, company_registration_number VARCHAR(64) NULL, company_vat_number VARCHAR(64) NULL, veterinary_approval_number VARCHAR(64) NULL, company_physical_address TEXT NULL, company_postal_address TEXT NULL, company_email VARCHAR(160) NULL, company_phone_number VARCHAR(64) NULL, company_mobile_number VARCHAR(64) NULL, company_whatsapp_number VARCHAR(64) NULL, company_website VARCHAR(160) NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) except Exception: # Ignore if DB variant differs slightly; table may already exist pass # Ensure banking_details table exists try: cur.execute("SHOW TABLES LIKE 'banking_details'") if not cur.fetchone(): cur.execute( """ CREATE TABLE banking_details ( id INT AUTO_INCREMENT PRIMARY KEY, bank_name VARCHAR(160) NOT NULL, account_name VARCHAR(160) NULL, account_number VARCHAR(64) NULL, account_type VARCHAR(64) NULL, branch_code VARCHAR(64) NULL, swift_code VARCHAR(64) NULL, iban VARCHAR(64) NULL, bank_address TEXT NULL, notes TEXT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) except Exception: pass # Ensure banking_details required columns exist (idempotent) try: cur2 = conn.cursor(dictionary=True) cur2.execute("SHOW COLUMNS FROM banking_details") cols = {r.get('Field') for r in (cur2.fetchall() or [])} alters = [] if 'bank_name' not in cols: alters.append("ADD COLUMN bank_name VARCHAR(160) NOT NULL") if 'account_name' not in cols: alters.append("ADD COLUMN account_name VARCHAR(160) NULL") if 'account_number' not in cols: alters.append("ADD COLUMN account_number VARCHAR(64) NULL") if 'account_type' not in cols: alters.append("ADD COLUMN account_type VARCHAR(64) NULL") if 'branch_code' not in cols: alters.append("ADD COLUMN branch_code VARCHAR(64) NULL") if 'swift_code' not in cols: alters.append("ADD COLUMN swift_code VARCHAR(64) NULL") if 'iban' not in cols: alters.append("ADD COLUMN iban VARCHAR(64) NULL") if 'bank_address' not in cols: alters.append("ADD COLUMN bank_address TEXT NULL") if 'notes' not in cols: alters.append("ADD COLUMN notes TEXT NULL") if alters: cur.execute("ALTER TABLE banking_details " + ", ".join(alters)) except Exception: pass # Ensure banking_terms table exists try: cur.execute("SHOW TABLES LIKE 'banking_terms'") if not cur.fetchone(): cur.execute( """ CREATE TABLE banking_terms ( id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(160) NOT NULL, term_text TEXT NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) except Exception: pass # Ensure banking_terms required columns exist (idempotent) try: cur2 = conn.cursor(dictionary=True) cur2.execute("SHOW COLUMNS FROM banking_terms") cols = {r.get('Field') for r in (cur2.fetchall() or [])} alters = [] if 'title' not in cols: alters.append("ADD COLUMN title VARCHAR(160) NOT NULL") if 'term_text' not in cols: alters.append("ADD COLUMN term_text TEXT NOT NULL") if alters: cur.execute("ALTER TABLE banking_terms " + ", ".join(alters)) except Exception: pass # Ensure master_animal table exists try: cur.execute("SHOW TABLES LIKE 'master_animal'") if not cur.fetchone(): cur.execute( """ CREATE TABLE master_animal ( id INT AUTO_INCREMENT PRIMARY KEY, common_name VARCHAR(100) NOT NULL UNIQUE, dip_pack TINYINT(1) NOT NULL DEFAULT 0, tanning_of_skins TINYINT(1) NOT NULL DEFAULT 0, felting_of_skins_add_on TINYINT(1) NOT NULL DEFAULT 0, full_mount TINYINT(1) NOT NULL DEFAULT 0, shoulder_mount TINYINT(1) NOT NULL DEFAULT 0, wall_pedestal_shoulder_mount TINYINT(1) NOT NULL DEFAULT 0, pedestal_shoulder_mount TINYINT(1) NOT NULL DEFAULT 0, skull TINYINT(1) NOT NULL DEFAULT 0, rug_mount TINYINT(1) NOT NULL DEFAULT 0, africa_map_mount TINYINT(1) NOT NULL DEFAULT 0, half_mount TINYINT(1) NOT NULL DEFAULT 0, open_mouth_add_on TINYINT(1) NOT NULL DEFAULT 0, miscellaneous_manufactured_items TINYINT(1) NOT NULL DEFAULT 0 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) except Exception: pass # Ensure master_animal_pricing table exists try: cur.execute("SHOW TABLES LIKE 'master_animal_pricing'") if not cur.fetchone(): cur.execute( """ CREATE TABLE master_animal_pricing ( id INT AUTO_INCREMENT PRIMARY KEY, common_name VARCHAR(120) NOT NULL UNIQUE, shoulder_mount_usd VARCHAR(64) NULL, shoulder_mount_eur VARCHAR(64) NULL, shoulder_mount_pula VARCHAR(64) NULL, shoulder_mount_zar VARCHAR(64) NULL, wall_pedestal_shoulder_mount_usd VARCHAR(64) NULL, wall_pedestal_shoulder_mount_eur VARCHAR(64) NULL, wall_pedestal_shoulder_mount_pula VARCHAR(64) NULL, wall_pedestal_shoulder_mount_zar VARCHAR(64) NULL, pedestal_shoulder_mount_usd VARCHAR(64) NULL, pedestal_shoulder_mount_eur VARCHAR(64) NULL, pedestal_shoulder_mount_pula VARCHAR(64) NULL, pedestal_shoulder_mount_zar VARCHAR(64) NULL, full_mount_usd VARCHAR(64) NULL, full_mount_eur VARCHAR(64) NULL, full_mount_pula VARCHAR(64) NULL, full_mount_zar VARCHAR(64) NULL, full_mount_note VARCHAR(160) NULL, skull_usd VARCHAR(64) NULL, skull_eur VARCHAR(64) NULL, skull_pula VARCHAR(64) NULL, skull_zar VARCHAR(64) NULL, rug_mount_usd VARCHAR(64) NULL, rug_mount_eur VARCHAR(64) NULL, rug_mount_pula VARCHAR(64) NULL, rug_mount_zar VARCHAR(64) NULL, tanning_of_skins_usd VARCHAR(64) NULL, tanning_of_skins_eur VARCHAR(64) NULL, tanning_of_skins_pula VARCHAR(64) NULL, tanning_of_skins_zar VARCHAR(64) NULL, felting_of_skins_add_on_usd VARCHAR(64) NULL, felting_of_skins_add_on_eur VARCHAR(64) NULL, felting_of_skins_add_on_pula VARCHAR(64) NULL, felting_of_skins_add_on_zar VARCHAR(64) NULL, africa_map_mount_usd VARCHAR(64) NULL, africa_map_mount_eur VARCHAR(64) NULL, africa_map_mount_pula VARCHAR(64) NULL, africa_map_mount_zar VARCHAR(64) NULL, half_mount_usd VARCHAR(64) NULL, half_mount_eur VARCHAR(64) NULL, half_mount_pula VARCHAR(64) NULL, half_mount_zar VARCHAR(64) NULL, dip_pack_price_usd VARCHAR(64) NULL, open_mouth_add_on_usd VARCHAR(64) NULL, open_mouth_add_on_eur VARCHAR(64) NULL, open_mouth_add_on_pula VARCHAR(64) NULL, open_mouth_add_on_zar VARCHAR(64) NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) except Exception: pass # Ensure customers table exists (MariaDB) cur.execute("SHOW TABLES LIKE 'customers'") if not cur.fetchone(): cur.execute( """ CREATE TABLE customers ( id INT AUTO_INCREMENT PRIMARY KEY, customer_unique_number VARCHAR(50) GENERATED ALWAYS AS (CONCAT('CUST-', LPAD(id, 6, '0'))) STORED UNIQUE, status ENUM('Active','Inactive','On-Hold') NOT NULL DEFAULT 'Active', customer_type ENUM('Export','Local') NOT NULL, first_names VARCHAR(255) NOT NULL, surname VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL UNIQUE, contact_number VARCHAR(50) NOT NULL, whatsapp_number VARCHAR(50) DEFAULT NULL, physical_country VARCHAR(100) DEFAULT NULL, physical_street VARCHAR(255) DEFAULT NULL, physical_province_state VARCHAR(100) DEFAULT NULL, physical_city VARCHAR(100) DEFAULT NULL, physical_postal_code VARCHAR(20) DEFAULT NULL, ship_country VARCHAR(100) DEFAULT NULL, ship_street VARCHAR(255) DEFAULT NULL, ship_province_state VARCHAR(100) DEFAULT NULL, ship_city VARCHAR(100) DEFAULT NULL, ship_postal_code VARCHAR(20) DEFAULT NULL, referred_by VARCHAR(255) DEFAULT NULL, preferred_currency ENUM('USD','EUR','BWP','ZAR') DEFAULT 'USD', citizenship ENUM('South African Citizen','Foreign National') NOT NULL, rsa_id VARCHAR(50) DEFAULT NULL, passport_number VARCHAR(50) DEFAULT NULL, documents TEXT DEFAULT NULL, office_notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, CONSTRAINT chk_citizenship_id CHECK ( (citizenship = 'South African Citizen' AND rsa_id IS NOT NULL AND passport_number IS NULL) OR (citizenship = 'Foreign National' AND passport_number IS NOT NULL AND rsa_id IS NULL) ) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Remove legacy triggers and enforce generated column for unique number try: cur.execute("DROP TRIGGER IF EXISTS customers_set_unique_number") except Exception: pass # Try to convert customer_unique_number to generated column (if not already) try: cur.execute( """ ALTER TABLE customers MODIFY customer_unique_number VARCHAR(50) GENERATED ALWAYS AS (CONCAT('CUST-', LPAD(id, 6, '0'))) STORED UNIQUE """ ) except Exception: # Likely already generated or DB variant without support; ignore pass # Ensure customer_documents table exists (single check) try: cur.execute("SHOW TABLES LIKE 'customer_documents'") if not cur.fetchone(): cur.execute( """ CREATE TABLE customer_documents ( id INT AUTO_INCREMENT PRIMARY KEY, customer_id INT NOT NULL, file_path VARCHAR(512) NOT NULL, file_name VARCHAR(255) DEFAULT NULL, mime_type VARCHAR(100) DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, deleted_at TIMESTAMP NULL DEFAULT NULL, INDEX idx_cust_docs_customer_id (customer_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) except Exception: pass # Ensure partners table exists cur.execute("SHOW TABLES LIKE 'partners'") if not cur.fetchone(): cur.execute( """ CREATE TABLE partners ( id INT AUTO_INCREMENT PRIMARY KEY, partner_unique_number VARCHAR(50) UNIQUE, status ENUM('Active','Inactive','On-Hold') NOT NULL DEFAULT 'Active', partner_type ENUM('Professional Hunter','Outfitter','Professional Hunter & Outfitter','Shipping') NOT NULL, company_name VARCHAR(255) NOT NULL, company_vat VARCHAR(100) DEFAULT NULL, company_email VARCHAR(255) NOT NULL, company_contact_number VARCHAR(50) NOT NULL, company_country VARCHAR(100) NOT NULL, company_street VARCHAR(255) NOT NULL, company_province VARCHAR(100) NOT NULL, company_city VARCHAR(100) NOT NULL, company_postal_code VARCHAR(20) NOT NULL, first_name VARCHAR(100) NOT NULL, surname VARCHAR(100) NOT NULL, email VARCHAR(255) NOT NULL, contact_number VARCHAR(50) NOT NULL, hunting_outfitter_id VARCHAR(100) DEFAULT NULL, comms_local_percent DECIMAL(5,2) DEFAULT NULL, comms_export_percent DECIMAL(5,2) DEFAULT NULL, outfitter_documents TEXT DEFAULT NULL, professional_hunter_id VARCHAR(100) DEFAULT NULL, professional_hunter_permit_type ENUM('Standard','Dangerous Game','Standard & Dangerous Game') DEFAULT NULL, professional_hunter_permit_issue_date DATE DEFAULT NULL, professional_hunter_permit_expiry_date DATE DEFAULT NULL, professional_hunter_documents TEXT DEFAULT NULL, office_notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Seed dummy partners if empty try: cur.execute("SELECT COUNT(*) FROM partners") count = cur.fetchone()[0] if count == 0: samples = [ ( 'Active', 'Outfitter', 'Savanna Safaris', 'VAT123', 'info@savanna.example', '012-345-6789', 'South Africa', '12 Main Rd', 'Limpopo', 'Polokwane', '0700', 'Peter', 'Masilela', 'peter@savanna.example', '082-000-1111', 'OUT-001', 10.0, 15.0, None, None, None, None, None, 'Trusted outfitter' ), ( 'Active', 'Professional Hunter', 'Kudu Plains', None, 'office@kuduplains.example', '011-222-3333', 'South Africa', '45 Bushveld Ave', 'North West', 'Rustenburg', '0299', 'Lerato', 'Khoza', 'lerato@kuduplains.example', '083-111-2222', None, None, None, None, 'PH-0099', 'Dangerous Game', '2024-01-10', '2026-01-09', 'Experienced PH' ), ( 'On-Hold', 'Shipping', 'WildShip Logistics', None, 'contact@wildship.example', '021-555-8888', 'South Africa', '78 Dock St', 'Western Cape', 'Cape Town', '8001', 'Aisha', 'Khan', 'aisha@wildship.example', '084-333-4444', None, None, None, None, None, None, None, None, 'Pending review' ), ] cur.executemany( """ INSERT INTO partners ( status, partner_type, company_name, company_vat, company_email, company_contact_number, company_country, company_street, company_province, company_city, company_postal_code, first_name, surname, email, contact_number, hunting_outfitter_id, comms_local_percent, comms_export_percent, outfitter_documents, professional_hunter_id, professional_hunter_permit_type, professional_hunter_permit_issue_date, professional_hunter_permit_expiry_date, professional_hunter_documents, office_notes ) VALUES ( %s,%s,%s,%s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s ) """, samples, ) except Exception: pass # Remove legacy trigger and enforce generated partner_unique_number try: cur.execute("DROP TRIGGER IF EXISTS partners_set_unique_number") except Exception: pass # Ensure partner_documents table exists try: cur.execute("SHOW TABLES LIKE 'partner_documents'") if not cur.fetchone(): cur.execute( """ CREATE TABLE partner_documents ( id INT AUTO_INCREMENT PRIMARY KEY, partner_id INT NOT NULL, file_path VARCHAR(512) NOT NULL, file_name VARCHAR(255) DEFAULT NULL, mime_type VARCHAR(100) DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, deleted_at TIMESTAMP NULL DEFAULT NULL, INDEX idx_partner_docs_partner_id (partner_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) except Exception: pass try: cur.execute( """ ALTER TABLE partners MODIFY partner_unique_number VARCHAR(50) GENERATED ALWAYS AS (CONCAT('PART-', LPAD(id, 6, '0'))) STORED UNIQUE """ ) except Exception: pass # Ensure quotes tables exist cur.execute("SHOW TABLES LIKE 'quotes'") if not cur.fetchone(): cur.execute( """ CREATE TABLE quotes ( id INT AUTO_INCREMENT PRIMARY KEY, quote_unique_number VARCHAR(50) UNIQUE, partner_id INT DEFAULT NULL, partner_name VARCHAR(255) DEFAULT NULL, order_no VARCHAR(50) NOT NULL, quote_date DATE NOT NULL, order_type ENUM('Local','Export') NOT NULL, customer_id INT DEFAULT NULL, customer_first_names VARCHAR(255) NOT NULL, customer_surname VARCHAR(255) NOT NULL, customer_email VARCHAR(255) NOT NULL, customer_contact_number VARCHAR(50) NOT NULL, customer_country VARCHAR(100) NOT NULL, customer_street VARCHAR(255) NOT NULL, customer_province_state VARCHAR(100) DEFAULT NULL, customer_city VARCHAR(100) NOT NULL, customer_postal_code VARCHAR(20) NOT NULL, company_name VARCHAR(255) NOT NULL, company_physical_address VARCHAR(255) NOT NULL, company_phone_number VARCHAR(50) NOT NULL, company_email VARCHAR(255) NOT NULL, company_vat_number VARCHAR(100) DEFAULT NULL, company_registration_number VARCHAR(100) DEFAULT NULL, company_logo_url VARCHAR(500) DEFAULT NULL, quote_status ENUM('Draft - In Progress','Draft - For Customer Review','Draft - Customer Requested Changes','Draft - Pending Office Approval','Rejected','Accepted - Deposit Due','Accepted - Deposit Received','Accepted - Converted to Order') NOT NULL DEFAULT 'Draft - In Progress', quote_currency ENUM('USD','EUR','BWP','ZAR') NOT NULL DEFAULT 'USD', sub_total DECIMAL(12,2) NOT NULL DEFAULT 0.00, vat_amount DECIMAL(12,2) NOT NULL DEFAULT 0.00, total_amount DECIMAL(12,2) NOT NULL DEFAULT 0.00, deposit_percent DECIMAL(5,2) DEFAULT NULL, deposit_amount_due DECIMAL(12,2) DEFAULT NULL, terms_and_conditions TEXT DEFAULT NULL, payment_details TEXT DEFAULT NULL, office_notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, deleted_at TIMESTAMP NULL DEFAULT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Remove legacy trigger and ensure quote_unique_number is a normal column (not generated) try: cur.execute("DROP TRIGGER IF EXISTS quotes_set_unique_number") except Exception: pass try: cur.execute( """ ALTER TABLE quotes MODIFY quote_unique_number VARCHAR(50) UNIQUE """ ) except Exception: pass # Backfill any missing quote_unique_number values in Q-0001 style try: # Determine current max sequence used in Q-XXXX pattern cur.execute( """ SELECT COALESCE(MAX(CAST(SUBSTRING(quote_unique_number, 3) AS UNSIGNED)), 0) FROM quotes WHERE quote_unique_number REGEXP '^Q-[0-9]+' """ ) row = cur.fetchone() max_seq = int(row[0] if row and row[0] is not None else 0) # Fetch rows that need backfilling cur.execute( """ SELECT id FROM quotes WHERE quote_unique_number IS NULL OR quote_unique_number = '' ORDER BY id """ ) missing = [r[0] for r in (cur.fetchall() or [])] seq = max_seq for quote_id_to_fill in missing: seq += 1 qno = f"Q-{seq:04d}" cur.execute( "UPDATE quotes SET quote_unique_number = %s WHERE id = %s", (qno, quote_id_to_fill), ) except Exception: pass # Ensure soft-delete column exists on quotes try: cur.execute( """ ALTER TABLE quotes ADD COLUMN deleted_at TIMESTAMP NULL DEFAULT NULL """ ) except Exception: pass # Line items table cur.execute("SHOW TABLES LIKE 'quote_line_items'") if not cur.fetchone(): cur.execute( """ CREATE TABLE quote_line_items ( id INT AUTO_INCREMENT PRIMARY KEY, quote_id INT NOT NULL, item_quantity INT NOT NULL DEFAULT 1, item_type ENUM('Animal','Bird','Administrative') NOT NULL, species VARCHAR(255) DEFAULT NULL, unit_price DECIMAL(12,2) NOT NULL DEFAULT 0.00, total_price DECIMAL(12,2) NOT NULL DEFAULT 0.00, detailed_instructions TEXT DEFAULT NULL, line_item_status ENUM('Draft','Approved','Cancelled') NOT NULL DEFAULT 'Draft', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_quote_id (quote_id), CONSTRAINT fk_quote_line_items_quote FOREIGN KEY (quote_id) REFERENCES quotes(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Processing selections table cur.execute("SHOW TABLES LIKE 'quote_line_item_processes'") if not cur.fetchone(): cur.execute( """ CREATE TABLE quote_line_item_processes ( id INT AUTO_INCREMENT PRIMARY KEY, line_item_id INT NOT NULL, processing_type ENUM('Dip & Pack','Skins','Full Mount','Shoulder Mount','Wall Pedestal Shoulder Mount','Pedestal Shoulder Mount','Skull Mount','Rug Mount','Africa Map Mount','Half Mount','Standalone Bases','Manufactured Product') NOT NULL, processing_subtype VARCHAR(255) DEFAULT NULL, INDEX idx_line_item_id (line_item_id), CONSTRAINT fk_qli_process_item FOREIGN KEY (line_item_id) REFERENCES quote_line_items(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Documents table cur.execute("SHOW TABLES LIKE 'quote_documents'") if not cur.fetchone(): cur.execute( """ CREATE TABLE quote_documents ( id INT AUTO_INCREMENT PRIMARY KEY, quote_id INT NOT NULL, file_url VARCHAR(500) NOT NULL, file_name VARCHAR(255) DEFAULT NULL, mime_type VARCHAR(100) DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_quote_doc_quote_id (quote_id), CONSTRAINT fk_quote_documents_quote FOREIGN KEY (quote_id) REFERENCES quotes(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # HS codes table cur.execute("SHOW TABLES LIKE 'quote_hs_codes'") if not cur.fetchone(): cur.execute( """ CREATE TABLE quote_hs_codes ( id INT AUTO_INCREMENT PRIMARY KEY, quote_id INT NOT NULL, hs_code VARCHAR(50) NOT NULL, INDEX idx_quote_hs_quote_id (quote_id), CONSTRAINT fk_quote_hs_codes_quote FOREIGN KEY (quote_id) REFERENCES quotes(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Ensure messages tables exist cur.execute("SHOW TABLES LIKE 'messages'") if not cur.fetchone(): cur.execute( """ CREATE TABLE messages ( id INT AUTO_INCREMENT PRIMARY KEY, message_unique_number VARCHAR(50) UNIQUE, customer_id INT NOT NULL, customer_name VARCHAR(255) NOT NULL, message_title VARCHAR(255) NOT NULL, message_status ENUM('Open','Awaiting Reply','Closed','Archived') NOT NULL DEFAULT 'Open', opened_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, last_message_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, assigned_staff_id INT DEFAULT NULL, assigned_staff_name VARCHAR(255) DEFAULT NULL, related_quote_id INT DEFAULT NULL, related_quote_number VARCHAR(50) DEFAULT NULL, message_priority ENUM('Normal','Urgent','Follow-Up') NOT NULL DEFAULT 'Normal', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_messages_customer_id (customer_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Remove legacy trigger and enforce generated message_unique_number try: cur.execute("DROP TRIGGER IF EXISTS messages_set_unique_number") except Exception: pass try: cur.execute( """ ALTER TABLE messages MODIFY message_unique_number VARCHAR(50) GENERATED ALWAYS AS (CONCAT('MSG-', LPAD(id, 6, '0'))) STORED UNIQUE """ ) except Exception: pass cur.execute("SHOW TABLES LIKE 'message_entries'") if not cur.fetchone(): cur.execute( """ CREATE TABLE message_entries ( id INT AUTO_INCREMENT PRIMARY KEY, message_id INT NOT NULL, sender_role ENUM('Customer','Staff') NOT NULL, sender_name VARCHAR(255) NOT NULL, message_header VARCHAR(255) DEFAULT NULL, message_content TEXT NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, INDEX idx_message_entries_message_id (message_id), CONSTRAINT fk_message_entries_message FOREIGN KEY (message_id) REFERENCES messages(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) cur.execute("SHOW TABLES LIKE 'message_attachments'") if not cur.fetchone(): cur.execute( """ CREATE TABLE message_attachments ( id INT AUTO_INCREMENT PRIMARY KEY, entry_id INT NOT NULL, file_url VARCHAR(500) NOT NULL, file_name VARCHAR(255) DEFAULT NULL, mime_type VARCHAR(100) DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_message_attachments_entry_id (entry_id), CONSTRAINT fk_message_attachments_entry FOREIGN KEY (entry_id) REFERENCES message_entries(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Ensure invoices tables exist cur.execute("SHOW TABLES LIKE 'invoices'") if not cur.fetchone(): cur.execute( """ CREATE TABLE invoices ( id INT AUTO_INCREMENT PRIMARY KEY, invoice_unique_number VARCHAR(50) UNIQUE, related_quote_id INT DEFAULT NULL, related_quote_number VARCHAR(50) DEFAULT NULL, partner_id INT DEFAULT NULL, partner_name VARCHAR(255) DEFAULT NULL, invoice_date DATE NOT NULL, order_no VARCHAR(50) DEFAULT NULL, order_type ENUM('Local','Export') NOT NULL, customer_id INT DEFAULT NULL, customer_first_names VARCHAR(255) NOT NULL, customer_surname VARCHAR(255) NOT NULL, customer_email VARCHAR(255) DEFAULT NULL, customer_contact_number VARCHAR(50) DEFAULT NULL, customer_country VARCHAR(100) NOT NULL, customer_street VARCHAR(255) DEFAULT NULL, customer_province_state VARCHAR(100) DEFAULT NULL, customer_city VARCHAR(100) DEFAULT NULL, customer_postal_code VARCHAR(20) DEFAULT NULL, company_name VARCHAR(255) NOT NULL, company_physical_address VARCHAR(255) NOT NULL, company_phone_number VARCHAR(50) NOT NULL, company_email VARCHAR(255) NOT NULL, company_vat_number VARCHAR(100) DEFAULT NULL, company_registration_number VARCHAR(100) DEFAULT NULL, company_logo_url VARCHAR(500) DEFAULT NULL, invoice_type ENUM('Pro-Forma Invoice','Tax Invoice') NOT NULL, payment_type ENUM('Deposit','Installment','Final') NOT NULL, invoice_status ENUM('Draft – In Progress','Issued','Partial Payment Received','Paid in Full','Withdrawn') NOT NULL DEFAULT 'Draft – In Progress', invoice_currency ENUM('USD','EUR','BWP','ZAR') NOT NULL, sub_total DECIMAL(12,2) NOT NULL DEFAULT 0.00, vat_amount DECIMAL(12,2) NOT NULL DEFAULT 0.00, total_amount DECIMAL(12,2) NOT NULL DEFAULT 0.00, deposit_percent DECIMAL(5,2) DEFAULT NULL, deposit_amount_due DECIMAL(12,2) DEFAULT NULL, terms_and_conditions TEXT DEFAULT NULL, payment_details TEXT DEFAULT NULL, office_notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Remove legacy trigger and enforce generated invoice_unique_number try: cur.execute("DROP TRIGGER IF EXISTS invoices_set_unique_number") except Exception: pass try: cur.execute( """ ALTER TABLE invoices MODIFY invoice_unique_number VARCHAR(50) GENERATED ALWAYS AS (CONCAT('INV-', LPAD(id, 6, '0'))) STORED UNIQUE """ ) except Exception: pass cur.execute("SHOW TABLES LIKE 'invoice_line_items'") if not cur.fetchone(): cur.execute( """ CREATE TABLE invoice_line_items ( id INT AUTO_INCREMENT PRIMARY KEY, invoice_id INT NOT NULL, item_quantity INT NOT NULL DEFAULT 1, item_type ENUM('Animal','Bird','Administrative') NOT NULL, species VARCHAR(255) DEFAULT NULL, processing_type ENUM('Dip & Pack','Skins','Full Mount','Shoulder Mount','Wall Pedestal Shoulder Mount','Pedestal Shoulder Mount','Skull Mount','Rug Mount','Africa Map Mount','Half Mount','Standalone Bases','Manufactured Product') DEFAULT NULL, processing_subtype VARCHAR(255) DEFAULT NULL, unit_price DECIMAL(12,2) NOT NULL DEFAULT 0.00, total_price DECIMAL(12,2) NOT NULL DEFAULT 0.00, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_invoice_id (invoice_id), CONSTRAINT fk_invoice_line_items_invoice FOREIGN KEY (invoice_id) REFERENCES invoices(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) cur.execute("SHOW TABLES LIKE 'invoice_hs_codes'") if not cur.fetchone(): cur.execute( """ CREATE TABLE invoice_hs_codes ( id INT AUTO_INCREMENT PRIMARY KEY, invoice_id INT NOT NULL, hs_code VARCHAR(50) NOT NULL, INDEX idx_invoice_hs_invoice_id (invoice_id), CONSTRAINT fk_invoice_hs_codes_invoice FOREIGN KEY (invoice_id) REFERENCES invoices(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) cur.execute("SHOW TABLES LIKE 'invoice_payments'") if not cur.fetchone(): cur.execute( """ CREATE TABLE invoice_payments ( id INT AUTO_INCREMENT PRIMARY KEY, invoice_id INT NOT NULL, payment_date DATE NOT NULL, amount DECIMAL(12,2) NOT NULL, notes VARCHAR(255) DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_invoice_payments_invoice_id (invoice_id), CONSTRAINT fk_invoice_payments_invoice FOREIGN KEY (invoice_id) REFERENCES invoices(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Ensure orders table exists cur.execute("SHOW TABLES LIKE 'orders'") if not cur.fetchone(): cur.execute( """ CREATE TABLE orders ( id INT AUTO_INCREMENT PRIMARY KEY, order_unique_number VARCHAR(50) UNIQUE, related_quote_id INT DEFAULT NULL, related_quote_number VARCHAR(50) DEFAULT NULL, related_invoice_id INT DEFAULT NULL, related_invoice_number VARCHAR(50) DEFAULT NULL, order_type ENUM('Export','Local') NOT NULL, processing_type ENUM('Fully Processed','Dip & Pack','Fully Processed + D&P','Other') NOT NULL, customer_id INT NOT NULL, customer_name VARCHAR(255) NOT NULL, order_delivery ENUM('Shipping Partner','Customer Collect') NOT NULL, deposit_received VARCHAR(100) DEFAULT 'No', instructions_finalised ENUM('Yes','No') NOT NULL DEFAULT 'No', order_paid_in_full VARCHAR(100) DEFAULT 'No', office_notes TEXT DEFAULT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_orders_customer_id (customer_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Remove legacy trigger and enforce generated order_unique_number try: cur.execute("DROP TRIGGER IF EXISTS orders_set_unique_number") except Exception: pass try: cur.execute( """ ALTER TABLE orders MODIFY order_unique_number VARCHAR(50) GENERATED ALWAYS AS (CONCAT('ORD-', LPAD(id, 6, '0'))) STORED UNIQUE """ ) except Exception: pass # Check if users table exists and has the right structure cur.execute("SHOW TABLES LIKE 'users'") if not cur.fetchone(): # Create users table if it doesn't exist cur.execute( """ CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(64) NOT NULL UNIQUE, email VARCHAR(255) DEFAULT NULL UNIQUE, password_hash VARCHAR(255) DEFAULT NULL, role ENUM('customer','partner','staff','admin','office') NOT NULL DEFAULT 'customer', customer_id INT DEFAULT NULL, partner_id INT DEFAULT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_login DATETIME DEFAULT NULL, password_reset_token VARCHAR(100) DEFAULT NULL, password_reset_expires DATETIME DEFAULT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) else: # Add missing columns if they don't exist try: cur.execute("ALTER TABLE users ADD COLUMN IF NOT EXISTS password_reset_token VARCHAR(100) DEFAULT NULL") except Exception: pass try: cur.execute("ALTER TABLE users ADD COLUMN IF NOT EXISTS password_reset_expires DATETIME DEFAULT NULL") except Exception: pass # Ensure password_hash allows NULL for accounts without an initial password try: cur.execute("ALTER TABLE users MODIFY COLUMN password_hash VARCHAR(255) NULL") except Exception: pass # Check if customer_orders table exists and has the right structure cur.execute("SHOW TABLES LIKE 'customer_orders'") if not cur.fetchone(): # Create customer_orders table if it doesn't exist cur.execute( """ CREATE TABLE customer_orders ( id INT AUTO_INCREMENT PRIMARY KEY, orderUniqueNumber VARCHAR(50) NOT NULL UNIQUE, relatedQuoteNumber VARCHAR(50), relatedInvoiceNumber VARCHAR(50), orderType ENUM('Export', 'Local') NOT NULL, processingType ENUM('Fully Processed', 'Dip & Pack', 'Fully Processed + D&P', 'Other') NOT NULL, customerName VARCHAR(255) NOT NULL, orderDelivery ENUM('Shipping Partner', 'Customer Collect') NOT NULL, depositReceived VARCHAR(100), instructionsFinalised ENUM('Yes', 'No') NOT NULL, orderPaidInFull VARCHAR(100), officeNotes TEXT, date DATE NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ ) # Insert sample data sample_orders = [ { 'orderUniqueNumber': 'SO-1001', 'relatedQuoteNumber': 'Q-501', 'relatedInvoiceNumber': 'INV-9001', 'orderType': 'Export', 'processingType': 'Fully Processed', 'customerName': 'John Doe', 'orderDelivery': 'Shipping Partner', 'depositReceived': 'Yes - 01-06-2024', 'instructionsFinalised': 'No', 'orderPaidInFull': 'No', 'officeNotes': 'Awaiting collection preference', 'date': '2024-05-01' }, { 'orderUniqueNumber': 'SO-1002', 'relatedQuoteNumber': 'Q-502', 'relatedInvoiceNumber': 'INV-9002', 'orderType': 'Local', 'processingType': 'Dip & Pack', 'customerName': 'Jane Smith', 'orderDelivery': 'Customer Collect', 'depositReceived': 'Yes - 15-06-2024', 'instructionsFinalised': 'Yes', 'orderPaidInFull': 'Yes - 30-06-2024', 'officeNotes': 'Paid and ready', 'date': '2024-06-10' }, { 'orderUniqueNumber': 'SO-1003', 'relatedQuoteNumber': 'Q-503', 'relatedInvoiceNumber': 'INV-9003', 'orderType': 'Export', 'processingType': 'Fully Processed + D&P', 'customerName': 'Robert Brown', 'orderDelivery': 'Shipping Partner', 'depositReceived': 'No', 'instructionsFinalised': 'No', 'orderPaidInFull': 'No', 'officeNotes': 'Pending deposit', 'date': '2024-06-20' } ] for order in sample_orders: cur.execute( """ INSERT INTO customer_orders ( orderUniqueNumber, relatedQuoteNumber, relatedInvoiceNumber, orderType, processingType, customerName, orderDelivery, depositReceived, instructionsFinalised, orderPaidInFull, officeNotes, date ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( order['orderUniqueNumber'], order['relatedQuoteNumber'], order['relatedInvoiceNumber'], order['orderType'], order['processingType'], order['customerName'], order['orderDelivery'], order['depositReceived'], order['instructionsFinalised'], order['orderPaidInFull'], order['officeNotes'], order['date'] ) ) finally: conn.close() def ensure_admin_user(): """Seed or update the default admin user in the database.""" conn = get_connection() try: cur = conn.cursor() admin_username = "admin" admin_email = "admin@localhost" admin_hash = pwd_context.hash("Admin@123") # If user exists, update its password/role; else insert cur.execute("SELECT id FROM users WHERE username = %s", (admin_username,)) row = cur.fetchone() if row: cur.execute( "UPDATE users SET password_hash=%s, role='admin', email=COALESCE(email, %s) WHERE username=%s", (admin_hash, admin_email, admin_username), ) else: cur.execute( "INSERT INTO users (username, email, password_hash, role) VALUES (%s, %s, %s, 'admin')", (admin_username, admin_email, admin_hash), ) finally: conn.close() class LoginRequest(BaseModel): identifier: str password: str class LoginResponse(BaseModel): message: str customer_id: int | None = None role: str | None = None class CustomerRecord(BaseModel): id: Optional[int] = None customer_unique_number: Optional[str] = None status: str customer_type: str first_names: str surname: str email: EmailStr contact_number: str whatsapp_number: Optional[str] = None physical_country: Optional[str] = None physical_street: Optional[str] = None physical_province_state: Optional[str] = None physical_city: Optional[str] = None physical_postal_code: Optional[str] = None ship_country: Optional[str] = None ship_street: Optional[str] = None ship_province_state: Optional[str] = None ship_city: Optional[str] = None ship_postal_code: Optional[str] = None referred_by: Optional[str] = None preferred_currency: Optional[str] = None citizenship: str rsa_id: Optional[str] = None passport_number: Optional[str] = None documents: Optional[str] = None office_notes: Optional[str] = None class ForgotPasswordRequest(BaseModel): email: EmailStr class ResetPasswordRequest(BaseModel): token: str new_password: str class OrderRecord(BaseModel): id: Optional[int] = None order_unique_number: Optional[str] = None related_quote_id: Optional[int] = None related_quote_number: Optional[str] = None related_invoice_id: Optional[int] = None related_invoice_number: Optional[str] = None order_type: str processing_type: str customer_id: int customer_name: str order_delivery: str deposit_received: Optional[str] = None instructions_finalised: Optional[str] = None order_paid_in_full: Optional[str] = None office_notes: Optional[str] = None created_at: Optional[str] = None updated_at: Optional[str] = None class PartnerRecord(BaseModel): id: Optional[int] = None partner_unique_number: Optional[str] = None status: str partner_type: str # Company company_name: str company_vat: Optional[str] = None company_email: EmailStr company_contact_number: str company_country: str company_street: str company_province: str company_city: str company_postal_code: str # Personal first_name: str surname: str email: EmailStr contact_number: str # Outfitter hunting_outfitter_id: Optional[str] = None comms_local_percent: Optional[float] = None comms_export_percent: Optional[float] = None outfitter_documents: Optional[str] = None # Professional Hunter professional_hunter_id: Optional[str] = None professional_hunter_permit_type: Optional[str] = None professional_hunter_permit_issue_date: Optional[str] = None professional_hunter_permit_expiry_date: Optional[str] = None professional_hunter_documents: Optional[str] = None # Notes office_notes: Optional[str] = None # Company Info model class CompanyInfoRecord(BaseModel): id: Optional[int] = None company_name: str company_registration_number: Optional[str] = None company_vat_number: Optional[str] = None veterinary_approval_number: Optional[str] = None company_physical_address: Optional[str] = None company_postal_address: Optional[str] = None company_email: Optional[EmailStr] = None company_phone_number: Optional[str] = None company_mobile_number: Optional[str] = None company_whatsapp_number: Optional[str] = None company_website: Optional[str] = None class BankingDetailsRecord(BaseModel): id: Optional[int] = None bank_name: str account_name: Optional[str] = None account_number: Optional[str] = None account_type: Optional[str] = None branch_code: Optional[str] = None swift_code: Optional[str] = None iban: Optional[str] = None bank_address: Optional[str] = None notes: Optional[str] = None class BankingTermRecord(BaseModel): id: Optional[int] = None title: str term_text: str class MasterAnimalRecord(BaseModel): id: Optional[int] = None common_name: str dip_pack: bool = False tanning_of_skins: bool = False felting_of_skins_add_on: bool = False full_mount: bool = False shoulder_mount: bool = False wall_pedestal_shoulder_mount: bool = False pedestal_shoulder_mount: bool = False skull: bool = False rug_mount: bool = False africa_map_mount: bool = False half_mount: bool = False open_mouth_add_on: bool = False miscellaneous_manufactured_items: bool = False class MasterAnimalPricingRecord(BaseModel): id: Optional[int] = None common_name: str shoulder_mount_usd: Optional[str] = None shoulder_mount_eur: Optional[str] = None shoulder_mount_pula: Optional[str] = None shoulder_mount_zar: Optional[str] = None wall_pedestal_shoulder_mount_usd: Optional[str] = None wall_pedestal_shoulder_mount_eur: Optional[str] = None wall_pedestal_shoulder_mount_pula: Optional[str] = None wall_pedestal_shoulder_mount_zar: Optional[str] = None pedestal_shoulder_mount_usd: Optional[str] = None pedestal_shoulder_mount_eur: Optional[str] = None pedestal_shoulder_mount_pula: Optional[str] = None pedestal_shoulder_mount_zar: Optional[str] = None full_mount_usd: Optional[str] = None full_mount_eur: Optional[str] = None full_mount_pula: Optional[str] = None full_mount_zar: Optional[str] = None full_mount_note: Optional[str] = None skull_usd: Optional[str] = None skull_eur: Optional[str] = None skull_pula: Optional[str] = None skull_zar: Optional[str] = None rug_mount_usd: Optional[str] = None rug_mount_eur: Optional[str] = None rug_mount_pula: Optional[str] = None rug_mount_zar: Optional[str] = None tanning_of_skins_usd: Optional[str] = None tanning_of_skins_eur: Optional[str] = None tanning_of_skins_pula: Optional[str] = None tanning_of_skins_zar: Optional[str] = None felting_of_skins_add_on_usd: Optional[str] = None felting_of_skins_add_on_eur: Optional[str] = None felting_of_skins_add_on_pula: Optional[str] = None felting_of_skins_add_on_zar: Optional[str] = None africa_map_mount_usd: Optional[str] = None africa_map_mount_eur: Optional[str] = None africa_map_mount_pula: Optional[str] = None africa_map_mount_zar: Optional[str] = None half_mount_usd: Optional[str] = None half_mount_eur: Optional[str] = None half_mount_pula: Optional[str] = None half_mount_zar: Optional[str] = None dip_pack_price_usd: Optional[str] = None open_mouth_add_on_usd: Optional[str] = None open_mouth_add_on_eur: Optional[str] = None open_mouth_add_on_pula: Optional[str] = None open_mouth_add_on_zar: Optional[str] = None class DipPackPricingRecord(BaseModel): id: Optional[int] = None animal_group: str composition: str notes: Optional[str] = None price_zar: Optional[str] = None price_usd: Optional[str] = None price_eur: Optional[str] = None price_bwp: Optional[str] = None hs_code: Optional[str] = None class EnumUpdateRequest(BaseModel): values: list[str] class DipPackPricingRecord(BaseModel): id: Optional[int] = None animal_group: str composition: str notes: Optional[str] = None price_zar: Optional[str] = None price_usd: Optional[str] = None price_eur: Optional[str] = None price_bwp: Optional[str] = None hs_code: Optional[str] = None # Quotes models class QuoteLineItemProcessRecord(BaseModel): processing_type: str processing_subtype: Optional[str] = None class QuoteLineItemRecord(BaseModel): item_quantity: int item_type: str species: Optional[str] = None unit_price: float total_price: float detailed_instructions: Optional[str] = None line_item_status: str = 'Draft' processes: list[QuoteLineItemProcessRecord] = [] class QuoteCreateRequest(BaseModel): # Partner / Order partner_id: Optional[int] = None partner_name: Optional[str] = None order_no: str quote_date: str # YYYY-MM-DD order_type: str # 'Local' | 'Export' # Customer snapshot customer_id: Optional[int] = None customer_first_names: str customer_surname: str customer_email: str customer_contact_number: str customer_country: str customer_street: str customer_province_state: Optional[str] = None customer_city: str customer_postal_code: str # Company snapshot company_name: str company_physical_address: str company_phone_number: str company_email: str company_vat_number: Optional[str] = None company_registration_number: Optional[str] = None company_logo_url: Optional[str] = None # Status & currency quote_status: Optional[str] = None quote_currency: str # Derived totals sub_total: Optional[float] = None vat_amount: Optional[float] = None total_amount: Optional[float] = None deposit_percent: Optional[float] = None deposit_amount_due: Optional[float] = None # T&Cs and payment terms_and_conditions: Optional[str] = None payment_details: Optional[str] = None office_notes: Optional[str] = None # Children line_items: list[QuoteLineItemRecord] = [] hs_codes: list[str] = [] @app.on_event("startup") def on_startup(): ensure_schema() ensure_admin_user() @app.get("/health") def health(): conn = None try: conn = get_connection() cur = conn.cursor() cur.execute("SELECT 1") cur.fetchone() return {"ok": True, "db": True} except Exception as e: return {"ok": True, "db": False, "error": str(e)} finally: try: if conn is not None: conn.close() except Exception: pass # Partners CRUD @app.get("/partners") def list_partners(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, partner_unique_number, status, partner_type, company_name, company_vat, company_email, company_contact_number, company_country, company_street, company_province, company_city, company_postal_code, first_name, surname, email, contact_number, hunting_outfitter_id, comms_local_percent, comms_export_percent, outfitter_documents, professional_hunter_id, professional_hunter_permit_type, professional_hunter_permit_issue_date, professional_hunter_permit_expiry_date, professional_hunter_documents, office_notes, created_at FROM partners ORDER BY created_at DESC """ ) return cur.fetchall() finally: conn.close() # Company Info CRUD @app.get("/company-info") def list_company_info(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, company_name, company_registration_number, company_vat_number, veterinary_approval_number, company_physical_address, company_postal_address, company_email, company_phone_number, company_mobile_number, company_whatsapp_number, company_website FROM company_info ORDER BY id DESC """ ) return cur.fetchall() finally: conn.close() @app.get("/company-info/{record_id}") def get_company_info(record_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, company_name, company_registration_number, company_vat_number, veterinary_approval_number, company_physical_address, company_postal_address, company_email, company_phone_number, company_mobile_number, company_whatsapp_number, company_website FROM company_info WHERE id = %s """, (record_id,), ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Record not found") return row finally: conn.close() @app.post("/company-info") def create_company_info(body: CompanyInfoRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO company_info ( company_name, company_registration_number, company_vat_number, veterinary_approval_number, company_physical_address, company_postal_address, company_email, company_phone_number, company_mobile_number, company_whatsapp_number, company_website ) VALUES ( %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s ) """, ( body.company_name, _blank_to_none(body.company_registration_number), _blank_to_none(body.company_vat_number), _blank_to_none(body.veterinary_approval_number), _blank_to_none(body.company_physical_address), _blank_to_none(body.company_postal_address), _blank_to_none(body.company_email), _blank_to_none(body.company_phone_number), _blank_to_none(body.company_mobile_number), _blank_to_none(body.company_whatsapp_number), _blank_to_none(body.company_website), ), ) new_id = cur.lastrowid conn.commit() return {"id": new_id} finally: conn.close() @app.put("/company-info/{record_id}") def update_company_info(record_id: int, body: CompanyInfoRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE company_info SET company_name=%s, company_registration_number=%s, company_vat_number=%s, veterinary_approval_number=%s, company_physical_address=%s, company_postal_address=%s, company_email=%s, company_phone_number=%s, company_mobile_number=%s, company_whatsapp_number=%s, company_website=%s WHERE id = %s """, ( body.company_name, _blank_to_none(body.company_registration_number), _blank_to_none(body.company_vat_number), _blank_to_none(body.veterinary_approval_number), _blank_to_none(body.company_physical_address), _blank_to_none(body.company_postal_address), _blank_to_none(body.company_email), _blank_to_none(body.company_phone_number), _blank_to_none(body.company_mobile_number), _blank_to_none(body.company_whatsapp_number), _blank_to_none(body.company_website), record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Record not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/company-info/{record_id}") def delete_company_info(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM company_info WHERE id = %s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Record not found") return {"success": True} finally: conn.close() # Banking Details CRUD @app.get("/banking-details") def list_banking_details(): conn = get_connection() try: cur = conn.cursor(dictionary=True) # Select all columns to be resilient if some columns are missing on older DBs cur.execute("SELECT * FROM banking_details ORDER BY id DESC") rows = cur.fetchall() or [] return rows finally: conn.close() @app.get("/banking-details/{record_id}") def get_banking_detail(record_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute("SELECT * FROM banking_details WHERE id = %s", (record_id,)) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Record not found") return row finally: conn.close() @app.post("/banking-details") def create_banking_detail(body: BankingDetailsRecord): conn = get_connection() try: cur = conn.cursor() # Build dynamic insert based on actual columns in DB cols = _get_table_columns('banking_details') allowed = [ ('bank_name', body.bank_name), ('account_name', _blank_to_none(body.account_name)), ('account_number', _blank_to_none(body.account_number)), ('account_type', _blank_to_none(body.account_type)), ('branch_code', _blank_to_none(body.branch_code)), ('swift_code', _blank_to_none(body.swift_code)), ('iban', _blank_to_none(body.iban)), ('bank_address', _blank_to_none(body.bank_address)), ('notes', _blank_to_none(body.notes)), ] insert_cols = [c for c, _ in allowed if c in cols] insert_vals = [v for c, v in allowed if c in cols] placeholders = ",".join(["%s"] * len(insert_vals)) cur.execute( f"INSERT INTO banking_details ({','.join(insert_cols)}) VALUES ({placeholders})", tuple(insert_vals), ) new_id = cur.lastrowid conn.commit() return {"id": new_id} finally: conn.close() @app.put("/banking-details/{record_id}") def update_banking_detail(record_id: int, body: BankingDetailsRecord): conn = get_connection() try: cur = conn.cursor() cols = _get_table_columns('banking_details') updates = [] params = [] if 'bank_name' in cols: updates.append('bank_name=%s'); params.append(body.bank_name) if 'account_name' in cols: updates.append('account_name=%s'); params.append(_blank_to_none(body.account_name)) if 'account_number' in cols: updates.append('account_number=%s'); params.append(_blank_to_none(body.account_number)) if 'account_type' in cols: updates.append('account_type=%s'); params.append(_blank_to_none(body.account_type)) if 'branch_code' in cols: updates.append('branch_code=%s'); params.append(_blank_to_none(body.branch_code)) if 'swift_code' in cols: updates.append('swift_code=%s'); params.append(_blank_to_none(body.swift_code)) if 'iban' in cols: updates.append('iban=%s'); params.append(_blank_to_none(body.iban)) if 'bank_address' in cols: updates.append('bank_address=%s'); params.append(_blank_to_none(body.bank_address)) if 'notes' in cols: updates.append('notes=%s'); params.append(_blank_to_none(body.notes)) if not updates: return {"success": True} params.append(record_id) cur.execute(f"UPDATE banking_details SET {', '.join(updates)} WHERE id = %s", tuple(params)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Record not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/banking-details/{record_id}") def delete_banking_detail(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM banking_details WHERE id = %s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Record not found") return {"success": True} finally: conn.close() # Banking Terms CRUD @app.get("/banking-terms") def list_banking_terms(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute("SELECT * FROM banking_terms ORDER BY id DESC") rows = cur.fetchall() or [] def normalize(row: dict) -> dict: # Compute normalized fields title = row.get('title') or None if not title: section_raw = row.get('section') or row.get('term_section') or row.get('heading') clause_code_raw = row.get('clause_code') or row.get('code') or row.get('name') or row.get('label') if section_raw and clause_code_raw: title = f"{section_raw} {clause_code_raw}".strip() else: title = clause_code_raw or section_raw or row.get('doc_context') or '' term_text = row.get('term_text') or None if term_text is None: term_text = row.get('clause_text') or row.get('text') or row.get('content') or row.get('value') or row.get('description') or '' note = row.get('note') if note: term_text = f"{term_text}\nNote: {note}" if term_text else f"Note: {note}" # Provide raw fields when present for richer clients (backward compatible) return { 'id': row.get('id'), 'title': title or '', 'term_text': term_text or '', 'doc_context': row.get('doc_context'), 'section': row.get('section') or row.get('term_section') or row.get('heading'), 'clause_code': row.get('clause_code') or row.get('code') or row.get('name') or row.get('label'), 'clause_text': row.get('clause_text') or row.get('text') or row.get('content') or row.get('value') or row.get('description'), 'note': row.get('note'), } return [normalize(r) for r in rows] finally: conn.close() @app.get("/banking-terms/{record_id}") def get_banking_term(record_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute("SELECT * FROM banking_terms WHERE id = %s", (record_id,)) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Record not found") # Normalize legacy columns to title + term_text section = row.get('section') or row.get('term_section') or row.get('heading') clause_code = row.get('clause_code') or row.get('code') or row.get('name') or row.get('label') title = row.get('title') or (f"{section} {clause_code}".strip() if section and clause_code else (clause_code or section or row.get('doc_context') or '')) term_text = row.get('term_text') or row.get('clause_text') or row.get('text') or row.get('content') or row.get('value') or row.get('description') or '' note = row.get('note') if note: term_text = f"{term_text}\nNote: {note}" if term_text else f"Note: {note}" return { 'id': row.get('id'), 'title': title or '', 'term_text': term_text or '' } finally: conn.close() @app.post("/banking-terms") def create_banking_term(body: BankingTermRecord): conn = get_connection() try: cur = conn.cursor() cols = _get_table_columns('banking_terms') allowed = [ ('title', body.title), ('term_text', body.term_text), ] insert_cols = [c for c, _ in allowed if c in cols] insert_vals = [v for c, v in allowed if c in cols] placeholders = ",".join(["%s"] * len(insert_vals)) cur.execute( f"INSERT INTO banking_terms ({','.join(insert_cols)}) VALUES ({placeholders})", tuple(insert_vals), ) new_id = cur.lastrowid conn.commit() return {"id": new_id} finally: conn.close() @app.put("/banking-terms/{record_id}") def update_banking_term(record_id: int, body: BankingTermRecord): conn = get_connection() try: cur = conn.cursor() cols = _get_table_columns('banking_terms') updates = [] params = [] if 'title' in cols: updates.append('title=%s'); params.append(body.title) if 'term_text' in cols: updates.append('term_text=%s'); params.append(body.term_text) if not updates: return {"success": True} params.append(record_id) cur.execute(f"UPDATE banking_terms SET {', '.join(updates)} WHERE id = %s", tuple(params)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Record not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/banking-terms/{record_id}") def delete_banking_term(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM banking_terms WHERE id = %s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Record not found") return {"success": True} finally: conn.close() # Master Animal CRUD @app.get("/master-animals") def list_master_animals(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, common_name, dip_pack, tanning_of_skins, felting_of_skins_add_on, full_mount, shoulder_mount, wall_pedestal_shoulder_mount, pedestal_shoulder_mount, skull, rug_mount, africa_map_mount, half_mount, open_mouth_add_on, miscellaneous_manufactured_items FROM master_animal ORDER BY common_name """ ) rows = cur.fetchall() or [] # Convert 0/1 to booleans for r in rows: for k in list(r.keys()): if k != 'id' and k != 'common_name': r[k] = bool(r[k]) return rows finally: conn.close() @app.post("/master-animals") def create_master_animal(body: MasterAnimalRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO master_animal ( common_name, dip_pack, tanning_of_skins, felting_of_skins_add_on, full_mount, shoulder_mount, wall_pedestal_shoulder_mount, pedestal_shoulder_mount, skull, rug_mount, africa_map_mount, half_mount, open_mouth_add_on, miscellaneous_manufactured_items ) VALUES ( %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s ) """, ( body.common_name.strip(), int(bool(body.dip_pack)), int(bool(body.tanning_of_skins)), int(bool(body.felting_of_skins_add_on)), int(bool(body.full_mount)), int(bool(body.shoulder_mount)), int(bool(body.wall_pedestal_shoulder_mount)), int(bool(body.pedestal_shoulder_mount)), int(bool(body.skull)), int(bool(body.rug_mount)), int(bool(body.africa_map_mount)), int(bool(body.half_mount)), int(bool(body.open_mouth_add_on)), int(bool(body.miscellaneous_manufactured_items)), ), ) new_id = cur.lastrowid conn.commit() return {"id": new_id} finally: conn.close() @app.put("/master-animals/{animal_id}") def update_master_animal(animal_id: int, body: MasterAnimalRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE master_animal SET common_name=%s, dip_pack=%s, tanning_of_skins=%s, felting_of_skins_add_on=%s, full_mount=%s, shoulder_mount=%s, wall_pedestal_shoulder_mount=%s, pedestal_shoulder_mount=%s, skull=%s, rug_mount=%s, africa_map_mount=%s, half_mount=%s, open_mouth_add_on=%s, miscellaneous_manufactured_items=%s WHERE id = %s """, ( body.common_name.strip(), int(bool(body.dip_pack)), int(bool(body.tanning_of_skins)), int(bool(body.felting_of_skins_add_on)), int(bool(body.full_mount)), int(bool(body.shoulder_mount)), int(bool(body.wall_pedestal_shoulder_mount)), int(bool(body.pedestal_shoulder_mount)), int(bool(body.skull)), int(bool(body.rug_mount)), int(bool(body.africa_map_mount)), int(bool(body.half_mount)), int(bool(body.open_mouth_add_on)), int(bool(body.miscellaneous_manufactured_items)), animal_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Animal not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/master-animals/{animal_id}") def delete_master_animal(animal_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM master_animal WHERE id = %s", (animal_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Animal not found") return {"success": True} finally: conn.close() @app.get("/master-animal-pricing") def list_master_animal_pricing(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT * FROM master_animal_pricing ORDER BY common_name """ ) return cur.fetchall() or [] finally: conn.close() @app.post("/master-animal-pricing") def create_master_animal_pricing(body: MasterAnimalPricingRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO master_animal_pricing ( common_name, shoulder_mount_usd, shoulder_mount_eur, shoulder_mount_pula, shoulder_mount_zar, wall_pedestal_shoulder_mount_usd, wall_pedestal_shoulder_mount_eur, wall_pedestal_shoulder_mount_pula, wall_pedestal_shoulder_mount_zar, pedestal_shoulder_mount_usd, pedestal_shoulder_mount_eur, pedestal_shoulder_mount_pula, pedestal_shoulder_mount_zar, full_mount_usd, full_mount_eur, full_mount_pula, full_mount_zar, full_mount_note, skull_usd, skull_eur, skull_pula, skull_zar, rug_mount_usd, rug_mount_eur, rug_mount_pula, rug_mount_zar, tanning_of_skins_usd, tanning_of_skins_eur, tanning_of_skins_pula, tanning_of_skins_zar, felting_of_skins_add_on_usd, felting_of_skins_add_on_eur, felting_of_skins_add_on_pula, felting_of_skins_add_on_zar, africa_map_mount_usd, africa_map_mount_eur, africa_map_mount_pula, africa_map_mount_zar, half_mount_usd, half_mount_eur, half_mount_pula, half_mount_zar, dip_pack_price_usd, open_mouth_add_on_usd, open_mouth_add_on_eur, open_mouth_add_on_pula, open_mouth_add_on_zar ) VALUES ( %s,%s,%s,%s,%s, %s,%s,%s,%s, %s,%s,%s,%s, %s,%s,%s,%s,%s, %s,%s,%s,%s, %s,%s,%s,%s, %s,%s,%s,%s, %s,%s,%s,%s, %s,%s,%s,%s, %s, %s,%s,%s,%s ) """, ( body.common_name, body.shoulder_mount_usd, body.shoulder_mount_eur, body.shoulder_mount_pula, body.shoulder_mount_zar, body.wall_pedestal_shoulder_mount_usd, body.wall_pedestal_shoulder_mount_eur, body.wall_pedestal_shoulder_mount_pula, body.wall_pedestal_shoulder_mount_zar, body.pedestal_shoulder_mount_usd, body.pedestal_shoulder_mount_eur, body.pedestal_shoulder_mount_pula, body.pedestal_shoulder_mount_zar, body.full_mount_usd, body.full_mount_eur, body.full_mount_pula, body.full_mount_zar, _blank_to_none(body.full_mount_note), body.skull_usd, body.skull_eur, body.skull_pula, body.skull_zar, body.rug_mount_usd, body.rug_mount_eur, body.rug_mount_pula, body.rug_mount_zar, body.tanning_of_skins_usd, body.tanning_of_skins_eur, body.tanning_of_skins_pula, body.tanning_of_skins_zar, body.felting_of_skins_add_on_usd, body.felting_of_skins_add_on_eur, body.felting_of_skins_add_on_pula, body.felting_of_skins_add_on_zar, body.africa_map_mount_usd, body.africa_map_mount_eur, body.africa_map_mount_pula, body.africa_map_mount_zar, body.half_mount_usd, body.half_mount_eur, body.half_mount_pula, body.half_mount_zar, body.dip_pack_price_usd, body.open_mouth_add_on_usd, body.open_mouth_add_on_eur, body.open_mouth_add_on_pula, body.open_mouth_add_on_zar, ), ) new_id = cur.lastrowid conn.commit() return {"id": new_id} finally: conn.close() @app.put("/master-animal-pricing/{pricing_id}") def update_master_animal_pricing(pricing_id: int, body: MasterAnimalPricingRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE master_animal_pricing SET common_name=%s, shoulder_mount_usd=%s, shoulder_mount_eur=%s, shoulder_mount_pula=%s, shoulder_mount_zar=%s, wall_pedestal_shoulder_mount_usd=%s, wall_pedestal_shoulder_mount_eur=%s, wall_pedestal_shoulder_mount_pula=%s, wall_pedestal_shoulder_mount_zar=%s, pedestal_shoulder_mount_usd=%s, pedestal_shoulder_mount_eur=%s, pedestal_shoulder_mount_pula=%s, pedestal_shoulder_mount_zar=%s, full_mount_usd=%s, full_mount_eur=%s, full_mount_pula=%s, full_mount_zar=%s, full_mount_note=%s, skull_usd=%s, skull_eur=%s, skull_pula=%s, skull_zar=%s, rug_mount_usd=%s, rug_mount_eur=%s, rug_mount_pula=%s, rug_mount_zar=%s, tanning_of_skins_usd=%s, tanning_of_skins_eur=%s, tanning_of_skins_pula=%s, tanning_of_skins_zar=%s, felting_of_skins_add_on_usd=%s, felting_of_skins_add_on_eur=%s, felting_of_skins_add_on_pula=%s, felting_of_skins_add_on_zar=%s, africa_map_mount_usd=%s, africa_map_mount_eur=%s, africa_map_mount_pula=%s, africa_map_mount_zar=%s, half_mount_usd=%s, half_mount_eur=%s, half_mount_pula=%s, half_mount_zar=%s, dip_pack_price_usd=%s, open_mouth_add_on_usd=%s, open_mouth_add_on_eur=%s, open_mouth_add_on_pula=%s, open_mouth_add_on_zar=%s WHERE id = %s """, ( body.common_name, body.shoulder_mount_usd, body.shoulder_mount_eur, body.shoulder_mount_pula, body.shoulder_mount_zar, body.wall_pedestal_shoulder_mount_usd, body.wall_pedestal_shoulder_mount_eur, body.wall_pedestal_shoulder_mount_pula, body.wall_pedestal_shoulder_mount_zar, body.pedestal_shoulder_mount_usd, body.pedestal_shoulder_mount_eur, body.pedestal_shoulder_mount_pula, body.pedestal_shoulder_mount_zar, body.full_mount_usd, body.full_mount_eur, body.full_mount_pula, body.full_mount_zar, _blank_to_none(body.full_mount_note), body.skull_usd, body.skull_eur, body.skull_pula, body.skull_zar, body.rug_mount_usd, body.rug_mount_eur, body.rug_mount_pula, body.rug_mount_zar, body.tanning_of_skins_usd, body.tanning_of_skins_eur, body.tanning_of_skins_pula, body.tanning_of_skins_zar, body.felting_of_skins_add_on_usd, body.felting_of_skins_add_on_eur, body.felting_of_skins_add_on_pula, body.felting_of_skins_add_on_zar, body.africa_map_mount_usd, body.africa_map_mount_eur, body.africa_map_mount_pula, body.africa_map_mount_zar, body.half_mount_usd, body.half_mount_eur, body.half_mount_pula, body.half_mount_zar, body.dip_pack_price_usd, body.open_mouth_add_on_usd, body.open_mouth_add_on_eur, body.open_mouth_add_on_pula, body.open_mouth_add_on_zar, pricing_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Pricing not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/master-animal-pricing/{pricing_id}") def delete_master_animal_pricing(pricing_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM master_animal_pricing WHERE id = %s", (pricing_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Pricing not found") return {"success": True} finally: conn.close() # Items (read-only lists) @app.get("/items/african-map-mount") def list_items_african_map_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, animal_common_name, frame, add_on, base_price_usd, add_on_price_usd, notes FROM african_map_mount ORDER BY animal_common_name """ ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/bird-species") def list_items_bird_species(): conn = get_connection() try: cur = conn.cursor(dictionary=True) # Prefer legacy table name if present: item_bird_species; otherwise fallback to bird_species try: cur.execute( """SELECT id, common_name, scientific_name, cites_status, tops_status FROM item_bird_species ORDER BY common_name""" ) return cur.fetchall() or [] except mysql.connector.Error as e1: if getattr(e1, 'errno', None) != 1146: # Not a missing table; re-raise raise # Fallback to bird_species table try: cur.execute( """SELECT id, common_name, scientific_name, cites_status, tops_status FROM bird_species ORDER BY common_name""" ) return cur.fetchall() or [] except mysql.connector.Error as e2: if getattr(e2, 'errno', None) == 1146: return [] raise finally: conn.close() @app.get("/items/dip-pack-pricing") def list_items_dip_pack_pricing(): conn = get_connection() try: cur = conn.cursor(dictionary=True) # Prefer new table dip_pack_pricing; fallback to legacy item_dip_pack if missing try: cur.execute( """SELECT id, animal_group, composition, notes, price_usd, hs_code FROM dip_pack_pricing ORDER BY animal_group, composition""" ) return cur.fetchall() or [] except mysql.connector.Error as e1: if getattr(e1, 'errno', None) != 1146: # Not missing table; bubble up raise # Fallback to legacy table name try: cur.execute( """SELECT id, animal_group, composition, notes, price_usd, hs_code FROM item_dip_pack ORDER BY animal_group, composition""" ) return cur.fetchall() or [] except mysql.connector.Error as e2: if getattr(e2, 'errno', None) == 1146: return [] raise finally: conn.close() @app.get("/items/line-item-dip-pack") def list_items_line_item_dip_pack(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """ SELECT id, category, size_type, name, feet_count, notes, price_zar, price_usd, price_eur, price_bwp, is_active FROM line_item_dip_pack WHERE is_active = 1 ORDER BY category, size_type, name """ ) except mysql.connector.Error as e: # If table does not exist, return empty list gracefully if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/line-item-dip-pack") def create_line_item_dip_pack(payload: dict): required = ["category", "name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO line_item_dip_pack (category, size_type, name, feet_count, notes, price_zar, price_usd, price_eur, price_bwp, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( payload.get("category"), _blank_to_none(payload.get("size_type")), payload.get("name"), _blank_to_none(payload.get("feet_count")), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/line-item-dip-pack/{record_id}") def update_line_item_dip_pack(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE line_item_dip_pack SET category=%s, size_type=%s, name=%s, feet_count=%s, notes=%s, price_zar=%s, price_usd=%s, price_eur=%s, price_bwp=%s, is_active=%s WHERE id=%s """, ( payload.get("category"), _blank_to_none(payload.get("size_type")), payload.get("name"), _blank_to_none(payload.get("feet_count")), _blank_to_none(payload.get("notes")), _blank_to_none(payload.get("price_zar")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("price_eur")), _blank_to_none(payload.get("price_bwp")), 1 if str(payload.get("is_active", "1")).strip() not in ("0", "false", "False") else 0, record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Dip & Pack line item not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/line-item-dip-pack/{record_id}") def delete_line_item_dip_pack(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM line_item_dip_pack WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Dip & Pack line item not found") conn.commit() return {"success": True} finally: conn.close() @app.post("/items/dip-pack-pricing") def create_items_dip_pack_pricing(body: DipPackPricingRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO dip_pack_pricing ( animal_group, composition, notes, price_zar, price_usd, price_eur, price_bwp, hs_code ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s) """, ( body.animal_group, body.composition, _blank_to_none(body.notes), body.price_zar or 'Not Applicable', body.price_usd or 'Not Applicable', body.price_eur or 'Not Applicable', body.price_bwp or 'Not Applicable', _blank_to_none(body.hs_code), ), ) new_id = cur.lastrowid conn.commit() return {"id": new_id} finally: conn.close() @app.put("/items/dip-pack-pricing/{record_id}") def update_items_dip_pack_pricing(record_id: int, body: DipPackPricingRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE dip_pack_pricing SET animal_group=%s, composition=%s, notes=%s, price_zar=%s, price_usd=%s, price_eur=%s, price_bwp=%s, hs_code=%s WHERE id=%s """, ( body.animal_group, body.composition, _blank_to_none(body.notes), body.price_zar or 'Not Applicable', body.price_usd or 'Not Applicable', body.price_eur or 'Not Applicable', body.price_bwp or 'Not Applicable', _blank_to_none(body.hs_code), record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Dip & Pack row not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/dip-pack-pricing/{record_id}") def delete_items_dip_pack_pricing(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM dip_pack_pricing WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Dip & Pack row not found") conn.commit() return {"success": True} finally: conn.close() SAFE_ENUMS: dict[tuple[str, str], tuple[str, str]] = { ("dip_pack_pricing", "animal_group"): ("dip_pack_pricing", "animal_group"), ("item_dip_pack", "animal_group"): ("item_dip_pack", "animal_group"), # Full Mount ("fullmount", "base_type"): ("fullmount", "base_type"), ("fullmount", "standard_pose"): ("fullmount", "standard_pose"), ("fullmount", "special_pose"): ("fullmount", "special_pose"), ("fullmount", "add_on"): ("fullmount", "add_on"), ("item_full_mount", "base_type"): ("item_full_mount", "base_type"), ("item_full_mount", "standard_pose"): ("item_full_mount", "standard_pose"), ("item_full_mount", "special_pose"): ("item_full_mount", "special_pose"), ("item_full_mount", "add_on"): ("item_full_mount", "add_on"), # African Map Mount ("african_map_mount", "frame"): ("african_map_mount", "frame"), ("african_map_mount", "add_on"): ("african_map_mount", "add_on"), ("item_african_map_mount", "frame"): ("item_african_map_mount", "frame"), ("item_african_map_mount", "add_on"): ("item_african_map_mount", "add_on"), # Bird Species ("bird_species", "cites_status"): ("bird_species", "cites_status"), ("bird_species", "tops_status"): ("bird_species", "tops_status"), ("item_bird_species", "cites_status"): ("item_bird_species", "cites_status"), ("item_bird_species", "tops_status"): ("item_bird_species", "tops_status"), # Half Mount ("half_mount", "variation"): ("half_mount", "variation"), ("half_mount", "add_on"): ("half_mount", "add_on"), ("item_half_mount", "variation"): ("item_half_mount", "variation"), ("item_half_mount", "add_on"): ("item_half_mount", "add_on"), # Misc Products ("misc_products", "sub_category"): ("misc_products", "sub_category"), ("item_misc_products", "sub_category"): ("item_misc_products", "sub_category"), # Rug Mount ("rug_mount", "type"): ("rug_mount", "type"), ("rug_mount", "add_on"): ("rug_mount", "add_on"), ("item_rug_mount", "type"): ("item_rug_mount", "type"), ("item_rug_mount", "add_on"): ("item_rug_mount", "add_on"), # Shoulder Mount ("shoulder_mount", "variation"): ("shoulder_mount", "variation"), ("shoulder_mount", "special_variation"): ("shoulder_mount", "special_variation"), ("shoulder_mount", "add_on"): ("shoulder_mount", "add_on"), ("item_shoulder_mount", "variation"): ("item_shoulder_mount", "variation"), ("item_shoulder_mount", "special_variation"): ("item_shoulder_mount", "special_variation"), ("item_shoulder_mount", "add_on"): ("item_shoulder_mount", "add_on"), # Skins ("skins", "skin_type"): ("skins", "skin_type"), ("skins", "tanning_type"): ("skins", "tanning_type"), ("skins", "leather_tan_color"): ("skins", "leather_tan_color"), ("skins", "add_on"): ("skins", "add_on"), ("item_skins", "skin_type"): ("item_skins", "skin_type"), ("item_skins", "tanning_type"): ("item_skins", "tanning_type"), ("item_skins", "leather_tan_color"): ("item_skins", "leather_tan_color"), ("item_skins", "add_on"): ("item_skins", "add_on"), # Skulls ("skulls", "skull_type"): ("skulls", "skull_type"), ("skulls", "add_on"): ("skulls", "add_on"), ("skulls", "surcharge_type"): ("skulls", "surcharge_type"), ("skulls", "special_base"): ("skulls", "special_base"), ("item_skulls", "skull_type"): ("item_skulls", "skull_type"), ("item_skulls", "add_on"): ("item_skulls", "add_on"), ("item_skulls", "surcharge_type"): ("item_skulls", "surcharge_type"), ("item_skulls", "special_base"): ("item_skulls", "special_base"), # Wall Pedestal Shoulder Mount ("wall_pedestal_shoulder_mount", "variation"): ("wall_pedestal_shoulder_mount", "variation"), ("wall_pedestal_shoulder_mount", "add_on"): ("wall_pedestal_shoulder_mount", "add_on"), ("item_wall_pedestal_shoulder_mount", "variation"): ("item_wall_pedestal_shoulder_mount", "variation"), ("item_wall_pedestal_shoulder_mount", "add_on"): ("item_wall_pedestal_shoulder_mount", "add_on"), } def _parse_enum_list(column_type: str) -> list[str]: if not column_type.lower().startswith("enum("): return [] inner = column_type[column_type.find('(')+1:column_type.rfind(')')] out: list[str] = [] current = '' in_quote = False i = 0 while i < len(inner): ch = inner[i] if ch == "'": if in_quote and i + 1 < len(inner) and inner[i+1] == "'": current += "'"; i += 1 else: in_quote = not in_quote elif ch == ',' and not in_quote: out.append(current) current = '' else: current += ch i += 1 if current: out.append(current) return [v for v in (s.strip() for s in out) if v != ''] def _quote_enum_values(values: list[str]) -> str: return ",".join(["'" + v.replace("'", "''") + "'" for v in values]) @app.get("/admin/enums/{table}/{column}") def get_enum_values(table: str, column: str): if (table, column) not in SAFE_ENUMS: raise HTTPException(status_code=400, detail="Unsupported enum column") target_table, target_column = SAFE_ENUMS[(table, column)] conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, (target_table, target_column), ) row = cur.fetchone() if not row: # Fallbacks: handle legacy/new table names if target_table == 'item_skins': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('skins', target_column), ) row = cur.fetchone() elif target_table == 'item_wall_pedestal_shoulder_mount': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('wall_pedestal_shoulder_mount', target_column), ) row = cur.fetchone() if not row: # Also support legacy skulls mapping if target_table == 'item_skulls': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('skulls', target_column), ) row = cur.fetchone() # And legacy fullmount mapping if not row and target_table == 'item_full_mount': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('fullmount', target_column), ) row = cur.fetchone() # And legacy african_map_mount mapping if not row and target_table == 'item_african_map_mount': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('african_map_mount', target_column), ) row = cur.fetchone() # And legacy half_mount mapping if not row and target_table == 'item_half_mount': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('half_mount', target_column), ) row = cur.fetchone() # And legacy bird_species mapping if not row and target_table == 'item_bird_species': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('bird_species', target_column), ) row = cur.fetchone() # And legacy rug_mount mapping if not row and target_table == 'item_rug_mount': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('rug_mount', target_column), ) row = cur.fetchone() # And legacy misc_products mapping if not row and target_table == 'item_misc_products': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('misc_products', target_column), ) row = cur.fetchone() # And legacy shoulder_mount mapping if not row and target_table == 'item_shoulder_mount': cur.execute( """ SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s """, ('shoulder_mount', target_column), ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Column not found") return { "values": _parse_enum_list(row["COLUMN_TYPE"] or ""), "nullable": row["IS_NULLABLE"] == 'YES', "default": row["COLUMN_DEFAULT"], } finally: conn.close() @app.put("/admin/enums/{table}/{column}") def update_enum_values(table: str, column: str, body: EnumUpdateRequest): if (table, column) not in SAFE_ENUMS: raise HTTPException(status_code=400, detail="Unsupported enum column") if not body.values: raise HTTPException(status_code=400, detail="Must include at least one value") target_table, target_column = SAFE_ENUMS[(table, column)] conn = get_connection() try: cur = conn.cursor() quoted = _quote_enum_values(body.values) sql = f"ALTER TABLE `{target_table}` MODIFY `{target_column}` ENUM({quoted}) NOT NULL" cur.execute(sql) conn.commit() return {"success": True} finally: conn.close() @app.get("/items/full-mount") def list_items_full_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, animal_common_name, base_type, standard_pose, special_pose, base_price_usd, add_on, add_on_price_usd, notes FROM fullmount ORDER BY animal_common_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/half-mount") def list_items_half_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, animal_common_name, variation, base_price_usd, add_on, add_on_price_usd, pricing_rule_note, notes FROM half_mount ORDER BY animal_common_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/misc-products") def list_items_misc_products(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, sub_category, product_name, option_name, price_usd, internal_note FROM misc_products ORDER BY sub_category, product_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/item-misc-products") def list_item_misc_products(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, sub_category, product_name, option_name, price_usd, internal_note FROM item_misc_products ORDER BY sub_category, product_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.post("/items/item-misc-products") def create_item_misc_product(payload: dict): required = ["sub_category", "product_name"] for k in required: if not (payload.get(k) or "").strip(): raise HTTPException(status_code=400, detail=f"Missing required field: {k}") conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO item_misc_products (sub_category, product_name, option_name, price_usd, internal_note) VALUES (%s, %s, %s, %s, %s) """, ( payload.get("sub_category"), payload.get("product_name"), _blank_to_none(payload.get("option_name")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("internal_note")), ), ) conn.commit() return {"id": cur.lastrowid} finally: conn.close() @app.put("/items/item-misc-products/{record_id}") def update_item_misc_product(record_id: int, payload: dict): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE item_misc_products SET sub_category=%s, product_name=%s, option_name=%s, price_usd=%s, internal_note=%s WHERE id=%s """, ( payload.get("sub_category"), payload.get("product_name"), _blank_to_none(payload.get("option_name")), _blank_to_none(payload.get("price_usd")), _blank_to_none(payload.get("internal_note")), record_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Misc product not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/items/item-misc-products/{record_id}") def delete_item_misc_product(record_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM item_misc_products WHERE id=%s", (record_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Misc product not found") conn.commit() return {"success": True} finally: conn.close() @app.get("/items/rug-mount") def list_items_rug_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, animal_common_name, type, base_price_usd, add_on, add_on_price_usd, notes FROM rug_mount ORDER BY animal_common_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/shoulder-mount") def list_items_shoulder_mount(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, animal_common_name, variation, special_variation, base_price_usd, add_on, add_on_price_usd, notes FROM shoulder_mount ORDER BY animal_common_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/skins") def list_items_skins(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, animal_common_name, skin_type, tanning_type, base_price_usd, add_on, add_on_price_usd, surcharge_note FROM skins ORDER BY animal_common_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: # Fallback: older/newer schema might use item_skins instead of skins cur.execute( """SELECT id, animal_common_name, skin_type, tanning_type, base_price_usd, add_on, add_on_price_usd, surcharge_note FROM item_skins ORDER BY animal_common_name""" ) return cur.fetchall() or [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/skulls") def list_items_skulls(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, animal_common_name, skull_type, base_price_usd, discount_percent, add_on, add_on_price_usd, surcharge_type, surcharge_percent, notes FROM skulls ORDER BY animal_common_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/items/wall-pedestal-shoulder-mount") def list_items_wpsm(): conn = get_connection() try: cur = conn.cursor(dictionary=True) try: cur.execute( """SELECT id, animal_common_name, variation, base_price_usd, add_on, add_on_price_usd, notes FROM wall_pedestal_shoulder_mount ORDER BY animal_common_name""" ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1146: return [] raise return cur.fetchall() or [] finally: conn.close() @app.get("/partners/{partner_id}") def get_partner(partner_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, partner_unique_number, status, partner_type, company_name, company_vat, company_email, company_contact_number, company_country, company_street, company_province, company_city, company_postal_code, first_name, surname, email, contact_number, hunting_outfitter_id, comms_local_percent, comms_export_percent, outfitter_documents, professional_hunter_id, professional_hunter_permit_type, professional_hunter_permit_issue_date, professional_hunter_permit_expiry_date, professional_hunter_documents, office_notes, created_at FROM partners WHERE id = %s """, (partner_id,), ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Partner not found") return row finally: conn.close() # Quotes endpoints @app.get("/quotes") def list_quotes(deleted: bool = Query(False, description="If true, return only soft-deleted quotes")): conn = get_connection() try: cur = conn.cursor(dictionary=True) if deleted: cur.execute( """ SELECT id, quote_unique_number, partner_id, partner_name, order_no, quote_date, order_type, customer_id, customer_first_names, customer_surname, quote_status, quote_currency, sub_total, vat_amount, total_amount, deposit_percent, deposit_amount_due, created_at FROM quotes WHERE deleted_at IS NOT NULL ORDER BY created_at DESC """ ) else: cur.execute( """ SELECT id, quote_unique_number, partner_id, partner_name, order_no, quote_date, order_type, customer_id, customer_first_names, customer_surname, quote_status, quote_currency, sub_total, vat_amount, total_amount, deposit_percent, deposit_amount_due, created_at FROM quotes WHERE deleted_at IS NULL ORDER BY created_at DESC """ ) return cur.fetchall() finally: conn.close() @app.get("/quotes/{quote_id}") def get_quote(quote_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT * FROM quotes WHERE id = %s AND deleted_at IS NULL """, (quote_id,), ) quote = cur.fetchone() if not quote: raise HTTPException(status_code=404, detail="Quote not found") # Line items cur.execute( """ SELECT * FROM quote_line_items WHERE quote_id = %s ORDER BY id """, (quote_id,), ) items = cur.fetchall() or [] # Processes per item item_id_to_processes: dict[int, list[dict]] = {} if items: item_ids = [row["id"] for row in items] format_strings = ",".join(["%s"] * len(item_ids)) cur.execute( f"SELECT * FROM quote_line_item_processes WHERE line_item_id IN ({format_strings})", tuple(item_ids), ) for p in cur.fetchall() or []: item_id_to_processes.setdefault(p["line_item_id"], []).append(p) for it in items: it["processes"] = item_id_to_processes.get(it["id"], []) # HS codes cur.execute("SELECT hs_code FROM quote_hs_codes WHERE quote_id = %s", (quote_id,)) hs_codes = [r["hs_code"] for r in (cur.fetchall() or [])] # Documents cur.execute( "SELECT id, file_url, file_name, mime_type, created_at FROM quote_documents WHERE quote_id = %s ORDER BY created_at DESC", (quote_id,), ) documents = cur.fetchall() or [] quote["line_items"] = items quote["hs_codes"] = hs_codes quote["documents"] = documents return quote finally: conn.close() @app.delete("/quotes/{quote_id}") def delete_quote(quote_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute( "UPDATE quotes SET deleted_at = NOW() WHERE id = %s AND deleted_at IS NULL", (quote_id,), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Quote not found") conn.commit() return {"success": True} finally: conn.close() @app.put("/quotes/{quote_id}") def update_quote(quote_id: int, payload: QuoteCreateRequest): # Recompute totals if not supplied (keep same rules as create) sub_total = payload.sub_total if payload.sub_total is not None else sum( (li.total_price or 0) for li in payload.line_items ) vat_rate = 0.15 if payload.order_type == 'Local' else 0.0 vat_amount = payload.vat_amount if payload.vat_amount is not None else round(sub_total * vat_rate, 2) total_amount = payload.total_amount if payload.total_amount is not None else round(sub_total + vat_amount, 2) deposit_percent = payload.deposit_percent if deposit_percent is None: if payload.order_type == 'Local' and payload.quote_currency == 'ZAR': deposit_percent = 70.0 elif payload.order_type == 'Export' and payload.quote_currency in ('USD', 'EUR'): deposit_percent = 50.0 elif payload.order_type == 'Export' and payload.quote_currency == 'BWP': deposit_percent = 70.0 else: deposit_percent = 50.0 deposit_amount_due = payload.deposit_amount_due if payload.deposit_amount_due is not None else round(total_amount * (deposit_percent / 100.0), 2) conn = get_connection() try: conn.start_transaction() cur = conn.cursor() # Verify quote exists cur.execute("SELECT quote_unique_number FROM quotes WHERE id = %s AND deleted_at IS NULL", (quote_id,)) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Quote not found") # Update main quote record cur.execute( """ UPDATE quotes SET partner_id=%s, partner_name=%s, order_no=%s, quote_date=%s, order_type=%s, customer_id=%s, customer_first_names=%s, customer_surname=%s, customer_email=%s, customer_contact_number=%s, customer_country=%s, customer_street=%s, customer_province_state=%s, customer_city=%s, customer_postal_code=%s, company_name=%s, company_physical_address=%s, company_phone_number=%s, company_email=%s, company_vat_number=%s, company_registration_number=%s, company_logo_url=%s, quote_status=%s, quote_currency=%s, sub_total=%s, vat_amount=%s, total_amount=%s, deposit_percent=%s, deposit_amount_due=%s, terms_and_conditions=%s, payment_details=%s, office_notes=%s WHERE id = %s """, ( payload.partner_id, payload.partner_name, payload.order_no, payload.quote_date, payload.order_type, payload.customer_id, payload.customer_first_names, payload.customer_surname, payload.customer_email, payload.customer_contact_number, payload.customer_country, payload.customer_street, payload.customer_province_state, payload.customer_city, payload.customer_postal_code, payload.company_name, payload.company_physical_address, payload.company_phone_number, payload.company_email, _blank_to_none(payload.company_vat_number), _blank_to_none(payload.company_registration_number), _blank_to_none(payload.company_logo_url), (payload.quote_status or 'Draft - In Progress'), payload.quote_currency, sub_total, vat_amount, total_amount, deposit_percent, deposit_amount_due, _blank_to_none(payload.terms_and_conditions), _blank_to_none(payload.payment_details), _blank_to_none(payload.office_notes), quote_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Quote not found") # Replace line items and processes # Delete processes for this quote's items cur.execute("SELECT id FROM quote_line_items WHERE quote_id = %s", (quote_id,)) item_ids = [r[0] for r in (cur.fetchall() or [])] if item_ids: in_clause = ",".join(["%s"] * len(item_ids)) cur.execute(f"DELETE FROM quote_line_item_processes WHERE line_item_id IN ({in_clause})", tuple(item_ids)) # Delete items cur.execute("DELETE FROM quote_line_items WHERE quote_id = %s", (quote_id,)) # Insert new items if payload.line_items: cur_li = conn.cursor() for li in payload.line_items: cur_li.execute( """ INSERT INTO quote_line_items ( quote_id, item_quantity, item_type, species, unit_price, total_price, detailed_instructions, line_item_status ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s) """, ( quote_id, li.item_quantity, li.item_type, _blank_to_none(li.species), li.unit_price, li.total_price, _blank_to_none(li.detailed_instructions), (li.line_item_status or 'Draft') ), ) line_item_id = cur_li.lastrowid if li.processes: cur_proc = conn.cursor() for pr in li.processes: cur_proc.execute( "INSERT INTO quote_line_item_processes (line_item_id, processing_type, processing_subtype) VALUES (%s,%s,%s)", (line_item_id, pr.processing_type, _blank_to_none(pr.processing_subtype)), ) # Replace HS codes cur.execute("DELETE FROM quote_hs_codes WHERE quote_id = %s", (quote_id,)) if payload.hs_codes: cur_hs = conn.cursor() for code in payload.hs_codes: cur_hs.execute( "INSERT INTO quote_hs_codes (quote_id, hs_code) VALUES (%s,%s)", (quote_id, code), ) conn.commit() # Return updated summary cur2 = conn.cursor(dictionary=True) cur2.execute("SELECT quote_unique_number FROM quotes WHERE id = %s", (quote_id,)) row2 = cur2.fetchone() or {} return {"id": quote_id, "quote_unique_number": row2.get("quote_unique_number"), "sub_total": sub_total, "vat_amount": vat_amount, "total_amount": total_amount, "deposit_percent": deposit_percent, "deposit_amount_due": deposit_amount_due} finally: conn.close() @app.post("/quotes/{quote_id}/restore") def restore_quote(quote_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute( "UPDATE quotes SET deleted_at = NULL WHERE id = %s AND deleted_at IS NOT NULL", (quote_id,), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Quote not found or not deleted") conn.commit() return {"success": True} finally: conn.close() @app.delete("/quotes/{quote_id}/permanent") def permanently_delete_quote(quote_id: int): conn = get_connection() try: cur = conn.cursor() # Hard delete; cascades will remove related rows cur.execute("DELETE FROM quotes WHERE id = %s", (quote_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Quote not found") conn.commit() return {"success": True} finally: conn.close() @app.post("/quotes") def create_quote(payload: QuoteCreateRequest): # Compute totals if not supplied sub_total = payload.sub_total if payload.sub_total is not None else sum( (li.total_price or 0) for li in payload.line_items ) # VAT rules vat_rate = 0.15 if payload.order_type == 'Local' else 0.0 vat_amount = payload.vat_amount if payload.vat_amount is not None else round(sub_total * vat_rate, 2) total_amount = payload.total_amount if payload.total_amount is not None else round(sub_total + vat_amount, 2) # Deposit defaults deposit_percent = payload.deposit_percent if deposit_percent is None: if payload.order_type == 'Local' and payload.quote_currency == 'ZAR': deposit_percent = 70.0 elif payload.order_type == 'Export' and payload.quote_currency in ('USD', 'EUR'): deposit_percent = 50.0 elif payload.order_type == 'Export' and payload.quote_currency == 'BWP': deposit_percent = 70.0 else: deposit_percent = 50.0 deposit_amount_due = payload.deposit_amount_due if payload.deposit_amount_due is not None else round(total_amount * (deposit_percent / 100.0), 2) conn = get_connection() try: conn.start_transaction() cur = conn.cursor() # Generate next quote_unique_number in the format Q-0001 cur.execute( """ SELECT COALESCE(MAX(CAST(SUBSTRING(quote_unique_number, 3) AS UNSIGNED)), 0) AS max_seq, COUNT(*) AS row_count FROM quotes """ ) row_max = cur.fetchone() max_seq = int(row_max[0] if row_max and row_max[0] is not None else 0) row_count = int(row_max[1] if row_max and row_max[1] is not None else 0) next_seq = max(max_seq, row_count) + 1 next_quote_number = f"Q-{next_seq:04d}" cur.execute( """ INSERT INTO quotes ( quote_unique_number, partner_id, partner_name, order_no, quote_date, order_type, customer_id, customer_first_names, customer_surname, customer_email, customer_contact_number, customer_country, customer_street, customer_province_state, customer_city, customer_postal_code, company_name, company_physical_address, company_phone_number, company_email, company_vat_number, company_registration_number, company_logo_url, quote_status, quote_currency, sub_total, vat_amount, total_amount, deposit_percent, deposit_amount_due, terms_and_conditions, payment_details, office_notes ) VALUES ( %s, %s,%s,%s,%s,%s, %s,%s,%s,%s,%s, %s,%s,%s,%s,%s, %s,%s,%s,%s, %s,%s,%s, %s,%s, %s,%s,%s,%s,%s, %s,%s,%s ) """, ( next_quote_number, payload.partner_id, payload.partner_name, payload.order_no, payload.quote_date, payload.order_type, payload.customer_id, payload.customer_first_names, payload.customer_surname, payload.customer_email, payload.customer_contact_number, payload.customer_country, payload.customer_street, payload.customer_province_state, payload.customer_city, payload.customer_postal_code, payload.company_name, payload.company_physical_address, payload.company_phone_number, payload.company_email, _blank_to_none(payload.company_vat_number), _blank_to_none(payload.company_registration_number), _blank_to_none(payload.company_logo_url), (payload.quote_status or 'Draft - In Progress'), payload.quote_currency, sub_total, vat_amount, total_amount, deposit_percent, deposit_amount_due, _blank_to_none(payload.terms_and_conditions), _blank_to_none(payload.payment_details), _blank_to_none(payload.office_notes), ), ) quote_id = cur.lastrowid # Insert line items if payload.line_items: cur_li = conn.cursor() for li in payload.line_items: cur_li.execute( """ INSERT INTO quote_line_items ( quote_id, item_quantity, item_type, species, unit_price, total_price, detailed_instructions, line_item_status ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s) """, ( quote_id, li.item_quantity, li.item_type, _blank_to_none(li.species), li.unit_price, li.total_price, _blank_to_none(li.detailed_instructions), (li.line_item_status or 'Draft') ), ) line_item_id = cur_li.lastrowid # Insert processes if li.processes: cur_proc = conn.cursor() for pr in li.processes: cur_proc.execute( "INSERT INTO quote_line_item_processes (line_item_id, processing_type, processing_subtype) VALUES (%s,%s,%s)", (line_item_id, pr.processing_type, _blank_to_none(pr.processing_subtype)), ) # Insert HS codes if payload.hs_codes: cur_hs = conn.cursor() for code in payload.hs_codes: cur_hs.execute( "INSERT INTO quote_hs_codes (quote_id, hs_code) VALUES (%s,%s)", (quote_id, code), ) conn.commit() # Return generated quote number return {"id": quote_id, "quote_unique_number": next_quote_number, "sub_total": sub_total, "vat_amount": vat_amount, "total_amount": total_amount, "deposit_percent": deposit_percent, "deposit_amount_due": deposit_amount_due} except Exception as e: try: conn.rollback() except Exception: pass raise e finally: conn.close() @app.get("/quotes/{quote_id}/documents") def list_quote_documents(quote_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( "SELECT id, file_url, file_name, mime_type, created_at FROM quote_documents WHERE quote_id = %s ORDER BY created_at DESC", (quote_id,), ) return cur.fetchall() finally: conn.close() @app.post("/quotes/{quote_id}/documents") def upload_quote_document(quote_id: int, file: UploadFile = File(...)): safe_dir = os.path.join(UPLOAD_DIR, f"quote_{quote_id}") os.makedirs(safe_dir, exist_ok=True) unique_name = f"{int(datetime.utcnow().timestamp())}_{secrets.token_hex(8)}_{file.filename}" file_path = os.path.join(safe_dir, unique_name) with open(file_path, 'wb') as out: out.write(file.file.read()) rel_path = os.path.relpath(file_path, UPLOAD_DIR).replace('\\', '/') conn = get_connection() try: cur = conn.cursor() cur.execute( "INSERT INTO quote_documents (quote_id, file_url, file_name, mime_type) VALUES (%s,%s,%s,%s)", (quote_id, rel_path, file.filename, file.content_type), ) doc_id = cur.lastrowid return {"id": doc_id, "file_url": f"/uploads/{rel_path}"} finally: conn.close() # HS Codes master (for selection) @app.get("/hs-codes") def list_hs_codes(): conn = get_connection() try: cur = conn.cursor(dictionary=True) # If master table exists, return from it; else return distinct from applied codes try: cur.execute("SHOW TABLES LIKE 'hs_codes_master'") if cur.fetchone(): cur.execute("SELECT code as hs_code, description FROM hs_codes_master ORDER BY code") return cur.fetchall() except Exception: pass cur.execute("SELECT DISTINCT hs_code as hs_code FROM quote_hs_codes ORDER BY hs_code") return cur.fetchall() finally: conn.close() @app.post("/partners") def create_partner(body: PartnerRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO partners ( status, partner_type, company_name, company_vat, company_email, company_contact_number, company_country, company_street, company_province, company_city, company_postal_code, first_name, surname, email, contact_number, hunting_outfitter_id, comms_local_percent, comms_export_percent, outfitter_documents, professional_hunter_id, professional_hunter_permit_type, professional_hunter_permit_issue_date, professional_hunter_permit_expiry_date, professional_hunter_documents, office_notes ) VALUES ( %s,%s,%s,%s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s ) """, ( (body.status or 'Active'), body.partner_type, body.company_name, body.company_vat, body.company_email, body.company_contact_number, body.company_country, body.company_street, body.company_province, body.company_city, body.company_postal_code, body.first_name, body.surname, body.email, body.contact_number, _blank_to_none(body.hunting_outfitter_id), body.comms_local_percent, body.comms_export_percent, _blank_to_none(body.outfitter_documents), _blank_to_none(body.professional_hunter_id), _blank_to_none(body.professional_hunter_permit_type), _blank_to_none(body.professional_hunter_permit_issue_date), _blank_to_none(body.professional_hunter_permit_expiry_date), _blank_to_none(body.professional_hunter_documents), _blank_to_none(body.office_notes), ), ) new_id = cur.lastrowid conn.commit() return {"id": new_id} finally: conn.close() @app.put("/partners/{partner_id}") def update_partner(partner_id: int, body: PartnerRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE partners SET status=%s, partner_type=%s, company_name=%s, company_vat=%s, company_email=%s, company_contact_number=%s, company_country=%s, company_street=%s, company_province=%s, company_city=%s, company_postal_code=%s, first_name=%s, surname=%s, email=%s, contact_number=%s, hunting_outfitter_id=%s, comms_local_percent=%s, comms_export_percent=%s, outfitter_documents=%s, professional_hunter_id=%s, professional_hunter_permit_type=%s, professional_hunter_permit_issue_date=%s, professional_hunter_permit_expiry_date=%s, professional_hunter_documents=%s, office_notes=%s WHERE id = %s """, ( body.status, body.partner_type, body.company_name, body.company_vat, body.company_email, body.company_contact_number, body.company_country, body.company_street, body.company_province, body.company_city, body.company_postal_code, body.first_name, body.surname, body.email, body.contact_number, _blank_to_none(body.hunting_outfitter_id), body.comms_local_percent, body.comms_export_percent, _blank_to_none(body.outfitter_documents), _blank_to_none(body.professional_hunter_id), _blank_to_none(body.professional_hunter_permit_type), _blank_to_none(body.professional_hunter_permit_issue_date), _blank_to_none(body.professional_hunter_permit_expiry_date), _blank_to_none(body.professional_hunter_documents), _blank_to_none(body.office_notes), partner_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Partner not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/partners/{partner_id}") def delete_partner(partner_id: int): # Soft delete -> mark status as Inactive conn = get_connection() try: cur = conn.cursor() cur.execute("UPDATE partners SET status = 'Inactive' WHERE id = %s", (partner_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Partner not found") conn.commit() return {"success": True} finally: conn.close() # Partner documents @app.get("/partners/{partner_id}/documents") def list_partner_documents(partner_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, file_path, file_name, mime_type, created_at FROM partner_documents WHERE partner_id = %s AND deleted_at IS NULL ORDER BY created_at DESC """, (partner_id,), ) return cur.fetchall() finally: conn.close() @app.post("/partners/{partner_id}/documents") def upload_partner_document(partner_id: int, file: UploadFile = File(...)): safe_dir = os.path.join(UPLOAD_DIR, f"partner_{partner_id}") os.makedirs(safe_dir, exist_ok=True) unique_name = f"{int(datetime.utcnow().timestamp())}_{secrets.token_hex(8)}_{file.filename}" file_path = os.path.join(safe_dir, unique_name) with open(file_path, 'wb') as out: out.write(file.file.read()) rel_path = os.path.relpath(file_path, UPLOAD_DIR).replace('\\', '/') conn = get_connection() try: cur = conn.cursor() cur.execute( "INSERT INTO partner_documents (partner_id, file_path, file_name, mime_type) VALUES (%s,%s,%s,%s)", (partner_id, rel_path, file.filename, file.content_type), ) doc_id = cur.lastrowid return {"id": doc_id, "file_path": f"/uploads/{rel_path}"} finally: conn.close() @app.put("/partners/{partner_id}/documents/{doc_id}") def rename_partner_document(partner_id: int, doc_id: int, body: dict): new_name = (body.get("file_name") or "").strip() if not new_name: raise HTTPException(status_code=400, detail="file_name is required") conn = get_connection() try: cur = conn.cursor() cur.execute( "UPDATE partner_documents SET file_name = %s WHERE id = %s AND partner_id = %s", (new_name, doc_id, partner_id), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Document not found") return {"success": True} finally: conn.close() @app.delete("/partners/{partner_id}/documents/{doc_id}") def soft_delete_partner_document(partner_id: int, doc_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute( "UPDATE partner_documents SET deleted_at = NOW() WHERE id = %s AND partner_id = %s", (doc_id, partner_id), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Document not found") return {"success": True} finally: conn.close() @app.get("/partners/{partner_id}/documents/{doc_id}/download") def download_partner_document(partner_id: int, doc_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( "SELECT file_path, file_name, mime_type FROM partner_documents WHERE id = %s AND partner_id = %s AND deleted_at IS NULL", (doc_id, partner_id), ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Document not found") abs_path = os.path.join(UPLOAD_DIR, row["file_path"]) if not os.path.isfile(abs_path): raise HTTPException(status_code=404, detail="File missing") filename = row["file_name"] or os.path.basename(abs_path) return FileResponse(abs_path, media_type=row.get("mime_type") or 'application/octet-stream', filename=filename) finally: conn.close() @app.post("/auth/login", response_model=LoginResponse) def login(payload: LoginRequest): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( "SELECT id, password_hash, role FROM users WHERE username = %s OR email = %s", (payload.identifier, payload.identifier), ) row = cur.fetchone() if not row or not row.get("password_hash"): raise HTTPException(status_code=401, detail="Invalid credentials") if not pwd_context.verify(payload.password, row["password_hash"]): raise HTTPException(status_code=401, detail="Invalid credentials") # Update last login cur.execute("UPDATE users SET last_login = NOW() WHERE id = %s", (row["id"],)) return {"message": "Login successful", "customer_id": row["id"], "role": row.get("role")} finally: conn.close() @app.post("/auth/forgot-password") def forgot_password(payload: ForgotPasswordRequest): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute("SELECT id FROM users WHERE email = %s", (payload.email,)) row = cur.fetchone() if not row: # Do not reveal existence return {"message": "If the email exists, a reset link has been sent"} token = secrets.token_urlsafe(48) expires_at = datetime.utcnow() + timedelta(hours=1) cur2 = conn.cursor() cur2.execute( "UPDATE users SET password_reset_token = %s, password_reset_expires = %s WHERE id = %s", (token, expires_at, row["id"]), ) reset_link = f"http://localhost:40773/reset-password/{token}" # Send email if SMTP is configured smtp_host = os.getenv("SMTP_HOST") smtp_port = int(os.getenv("SMTP_PORT", "587")) smtp_starttls = os.getenv("SMTP_STARTTLS", "true").lower() in ("1", "true", "yes", "on") smtp_user = os.getenv("SMTP_USER") smtp_password = os.getenv("SMTP_PASSWORD") if smtp_host and smtp_user and smtp_password: try: msg = EmailMessage() msg["Subject"] = "Password Reset Instructions" msg["From"] = smtp_user msg["To"] = payload.email msg.set_content( f"Hello,\n\nA password reset was requested for your account. " f"If you made this request, click the link below to reset your password. " f"This link will expire in 1 hour.\n\n{reset_link}\n\n" f"If you did not request this, you can safely ignore this email.\n" ) with smtplib.SMTP(smtp_host, smtp_port, timeout=20) as server: if smtp_starttls: server.starttls() server.login(smtp_user, smtp_password) server.send_message(msg) except Exception as e: # Do not fail the endpoint on email errors, just log print(f"[forgot-password] Email send failed: {e}") return {"message": "If the email exists, a reset link has been sent", "token": token, "reset_link": reset_link} finally: conn.close() @app.post("/auth/reset-password") def reset_password(payload: ResetPasswordRequest): if len(payload.new_password) < 8: raise HTTPException(status_code=400, detail="Password must be at least 8 characters") conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( "SELECT id, password_reset_expires FROM users WHERE password_reset_token = %s", (payload.token,), ) row = cur.fetchone() if not row: raise HTTPException(status_code=400, detail="Invalid token") if row["password_reset_expires"] < datetime.utcnow(): raise HTTPException(status_code=400, detail="Token expired") password_hash = pwd_context.hash(payload.new_password) cur2 = conn.cursor() cur2.execute( "UPDATE users SET password_hash = %s, password_reset_token = NULL, password_reset_expires = NULL WHERE id = %s", (password_hash, row["id"]), ) return {"message": "Password reset successful"} finally: conn.close() # User management endpoints @app.get("/users") def list_users(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, username, email, role, last_login, created_at FROM users ORDER BY username """ ) return cur.fetchall() finally: conn.close() class UpdateRoleRequest(BaseModel): role: str @app.put("/users/{user_id}/role") def update_user_role(user_id: int, body: UpdateRoleRequest): role = body.role.strip().lower() valid = {"admin", "office", "user", "customer", "staff"} if role not in valid: raise HTTPException(status_code=400, detail="Invalid role") # Normalize capitalization role_db = role if role in {"admin", "office", "user"} else role conn = get_connection() try: cur = conn.cursor() cur.execute("UPDATE users SET role = %s WHERE id = %s", (role_db, user_id)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="User not found") conn.commit() return {"success": True} finally: conn.close() class CreateUserRequest(BaseModel): username: str email: Optional[EmailStr] = None role: str password: Optional[str] = None @app.post("/users") def create_user(body: CreateUserRequest): role = body.role.strip().lower() valid = {"admin", "office", "staff"} if role not in valid: raise HTTPException(status_code=400, detail="Invalid role") username = body.username.strip() if not username: raise HTTPException(status_code=400, detail="Username is required") if body.password is not None and len(body.password) < 8: raise HTTPException(status_code=400, detail="Password must be at least 8 characters") conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute("SELECT id FROM users WHERE username = %s", (username,)) if cur.fetchone(): raise HTTPException(status_code=400, detail="Username already in use") if body.email: cur.execute("SELECT id FROM users WHERE email = %s", (body.email,)) if cur.fetchone(): raise HTTPException(status_code=400, detail="Email already in use") password_hash_value = pwd_context.hash(body.password) if body.password else None cur2 = conn.cursor() cur2.execute( "INSERT INTO users (username, email, password_hash, role) VALUES (%s, %s, %s, %s)", (username, body.email, password_hash_value, role) ) user_id = cur2.lastrowid conn.commit() return {"id": user_id} finally: conn.close() class ChangePasswordRequest(BaseModel): new_password: str @app.put("/users/{user_id}/password") def change_user_password(user_id: int, body: ChangePasswordRequest): if len(body.new_password) < 8: raise HTTPException(status_code=400, detail="Password must be at least 8 characters") password_hash = pwd_context.hash(body.new_password) conn = get_connection() try: cur = conn.cursor() cur.execute("UPDATE users SET password_hash = %s WHERE id = %s", (password_hash, user_id)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="User not found") conn.commit() return {"success": True} finally: conn.close() class UpdateUserRequest(BaseModel): username: Optional[str] = None email: Optional[EmailStr] = None @app.put("/users/{user_id}") def update_user(user_id: int, body: UpdateUserRequest): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute("SELECT id, username, email FROM users WHERE id = %s", (user_id,)) existing = cur.fetchone() if not existing: raise HTTPException(status_code=404, detail="User not found") new_username = (body.username or existing["username"]).strip() new_email = (body.email if body.email is not None else existing["email"]) # may be None # Uniqueness checks cur.execute("SELECT id FROM users WHERE username = %s AND id <> %s", (new_username, user_id)) if cur.fetchone(): raise HTTPException(status_code=400, detail="Username already in use") if new_email: cur.execute("SELECT id FROM users WHERE email = %s AND id <> %s", (new_email, user_id)) if cur.fetchone(): raise HTTPException(status_code=400, detail="Email already in use") cur2 = conn.cursor() cur2.execute( "UPDATE users SET username = %s, email = %s WHERE id = %s", (new_username, new_email, user_id) ) if cur2.rowcount == 0: raise HTTPException(status_code=404, detail="User not found") conn.commit() return {"success": True} finally: conn.close() # Order endpoints (using 'orders' table) @app.get("/orders") def get_orders(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, order_unique_number, related_quote_id, related_quote_number, related_invoice_id, related_invoice_number, order_type, processing_type, customer_id, customer_name, order_delivery, deposit_received, instructions_finalised, order_paid_in_full, office_notes, created_at, updated_at FROM orders ORDER BY created_at DESC """ ) return cur.fetchall() finally: conn.close() @app.get("/orders/{order_id}") def get_order(order_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, order_unique_number, related_quote_id, related_quote_number, related_invoice_id, related_invoice_number, order_type, processing_type, customer_id, customer_name, order_delivery, deposit_received, instructions_finalised, order_paid_in_full, office_notes, created_at, updated_at FROM orders WHERE id = %s """, (order_id,), ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Order not found") return row finally: conn.close() @app.post("/orders") def create_order(order: OrderRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ INSERT INTO orders ( related_quote_id, related_quote_number, related_invoice_id, related_invoice_number, order_type, processing_type, customer_id, customer_name, order_delivery, deposit_received, instructions_finalised, order_paid_in_full, office_notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( order.related_quote_id, order.related_quote_number, order.related_invoice_id, order.related_invoice_number, order.order_type, order.processing_type, order.customer_id, order.customer_name, order.order_delivery, order.deposit_received or 'No', order.instructions_finalised or 'No', order.order_paid_in_full or 'No', order.office_notes, ), ) new_id = cur.lastrowid # Fetch generated order_unique_number cur2 = conn.cursor(dictionary=True) cur2.execute("SELECT order_unique_number FROM orders WHERE id = %s", (new_id,)) row = cur2.fetchone() return {"id": new_id, "order_unique_number": (row or {}).get("order_unique_number")} finally: conn.close() @app.put("/orders/{order_id}") def update_order(order_id: int, order: OrderRecord): conn = get_connection() try: cur = conn.cursor() cur.execute( """ UPDATE orders SET related_quote_id = %s, related_quote_number = %s, related_invoice_id = %s, related_invoice_number = %s, order_type = %s, processing_type = %s, customer_id = %s, customer_name = %s, order_delivery = %s, deposit_received = %s, instructions_finalised = %s, order_paid_in_full = %s, office_notes = %s WHERE id = %s """, ( order.related_quote_id, order.related_quote_number, order.related_invoice_id, order.related_invoice_number, order.order_type, order.processing_type, order.customer_id, order.customer_name, order.order_delivery, order.deposit_received, order.instructions_finalised, order.order_paid_in_full, order.office_notes, order_id, ), ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Order not found") return {"success": True} finally: conn.close() @app.delete("/orders/{order_id}") def delete_order(order_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute("DELETE FROM orders WHERE id = %s", (order_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Order not found") return {"success": True} finally: conn.close() @app.get("/customers/{customer_id}/documents") def list_customer_documents(customer_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, file_path, file_name, mime_type, created_at FROM customer_documents WHERE customer_id = %s AND deleted_at IS NULL ORDER BY created_at DESC """, (customer_id,) ) rows = cur.fetchall() return rows finally: conn.close() @app.post("/customers/{customer_id}/documents") def upload_customer_document(customer_id: int, file: UploadFile = File(...)): safe_dir = os.path.join(UPLOAD_DIR, str(customer_id)) os.makedirs(safe_dir, exist_ok=True) unique_name = f"{int(datetime.utcnow().timestamp())}_{secrets.token_hex(8)}_{file.filename}" file_path = os.path.join(safe_dir, unique_name) with open(file_path, 'wb') as out: out.write(file.file.read()) rel_path = os.path.relpath(file_path, UPLOAD_DIR).replace('\\', '/') conn = get_connection() try: cur = conn.cursor() cur.execute( "INSERT INTO customer_documents (customer_id, file_path, file_name, mime_type) VALUES (%s,%s,%s,%s)", (customer_id, rel_path, file.filename, file.content_type) ) doc_id = cur.lastrowid return {"id": doc_id, "file_path": f"/uploads/{rel_path}"} finally: conn.close() @app.put("/customers/{customer_id}/documents/{doc_id}") def rename_customer_document(customer_id: int, doc_id: int, body: dict): new_name = (body.get("file_name") or "").strip() if not new_name: raise HTTPException(status_code=400, detail="file_name is required") conn = get_connection() try: cur = conn.cursor() cur.execute( "UPDATE customer_documents SET file_name = %s WHERE id = %s AND customer_id = %s", (new_name, doc_id, customer_id) ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Document not found") return {"success": True} finally: conn.close() @app.delete("/customers/{customer_id}/documents/{doc_id}") def soft_delete_customer_document(customer_id: int, doc_id: int): conn = get_connection() try: cur = conn.cursor() cur.execute( "UPDATE customer_documents SET deleted_at = NOW() WHERE id = %s AND customer_id = %s", (doc_id, customer_id) ) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Document not found") return {"success": True} finally: conn.close() @app.get("/customers/{customer_id}/documents/{doc_id}/download") def download_customer_document(customer_id: int, doc_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( "SELECT file_path, file_name, mime_type FROM customer_documents WHERE id = %s AND customer_id = %s AND deleted_at IS NULL", (doc_id, customer_id) ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Document not found") abs_path = os.path.join(UPLOAD_DIR, row["file_path"]) if not os.path.isfile(abs_path): raise HTTPException(status_code=404, detail="File missing") filename = row["file_name"] or os.path.basename(abs_path) return FileResponse(abs_path, media_type=row.get("mime_type") or 'application/octet-stream', filename=filename) finally: conn.close() @app.get("/customers") def get_customers(): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, customer_unique_number, status, customer_type, first_names, surname, email, contact_number, whatsapp_number, physical_country, physical_street, physical_province_state, physical_city, physical_postal_code, ship_country, ship_street, ship_province_state, ship_city, ship_postal_code, referred_by, preferred_currency, citizenship, rsa_id, passport_number, documents, office_notes, created_at FROM customers ORDER BY created_at DESC """ ) result = cur.fetchall() return result finally: conn.close() @app.post("/customers") def create_customer(customer: CustomerRecord): conn = get_connection() try: cur = conn.cursor() try: cur.execute( """ INSERT INTO customers ( status, customer_type, first_names, surname, email, contact_number, whatsapp_number, physical_country, physical_street, physical_province_state, physical_city, physical_postal_code, ship_country, ship_street, ship_province_state, ship_city, ship_postal_code, referred_by, preferred_currency, citizenship, rsa_id, passport_number, documents, office_notes ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, ( # Respect provided status; default to Active if missing (customer.status or 'Active'), customer.customer_type, customer.first_names, customer.surname, customer.email, customer.contact_number, customer.whatsapp_number, customer.physical_country, customer.physical_street, customer.physical_province_state, customer.physical_city, customer.physical_postal_code, customer.ship_country, customer.ship_street, customer.ship_province_state, customer.ship_city, customer.ship_postal_code, customer.referred_by, customer.preferred_currency, customer.citizenship, customer.rsa_id, customer.passport_number, customer.documents, customer.office_notes ), ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1062: raise HTTPException(status_code=400, detail="Email already in use") raise new_id = cur.lastrowid # Create associated user login with role 'customer' and inactive (no login until password is set) # Username default: email if present, else generated "customer{new_id}" username = customer.email if customer.email else f"customer{new_id}" # Ensure username uniqueness by appending suffix if needed cur2 = conn.cursor() candidate = username suffix = 1 while True: cur2.execute("SELECT id FROM users WHERE username = %s", (candidate,)) if not cur2.fetchone(): break suffix += 1 candidate = f"{username}{suffix}" username_final = candidate # Mark account as inactive by leaving password_hash NULL initially; login endpoint rejects when no password password_hash_value = None try: cur2.execute( """ INSERT INTO users (username, email, password_hash, role, customer_id) VALUES (%s, %s, %s, %s, %s) """, (username_final, customer.email, password_hash_value, 'customer', new_id) ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1062: # Roll back user creation impact by surfacing clean error raise HTTPException(status_code=400, detail="Email already in use") raise conn.commit() return {"id": new_id} finally: conn.close() @app.get("/customers/{customer_id}") def get_customer(customer_id: int): conn = get_connection() try: cur = conn.cursor(dictionary=True) cur.execute( """ SELECT id, customer_unique_number, status, customer_type, first_names, surname, email, contact_number, whatsapp_number, physical_country, physical_street, physical_province_state, physical_city, physical_postal_code, ship_country, ship_street, ship_province_state, ship_city, ship_postal_code, referred_by, preferred_currency, citizenship, rsa_id, passport_number, documents, office_notes, created_at FROM customers WHERE id = %s """, (customer_id,), ) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="Customer not found") return row finally: conn.close() @app.put("/customers/{customer_id}") def update_customer(customer_id: int, customer: CustomerRecord): conn = get_connection() try: cur = conn.cursor() try: cur.execute( """ UPDATE customers SET status = %s, customer_type = %s, first_names = %s, surname = %s, email = %s, contact_number = %s, whatsapp_number = %s, physical_country = %s, physical_street = %s, physical_province_state = %s, physical_city = %s, physical_postal_code = %s, ship_country = %s, ship_street = %s, ship_province_state = %s, ship_city = %s, ship_postal_code = %s, referred_by = %s, preferred_currency = %s, citizenship = %s, rsa_id = %s, passport_number = %s, documents = %s, office_notes = %s WHERE id = %s """, ( customer.status, customer.customer_type, customer.first_names, customer.surname, customer.email, customer.contact_number, customer.whatsapp_number, customer.physical_country, customer.physical_street, customer.physical_province_state, customer.physical_city, customer.physical_postal_code, customer.ship_country, customer.ship_street, customer.ship_province_state, customer.ship_city, customer.ship_postal_code, customer.referred_by, customer.preferred_currency, customer.citizenship, customer.rsa_id, customer.passport_number, customer.documents, customer.office_notes, customer_id, ), ) except mysql.connector.Error as e: if getattr(e, 'errno', None) == 1062: raise HTTPException(status_code=400, detail="Email already in use") raise if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Customer not found") conn.commit() return {"success": True} finally: conn.close() @app.delete("/customers/{customer_id}") def delete_customer(customer_id: int): conn = get_connection() try: cur = conn.cursor() # Soft delete: mark as Inactive instead of removing the row cur.execute("UPDATE customers SET status = 'Inactive' WHERE id = %s", (customer_id,)) if cur.rowcount == 0: raise HTTPException(status_code=404, detail="Customer not found") conn.commit() return {"success": True} finally: conn.close() if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="0.0.0.0", port=4000, reload=True, log_level="info")