matt/inventory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update import scripts through products

Browse Source
This commit is contained in:
Matt
2025-02-14 21:46:50 -05:00
parent 9623681a15
commit 9b8577f258
3 changed files with 431 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
inventory-server/scripts/import/orders.js
View File

@@ -268,11 +268,18 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
`, [batchIds]); `, [batchIds]);
if (discounts.length > 0) { if (discounts.length > 0) {
const placeholders = discounts.map((_, idx) => const uniqueDiscounts = new Map();
`($${idx * 3 + 1}, $${idx * 3 + 2}, $${idx * 3 + 3})` discounts.forEach(d => {
).join(","); const key = `${d.order_id}-${d.pid}`;
const values = discounts.flatMap(d => [d.order_id, d.pid, d.discount]); uniqueDiscounts.set(key, d);
});
const values = Array.from(uniqueDiscounts.values()).flatMap(d => [d.order_id, d.pid, d.discount || 0]);
if (values.length > 0) {
const placeholders = Array.from({length: uniqueDiscounts.size}, (_, idx) => {
const base = idx * 3;
return `($${base + 1}, $${base + 2}, $${base + 3})`;
}).join(",");
await localConnection.query(` await localConnection.query(`
INSERT INTO temp_order_discounts (order_id, pid, discount) INSERT INTO temp_order_discounts (order_id, pid, discount)
VALUES ${placeholders} VALUES ${placeholders}
@@ -281,6 +288,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
`, values); `, values);
} }
} }
}
// Get tax information in batches - Keep MySQL compatible for production // Get tax information in batches - Keep MySQL compatible for production
for (let i = 0; i < orderIds.length; i += 5000) { for (let i = 0; i < orderIds.length; i += 5000) {
@@ -404,7 +412,6 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
// Filter orders and track missing products // Filter orders and track missing products
const validOrders = []; const validOrders = [];
const values = [];
const processedOrderItems = new Set(); const processedOrderItems = new Set();
const processedOrders = new Set(); const processedOrders = new Set();
@@ -425,7 +432,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
return `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}, $${base + 5}, $${base + 6}, $${base + 7}, $${base + 8}, $${base + 9}, $${base + 10}, $${base + 11}, $${base + 12}, $${base + 13}, $${base + 14}, $${base + 15})`; return `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}, $${base + 5}, $${base + 6}, $${base + 7}, $${base + 8}, $${base + 9}, $${base + 10}, $${base + 11}, $${base + 12}, $${base + 13}, $${base + 14}, $${base + 15})`;
}).join(','); }).join(',');
const values = validOrders.flatMap(o => [ const batchValues = validOrders.flatMap(o => [
o.order_number, o.order_number,
o.pid, o.pid,
o.SKU, o.SKU,
@@ -471,7 +478,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
COUNT(*) FILTER (WHERE xmax = 0) as inserted, COUNT(*) FILTER (WHERE xmax = 0) as inserted,
COUNT(*) FILTER (WHERE xmax <> 0) as updated COUNT(*) FILTER (WHERE xmax <> 0) as updated
FROM inserted_orders FROM inserted_orders
`, values); `, batchValues);
const { inserted, updated } = result.rows[0]; const { inserted, updated } = result.rows[0];
recordsAdded += inserted; recordsAdded += inserted;

333
inventory-server/scripts/import/products.js
View File

@@ -35,17 +35,28 @@ async function withRetry(operation, errorMessage) {
throw lastError; throw lastError;
} }
async function setupTemporaryTables(connection) { // Add helper function at the top of the file
await connection.query(` function validateDate(mysqlDate) {
DROP TABLE IF EXISTS temp_products; if (!mysqlDate || mysqlDate === '0000-00-00' || mysqlDate === '0000-00-00 00:00:00') {
return null;
}
// Check if the date is valid
const date = new Date(mysqlDate);
return isNaN(date.getTime()) ? null : mysqlDate;
}
async function setupTemporaryTables(connection) {
// Drop the table if it exists
await connection.query('DROP TABLE IF EXISTS temp_products');
// Create the temporary table
await connection.query(`
CREATE TEMP TABLE temp_products ( CREATE TEMP TABLE temp_products (
pid BIGINT NOT NULL, pid BIGINT NOT NULL,
title VARCHAR(255), title VARCHAR(255),
description TEXT, description TEXT,
SKU VARCHAR(50), SKU VARCHAR(50),
stock_quantity INTEGER DEFAULT 0, stock_quantity INTEGER DEFAULT 0,
pending_qty INTEGER DEFAULT 0,
preorder_count INTEGER DEFAULT 0, preorder_count INTEGER DEFAULT 0,
notions_inv_count INTEGER DEFAULT 0, notions_inv_count INTEGER DEFAULT 0,
price DECIMAL(10,3) NOT NULL DEFAULT 0, price DECIMAL(10,3) NOT NULL DEFAULT 0,
@@ -58,7 +69,7 @@ async function setupTemporaryTables(connection) {
line VARCHAR(100), line VARCHAR(100),
subline VARCHAR(100), subline VARCHAR(100),
artist VARCHAR(100), artist VARCHAR(100),
category_ids TEXT, categories TEXT,
created_at TIMESTAMP, created_at TIMESTAMP,
first_received TIMESTAMP, first_received TIMESTAMP,
landing_cost_price DECIMAL(10,3), landing_cost_price DECIMAL(10,3),
@@ -66,9 +77,11 @@ async function setupTemporaryTables(connection) {
harmonized_tariff_code VARCHAR(50), harmonized_tariff_code VARCHAR(50),
updated_at TIMESTAMP, updated_at TIMESTAMP,
visible BOOLEAN, visible BOOLEAN,
managing_stock BOOLEAN DEFAULT true,
replenishable BOOLEAN, replenishable BOOLEAN,
permalink VARCHAR(255), permalink VARCHAR(255),
moq DECIMAL(10,3), moq INTEGER DEFAULT 1,
uom INTEGER DEFAULT 1,
rating DECIMAL(10,2), rating DECIMAL(10,2),
reviews INTEGER, reviews INTEGER,
weight DECIMAL(10,3), weight DECIMAL(10,3),
@@ -81,12 +94,17 @@ async function setupTemporaryTables(connection) {
baskets INTEGER, baskets INTEGER,
notifies INTEGER, notifies INTEGER,
date_last_sold TIMESTAMP, date_last_sold TIMESTAMP,
image VARCHAR(255),
image_175 VARCHAR(255),
image_full VARCHAR(255),
options TEXT,
tags TEXT,
needs_update BOOLEAN DEFAULT TRUE, needs_update BOOLEAN DEFAULT TRUE,
PRIMARY KEY (pid) PRIMARY KEY (pid)
); )`);
CREATE INDEX idx_temp_products_needs_update ON temp_products (needs_update); // Create the index
`); await connection.query('CREATE INDEX idx_temp_products_needs_update ON temp_products (needs_update)');
} }
async function cleanupTemporaryTables(connection) { async function cleanupTemporaryTables(connection) {
@@ -205,17 +223,18 @@ async function importMissingProducts(prodConnection, localConnection, missingPid
const batch = prodData.slice(i, i + BATCH_SIZE); const batch = prodData.slice(i, i + BATCH_SIZE);
const placeholders = batch.map((_, idx) => { const placeholders = batch.map((_, idx) => {
const base = idx * 41; // 41 columns const base = idx * 47; // 47 columns
return `(${Array.from({ length: 41 }, (_, i) => `$${base + i + 1}`).join(', ')})`; return `(${Array.from({ length: 47 }, (_, i) => `$${base + i + 1}`).join(', ')})`;
}).join(','); }).join(',');
const values = batch.flatMap(row => [ const values = batch.flatMap(row => {
const imageUrls = getImageUrls(row.pid);
return [
row.pid, row.pid,
row.title, row.title,
row.description, row.description,
row.SKU, row.itemnumber || '',
row.stock_quantity, row.stock_quantity > 5000 ? 0 : Math.max(0, row.stock_quantity),
row.pending_qty,
row.preorder_count, row.preorder_count,
row.notions_inv_count, row.notions_inv_count,
row.price, row.price,
@@ -229,16 +248,18 @@ async function importMissingProducts(prodConnection, localConnection, missingPid
row.subline, row.subline,
row.artist, row.artist,
row.category_ids, row.category_ids,
row.date_created, validateDate(row.date_created),
row.first_received, validateDate(row.first_received),
row.landing_cost_price, row.landing_cost_price,
row.barcode, row.barcode,
row.harmonized_tariff_code, row.harmonized_tariff_code,
row.updated_at, validateDate(row.updated_at),
row.visible, row.visible,
true,
row.replenishable, row.replenishable,
row.permalink, row.permalink,
row.moq, Math.max(1, Math.round(row.moq || 1)),
1,
row.rating, row.rating,
row.reviews, row.reviews,
row.weight, row.weight,
@@ -250,19 +271,25 @@ async function importMissingProducts(prodConnection, localConnection, missingPid
row.total_sold, row.total_sold,
row.baskets, row.baskets,
row.notifies, row.notifies,
row.date_last_sold validateDate(row.date_last_sold),
]); imageUrls.image,
imageUrls.image_175,
imageUrls.image_full,
null,
null
];
});
const [result] = await localConnection.query(` const [result] = await localConnection.query(`
WITH inserted_products AS ( WITH inserted_products AS (
INSERT INTO products ( INSERT INTO products (
pid, title, description, SKU, stock_quantity, pending_qty, preorder_count, pid, title, description, SKU, stock_quantity, preorder_count, notions_inv_count,
notions_inv_count, price, regular_price, cost_price, vendor, vendor_reference, price, regular_price, cost_price, vendor, vendor_reference, notions_reference,
notions_reference, brand, line, subline, artist, category_ids, created_at, brand, line, subline, artist, categories, created_at, first_received,
first_received, landing_cost_price, barcode, harmonized_tariff_code, landing_cost_price, barcode, harmonized_tariff_code, updated_at, visible,
updated_at, visible, replenishable, permalink, moq, rating, reviews, managing_stock, replenishable, permalink, moq, uom, rating, reviews,
weight, length, width, height, country_of_origin, location, total_sold, weight, length, width, height, country_of_origin, location, total_sold,
baskets, notifies, date_last_sold baskets, notifies, date_last_sold, image, image_175, image_full, options, tags
) )
VALUES ${placeholders} VALUES ${placeholders}
ON CONFLICT (pid) DO NOTHING ON CONFLICT (pid) DO NOTHING
@@ -285,7 +312,7 @@ async function importMissingProducts(prodConnection, localConnection, missingPid
} }
} }
async function materializeCalculations(prodConnection, localConnection, incrementalUpdate = true, lastSyncTime = '1970-01-01') { async function materializeCalculations(prodConnection, localConnection, incrementalUpdate = true, lastSyncTime = '1970-01-01', startTime = Date.now()) {
outputProgress({ outputProgress({
status: "running", status: "running",
operation: "Products import", operation: "Products import",
@@ -315,36 +342,8 @@ async function materializeCalculations(prodConnection, localConnection, incremen
) THEN 0 ) THEN 0
ELSE 1 ELSE 1
END AS replenishable, END AS replenishable,
COALESCE(si.available_local, 0) - COALESCE( COALESCE(si.available_local, 0) as stock_quantity,
(SELECT SUM(oi.qty_ordered - oi.qty_placed) 0 as pending_qty,
FROM order_items oi
JOIN _order o ON oi.order_id = o.order_id
WHERE oi.prod_pid = p.pid
AND o.date_placed != '0000-00-00 00:00:00'
AND o.date_shipped = '0000-00-00 00:00:00'
AND oi.pick_finished = 0
AND oi.qty_back = 0
AND o.order_status != 15
AND o.order_status < 90
AND oi.qty_ordered >= oi.qty_placed
AND oi.qty_ordered > 0
), 0
) as stock_quantity,
COALESCE(
(SELECT SUM(oi.qty_ordered - oi.qty_placed)
FROM order_items oi
JOIN _order o ON oi.order_id = o.order_id
WHERE oi.prod_pid = p.pid
AND o.date_placed != '0000-00-00 00:00:00'
AND o.date_shipped = '0000-00-00 00:00:00'
AND oi.pick_finished = 0
AND oi.qty_back = 0
AND o.order_status != 15
AND o.order_status < 90
AND oi.qty_ordered >= oi.qty_placed
AND oi.qty_ordered > 0
), 0
) as pending_qty,
COALESCE(ci.onpreorder, 0) as preorder_count, COALESCE(ci.onpreorder, 0) as preorder_count,
COALESCE(pnb.inventory, 0) as notions_inv_count, COALESCE(pnb.inventory, 0) as notions_inv_count,
COALESCE(pcp.price_each, 0) as price, COALESCE(pcp.price_each, 0) as price,
@@ -423,18 +422,18 @@ async function materializeCalculations(prodConnection, localConnection, incremen
await withRetry(async () => { await withRetry(async () => {
const placeholders = batch.map((_, idx) => { const placeholders = batch.map((_, idx) => {
const offset = idx * 41; // 41 columns const base = idx * 47; // 47 columns
return `(${Array.from({ length: 41 }, (_, i) => `$${offset + i + 1}`).join(', ')})`; return `(${Array.from({ length: 47 }, (_, i) => `$${base + i + 1}`).join(', ')})`;
}).join(','); }).join(',');
const values = batch.flatMap(row => [ const values = batch.flatMap(row => {
const imageUrls = getImageUrls(row.pid);
return [
row.pid, row.pid,
row.title, row.title,
row.description, row.description,
row.SKU, row.itemnumber || '',
// Set stock quantity to 0 if it's over 5000
row.stock_quantity > 5000 ? 0 : Math.max(0, row.stock_quantity), row.stock_quantity > 5000 ? 0 : Math.max(0, row.stock_quantity),
row.pending_qty,
row.preorder_count, row.preorder_count,
row.notions_inv_count, row.notions_inv_count,
row.price, row.price,
@@ -448,17 +447,19 @@ async function materializeCalculations(prodConnection, localConnection, incremen
row.subline, row.subline,
row.artist, row.artist,
row.category_ids, row.category_ids,
row.date_created, // map to created_at validateDate(row.date_created),
row.first_received, validateDate(row.first_received),
row.landing_cost_price, row.landing_cost_price,
row.barcode, row.barcode,
row.harmonized_tariff_code, row.harmonized_tariff_code,
row.updated_at, validateDate(row.updated_at),
row.visible, row.visible,
true,
row.replenishable, row.replenishable,
row.permalink, row.permalink,
row.moq, Math.max(1, Math.round(row.moq || 1)),
row.rating ? Number(row.rating).toFixed(2) : null, 1,
row.rating,
row.reviews, row.reviews,
row.weight, row.weight,
row.length, row.length,
@@ -469,25 +470,30 @@ async function materializeCalculations(prodConnection, localConnection, incremen
row.total_sold, row.total_sold,
row.baskets, row.baskets,
row.notifies, row.notifies,
row.date_last_sold validateDate(row.date_last_sold),
]); imageUrls.image,
imageUrls.image_175,
imageUrls.image_full,
null,
null
];
});
await localConnection.query(` await localConnection.query(`
INSERT INTO temp_products ( INSERT INTO temp_products (
pid, title, description, SKU, stock_quantity, pending_qty, preorder_count, pid, title, description, SKU, stock_quantity, preorder_count, notions_inv_count,
notions_inv_count, price, regular_price, cost_price, vendor, vendor_reference, price, regular_price, cost_price, vendor, vendor_reference, notions_reference,
notions_reference, brand, line, subline, artist, category_ids, created_at, brand, line, subline, artist, categories, created_at, first_received,
first_received, landing_cost_price, barcode, harmonized_tariff_code, landing_cost_price, barcode, harmonized_tariff_code, updated_at, visible,
updated_at, visible, replenishable, permalink, moq, rating, reviews, managing_stock, replenishable, permalink, moq, uom, rating, reviews,
weight, length, width, height, country_of_origin, location, total_sold, weight, length, width, height, country_of_origin, location, total_sold,
baskets, notifies, date_last_sold baskets, notifies, date_last_sold, image, image_175, image_full, options, tags
) VALUES ${placeholders} ) VALUES ${placeholders}
ON CONFLICT (pid) DO UPDATE SET ON CONFLICT (pid) DO UPDATE SET
title = EXCLUDED.title, title = EXCLUDED.title,
description = EXCLUDED.description, description = EXCLUDED.description,
SKU = EXCLUDED.SKU, SKU = EXCLUDED.SKU,
stock_quantity = EXCLUDED.stock_quantity, stock_quantity = EXCLUDED.stock_quantity,
pending_qty = EXCLUDED.pending_qty,
preorder_count = EXCLUDED.preorder_count, preorder_count = EXCLUDED.preorder_count,
notions_inv_count = EXCLUDED.notions_inv_count, notions_inv_count = EXCLUDED.notions_inv_count,
price = EXCLUDED.price, price = EXCLUDED.price,
@@ -500,7 +506,6 @@ async function materializeCalculations(prodConnection, localConnection, incremen
line = EXCLUDED.line, line = EXCLUDED.line,
subline = EXCLUDED.subline, subline = EXCLUDED.subline,
artist = EXCLUDED.artist, artist = EXCLUDED.artist,
category_ids = EXCLUDED.category_ids,
created_at = EXCLUDED.created_at, created_at = EXCLUDED.created_at,
first_received = EXCLUDED.first_received, first_received = EXCLUDED.first_received,
landing_cost_price = EXCLUDED.landing_cost_price, landing_cost_price = EXCLUDED.landing_cost_price,
@@ -508,9 +513,11 @@ async function materializeCalculations(prodConnection, localConnection, incremen
harmonized_tariff_code = EXCLUDED.harmonized_tariff_code, harmonized_tariff_code = EXCLUDED.harmonized_tariff_code,
updated_at = EXCLUDED.updated_at, updated_at = EXCLUDED.updated_at,
visible = EXCLUDED.visible, visible = EXCLUDED.visible,
managing_stock = EXCLUDED.managing_stock,
replenishable = EXCLUDED.replenishable, replenishable = EXCLUDED.replenishable,
permalink = EXCLUDED.permalink, permalink = EXCLUDED.permalink,
moq = EXCLUDED.moq, moq = EXCLUDED.moq,
uom = EXCLUDED.uom,
rating = EXCLUDED.rating, rating = EXCLUDED.rating,
reviews = EXCLUDED.reviews, reviews = EXCLUDED.reviews,
weight = EXCLUDED.weight, weight = EXCLUDED.weight,
@@ -522,7 +529,12 @@ async function materializeCalculations(prodConnection, localConnection, incremen
total_sold = EXCLUDED.total_sold, total_sold = EXCLUDED.total_sold,
baskets = EXCLUDED.baskets, baskets = EXCLUDED.baskets,
notifies = EXCLUDED.notifies, notifies = EXCLUDED.notifies,
date_last_sold = EXCLUDED.date_last_sold date_last_sold = EXCLUDED.date_last_sold,
image = EXCLUDED.image,
image_175 = EXCLUDED.image_175,
image_full = EXCLUDED.image_full,
options = EXCLUDED.options,
tags = EXCLUDED.tags
`, values); `, values);
}, `Error inserting batch ${i} to ${i + batch.length}`); }, `Error inserting batch ${i} to ${i + batch.length}`);
@@ -560,11 +572,15 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
} }
} }
// Start a transaction to ensure temporary tables persist
await localConnection.beginTransaction();
try {
// Setup temporary tables // Setup temporary tables
await setupTemporaryTables(localConnection); await setupTemporaryTables(localConnection);
// Materialize calculations into temp table // Materialize calculations into temp table
await materializeCalculations(prodConnection, localConnection, incrementalUpdate, lastSyncTime); await materializeCalculations(prodConnection, localConnection, incrementalUpdate, lastSyncTime, startTime);
// Get the list of products that need updating // Get the list of products that need updating
const [products] = await localConnection.query(` const [products] = await localConnection.query(`
@@ -574,7 +590,6 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
t.description, t.description,
t.SKU, t.SKU,
t.stock_quantity, t.stock_quantity,
t.pending_qty,
t.preorder_count, t.preorder_count,
t.notions_inv_count, t.notions_inv_count,
t.price, t.price,
@@ -587,7 +602,7 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
t.line, t.line,
t.subline, t.subline,
t.artist, t.artist,
t.category_ids, t.categories,
t.created_at, t.created_at,
t.first_received, t.first_received,
t.landing_cost_price, t.landing_cost_price,
@@ -595,6 +610,7 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
t.harmonized_tariff_code, t.harmonized_tariff_code,
t.updated_at, t.updated_at,
t.visible, t.visible,
t.managing_stock,
t.replenishable, t.replenishable,
t.permalink, t.permalink,
t.moq, t.moq,
@@ -609,30 +625,36 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
t.total_sold, t.total_sold,
t.baskets, t.baskets,
t.notifies, t.notifies,
t.date_last_sold t.date_last_sold,
t.image,
t.image_175,
t.image_full,
t.options,
t.tags
FROM temp_products t FROM temp_products t
WHERE t.needs_update = true
`); `);
// Process products in batches
let recordsAdded = 0; let recordsAdded = 0;
let recordsUpdated = 0; let recordsUpdated = 0;
// Process products in batches for (let i = 0; i < products.rows.length; i += BATCH_SIZE) {
for (let i = 0; i < products.length; i += BATCH_SIZE) { const batch = products.rows.slice(i, i + BATCH_SIZE);
const batch = products.slice(i, Math.min(i + BATCH_SIZE, products.length));
await withRetry(async () => {
const placeholders = batch.map((_, idx) => { const placeholders = batch.map((_, idx) => {
const offset = idx * 41; // 41 columns const base = idx * 47; // 47 columns
return `(${Array.from({ length: 41 }, (_, i) => `$${offset + i + 1}`).join(', ')})`; return `(${Array.from({ length: 47 }, (_, i) => `$${base + i + 1}`).join(', ')})`;
}).join(','); }).join(',');
const values = batch.flatMap(row => [ const values = batch.flatMap(row => {
const imageUrls = getImageUrls(row.pid);
return [
row.pid, row.pid,
row.title, row.title,
row.description, row.description,
row.SKU, row.SKU || '',
row.stock_quantity, row.stock_quantity > 5000 ? 0 : Math.max(0, row.stock_quantity),
row.pending_qty,
row.preorder_count, row.preorder_count,
row.notions_inv_count, row.notions_inv_count,
row.price, row.price,
@@ -645,17 +667,19 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
row.line, row.line,
row.subline, row.subline,
row.artist, row.artist,
row.category_ids, row.categories,
row.created_at, validateDate(row.created_at),
row.first_received, validateDate(row.first_received),
row.landing_cost_price, row.landing_cost_price,
row.barcode, row.barcode,
row.harmonized_tariff_code, row.harmonized_tariff_code,
row.updated_at, validateDate(row.updated_at),
row.visible, row.visible,
row.managing_stock,
row.replenishable, row.replenishable,
row.permalink, row.permalink,
row.moq, row.moq,
1,
row.rating, row.rating,
row.reviews, row.reviews,
row.weight, row.weight,
@@ -667,26 +691,32 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
row.total_sold, row.total_sold,
row.baskets, row.baskets,
row.notifies, row.notifies,
row.date_last_sold validateDate(row.date_last_sold),
]); imageUrls.image,
imageUrls.image_175,
imageUrls.image_full,
row.options,
row.tags
];
});
const result = await localConnection.query(` const [result] = await localConnection.query(`
WITH upserted_products AS ( WITH upserted AS (
INSERT INTO products ( INSERT INTO products (
pid, title, description, SKU, stock_quantity, pending_qty, preorder_count, pid, title, description, SKU, stock_quantity, preorder_count, notions_inv_count,
notions_inv_count, price, regular_price, cost_price, vendor, vendor_reference, price, regular_price, cost_price, vendor, vendor_reference, notions_reference,
notions_reference, brand, line, subline, artist, category_ids, created_at, brand, line, subline, artist, categories, created_at, first_received,
first_received, landing_cost_price, barcode, harmonized_tariff_code, landing_cost_price, barcode, harmonized_tariff_code, updated_at, visible,
updated_at, visible, replenishable, permalink, moq, rating, reviews, managing_stock, replenishable, permalink, moq, uom, rating, reviews,
weight, length, width, height, country_of_origin, location, total_sold, weight, length, width, height, country_of_origin, location, total_sold,
baskets, notifies, date_last_sold baskets, notifies, date_last_sold, image, image_175, image_full, options, tags
) VALUES ${placeholders} )
VALUES ${placeholders}
ON CONFLICT (pid) DO UPDATE SET ON CONFLICT (pid) DO UPDATE SET
title = EXCLUDED.title, title = EXCLUDED.title,
description = EXCLUDED.description, description = EXCLUDED.description,
SKU = EXCLUDED.SKU, SKU = EXCLUDED.SKU,
stock_quantity = EXCLUDED.stock_quantity, stock_quantity = EXCLUDED.stock_quantity,
pending_qty = EXCLUDED.pending_qty,
preorder_count = EXCLUDED.preorder_count, preorder_count = EXCLUDED.preorder_count,
notions_inv_count = EXCLUDED.notions_inv_count, notions_inv_count = EXCLUDED.notions_inv_count,
price = EXCLUDED.price, price = EXCLUDED.price,
@@ -699,7 +729,6 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
line = EXCLUDED.line, line = EXCLUDED.line,
subline = EXCLUDED.subline, subline = EXCLUDED.subline,
artist = EXCLUDED.artist, artist = EXCLUDED.artist,
category_ids = EXCLUDED.category_ids,
created_at = EXCLUDED.created_at, created_at = EXCLUDED.created_at,
first_received = EXCLUDED.first_received, first_received = EXCLUDED.first_received,
landing_cost_price = EXCLUDED.landing_cost_price, landing_cost_price = EXCLUDED.landing_cost_price,
@@ -707,9 +736,11 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
harmonized_tariff_code = EXCLUDED.harmonized_tariff_code, harmonized_tariff_code = EXCLUDED.harmonized_tariff_code,
updated_at = EXCLUDED.updated_at, updated_at = EXCLUDED.updated_at,
visible = EXCLUDED.visible, visible = EXCLUDED.visible,
managing_stock = EXCLUDED.managing_stock,
replenishable = EXCLUDED.replenishable, replenishable = EXCLUDED.replenishable,
permalink = EXCLUDED.permalink, permalink = EXCLUDED.permalink,
moq = EXCLUDED.moq, moq = EXCLUDED.moq,
uom = EXCLUDED.uom,
rating = EXCLUDED.rating, rating = EXCLUDED.rating,
reviews = EXCLUDED.reviews, reviews = EXCLUDED.reviews,
weight = EXCLUDED.weight, weight = EXCLUDED.weight,
@@ -721,58 +752,82 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
total_sold = EXCLUDED.total_sold, total_sold = EXCLUDED.total_sold,
baskets = EXCLUDED.baskets, baskets = EXCLUDED.baskets,
notifies = EXCLUDED.notifies, notifies = EXCLUDED.notifies,
date_last_sold = EXCLUDED.date_last_sold date_last_sold = EXCLUDED.date_last_sold,
image = EXCLUDED.image,
image_175 = EXCLUDED.image_175,
image_full = EXCLUDED.image_full,
options = EXCLUDED.options,
tags = EXCLUDED.tags
RETURNING RETURNING
CASE WHEN xmax::text::int > 0 THEN 0 ELSE 1 END AS inserted, xmax = 0 as inserted
CASE WHEN xmax::text::int > 0 THEN 1 ELSE 0 END AS updated
) )
SELECT SELECT
COUNT(*) FILTER (WHERE inserted = 1) as inserted, COUNT(*) FILTER (WHERE inserted) as inserted,
COUNT(*) FILTER (WHERE updated = 1) as updated COUNT(*) FILTER (WHERE NOT inserted) as updated
FROM upserted_products FROM upserted
`, values); `, values);
recordsAdded += result.rows[0].inserted; recordsAdded += parseInt(result.rows[0].inserted, 10) || 0;
recordsUpdated += result.rows[0].updated; recordsUpdated += parseInt(result.rows[0].updated, 10) || 0;
}, `Error inserting batch ${i} to ${i + batch.length}`);
// Process category relationships for each product in the batch
for (const row of batch) {
if (row.categories) {
const categoryIds = row.categories.split(',').filter(id => id && id.trim());
if (categoryIds.length > 0) {
const catPlaceholders = categoryIds.map((_, idx) =>
`($${idx * 2 + 1}, $${idx * 2 + 2})`
).join(',');
const catValues = categoryIds.flatMap(catId => [row.pid, parseInt(catId.trim(), 10)]);
// First delete existing relationships for this product
await localConnection.query(
'DELETE FROM product_categories WHERE pid = $1',
[row.pid]
);
// Then insert the new relationships
await localConnection.query(`
INSERT INTO product_categories (pid, cat_id)
VALUES ${catPlaceholders}
ON CONFLICT (pid, cat_id) DO NOTHING
`, catValues);
}
}
}
outputProgress({ outputProgress({
status: "running", status: "running",
operation: "Products import", operation: "Products import",
message: `Imported ${i + batch.length} of ${products.length} products`, message: `Processing products: ${i + batch.length} of ${products.rows.length}`,
current: i + batch.length, current: i + batch.length,
total: products.length, total: products.rows.length,
elapsed: formatElapsedTime((Date.now() - startTime) / 1000), elapsed: formatElapsedTime((Date.now() - startTime) / 1000),
remaining: estimateRemaining(startTime, i + batch.length, products.length), remaining: estimateRemaining(startTime, i + batch.length, products.rows.length),
rate: calculateRate(startTime, i + batch.length) rate: calculateRate(startTime, i + batch.length)
}); });
} }
// Update sync status
await localConnection.query(`
INSERT INTO sync_status (table_name, last_sync_timestamp)
VALUES ('products', NOW())
ON CONFLICT (table_name) DO UPDATE SET
last_sync_timestamp = NOW()
`);
// Cleanup temporary tables // Cleanup temporary tables
await cleanupTemporaryTables(localConnection); await cleanupTemporaryTables(localConnection);
// Commit the transaction
await localConnection.commit();
return { return {
status: "complete", status: 'complete',
recordsAdded, recordsAdded,
recordsUpdated, recordsUpdated,
totalRecords: products.length totalRecords: products.rows.length,
duration: formatElapsedTime(Date.now() - startTime)
}; };
} catch (error) { } catch (error) {
console.error('Error in importProducts:', error); // Rollback on error
// Attempt cleanup on error await localConnection.rollback();
try { throw error;
await cleanupTemporaryTables(localConnection);
} catch (cleanupError) {
console.error('Error during cleanup:', cleanupError);
} }
} catch (error) {
console.error('Error in importProducts:', error);
throw error; throw error;
} }
} }

53
inventory-server/scripts/import/utils.js
View File

@@ -64,7 +64,12 @@ async function setupConnections(sshConfig) {
// Create a wrapper for the PostgreSQL pool to match MySQL interface // Create a wrapper for the PostgreSQL pool to match MySQL interface
const localConnection = { const localConnection = {
_client: null,
_transactionActive: false,
query: async (text, params) => { query: async (text, params) => {
// If we're not in a transaction, use the pool directly
if (!localConnection._transactionActive) {
const client = await localPool.connect(); const client = await localPool.connect();
try { try {
const result = await client.query(text, params); const result = await client.query(text, params);
@@ -72,17 +77,55 @@ async function setupConnections(sshConfig) {
} finally { } finally {
client.release(); client.release();
} }
}
// If we're in a transaction, use the dedicated client
if (!localConnection._client) {
throw new Error('No active transaction client');
}
const result = await localConnection._client.query(text, params);
return [result];
}, },
beginTransaction: async () => {
if (localConnection._transactionActive) {
throw new Error('Transaction already active');
}
localConnection._client = await localPool.connect();
await localConnection._client.query('BEGIN');
localConnection._transactionActive = true;
},
commit: async () => {
if (!localConnection._transactionActive) {
throw new Error('No active transaction to commit');
}
await localConnection._client.query('COMMIT');
localConnection._client.release();
localConnection._client = null;
localConnection._transactionActive = false;
},
rollback: async () => {
if (!localConnection._transactionActive) {
throw new Error('No active transaction to rollback');
}
await localConnection._client.query('ROLLBACK');
localConnection._client.release();
localConnection._client = null;
localConnection._transactionActive = false;
},
end: async () => { end: async () => {
if (localConnection._client) {
localConnection._client.release();
localConnection._client = null;
}
await localPool.end(); await localPool.end();
} }
}; };
return { return { prodConnection, localConnection, tunnel };
ssh: tunnel.ssh,
prodConnection,
localConnection
};
} }
// Helper function to close connections // Helper function to close connections