groceries: nightly AH/Jumbo catalogue import, unit-price normalisation, product groups and compare UI
Deploy production / deploy (push) Successful in 2m48s

This commit is contained in:
stan committed 2026-10-24 14:01:26 +00:00
1 parent 8dde5aa302
commit 427f4ed095
20 files changed
+34891 -699

No files matched your search

+397
View File
@@ -0,0 +1,397 @@
// DB side of the grocery importer (used by scripts/groceries-import.ts). Uses a plain postgres.js
// client so it runs outside SvelteKit. All writes are idempotent; re-running is safe.
import type { Sql } from 'postgres';
import type { PoliteFetcher } from './http';
import { clusterItems, type MatchItem } from './match';
import {
brandClass,
brandLabel,
displayName,
matchKey,
nameTokens,
normaliseUnit
} from './normalize';
import { AhClient, type RawItem, type Source } from './sources';
type Log = (m: string) => void;
export async function storeId(sql: Sql, slug: string): Promise<number> {
const [s] = await sql<{ id: number }[]>`select id from groceries.stores where slug = ${slug}`;
if (!s) throw new Error(`unknown store ${slug}`);
return s.id;
}
function keyFor(name: string, brand: string | null, unit: string | null) {
const cls = brandClass(brand);
const tokens = nameTokens(name, brand);
return { cls, tokens, key: matchKey(cls, unit, tokens) };
}
/** Fetch a store's catalogue and upsert it; returns counts. */
export async function importStore(sql: Sql, source: Source, f: PoliteFetcher, log: Log) {
const sid = await storeId(sql, source.slug);
const [run] = await sql<{ id: number; started_at: Date }[]>`
insert into groceries.import_runs (store_id) values (${sid}) returning id, started_at`;
try {
const raw = await source.fetchAll(f, log);
const [prev] = await sql<{ n: number }[]>`
select count(*)::int as n from groceries.items where store_id = ${sid} and available`;
const res = await upsertItems(sql, sid, raw, run.started_at);
// Only mark unseen items unavailable when the crawl looks complete.
let delisted = 0;
if (raw.length >= prev.n * 0.8) {
const r = await sql`
update groceries.items set available = false
where store_id = ${sid} and available and last_seen_at < ${run.started_at}`;
delisted = r.count;
}
await sql`
update groceries.import_runs set finished_at = now(), status = 'ok', requests = ${f.requests},
fetched = ${raw.length}, inserted = ${res.inserted}, price_changes = ${res.priceChanges}
where id = ${run.id}`;
return { fetched: raw.length, ...res, delisted, requests: f.requests, cacheHits: f.cacheHits };
} catch (e) {
await sql`
update groceries.import_runs set finished_at = now(), status = 'error', requests = ${f.requests},
error = ${String(e instanceof Error ? e.message : e).slice(0, 2000)}
where id = ${run.id}`;
throw e;
}
}
async function upsertItems(sql: Sql, sid: number, raw: RawItem[], runStart: Date) {
let inserted = 0;
for (let i = 0; i < raw.length; i += 500) {
const rows = raw.slice(i, i + 500).map((r) => {
const n = normaliseUnit({
sizeText: r.sizeText,
title: r.name,
priceCents: r.priceCents,
promoPriceCents: r.promoPriceCents,
storeUnitPrice: r.storeUnitPrice,
storeUnitBasisCents: r.storeUnitBasisCents
});
return {
store_id: sid,
store_product_id: r.storeProductId,
name: r.name,
brand: r.brand,
size_text: r.sizeText,
unit: n.unit,
unit_amount: n.unitAmount,
unit_source: n.unitSource,
price_cents: r.priceCents,
promo_price_cents: r.promoPriceCents,
promo_label: r.promoLabel,
unit_price_cents: n.unitPriceCents,
promo_unit_price_cents: n.promoUnitPriceCents,
category: r.category,
category_path: r.categoryPath,
gtin: r.gtin,
image_url: r.imageUrl,
url: r.url,
available: r.available,
match_key: keyFor(r.name, r.brand, n.unit).key
};
});
const res = await sql<{ inserted: boolean }[]>`
insert into groceries.items ${sql(rows)}
on conflict (store_id, store_product_id) do update set
name = excluded.name, brand = excluded.brand, size_text = excluded.size_text,
unit = excluded.unit, unit_amount = excluded.unit_amount, unit_source = excluded.unit_source,
price_cents = excluded.price_cents, promo_price_cents = excluded.promo_price_cents,
promo_label = excluded.promo_label, unit_price_cents = excluded.unit_price_cents,
promo_unit_price_cents = excluded.promo_unit_price_cents, category = excluded.category,
category_path = excluded.category_path, gtin = coalesce(excluded.gtin, items.gtin),
image_url = excluded.image_url, url = excluded.url, available = excluded.available,
match_key = excluded.match_key, last_seen_at = now()
returning (xmax = 0) as inserted`;
inserted += res.filter((r) => r.inserted).length;
}
// Price history: a row when the price or promo differs from the item's last recorded row.
const hist = await sql<{ changed: boolean }[]>`
insert into groceries.item_prices (item_id, observed_on, price_cents, promo_price_cents, promo_label)
select i.id, (now() at time zone 'Europe/Amsterdam')::date, i.price_cents, i.promo_price_cents, i.promo_label
from groceries.items i
left join lateral (
select p.price_cents, p.promo_price_cents, p.promo_label from groceries.item_prices p
where p.item_id = i.id order by p.observed_on desc limit 1
) last on true
where i.store_id = ${sid} and i.last_seen_at >= ${runStart}
and (last is null
or last.price_cents is distinct from i.price_cents
or last.promo_price_cents is distinct from i.promo_price_cents
or last.promo_label is distinct from i.promo_label)
on conflict (item_id, observed_on) do update set
price_cents = excluded.price_cents, promo_price_cents = excluded.promo_price_cents,
promo_label = excluded.promo_label
returning exists (
select 1 from groceries.item_prices o
where o.item_id = item_prices.item_id and o.observed_on < item_prices.observed_on
) as changed`;
return { inserted, priceChanges: hist.filter((h) => h.changed).length };
}
/** Fill in AH EANs from the product detail endpoint, `limit` products per run (A-brands first). */
export async function enrichAhGtins(sql: Sql, f: PoliteFetcher, limit: number, log: Log) {
if (limit <= 0) return { checked: 0, found: 0 };
const sid = await storeId(sql, 'ah');
const todo = await sql<{ id: number; store_product_id: string }[]>`
select id, store_product_id from groceries.items
where store_id = ${sid} and gtin is null and available
and (gtin_checked_at is null or gtin_checked_at < now() - interval '90 days')
order by (lower(coalesce(brand, '')) like 'ah%') asc, gtin_checked_at asc nulls first, id
limit ${limit}`;
const ah = new AhClient(f);
let found = 0;
for (const [n, t] of todo.entries()) {
let gtin: string | null;
try {
gtin = await ah.gtin(t.store_product_id);
} catch (e) {
log(` ah gtin ${t.store_product_id}: ${e instanceof Error ? e.message : e}`);
if (n > 20 && found === 0) throw e; // something is wrong; stop hammering
continue;
}
if (gtin) found++;
await sql`update groceries.items set gtin = ${gtin}, gtin_checked_at = now() where id = ${t.id}`;
if ((n + 1) % 250 === 0) log(` ah gtin: ${n + 1}/${todo.length} checked, ${found} found`);
}
return { checked: todo.length, found };
}
// ---- regroup --------------------------------------------------------------------------------
type ItemRow = {
id: number;
store_id: number;
store_slug: string;
name: string;
brand: string | null;
unit: string | null;
gtin: string | null;
category: string | null;
group_id: number | null;
};
type GroupRow = { id: number; key: string | null; merged_into_id: number | null; manual: boolean };
/**
* Recompute groups for all unlocked items: cluster, map each cluster to a stable group (by key,
* following manual merges, else the group most of its items already had), create/update/delete
* groups, then refresh the denormalised stats. Locked items (moved by hand) are left alone.
*/
export async function regroup(sql: Sql, log: Log) {
return sql.begin(async (tx) => {
const items = await tx<ItemRow[]>`
select i.id, i.store_id, s.slug as store_slug, i.name, i.brand, i.unit, i.gtin, i.category, i.group_id
from groceries.items i join groceries.stores s on s.id = i.store_id
where not i.match_locked
order by i.id`;
const groups = await tx<
GroupRow[]
>`select id, key, merged_into_id, manual from groceries.groups`;
const byId = new Map(groups.map((g) => [g.id, g]));
const byKey = new Map(groups.filter((g) => g.key).map((g) => [g.key!, g]));
const resolve = (g: GroupRow | undefined) => {
for (let hops = 0; g?.merged_into_id && hops < 20; hops++) g = byId.get(g.merged_into_id);
return g;
};
const meta = new Map<number, ItemRow & { cls: string; tokens: string[]; key: string }>();
const matchItems: MatchItem[] = items.map((it) => {
const k = keyFor(it.name, it.brand, it.unit);
meta.set(it.id, { ...it, ...k });
return {
id: it.id,
storeId: it.store_id,
gtin: it.gtin,
key: k.key,
cls: k.cls,
unit: it.unit,
tokens: k.tokens
};
});
const clusters = clusterItems(matchItems);
const claimed = new Set<number>();
const keysInUse = new Set(clusters.map((c) => c.key));
type Plan = { cluster: (typeof clusters)[number]; groupId: number | null; info: GroupInfo };
const plans: Plan[] = [];
// Pass 1: exact key (or merged redirect).
for (const c of clusters) {
const g = resolve(byKey.get(c.key));
const info = groupInfo(
c.itemIds.map((id) => meta.get(id)!),
c.key
);
if (g) claimed.add(g.id);
plans.push({ cluster: c, groupId: g?.id ?? null, info });
}
// Pass 2: the group most items already belong to (keeps ids stable when names change).
for (const p of plans) {
if (p.groupId) continue;
const counts = new Map<number, number>();
for (const id of p.cluster.itemIds) {
const gid = meta.get(id)!.group_id;
if (gid) counts.set(gid, (counts.get(gid) ?? 0) + 1);
}
const best = [...counts].sort((a, b) => b[1] - a[1] || a[0] - b[0])[0];
const g = best ? byId.get(best[0]) : undefined;
if (
g &&
!g.manual &&
!g.merged_into_id &&
!claimed.has(g.id) &&
!(g.key && keysInUse.has(g.key))
) {
claimed.add(g.id);
p.groupId = g.id;
}
}
// Create missing groups.
const toCreate = plans.filter((p) => !p.groupId);
for (let i = 0; i < toCreate.length; i += 1000) {
const chunk = toCreate.slice(i, i + 1000);
const ids = await tx<{ id: number }[]>`
insert into groceries.groups ${tx(
chunk.map((p) => ({
key: p.cluster.key,
name: p.info.name,
brand: p.info.brand,
category: p.info.category,
unit: p.info.unit,
family_key: p.info.familyKey
}))
)} returning id`;
chunk.forEach((p, j) => (p.groupId = ids[j].id));
}
// Update automatic groups' descriptive fields (manual groups keep Stan's name).
const toUpdate = plans.filter((p) => !toCreate.includes(p) && !byId.get(p.groupId!)?.manual);
const seen = new Set<number>();
for (let i = 0; i < toUpdate.length; i += 1000) {
const rows = toUpdate
.slice(i, i + 1000)
.filter((p) => !seen.has(p.groupId!) && seen.add(p.groupId!))
.map((p) => {
const g = byId.get(p.groupId!)!;
// Only take the cluster key when no other group has it.
const key = g.key === p.cluster.key || !byKey.has(p.cluster.key) ? p.cluster.key : g.key;
return [
p.groupId!,
key,
p.info.name,
p.info.brand,
p.info.category,
p.info.unit,
p.info.familyKey
];
});
if (!rows.length) continue;
await tx`
update groceries.groups g set key = v.key, name = v.name, brand = v.brand,
category = v.category, unit = v.unit, family_key = v.family_key
from (values ${tx(rows as never)}) as v(id, key, name, brand, category, unit, family_key)
where g.id = v.id::int and (g.key is distinct from v.key or g.name is distinct from v.name
or g.brand is distinct from v.brand or g.category is distinct from v.category
or g.unit is distinct from v.unit or g.family_key is distinct from v.family_key)`;
}
// Assign items.
const assignments: [number, number, string, string, number | null][] = [];
for (const p of plans)
for (const id of p.cluster.itemIds) {
const m = p.cluster.method.get(id)!;
assignments.push([id, p.groupId!, meta.get(id)!.key, m.method, m.confidence]);
}
let moved = 0;
for (let i = 0; i < assignments.length; i += 2000) {
const r = await tx`
update groceries.items i set group_id = v.group_id::int, match_key = v.match_key,
match_method = v.method, match_confidence = v.confidence::real
from (values ${tx(assignments.slice(i, i + 2000) as never)}) as v(id, group_id, match_key, method, confidence)
where i.id = v.id::int and (i.group_id is distinct from v.group_id::int
or i.match_key is distinct from v.match_key or i.match_method is distinct from v.method
or i.match_confidence is distinct from v.confidence::real)`;
moved += r.count;
}
// Drop empty automatic groups (favourites, manual groups and merge tombstones stay).
const del = await tx`
delete from groceries.groups g
where not g.favorite and not g.manual and g.merged_into_id is null and g.notes is null
and not exists (select 1 from groceries.items i where i.group_id = g.id)
and not exists (select 1 from groceries.groups m where m.merged_into_id = g.id)`;
await refreshGroupStats(tx as unknown as Sql);
log(
`regroup: ${items.length} items in ${clusters.length} clusters; ${toCreate.length} groups created, ${moved} items (re)assigned, ${del.count} empty groups removed`
);
return { clusters: clusters.length, created: toCreate.length, moved, removed: del.count };
});
}
type GroupInfo = {
name: string;
brand: string | null;
category: string | null;
unit: string | null;
familyKey: string;
};
function groupInfo(
members: (ItemRow & { cls: string; tokens: string[]; key: string })[],
key: string
): GroupInfo {
// Representative: an item with the canonical key, preferring AH (cleanest titles), then Jumbo.
const order = (s: string) => (s === 'ah' ? 0 : s === 'jumbo' ? 1 : 2);
const rep =
[...members]
.filter((m) => m.key === key)
.sort(
(a, b) =>
order(a.store_slug) - order(b.store_slug) || a.name.length - b.name.length || a.id - b.id
)[0] ?? members[0];
const cats = new Map<string, number>();
for (const m of members) if (m.category) cats.set(m.category, (cats.get(m.category) ?? 0) + 1);
const category = [...cats].sort((a, b) => b[1] - a[1] || (a[0] < b[0] ? -1 : 1))[0]?.[0] ?? null;
return {
name: displayName(rep.name, rep.brand).slice(0, 200),
brand: brandLabel(rep.cls, rep.brand),
category,
unit: rep.unit,
familyKey: `${rep.unit ?? '?'}|${[...rep.tokens].sort().join(' ')}`
};
}
// Accent folding for search_text (no unaccent extension needed); the query side uses fold().
const ACCENTS_FROM = 'áàäâãéèëêíìïîóòöôõúùüûçñ';
const ACCENTS_TO = 'aaaaaeeeeiiiiooooouuuucn';
/** Recompute denormalised group stats (all groups, or just `ids`). */
export async function refreshGroupStats(sql: Sql, ids?: number[]) {
const only = ids ? sql`where g2.id = any(${ids}::int[])` : sql``;
await sql`
update groceries.groups g set
item_count = s.item_count, store_count = s.store_count, store_ids = s.store_ids,
min_unit_price_cents = s.min_unit, min_price_cents = s.min_price, promo_count = s.promo_count,
image_url = s.image_url, search_text = s.search_text
from (
select g2.id,
count(i.id)::int as item_count,
count(distinct i.store_id)::int as store_count,
coalesce(array_agg(distinct i.store_id) filter (where i.id is not null), '{}')::int[] as store_ids,
min(coalesce(i.promo_unit_price_cents, i.unit_price_cents)) as min_unit,
min(coalesce(i.promo_price_cents, i.price_cents)) as min_price,
(count(*) filter (where i.promo_price_cents is not null or i.promo_label is not null))::int as promo_count,
(array_agg(i.image_url order by i.store_id, i.id) filter (where i.image_url is not null))[1] as image_url,
translate(lower(concat_ws(' ', g2.name, g2.brand, g2.category,
string_agg(distinct concat_ws(' ', i.brand, i.name), ' '))), ${ACCENTS_FROM}, ${ACCENTS_TO}) as search_text
from groceries.groups g2
left join groceries.items i on i.group_id = g2.id and i.available
${only}
group by g2.id
) s
where g.id = s.id and (g.item_count, g.store_count, g.store_ids, g.min_unit_price_cents, g.min_price_cents,
g.promo_count, g.image_url, g.search_text) is distinct from
(s.item_count, s.store_count, s.store_ids, s.min_unit, s.min_price, s.promo_count, s.image_url, s.search_text)`;
}