优化

scripts/real-order-xlsx-test-env.mjs

@@ -2,7 +2,7 @@
 // Explicitly authorized non-production HTTP import. Never resets a database.
 import { createHash } from 'node:crypto'
 import { execFileSync } from 'node:child_process'
-import { mkdir, readFile, writeFile, rename, open } from 'node:fs/promises'
+import { mkdir, readFile, writeFile, rename, open, unlink } from 'node:fs/promises'
 import { resolve } from 'node:path'
 import { pathToFileURL } from 'node:url'
 import { AcceptanceRun, parseArguments, validateExecutionOptions } from './admin-test-env-acceptance.mjs'
@@ -45,7 +45,7 @@ export function normalizeRows(rows) {
     for (const name of ['实际件数', '实际毛重', '实际立方']) {
       if (entries.some(({ row }) => !positive(get(row, name)))) problems.push(`MISSING_${name}`)
     }
-    const sum = (name) => round(entries.reduce((total, { row }) => total + (Number(get(row, name)) || 0), 0))
+    const sum = (name, digits = 6) => round(entries.reduce((total, { row }) => total + (Number(get(row, name)) || 0), 0), digits)
     const packageCount = sum('实际件数')
     if (!Number.isInteger(packageCount) || packageCount < 1) problems.push('INVALID_PACKAGE_COUNT')
     if (entries.length > 1) {
@@ -68,7 +68,10 @@ export function normalizeRows(rows) {
       sourceCustomer: text(get(first, '客户')), sourceChannel, channelName: channelName(sourceChannel),
       destination, destinationType, pricingTaxMode, taxMode, chargeMode,
       fulfillmentType: destination.includes('一件代发') ? 'ONE_CLICK_DROPSHIP' : 'STANDARD',
-      packageCount, weightKg: sum('实际毛重'), volumeCbm: sum('实际立方'),
+      // Match the persisted order schema: kg has 3 decimals and CBM has 4.
+      // Split source rows can contain longer allocation decimals whose exact sum
+      // cannot be represented by the order table.
+      packageCount, weightKg: sum('实际毛重', 3), volumeCbm: sum('实际立方', 4),
       cargoName: distinct('品名').filter(Boolean).join(' / '),
       sku: distinct('SKU').filter(Boolean).join('\n'), shipmentId: distinct('FBA货件编号').filter(Boolean).join('\n'),
       referenceId: distinct('跟踪编号').filter(Boolean).join('\n'), problems }
@@ -199,7 +202,7 @@ export function requireSuccess(response) {
 
 async function main() {
   const args = new Map()
-  const allowed = new Set(['input', 'sheet', 'python', 'output-dir', 'batch-id', 'stage', 'limit', 'customer-name', 'origin-warehouse-name', 'allow-test-placeholders', 'expected-host'])
+  const allowed = new Set(['input', 'sheet', 'python', 'output-dir', 'batch-id', 'stage', 'limit', 'concurrency', 'customer-name', 'origin-warehouse-name', 'allow-test-placeholders', 'expected-host'])
   for (let i = 2; i < process.argv.length; i += 2) {
     const name = process.argv[i].replace(/^--/, ''), value = process.argv[i + 1]
     if (!allowed.has(name) || !value || value.startsWith('--')) throw Error(`无效参数：${name}`)
@@ -207,10 +210,14 @@ async function main() {
   }
   for (const key of ['input', 'python', 'output-dir', 'batch-id', 'expected-host']) if (!args.get(key)) throw Error(`缺少 --${key}`)
   const stage = args.get('stage') || 'preflight'
-  if (!['preflight', 'create', 'wait-split'].includes(stage)) throw Error('stage 必须为 preflight/create/wait-split')
-  if (stage === 'wait-split' && args.get('allow-test-placeholders') !== 'true') throw Error('此源表缺少必填字段，模拟测量须显式允许测试占位')
+  if (!['preflight', 'create', 'pending-review', 'wait-split'].includes(stage)) throw Error('stage 必须为 preflight/create/pending-review/wait-split')
+  if (['pending-review', 'wait-split'].includes(stage) && args.get('allow-test-placeholders') !== 'true') {
+    throw Error('此源表缺少提交必填字段，须显式允许测试占位')
+  }
   const limit = Number(args.get('limit') || 1)
   if (!Number.isInteger(limit) || limit < 1) throw Error('limit 必须为正整数')
+  const concurrency = Number(args.get('concurrency') || 1)
+  if (!Number.isInteger(concurrency) || concurrency < 1 || concurrency > 10) throw Error('concurrency 必须是1到10的整数')
   const options = validateExecutionOptions(parseArguments(['--modules', 'orders,inbound-measurement', '--quiet', 'true']))
   if (new URL(options.gateway).hostname !== args.get('expected-host') || !/^test[-.]/.test(args.get('expected-host'))) throw Error('只允许明确匹配的测试域名')
   const input = resolve(args.get('input'))
@@ -229,16 +236,22 @@ async function main() {
   try { state = JSON.parse(await readFile(statePath, 'utf8')) } catch (error) { if (error.code !== 'ENOENT') throw error }
   if (state && JSON.stringify(state.identity) !== JSON.stringify(identity)) throw Error('批次/源文件/目标发生变化，请使用独立输出目录')
   state ||= { identity, orders: {}, startedAt: new Date().toISOString() }
-  const lock = await open(resolve(directory, 'run.lock'), 'wx')
-  const save = async () => {
-    await writeFile(`${statePath}.tmp`, JSON.stringify(state, null, 2), 'utf8')
+  const lockPath = resolve(directory, 'run.lock')
+  const lock = await acquireRunLock(lockPath)
+  let saveQueue = Promise.resolve()
+  const save = () => {
+    saveQueue = saveQueue.then(async () => {
+      const temporaryPath = `${statePath}.${process.pid}.tmp`
+      await writeFile(temporaryPath, JSON.stringify(state, null, 2), 'utf8')
     for (let attempt = 0; ; attempt++) {
-      try { await rename(`${statePath}.tmp`, statePath); break }
+        try { await rename(temporaryPath, statePath); break }
       catch (error) {
         if (!['EPERM', 'EBUSY', 'EACCES'].includes(error.code) || attempt >= 5) throw error
         await new Promise((resolve) => setTimeout(resolve, 100 * (attempt + 1)))
       }
     }
+    })
+    return saveQueue
   }
   const ctx = new AcceptanceRun(options)
   const api = async (method, path, body, token = ctx.token, headers = {}) => requireSuccess(await ctx.request(method, path, body, token, headers))
@@ -264,14 +277,35 @@ async function main() {
     const channels = await api('GET', `/api/merchant/v1/first-leg/pricing/channels?originWarehouseId=${receiving.id}`, undefined, merchant)
     const catalog = { receiving, channels, destinations: [...await list('FBA'), ...await list('OVERSEAS')] }
     const pallets = []
-    for (const code of PALLETS) {
-      const page = await api('GET', `/api/platform/v1/pallets?page=1&size=20&tenantId=0&warehouseId=${receiving.id}&palletCode=${encodeURIComponent(code)}`)
-      const found = page.records?.filter((item) => item.palletCode === code && String(item.warehouseId) === String(receiving.id))
-      if (found?.length !== 1) throw Error(`REQUIRED_PALLET_MISSING_${code}`)
-      pallets.push(found[0])
+    if (stage === 'wait-split') {
+      for (const code of PALLETS) {
+        const page = await api('GET', `/api/platform/v1/pallets?page=1&size=20&tenantId=0&warehouseId=${receiving.id}&palletCode=${encodeURIComponent(code)}`)
+        const found = page.records?.filter((item) => item.palletCode === code && String(item.warehouseId) === String(receiving.id))
+        if (found?.length !== 1) throw Error(`REQUIRED_PALLET_MISSING_${code}`)
+        pallets.push(found[0])
+      }
     }
     const policy = await api('GET', '/api/merchant/v1/orders/first-leg/inbound-code-policy', undefined, merchant)
     if (policy.mode !== 'AUTO') throw Error('INBOUND_CODE_POLICY_NOT_AUTO')
+    let sharedResources = null
+    if (['pending-review', 'wait-split'].includes(stage)) {
+      let dimensionRules = null
+      if (stage === 'wait-split') {
+        const dimensionConfig = await api('GET', `/api/platform/v1/first-leg/customer-dimension-config?tenantId=${profile.id}`)
+        totalPreservingShapes(normalized.orders.find((order) => order.problems.length === 0), dimensionConfig.rules)
+        dimensionRules = dimensionConfig.rules
+      }
+      let productImageFileId = state.sharedResources?.productImageFileId
+      if (!productImageFileId) {
+        const image = requireSuccess(await ctx.requestMultipart('POST', '/api/merchant/v1/files/product-images',
+          { content: PNG, type: 'image/png', name: `TEST-XLSX-${state.identity.batchId}.png` }, merchant))
+        productImageFileId = image?.fileId
+        if (!productImageFileId) throw Error('SHARED_PRODUCT_IMAGE_ID_MISSING')
+      }
+      state.sharedResources = { productImageFileId, ...(dimensionRules ? { dimensionRules } : {}) }
+      sharedResources = state.sharedResources
+      await save()
+    }
     const candidates = [], rejected = [...normalized.excluded]
     for (const order of normalized.orders) {
       try { candidates.push({ order, route: matchRoute(order, catalog) }) }
@@ -284,47 +318,88 @@ async function main() {
     console.log(JSON.stringify({ stage, sourceRows: preflight.sourceRows, orders: preflight.orders, matched: candidates.length, rejected: rejected.length }))
     await save()
     if (stage === 'preflight') return
-    await executeCandidates({ candidates, limit, state, save, api, ctx, merchant, profile, pallets, stage })
+    await executeCandidates({ candidates, limit, concurrency, state, save, api, ctx, merchant, profile, pallets,
+      sharedResources, stage })
   } finally {
+    await saveQueue.catch(() => {})
     await lock.close()
-    const { unlink } = await import('node:fs/promises')
-    await unlink(resolve(directory, 'run.lock'))
+    await unlink(lockPath)
   }
 }
 
-export async function executeCandidates({ candidates, limit, state, save, api, ctx, merchant, profile, pallets, stage }) {
-  let attempted = 0, succeeded = 0, failed = 0, skipped = 0
+async function acquireRunLock(lockPath) {
+  try {
+    const handle = await open(lockPath, 'wx')
+    await handle.writeFile(String(process.pid), 'utf8')
+    return handle
+  } catch (error) {
+    if (error.code !== 'EEXIST') throw error
+    const existingPid = Number(await readFile(lockPath, 'utf8').catch(() => ''))
+    let alive = false
+    if (Number.isInteger(existingPid) && existingPid > 0) {
+      try { process.kill(existingPid, 0); alive = true } catch { /* Stale PID. */ }
+    }
+    if (alive) throw Error(`批次正在运行，pid=${existingPid}`)
+    await unlink(lockPath)
+    const handle = await open(lockPath, 'wx')
+    await handle.writeFile(String(process.pid), 'utf8')
+    return handle
+  }
+}
+
+export async function executeCandidates({ candidates, limit, concurrency = 1, state, save, api, ctx, merchant, profile, pallets,
+  sharedResources, stage }) {
+  let attempted = 0, completed = 0, succeeded = 0, failed = 0, skipped = 0, stopReason = null
+  const pending = []
   for (const { order, route } of candidates) {
     // An interrupted write is not permission to create another order. Keep its last
     // checkpoint for inspection. A reset database is always a NEW batch/directory.
-    if (state.orders[order.key]) { skipped++; continue }
-    if (attempted >= limit) break
+    const previous = state.orders[order.key]
+    const retryablePricingWarning = stage === 'pending-review' && previous?.status === 'FAILED'
+      && previous.stage === 'estimate' && !previous.orderId && ['A0404', 'A0422'].includes(previous.evidence?.code)
+    if (previous && !retryablePricingWarning) { skipped++; continue }
+    if (pending.length >= limit) break
+    pending.push({ order, route, previous })
+  }
+  for (let offset = 0; offset < pending.length; offset += concurrency) {
+    const batch = pending.slice(offset, offset + concurrency)
+    await Promise.all(batch.map(async ({ order, route, previous }) => {
     const marker = `XLSX-${hash(`${state.identity.batchId}:${order.key}`).slice(0, 24)}`
-    const entry = { rows: order.rows, marker, status: 'STARTED', stage: 'estimate', startedAt: new Date().toISOString() }
+    const entry = { rows: order.rows, marker, status: 'STARTED', stage: 'estimate', startedAt: new Date().toISOString(),
+      ...(previous ? { priorAttempts: [...(previous.priorAttempts || []), { status: previous.status, stage: previous.stage,
+        error: previous.error, evidence: previous.evidence, finishedAt: previous.finishedAt }] } : {}) }
     state.orders[order.key] = entry
     await save()
     attempted++
     try {
-      const estimate = await api('POST', '/api/merchant/v1/first-leg/pricing/estimate', {
-        originWarehouseId: route.receiving.id, channelCode: route.channel.channelCode,
-        destinationCode: route.destination.warehouseCode, destinationType: order.destinationType,
-        destinationWarehouseId: route.destination.id, destinationName: route.destination.warehouseName,
-        pricingTaxMode: order.pricingTaxMode, chargeMode: order.chargeMode,
-        actualWeightKg: order.weightKg, actualVolumeCbm: order.volumeCbm,
-      }, merchant)
-      if (estimate?.channelCode !== route.channel.channelCode || estimate?.pricingTaxMode !== order.pricingTaxMode) throw Error('ESTIMATE_ROUTE_MISMATCH')
+      let estimate = null
+      try {
+        estimate = await api('POST', '/api/merchant/v1/first-leg/pricing/estimate', {
+          originWarehouseId: route.receiving.id, channelCode: route.channel.channelCode,
+          destinationCode: route.destination.warehouseCode, destinationType: order.destinationType,
+          destinationWarehouseId: route.destination.id, destinationName: route.destination.warehouseName,
+          pricingTaxMode: order.pricingTaxMode, chargeMode: order.chargeMode,
+          actualWeightKg: order.weightKg, actualVolumeCbm: order.volumeCbm,
+        }, merchant)
+      } catch (error) {
+        if (stage !== 'pending-review' || !['A0404', 'A0422'].includes(error.evidence?.code)) throw error
+        entry.pricingWarning = { code: error.evidence.code, message: error.evidence.message,
+          requestId: error.evidence.requestId, traceId: error.evidence.traceId }
+      }
+      if (estimate && (estimate.channelCode !== route.channel.channelCode || estimate.pricingTaxMode !== order.pricingTaxMode)) {
+        throw Error('ESTIMATE_ROUTE_MISMATCH')
+      }
       let payload
       if (stage === 'create') {
         payload = buildSourceDraft(order, route, profile.tenantName, marker)
       } else {
-        const config = await api('GET', `/api/platform/v1/first-leg/customer-dimension-config?tenantId=${profile.id}`)
-        entry.dimensionPlan = totalPreservingShapes(order, config.rules)
-        entry.stage = 'product-image'; await save()
-        const image = requireSuccess(await ctx.requestMultipart('POST', '/api/merchant/v1/files/product-images',
-          { content: PNG, type: 'image/png', name: `${marker}-TEST.png` }, merchant))
-        if (!image?.fileId) throw Error('PRODUCT_IMAGE_ID_MISSING')
-        entry.productImageFileId = image.fileId
-        payload = buildPayload(order, route, profile.tenantName, image.fileId, marker)
+        if (!sharedResources?.productImageFileId) throw Error('SHARED_PRODUCT_IMAGE_MISSING')
+        if (stage === 'wait-split') {
+          if (!Array.isArray(sharedResources.dimensionRules)) throw Error('SHARED_DIMENSION_RULES_MISSING')
+          entry.dimensionPlan = totalPreservingShapes(order, sharedResources.dimensionRules)
+        }
+        entry.productImageFileId = sharedResources.productImageFileId
+        payload = buildPayload(order, route, profile.tenantName, sharedResources.productImageFileId, marker)
       }
       entry.stage = 'create'; entry.payloadHash = hash(JSON.stringify(payload)); await save()
       entry.orderId = String(await api('POST', '/api/merchant/v1/orders/first-leg', payload, merchant,
@@ -340,9 +415,25 @@ export async function executeCandidates({ candidates, limit, state, save, api, c
           || Math.abs(Number(cargo.volumeCbm) - order.volumeCbm) > 0.00001) throw Error('DRAFT_SOURCE_FACTS_MISMATCH')
       entry.orderNo = draft.order.orderNo
       entry.declarationVerified = true
-      if (stage === 'wait-split') {
+      if (['pending-review', 'wait-split'].includes(stage)) {
         entry.stage = 'submit'; await save()
         await api('POST', `${clientPath}/submit`, undefined, merchant, { 'Idempotency-Key': `${marker}-submit` })
+      }
+      if (stage === 'pending-review') {
+        const submitted = await api('GET', clientPath, undefined, merchant)
+        const unified = await api('GET', `/api/merchant/v1/orders/${entry.orderId}`, undefined, merchant)
+        if (submitted.order?.status !== 'SUBMITTED' || unified.summary?.mainStatus !== 'PENDING_REVIEW') {
+          const error = Error('PENDING_REVIEW_NOT_CONFIRMED')
+          error.evidence = { clientStatus: submitted.order?.status, mainStatus: unified.summary?.mainStatus }
+          throw error
+        }
+        entry.stage = 'PENDING_REVIEW'
+        entry.verification = { clientStatus: submitted.order.status, mainStatus: unified.summary.mainStatus,
+          expectedPackages: order.packageCount, actualPackages: Number(submitted.cargoes?.[0]?.packageCount),
+          expectedCargoWeightKg: order.weightKg, actualCargoWeightKg: Number(submitted.cargoes?.[0]?.grossWeightKg),
+          expectedVolumeCbm: order.volumeCbm, actualVolumeCbm: Number(submitted.cargoes?.[0]?.volumeCbm) }
+      }
+      if (stage === 'wait-split') {
         entry.stage = 'approve'; await save()
         const platformPath = `/api/platform/v1/orders/first-leg/${entry.orderId}`
         const q = `tenantId=${profile.id}`
@@ -358,14 +449,22 @@ export async function executeCandidates({ candidates, limit, state, save, api, c
       entry.status = 'SUCCEEDED'; succeeded++
     } catch (error) {
       entry.status = 'FAILED'; entry.error = error.message; entry.evidence = error.evidence; failed++
+      if (error.evidence?.code === 'A0401') stopReason ||= 'AUTHENTICATION_EXPIRED'
     }
+    completed++
     entry.finishedAt = new Date().toISOString(); await save()
-    console.log(JSON.stringify({ attempted, succeeded, failed, skipped, rows: entry.rows,
-      stage: entry.stage, status: entry.status, orderNo: entry.orderNo, inboundCode: entry.inboundCode, error: entry.error, evidence: entry.evidence }))
-    // A small pilot must expose a common deployment/data defect before multiplying it.
+    if (entry.status === 'FAILED' || completed % 100 === 0 || completed === pending.length) {
+      console.log(JSON.stringify({ attempted, completed, succeeded, failed, skipped, rows: entry.rows,
+        stage: entry.stage, status: entry.status, orderNo: entry.orderNo, inboundCode: entry.inboundCode,
+        error: entry.error, evidence: entry.evidence }))
+    }
+    }))
+    if (stopReason) break
+    // A small concurrent pilot must expose a common deployment/data defect before multiplying it.
     if (failed >= 3 && succeeded === 0) break
   }
-  state.lastRun = { attempted, succeeded, failed, skipped, stage, finishedAt: new Date().toISOString() }
+  state.lastRun = { attempted, completed, succeeded, failed, skipped, concurrency, stage, stopReason,
+    finishedAt: new Date().toISOString() }
   await save()
   if (failed) process.exitCode = 1
 }
@@ -383,6 +482,34 @@ async function poll(read, predicate) {
   throw error
 }
 
+export async function saveMeasurementDraft({ api, measurementPath, base, q, order, entry, marker,
+  attempts = 3, wait = (milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds)), onRetry }) {
+  for (let attempt = 1; attempt <= attempts; attempt++) {
+    const measured = await api('GET', measurementPath)
+    if (measured.items?.length !== entry.dimensionPlan.length) throw Error('UNEXPECTED_MEASUREMENT_ITEMS')
+    const payload = {
+      measurementId: measured.measurementId, version: Number(measured.version),
+      actualGoodsName: oneValue(order.cargoName, '测试导入汇总货物', 128),
+      items: entry.dimensionPlan.map((group) => {
+        const item = measured.items.find((value) => String(value.id) === group.itemId)
+        if (!item) throw Error('MEASUREMENT_GROUP_MISSING')
+        return { inboundDetailId: item.inboundDetailId, itemNo: item.itemNo, boxSpecification: item.boxSpecification,
+          packageCount: group.count, lengthCm: group.length, widthCm: group.width, heightCm: group.height,
+          intelligentImageFileId: item.intelligentImageFileId, overLimitWarning: item.overLimitWarning ?? null,
+          outerLengthCm: group.length, outerWidthCm: group.width, outerHeightCm: group.height,
+          coefficientMode: 'ADD', remark: `${marker} TEST按总体积分组反算箱规，非实测` }
+      }),
+    }
+    try { return await api('PUT', `${base}/measurement/draft?${q}`, payload) }
+    catch (error) {
+      if (error.evidence?.code !== 'S0500' || attempt === attempts) throw error
+      await onRetry?.({ attempt, code: error.evidence.code, requestId: error.evidence.requestId, traceId: error.evidence.traceId })
+      await wait(250 * attempt)
+    }
+  }
+  throw Error('MEASUREMENT_DRAFT_RETRY_EXHAUSTED')
+}
+
 async function measureOrder({ order, entry, ctx, api, pallets, tenantId, save }) {
   const inboundId = entry.inboundId, q = `tenantId=${tenantId}`, marker = entry.marker
   const inboundPath = `/api/platform/inbounds/${inboundId}?${q}`
@@ -411,21 +538,8 @@ async function measureOrder({ order, entry, ctx, api, pallets, tenantId, save })
   group.itemId = String(scan.itemId)
   cartonIndex += group.count
   }
-  const measured = await api('GET', measurementPath)
-  if (measured.items?.length !== entry.dimensionPlan.length) throw Error('UNEXPECTED_MEASUREMENT_ITEMS')
-  const draft = await api('PUT', `${base}/measurement/draft?${q}`, {
-    measurementId: measured.measurementId, version: Number(measured.version),
-    actualGoodsName: oneValue(order.cargoName, '测试导入汇总货物', 128),
-    items: entry.dimensionPlan.map((group) => {
-      const item = measured.items.find((value) => String(value.id) === group.itemId)
-      if (!item) throw Error('MEASUREMENT_GROUP_MISSING')
-      return { inboundDetailId: item.inboundDetailId, itemNo: item.itemNo, boxSpecification: item.boxSpecification,
-      packageCount: group.count, lengthCm: group.length, widthCm: group.width, heightCm: group.height,
-      intelligentImageFileId: item.intelligentImageFileId, overLimitWarning: item.overLimitWarning ?? null,
-      outerLengthCm: group.length, outerWidthCm: group.width, outerHeightCm: group.height,
-      coefficientMode: 'ADD',
-      remark: `${marker} TEST按总体积分组反算箱规，非实测` } }),
-  })
+  const draft = await saveMeasurementDraft({ api, measurementPath, base, q, order, entry, marker,
+    onRetry: async (retry) => { entry.retries ||= []; entry.retries.push({ stage: 'measurement-draft', ...retry }); await save() } })
   const draftCheck = await api('GET', measurementPath)
   if (Math.abs(Number(draftCheck.actualVolumeCbm) - order.volumeCbm) > 0.0000001
       || Number(draftCheck.actualPackageCount) !== order.packageCount) throw Error('DRAFT_VOLUME_OR_COUNT_MISMATCH')