import re import secrets from datetime import datetime, timedelta import httpx from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel from sqlalchemy import text from . import vault from .auth import require_token, require_admin, hash_password from .db import get_db from .intake_routes import _resolve_barcode # The MEGA admin — back-office cockpit, all admin-gated. Inspired by WowPlatter's wp-admin nav # (Dashboard / Inventory / Orders / Import / Sales / Reports), but on RecordGod's own data. router = APIRouter(prefix="/admin", tags=["admin"]) @router.get("/stats") async def stats(ident=Depends(require_token), db=Depends(get_db)): s = ident["store_id"] row = (await db.execute(text(""" SELECT (SELECT count(*) FROM inventory WHERE store_id=:s) AS items, (SELECT count(*) FROM inventory WHERE store_id=:s AND in_stock) AS in_stock, (SELECT count(*) FROM inventory WHERE store_id=:s AND kind='vinyl') AS vinyl, (SELECT count(*) FROM inventory WHERE store_id=:s AND kind<>'vinyl') AS other_goods, (SELECT coalesce(sum(price),0)::float FROM inventory WHERE store_id=:s AND in_stock) AS stock_value, (SELECT count(*) FROM inventory WHERE staged) AS staged, (SELECT count(*) FROM crate) AS crates, (SELECT count(*) FROM sales) AS sales, (SELECT coalesce(sum(total),0)::float FROM sales) AS sales_total """), {"s": s})).mappings().first() return {"ok": True, **dict(row)} @router.get("/me") async def me(ident=Depends(require_token), db=Depends(get_db)): """Who am I + role — the dash hides admin-only tabs for staff (the backend still enforces).""" if ident.get("staff_id"): await db.execute(text("UPDATE staff SET last_seen=now() WHERE id=:i"), {"i": ident["staff_id"]}) await db.commit() from . import packs return {"name": ident.get("name"), "role": ident.get("role"), "staff_id": ident.get("staff_id"), **packs.info()} # ── Staff accounts (ADMIN ONLY) — operators get a token + role; tokens shown once on create/reset ── class StaffIn(BaseModel): name: str role: str = "staff" email: str | None = None phone: str | None = None pay_rate: float | None = None password: str | None = None class StaffEditIn(BaseModel): name: str | None = None role: str | None = None active: bool | None = None email: str | None = None phone: str | None = None pay_rate: float | None = None @router.get("/staff") async def staff_list(ident=Depends(require_admin), db=Depends(get_db)): rows = await db.execute(text( "SELECT id, name, role, active, email, phone, pay_rate, password_hash IS NOT NULL AS has_password, " "created_at, last_seen FROM staff ORDER BY active DESC, name")) return {"staff": [dict(r) for r in rows.mappings()]} @router.post("/staff") async def staff_add(body: StaffIn, ident=Depends(require_admin), db=Depends(get_db)): if not body.name.strip(): raise HTTPException(400, "name required") role = body.role if body.role in ("admin", "staff") else "staff" token = "rg_" + secrets.token_urlsafe(18) ph = hash_password(body.password) if body.password else None r = (await db.execute(text( "INSERT INTO staff (name, token, role, email, phone, pay_rate, password_hash) " "VALUES (:n, :t, :r, :e, :p, :pr, :ph) RETURNING id"), {"n": body.name.strip(), "t": token, "r": role, "e": (body.email or None), "p": body.phone, "pr": body.pay_rate, "ph": ph})).scalar() await db.commit() return {"ok": True, "id": r, "token": token} # token returned ONCE (break-glass); normal sign-in = email + password class SetPwIn(BaseModel): password: str @router.post("/staff/{sid}/password") async def staff_set_password(sid: int, body: SetPwIn, ident=Depends(require_admin), db=Depends(get_db)): """Admin sets/resets a staff member's password (e.g. onboarding, or a forgotten password).""" if len(body.password) < 6: raise HTTPException(400, "password too short (min 6 characters)") r = await db.execute(text("UPDATE staff SET password_hash=:h WHERE id=:i"), {"h": hash_password(body.password), "i": sid}) await db.commit() if not r.rowcount: raise HTTPException(404, "not found") return {"ok": True} @router.post("/staff/{sid}") async def staff_edit(sid: int, body: StaffEditIn, ident=Depends(require_admin), db=Depends(get_db)): fields = {k: v for k, v in body.model_dump().items() if v is not None} if "role" in fields and fields["role"] not in ("admin", "staff"): raise HTTPException(400, "bad role") if not fields: return {"ok": True, "unchanged": True} sets = ", ".join(f"{k} = :{k}" for k in fields) r = await db.execute(text(f"UPDATE staff SET {sets} WHERE id = :i"), {**fields, "i": sid}) await db.commit() if not r.rowcount: raise HTTPException(404, "not found") return {"ok": True} @router.post("/staff/{sid}/token") async def staff_reset_token(sid: int, ident=Depends(require_admin), db=Depends(get_db)): token = "rg_" + secrets.token_urlsafe(18) r = await db.execute(text("UPDATE staff SET token = :t WHERE id = :i"), {"t": token, "i": sid}) await db.commit() if not r.rowcount: raise HTTPException(404, "not found") return {"ok": True, "token": token} # ── Time clock — staff clock on/off; the open shift + today's total drive the workstation widget ── @router.get("/clock") async def clock_status(ident=Depends(require_token), db=Depends(get_db)): sid = ident.get("staff_id") if not sid: return {"staff": False} cur = (await db.execute(text( "SELECT id, clock_in FROM staff_shift WHERE staff_id=:s AND clock_out IS NULL ORDER BY clock_in DESC LIMIT 1" ), {"s": sid})).mappings().first() today = (await db.execute(text(""" SELECT coalesce(sum(extract(epoch FROM (coalesce(clock_out, now()) - clock_in))), 0)::bigint FROM staff_shift WHERE staff_id=:s AND clock_in >= date_trunc('day', now())"""), {"s": sid})).scalar() return {"staff": True, "name": ident["name"], "open": dict(cur) if cur else None, "today_seconds": today} @router.post("/clock/in") async def clock_in(ident=Depends(require_token), db=Depends(get_db)): sid = ident.get("staff_id") if not sid: raise HTTPException(400, "only staff clock in") cur = (await db.execute(text( "SELECT id FROM staff_shift WHERE staff_id=:s AND clock_out IS NULL LIMIT 1"), {"s": sid})).first() if cur: return {"ok": True, "already_open": True} await db.execute(text("INSERT INTO staff_shift (staff_id) VALUES (:s)"), {"s": sid}) await db.commit() return {"ok": True, "clocked_in": True} @router.post("/clock/out") async def clock_out(ident=Depends(require_token), db=Depends(get_db)): sid = ident.get("staff_id") if not sid: raise HTTPException(400, "only staff clock out") r = await db.execute(text( "UPDATE staff_shift SET clock_out=now() WHERE staff_id=:s AND clock_out IS NULL"), {"s": sid}) await db.commit() return {"ok": True, "closed": r.rowcount} @router.get("/timecards") async def timecards(days: int = Query(14), staff_id: int | None = None, ident=Depends(require_admin), db=Depends(get_db)): """The timecard report (admin): per-staff hour totals + recent shifts over the window.""" where = ["sh.clock_in >= now() - make_interval(days => :d)"] params = {"d": days} if staff_id: where.append("sh.staff_id = :sid"); params["sid"] = staff_id w = " AND ".join(where) shifts = [dict(r) for r in (await db.execute(text(f""" SELECT sh.id, sh.staff_id, st.name, sh.clock_in, sh.clock_out, round((extract(epoch FROM (coalesce(sh.clock_out, now()) - sh.clock_in))/3600.0)::numeric, 2) AS hours FROM staff_shift sh JOIN staff st ON st.id = sh.staff_id WHERE {w} ORDER BY sh.clock_in DESC LIMIT 500"""), params)).mappings()] totals = [dict(r) for r in (await db.execute(text(f""" SELECT sh.staff_id, st.name, st.pay_rate, count(*) AS shifts, round((sum(extract(epoch FROM (coalesce(sh.clock_out, now()) - sh.clock_in)))/3600.0)::numeric, 2) AS hours, round((sum(extract(epoch FROM (coalesce(sh.clock_out, now()) - sh.clock_in)))/3600.0 * coalesce(st.pay_rate, 0))::numeric, 2) AS pay FROM staff_shift sh JOIN staff st ON st.id = sh.staff_id WHERE {w} GROUP BY sh.staff_id, st.name, st.pay_rate ORDER BY hours DESC"""), params)).mappings()] return {"shifts": shifts, "totals": totals, "days": days} # ── New stock (distro purchases awaiting approval) — confirm release_id + set retail, then publish ── @router.get("/newstock") async def newstock_list(ident=Depends(require_token), db=Depends(get_db)): rows = [dict(r) for r in (await db.execute(text(""" SELECT i.sku, i.title, i.identifier AS barcode, i.release_id, i.cost_price, i.price, i.condition_type, i.cost_source, i.est_market_value, COALESCE(dc.thumb, dr.thumb) AS thumb, dr.artists_sort AS artist, dr.year, (i.attributes->>'resolved')::bool AS resolved, i.attributes->>'slug' AS slug FROM inventory i LEFT JOIN disc_cache dc ON dc.release_id = i.release_id LEFT JOIN disc_release dr ON dr.id = i.release_id WHERE i.store_id = :s AND i.staged AND i.cost_source IS NOT NULL ORDER BY i.cost_source DESC, i.title NULLS LAST, i.sku"""), {"s": ident["store_id"]})).mappings()] return {"items": rows, "count": len(rows)} class NewStockIn(BaseModel): release_id: int | None = None title: str | None = None price: float | None = None condition_type: str | None = None publish: bool = False @router.post("/newstock/{sku}") async def newstock_update(sku: str, body: NewStockIn, ident=Depends(require_token), db=Depends(get_db)): """Staff approve a distro copy: re-pick the release / set retail / flip new-used, then publish (un-stage).""" sets, params = [], {"sku": sku, "sid": ident["store_id"]} if body.release_id is not None: sets.append("release_id = :rid"); params["rid"] = body.release_id if body.title is not None: sets.append("title = :t"); params["t"] = body.title if body.price is not None: sets.append("price = :p"); params["p"] = body.price if body.condition_type in ("new", "used"): sets.append("condition_type = :ct"); params["ct"] = body.condition_type if body.publish: sets += ["staged = false", "status = 'publish'", "in_stock = true"] if not sets: return {"ok": True, "unchanged": True} sets.append("updated_at = now()") r = await db.execute(text( f"UPDATE inventory SET {', '.join(sets)} WHERE sku = :sku AND store_id = :sid"), params) await db.commit() if not r.rowcount: raise HTTPException(404, "not found") return {"ok": True, "published": body.publish} @router.post("/newstock/publish-all") async def newstock_publish_all(ident=Depends(require_token), db=Depends(get_db)): """Publish every resolved copy that already has a retail price (the bulk 'ship it' button).""" r = await db.execute(text(""" UPDATE inventory SET staged = false, status = 'publish', in_stock = true, updated_at = now() WHERE store_id = :s AND staged AND cost_source IS NOT NULL AND release_id IS NOT NULL AND price IS NOT NULL"""), {"s": ident["store_id"]}) await db.commit() return {"ok": True, "published": r.rowcount} # ── Wishlist buy-list — scrape a distro wishlist → scarcity-rank (DealGod /api/supply) → what to buy ── _COLOUR_RE = re.compile( r"(colou?red|green|blue|red|white|clear|splatter|marble|gold|silver|pink|orange|yellow|purple|" r"transparent|smoke|translucent|coke[- ]?bottle|cream|crystal|neon|glow)", re.I) async def _dealgod_supply(db, release_ids): """Batch DealGod /api/supply → {release_id(str): {store_count, discogs_seller_count, …}}. Now delegates to the shared DealGod client (app/dealgod.py) — the single seam to the brain.""" from . import dealgod return await dealgod.supply(db, release_ids) class WishItem(BaseModel): barcode: str | None = None release_id: int | None = None title: str | None = None slug: str | None = None class WishIn(BaseModel): source: str = "rarewaves" items: list[WishItem] @router.post("/wishlist/scarcity") async def wishlist_scarcity(body: WishIn, ident=Depends(require_token), db=Depends(get_db)): """Resolve a scraped wishlist → release_ids → DealGod supply → upsert the buy-list (scarcity-ranked).""" sid = ident["store_id"] resolved = [] for it in body.items: rid = it.release_id or await _resolve_barcode(db, it.barcode) if not rid: continue m = _COLOUR_RE.search(it.slug or "") resolved.append((rid, it.barcode, it.title, (m.group(1).lower() if m else "black"))) rids = list({r[0] for r in resolved}) supply = await _dealgod_supply(db, rids) stocked = set() if rids: stocked = {x[0] for x in (await db.execute(text( "SELECT DISTINCT release_id FROM inventory WHERE release_id = ANY(:r) AND store_id=:s AND in_stock"), {"r": rids, "s": sid}))} for rid, barcode, title, colour in resolved: sup = supply.get(str(rid), {}) await db.execute(text(""" INSERT INTO buylist (store_id, release_id, barcode, title, colour, source, store_count, au_copies, lowest_au, median_au, discogs_seller_count, discogs_lowest, already_stocked, updated_at) VALUES (:s,:rid,:bc,:t,:col,:src,:sc,:auc,:lau,:mau,:dsc,:dlo,:stk,now()) ON CONFLICT (store_id, release_id) DO UPDATE SET store_count=EXCLUDED.store_count, au_copies=EXCLUDED.au_copies, lowest_au=EXCLUDED.lowest_au, median_au=EXCLUDED.median_au, discogs_seller_count=EXCLUDED.discogs_seller_count, discogs_lowest=EXCLUDED.discogs_lowest, already_stocked=EXCLUDED.already_stocked, title=COALESCE(EXCLUDED.title, buylist.title), colour=EXCLUDED.colour, updated_at=now()"""), {"s": sid, "rid": rid, "bc": barcode, "t": title, "col": colour, "src": body.source, "sc": sup.get("store_count"), "auc": sup.get("au_copies"), "lau": sup.get("lowest_au"), "mau": sup.get("median_au"), "dsc": sup.get("discogs_seller_count"), "dlo": sup.get("discogs_lowest"), "stk": rid in stocked}) await db.commit() return {"ok": True, "resolved": len(resolved), "unresolved": len(body.items) - len(resolved), "with_supply": sum(1 for r in resolved if str(r[0]) in supply)} @router.get("/buylist") async def buylist(ident=Depends(require_token), db=Depends(get_db)): rows = [dict(r) for r in (await db.execute(text(""" SELECT b.release_id, b.barcode, b.title, b.colour, b.store_count, b.au_copies, b.lowest_au::float AS lowest_au, b.median_au::float AS median_au, b.discogs_seller_count, b.discogs_lowest::float AS discogs_lowest, b.already_stocked, b.source, COALESCE(dc.thumb, dr.thumb) AS thumb, dr.artists_sort AS artist FROM buylist b LEFT JOIN disc_cache dc ON dc.release_id = b.release_id LEFT JOIN disc_release dr ON dr.id = b.release_id WHERE b.store_id = :s ORDER BY b.store_count ASC NULLS LAST, b.discogs_seller_count ASC NULLS LAST, b.title NULLS LAST"""), {"s": ident["store_id"]})).mappings()] return {"items": rows, "count": len(rows)} @router.post("/buylist/clear") async def buylist_clear(ident=Depends(require_token), db=Depends(get_db)): r = await db.execute(text("DELETE FROM buylist WHERE store_id = :s"), {"s": ident["store_id"]}) await db.commit() return {"ok": True, "cleared": r.rowcount} @router.get("/inventory") async def inventory(q: str = Query(""), kind: str = Query(""), crate_id: int | None = Query(None), page: int = Query(1, ge=1), ident=Depends(require_token), db=Depends(get_db)): per = 50 where = ["i.store_id = :s"] params = {"s": ident["store_id"], "per": per, "off": (page - 1) * per} if q: where.append("(i.sku ILIKE :q OR coalesce(i.title,dc.title) ILIKE :q OR dc.artist ILIKE :q)") params["q"] = f"%{q}%" if kind: where.append("i.kind = :k") params["k"] = kind if crate_id: where.append("i.crate_id = :cid") params["cid"] = crate_id w = " AND ".join(where) items = [dict(r) for r in (await db.execute(text(f""" SELECT i.sku, i.kind, i.release_id, coalesce(i.title, dc.title) AS title, dc.artist, i.price::float AS price, i.cost_price::float AS cost_price, i.condition_type, i.condition, i.sleeve_cond, i.slot_number, i.notes, i.in_stock, i.crate_id, c.label_text AS crate, dc.thumb, i.target_price::float AS target FROM inventory i LEFT JOIN disc_cache dc ON dc.release_id = i.release_id LEFT JOIN crate c ON c.id = i.crate_id WHERE {w} ORDER BY i.updated_at DESC NULLS LAST LIMIT :per OFFSET :off """), params)).mappings()] cparams = {k: v for k, v in params.items() if k not in ("per", "off")} total = (await db.execute(text( f"SELECT count(*) FROM inventory i LEFT JOIN disc_cache dc ON dc.release_id=i.release_id WHERE {w}" ), cparams)).scalar() return {"ok": True, "items": items, "page": page, "per": per, "total": total} # ── Inventory CRUD (the WowPlatter Inventory screen, RecordGod-side) ─────────────────────────── class InvItemIn(BaseModel): sku: str | None = None kind: str = "vinyl" release_id: int | None = None identifier: str | None = None title: str | None = None price: float | None = None qty: int = 1 condition: str | None = None sleeve_cond: str | None = None brand: str | None = None crate_id: int | None = None slot_number: int | None = None notes: str | None = None in_stock: bool = True @router.get("/value/{sku}") async def dealgod_value(sku: str, ident=Depends(require_token), db=Depends(get_db)): """DealGod value overlay for one item — records resolve via release_id, everything else via canon_id (the StoreGod generalisation). Canon goods also get lore (tier/faults). Gated on the pack's 'value' feature.""" from . import dealgod, packs if not packs.has_feature("value"): return {"ok": False, "reason": "value not enabled for this pack"} row = (await db.execute(text( "SELECT release_id, canon_id, kind, condition_type FROM inventory WHERE sku = :s AND store_id = :st"), {"s": sku, "st": ident["store_id"]})).mappings().first() if not row: raise HTTPException(404, "not found") cond = "new" if row["condition_type"] == "new" else "used-good" if row["release_id"]: val = await dealgod.value(db, release_id=row["release_id"], condition=cond) return {"ok": True, "value": val, "ref": {"type": "release", "id": row["release_id"]}} if row["canon_id"]: val = await dealgod.value(db, canon_id=row["canon_id"], condition=cond) lore = await dealgod.lore(db, row["canon_id"]) if packs.has_feature("lore") else None return {"ok": True, "value": val, "lore": lore, "ref": {"type": "canon", "id": row["canon_id"]}} return {"ok": True, "value": None, "note": "no DealGod ref — identify it first (records→release_id, others→canon_id)"} class IdentifyIn(BaseModel): clues: dict = {} scope: list[str] | None = None @router.post("/identify") async def dealgod_identify(body: IdentifyIn, ident=Depends(require_token), db=Depends(get_db)): """Universal 'what is this' for the back-office → DealGod /api/identify. Defaults scope to the active pack's domains (records bypass this — they use the Discogs matcher). Returns normalised candidates.""" from . import dealgod, packs if not packs.has_feature("identify"): return {"ok": False, "reason": "identify not enabled for this pack"} scope = body.scope or packs.active_scope() if scope == ["*"]: scope = None # vanilla StoreGod base → no domain filter r = await dealgod.identify(db, body.clues, scope=scope) return {"ok": True, **r} class StageIdentifyIn(BaseModel): ref_type: str = "canon" # canon | release ref_id: int title: str | None = None kind: str = "other" identifier: str | None = None # the scanned barcode/ISBN price: float | None = None condition: str | None = None @router.post("/intake/identify") async def intake_identify(body: StageIdentifyIn, ident=Depends(require_token), db=Depends(get_db)): """Stage an item from an /api/identify candidate — the universal intake for non-record packs (records keep their Discogs intake). release → release_id, canon → canon_id.""" from .intake_routes import _stage rid = body.ref_id if body.ref_type == "release" else None canon = body.ref_id if body.ref_type == "canon" else None res = await _stage(db, ident["store_id"], rid, None, body.condition, None, body.price, None, kind=body.kind, identifier=body.identifier, canon_id=canon, title=body.title) await db.commit() return {"ok": True, **res} def _purity(metal: str, karat: float) -> float: if metal == "gold": return max(0.0, min(1.0, karat / 24)) # karat out of 24 (9/14/18/22/24) k = karat # silver/platinum: fineness — 925 / 92.5 / 0.925 all → 0.925 if k > 100: k /= 1000 elif k > 1: k /= 100 return max(0.0, min(1.0, k)) @router.get("/melt") async def melt(metal: str = "gold", karat: float = 18, grams: float = 0, ident=Depends(require_token), db=Depends(get_db)): """The jewellery module's signature tool — live metal MELT value (the floor you never sell below). melt = grams × spot/oz × purity, using DealGod's live AU metal prices. Gated on the 'melt' feature.""" from . import dealgod, packs if not packs.has_feature("melt"): return {"ok": False, "reason": "melt not enabled — unlock the jewellery module"} mp = await dealgod.metal_prices(db) if not mp: return {"ok": False, "reason": "metal prices unavailable"} metal = (metal or "gold").lower() spot = {"gold": mp.get("gold_aud_oz"), "silver": mp.get("silver_aud_oz"), "platinum": mp.get("platinum_aud_oz")}.get(metal) if not spot: return {"ok": False, "reason": "metal must be gold|silver|platinum"} purity = _purity(metal, karat) per_gram = round(spot / 31.1035 * purity, 3) return {"ok": True, "metal": metal, "karat": karat, "grams": grams, "purity": round(purity, 4), "per_gram": per_gram, "melt": round(per_gram * (grams or 0), 2), "spot_oz": spot, "captured_at": mp.get("captured_at")} class PriceGuideIn(BaseModel): query: str scope: list[str] | None = None @router.post("/priceguide") async def priceguide(body: PriceGuideIn, ident=Depends(require_token), db=Depends(get_db)): """Specialist intel lookup — identify the thing, then compose value + lore (variants/faults/id_keys) + supply. The price guide a reseller checks BEFORE they buy. Scope defaults to the active modules' domains.""" from . import dealgod, packs if not packs.has_feature("value"): return {"ok": False, "reason": "value not enabled"} q = (body.query or "").strip() if not q: return {"ok": True, "candidate": None} digits = re.sub(r"\D", "", q) clue = ({"barcode": digits} if len(digits) >= 12 else {"isbn": digits}) if re.fullmatch(r"[0-9\s-]{8,}", q) else {"text": q} scope = body.scope or packs.active_scope() if scope == ["*"]: scope = None idr = await dealgod.identify(db, clue, scope=scope) cands = idr.get("candidates") or [] if not cands: return {"ok": True, "candidate": None, "reason": idr.get("reason") or idr.get("status")} c = cands[0] val = await dealgod.value(db, ref_type=c.get("ref_type"), ref_id=c.get("ref_id")) lore = await dealgod.lore(db, c.get("ref_id")) if (c.get("ref_type") == "canon" and packs.has_feature("lore")) else None sup = None if c.get("ref_type") == "release": sup = (await dealgod.supply(db, [c["ref_id"]])).get(str(c["ref_id"])) return {"ok": True, "candidate": c, "value": val, "lore": lore, "supply": sup, "others": cands[1:4]} class InvEditIn(BaseModel): price: float | None = None cost_price: float | None = None condition: str | None = None condition_type: str | None = None # 'new' | 'used' sleeve_cond: str | None = None title: str | None = None crate_id: int | None = None slot_number: int | None = None notes: str | None = None in_stock: bool | None = None target_price: float | None = None kind: str | None = None class BulkIn(BaseModel): skus: list[str] action: str # set_price | adjust_pct | set_crate | mark_sold | restock | delete value: float | int | str | None = None async def _fresh_sku(db): """Canonical SKU = 14-digit product-added timestamp (YYYYMMDDHHMMSS) — the format SCANGOD's EPC/RFID codec requires (packs SKU 14 digits × 10^9 + release_id onto the UHF chip). Bumps a second on collision so a same-second batch stays unique against the sku PRIMARY KEY. ponytail: check-then-insert; a truly concurrent same-second add on another connection could still hit the PK and be ON CONFLICT DO NOTHING'd. Fine for counter/admin stock entry; add retry-on-conflict if bulk-import concurrency grows.""" base = datetime.now() for i in range(120): sku = (base + timedelta(seconds=i)).strftime("%Y%m%d%H%M%S") if not (await db.execute(text("SELECT 1 FROM inventory WHERE sku = :s"), {"s": sku})).first(): return sku return (base + timedelta(seconds=120)).strftime("%Y%m%d%H%M%S") @router.post("/inventory/add") async def inv_add(body: InvItemIn, ident=Depends(require_token), db=Depends(get_db)): sku = (body.sku or "").strip() or await _fresh_sku(db) title = body.title if body.release_id and not title: # pull the cached title if intaking by release r = (await db.execute(text("SELECT title FROM disc_cache WHERE release_id=:r"), {"r": body.release_id})).first() title = r[0] if r else None await db.execute(text(""" INSERT INTO inventory (sku, store_id, kind, release_id, identifier, title, price, qty, in_stock, condition, sleeve_cond, brand, crate_id, slot_number, notes, status, created_at, updated_at) VALUES (:sku, :s, :kind, :rid, :ident, :title, :price, :qty, :in_stock, :cond, :sleeve, :brand, :crate, :slot, :notes, :status, now(), now()) ON CONFLICT (sku) DO NOTHING """), {"sku": sku, "s": ident["store_id"], "kind": body.kind, "rid": body.release_id, "ident": body.identifier, "title": title, "price": body.price, "qty": body.qty, "in_stock": body.in_stock, "cond": body.condition, "sleeve": body.sleeve_cond, "brand": body.brand, "crate": body.crate_id, "slot": body.slot_number, "notes": body.notes, "status": "publish" if body.in_stock else "sold"}) await db.commit() return {"ok": True, "sku": sku} @router.post("/inventory/{sku}/edit") async def inv_edit(sku: str, body: InvEditIn, ident=Depends(require_token), db=Depends(get_db)): fields = {k: v for k, v in body.model_dump().items() if v is not None} if not fields: return {"ok": True, "unchanged": True} sets = ", ".join(f"{k} = :{k}" for k in fields) + ", updated_at = now()" r = await db.execute(text(f"UPDATE inventory SET {sets} WHERE sku = :sku AND store_id = :s"), {**fields, "sku": sku, "s": ident["store_id"]}) await db.commit() if not r.rowcount: raise HTTPException(404, "sku not found") return {"ok": True, "updated": r.rowcount} @router.post("/inventory/{sku}/delete") async def inv_delete(sku: str, ident=Depends(require_token), db=Depends(get_db)): r = await db.execute(text("DELETE FROM inventory WHERE sku = :sku AND store_id = :s"), {"sku": sku, "s": ident["store_id"]}) await db.commit() return {"ok": True, "deleted": r.rowcount} class ImportIn(BaseModel): lines: list[str] crate_id: int | None = None price: float | None = None condition: str | None = None kind: str = "vinyl" @router.post("/inventory/import") async def inv_import(body: ImportIn, ident=Depends(require_token), db=Depends(get_db)): """Bulk stock-IN: each line = Release ID / barcode / catno → resolved against the local disc_release mirror → a new inventory item. (Distinct from the POS Import, which marks items SOLD.)""" added, not_found = [], [] for raw in body.lines: ln = raw.strip() if not ln: continue rid = title = None if ln.isdigit(): r = (await db.execute(text("SELECT id, title FROM disc_release WHERE id=:r"), {"r": int(ln)})).first() if r: rid, title = r[0], r[1] if not rid: # barcode / identifier r = (await db.execute(text("""SELECT dr.id, dr.title FROM disc_release_identifier i JOIN disc_release dr ON dr.id=i.release_id WHERE i.value=:v LIMIT 1"""), {"v": ln})).first() if r: rid, title = r[0], r[1] if not rid: # catalogue number r = (await db.execute(text("""SELECT dr.id, dr.title FROM disc_release_label l JOIN disc_release dr ON dr.id=l.release_id WHERE l.catno=:v LIMIT 1"""), {"v": ln})).first() if r: rid, title = r[0], r[1] if not rid: not_found.append(ln) continue sku = await _fresh_sku(db) await db.execute(text(""" INSERT INTO inventory (sku, store_id, kind, release_id, title, price, qty, in_stock, condition, crate_id, status, created_at, updated_at) VALUES (:sku,:s,:kind,:rid,:title,:price,1,true,:cond,:crate,'publish',now(),now()) ON CONFLICT (sku) DO NOTHING """), {"sku": sku, "s": ident["store_id"], "kind": body.kind, "rid": rid, "title": title, "price": body.price, "cond": body.condition, "crate": body.crate_id}) added.append({"sku": sku, "release_id": rid, "title": title}) await db.commit() return {"ok": True, "added": len(added), "items": added, "not_found": not_found} @router.post("/inventory/bulk") async def inv_bulk(body: BulkIn, ident=Depends(require_token), db=Depends(get_db)): if not body.skus: raise HTTPException(400, "no skus") p = {"skus": body.skus, "s": ident["store_id"], "v": body.value} sql = { "set_price": "UPDATE inventory SET price=:v, updated_at=now() WHERE sku=ANY(:skus) AND store_id=:s", "adjust_pct": "UPDATE inventory SET price=round(price*(1+:v/100.0),2), updated_at=now() WHERE sku=ANY(:skus) AND store_id=:s", "set_crate": "UPDATE inventory SET crate_id=:v, updated_at=now() WHERE sku=ANY(:skus) AND store_id=:s", "mark_sold": "UPDATE inventory SET in_stock=false, status='sold', sold_date=now(), updated_at=now() WHERE sku=ANY(:skus) AND store_id=:s", "restock": "UPDATE inventory SET in_stock=true, status='publish', sold_date=NULL, updated_at=now() WHERE sku=ANY(:skus) AND store_id=:s", "delete": "DELETE FROM inventory WHERE sku=ANY(:skus) AND store_id=:s", }.get(body.action) if not sql: raise HTTPException(400, f"unknown action {body.action}") r = await db.execute(text(sql), p) await db.commit() return {"ok": True, "affected": r.rowcount} class CustEditIn(BaseModel): first_name: str | None = None last_name: str | None = None email: str | None = None phone: str | None = None address: str | None = None notes: str | None = None mailing_optin: bool | None = None @router.get("/customers") async def customers_list(q: str = Query(""), ident=Depends(require_token), db=Depends(get_db)): where, params = "", {} if q.strip(): where = ("WHERE c.first_name ILIKE :q OR c.last_name ILIKE :q OR c.email ILIKE :q " "OR (c.first_name || ' ' || coalesce(c.last_name,'')) ILIKE :q") params = {"q": f"%{q.strip()}%"} rows = await db.execute(text(f""" SELECT c.id, trim(c.first_name || ' ' || coalesce(c.last_name,'')) AS name, c.email, c.phone, c.is_guest, c.mailing_optin, count(s.id) AS orders, coalesce(sum(s.total),0)::float AS spent, max(s.sale_date) AS last_order FROM customer c LEFT JOIN sales s ON s.customer_id = c.id {where} GROUP BY c.id ORDER BY spent DESC NULLS LAST, name LIMIT 300 """), params) return {"customers": [dict(r) for r in rows.mappings()]} @router.get("/customers/{cid}") async def customer_detail(cid: int, ident=Depends(require_token), db=Depends(get_db)): c = (await db.execute(text("SELECT * FROM customer WHERE id=:i"), {"i": cid})).mappings().first() if not c: raise HTTPException(404, "customer not found") sales = [dict(r) for r in (await db.execute(text(""" SELECT s.id, s.sale_number, s.total::float AS total, s.sale_date, s.payment_method, s.status, (SELECT count(*) FROM sale_items si WHERE si.sale_id = s.id) AS items FROM sales s WHERE s.customer_id = :i ORDER BY s.sale_date DESC NULLS LAST LIMIT 100 """), {"i": cid})).mappings()] stats = dict((await db.execute(text(""" SELECT count(*) AS orders, coalesce(sum(total),0)::float AS spent, max(sale_date) AS last_order FROM sales WHERE customer_id = :i """), {"i": cid})).mappings().first()) return {"customer": dict(c), "sales": sales, "stats": stats} @router.post("/customers/{cid}") async def customer_edit(cid: int, body: CustEditIn, ident=Depends(require_token), db=Depends(get_db)): fields = {k: v for k, v in body.model_dump().items() if v is not None} if not fields: return {"ok": True, "unchanged": True} sets = ", ".join(f"{k} = :{k}" for k in fields) r = await db.execute(text(f"UPDATE customer SET {sets} WHERE id = :i"), {**fields, "i": cid}) await db.commit() if not r.rowcount: raise HTTPException(404, "not found") return {"ok": True, "updated": r.rowcount} @router.get("/wantlist") async def wantlist_list(status: str = Query("pending"), ident=Depends(require_token), db=Depends(get_db)): """Customer record-requests captured by the public storefront /wantlist form.""" where, params = "", {} if status and status != "all": where = "WHERE w.status = :s"; params = {"s": status} rows = await db.execute(text(f""" SELECT w.id, w.release_id, w.artist, w.title, w.format, w.max_price::float AS max_price, w.name, w.email, w.phone, w.delivery_preference, w.postcode, w.notes, w.status, w.created_at, w.actioned_at FROM wantlist w {where} ORDER BY w.created_at DESC LIMIT 500 """), params) counts = {r["status"]: r["n"] for r in (await db.execute(text( "SELECT status, count(*) AS n FROM wantlist GROUP BY status"))).mappings()} return {"requests": [dict(r) for r in rows.mappings()], "counts": counts} class WantStatusIn(BaseModel): status: str # pending | found | fulfilled | cancelled @router.post("/wantlist/{wid}") async def wantlist_action(wid: int, body: WantStatusIn, ident=Depends(require_token), db=Depends(get_db)): if body.status not in ("pending", "found", "fulfilled", "cancelled"): raise HTTPException(422, "bad status") actioned = "now()" if body.status != "pending" else "NULL" r = await db.execute(text( f"UPDATE wantlist SET status=:s, actioned_at={actioned} WHERE id=:i"), {"s": body.status, "i": wid}) await db.commit() if not r.rowcount: raise HTTPException(404, "not found") return {"ok": True} @router.get("/crates") async def crates(ident=Depends(require_token), db=Depends(get_db)): rows = await db.execute(text(""" SELECT c.id, c.name, c.label_text, c.purpose, count(i.sku) FILTER (WHERE i.in_stock) AS items, coalesce(sum(i.price) FILTER (WHERE i.in_stock), 0)::float AS value FROM crate c LEFT JOIN inventory i ON i.crate_id = c.id GROUP BY c.id, c.name, c.label_text, c.purpose ORDER BY items DESC NULLS LAST, c.id """)) return {"ok": True, "crates": [dict(r) for r in rows.mappings()]} @router.get("/storemap") async def storemap(ident=Depends(require_token), db=Depends(get_db)): """Top-down store map: room + racks + crates at their real positions, with live item counts. Powers the clickable store-navigator (the WowPlatter 'Search' workstation, RecordGod-side).""" sp = (await db.execute(text( "SELECT * FROM virtual_space WHERE visible='y' ORDER BY (is_default='y') DESC LIMIT 1" ))).mappings().first() space = {"room_width": float(sp["room_width"] or 12), "room_depth": float(sp["room_depth"] or 12)} if sp else {"room_width": 12, "room_depth": 12} ctypes = {r["id"]: {"w": float(r["width"] or 0.35), "d": float(r["depth"] or 0.5)} for r in (await db.execute(text("SELECT id, width, depth FROM virtual_crate_type"))).mappings()} crates = [dict(r) for r in (await db.execute(text(""" SELECT c.id, c.name, c.label_text, c.crate_purpose, c.crate_type_id, c.pos_x::float AS x, c.pos_z::float AS z, c.rotation_y::float AS rot, (SELECT count(*) FROM inventory i WHERE i.crate_id = c.id AND i.in_stock) AS items FROM virtual_crate c WHERE c.visible='y' """))).mappings()] racks = [dict(r) for r in (await db.execute(text(""" SELECT r.pos_x::float AS x, r.pos_z::float AS z, r.rotation_y::float AS rot, rt.width::float AS w, rt.depth::float AS d FROM virtual_rack r LEFT JOIN virtual_rack_type rt ON rt.id = r.rack_type_id WHERE r.visible='y' """))).mappings()] return {"ok": True, "space": space, "crate_types": ctypes, "crates": crates, "racks": racks} @router.get("/reports") async def reports(ident=Depends(require_token), db=Depends(get_db)): summary = dict((await db.execute(text(""" SELECT count(*) AS orders, coalesce(sum(total),0)::float AS revenue, coalesce(avg(total),0)::float AS avg_order FROM sales """))).mappings().first()) by_month = [dict(r) for r in (await db.execute(text(""" SELECT to_char(date_trunc('month', sale_date),'YYYY-MM') AS month, count(*) AS orders, coalesce(sum(total),0)::float AS revenue FROM sales WHERE sale_date IS NOT NULL GROUP BY 1 ORDER BY 1 DESC LIMIT 12 """))).mappings()] top = [dict(r) for r in (await db.execute(text(""" SELECT item_name, count(*) AS qty, coalesce(sum(line_total),0)::float AS revenue FROM sale_items WHERE item_name IS NOT NULL GROUP BY item_name ORDER BY revenue DESC LIMIT 10 """))).mappings()] return {"ok": True, "summary": summary, "by_month": by_month, "top": top} # ── Reports workbench — parameterized + lazy (each chart fetched on demand) ─────────────────── # migrated history is status='paid'; new POS sales are 'completed' — count both, exclude open holds _DONE = "s.status IN ('completed','paid')" def _period(days: int, frm, to): if frm and to: return "s.sale_date >= :p_from AND s.sale_date < (:p_to::date + 1)", {"p_from": frm, "p_to": to} if days and days > 0: return "s.sale_date >= now() - make_interval(days => :p_days)", {"p_days": days} return "TRUE", {} # days<=0 = all time @router.get("/report/kpis") async def report_kpis(days: int = Query(30), frm: str | None = Query(None, alias="from"), to: str | None = None, ident=Depends(require_token), db=Depends(get_db)): where, p = _period(days, frm, to) row = dict((await db.execute(text(f""" SELECT count(*) AS sales, coalesce(sum(s.total),0)::float AS revenue, coalesce(sum(s.discount_amount),0)::float AS discounts, coalesce(sum(s.tax_amount),0)::float AS tax, coalesce(avg(s.total),0)::float AS avg_sale, count(DISTINCT s.customer_id) FILTER (WHERE s.customer_id IS NOT NULL AND s.customer_id<>0) AS customers FROM sales s WHERE {where} AND {_DONE}"""), p)).mappings().first()) row["units"] = (await db.execute(text( f"SELECT coalesce(sum(si.qty),0) FROM sale_items si JOIN sales s ON s.id=si.sale_id WHERE {where} AND {_DONE}" ), p)).scalar() snap = dict((await db.execute(text(""" SELECT count(*) FILTER (WHERE in_stock) AS in_stock, coalesce(sum(price) FILTER (WHERE in_stock),0)::float AS stock_value, count(*) FILTER (WHERE in_stock AND crate_id IS NOT NULL) AS located FROM inventory WHERE store_id=1"""))).mappings().first()) row["in_stock"] = snap["in_stock"]; row["stock_value"] = snap["stock_value"]; row["located"] = snap["located"] row["open_laybys"] = (await db.execute(text("SELECT count(*) FROM sales WHERE payment_status='hold'"))).scalar() return row @router.get("/report/series") async def report_series(days: int = Query(30), bucket: str = Query("day"), frm: str | None = Query(None, alias="from"), to: str | None = None, ident=Depends(require_token), db=Depends(get_db)): where, p = _period(days, frm, to) bucket = bucket if bucket in ("day", "week", "month") else "day" fmt = {"day": "YYYY-MM-DD", "week": 'IYYY-"W"IW', "month": "YYYY-MM"}[bucket] rows = [dict(r) for r in (await db.execute(text(f""" SELECT to_char(date_trunc('{bucket}', s.sale_date), '{fmt}') AS bucket, count(*) AS sales, coalesce(sum(s.total),0)::float AS revenue FROM sales s WHERE {where} AND {_DONE} AND s.sale_date IS NOT NULL GROUP BY 1 ORDER BY 1"""), p)).mappings()] return {"series": rows, "bucket": bucket} _BREAKDOWN = { "payment": "SELECT coalesce(nullif(s.payment_method,''),'?') AS name, count(*) AS count, coalesce(sum(s.total),0)::float AS revenue FROM sales s WHERE {where} AND {done} GROUP BY 1 ORDER BY {order} DESC LIMIT {lim}", "weekday": "SELECT trim(to_char(s.sale_date,'Dy')) AS name, extract(dow from s.sale_date) AS _o, count(*) AS count, coalesce(sum(s.total),0)::float AS revenue FROM sales s WHERE {where} AND {done} GROUP BY 1,2 ORDER BY _o", "hour": "SELECT to_char(s.sale_date,'HH24')||'h' AS name, count(*) AS count, coalesce(sum(s.total),0)::float AS revenue FROM sales s WHERE {where} AND {done} AND s.sale_date IS NOT NULL GROUP BY 1 ORDER BY 1", "format": "SELECT coalesce(nullif(f.name,''),'—') AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku JOIN disc_release_format f ON f.release_id=i.release_id WHERE {where} AND {done} GROUP BY 1 ORDER BY {order} DESC LIMIT {lim}", "condition": "SELECT coalesce(nullif(i.condition,''),'—') AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku WHERE {where} AND {done} GROUP BY 1 ORDER BY {order} DESC LIMIT {lim}", "kind": "SELECT coalesce(nullif(i.kind,''),'—') AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku WHERE {where} AND {done} GROUP BY 1 ORDER BY {order} DESC LIMIT {lim}", "genre": "SELECT g.genre_name AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku JOIN disc_release_genre g ON g.release_id=i.release_id WHERE {where} AND {done} AND g.genre_name<>'' GROUP BY 1 ORDER BY {order} DESC LIMIT {lim}", "style": "SELECT st.style_name AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku JOIN disc_release_style st ON st.release_id=i.release_id WHERE {where} AND {done} AND st.style_name<>'' GROUP BY 1 ORDER BY {order} DESC LIMIT {lim}", } @router.get("/report/breakdown") async def report_breakdown(dim: str, days: int = Query(30), metric: str = Query("revenue"), limit: int = Query(12), frm: str | None = Query(None, alias="from"), to: str | None = None, ident=Depends(require_token), db=Depends(get_db)): if dim not in _BREAKDOWN: raise HTTPException(400, "bad dim") where, p = _period(days, frm, to) sql = _BREAKDOWN[dim].format(where=where, done=_DONE, order=("revenue" if metric == "revenue" else "count"), lim=max(1, min(limit, 50))) return {"dim": dim, "metric": metric, "rows": [dict(r) for r in (await db.execute(text(sql), p)).mappings()]} _TOP = { "release": "SELECT coalesce(min(si.item_name),'Release '||i.release_id::text) AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku WHERE {where} AND {done} AND i.release_id IS NOT NULL GROUP BY i.release_id ORDER BY revenue DESC LIMIT {lim}", "artist": "SELECT da.name AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku JOIN disc_release_artist ra ON ra.release_id=i.release_id JOIN disc_artist da ON da.id=ra.artist_id WHERE {where} AND {done} AND da.name<>'' GROUP BY da.name ORDER BY revenue DESC LIMIT {lim}", "label": "SELECT rl.label_name AS name, count(*) AS count, coalesce(sum(si.line_total),0)::float AS revenue FROM sales s JOIN sale_items si ON si.sale_id=s.id JOIN inventory i ON i.sku=si.sku JOIN disc_release_label rl ON rl.release_id=i.release_id WHERE {where} AND {done} AND rl.label_name<>'' GROUP BY rl.label_name ORDER BY revenue DESC LIMIT {lim}", "customer": "SELECT trim(c.first_name||' '||coalesce(c.last_name,'')) AS name, count(*) AS count, coalesce(sum(s.total),0)::float AS revenue FROM sales s JOIN customer c ON c.id=s.customer_id WHERE {where} AND {done} AND s.customer_id IS NOT NULL AND s.customer_id<>0 GROUP BY c.id ORDER BY revenue DESC LIMIT {lim}", } @router.get("/report/top") async def report_top(dim: str, days: int = Query(30), limit: int = Query(10), frm: str | None = Query(None, alias="from"), to: str | None = None, ident=Depends(require_token), db=Depends(get_db)): if dim not in _TOP: raise HTTPException(400, "bad dim") where, p = _period(days, frm, to) sql = _TOP[dim].format(where=where, done=_DONE, lim=max(1, min(limit, 50))) return {"dim": dim, "rows": [dict(r) for r in (await db.execute(text(sql), p)).mappings()]} _STOCK = { "format": "SELECT coalesce(nullif(f.name,''),'—') AS name, count(DISTINCT i.sku) AS count, coalesce(sum(i.price),0)::float AS value FROM inventory i JOIN disc_release_format f ON f.release_id=i.release_id WHERE i.in_stock AND i.store_id=1 GROUP BY 1 ORDER BY count DESC LIMIT 12", "genre": "SELECT g.genre_name AS name, count(DISTINCT i.sku) AS count, coalesce(sum(i.price),0)::float AS value FROM inventory i JOIN disc_release_genre g ON g.release_id=i.release_id WHERE i.in_stock AND i.store_id=1 AND g.genre_name<>'' GROUP BY 1 ORDER BY count DESC LIMIT 12", "condition": "SELECT coalesce(nullif(condition,''),'—') AS name, count(*) AS count, coalesce(sum(price),0)::float AS value FROM inventory WHERE in_stock AND store_id=1 GROUP BY 1 ORDER BY count DESC", "kind": "SELECT coalesce(nullif(kind,''),'—') AS name, count(*) AS count, coalesce(sum(price),0)::float AS value FROM inventory WHERE in_stock AND store_id=1 GROUP BY 1 ORDER BY count DESC", "price": "SELECT CASE WHEN price<10 THEN '< $10' WHEN price<20 THEN '$10–20' WHEN price<35 THEN '$20–35' WHEN price<60 THEN '$35–60' WHEN price<100 THEN '$60–100' ELSE '$100+' END AS name, min(price) AS _o, count(*) AS count, coalesce(sum(price),0)::float AS value FROM inventory WHERE in_stock AND store_id=1 AND price IS NOT NULL GROUP BY 1 ORDER BY _o", } @router.get("/report/stock") async def report_stock(dim: str, ident=Depends(require_token), db=Depends(get_db)): if dim not in _STOCK: raise HTTPException(400, "bad dim") return {"dim": dim, "rows": [dict(r) for r in (await db.execute(text(_STOCK[dim]))).mappings()]} # ── System — read-only Postgres health (admin). No maintenance buttons: PG autovacuums, infra owns backups. @router.get("/system") async def system(ident=Depends(require_admin), db=Depends(get_db)): info = dict((await db.execute(text(""" SELECT pg_database_size(current_database()) AS db_bytes, split_part(version(), ' ', 2) AS pg_version, extract(epoch FROM (now() - pg_postmaster_start_time()))::bigint AS uptime_s, (SELECT count(*) FROM pg_stat_activity WHERE datname = current_database()) AS conns, (SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() AND state = 'active') AS active, (SELECT round(sum(blks_hit) * 100.0 / nullif(sum(blks_hit + blks_read), 0), 1) FROM pg_stat_database WHERE datname = current_database()) AS cache_hit """))).mappings().first()) tables = [dict(r) for r in (await db.execute(text(""" SELECT relname AS name, pg_total_relation_size(relid) AS bytes, n_live_tup AS rows FROM pg_stat_user_tables ORDER BY pg_total_relation_size(relid) DESC LIMIT 15 """))).mappings()] fresh = dict((await db.execute(text(""" SELECT (SELECT max(updated_at) FROM inventory) AS inventory_updated, (SELECT max(sale_date) FROM sales) AS last_sale, (SELECT count(*) FROM inventory) AS inventory_rows, (SELECT count(*) FROM disc_release) AS releases """))).mappings().first()) return {"db": info, "tables": tables, "fresh": fresh} @router.get("/slow-queries") async def slow_queries(ident=Depends(require_admin), db=Depends(get_db)): """The Query Monitor replacement — slowest SQL by mean time, straight from pg_stat_statements (the real tool).""" try: rows = [dict(r) for r in (await db.execute(text(""" SELECT left(query, 200) AS query, calls, round(mean_exec_time::numeric, 2) AS mean_ms, round(total_exec_time::numeric, 0) AS total_ms, round(max_exec_time::numeric, 1) AS max_ms, rows FROM pg_stat_statements WHERE query NOT ILIKE '%pg_stat_statements%' AND query NOT ILIKE 'BEGIN%' AND query NOT ILIKE 'COMMIT%' AND query NOT ILIKE 'SET %' ORDER BY mean_exec_time DESC LIMIT 30 """))).mappings()] return {"ok": True, "queries": rows} except Exception as e: return {"ok": False, "reason": f"pg_stat_statements unavailable: {e}"} @router.post("/slow-queries/reset") async def slow_queries_reset(ident=Depends(require_admin), db=Depends(get_db)): try: await db.execute(text("SELECT pg_stat_statements_reset()")) await db.commit() return {"ok": True} except Exception as e: return {"ok": False, "reason": str(e)} @router.get("/payment-providers") async def payment_providers(ident=Depends(require_token), db=Depends(get_db)): """Which processors this store can take — the checkout offers exactly these (provider-agnostic: the anti-Ghost).""" from . import payments return {"ok": True, "providers": await payments.enabled_providers(db)} class PayIn(BaseModel): provider: str = "mock" token: str # the tokenised card from the frontend SDK (never raw card data) amount: float # dollars currency: str = "AUD" reference: str | None = None @router.post("/pay") async def pay(body: PayIn, ident=Depends(require_admin), db=Depends(get_db)): """Virtual terminal — charge a tokenised card via ANY configured processor (Square/Stripe/PayPal/mock). Card data is tokenised client-side; we only ever see the token. Amount in dollars → charged in cents.""" from . import payments return await payments.charge(db, body.provider, body.token, int(round(body.amount * 100)), body.currency, body.reference) # ── Content pages (the WordPress Pages replacement) ─────────────────────────────────────────── class PageIn(BaseModel): slug: str title: str = "" body: str = "" published: bool = True seo_title: str | None = None seo_description: str | None = None nav_order: int = 0 @router.get("/pages") async def pages_list(ident=Depends(require_token), db=Depends(get_db)): rows = [dict(r) for r in (await db.execute(text( "SELECT id, slug, title, published, nav_order, updated_at, length(body) AS body_len " "FROM page ORDER BY nav_order, slug"))).mappings()] return {"ok": True, "pages": rows} @router.get("/pages/{slug}") async def page_get(slug: str, ident=Depends(require_token), db=Depends(get_db)): r = (await db.execute(text("SELECT * FROM page WHERE slug = :s"), {"s": slug})).mappings().first() return {"ok": True, "page": dict(r) if r else None} @router.post("/pages") async def page_save(body: PageIn, ident=Depends(require_admin), db=Depends(get_db)): slug = re.sub(r"[^a-z0-9-]+", "-", body.slug.lower()).strip("-") or "page" await db.execute(text(""" INSERT INTO page (slug, title, body, published, seo_title, seo_description, nav_order, updated_at) VALUES (:s, :t, :b, :p, :st, :sd, :no, now()) ON CONFLICT (slug) DO UPDATE SET title=:t, body=:b, published=:p, seo_title=:st, seo_description=:sd, nav_order=:no, updated_at=now()"""), {"s": slug, "t": body.title, "b": body.body, "p": body.published, "st": body.seo_title, "sd": body.seo_description, "no": body.nav_order}) await db.commit() return {"ok": True, "slug": slug} @router.delete("/pages/{slug}") async def page_delete(slug: str, ident=Depends(require_admin), db=Depends(get_db)): await db.execute(text("DELETE FROM page WHERE slug = :s"), {"s": slug}) await db.commit() return {"ok": True} # ── 301 redirect map (old WP/Woo URLs → new) — SEO preservation at cutover ───────────────────── class RedirectRow(BaseModel): from_path: str to_path: str code: int = 301 class RedirectsIn(BaseModel): redirects: list[RedirectRow] @router.get("/redirects") async def redirects_list(ident=Depends(require_admin), db=Depends(get_db)): rows = [dict(r) for r in (await db.execute(text( "SELECT from_path, to_path, code, hits FROM url_redirect ORDER BY hits DESC, from_path LIMIT 500"))).mappings()] return {"ok": True, "redirects": rows} @router.post("/redirects") async def redirects_save(body: RedirectsIn, ident=Depends(require_admin), db=Depends(get_db)): n = 0 for r in body.redirects: fp, tp = r.from_path.strip(), r.to_path.strip() if not fp or not tp or fp == tp: continue if not fp.startswith("/"): fp = "/" + fp await db.execute(text("""INSERT INTO url_redirect (from_path, to_path, code) VALUES (:f, :t, :c) ON CONFLICT (from_path) DO UPDATE SET to_path = :t, code = :c"""), {"f": fp, "t": tp, "c": r.code or 301}) n += 1 await db.commit() return {"ok": True, "saved": n} @router.delete("/redirects") async def redirect_delete(from_path: str = Query(...), ident=Depends(require_admin), db=Depends(get_db)): await db.execute(text("DELETE FROM url_redirect WHERE from_path = :f"), {"f": from_path}) await db.commit() return {"ok": True} # ── Discogs Marketplace Manager (manage YOUR listings + orders from StoreGod) ────────────────── @router.get("/discogs") async def discogs_status(ident=Depends(require_token), db=Depends(get_db)): from . import discogs_mp return await discogs_mp.status(db) @router.get("/discogs/orders") async def discogs_orders(page: int = 1, ident=Depends(require_token), db=Depends(get_db)): from . import discogs_mp return await discogs_mp.orders(db, page) class DiscogsListIn(BaseModel): status: str = "For Sale" # 'Draft' = safe/not live (use to test); 'For Sale' goes live price: float | None = None @router.post("/discogs/list/{sku}") async def discogs_list(sku: str, body: DiscogsListIn, ident=Depends(require_admin), db=Depends(get_db)): """List a StoreGod item on Discogs. Records only (needs a release_id). Stores the listing id for cross-channel sync.""" from . import discogs_mp row = (await db.execute(text( "SELECT release_id, price::float AS price, condition, sleeve_cond, discogs_listing_id " "FROM inventory WHERE sku = :s AND store_id = :st"), {"s": sku, "st": ident["store_id"]})).mappings().first() if not row: raise HTTPException(404, "not found") if not row["release_id"]: return {"ok": False, "error": "no release_id — Discogs listings need a Discogs release"} if row["discogs_listing_id"]: return {"ok": False, "error": f"already listed (#{row['discogs_listing_id']})"} price = body.price if body.price is not None else (row["price"] or 0) if price <= 0: return {"ok": False, "error": "set a price first"} res = await discogs_mp.create_listing(db, row["release_id"], price, row["condition"], row["sleeve_cond"], body.status) if res.get("ok"): await db.execute(text("UPDATE inventory SET discogs_listing_id = :l WHERE sku = :s"), {"l": res["listing_id"], "s": sku}) await db.commit() return res @router.delete("/discogs/list/{sku}") async def discogs_delist(sku: str, ident=Depends(require_admin), db=Depends(get_db)): from . import discogs_mp lid = (await db.execute(text("SELECT discogs_listing_id FROM inventory WHERE sku = :s AND store_id = :st"), {"s": sku, "st": ident["store_id"]})).scalar() if not lid: return {"ok": True, "note": "not listed"} res = await discogs_mp.delete_listing(db, lid) if res.get("ok"): await db.execute(text("UPDATE inventory SET discogs_listing_id = NULL WHERE sku = :s"), {"s": sku}) await db.commit() return res @router.post("/discogs/sync") async def discogs_sync(ident=Depends(require_admin), db=Depends(get_db)): """Pull new Discogs orders → mark sold + record the sale (idempotent). The other half of anti-oversell.""" from . import discogs_mp return await discogs_mp.sync_orders(db) @router.get("/orders") async def orders(ident=Depends(require_token), db=Depends(get_db)): base = await vault.get_secret(db, "woo_base_url") ck = await vault.get_secret(db, "woo_consumer_key") cs = await vault.get_secret(db, "woo_consumer_secret") if not (base and ck and cs): return {"ok": False, "reason": "WooCommerce not connected — add keys in Connections"} try: async with httpx.AsyncClient(timeout=20, follow_redirects=True) as c: r = await c.get(base.rstrip("/") + "/wp-json/wc/v3/orders", params={"per_page": 20, "orderby": "date", "order": "desc"}, auth=(ck, cs)) except Exception as e: return {"ok": False, "reason": f"connection error: {e}"} if r.status_code != 200: return {"ok": False, "reason": f"Woo API HTTP {r.status_code}"} out = [] for o in r.json(): b = o.get("billing", {}) name = (b.get("first_name", "") + " " + b.get("last_name", "")).strip() or b.get("email", "") out.append({"id": o["id"], "number": o.get("number"), "status": o.get("status"), "total": o.get("total"), "currency": o.get("currency"), "date": (o.get("date_created") or "")[:10], "customer": name, "items": len(o.get("line_items", []))}) return {"ok": True, "orders": out}