88172924eb
Replaces the bare release-id box with a tabbed Intake that all feeds one _stage() core: - Find & stage: search the local Discogs mirror (instant) with a Discogs-API fallback for new titles; pick a release → condition/price → stage. Cards show cover/artist/year/label/format. - Discogs collection: lists the OAuth token's own folders (mrp3, 152 with stock) → page through releases, reads media/sleeve/price from the collection's custom fields, stage-all-on-page. - Google Sheet: paste the share URL → public CSV export (no Google creds) → auto-detect columns (release id/media/sleeve/price/comment/timestamp) → preview → bulk import. SKU from the Google-Form timestamp (YYYYMMDDHHMMSS), matching migrated stock. Staging enriches title/artist/cover on demand: a disc_cache miss fetches the release from the Discogs API and grows the local mirror (disc_cache + disc_release + artist + label), so internal search + storefront get richer with every intake. Verified end-to-end on the VPS (incl. enrich of a not-in-cache release) + sheet-parser self-check. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
422 lines
18 KiB
Python
422 lines
18 KiB
Python
import asyncio
|
|
import base64
|
|
import csv
|
|
import io
|
|
import json
|
|
import re
|
|
import secrets
|
|
|
|
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)
|
|
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 — the fast daily path (re-stocks resolve instantly)."""
|
|
where, params = ["TRUE"], {}
|
|
if q:
|
|
where.append("(dr.title ILIKE :q OR dr.artists_sort ILIKE :q)"); params["q"] = f"%{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 dr.title 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:
|
|
m = re.match(r"(\d{4})-(\d{2})-(\d{2})[T ](\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_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
|
|
out = []
|
|
for r in rows[1:]:
|
|
def g(k):
|
|
i = col.get(k)
|
|
return r[i].strip() if i is not None and i < len(r) and r[i] else None
|
|
rid = g("release_id")
|
|
if not rid or not 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, 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]}
|
|
|
|
|
|
@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 = 0
|
|
for row in rows:
|
|
await _stage(db, ident["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 {"ok": True, "staged": staged}
|
|
|
|
|
|
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("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()
|