feat(pipeline): 渠道判定后立即归一商品并算成本

- 新增按订单项定向归一，避免新判定订单等冷通道 - 判定/指定渠道后立即归一 → 算成本 → 按业务日回填事实与总账 - 提取 refreshFactsForDates 供管道与渠道服务复用
2026-11-01 15:21:53 +08:00
parent 9e154cb16d
commit fd6d2d561b
4 changed files with 174 additions and 55 deletions
@@ -319,3 +319,69 @@ async function loadMappingIndex(): Promise<MappingIndex> {
    })),
  )
 }
+
+/** 指定订单项立即归一（渠道判定后调用，保证成本能马上算出来）。 */
+export async function normalizeProductsForItems(itemIds: number[]): Promise<{
+  matched: number
+  unmatched: number
+}> {
+  const ids = [
+    ...new Set(itemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0)),
+  ].slice(0, 2000)
+  if (ids.length === 0) return { matched: 0, unmatched: 0 }
+  const index = await loadMappingIndex()
+  const rows = await query<ItemRow>(
+    `
+      SELECT
+        oi.id,
+        o.provider,
+        o.shop_id,
+        oi.sku_code,
+        oi.sku_name,
+        oi.normalized_sku_name,
+        oi.normalized_sku_series,
+        oi.fulfillment_channel,
+        wpr.rule_key AS work_rule_key,
+        cfg.canonical_product_id AS manual_config_product_id
+      FROM order_items oi
+      JOIN orders o ON o.id = oi.order_id
+      LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
+      LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
+      LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
+      WHERE oi.fulfillment_status = 'assigned'
+        AND oi.canonical_product_id IS NULL
+        AND oi.id = ANY($1::bigint[])
+    `,
+    [ids],
+  )
+  const matched: Array<{ id: number; productId: number }> = []
+  let unmatched = 0
+  for (const item of rows.rows) {
+    const productId = resolveProductId(item, index)
+    if (!productId) {
+      unmatched += 1
+      await upsertAggregatedException({
+        reason: 'product_mapping_missing',
+        scopeKey: `product_mapping_missing|${item.provider}|${item.shop_id}|${item.sku_code}|${item.fulfillment_channel || '-'}`,
+        orderItemId: item.id,
+        detail: { skuName: item.sku_name, channel: item.fulfillment_channel },
+      })
+      continue
+    }
+    matched.push({ id: item.id, productId })
+  }
+  if (matched.length > 0) {
+    await query(
+      `
+        UPDATE order_items oi
+        SET canonical_product_id = input.product_id, updated_at = NOW()
+        FROM unnest($1::bigint[], $2::bigint[]) AS input(id, product_id)
+        WHERE oi.id = input.id
+          AND oi.canonical_product_id IS NULL
+      `,
+      [matched.map((row) => row.id), matched.map((row) => row.productId)],
+    )
+    await resolveExceptionsForItems(matched.map((row) => row.id))
+  }
+  return { matched: matched.length, unmatched }
+}
@@ -0,0 +1,71 @@
+import { query } from '../../db/client.js'
+import { rebuildChannelSalesDailyFacts } from './sales-facts-channel-service.js'
+import { syncFinanceLedgerFromFacts } from './finance-ledger-service.js'
+
+/**
+ * 按业务日回填销售事实与总账。
+ *
+ * - 日期较多（>31）时按 min~max 整段重算，避免截断导致部分日期永远不更新；
+ * - 已关账期间会被跳过，返回 skippedClosed 供调用方提示。
+ */
+export async function refreshFactsForDates(
+  dates: string[],
+): Promise<{ days: number; skippedClosed: number }> {
+  const unique = [...new Set(dates.filter(Boolean))].sort()
+  if (unique.length === 0) return { days: 0, skippedClosed: 0 }
+
+  if (unique.length > 31) {
+    const dateFrom = unique[0]!
+    const dateTo = unique[unique.length - 1]!
+    try {
+      const facts = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo })
+      await syncFinanceLedgerFromFacts({ dateFrom, dateTo })
+      if (facts.skippedClosedPeriods.length > 0) {
+        console.warn('[finance] 部分日期已关账，事实未回填', {
+          dateFrom,
+          dateTo,
+          skipped: facts.skippedClosedPeriods,
+        })
+      }
+      return { days: unique.length, skippedClosed: facts.skippedClosedPeriods.length }
+    } catch {
+      return { days: 0, skippedClosed: 0 }
+    }
+  }
+
+  let done = 0
+  let skippedClosed = 0
+  for (const date of unique) {
+    try {
+      const facts = await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date })
+      if (facts.skippedClosedPeriods.length > 0) {
+        skippedClosed += facts.skippedClosedPeriods.length
+        console.warn('[finance] 日期已关账，事实未回填', { date })
+        continue
+      }
+      await syncFinanceLedgerFromFacts({ dateFrom: date, dateTo: date })
+      done += 1
+    } catch {
+      // 单日失败不阻断调用方，下一轮或人工重跑会再处理。
+    }
+  }
+  return { days: done, skippedClosed }
+}
+
+/** 批量取订单项所属业务日（Asia/Shanghai）。 */
+export async function orderItemBasisDates(itemIds: number[]): Promise<string[]> {
+  const ids = [...new Set(itemIds.filter((id) => Number.isInteger(id) && id > 0))]
+  if (ids.length === 0) return []
+  const rows = await query<{ basis_date: string }>(
+    `
+      SELECT DISTINCT
+        (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
+          AS basis_date
+      FROM order_items oi
+      JOIN orders o ON o.id = oi.order_id
+      WHERE oi.id = ANY($1::bigint[])
+    `,
+    [ids],
+  )
+  return rows.rows.map((row) => row.basis_date)
+}
@@ -3,6 +3,9 @@ import { createHttpError } from '../../utils/http.js'
 import { nowIso } from '../../utils/time.js'
 import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
 import { readWatermarkItemId } from '../watermark-store.js'
+import { syncCostSnapshotsForItems } from '../cost-center/cost-snapshot-service.js'
+import { normalizeProductsForItems } from '../cost-center/product-normalize-service.js'
+import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js'
 import {
  FULFILLMENT_CHANNELS,
  MATCH_KEY_TYPES,
@@ -145,6 +148,8 @@ export type AssignChannelsResult = {
  byChannel: Record<string, number>
  /** 本轮被判定的订单所属业务日，供事实/总账按日回填。 */
  affectedDates: string[]
+  /** 本轮新判定的订单项（供立即计算成本，最多 2000 条）。 */
+  assignedItemIds: number[]
 }

 /** 批量判定：默认只扫 pending 且未锁定的订单项。 */
@@ -169,6 +174,7 @@ export async function assignPendingOrderItemChannels(
    nextCursor: cursor,
    byChannel: {},
    affectedDates: [],
+    assignedItemIds: [],
  }
  const affectedDates = new Set<string>()

@@ -262,6 +268,7 @@ export async function assignPendingOrderItemChannels(
        ],
      )
      result.assigned += updated.rowCount || 0
+      result.assignedItemIds = assignments.map((row) => row.id).slice(0, 2000)
    }
  }

@@ -305,7 +312,17 @@ export async function assignChannelToOrderItems(input: {
    [ids, input.channel, input.adminUserId, String(input.note || '').slice(0, 200)],
  )

-  return { updated: result.rowCount || 0 }
+  const updated = result.rowCount || 0
+  if (updated > 0) {
+    // 运营指定后先归一商品，再立即算成本并回填事实，避免旧单等冷通道。
+    await normalizeProductsForItems(ids).catch(() => undefined)
+    const cost = await syncCostSnapshotsForItems(ids).catch(() => null)
+    const dates = [
+      ...new Set([...(await orderItemBasisDates(ids)), ...(cost?.affectedDates || [])]),
+    ]
+    await refreshFactsForDates(dates).catch(() => undefined)
+  }
+  return { updated }
 }

 /** 运营解锁：允许自动判定重新接管。 */
@@ -3,10 +3,17 @@ import { query } from '../../db/client.js'
 import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
 import { resolveExceptionsForItemsByReason } from '../cost-center/cost-exception-service.js'
 import { readWatermarkItemId } from '../watermark-store.js'
+import { refreshFactsForDates } from '../finance/refresh-facts.js'
 import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js'
 import { syncFinanceLedgerFromFacts } from '../finance/finance-ledger-service.js'
-import { normalizeOrderItemProducts } from '../cost-center/product-normalize-service.js'
-import { syncChannelCostSnapshots } from '../cost-center/cost-snapshot-service.js'
+import {
+  normalizeOrderItemProducts,
+  normalizeProductsForItems,
+} from '../cost-center/product-normalize-service.js'
+import {
+  syncChannelCostSnapshots,
+  syncCostSnapshotsForItems,
+} from '../cost-center/cost-snapshot-service.js'
 import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
 import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
 import {
@@ -47,7 +54,16 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
  })

  await touchWatermark('channel_assign', { lastItemId: result.nextCursor })
-  const refreshed = await refreshFactsForDates(result.affectedDates)
+
+  // 刚判定渠道的订单项立刻算成本，不等冷通道扫到。
+  let costDates: string[] = []
+  if (result.assignedItemIds.length > 0) {
+    // 先归一商品，再算成本：卡券/自动发货没有商品就无法解析采购价。
+    await normalizeProductsForItems(result.assignedItemIds).catch(() => undefined)
+    const cost = await syncCostSnapshotsForItems(result.assignedItemIds).catch(() => null)
+    costDates = cost?.affectedDates || []
+  }
+  const refreshed = await refreshFactsForDates([...result.affectedDates, ...costDates])
  return {
    status: 'ok',
    message:
@@ -60,57 +76,6 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
  }
 }

-/**
- * 渠道/成本变更后，按业务日回填销售事实与总账。
- *
- * 事实任务只回看最近几天；补配价格、迟到分类这类变更必须按受影响日期重算，
- * 否则旧日期的毛利会一直停在缺失成本/未分类。
- */
-async function refreshFactsForDates(
-  dates: string[],
-): Promise<{ days: number; skippedClosed: number }> {
-  const unique = [...new Set(dates.filter(Boolean))].sort()
-  if (unique.length === 0) return { days: 0, skippedClosed: 0 }
-
-  // 日期过多时按整段重算，避免截断导致部分日期永远不更新。
-  if (unique.length > 31) {
-    const dateFrom = unique[0]!
-    const dateTo = unique[unique.length - 1]!
-    try {
-      const facts = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo })
-      await syncFinanceLedgerFromFacts({ dateFrom, dateTo })
-      if (facts.skippedClosedPeriods.length > 0) {
-        console.warn('[finance-pipeline] 部分日期已关账，事实未回填', {
-          dateFrom,
-          dateTo,
-          skipped: facts.skippedClosedPeriods,
-        })
-      }
-      return { days: unique.length, skippedClosed: facts.skippedClosedPeriods.length }
-    } catch {
-      return { days: 0, skippedClosed: 0 }
-    }
-  }
-
-  let done = 0
-  let skippedClosed = 0
-  for (const date of unique) {
-    try {
-      const facts = await rebuildChannelSalesDailyFacts({ dateFrom: date, dateTo: date })
-      if (facts.skippedClosedPeriods.length > 0) {
-        skippedClosed += facts.skippedClosedPeriods.length
-        console.warn('[finance-pipeline] 日期已关账，事实未回填', { date })
-        continue
-      }
-      await syncFinanceLedgerFromFacts({ dateFrom: date, dateTo: date })
-      done += 1
-    } catch {
-      // 单日失败不影响管道本身，下一轮或人工重跑会再处理。
-    }
-  }
-  return { days: done, skippedClosed }
-}
-
 export async function runCarrierDispatchJob(job: JsonObject = {}) {
  const config = jobConfig(job)
  const limit = clampInt(config.limit, 200, 1, 2000)