← back to Govarbitrage

src/importers/ingest.ts

160 lines

import { prisma } from "@/lib/db";
import { runResearch } from "@/pipeline/research";
import type { AuctionSource, Condition } from "@prisma/client";

// Normalized import shape every importer (CSV, extension, URL scrape) produces.
export interface RawListing {
  source: string;
  sourceAuctionId: string;
  sourceUrl?: string;
  title: string;
  description?: string;
  category?: string;
  manufacturer?: string;
  model?: string;
  condition?: string;
  quantity?: number;
  weightLbs?: number;
  dimensions?: string;
  locationCity?: string;
  locationState?: string;
  locationZip?: string;
  currentBid?: number;
  bidCount?: number;
  closingAt?: string | Date | null;
  imageUrls?: string[];
  auctionTerms?: string;
}

const SOURCE_MAP: Record<string, AuctionSource> = {
  GOVDEALS: "GOVDEALS",
  PUBLIC_SURPLUS: "PUBLIC_SURPLUS",
  PUBLICSURPLUS: "PUBLIC_SURPLUS",
  GSA_AUCTIONS: "GSA_AUCTIONS",
  GSA: "GSA_AUCTIONS",
  COUNTY: "COUNTY",
  STATE_SURPLUS: "STATE_SURPLUS",
  UNIVERSITY_SURPLUS: "UNIVERSITY_SURPLUS",
  MUNICIBID: "MUNICIBID",
  BID4ASSETS: "BID4ASSETS",
  GOINDUSTRY: "GOINDUSTRY",
  NETWORK_INTL: "NETWORK_INTL",
  GRAYS_AU: "GRAYS_AU",
  GOVPLANET: "GOVPLANET",
  CSV: "CSV",
  EXTENSION: "EXTENSION",
};

const CONDITION_MAP: Record<string, Condition> = {
  NEW: "NEW",
  LIKE_NEW: "LIKE_NEW",
  LIKENEW: "LIKE_NEW",
  USED_GOOD: "USED_GOOD",
  GOOD: "USED_GOOD",
  USED: "USED_GOOD",
  USED_FAIR: "USED_FAIR",
  FAIR: "USED_FAIR",
  FOR_PARTS: "FOR_PARTS",
  PARTS: "FOR_PARTS",
  SALVAGE: "FOR_PARTS",
  UNKNOWN: "UNKNOWN",
};

export function normalizeSource(s: string | undefined): AuctionSource {
  return SOURCE_MAP[(s || "").toUpperCase().replace(/[\s-]/g, "_")] ?? "OTHER";
}

export function normalizeCondition(c: string | undefined): Condition {
  return CONDITION_MAP[(c || "").toUpperCase().replace(/[\s-]/g, "_")] ?? "UNKNOWN";
}

export interface IngestResult {
  listingId: string;
  created: boolean;
  research?: Awaited<ReturnType<typeof runResearch>>;
}

/**
 * Ingest one normalized listing: upsert by (source, sourceAuctionId), then run
 * the research pipeline (local AI identification + engines). Idempotent.
 */
export async function ingest(
  raw: RawListing,
  opts: { research?: boolean; useAI?: boolean } = {},
): Promise<IngestResult> {
  if (!raw.title || !raw.sourceAuctionId) {
    throw new Error("Import requires at least title and sourceAuctionId");
  }
  const source = normalizeSource(raw.source);
  const closingAt = raw.closingAt ? new Date(raw.closingAt) : null;

  const existing = await prisma.listing.findUnique({
    where: { source_sourceAuctionId: { source, sourceAuctionId: raw.sourceAuctionId } },
  });

  const data = {
    source,
    sourceAuctionId: raw.sourceAuctionId,
    sourceUrl: raw.sourceUrl ?? null,
    title: raw.title,
    description: raw.description ?? null,
    category: raw.category ?? null,
    manufacturer: raw.manufacturer ?? null,
    model: raw.model ?? null,
    condition: normalizeCondition(raw.condition),
    quantity: raw.quantity && raw.quantity > 0 ? Math.floor(raw.quantity) : 1,
    weightLbs: raw.weightLbs ?? null,
    dimensions: raw.dimensions ?? null,
    locationCity: raw.locationCity ?? null,
    locationState: raw.locationState ?? null,
    locationZip: raw.locationZip ?? null,
    currentBid: raw.currentBid ?? 0,
    bidCount: raw.bidCount ?? 0,
    closingAt: closingAt && !isNaN(closingAt.getTime()) ? closingAt : null,
    imageUrls: raw.imageUrls ?? [],
    auctionTerms: raw.auctionTerms ?? null,
    // Seeing the item in an import = it's live on the source right now.
    // Refresh lastSeenAt and reactivate (an item that reappears is live again).
    lastSeenAt: new Date(),
    listingStatus: "ACTIVE" as const,
    endedAt: null,
  };

  const listing = existing
    ? await prisma.listing.update({ where: { id: existing.id }, data })
    : await prisma.listing.create({ data: { ...data, researchStatus: "PENDING" } });

  await prisma.listingEvent.create({
    data: {
      listingId: listing.id,
      type: existing ? "MANUAL_EDIT" : "IMPORTED",
      message: existing ? `Re-imported from ${source}` : `Imported from ${source}`,
    },
  });

  let research;
  if (opts.research !== false) {
    // Live single imports use the local AI identifier; bulk imports pass
    // useAI:false for speed (the worker AI-enriches later). Failures degrade
    // to the heuristic path inside runResearch either way.
    const useAI = opts.useAI !== false;
    research = await runResearch(listing.id, { useAI, writeIdentity: useAI });
  }

  return { listingId: listing.id, created: !existing, research };
}

/** Bulk ingest with per-row error capture. */
export async function ingestMany(rows: RawListing[], opts: { research?: boolean; useAI?: boolean } = {}) {
  const results: { ok: boolean; listingId?: string; created?: boolean; error?: string; title: string }[] = [];
  for (const row of rows) {
    try {
      const r = await ingest(row, opts);
      results.push({ ok: true, listingId: r.listingId, created: r.created, title: row.title });
    } catch (e) {
      results.push({ ok: false, error: (e as Error).message, title: row.title || "(untitled)" });
    }
  }
  return results;
}