25127b1b72
The intake sheet is read straight from the stored Google service account (discotag@…
inventory-sheets, migrated out of WowPlatter's wp_rmp_disc_google_service into the vault) — a
signed JWT (cryptography, no new dep) → Sheets API → private sheet, no public-share needed. Uses
the saved column_mappings (timestamp/release_id/media/sleeve/price/comment, 1-indexed). SKU from
the sheet's locale timestamp '2025-01-30, 14:01:02' → 20250130140102 (matches migrated stock).
Bulk import (both connected + paste-URL paths) now SKIPS SKUs already in inventory, so importing
the 34,752-row form log only stages the genuinely-new rows (live: 2,006 new / 32,746 skipped).
Preview shows new-of-total. Endpoints: GET/POST /admin/intake/gsheet/{preview,import}.
Also: .dockerignore now excludes disc_images/ (volume-mounted, was 916MB) — build context
918MB→1MB, deploy 13min(hung)→24s. The COPY . . was baking the image store into every image.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
542 lines
23 KiB
Python
542 lines
23 KiB
Python
import asyncio
|
|
import base64
|
|
import csv
|
|
import io
|
|
import json
|
|
import re
|
|
import secrets
|
|
import time
|
|
|
|
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:
|
|
# 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}
|
|
|
|
|
|
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()
|