More import script updates and fixes, better import_history tracking

This commit is contained in:
2025-01-31 13:12:43 -05:00
parent a867117c3c
commit 1c932e0df5
6 changed files with 41 additions and 14 deletions

View File

@@ -184,6 +184,7 @@ CREATE TABLE IF NOT EXISTS import_history (
start_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, start_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
end_time TIMESTAMP NULL, end_time TIMESTAMP NULL,
duration_seconds INT, duration_seconds INT,
duration_minutes DECIMAL(10,2) GENERATED ALWAYS AS (duration_seconds / 60.0) STORED,
records_added INT DEFAULT 0, records_added INT DEFAULT 0,
records_updated INT DEFAULT 0, records_updated INT DEFAULT 0,
is_incremental BOOLEAN DEFAULT FALSE, is_incremental BOOLEAN DEFAULT FALSE,

View File

@@ -118,6 +118,7 @@ CREATE TABLE IF NOT EXISTS orders (
status VARCHAR(20) DEFAULT 'pending', status VARCHAR(20) DEFAULT 'pending',
canceled TINYINT(1) DEFAULT 0, canceled TINYINT(1) DEFAULT 0,
PRIMARY KEY (id), PRIMARY KEY (id),
UNIQUE KEY unique_order_line (order_number, pid),
KEY order_number (order_number), KEY order_number (order_number),
KEY pid (pid), KEY pid (pid),
KEY customer (customer), KEY customer (customer),

View File

@@ -161,6 +161,7 @@ async function main() {
results.categories = await importCategories(prodConnection, localConnection); results.categories = await importCategories(prodConnection, localConnection);
if (isImportCancelled) throw new Error("Import cancelled"); if (isImportCancelled) throw new Error("Import cancelled");
completedSteps++; completedSteps++;
console.log('Categories import result:', results.categories);
totalRecordsAdded += results.categories?.recordsAdded || 0; totalRecordsAdded += results.categories?.recordsAdded || 0;
totalRecordsUpdated += results.categories?.recordsUpdated || 0; totalRecordsUpdated += results.categories?.recordsUpdated || 0;
} }
@@ -169,6 +170,7 @@ async function main() {
results.products = await importProducts(prodConnection, localConnection, INCREMENTAL_UPDATE); results.products = await importProducts(prodConnection, localConnection, INCREMENTAL_UPDATE);
if (isImportCancelled) throw new Error("Import cancelled"); if (isImportCancelled) throw new Error("Import cancelled");
completedSteps++; completedSteps++;
console.log('Products import result:', results.products);
totalRecordsAdded += results.products?.recordsAdded || 0; totalRecordsAdded += results.products?.recordsAdded || 0;
totalRecordsUpdated += results.products?.recordsUpdated || 0; totalRecordsUpdated += results.products?.recordsUpdated || 0;
} }
@@ -177,6 +179,7 @@ async function main() {
results.orders = await importOrders(prodConnection, localConnection, INCREMENTAL_UPDATE); results.orders = await importOrders(prodConnection, localConnection, INCREMENTAL_UPDATE);
if (isImportCancelled) throw new Error("Import cancelled"); if (isImportCancelled) throw new Error("Import cancelled");
completedSteps++; completedSteps++;
console.log('Orders import result:', results.orders);
totalRecordsAdded += results.orders?.recordsAdded || 0; totalRecordsAdded += results.orders?.recordsAdded || 0;
totalRecordsUpdated += results.orders?.recordsUpdated || 0; totalRecordsUpdated += results.orders?.recordsUpdated || 0;
} }
@@ -185,6 +188,7 @@ async function main() {
results.purchaseOrders = await importPurchaseOrders(prodConnection, localConnection, INCREMENTAL_UPDATE); results.purchaseOrders = await importPurchaseOrders(prodConnection, localConnection, INCREMENTAL_UPDATE);
if (isImportCancelled) throw new Error("Import cancelled"); if (isImportCancelled) throw new Error("Import cancelled");
completedSteps++; completedSteps++;
console.log('Purchase orders import result:', results.purchaseOrders);
totalRecordsAdded += results.purchaseOrders?.recordsAdded || 0; totalRecordsAdded += results.purchaseOrders?.recordsAdded || 0;
totalRecordsUpdated += results.purchaseOrders?.recordsUpdated || 0; totalRecordsUpdated += results.purchaseOrders?.recordsUpdated || 0;
} }

View File

@@ -73,6 +73,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
`); `);
// Get base order items first // Get base order items first
console.log('Last sync time:', lastSyncTime);
const [orderItems] = await prodConnection.query(` const [orderItems] = await prodConnection.query(`
SELECT SELECT
oi.order_id, oi.order_id,
@@ -89,15 +90,13 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
AND o.date_placed_onlydate IS NOT NULL AND o.date_placed_onlydate IS NOT NULL
${incrementalUpdate ? ` ${incrementalUpdate ? `
AND ( AND (
o.stamp > ? o.stamp > ?
OR oi.stamp > ? OR oi.stamp > ?
OR o.date_placed > ?
OR o.date_shipped > ?
OR o.date_cancelled > ?
OR o.date_updated > ?
) )
` : ''} ` : ''}
`, incrementalUpdate ? [lastSyncTime, lastSyncTime, lastSyncTime, lastSyncTime, lastSyncTime, lastSyncTime] : []); `, incrementalUpdate ? [lastSyncTime, lastSyncTime] : []);
console.log('Found', orderItems.length, 'orders to process');
const totalOrders = orderItems.length; const totalOrders = orderItems.length;
let processed = 0; let processed = 0;
@@ -111,7 +110,13 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
]); ]);
await localConnection.query(` await localConnection.query(`
INSERT INTO temp_order_items VALUES ${placeholders} INSERT INTO temp_order_items (order_id, pid, SKU, price, quantity, base_discount)
VALUES ${placeholders}
ON DUPLICATE KEY UPDATE
SKU = VALUES(SKU),
price = VALUES(price),
quantity = VALUES(quantity),
base_discount = VALUES(base_discount)
`, values); `, values);
processed += batch.length; processed += batch.length;
@@ -279,12 +284,26 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
INSERT INTO orders (${columnNames.join(",")}) INSERT INTO orders (${columnNames.join(",")})
VALUES ${placeholders} VALUES ${placeholders}
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE
${columnNames.map(col => `${col} = VALUES(${col})`).join(",")} SKU = VALUES(SKU),
date = VALUES(date),
price = VALUES(price),
quantity = VALUES(quantity),
discount = VALUES(discount),
tax = VALUES(tax),
tax_included = VALUES(tax_included),
shipping = VALUES(shipping),
customer = VALUES(customer),
customer_name = VALUES(customer_name),
status = VALUES(status),
canceled = VALUES(canceled)
`; `;
const result = await localConnection.query(query, values.flat()); const result = await localConnection.query(query, values);
recordsAdded += result.affectedRows - result.changedRows; // For INSERT ... ON DUPLICATE KEY UPDATE:
recordsUpdated += result.changedRows; // - affectedRows is 1 for each inserted row and 2 for each updated row
// - changedRows is 1 for each row that was actually changed during update
recordsAdded += result[0].affectedRows - (2 * result[0].changedRows); // New rows
recordsUpdated += result[0].changedRows; // Actually changed rows
importedCount += validOrders.length; importedCount += validOrders.length;
} }

View File

@@ -200,6 +200,8 @@ async function materializeCalculations(prodConnection, localConnection) {
async function importProducts(prodConnection, localConnection, incrementalUpdate = true) { async function importProducts(prodConnection, localConnection, incrementalUpdate = true) {
const startTime = Date.now(); const startTime = Date.now();
let recordsAdded = 0;
let recordsUpdated = 0;
try { try {
// Get column names first // Get column names first
@@ -471,8 +473,8 @@ async function importProducts(prodConnection, localConnection, incrementalUpdate
`; `;
const result = await localConnection.query(insertQuery, productValues); const result = await localConnection.query(insertQuery, productValues);
recordsAdded += result.affectedRows - result.changedRows; recordsAdded += result.affectedRows - (2 * result.changedRows); // New rows
recordsUpdated += result.changedRows; recordsUpdated += result.changedRows; // Actually changed rows
} }
// Insert category relationships // Insert category relationships

View File

@@ -329,7 +329,7 @@ async function importPurchaseOrders(prodConnection, localConnection, incremental
`; `;
const result = await localConnection.query(query, values.flat()); const result = await localConnection.query(query, values.flat());
recordsAdded += result.affectedRows - result.changedRows; recordsAdded += result.affectedRows - (2 * result.changedRows);
recordsUpdated += result.changedRows; recordsUpdated += result.changedRows;
} }