database.py

"""
database.py — CURT inventory database + data-access layer.

Running this file directly creates curt_inventory.db, creates the parts
table, and seeds it with 10+ realistic CURT parts.

The assistant (Phase 1 & 2) calls these functions instead of writing raw SQL.
"""

import difflib
import sqlite3

DB_NAME = "curt_inventory.db"

# --------------------------------------------------------------------------
# Schema
# --------------------------------------------------------------------------

CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS parts (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    name        TEXT    NOT NULL UNIQUE COLLATE NOCASE,
    quantity    INTEGER NOT NULL DEFAULT 0 CHECK (quantity >= 0),
    category    TEXT    NOT NULL,
    location    TEXT    NOT NULL
);
"""

# --------------------------------------------------------------------------
# Seed data (realistic CURT manufacturing parts)
# --------------------------------------------------------------------------

SEED_PARTS = [
    # name, quantity, category, location
    ("Brake Pads",            42,  "Braking",       "Aisle 1, Shelf B2"),
    ("Brake Rotors",          18,  "Braking",       "Aisle 1, Shelf B5"),
    ("ECU",                    7,  "Electronics",   "Cabinet C1, Bin 4"),
    ("Wiring Harness",        25,  "Electronics",   "Cabinet C1, Bin 9"),
    ("Trailer Hitch Ball",    60,  "Hitches",       "Aisle 3, Shelf A1"),
    ("Class III Hitch",       12,  "Hitches",       "Aisle 3, Rack R2"),
    ("Sway Control Kit",      15,  "Hitches",       "Aisle 3, Rack R3"),
    ("Wheel Bearing Kit",     30,  "Suspension",    "Aisle 2, Shelf D4"),
    ("Leaf Spring",           20,  "Suspension",    "Aisle 2, Rack R1"),
    ("Shock Absorber",        35,  "Suspension",    "Aisle 2, Shelf D7"),
    ("LED Tail Light",        48,  "Lighting",      "Aisle 4, Bin L2"),
    ("7-Way Connector",       22,  "Electrical",    "Cabinet C2, Bin 3"),
    ("Safety Chains",         50,  "Safety",        "Aisle 5, Shelf S1"),
    ("Breakaway Kit",         10,  "Safety",        "Aisle 5, Shelf S3"),
]


def get_connection():
    """Open a connection to the inventory database."""
    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row  # rows behave like dicts
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def init_db():
    """Create the database schema if it doesn't exist yet."""
    with get_connection() as conn:
        conn.execute(CREATE_TABLE_SQL)


def seed_db(force: bool = False):
    """Insert the seed parts. Skips if data already exists unless force=True."""
    with get_connection() as conn:
        count = conn.execute("SELECT COUNT(*) FROM parts").fetchone()[0]
        if count > 0 and not force:
            return False

        conn.executemany(
            "INSERT INTO parts (name, quantity, category, location) VALUES (?, ?, ?, ?)",
            SEED_PARTS,
        )
    return True


def reset_db():
    """Drop and rebuild the table, then re-seed. Useful during development."""
    with get_connection() as conn:
        conn.execute("DROP TABLE IF EXISTS parts")
        conn.execute(CREATE_TABLE_SQL)
    seed_db(force=True)


# --------------------------------------------------------------------------
# Data-access layer — the ONLY way the assistant talks to the database
# --------------------------------------------------------------------------

def get_part(name: str):
    """Look up a single part by (case-insensitive) name. Returns dict or None."""
    with get_connection() as conn:
        row = conn.execute(
            "SELECT * FROM parts WHERE name = ? COLLATE NOCASE", (name,)
        ).fetchone()
    return dict(row) if row else None


def _singular(text: str) -> str:
    """Very light plural handling: 'pads' -> 'pad', 'rotors' -> 'rotor'."""
    return " ".join(
        w[:-1] if w.endswith("s") and len(w) > 3 else w for w in text.lower().split()
    )


def find_part(query: str) -> dict:
    """
    Resolve a user-supplied part name to an inventory item, tolerating
    misspellings, partial names, plurals and casing.

    Resolution order:
      1. Exact match (case-insensitive)            -> "exact"
      2. Substring, both directions ("rotor" -> "Brake Rotors") -> "match"/"ambiguous"
      3. Fuzzy spelling ("brak pads" -> "Brake Pads") via difflib (cutoff 0.6)
      4. Plural-insensitive retry of steps 2-3

    Returns one of:
        {"status": "exact",     "part": {...}}
        {"status": "match",     "part": {...}}              # one confident non-exact hit
        {"status": "ambiguous", "candidates": [{...}, ...]}  # several plausible hits
        {"status": "none",      "candidates": []}
    """
    q = (query or "").strip().lower()
    if not q:
        return {"status": "none", "candidates": []}

    all_parts = get_all_parts()

    # 1) Exact (case-insensitive)
    for p in all_parts:
        if p["name"].lower() == q:
            return {"status": "exact", "part": p}

    # 2) Substring in either direction ("rotor" -> "Brake Rotors", "ecu kit" -> "ECU")
    subs = [p for p in all_parts if q in p["name"].lower() or p["name"].lower() in q]
    if len(subs) == 1:
        return {"status": "match", "part": subs[0]}
    if len(subs) > 1:
        return {"status": "ambiguous", "candidates": subs}

    # 3) Fuzzy spelling correction ("brak pads" -> "brake pads")
    close = difflib.get_close_matches(
        q, [p["name"].lower() for p in all_parts], n=2, cutoff=0.6
    )
    matches = [p for p in all_parts if p["name"].lower() in close]
    if len(matches) == 1:
        return {"status": "match", "part": matches[0]}
    if len(matches) > 1:
        return {"status": "ambiguous", "candidates": matches}

    # 4) Retry substring/fuzzy with plurals stripped ("brake pad" vs "Brake Pads")
    qs = _singular(q)
    subs2 = [p for p in all_parts if qs in _singular(p["name"]) or _singular(p["name"]) in qs]
    if len(subs2) == 1:
        return {"status": "match", "part": subs2[0]}
    if len(subs2) > 1:
        return {"status": "ambiguous", "candidates": subs2}

    return {"status": "none", "candidates": []}


def get_categories() -> list[str]:
    """Return the distinct category names in the inventory, sorted."""
    return sorted({p["category"] for p in get_all_parts()})


def find_category(query: str) -> dict:
    """
    Resolve a user-supplied category name, tolerating partial input
    ("electronic" -> "Electronics") and misspellings ("breaks" -> "Braking").

    Returns one of:
        {"status": "match",      "category": "Electronics"}
        {"status": "ambiguous",  "candidates": ["Electrical", "Electronics"]}
        {"status": "none",       "candidates": []}   # candidates = all categories
    """
    q = (query or "").strip().lower()
    if not q:
        return {"status": "none", "candidates": []}

    categories = get_categories()

    if q in [c.lower() for c in categories]:
        return {"status": "match", "category": next(c for c in categories if c.lower() == q)}

    # Partial / substring match
    subs = [c for c in categories if q in c.lower() or c.lower() in q]
    if len(subs) == 1:
        return {"status": "match", "category": subs[0]}
    if len(subs) > 1:
        return {"status": "ambiguous", "candidates": subs}

    # Fuzzy spelling ("breaks" -> "Braking")
    close = difflib.get_close_matches(q, [c.lower() for c in categories], n=2, cutoff=0.6)
    matches = sorted({c for c in categories if c.lower() in close})
    if len(matches) == 1:
        return {"status": "match", "category": matches[0]}
    if len(matches) > 1:
        return {"status": "ambiguous", "candidates": matches}

    return {"status": "none", "candidates": categories}


def get_by_category(category: str) -> list[dict]:
    """Return all parts in a category (case-insensitive, partial match)."""
    with get_connection() as conn:
        rows = conn.execute(
            "SELECT * FROM parts WHERE category LIKE ? COLLATE NOCASE ORDER BY name",
            (f"%{category}%",),
        ).fetchall()
    return [dict(r) for r in rows]


def get_all_parts() -> list[dict]:
    """Return every part in the inventory."""
    with get_connection() as conn:
        rows = conn.execute("SELECT * FROM parts ORDER BY category, name").fetchall()
    return [dict(r) for r in rows]


def get_low_stock(threshold: int = 10) -> list[dict]:
    """Return parts whose quantity is at or below the threshold."""
    with get_connection() as conn:
        rows = conn.execute(
            "SELECT * FROM parts WHERE quantity <= ? ORDER BY quantity ASC", (threshold,)
        ).fetchall()
    return [dict(r) for r in rows]


def update_quantity(name: str, delta: int) -> dict | None:
    """
    Add `delta` to a part's quantity (negative delta = remove stock).
    Returns the updated part, or None if the part doesn't exist or the
    update would push quantity below zero (CHECK constraint blocks it).
    """
    try:
        with get_connection() as conn:
            cur = conn.execute(
                "UPDATE parts SET quantity = quantity + ? WHERE name = ? COLLATE NOCASE",
                (delta, name),
            )
            if cur.rowcount == 0:
                return None

            row = conn.execute(
                "SELECT * FROM parts WHERE name = ? COLLATE NOCASE", (name,)
            ).fetchone()
    except sqlite3.IntegrityError:
        return None  # result would be negative

    return dict(row)


def set_quantity(name: str, quantity: int) -> dict | None:
    """Set a part's quantity to an explicit value. Returns updated part or None."""
    if quantity < 0:
        return None

    try:
        with get_connection() as conn:
            cur = conn.execute(
                "UPDATE parts SET quantity = ? WHERE name = ? COLLATE NOCASE",
                (quantity, name),
            )
            if cur.rowcount == 0:
                return None

            row = conn.execute(
                "SELECT * FROM parts WHERE name = ? COLLATE NOCASE", (name,)
            ).fetchone()
    except sqlite3.IntegrityError:
        return None

    return dict(row)


def add_part(name: str, quantity: int, category: str, location: str) -> dict:
    """Insert a new part. Returns the created part as a dict."""
    with get_connection() as conn:
        cur = conn.execute(
            "INSERT INTO parts (name, quantity, category, location) VALUES (?, ?, ?, ?)",
            (name, quantity, category, location),
        )
        row = conn.execute("SELECT * FROM parts WHERE id = ?", (cur.lastrowid,)).fetchone()
    return dict(row)


def delete_part(name: str) -> bool:
    """Remove a part by name. Returns True if something was deleted."""
    with get_connection() as conn:
        cur = conn.execute("DELETE FROM parts WHERE name = ? COLLATE NOCASE", (name,))
    return cur.rowcount > 0


# --------------------------------------------------------------------------
# Entry point
# --------------------------------------------------------------------------

if __name__ == "__main__":
    init_db()
    created = seed_db()
    print(f"Database ready: {DB_NAME}")
    print(f"Seeded: {created}")
    for p in get_all_parts():
        print(f"  {p['name']:<22} qty={p['quantity']:<4} {p['category']:<12} {p['location']}")