ef37063b2e
(1) Inventory edit modal gains Cost + New/Used + a live Margin readout (price−cost, %); inv_edit accepts cost_price + condition_type; inventory list returns them. (2) Inertia/Warner SOH list = an xlsx that doubles as the order form (fill ORDER QTY, upload). New /admin/intake/distro-sheet[/preview]: openpyxl parses header-mapped columns (ORDER QTY/UPC/ARTIST/TITLE/ FORMAT/PPD), takes rows with QTY>0 → shared _ingest_distro core (refactored out of the RareWaves path) → staged NEW stock with cost_price=PPD, kind from FORMAT (cd/vinyl), catno kept. Upload+preview UI in New stock. DistroItem gains kind+catno; RareWaves path now passes kind too. Verified on the real 3,368-row Warner sheet: 3 ordered titles → 4 copies, all resolved, kind=cd, cost from PPD. Added openpyxl to requirements. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
909 lines
40 KiB
Python
909 lines
40 KiB
Python
import asyncio
|
|
import base64
|
|
import csv
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import text
|
|
|
|
from . import vault
|
|
from .auth import require_token
|
|
from .db import get_db
|
|
|
|
# Intake — stage new stock from three sources, all feeding one _stage() core:
|
|
# 1. internal lookup over the local Discogs mirror (the daily driver) + Discogs-API fallback
|
|
# 2. a Discogs collection folder (OAuth token's own collection)
|
|
# 3. a Google Sheet (public CSV export — no Google creds; SKU from the form timestamp)
|
|
# Staging enriches title/artist/cover-art on demand; a cache miss fetches the release from the
|
|
# Discogs API and grows the local mirror (release + artist + label + thumb).
|
|
router = APIRouter(prefix="/admin/intake", tags=["intake"])
|
|
|
|
DISCOGS = "https://api.discogs.com"
|
|
UA = "RecordGod/0.1 +https://recordgod.com"
|
|
|
|
|
|
def _new_sku() -> str:
|
|
return "MRP-" + base64.b32encode(secrets.token_bytes(5)).decode().rstrip("=")[:8]
|
|
|
|
|
|
async def _client(db):
|
|
tok = await vault.get_secret(db, "discogs_token")
|
|
headers = {"User-Agent": UA}
|
|
if tok:
|
|
headers["Authorization"] = f"Discogs token={tok}"
|
|
return httpx.AsyncClient(base_url=DISCOGS, headers=headers, timeout=20,
|
|
follow_redirects=True), bool(tok)
|
|
|
|
|
|
_user_cache = {}
|
|
|
|
|
|
async def _discogs_user(db):
|
|
if "u" not in _user_cache:
|
|
c, ok = await _client(db)
|
|
async with c:
|
|
if not ok:
|
|
return None
|
|
r = await c.get("/oauth/identity")
|
|
_user_cache["u"] = r.json().get("username") if r.status_code == 200 else None
|
|
return _user_cache["u"]
|
|
|
|
|
|
# --- enrichment ------------------------------------------------------------
|
|
|
|
async def _cache_hit(db, rid):
|
|
return (await db.execute(text(
|
|
"SELECT title, artist, thumb, weight FROM disc_cache WHERE release_id=:r"),
|
|
{"r": rid})).mappings().first()
|
|
|
|
|
|
async def _enrich(db, rid):
|
|
"""{title, artist, thumb, weight} for a release_id. disc_cache first; on a miss, fetch the
|
|
release from Discogs and grow the local mirror (best-effort — never fails the stage)."""
|
|
if not rid:
|
|
return None
|
|
hit = await _cache_hit(db, rid)
|
|
if hit:
|
|
return dict(hit)
|
|
return await _fetch_release(db, rid)
|
|
|
|
|
|
async def _fetch_release(db, rid):
|
|
"""ALWAYS hit the Discogs API for this release and grow the mirror (release+artist+label+thumb),
|
|
ignoring any existing cache row. Used by Heal to backfill missing cover art / metadata."""
|
|
try:
|
|
c, ok = await _client(db)
|
|
async with c:
|
|
if not ok:
|
|
return None
|
|
r = await c.get(f"/releases/{rid}")
|
|
if r.status_code != 200:
|
|
return None
|
|
d = r.json()
|
|
except Exception:
|
|
return None
|
|
artist = ", ".join(a["name"] for a in d.get("artists", []) if a.get("name")) or None
|
|
title = d.get("title")
|
|
thumb = (d.get("images") or [{}])[0].get("uri") or d.get("thumb") or None
|
|
weight = int(d["estimated_weight"]) if d.get("estimated_weight") else None
|
|
year = d.get("year") or None
|
|
await _grow_mirror(db, d, artist, title, thumb, weight, year)
|
|
return {"title": title, "artist": artist, "thumb": thumb, "weight": weight}
|
|
|
|
|
|
async def _grow_mirror(db, d, artist, title, thumb, weight, year):
|
|
"""Upsert the fetched release + its artists/labels into disc_* so it's findable next time
|
|
and the storefront has artist/label/cover. Minimal columns; best-effort."""
|
|
rid = d["id"]
|
|
await db.execute(text("""
|
|
INSERT INTO disc_cache (release_id, title, artist, thumb, weight)
|
|
VALUES (:r,:t,:a,:th,:w) ON CONFLICT (release_id) DO UPDATE
|
|
SET title=EXCLUDED.title, artist=EXCLUDED.artist,
|
|
thumb=COALESCE(EXCLUDED.thumb, disc_cache.thumb)"""),
|
|
{"r": rid, "t": title, "a": artist, "th": thumb, "w": weight})
|
|
await db.execute(text("""
|
|
INSERT INTO disc_release (id, title, artists_sort, country, year, thumb, master_id)
|
|
VALUES (:r,:t,:a,:c,:y,:th,:m) ON CONFLICT (id) DO UPDATE
|
|
SET title=EXCLUDED.title, artists_sort=EXCLUDED.artists_sort,
|
|
thumb=COALESCE(EXCLUDED.thumb, disc_release.thumb)"""),
|
|
{"r": rid, "t": title, "a": artist, "c": d.get("country"), "y": year,
|
|
"th": thumb, "m": (d.get("master_id") or None)})
|
|
for pos, a in enumerate(d.get("artists", []) or [], 1):
|
|
if not a.get("id"):
|
|
continue
|
|
await db.execute(text(
|
|
"INSERT INTO disc_artist (id,name) VALUES (:i,:n) ON CONFLICT (id) DO NOTHING"),
|
|
{"i": a["id"], "n": a.get("name")})
|
|
await db.execute(text("""INSERT INTO disc_release_artist (release_id,artist_id,artist_name,position)
|
|
SELECT :r,:i,:n,:p WHERE NOT EXISTS (SELECT 1 FROM disc_release_artist
|
|
WHERE release_id=:r AND artist_id=:i)"""),
|
|
{"r": rid, "i": a["id"], "n": a.get("name"), "p": str(pos)})
|
|
for lab in d.get("labels", []) or []:
|
|
if not lab.get("id"):
|
|
continue
|
|
await db.execute(text(
|
|
"INSERT INTO disc_label (id,name) VALUES (:i,:n) ON CONFLICT (id) DO NOTHING"),
|
|
{"i": lab["id"], "n": lab.get("name")})
|
|
await db.execute(text("""INSERT INTO disc_release_label (release_id,label_id,label_name,catno)
|
|
SELECT :r,:i,:n,:c WHERE NOT EXISTS (SELECT 1 FROM disc_release_label
|
|
WHERE release_id=:r AND label_id=:i)"""),
|
|
{"r": rid, "i": lab["id"], "n": lab.get("name"), "c": lab.get("catno")})
|
|
|
|
|
|
# --- staging core ----------------------------------------------------------
|
|
|
|
async def _stage(db, store_id, rid, sku, condition, sleeve, price, notes, kind="vinyl",
|
|
identifier=None):
|
|
meta = await _enrich(db, rid) if rid else None
|
|
sku = sku or _new_sku()
|
|
await db.execute(text("""
|
|
INSERT INTO inventory (sku, store_id, kind, release_id, identifier, title, price,
|
|
condition, sleeve_cond, weight_g, notes, staged, status)
|
|
VALUES (:sku,:sid,:kind,:rid,:ident,:title,:price,:cond,:sleeve,:wt,:notes,true,'staged')
|
|
ON CONFLICT (sku) DO UPDATE SET
|
|
release_id=EXCLUDED.release_id, title=EXCLUDED.title, price=EXCLUDED.price,
|
|
condition=EXCLUDED.condition, sleeve_cond=EXCLUDED.sleeve_cond,
|
|
notes=EXCLUDED.notes, updated_at=now()"""),
|
|
{"sku": sku, "sid": store_id, "kind": kind, "rid": rid, "ident": identifier,
|
|
"title": (meta or {}).get("title"), "price": price, "cond": condition,
|
|
"sleeve": sleeve, "wt": (meta or {}).get("weight"), "notes": notes})
|
|
return {"sku": sku, "title": (meta or {}).get("title"),
|
|
"artist": (meta or {}).get("artist"), "enriched": bool(meta)}
|
|
|
|
|
|
class StageIn(BaseModel):
|
|
release_id: int | None = None
|
|
identifier: str | None = None
|
|
condition: str | None = "VG+"
|
|
sleeve: str | None = None
|
|
price: float | None = None
|
|
sku: str | None = None
|
|
notes: str | None = None
|
|
kind: str = "vinyl"
|
|
|
|
|
|
@router.post("/stage")
|
|
async def stage(body: StageIn, ident=Depends(require_token), db=Depends(get_db)):
|
|
res = await _stage(db, ident["store_id"], body.release_id, body.sku, body.condition,
|
|
body.sleeve, body.price, body.notes, body.kind, body.identifier)
|
|
await db.commit()
|
|
return {"ok": True, "staged": True, **res}
|
|
|
|
|
|
# --- 1. internal lookup ----------------------------------------------------
|
|
|
|
@router.get("/search")
|
|
async def search(q: str = Query(""), label: str = Query(""), year: int | None = None,
|
|
country: str = Query(""), fmt: str = Query(""),
|
|
ident=Depends(require_token), db=Depends(get_db)):
|
|
"""Search the local Discogs mirror — fast + typo-tolerant via pg_trgm on search_text
|
|
(word_similarity: multi-word, any order, fuzzy; GIN-indexed; threshold 0.3 pinned on the DB)."""
|
|
where, params, order = ["TRUE"], {}, "dr.title"
|
|
if q:
|
|
where.append(":q <% dr.search_text") # typo-tolerant word match, GIN-accelerated
|
|
order = ":q <<-> dr.search_text" # rank by word distance (closest first)
|
|
params["q"] = q
|
|
if year:
|
|
where.append("dr.year = :y"); params["y"] = year
|
|
if country:
|
|
where.append("dr.country ILIKE :c"); params["c"] = f"%{country}%"
|
|
if label:
|
|
where.append("EXISTS (SELECT 1 FROM disc_release_label rl WHERE rl.release_id=dr.id AND rl.label_name ILIKE :lab)")
|
|
params["lab"] = f"%{label}%"
|
|
if fmt:
|
|
where.append("EXISTS (SELECT 1 FROM disc_release_format f WHERE f.release_id=dr.id AND f.name ILIKE :fmt)")
|
|
params["fmt"] = f"%{fmt}%"
|
|
rows = [dict(r) for r in (await db.execute(text(f"""
|
|
SELECT dr.id AS release_id, dr.title, dr.artists_sort AS artist, dr.year, dr.country,
|
|
COALESCE(dc.thumb, dr.thumb) AS thumb,
|
|
(SELECT string_agg(label_name || COALESCE(' ('||catno||')',''), ', ')
|
|
FROM disc_release_label WHERE release_id=dr.id) AS label,
|
|
(SELECT string_agg(name, ', ') FROM disc_release_format WHERE release_id=dr.id) AS format,
|
|
EXISTS (SELECT 1 FROM inventory i WHERE i.release_id=dr.id AND i.in_stock AND i.store_id=1) AS in_stock
|
|
FROM disc_release dr LEFT JOIN disc_cache dc ON dc.release_id=dr.id
|
|
WHERE {' AND '.join(where)}
|
|
ORDER BY {order} LIMIT 40"""), params)).mappings()]
|
|
return {"items": rows, "source": "local"}
|
|
|
|
|
|
@router.get("/discogs/search")
|
|
async def discogs_search(q: str = Query(...), ident=Depends(require_token), db=Depends(get_db)):
|
|
"""Fallback for brand-new titles not yet in the local mirror — Discogs API search."""
|
|
c, ok = await _client(db)
|
|
async with c:
|
|
if not ok:
|
|
return {"items": [], "source": "discogs", "error": "no Discogs token saved"}
|
|
r = await c.get("/database/search", params={"q": q, "type": "release", "per_page": 40})
|
|
if r.status_code != 200:
|
|
return {"items": [], "source": "discogs", "error": f"HTTP {r.status_code}"}
|
|
out = []
|
|
for d in r.json().get("results", []):
|
|
title = d.get("title", "")
|
|
artist, _, rest = title.partition(" - ")
|
|
out.append({"release_id": d.get("id"), "title": rest or title, "artist": artist,
|
|
"year": d.get("year"), "country": d.get("country"),
|
|
"thumb": d.get("thumb"), "label": ", ".join(d.get("label", []) or []),
|
|
"format": ", ".join(d.get("format", []) or []), "in_stock": False})
|
|
return {"items": out, "source": "discogs"}
|
|
|
|
|
|
# --- 2. Discogs collection folder -----------------------------------------
|
|
|
|
@router.get("/discogs/folders")
|
|
async def folders(ident=Depends(require_token), db=Depends(get_db)):
|
|
user = await _discogs_user(db)
|
|
if not user:
|
|
return {"folders": [], "error": "no Discogs token saved"}
|
|
c, _ = await _client(db)
|
|
async with c:
|
|
r = await c.get(f"/users/{user}/collection/folders")
|
|
fs = [{"id": f["id"], "name": f["name"], "count": f["count"]}
|
|
for f in r.json().get("folders", []) if f["count"]]
|
|
return {"user": user, "folders": fs}
|
|
|
|
|
|
async def _field_map(c, user):
|
|
"""Discogs collection custom-field ids → our slots (media/sleeve/price/notes), matched by name."""
|
|
r = await c.get(f"/users/{user}/collection/fields")
|
|
out = {}
|
|
for f in r.json().get("fields", []):
|
|
n = f.get("name", "").lower()
|
|
if "sleeve" in n or "cover" in n:
|
|
out[f["id"]] = "sleeve"
|
|
elif "media" in n or "record" in n or "grade" in n or "condition" in n:
|
|
out[f["id"]] = "media"
|
|
elif "price" in n or "$" in n:
|
|
out[f["id"]] = "price"
|
|
elif "note" in n or "comment" in n:
|
|
out[f["id"]] = "notes"
|
|
return out
|
|
|
|
|
|
@router.get("/discogs/folder/{folder_id}")
|
|
async def folder_items(folder_id: int, page: int = Query(1, ge=1),
|
|
ident=Depends(require_token), db=Depends(get_db)):
|
|
user = await _discogs_user(db)
|
|
if not user:
|
|
return {"items": [], "error": "no Discogs token saved"}
|
|
c, _ = await _client(db)
|
|
async with c:
|
|
fmap = await _field_map(c, user)
|
|
r = await c.get(f"/users/{user}/collection/folders/{folder_id}/releases",
|
|
params={"per_page": 100, "page": page, "sort": "added", "sort_order": "desc"})
|
|
j = r.json()
|
|
items = []
|
|
for it in j.get("releases", []):
|
|
bi = it.get("basic_information", {})
|
|
slots = {}
|
|
for note in it.get("notes", []) or []:
|
|
slot = fmap.get(note.get("field_id"))
|
|
if slot and note.get("value"):
|
|
slots[slot] = note["value"]
|
|
items.append({
|
|
"release_id": bi.get("id"),
|
|
"title": bi.get("title"),
|
|
"artist": ", ".join(a["name"] for a in bi.get("artists", []) if a.get("name")),
|
|
"year": bi.get("year"), "thumb": bi.get("thumb"),
|
|
"media": slots.get("media"), "sleeve": slots.get("sleeve"),
|
|
"price": slots.get("price"), "notes": slots.get("notes"),
|
|
})
|
|
pg = j.get("pagination", {})
|
|
return {"items": items, "page": pg.get("page", page), "pages": pg.get("pages", 1)}
|
|
|
|
|
|
# --- 3. Google Sheet (public CSV export) ----------------------------------
|
|
|
|
def _csv_url(url: str) -> str | None:
|
|
m = re.search(r"/spreadsheets/d/([a-zA-Z0-9-_]+)", url)
|
|
if not m:
|
|
return None
|
|
gid = (re.search(r"[#&?]gid=(\d+)", url) or [None, "0"])[1]
|
|
return f"https://docs.google.com/spreadsheets/d/{m.group(1)}/export?format=csv&gid={gid}"
|
|
|
|
|
|
def _detect(headers):
|
|
"""Map a sheet's header row → our fields by keyword (handles the Google-Form layout)."""
|
|
col = {}
|
|
for i, h in enumerate(headers):
|
|
n = (h or "").lower()
|
|
if "release" in n and ("id" in n or "discogs" in n): col.setdefault("release_id", i)
|
|
elif n in ("release_id", "releaseid", "discogs"): col.setdefault("release_id", i)
|
|
elif "sleeve" in n or "cover" in n: col.setdefault("sleeve", i)
|
|
elif "media" in n or "record cond" in n or n == "condition" or "grade" in n: col.setdefault("media", i)
|
|
elif "price" in n or "aud" in n or "$" in n: col.setdefault("price", i)
|
|
elif "sku" in n: col.setdefault("sku", i)
|
|
elif "comment" in n or "note" in n: col.setdefault("notes", i)
|
|
elif "time" in n or "stamp" in n: col.setdefault("timestamp", i)
|
|
return col
|
|
|
|
|
|
def _sku_from_ts(ts: str) -> str | None:
|
|
# separator between date and time is T, space, or ", " depending on the sheet's locale format
|
|
m = re.match(r"(\d{4})-(\d{2})-(\d{2})[T,\s]+(\d{2}):(\d{2}):(\d{2})", ts or "")
|
|
if m:
|
|
return "".join(m.groups()) # YYYYMMDDHHMMSS — matches WowPlatter / migrated stock
|
|
m = re.match(r"(\d{1,2})/(\d{1,2})/(\d{4})\s+(\d{1,2}):(\d{2}):(\d{2})", ts or "")
|
|
if m:
|
|
d, mo, y, h, mi, s = m.groups()
|
|
return f"{y}{int(mo):02d}{int(d):02d}{int(h):02d}{mi}{s}"
|
|
return None
|
|
|
|
|
|
def _parse_rows(rows, col):
|
|
"""rows[0] is the header (skipped); col maps our field → 0-indexed column. Shared by the
|
|
public-CSV path (col from header detection) and the service-account path (col from the
|
|
stored Google column_mappings)."""
|
|
out = []
|
|
for r in rows[1:]:
|
|
def g(k):
|
|
i = col.get(k)
|
|
v = r[i] if i is not None and i < len(r) else None
|
|
return v.strip() if isinstance(v, str) and v.strip() else None
|
|
rid = g("release_id")
|
|
if not rid or not str(rid).isdigit():
|
|
continue
|
|
price = g("price")
|
|
try:
|
|
price = float(re.sub(r"[^0-9.]", "", price)) if price else None
|
|
except ValueError:
|
|
price = None
|
|
sku = g("sku") or _sku_from_ts(g("timestamp") or "")
|
|
out.append({"release_id": int(rid), "sku": sku, "price": price,
|
|
"media": g("media") or "VG+", "sleeve": g("sleeve"),
|
|
"notes": g("notes")})
|
|
return out
|
|
|
|
|
|
def _parse_sheet(text_csv):
|
|
rows = list(csv.reader(io.StringIO(text_csv)))
|
|
if not rows:
|
|
return [], {}
|
|
col = _detect(rows[0])
|
|
if "release_id" not in col:
|
|
return [], col
|
|
return _parse_rows(rows, col), col
|
|
|
|
|
|
async def _fetch_sheet(url):
|
|
csv_url = _csv_url(url)
|
|
if not csv_url:
|
|
return None, "not a Google Sheets URL"
|
|
async with httpx.AsyncClient(timeout=30, follow_redirects=True) as c:
|
|
r = await c.get(csv_url)
|
|
if r.status_code != 200 or r.text.lstrip().startswith("<"):
|
|
return None, "sheet not public (set Share → anyone with link can view)"
|
|
return r.text, None
|
|
|
|
|
|
class SheetIn(BaseModel):
|
|
url: str
|
|
|
|
|
|
@router.post("/sheet/preview")
|
|
async def sheet_preview(body: SheetIn, ident=Depends(require_token), db=Depends(get_db)):
|
|
csv_text, err = await _fetch_sheet(body.url)
|
|
if err:
|
|
return {"ok": False, "error": err}
|
|
rows, col = _parse_sheet(csv_text)
|
|
if not rows and "release_id" not in col:
|
|
return {"ok": False, "error": "no release_id column detected", "columns": col}
|
|
return {"ok": True, "total": len(rows), "columns": col, "sample": rows[:10]}
|
|
|
|
|
|
async def _bulk_stage(db, store_id, rows):
|
|
"""Stage sheet rows, SKIPPING any SKU already in inventory — so re-importing a 34k form-log
|
|
only touches genuinely-new stock (the bulk is already-migrated). Rows without a derived SKU
|
|
always stage (can't dedup them). Returns (staged, skipped)."""
|
|
skus = [r["sku"] for r in rows if r["sku"]]
|
|
existing = set()
|
|
if skus:
|
|
existing = {x[0] for x in (await db.execute(
|
|
text("SELECT sku FROM inventory WHERE sku = ANY(:s)"), {"s": skus}))}
|
|
staged = skipped = 0
|
|
for row in rows:
|
|
if row["sku"] and row["sku"] in existing:
|
|
skipped += 1
|
|
continue
|
|
await _stage(db, store_id, row["release_id"], row["sku"], row["media"],
|
|
row["sleeve"], row["price"], row["notes"])
|
|
staged += 1
|
|
if staged % 25 == 0:
|
|
await db.commit()
|
|
await asyncio.sleep(0.2) # ponytail: gentle on the Discogs API for cache-miss enrich
|
|
await db.commit()
|
|
return staged, skipped
|
|
|
|
|
|
@router.post("/sheet/import")
|
|
async def sheet_import(body: SheetIn, ident=Depends(require_token), db=Depends(get_db)):
|
|
csv_text, err = await _fetch_sheet(body.url)
|
|
if err:
|
|
return {"ok": False, "error": err}
|
|
rows, _ = _parse_sheet(csv_text)
|
|
staged, skipped = await _bulk_stage(db, ident["store_id"], rows)
|
|
return {"ok": True, "staged": staged, "skipped": skipped}
|
|
|
|
|
|
# --- 3b. the CONNECTED Google sheet via service account (private, no public share) ---
|
|
|
|
_gtoken = {}
|
|
|
|
|
|
async def _google_token(db):
|
|
"""Service-account access token for the Sheets API (signed JWT → OAuth token). Cached ~hour."""
|
|
if _gtoken.get("exp", 0) > time.time() + 60:
|
|
return _gtoken["tok"]
|
|
email = await vault.get_secret(db, "google_sa_email")
|
|
pem = await vault.get_secret(db, "google_sa_private_key")
|
|
if not (email and pem):
|
|
return None
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
now = int(time.time())
|
|
b64 = lambda d: base64.urlsafe_b64encode(json.dumps(d, separators=(",", ":")).encode()).rstrip(b"=")
|
|
seg = b64({"alg": "RS256", "typ": "JWT"}) + b"." + b64(
|
|
{"iss": email, "scope": "https://www.googleapis.com/auth/spreadsheets.readonly",
|
|
"aud": "https://oauth2.googleapis.com/token", "iat": now, "exp": now + 3600})
|
|
key = serialization.load_pem_private_key(pem.encode(), password=None)
|
|
sig = base64.urlsafe_b64encode(key.sign(seg, padding.PKCS1v15(), hashes.SHA256())).rstrip(b"=")
|
|
async with httpx.AsyncClient(timeout=20) as c:
|
|
r = await c.post("https://oauth2.googleapis.com/token", data={
|
|
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
|
|
"assertion": (seg + b"." + sig).decode()})
|
|
if r.status_code != 200:
|
|
return None
|
|
j = r.json()
|
|
_gtoken.update(tok=j["access_token"], exp=now + j.get("expires_in", 3600))
|
|
return _gtoken["tok"]
|
|
|
|
|
|
# WowPlatter's stored column_mappings keys → our fields
|
|
_GCOL = {"release_id": "release_id", "media_condition": "media", "sleeve_condition": "sleeve",
|
|
"price": "price", "comment": "notes", "timestamp": "timestamp", "sku": "sku"}
|
|
|
|
|
|
async def _configured_sheet(db):
|
|
"""Read the connected sheet via the service account → (items, error). Uses the stored
|
|
column map (1-indexed) when present, else header auto-detect."""
|
|
tok = await _google_token(db)
|
|
if not tok:
|
|
return None, "Google service account not connected (save its creds in Connections)"
|
|
sid = await vault.get_secret(db, "google_sheet_id")
|
|
if not sid:
|
|
return None, "no google_sheet_id saved"
|
|
name = await vault.get_secret(db, "google_sheet_name") or "Sheet1"
|
|
raw = await vault.get_secret(db, "google_column_map")
|
|
colmap = {}
|
|
if raw:
|
|
for gk, idx in json.loads(raw).items():
|
|
ours = _GCOL.get(gk)
|
|
if ours and str(idx).isdigit():
|
|
colmap[ours] = int(idx) - 1
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.get(f"https://sheets.googleapis.com/v4/spreadsheets/{sid}/values/{name}",
|
|
headers={"Authorization": "Bearer " + tok})
|
|
if r.status_code != 200:
|
|
return None, f"Sheets API HTTP {r.status_code}"
|
|
values = r.json().get("values", [])
|
|
if not values:
|
|
return [], None
|
|
return _parse_rows(values, colmap or _detect(values[0])), None
|
|
|
|
|
|
async def _count_new(db, rows):
|
|
skus = [r["sku"] for r in rows if r["sku"]]
|
|
existing = set()
|
|
if skus:
|
|
existing = {x[0] for x in (await db.execute(
|
|
text("SELECT sku FROM inventory WHERE sku = ANY(:s)"), {"s": skus}))}
|
|
return sum(1 for r in rows if not (r["sku"] and r["sku"] in existing))
|
|
|
|
|
|
@router.get("/gsheet/preview")
|
|
async def gsheet_preview(ident=Depends(require_token), db=Depends(get_db)):
|
|
rows, err = await _configured_sheet(db)
|
|
if err:
|
|
return {"ok": False, "error": err}
|
|
return {"ok": True, "total": len(rows), "new": await _count_new(db, rows), "sample": rows[:10],
|
|
"sheet": await vault.get_secret(db, "google_sheet_name") or "Sheet1"}
|
|
|
|
|
|
@router.post("/gsheet/import")
|
|
async def gsheet_import(ident=Depends(require_token), db=Depends(get_db)):
|
|
rows, err = await _configured_sheet(db)
|
|
if err:
|
|
return {"ok": False, "error": err}
|
|
staged, skipped = await _bulk_stage(db, ident["store_id"], rows)
|
|
return {"ok": True, "staged": staged, "skipped": skipped}
|
|
|
|
|
|
# --- Heal: backfill missing catalog metadata (the salvaged self-healing pass) ----
|
|
# WowPlatter healed gaps inline during a 3-hour import; on metal it's a bounded one-button sweep.
|
|
# Phase 1 = LOCAL backfill from the mirror (free, instant). Phase 2 = bounded Discogs re-fetch for
|
|
# release_ids missing from the mirror or without cover art (rate-limited → re-run for the rest).
|
|
|
|
# in-stock rows in this store whose release_id is missing from the mirror, or whose cached cover is blank
|
|
_NEEDS_API = """release_id IS NOT NULL AND (
|
|
NOT EXISTS (SELECT 1 FROM disc_cache dc WHERE dc.release_id=i.release_id)
|
|
OR EXISTS (SELECT 1 FROM disc_cache dc WHERE dc.release_id=i.release_id
|
|
AND (dc.thumb IS NULL OR dc.thumb='')))"""
|
|
|
|
|
|
@router.get("/heal/scan")
|
|
async def heal_scan(ident=Depends(require_token), db=Depends(get_db)):
|
|
"""Dry-run: count the gaps before healing anything."""
|
|
sid = {"sid": ident["store_id"]}
|
|
base = "FROM inventory i WHERE i.store_id=:sid AND i.in_stock"
|
|
|
|
async def n(extra):
|
|
return (await db.execute(text(f"SELECT count(*) {base} AND {extra}"), sid)).scalar()
|
|
|
|
async def nd(extra):
|
|
return (await db.execute(text(f"SELECT count(DISTINCT i.release_id) {base} AND {extra}"), sid)).scalar()
|
|
|
|
return {
|
|
"ok": True,
|
|
"no_release_id": await n("i.release_id IS NULL"), # can't auto-heal (needs matching)
|
|
"missing_title": await n("i.release_id IS NOT NULL AND i.title IS NULL"),
|
|
"missing_weight": await n("i.release_id IS NOT NULL AND i.weight_g IS NULL"),
|
|
"local_fixable": await n("i.release_id IS NOT NULL AND (i.title IS NULL OR i.weight_g IS NULL) "
|
|
"AND EXISTS (SELECT 1 FROM disc_cache dc WHERE dc.release_id=i.release_id "
|
|
"AND (dc.title IS NOT NULL OR dc.weight IS NOT NULL))"),
|
|
"api_needed": await nd(_NEEDS_API), # distinct releases needing Discogs
|
|
}
|
|
|
|
|
|
class HealIn(BaseModel):
|
|
limit: int = 60 # cap Discogs fetches per run (rate limit) — re-run for the remainder
|
|
|
|
|
|
@router.post("/heal/run")
|
|
async def heal_run(body: HealIn = HealIn(), ident=Depends(require_token), db=Depends(get_db)):
|
|
sid = ident["store_id"]
|
|
# Phase 1 — local backfill from the mirror (instant, unbounded)
|
|
local = (await db.execute(text("""
|
|
UPDATE inventory i SET title = COALESCE(i.title, dc.title),
|
|
weight_g = COALESCE(i.weight_g, dc.weight), updated_at = now()
|
|
FROM disc_cache dc
|
|
WHERE dc.release_id = i.release_id AND i.store_id = :sid AND i.in_stock
|
|
AND (i.title IS NULL OR i.weight_g IS NULL)
|
|
AND (dc.title IS NOT NULL OR dc.weight IS NOT NULL)"""), {"sid": sid})).rowcount
|
|
await db.commit()
|
|
|
|
# Phase 2 — bounded Discogs re-fetch for releases missing from the mirror / without cover
|
|
rids = [r[0] for r in (await db.execute(text(
|
|
f"SELECT DISTINCT i.release_id FROM inventory i "
|
|
f"WHERE i.store_id=:sid AND i.in_stock AND {_NEEDS_API} LIMIT :lim"),
|
|
{"sid": sid, "lim": body.limit}))]
|
|
enriched = 0
|
|
for rid in rids:
|
|
meta = await _fetch_release(db, rid) # always fetches → fills cover + grows mirror
|
|
if meta:
|
|
await db.execute(text(
|
|
"UPDATE inventory SET title=COALESCE(title,:t), weight_g=COALESCE(weight_g,:w), "
|
|
"updated_at=now() WHERE release_id=:r AND store_id=:sid AND in_stock"),
|
|
{"t": meta.get("title"), "w": meta.get("weight"), "r": rid, "sid": sid})
|
|
enriched += 1
|
|
await asyncio.sleep(0.2) # ponytail: gentle on the Discogs rate limit
|
|
await db.commit()
|
|
|
|
remaining = (await db.execute(text(
|
|
f"SELECT count(DISTINCT i.release_id) FROM inventory i "
|
|
f"WHERE i.store_id=:sid AND i.in_stock AND {_NEEDS_API}"), {"sid": sid})).scalar()
|
|
return {"ok": True, "local_healed": local, "api_enriched": enriched, "remaining_api": remaining}
|
|
|
|
|
|
# --- ScanGod bridge: photograph stock → staged RecordGod products -----------------
|
|
# DealGod's ScanGod (vision + bench measurement) POSTs reviewed items here; we resolve the
|
|
# barcode → release_id (local disc_release_identifier, Discogs fallback), enrich, store the
|
|
# MEASURED weight + dims + condition photos, and stage for the human review/publish queue.
|
|
# Contract: SCANGOD_BRIDGE_BRIEF.md / SCANGOD_BRIDGE_REPLY.md.
|
|
|
|
ITEM_IMG_DIR = Path(os.getenv("DISC_IMAGE_DIR", "/app/disc_images")) / "items"
|
|
|
|
|
|
async def _resolve_barcode(db, barcode):
|
|
"""barcode → release_id. Local disc_release_identifier first — NORMALISED (DB barcodes are stored
|
|
inconsistently: '5 018775 901762' vs '042285768916'), matched on a functional index over the
|
|
digits-only form, trying the EAN-13/UPC-A leading-zero variants. Discogs barcode search on a miss."""
|
|
if not barcode:
|
|
return None
|
|
digits = re.sub(r"\D", "", str(barcode))
|
|
if len(digits) < 6:
|
|
return None
|
|
cands = list({digits, digits.lstrip("0"), "0" + digits})
|
|
row = (await db.execute(text(
|
|
"SELECT release_id FROM disc_release_identifier "
|
|
"WHERE type='Barcode' AND regexp_replace(value,'[^0-9]','','g') = ANY(:c) LIMIT 1"),
|
|
{"c": cands})).first()
|
|
if row:
|
|
return row[0]
|
|
try:
|
|
c, ok = await _client(db)
|
|
async with c:
|
|
if not ok:
|
|
return None
|
|
r = await c.get("/database/search", params={"barcode": digits, "type": "release", "per_page": 1})
|
|
if r.status_code == 200:
|
|
res = r.json().get("results", [])
|
|
if res:
|
|
return res[0].get("id")
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _save_item_images(sku, images):
|
|
"""Decode base64 condition photos → DISC_IMAGE_DIR/items/<sku>/<n>.jpg; return their /img/item URLs."""
|
|
out = []
|
|
if not images:
|
|
return out
|
|
d = ITEM_IMG_DIR / sku
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
for i, img in enumerate(images):
|
|
if not isinstance(img, str):
|
|
continue
|
|
b64 = img.split(",", 1)[1] if img.startswith("data:") else img
|
|
try:
|
|
(d / f"{i}.jpg").write_bytes(base64.b64decode(b64))
|
|
out.append(f"/img/item/{sku}/{i}")
|
|
except Exception:
|
|
continue
|
|
return out
|
|
|
|
|
|
class ScanItem(BaseModel):
|
|
kind: str = "vinyl"
|
|
barcode: str | None = None
|
|
release_id: int | None = None
|
|
title: str | None = None
|
|
artist: str | None = None
|
|
condition: str | None = "VG+"
|
|
sleeve: str | None = None
|
|
price: float | None = None
|
|
notes: str | None = None
|
|
weight_g: int | None = None # MEASURED on the bench scale — overrides the enrich default
|
|
dims_mm: dict | None = None # MEASURED → inventory.attributes
|
|
est_market_value: float | None = None
|
|
images: list[str] | None = None # base64 condition photos (data: URLs or raw b64)
|
|
|
|
|
|
class ScanIn(BaseModel):
|
|
source: str = "scangod"
|
|
scan_id: int | None = None
|
|
items: list[ScanItem]
|
|
|
|
|
|
@router.post("/scan")
|
|
async def intake_scan(body: ScanIn, ident=Depends(require_token), db=Depends(get_db)):
|
|
"""Stage a batch of camera-captured items. One physical copy = one new SKU (condition varies per
|
|
copy, so no barcode dedup). Returns a per-item review_url. Non-destructive: lands as staged."""
|
|
staged, errors = [], []
|
|
for it in body.items:
|
|
try:
|
|
rid = it.release_id or await _resolve_barcode(db, it.barcode)
|
|
meta = await _enrich(db, rid) if rid else None
|
|
sku = _new_sku()
|
|
title = it.title or (meta or {}).get("title")
|
|
weight = it.weight_g if it.weight_g is not None else (meta or {}).get("weight")
|
|
attrs = {"source": body.source}
|
|
if body.scan_id is not None:
|
|
attrs["scan_id"] = body.scan_id
|
|
if it.dims_mm:
|
|
attrs["dims_mm"] = it.dims_mm
|
|
imgs = _save_item_images(sku, it.images)
|
|
await db.execute(text("""
|
|
INSERT INTO inventory (sku, store_id, kind, release_id, identifier, title, price,
|
|
condition, sleeve_cond, weight_g, notes, est_market_value, attributes, images,
|
|
staged, status)
|
|
VALUES (:sku,:sid,:kind,:rid,:bc,:title,:price,:cond,:sleeve,:wt,:notes,:emv,
|
|
CAST(:attrs AS jsonb), CAST(:imgs AS jsonb), true, 'staged')
|
|
ON CONFLICT (sku) DO NOTHING"""),
|
|
{"sku": sku, "sid": ident["store_id"], "kind": it.kind, "rid": rid,
|
|
"bc": it.barcode, "title": title, "price": it.price, "cond": it.condition,
|
|
"sleeve": it.sleeve, "wt": weight, "notes": it.notes,
|
|
"emv": it.est_market_value, "attrs": json.dumps(attrs), "imgs": json.dumps(imgs)})
|
|
staged.append({"sku": sku, "release_id": rid, "title": title,
|
|
"review_url": f"/admin?review={sku}"})
|
|
except Exception as e:
|
|
errors.append({"barcode": it.barcode, "error": str(e)})
|
|
await db.commit()
|
|
return {"staged": staged, "errors": errors}
|
|
|
|
|
|
# --- Distro purchase ingest: scrape a distributor order → cost-tracked NEW stock ------------------
|
|
# RareWaves (Shopify) order pages give barcode (in the /products/<ean>- handle) + title + qty +
|
|
# unit cost ($X.XX/ea). The PRICEGOD extension scrapes the order and POSTs it here; we resolve the
|
|
# barcode → release_id, stage one NEW copy per qty with cost_price + cost_source, for staff approval.
|
|
|
|
class DistroItem(BaseModel):
|
|
barcode: str | None = None
|
|
release_id: int | None = None
|
|
title: str | None = None
|
|
artist: str | None = None
|
|
year: int | None = None
|
|
qty: int = 1
|
|
unit_cost: float | None = None
|
|
kind: str | None = None # cd | vinyl | … (Inertia gives FORMAT; RareWaves = vinyl)
|
|
catno: str | None = None
|
|
slug: str | None = None
|
|
variant_id: str | None = None
|
|
image: str | None = None
|
|
|
|
|
|
class DistroIn(BaseModel):
|
|
source: str = "rarewaves"
|
|
order_ref: str
|
|
items: list[DistroItem]
|
|
|
|
|
|
async def _ingest_distro(db, sid, source, order_ref, items):
|
|
"""Shared core for every distro source (web-scrape or spreadsheet). `items` = list of dicts.
|
|
Idempotent per order (cost_source). Stages one NEW copy per qty; unresolved barcodes still stage."""
|
|
cost_source = f"{source} #{order_ref}"
|
|
existing = (await db.execute(text(
|
|
"SELECT count(*) FROM inventory WHERE cost_source=:cs AND store_id=:sid"),
|
|
{"cs": cost_source, "sid": sid})).scalar()
|
|
if existing:
|
|
return {"ok": True, "already_ingested": True, "existing": existing, "cost_source": cost_source}
|
|
|
|
staged, errors = [], []
|
|
for it in items:
|
|
try:
|
|
rid = it.get("release_id") or await _resolve_barcode(db, it.get("barcode"))
|
|
meta = await _enrich(db, rid) if rid else None
|
|
title = it.get("title") or (meta or {}).get("title")
|
|
weight = (meta or {}).get("weight")
|
|
kind = it.get("kind") or "vinyl"
|
|
for _ in range(max(1, int(it.get("qty") or 1))):
|
|
sku = _new_sku()
|
|
attrs = {"source": source, "order_ref": order_ref, "slug": it.get("slug"),
|
|
"variant_id": it.get("variant_id"), "year": it.get("year"),
|
|
"catno": it.get("catno"), "artist": it.get("artist"),
|
|
"resolved": rid is not None, "scrape_image": it.get("image")}
|
|
await db.execute(text("""
|
|
INSERT INTO inventory (sku, store_id, kind, release_id, identifier, title,
|
|
cost_price, cost_source, condition_type, condition, weight_g, attributes,
|
|
staged, in_stock, status)
|
|
VALUES (:sku,:sid,:kind,:rid,:bc,:title,:cost,:cs,'new','M',:wt,
|
|
CAST(:attrs AS jsonb), true, false, 'staged')
|
|
ON CONFLICT (sku) DO NOTHING"""),
|
|
{"sku": sku, "sid": sid, "kind": kind, "rid": rid, "bc": it.get("barcode"),
|
|
"title": title, "cost": it.get("unit_cost"), "cs": cost_source,
|
|
"wt": weight, "attrs": json.dumps(attrs)})
|
|
staged.append({"sku": sku, "release_id": rid, "title": title, "resolved": rid is not None})
|
|
except Exception as e:
|
|
errors.append({"barcode": it.get("barcode"), "error": str(e)})
|
|
await db.commit()
|
|
resolved = sum(1 for s in staged if s["resolved"])
|
|
return {"ok": True, "source": source, "order_ref": order_ref, "cost_source": cost_source,
|
|
"staged": len(staged), "resolved": resolved, "unresolved": len(staged) - resolved, "errors": errors}
|
|
|
|
|
|
@router.post("/distro")
|
|
async def intake_distro(body: DistroIn, ident=Depends(require_token), db=Depends(get_db)):
|
|
"""Web-scraped distributor order (RareWaves) → staged NEW stock with per-copy cost."""
|
|
return await _ingest_distro(db, ident["store_id"], body.source, body.order_ref,
|
|
[it.model_dump() for it in body.items])
|
|
|
|
|
|
# --- Spreadsheet distro ingest (Inertia/Warner SOH list — barcode/title/format/cost in columns) ---
|
|
_DISTRO_COLS = { # our field -> header keywords (first match wins)
|
|
"qty": ["order qty", "qty", "quantity"],
|
|
"barcode": ["upc", "barcode", "ean", "bar code"],
|
|
"catno": ["catalogue", "catalog", "cat#", "cat no", "catno"],
|
|
"artist": ["artist"],
|
|
"title": ["title"],
|
|
"format": ["format"],
|
|
"unit_cost": ["ppd", "wsp", "cost", "price", "dealer"],
|
|
}
|
|
|
|
|
|
def _parse_distro_xlsx(data: bytes):
|
|
"""Inertia-style stock-list/order sheet → distro items for the rows with ORDER QTY > 0.
|
|
Header-mapped (works across distro sheets); FORMAT → kind (cd/vinyl)."""
|
|
import io
|
|
import openpyxl
|
|
wb = openpyxl.load_workbook(io.BytesIO(data), read_only=True, data_only=True)
|
|
items = []
|
|
for ws in wb.worksheets:
|
|
col = None
|
|
for row in ws.iter_rows(values_only=True):
|
|
cells = [(str(c).strip() if c is not None else "") for c in row]
|
|
if col is None:
|
|
up = [c.lower() for c in cells]
|
|
if any("title" in c for c in up) and any(("upc" in c or "barcode" in c or "ean" in c) for c in up):
|
|
col = {}
|
|
for i, h in enumerate(up):
|
|
for field, keys in _DISTRO_COLS.items():
|
|
if field not in col and any(k in h for k in keys):
|
|
col[field] = i
|
|
continue
|
|
def g(k):
|
|
i = col.get(k)
|
|
return cells[i] if i is not None and i < len(cells) else ""
|
|
try:
|
|
qty = int(float(g("qty"))) if g("qty") else 0
|
|
except ValueError:
|
|
qty = 0
|
|
bc = re.sub(r"\s", "", g("barcode"))
|
|
if qty <= 0 or not bc.isdigit():
|
|
continue
|
|
try:
|
|
cost = float(re.sub(r"[^0-9.]", "", g("unit_cost"))) if g("unit_cost") else None
|
|
except ValueError:
|
|
cost = None
|
|
fmt = g("format").lower()
|
|
kind = "cd" if "cd" in fmt else ("vinyl" if ("lp" in fmt or "vinyl" in fmt) else (fmt or "vinyl"))
|
|
items.append({"barcode": bc, "title": g("title") or None, "artist": g("artist") or None,
|
|
"catno": g("catno") or None, "qty": qty, "unit_cost": cost, "kind": kind})
|
|
return items
|
|
|
|
|
|
class DistroSheetIn(BaseModel):
|
|
source: str = "inertia"
|
|
order_ref: str | None = None
|
|
filename: str | None = None
|
|
xlsx_b64: str
|
|
|
|
|
|
@router.post("/distro-sheet")
|
|
async def intake_distro_sheet(body: DistroSheetIn, ident=Depends(require_token), db=Depends(get_db)):
|
|
"""Upload a distro stock-list/order sheet (xlsx) with ORDER QTY filled → stage those rows."""
|
|
try:
|
|
raw = base64.b64decode(body.xlsx_b64.split(",", 1)[-1])
|
|
items = _parse_distro_xlsx(raw)
|
|
except Exception as e:
|
|
return {"ok": False, "error": f"could not read the sheet: {e}"}
|
|
if not items:
|
|
return {"ok": False, "error": "no rows with ORDER QTY > 0 — fill the order-qty column and re-upload"}
|
|
order_ref = (body.order_ref or body.filename or "sheet").replace("#", "").strip()[:60]
|
|
return await _ingest_distro(db, ident["store_id"], body.source, order_ref, items)
|
|
|
|
|
|
@router.post("/distro-sheet/preview")
|
|
async def intake_distro_sheet_preview(body: DistroSheetIn, ident=Depends(require_token), db=Depends(get_db)):
|
|
"""Dry-run: how many order rows + a sample, before committing."""
|
|
try:
|
|
raw = base64.b64decode(body.xlsx_b64.split(",", 1)[-1])
|
|
items = _parse_distro_xlsx(raw)
|
|
except Exception as e:
|
|
return {"ok": False, "error": f"could not read the sheet: {e}"}
|
|
total = sum(int(i.get("qty") or 1) for i in items)
|
|
return {"ok": True, "rows": len(items), "copies": total, "sample": items[:10]}
|
|
|
|
|
|
def _selfcheck():
|
|
# SKU from a Google-Form timestamp (both ISO and en-AU slash formats) → YYYYMMDDHHMMSS
|
|
assert _sku_from_ts("2025-01-30T14:01:02.821Z") == "20250130140102"
|
|
assert _sku_from_ts("2025-01-30, 14:01:02") == "20250130140102" # Sheet1's locale format
|
|
assert _sku_from_ts("30/01/2025 14:01:02") == "20250130140102"
|
|
assert _sku_from_ts("") is None
|
|
# CSV export URL derivation (id + gid)
|
|
assert _csv_url("https://docs.google.com/spreadsheets/d/ABC_123/edit#gid=42") == \
|
|
"https://docs.google.com/spreadsheets/d/ABC_123/export?format=csv&gid=42"
|
|
assert _csv_url("https://example.com/nope") is None
|
|
# header detection + row parse (the Google-Form layout)
|
|
csv_text = ("Timestamp,Discogs Release ID,Media Condition,Sleeve Condition,Price (AUD),Comment\n"
|
|
"2025-01-30T14:01:02.821Z,249504,VG+,VG,25.50,nice copy\n"
|
|
",,,,,\n" # blank → skipped
|
|
"2025-02-01T09:00:00.000Z,bad,NM,NM,10,\n") # non-numeric release_id → skipped
|
|
rows, col = _parse_sheet(csv_text)
|
|
assert col["release_id"] == 1 and col["price"] == 4, col
|
|
assert len(rows) == 1, rows
|
|
assert rows[0] == {"release_id": 249504, "sku": "20250130140102", "price": 25.5,
|
|
"media": "VG+", "sleeve": "VG", "notes": "nice copy"}, rows[0]
|
|
print("intake selfcheck OK")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_selfcheck()
|