More incremental import fixes

Incremental import order fixes
2025-02-01 11:42:51 -05:00 · 2025-02-01 11:03:42 -05:00
4 changed files with 110 additions and 57 deletions
--- a/inventory-server/scripts/import-from-prod.js
+++ b/inventory-server/scripts/import-from-prod.js
@@ -10,9 +10,9 @@ const importPurchaseOrders = require('./import/purchase-orders');
 dotenv.config({ path: path.join(__dirname, "../.env") });

 // Constants to control which imports run
-const IMPORT_CATEGORIES = false;
-const IMPORT_PRODUCTS = false;
-const IMPORT_ORDERS = false;
+const IMPORT_CATEGORIES = true;
+const IMPORT_PRODUCTS = true;
+const IMPORT_ORDERS = true;
 const IMPORT_PURCHASE_ORDERS = true;

 // Add flag for incremental updates
--- a/inventory-server/scripts/import/orders.js
+++ b/inventory-server/scripts/import/orders.js
@@ -19,6 +19,10 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
  const missingProducts = new Set();
  let recordsAdded = 0;
  let recordsUpdated = 0;
+  let processedCount = 0;
+  let importedCount = 0;
+  let totalOrderItems = 0;
+  let totalUniqueOrders = 0;

  try {
    // Insert temporary table creation queries
@@ -86,7 +90,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =

    console.log('Orders: Using last sync time:', lastSyncTime);

-    // First get all relevant order items with basic info
+    // First get count of order items
    const [[{ total }]] = await prodConnection.query(`
      SELECT COUNT(*) as total
      FROM order_items oi
@@ -115,7 +119,8 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
        ` : ''}
    `, incrementalUpdate ? [lastSyncTime, lastSyncTime, lastSyncTime] : []);

-    console.log('Orders: Found changes:', total);
+    totalOrderItems = total;
+    console.log('Orders: Found changes:', totalOrderItems);

    // Get order items in batches
    const [orderItems] = await prodConnection.query(`
@@ -155,9 +160,6 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =

    console.log('Orders: Processing', orderItems.length, 'order items');

-    const totalOrders = orderItems.length;
-    let processed = 0;
-
    // Insert order items in batches
    for (let i = 0; i < orderItems.length; i += 5000) {
      const batch = orderItems.slice(i, Math.min(i + 5000, orderItems.length));
@@ -176,22 +178,30 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
          base_discount = VALUES(base_discount)
      `, values);

-      processed += batch.length;
+      processedCount = i + batch.length;
      outputProgress({
        status: "running",
        operation: "Orders import",
-        message: `Loading order items: ${processed} of ${totalOrders}`,
-        current: processed,
-        total: totalOrders
+        message: `Loading order items: ${processedCount} of ${totalOrderItems}`,
+        current: processedCount,
+        total: totalOrderItems
      });
    }

    // Get unique order IDs
    const orderIds = [...new Set(orderItems.map(item => item.order_id))];
+    totalUniqueOrders = orderIds.length;
+    console.log('Total unique order IDs:', totalUniqueOrders);
+
+    // Reset processed count for order processing phase
+    processedCount = 0;

    // Get order metadata in batches
    for (let i = 0; i < orderIds.length; i += 5000) {
      const batchIds = orderIds.slice(i, i + 5000);
+      console.log(`Processing batch ${i/5000 + 1}, size: ${batchIds.length}`);
+      console.log('Sample of batch IDs:', batchIds.slice(0, 5));
+      
      const [orders] = await prodConnection.query(`
        SELECT 
          o.order_id,
@@ -204,6 +214,14 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
        LEFT JOIN users u ON o.order_cid = u.cid
        WHERE o.order_id IN (?)
      `, [batchIds]);
+      
+      console.log(`Retrieved ${orders.length} orders for ${batchIds.length} IDs`);
+      const duplicates = orders.filter((order, index, self) => 
+        self.findIndex(o => o.order_id === order.order_id) !== index
+      );
+      if (duplicates.length > 0) {
+        console.log('Found duplicates:', duplicates);
+      }

      const placeholders = orders.map(() => "(?, ?, ?, ?, ?, ?)").join(",");
      const values = orders.flatMap(order => [
@@ -212,17 +230,27 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =

      await localConnection.query(`
        INSERT INTO temp_order_meta VALUES ${placeholders}
+        ON DUPLICATE KEY UPDATE
+          date = VALUES(date),
+          customer = VALUES(customer),
+          customer_name = VALUES(customer_name),
+          status = VALUES(status),
+          canceled = VALUES(canceled)
      `, values);

+      processedCount = i + orders.length;
      outputProgress({
        status: "running",
        operation: "Orders import",
-        message: `Loading order metadata: ${i + orders.length} of ${orderIds.length}`,
-        current: i + orders.length,
-        total: orderIds.length
+        message: `Loading order metadata: ${processedCount} of ${totalUniqueOrders}`,
+        current: processedCount,
+        total: totalUniqueOrders
      });
    }

+    // Reset processed count for final phase
+    processedCount = 0;
+
    // Get promotional discounts in batches
    for (let i = 0; i < orderIds.length; i += 5000) {
      const batchIds = orderIds.slice(i, i + 5000);
@@ -239,6 +267,8 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =

        await localConnection.query(`
          INSERT INTO temp_order_discounts VALUES ${placeholders}
+          ON DUPLICATE KEY UPDATE
+            discount = VALUES(discount)
        `, values);
      }
    }
@@ -274,6 +304,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
          const placeholders = Array(uniqueTaxes.size).fill("(?, ?, ?)").join(",");
          await localConnection.query(`
            INSERT INTO temp_order_taxes VALUES ${placeholders}
+            ON DUPLICATE KEY UPDATE tax = VALUES(tax)
          `, values);
        }
      }
@@ -300,8 +331,6 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
    }

    // Now combine all the data and insert into orders table
-    let importedCount = 0;
-
    // Pre-check all products at once instead of per batch
    const allOrderPids = [...new Set(orderItems.map(item => item.pid))];
    const [existingProducts] = allOrderPids.length > 0 ? await localConnection.query(
@@ -382,24 +411,10 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
              return newVal !== oldVal;
            });
            if (hasChanges) {
-              acc.updates.push({
-                order_number: order.order_number,
-                pid: order.pid,
-                values: columnNames.map(col => order[col] ?? null)
-              });
-            } else {
-              acc.inserts.push({
-                order_number: order.order_number,
-                pid: order.pid,
-                values: columnNames.map(col => order[col] ?? null)
-              });
+              acc.updates.push(order);
            }
          } else {
-            acc.inserts.push({
-              order_number: order.order_number,
-              pid: order.pid,
-              values: columnNames.map(col => order[col] ?? null)
-            });
+            acc.inserts.push(order);
          }
          return acc;
        }, { inserts: [], updates: [] });
@@ -411,9 +426,10 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
          const insertResult = await localConnection.query(`
            INSERT INTO orders (${columnNames.join(",")})
            VALUES ${insertPlaceholders}
-          `, insertsAndUpdates.inserts.map(i => i.values).flat());
+          `, insertsAndUpdates.inserts.map(i => columnNames.map(col => i[col] ?? null)).flat());
          
          recordsAdded += insertResult[0].affectedRows;
+          importedCount += insertResult[0].affectedRows;
        }

        // Handle updates - now we know these actually have changes
@@ -435,24 +451,26 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
              customer = VALUES(customer),
              customer_name = VALUES(customer_name),
              status = VALUES(status),
-              canceled = VALUES(canceled)
-          `, insertsAndUpdates.updates.map(u => u.values).flat());
+              canceled = VALUES(canceled),
+              costeach = VALUES(costeach)
+          `, insertsAndUpdates.updates.map(u => columnNames.map(col => u[col] ?? null)).flat());
          
          recordsUpdated += updateResult[0].affectedRows / 2;  // Each update counts as 2 in affectedRows
+          importedCount += updateResult[0].affectedRows / 2;
        }
-
-        importedCount += validOrders.length;
      }

+      // Update progress based on batch size - this is the number of order items we've processed
+      processedCount = i + batchIds.length * (totalOrderItems / totalUniqueOrders);
      outputProgress({
        status: "running",
        operation: "Orders import",
-        message: `Imported ${importedCount} of ${totalOrders} orders`,
-        current: importedCount,
-        total: totalOrders,
+        message: `Imported ${Math.floor(importedCount)} orders (${Math.floor(processedCount)} of ${totalOrderItems} items processed)`,
+        current: Math.floor(processedCount),
+        total: totalOrderItems,
        elapsed: formatElapsedTime((Date.now() - startTime) / 1000),
-        remaining: estimateRemaining(startTime, importedCount, totalOrders),
-        rate: calculateRate(startTime, importedCount)
+        remaining: estimateRemaining(startTime, processedCount, totalOrderItems),
+        rate: calculateRate(startTime, processedCount)
      });
    }

@@ -556,21 +574,34 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
            INSERT INTO orders (${columnNames.join(", ")})
            VALUES ${skippedPlaceholders}
            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),
+              costeach = VALUES(costeach)
          `;

          // Execute the insert query
          if (skippedOrderValues.length > 0) {
-            await localConnection.query(skippedInsertQuery, skippedOrderValues.flat());
+            const result = await localConnection.query(skippedInsertQuery, skippedOrderValues.flat());
+            const addedOrUpdated = Math.floor(result[0].affectedRows / 2); // Round down to avoid fractional orders
+            importedCount += addedOrUpdated;
+            recordsUpdated += addedOrUpdated;
+
+            outputProgress({
+              status: "running",
+              operation: "Orders import",
+              message: `Successfully imported ${addedOrUpdated} previously skipped orders`,
+            });
          }
-
-          importedCount += skippedProdOrders.length;
-
-          outputProgress({
-            status: "running",
-            operation: "Orders import",
-            message: `Successfully imported ${skippedProdOrders.length} previously skipped orders`,
-          });
        }
      } catch (error) {
        console.warn('Warning: Failed to import missing products:', error.message);
@@ -587,9 +618,9 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =

    return {
      status: "complete",
-      totalImported: importedCount,
+      totalImported: Math.floor(importedCount), // Round down to avoid fractional orders
      recordsAdded: recordsAdded || 0,
-      recordsUpdated: recordsUpdated || 0,
+      recordsUpdated: Math.floor(recordsUpdated), // Round down to avoid fractional orders
      totalSkipped: skippedOrders.size,
      missingProducts: missingProducts.size,
      incrementalUpdate,
--- a/inventory-server/scripts/import/products.js
+++ b/inventory-server/scripts/import/products.js
@@ -582,7 +582,21 @@ async function importMissingProducts(prodConnection, localConnection, missingPid
          ELSE 1
        END AS replenishable,
        COALESCE(si.available_local, 0) as stock_quantity,
-        COALESCE(pq.qty, 0) as pending_qty,
+        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(pnb.inventory, 0) as notions_inv_count,
        COALESCE(pcp.price_each, 0) as price,
--- a/inventory-server/scripts/import/purchase-orders.js
+++ b/inventory-server/scripts/import/purchase-orders.js
@@ -117,7 +117,12 @@ async function importPurchaseOrders(prodConnection, localConnection, incremental
          WHEN r.receiving_id IS NOT NULL THEN
            DATE(r.date_created)
        END as date,
-        NULLIF(p.date_estin, '0000-00-00') as expected_date,
+        CASE 
+          WHEN p.date_estin = '0000-00-00' THEN NULL
+          WHEN p.date_estin IS NULL THEN NULL
+          WHEN p.date_estin NOT REGEXP '^[0-9]{4}-[0-9]{2}-[0-9]{2}$' THEN NULL
+          ELSE p.date_estin
+        END as expected_date,
        COALESCE(p.status, 50) as status,
        p.short_note as notes,
        p.notes as long_note
@@ -359,9 +364,12 @@ async function importPurchaseOrders(prodConnection, localConnection, incremental

            function formatDate(dateStr) {
              if (!dateStr) return null;
+              if (dateStr === '0000-00-00' || dateStr === '0000-00-00 00:00:00') return null;
+              if (typeof dateStr === 'string' && !dateStr.match(/^\d{4}-\d{2}-\d{2}/)) return null;
              try {
                const date = new Date(dateStr);
                if (isNaN(date.getTime())) return null;
+                if (date.getFullYear() < 1900 || date.getFullYear() > 2100) return null;
                return date.toISOString().split('T')[0];
              } catch (e) {
                return null;
Author	SHA1	Message	Date
Matt	1003ff3cf2	More incremental import fixes	2025-02-01 11:42:51 -05:00
Matt	2d0089dc52	Incremental import order fixes	2025-02-01 11:03:42 -05:00