feat(finance): 统一内部 SKU 驱动的履约与财务分流

- 新增 internal_sku_mappings / internal_sku_channels，平台字段统一映射到内部中文 SKU - 下单入口按内部 SKU 渠道决策修正默认人工路由，自动发货补全 dash sku - 财务 channel_assign 读取统一渠道表；卡券/人工能力兜底，自动发货/接单要求真实载体 - carrier reconciliation 自动补自动发货任务和接单工单 - pending 落明确原因：无内部 SKU / 无渠道能力 / 无履约载体 - 合并、重判、旧 sku_code 规则清理、部署后 preflight 核对脚本 - 待分类队列按内部 SKU 聚合并展示原因
2026-11-02 14:17:08 +08:00
parent 531a23d78b
commit 5fe33b0b46
18 changed files with 1020 additions and 15 deletions
@@ -0,0 +1,133 @@
+/**
+ * 把旧的 sku_code 渠道规则转换为 internal_sku 全局规则，并停用旧规则。
+ *
+ * 用法：
+ *   npx tsx scripts/cleanup-legacy-sku-code-rules.ts
+ *   npx tsx scripts/cleanup-legacy-sku-code-rules.ts --apply
+ */
+import { closeDb, query } from '../src/db/client.js'
+
+type LegacyRule = {
+  id: number
+  provider: string
+  shop_id: string
+  match_key: string
+  channel: string
+  priority: number
+  source: string
+}
+
+async function resolveInternalSku(rule: LegacyRule): Promise<string> {
+  if (rule.channel === 'affiliate_dash') {
+    const product = await query<{ canonical_name: string }>(
+      `SELECT canonical_name FROM product_catalog
+       WHERE canonical_sku = $1 AND status = 'active' AND btrim(canonical_name) <> ''
+       LIMIT 1`,
+      [`aff:${rule.match_key}`],
+    )
+    if (product.rows[0]?.canonical_name) return String(product.rows[0].canonical_name).trim()
+  }
+
+  const params: unknown[] = [rule.match_key]
+  let scope = ''
+  if (rule.provider) {
+    params.push(rule.provider)
+    scope += ` AND o.provider = $${params.length}`
+  }
+  if (rule.shop_id) {
+    params.push(rule.shop_id)
+    scope += ` AND o.shop_id = $${params.length}`
+  }
+  const items = await query<{ internal_sku: string }>(
+    `
+      SELECT DISTINCT btrim(oi.manual_sku_key) AS internal_sku
+      FROM order_items oi
+      JOIN orders o ON o.id = oi.order_id
+      WHERE oi.sku_code = $1
+        AND btrim(COALESCE(oi.manual_sku_key, '')) ~ '[一-龥]'
+        ${scope}
+      LIMIT 2
+    `,
+    params,
+  )
+  const values = items.rows.map((row) => String(row.internal_sku || '').trim()).filter(Boolean)
+  return values.length === 1 ? values[0]! : ''
+}
+
+async function main() {
+  const apply = process.argv.includes('--apply')
+  const rules = await query<LegacyRule>(
+    `
+      SELECT id, provider, shop_id, match_key, channel, priority, source
+      FROM sku_channel_rules
+      WHERE match_key_type = 'sku_code' AND enabled = TRUE
+      ORDER BY id
+    `,
+  )
+
+  const converted: Array<{ id: number; matchKey: string; internalSku: string; channel: string }> =
+    []
+  const unresolved: LegacyRule[] = []
+  for (const rule of rules.rows) {
+    const internalSku = await resolveInternalSku(rule)
+    if (!internalSku) {
+      unresolved.push(rule)
+      continue
+    }
+    converted.push({ id: rule.id, matchKey: rule.match_key, internalSku, channel: rule.channel })
+    if (!apply) continue
+
+    await query(
+      `
+        INSERT INTO sku_channel_rules (
+          provider, shop_id, match_key_type, match_key, channel, priority,
+          source, enabled, note, created_at, updated_at
+        ) VALUES ('', '', 'internal_sku', $1, $2, $3, $4, TRUE,
+                  '由旧 sku_code 规则迁移', NOW(), NOW())
+        ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING
+      `,
+      [internalSku, rule.channel, rule.priority, rule.source || 'operator'],
+    )
+    await query(
+      `
+        INSERT INTO internal_sku_channels (
+          internal_sku, channel, priority, is_fallback, enabled, source
+        ) VALUES ($1, $2, $3, FALSE, TRUE, 'legacy-rule')
+        ON CONFLICT (internal_sku, channel) DO UPDATE SET
+          priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
+          enabled = TRUE,
+          updated_at = NOW()
+      `,
+      [internalSku, rule.channel, rule.priority],
+    )
+    await query(
+      `UPDATE sku_channel_rules
+       SET enabled = FALSE,
+           note = CONCAT(COALESCE(note, ''), ' [已迁移 internal_sku:', $2::text, ']'),
+           updated_at = NOW()
+       WHERE id = $1`,
+      [rule.id, internalSku],
+    )
+  }
+
+  console.log(
+    JSON.stringify(
+      {
+        apply,
+        legacyRules: rules.rows.length,
+        convertible: converted.length,
+        converted,
+        unresolved,
+      },
+      null,
+      2,
+    ),
+  )
+}
+
+main()
+  .catch((error) => {
+    console.error(error)
+    process.exitCode = 1
+  })
+  .finally(() => closeDb())
@@ -147,6 +147,22 @@ async function mergeGroup(
    [group.keeper_id],
  )

+  await query(
+    `
+      INSERT INTO internal_sku_channels (
+        internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source
+      )
+      VALUES ($1, 'affiliate_dash', 10, FALSE, TRUE, $2, 'merged'),
+             ($1, 'manual', 20, TRUE, TRUE, $2, 'merged')
+      ON CONFLICT (internal_sku, channel) DO UPDATE SET
+        priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
+        is_fallback = internal_sku_channels.is_fallback OR EXCLUDED.is_fallback,
+        canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
+        updated_at = NOW()
+    `,
+    [group.canonical_name, group.keeper_id],
+  )
+
  await query(
    `
      INSERT INTO sku_channel_rules (
@@ -0,0 +1,99 @@
+/**
+ * 履约/财务统一分流部署后核对（只读）。
+ *
+ * 用法：
+ *   npx tsx scripts/preflight-fulfillment-routing.ts
+ *   npx tsx scripts/preflight-fulfillment-routing.ts --strict
+ */
+import { closeDb, query } from '../src/db/client.js'
+
+type CountRow = { label: string; total: number }
+
+async function main() {
+  const strict = process.argv.includes('--strict')
+  const checks: Record<string, unknown> = {}
+
+  checks.migrations = (
+    await query<{ name: string }>(`SELECT name FROM schema_migrations ORDER BY id DESC LIMIT 5`)
+  ).rows.map((row) => row.name)
+
+  checks.pendingReasons = (
+    await query<CountRow>(
+      `SELECT COALESCE(fulfillment_pending_reason, '(null)') AS label, count(*)::int AS total
+       FROM order_items WHERE fulfillment_status = 'pending'
+       GROUP BY 1 ORDER BY 2 DESC`,
+    )
+  ).rows
+
+  checks.renameCardChannels = (
+    await query<CountRow>(
+      `SELECT COALESCE(fulfillment_channel, '(pending)') AS label, count(*)::int AS total
+       FROM order_items
+       WHERE sku_name = '11月改名卡'
+       GROUP BY 1 ORDER BY 2 DESC`,
+    )
+  ).rows
+
+  checks.rules = (
+    await query<CountRow>(
+      `SELECT match_key_type || '/' || channel || '/' ||
+              CASE WHEN enabled THEN 'on' ELSE 'off' END AS label, count(*)::int AS total
+       FROM sku_channel_rules GROUP BY 1 ORDER BY 2 DESC`,
+    )
+  ).rows
+
+  checks.internalChannels = (
+    await query<CountRow>(
+      `SELECT channel AS label, count(*)::int AS total
+       FROM internal_sku_channels WHERE enabled GROUP BY 1 ORDER BY 2 DESC`,
+    )
+  ).rows
+
+  checks.pendingWithoutChannel = (
+    await query<CountRow>(
+      `SELECT 'pending_no_internal_channel' AS label, count(*)::int AS total
+       FROM order_items oi
+       WHERE oi.fulfillment_status = 'pending'
+         AND btrim(COALESCE(oi.manual_sku_key, '')) <> ''
+         AND NOT EXISTS (
+           SELECT 1 FROM internal_sku_channels c
+           WHERE c.enabled AND btrim(c.internal_sku) = btrim(oi.manual_sku_key)
+         )`,
+    )
+  ).rows[0]
+
+  checks.channelCarrier = (
+    await query<{
+      channel: string
+      total: number
+      work_orders: number
+      affiliate_tasks: number
+      manual_tasks: number
+    }>(
+      `SELECT COALESCE(fulfillment_channel, '(pending)') AS channel,
+              count(*)::int AS total,
+              count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id))::int AS work_orders,
+              count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'))::int AS affiliate_tasks,
+              count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'))::int AS manual_tasks
+       FROM order_items oi
+       WHERE oi.fulfillment_status = 'assigned'
+       GROUP BY 1 ORDER BY 2 DESC`,
+    )
+  ).rows
+
+  const failed: string[] = []
+  const rename = checks.renameCardChannels as CountRow[]
+  if (rename.some((row) => row.label === 'pending')) failed.push('11月改名卡仍有 pending')
+  if ((checks.pendingWithoutChannel as CountRow).total > 0)
+    failed.push('存在有内部 SKU 但无渠道能力的 pending')
+
+  console.log(JSON.stringify({ strict, failed, ...checks }, null, 2))
+  if (strict && failed.length > 0) process.exitCode = 2
+}
+
+main()
+  .catch((error) => {
+    console.error(error)
+    process.exitCode = 1
+  })
+  .finally(() => closeDb())
@@ -33,11 +33,20 @@ async function loadCandidates(limit: number): Promise<Candidate[]> {
      JOIN orders o ON o.id = oi.order_id
      JOIN LATERAL (
        SELECT c.channel
-        FROM product_channel_capabilities c
-        WHERE c.product_id = oi.canonical_product_id
-          AND c.enabled
+        FROM internal_sku_channels c
+        WHERE c.enabled
          AND c.channel IN ('affiliate_dash', 'manual', 'third_party_voucher')
-        ORDER BY c.priority ASC
+          AND btrim(c.internal_sku) = COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), btrim(oi.sku_name))
+        ORDER BY
+          c.priority ASC,
+          CASE c.channel
+            WHEN 'affiliate_dash' THEN 1
+            WHEN 'work_order' THEN 2
+            WHEN 'third_party_voucher' THEN 3
+            WHEN 'manual' THEN 4
+            ELSE 9
+          END,
+          c.channel ASC
        LIMIT 1
      ) cap ON TRUE
      WHERE oi.fulfillment_status = 'assigned'
@@ -0,0 +1,67 @@
+-- 118_internal_sku_channels.sql
+--
+-- 内部中文 SKU 的渠道决策单一真源。
+-- 从商品渠道能力 + 现有 internal_sku 全局规则回填；后续 intake/finance 都读这张表。
+
+CREATE TABLE IF NOT EXISTS internal_sku_channels (
+  internal_sku TEXT NOT NULL,
+  channel TEXT NOT NULL CHECK (channel IN ('work_order', 'affiliate_dash', 'third_party_voucher', 'manual')),
+  priority INTEGER NOT NULL DEFAULT 100,
+  is_fallback BOOLEAN NOT NULL DEFAULT FALSE,
+  enabled BOOLEAN NOT NULL DEFAULT TRUE,
+  canonical_product_id BIGINT REFERENCES product_catalog(id) ON DELETE SET NULL,
+  config_json JSONB NOT NULL DEFAULT '{}'::jsonb,
+  source TEXT NOT NULL DEFAULT 'backfill',
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (internal_sku, channel)
+);
+
+CREATE INDEX IF NOT EXISTS idx_internal_sku_channels_priority
+  ON internal_sku_channels (internal_sku, priority, channel)
+  WHERE enabled = TRUE;
+
+COMMENT ON TABLE internal_sku_channels IS
+  '内部中文 SKU 渠道能力：priority 越小越优先，is_fallback 表示兜底渠道';
+
+WITH capability_source AS (
+  SELECT
+    btrim(p.canonical_name) AS internal_sku,
+    c.channel,
+    MIN(c.priority) AS priority,
+    BOOL_AND(c.is_fallback) AS is_fallback,
+    MIN(p.id) AS canonical_product_id
+  FROM product_catalog p
+  JOIN product_channel_capabilities c ON c.product_id = p.id AND c.enabled = TRUE
+  WHERE p.status = 'active' AND btrim(p.canonical_name) <> ''
+  GROUP BY 1, 2
+)
+INSERT INTO internal_sku_channels (
+  internal_sku, channel, priority, is_fallback, canonical_product_id, source
+)
+SELECT internal_sku, channel, priority, is_fallback, canonical_product_id, 'capability'
+FROM capability_source
+ON CONFLICT (internal_sku, channel) DO UPDATE SET
+  priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
+  is_fallback = internal_sku_channels.is_fallback AND EXCLUDED.is_fallback,
+  canonical_product_id = COALESCE(internal_sku_channels.canonical_product_id, EXCLUDED.canonical_product_id),
+  updated_at = NOW();
+
+WITH rule_source AS (
+  SELECT btrim(match_key) AS internal_sku, channel, MIN(priority) AS priority
+  FROM sku_channel_rules
+  WHERE match_key_type = 'internal_sku'
+    AND provider = '' AND shop_id = ''
+    AND enabled = TRUE
+    AND btrim(match_key) <> ''
+  GROUP BY 1, 2
+)
+INSERT INTO internal_sku_channels (
+  internal_sku, channel, priority, is_fallback, canonical_product_id, source
+)
+SELECT internal_sku, channel, priority, FALSE, NULL, 'rule'
+FROM rule_source
+ON CONFLICT (internal_sku, channel) DO UPDATE SET
+  priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
+  is_fallback = FALSE,
+  updated_at = NOW();
@@ -0,0 +1,14 @@
+-- 119_fulfillment_pending_reason.sql
+--
+-- 待分类不再是一个模糊状态：落明确原因，供队列/审计区分。
+--   no_internal_sku      没有内部中文 SKU
+--   no_channel_capability 有内部 SKU，但没有渠道能力/规则
+--   no_carrier           有渠道能力，但缺任务/工单载体
+
+ALTER TABLE order_items
+  ADD COLUMN IF NOT EXISTS fulfillment_pending_reason TEXT,
+  ADD COLUMN IF NOT EXISTS fulfillment_pending_detail JSONB;
+
+CREATE INDEX IF NOT EXISTS idx_order_items_pending_reason
+  ON order_items (fulfillment_pending_reason)
+  WHERE fulfillment_status = 'pending';
@@ -1,6 +1,7 @@
 import { query } from '../../db/client.js'
 import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js'
 import { readWatermarkItemId } from '../watermark-store.js'
+import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js'

 /**
 * 管道 B：商品归一。
@@ -191,6 +192,15 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise<number | null
    [productId],
  )

+  await upsertInternalSkuChannel({
+    internalSku,
+    channel: 'third_party_voucher',
+    priority: 10,
+    isFallback: false,
+    canonicalProductId: productId,
+    source: 'auto',
+  }).catch(() => undefined)
+
  // 同一内部中文 SKU 跨平台码统一分流到卡券渠道。
  await query(
    `
@@ -0,0 +1,195 @@
+import { query } from '../../db/client.js'
+import { getOrderById } from '../../repositories/order-repo.js'
+import { syncDeliveryTasksForOrder } from '../order/delivery-task-service.js'
+import { syncWorkerOrdersForSourceOrder } from '../worker-platform/sync-worker-orders-service.js'
+import { applyIntakeChannelDecision } from './intake-channel-decision-service.js'
+import type { ResolvedFulfillmentItem } from './product-resolution-service.js'
+import { logWarn } from '../../utils/logger.js'
+
+type PendingCarrierItem = {
+  id: number
+  order_id: number
+  sku_code: string
+  sku_name: string
+  quantity: number
+  item_snapshot_json: unknown
+  spec_json: unknown
+  channel: 'affiliate_dash' | 'work_order'
+}
+
+function parseObject(value: unknown): Record<string, unknown> {
+  if (value && typeof value === 'object' && !Array.isArray(value)) {
+    return value as Record<string, unknown>
+  }
+  try {
+    const parsed = JSON.parse(String(value || '{}'))
+    return parsed && typeof parsed === 'object' && !Array.isArray(parsed)
+      ? (parsed as Record<string, unknown>)
+      : {}
+  } catch {
+    return {}
+  }
+}
+
+/**
+ * 为“内部 SKU 有渠道能力、但还没有载体”的 pending 单补载体。
+ *
+ * - 接单平台：走现有工单匹配器，能匹配模板就建工单；
+ * - 自动发货：补 fulfillmentRoute/affiliateDash 后建任务；
+ * - 补不出来的继续保留 no_carrier，等运营补模板/映射。
+ *
+ * 财务渠道判定不在本服务内做；调用方在补完载体后再跑一次 channel_assign。
+ */
+export async function reconcilePendingCarriers(input: { limit?: number } = {}): Promise<{
+  scanned: number
+  workOrdersCreated: number
+  affiliateTasksCreated: number
+  remaining: number
+}> {
+  const limit = Math.min(500, Math.max(1, Math.trunc(Number(input.limit) || 200)))
+  const rows = await query<PendingCarrierItem>(
+    `
+      SELECT oi.id, oi.order_id, oi.sku_code, oi.sku_name, oi.quantity,
+             oi.item_snapshot_json, oi.spec_json,
+             cap.channel
+      FROM order_items oi
+      JOIN LATERAL (
+        SELECT c.channel
+        FROM internal_sku_channels c
+        WHERE c.enabled = TRUE
+          AND c.channel IN ('affiliate_dash', 'work_order')
+          AND btrim(c.internal_sku) = COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), btrim(oi.sku_name))
+        ORDER BY
+          c.priority ASC,
+          CASE c.channel
+            WHEN 'affiliate_dash' THEN 1
+            WHEN 'work_order' THEN 2
+            ELSE 9
+          END,
+          c.channel ASC
+        LIMIT 1
+      ) cap ON TRUE
+      WHERE oi.fulfillment_status = 'pending'
+        AND oi.fulfillment_locked = FALSE
+        AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+        AND NOT EXISTS (
+          SELECT 1 FROM fulfillment_tasks ft
+          WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
+        )
+      ORDER BY oi.id
+      LIMIT $1
+    `,
+    [limit],
+  )
+
+  let workOrdersCreated = 0
+  let affiliateTasksCreated = 0
+  const byOrder = new Map<number, PendingCarrierItem[]>()
+  for (const row of rows.rows) {
+    const list = byOrder.get(Number(row.order_id)) || []
+    list.push(row)
+    byOrder.set(Number(row.order_id), list)
+  }
+
+  for (const [orderId, items] of byOrder) {
+    const order = await getOrderById(orderId)
+    if (!order) continue
+    const itemRows = await query<{
+      id: number
+      order_id: number
+      sku_code: string
+      sku_name: string
+      quantity: number
+      item_snapshot_json: unknown
+      spec_json: unknown
+    }>(
+      `SELECT id, order_id, sku_code, sku_name, quantity, item_snapshot_json, spec_json
+       FROM order_items WHERE order_id = $1`,
+      [orderId],
+    )
+    const itemById = new Map(itemRows.rows.map((item) => [Number(item.id), item]))
+
+    for (const target of items) {
+      const item = itemById.get(Number(target.id))
+      if (!item) continue
+
+      try {
+        if (target.channel === 'work_order') {
+          const result = await syncWorkerOrdersForSourceOrder(order, [item as never], {
+            source: 'carrier_reconcile',
+            autoOnly: true,
+          })
+          workOrdersCreated += result.created.length
+          continue
+        }
+
+        const resolved = {
+          ...item,
+          itemId: String(item.id),
+          externalItemId: String(item.id),
+          externalSkuCode: item.sku_code,
+          externalSkuName: item.sku_name,
+          skuCode: item.sku_code,
+          skuName: item.sku_name,
+          spec: parseObject(item.spec_json),
+          snapshot: parseObject(item.item_snapshot_json),
+          isConfigured: false,
+        } as unknown as ResolvedFulfillmentItem
+        const enriched = await applyIntakeChannelDecision(resolved, {
+          provider: order.provider,
+          platform: order.platform,
+          shopId: order.shop_id,
+        })
+        const snapshot = parseObject(enriched.snapshot)
+        const route = parseObject(snapshot.fulfillmentRoute)
+        if (route.selectedExecutorKey !== 'affiliate_dash') continue
+
+        await query(
+          `UPDATE order_items
+           SET sku_code = $2,
+               spec_json = $3,
+               item_snapshot_json = $4,
+               updated_at = NOW()
+           WHERE id = $1`,
+          [
+            item.id,
+            String(enriched.skuCode || item.sku_code),
+            JSON.stringify(enriched.spec || {}),
+            JSON.stringify(snapshot),
+          ],
+        )
+        const tasks = await syncDeliveryTasksForOrder(order, [
+          {
+            ...item,
+            sku_code: String(enriched.skuCode || item.sku_code),
+            spec_json: enriched.spec || {},
+            item_snapshot_json: snapshot,
+          } as never,
+        ])
+        affiliateTasksCreated += tasks.length
+      } catch (error) {
+        logWarn('[carrier-reconciliation]', '补履约载体失败', {
+          orderItemId: target.id,
+          channel: target.channel,
+          error: error instanceof Error ? error.message : String(error),
+        })
+      }
+    }
+  }
+
+  const remainingResult = await query<{ count: number }>(
+    `
+      SELECT count(*)::int AS count
+      FROM order_items oi
+      WHERE oi.fulfillment_status = 'pending'
+        AND oi.fulfillment_pending_reason = 'no_carrier'
+    `,
+  )
+
+  return {
+    scanned: rows.rows.length,
+    workOrdersCreated,
+    affiliateTasksCreated,
+    remaining: Number(remainingResult.rows[0]?.count || 0),
+  }
+}
@@ -15,9 +15,12 @@ function buildItem(patch: Partial<ChannelCandidateItem> = {}): ChannelCandidateI
    sku_code: 'suit_xxx',
    sku_name: '套装-测试',
    manual_sku_key: 'suit_xxx',
+    internal_sku: '套装-测试',
    normalized_sku_name: '套装-测试',
    has_work_order: false,
    has_affiliate_task: false,
+    has_manual_task: false,
+    capability_channel: null,
    basis_date: '2026-10-31',
    ...patch,
  }
@@ -98,7 +101,9 @@ test('渠道判定：内部 SKU 与 SKU 名称两种键都能命中', () => {
    channel: 'manual',
  })
  assert.equal(
-    resolveChannelDecision(buildItem({ manual_sku_key: 'nick-001' }), [byInternal]).channel,
+    resolveChannelDecision(buildItem({ manual_sku_key: 'nick-001', internal_sku: 'nick-001' }), [
+      byInternal,
+    ]).channel,
    'manual',
  )

@@ -117,3 +122,28 @@ test('渠道判定：停用的规则不参与', () => {
  const decision = resolveChannelDecision(buildItem(), [buildRule({ enabled: false })])
  assert.equal(decision.channel, null)
 })
+
+test('渠道判定：卡券/人工能力可兜底，自动发货/接单必须靠载体', () => {
+  const voucher = resolveChannelDecision(
+    buildItem({ capability_channel: 'third_party_voucher' }),
+    [],
+  )
+  assert.equal(voucher.channel, 'third_party_voucher')
+  assert.equal(voucher.source, 'auto_mapping')
+
+  const manual = resolveChannelDecision(buildItem({ capability_channel: 'manual' }), [])
+  assert.equal(manual.channel, 'manual')
+  assert.equal(manual.source, 'auto_mapping')
+
+  const affiliate = resolveChannelDecision(buildItem({ capability_channel: 'affiliate_dash' }), [])
+  assert.equal(affiliate.channel, null, '自动发货能力没有任务时不能只改财务口径')
+
+  const workOrder = resolveChannelDecision(buildItem({ capability_channel: 'work_order' }), [])
+  assert.equal(workOrder.channel, null, '接单能力没有工单时不能只改财务口径')
+})
+
+test('渠道判定：人工任务作为人工载体证据', () => {
+  const decision = resolveChannelDecision(buildItem({ has_manual_task: true }), [])
+  assert.equal(decision.channel, 'manual')
+  assert.equal(decision.source, 'carrier_evidence')
+})
@@ -63,6 +63,7 @@ export type ChannelCandidateItem = {
  normalized_sku_name: string | null
  has_work_order: boolean
  has_affiliate_task: boolean
+  has_manual_task: boolean
  capability_channel: FulfillmentChannel | null
  basis_date: string
 }
@@ -91,6 +92,25 @@ function itemKey(item: ChannelCandidateItem, keyType: MatchKeyType): string {
  return normalizeKey(item.normalized_sku_name) || normalizeKey(item.sku_name).toLowerCase()
 }

+function resolvePendingReason(item: ChannelCandidateItem): {
+  reason: string
+  detail: Record<string, unknown>
+} {
+  if (!String(item.internal_sku || '').trim()) {
+    return { reason: 'no_internal_sku', detail: { skuCode: item.sku_code } }
+  }
+  if (!item.capability_channel) {
+    return {
+      reason: 'no_channel_capability',
+      detail: { internalSku: item.internal_sku },
+    }
+  }
+  return {
+    reason: 'no_carrier',
+    detail: { internalSku: item.internal_sku, channel: item.capability_channel },
+  }
+}
+
 /** 纯函数：给定规则与订单项，得出渠道决策（便于单测与文档对照）。 */
 export function resolveChannelDecision(
  item: ChannelCandidateItem,
@@ -144,6 +164,15 @@ export function resolveChannelDecision(
      matchedRuleIds: [],
    }
  }
+  if (item.has_manual_task) {
+    return {
+      channel: 'manual',
+      source: 'carrier_evidence',
+      conflict: false,
+      conflictChannels: [],
+      matchedRuleIds: [],
+    }
+  }

  // 卡券/人工没有必须创建的履约任务，可直接用商品渠道能力兜底；
  // 自动发货/接单平台仍要求真实载体，避免只改口径不派单。
@@ -173,6 +202,8 @@ export type AssignChannelsResult = {
  affectedDates: string[]
  /** 本轮新判定的订单项（供立即计算成本，最多 2000 条）。 */
  assignedItemIds: number[]
+  /** 待分类原因计数。 */
+  pendingReasons: Record<string, number>
 }

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

@@ -222,17 +254,20 @@ export async function assignPendingOrderItemChannels(
            SELECT 1 FROM fulfillment_tasks ft
            WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
          ) AS has_affiliate_task,
+          EXISTS (
+            SELECT 1 FROM fulfillment_tasks ft
+            WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'
+          ) AS has_manual_task,
          cap.channel AS capability_channel
        FROM order_items oi
        JOIN orders o ON o.id = oi.order_id
        LEFT JOIN LATERAL (
          SELECT c.channel
-          FROM product_catalog p
-          JOIN product_channel_capabilities c
-            ON c.product_id = p.id AND c.enabled = TRUE
-          WHERE p.status = 'active'
-            AND btrim(p.canonical_name) = ${INTERNAL_SKU_SQL}
+          FROM internal_sku_channels c
+          WHERE c.enabled = TRUE
+            AND btrim(c.internal_sku) = ${INTERNAL_SKU_SQL}
          ORDER BY
+            c.priority ASC,
            CASE c.channel
              WHEN 'affiliate_dash' THEN 1
              WHEN 'work_order' THEN 2
@@ -240,7 +275,7 @@ export async function assignPendingOrderItemChannels(
              WHEN 'manual' THEN 4
              ELSE 9
            END,
-            c.priority ASC
+            c.channel ASC
          LIMIT 1
        ) cap ON TRUE
        WHERE oi.fulfillment_status = 'pending'
@@ -282,6 +317,16 @@ export async function assignPendingOrderItemChannels(
      }
      if (!decision.channel) {
        result.stillPending += 1
+        const pending = resolvePendingReason(item)
+        result.pendingReasons[pending.reason] = (result.pendingReasons[pending.reason] || 0) + 1
+        await query(
+          `UPDATE order_items
+           SET fulfillment_pending_reason = $2,
+               fulfillment_pending_detail = $3,
+               updated_at = NOW()
+           WHERE id = $1 AND fulfillment_status = 'pending'`,
+          [item.id, pending.reason, JSON.stringify(pending.detail)],
+        )
        continue
      }
      assignments.push({ id: item.id, channel: decision.channel, source: decision.source })
@@ -297,6 +342,8 @@ export async function assignPendingOrderItemChannels(
            fulfillment_status = 'assigned',
            fulfillment_source = input.source,
            fulfillment_decided_at = NOW(),
+            fulfillment_pending_reason = NULL,
+            fulfillment_pending_detail = NULL,
            updated_at = NOW()
          FROM unnest($1::bigint[], $2::text[], $3::text[])
            AS input(id, channel, source)
@@ -385,6 +432,7 @@ export type PendingGroupRow = {
  internal_sku: string
  sku_name: string
  sku_codes: string[]
+  pending_reasons: string[]
  item_count: number
  revenue_fen: number
  first_seen_at: string
@@ -432,6 +480,7 @@ export async function listPendingChannelGroups(input: {
        ${INTERNAL_SKU_SQL} AS internal_sku,
        MIN(oi.sku_name) AS sku_name,
        array_agg(DISTINCT oi.sku_code) AS sku_codes,
+        array_remove(array_agg(DISTINCT oi.fulfillment_pending_reason), NULL) AS pending_reasons,
        COUNT(*)::int AS item_count,
        COALESCE(SUM(oi.revenue_amount_fen), 0)::bigint AS revenue_fen,
        MIN(oi.created_at) AS first_seen_at,
@@ -0,0 +1,125 @@
+import {
+  listInternalSkuChannels,
+  type InternalSkuChannelRow,
+} from './internal-sku-channel-service.js'
+import type { ResolvedFulfillmentItem } from './product-resolution-service.js'
+import { matchAffiliateDashSkuByInternalSku } from './product-resolution-service.js'
+import { logWarn } from '../../utils/logger.js'
+
+type IntakeChannelDecisionInput = {
+  provider?: string
+  platform?: string
+  shopId?: string
+}
+
+function parseObject(value: unknown): Record<string, unknown> {
+  return value && typeof value === 'object' && !Array.isArray(value)
+    ? (value as Record<string, unknown>)
+    : {}
+}
+
+function resolveInternalSku(item: ResolvedFulfillmentItem): string {
+  const snapshot = parseObject(item.snapshot)
+  const spec = parseObject(item.spec)
+  return String(
+    snapshot.internalSku || spec.skuNick || spec.outSkuSn || item.skuCode || item.skuName || '',
+  ).trim()
+}
+
+/**
+ * 下单入口读取内部 SKU 渠道决策。
+ *
+ * 目前只补“自动发货”载体：若内部 SKU 能力是 affiliate_dash 且能解析出 dash sku，
+ * 直接写 fulfillmentRoute + affiliateDash，让 delivery-task-service 建任务。
+ * 接单平台仍由 work_order 匹配器创建；卡券/人工无任务，由财务侧兜底。
+ */
+export async function applyIntakeChannelDecision(
+  item: ResolvedFulfillmentItem,
+  _input: IntakeChannelDecisionInput,
+): Promise<ResolvedFulfillmentItem> {
+  const internalSku = resolveInternalSku(item)
+  if (!internalSku) return item
+
+  let channels: InternalSkuChannelRow[] = []
+  try {
+    channels = await listInternalSkuChannels(internalSku)
+  } catch (error) {
+    // 单测/旧库没有 118 迁移时不影响下单主流程，生产由迁移保证。
+    logWarn('[intake-channel-decision]', '读取内部 SKU 渠道失败', {
+      internalSku,
+      error: error instanceof Error ? error.message : String(error),
+    })
+    return item
+  }
+  if (channels.length === 0) return item
+
+  const snapshot = parseObject(item.snapshot)
+  const route = parseObject(snapshot.fulfillmentRoute)
+  const selected = String(route.selectedExecutorKey || '').trim()
+  // 路由配置未命中时会默认落 manual_dispatch；这不是运营显式规则，不能覆盖内部 SKU 能力。
+  const isDefaultManualFallback = selected === 'manual_dispatch' && !route.matchedRule
+  const primary = channels[0]!
+  const nextSnapshot: Record<string, unknown> = {
+    ...snapshot,
+    internalSku: String(snapshot.internalSku || internalSku),
+    internalSkuChannel: primary.channel,
+    internalSkuChannelSource: primary.source,
+  }
+
+  const canOverrideDefaultManual = !selected || isDefaultManualFallback
+  const existingAffiliateSku = String(parseObject(snapshot.affiliateDash).sku || '').trim()
+
+  if (primary.channel === 'affiliate_dash' || selected === 'affiliate_dash') {
+    const dashSku = existingAffiliateSku || matchAffiliateDashSkuByInternalSku(internalSku)
+    if (dashSku) {
+      if (canOverrideDefaultManual || selected === 'affiliate_dash') {
+        nextSnapshot.fulfillmentRoute = {
+          ...route,
+          selectedExecutorKey: 'affiliate_dash',
+          selectedRuleId: String(route.selectedRuleId || ''),
+          reason: canOverrideDefaultManual
+            ? '内部 SKU 渠道能力指定自动发货'
+            : String(route.reason || '自动发货路由'),
+          matchedRule: route.matchedRule ?? null,
+          candidates: Array.isArray(route.candidates) ? route.candidates : [],
+          skipped: Array.isArray(route.skipped) ? route.skipped : [],
+        }
+        nextSnapshot.affiliateDash = {
+          matchMode: 'affiliate_dash_sku_mapping',
+          productNo: internalSku,
+          sku: dashSku,
+          skuName: internalSku,
+        }
+        // 商品归一按 aff:<dashSku> 找商品；内部中文 SKU 固定落 spec.skuNick。
+        return {
+          ...item,
+          skuCode: dashSku,
+          spec: {
+            ...(item.spec && typeof item.spec === 'object' ? item.spec : {}),
+            skuNick: internalSku,
+          },
+          isConfigured: true,
+          snapshot: nextSnapshot,
+        }
+      }
+    }
+    nextSnapshot.internalSkuChannelReason = 'missing_affiliate_dash_mapping'
+    return { ...item, snapshot: nextSnapshot }
+  }
+
+  // 内部能力是接单/卡券，但路由只是默认人工兜底：清掉人工任务，交给工单或财务能力兜底。
+  if (
+    (primary.channel === 'work_order' || primary.channel === 'third_party_voucher') &&
+    isDefaultManualFallback
+  ) {
+    nextSnapshot.fulfillmentRoute = {
+      ...route,
+      selectedExecutorKey: '',
+      reason: `内部 SKU 渠道能力为 ${primary.channel}，不使用默认人工兜底`,
+      matchedRule: null,
+    }
+    return { ...item, isConfigured: false, snapshot: nextSnapshot }
+  }
+
+  return { ...item, snapshot: nextSnapshot }
+}
@@ -0,0 +1,83 @@
+import { query } from '../../db/client.js'
+import type { FulfillmentChannel } from './channel-types.js'
+
+export type InternalSkuChannelRow = {
+  internal_sku: string
+  channel: FulfillmentChannel
+  priority: number
+  is_fallback: boolean
+  enabled: boolean
+  canonical_product_id: number | null
+  source: string
+}
+
+export async function listInternalSkuChannels(
+  internalSku: string,
+): Promise<InternalSkuChannelRow[]> {
+  const key = String(internalSku || '').trim()
+  if (!key) return []
+  const result = await query<InternalSkuChannelRow>(
+    `
+      SELECT internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source
+      FROM internal_sku_channels
+      WHERE btrim(internal_sku) = $1 AND enabled = TRUE
+      ORDER BY
+        priority ASC,
+        CASE channel
+          WHEN 'affiliate_dash' THEN 1
+          WHEN 'work_order' THEN 2
+          WHEN 'third_party_voucher' THEN 3
+          WHEN 'manual' THEN 4
+          ELSE 9
+        END,
+        channel ASC
+    `,
+    [key],
+  )
+  return result.rows
+}
+
+/** 财务侧可直接兜底的渠道：卡券/人工不要求新建履约任务。 */
+export async function resolveInternalSkuFallbackChannel(
+  internalSku: string,
+): Promise<FulfillmentChannel | null> {
+  const channels = await listInternalSkuChannels(internalSku)
+  const safe = channels.find(
+    (row) => row.channel === 'third_party_voucher' || row.channel === 'manual',
+  )
+  return safe?.channel || null
+}
+
+export async function upsertInternalSkuChannel(input: {
+  internalSku: string
+  channel: FulfillmentChannel
+  priority?: number
+  isFallback?: boolean
+  canonicalProductId?: number | null
+  source?: string
+}): Promise<void> {
+  const internalSku = String(input.internalSku || '').trim()
+  if (!internalSku) return
+  await query(
+    `
+      INSERT INTO internal_sku_channels (
+        internal_sku, channel, priority, is_fallback, enabled,
+        canonical_product_id, source, created_at, updated_at
+      ) VALUES ($1, $2, $3, $4, TRUE, $5, $6, NOW(), NOW())
+      ON CONFLICT (internal_sku, channel) DO UPDATE SET
+        priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
+        is_fallback = internal_sku_channels.is_fallback AND EXCLUDED.is_fallback,
+        canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
+        enabled = TRUE,
+        updated_at = NOW()
+    `,
+    [
+      internalSku,
+      input.channel,
+      Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100,
+      input.isFallback === true,
+      input.canonicalProductId ?? null,
+      String(input.source || 'auto').trim() || 'auto',
+    ],
+  )
+}
@@ -377,6 +377,15 @@ function resolveDouyinMashidaItemCandidate({
 * 映射来源 = **权威常量表 + 后台增量覆盖**（`affiliate-dash` 配置面板的 `skuMapping`），
 * 与 91 分支同一套来源：新增抖音商品在后台加一行即可，不必改代码发版。
 */
+export function matchAffiliateDashSkuByInternalSku(internalSku: unknown): string {
+  const raw = String(internalSku ?? '').trim()
+  if (!raw) return ''
+  return matchAffiliateDashSku(
+    mergeAffiliateDashSkuMapping(getAffiliateDashConfig().skuMapping),
+    raw,
+  )
+}
+
 export function matchDouyinMashidaAffiliateDashSku(internalSku: unknown): string {
  const raw = String(internalSku ?? '').trim()
  if (!raw) {
@@ -10,6 +10,7 @@ import {
 import { withTransaction } from '../../db/client.js'
 import { syncDeliveryTasksForOrder } from './delivery-task-service.js'
 import { resolveOrderItemForFulfillment } from '../fulfillment/product-resolution-service.js'
+import { applyIntakeChannelDecision } from '../fulfillment/intake-channel-decision-service.js'
 import {
  resolveOrderFulfillmentReadiness,
  syncOrderFulfillmentAttachments,
@@ -113,14 +114,19 @@ export async function upsertOrderFromSource(
  })

  const resolvedItems = await Promise.all(
-    event.items.map((item) =>
-      resolveOrderItemForFulfillment({
+    event.items.map(async (item) => {
+      const resolved = await resolveOrderItemForFulfillment({
        provider: event.provider,
        platform: event.platform,
        shopId: event.shopId,
        item,
-      }),
-    ),
+      })
+      return applyIntakeChannelDecision(resolved, {
+        provider: event.provider,
+        platform: event.platform,
+        shopId: event.shopId,
+      })
+    }),
  )
  const resolvedFulfillmentItems = resolvedItems as FulfillmentOrderItem[]
  const configuredItems = resolvedFulfillmentItems.filter((item) => item.isConfigured)
@@ -2,6 +2,7 @@ import type { JsonObject } from '../../types/json.js'
 import { query } from '../../db/client.js'
 import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
 import { runAffiliateFallback } from '../fulfillment/affiliate-fallback-service.js'
+import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js'
 import {
  resolveExceptionsForItemsByReason,
  resolveOrphanCostExceptions,
@@ -84,6 +85,10 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
 export async function runCarrierDispatchJob(job: JsonObject = {}) {
  const config = jobConfig(job)
  const limit = clampInt(config.limit, 200, 1, 2000)
+  const reconciled = await reconcilePendingCarriers({ limit: Math.min(limit, 200) })
+  if (reconciled.affiliateTasksCreated > 0) {
+    await assignPendingOrderItemChannels({ batchSize: 200, maxBatches: 1 }).catch(() => undefined)
+  }
  const cursor =
    config.afterId != null
      ? Math.max(0, Math.trunc(Number(config.afterId)))
@@ -126,6 +126,19 @@ function PendingPanel() {
      render: (value: string[] | undefined) => (value?.length ? value.join(' / ') : '-'),
    },
    { title: '商品名', dataIndex: 'sku_name', ellipsis: true },
+    {
+      title: '待分类原因',
+      dataIndex: 'pending_reasons',
+      width: 160,
+      render: (value: string[] | undefined) => {
+        const labels: Record<string, string> = {
+          no_internal_sku: '无内部SKU',
+          no_channel_capability: '无渠道能力',
+          no_carrier: '无履约载体',
+        }
+        return value?.length ? value.map((reason) => labels[reason] || reason).join(' / ') : '-'
+      },
+    },
    { title: '订单数', dataIndex: 'item_count', width: 90 },
    {
      title: '收入',
@@ -7,6 +7,7 @@ export type PendingChannelGroup = {
  internal_sku: string
  sku_name: string
  sku_codes: string[]
+  pending_reasons: string[]
  item_count: number
  revenue_fen: number
  first_seen_at: string
@@ -0,0 +1,141 @@
+# 生产部署核对清单（履约与财务分流统一）
+
+日期：2026-11-02
+适用：生产当前在迁移 115，本次新增 116–119 + backend/frontend。
+
+## 1. 部署前
+
+1. 备份生产数据库（pg_dump，报告文件大小和 sha256）。
+2. 记录当前：
+   - `schema_migrations` 最大版本；
+   - pending 数量及原因；
+   - `order_items` 渠道分布；
+   - `sku_channel_rules` 匹配键分布。
+3. 确认没有正在跑的财务管道（或暂停定时任务）。
+4. 本地/预发已通过：
+   - 后端测试、前端测试、build、lint、typecheck、SQL guard；
+   - 三平台 E2E（自动发货 / 卡券 / 接单 / 抖音自动发货）；
+   - 成本、销售事实金额正确。
+
+## 2. 迁移顺序
+
+```text
+116_voucher_internal_sku_backfill
+117_internal_sku_mappings
+118_internal_sku_channels
+119_fulfillment_pending_reason
+```
+
+重点检查：
+
+- 116：给已有 `tpv:` 商品补 capability 和 internal_sku 规则；
+- 118：按商品能力/规则回填 `internal_sku_channels`；
+  - SQL 已做聚合，避免同名商品同渠道 `ON CONFLICT` 二次更新报错；
+- 119：只加列和索引，风险低。
+
+## 3. 部署顺序
+
+1. 执行迁移 116–119；
+2. 部署 backend；
+3. 部署 frontend；
+4. 启动/恢复定时任务。
+
+## 4. 部署后核对
+
+### 待分类
+
+```sql
+SELECT fulfillment_pending_reason, count(*)
+FROM order_items
+WHERE fulfillment_status = 'pending'
+GROUP BY 1;
+```
+
+预期只剩：
+
+- `no_internal_sku`
+- `no_channel_capability`
+- `no_carrier`
+
+### 11月改名卡
+
+```sql
+SELECT oi.fulfillment_channel, count(*)
+FROM order_items oi
+WHERE oi.sku_name = '11月改名卡'
+  AND oi.fulfillment_status = 'assigned'
+GROUP BY 1;
+```
+
+预期：`third_party_voucher`，不再 manual / pending。
+
+### 财务与载体一致
+
+```sql
+SELECT
+  oi.fulfillment_channel,
+  count(*) FILTER (WHERE EXISTS (
+    SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id
+  )) AS work_orders,
+  count(*) FILTER (WHERE EXISTS (
+    SELECT 1 FROM fulfillment_tasks ft
+    WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
+  )) AS affiliate_tasks,
+  count(*) FILTER (WHERE EXISTS (
+    SELECT 1 FROM fulfillment_tasks ft
+    WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'
+  )) AS manual_tasks
+FROM order_items oi
+WHERE oi.fulfillment_status = 'assigned'
+GROUP BY 1;
+```
+
+### 规则
+
+```sql
+SELECT match_key_type, channel, enabled, count(*)
+FROM sku_channel_rules
+GROUP BY 1,2,3
+ORDER BY 1,2,3;
+```
+
+旧 `sku_code` 规则应只剩无法定位内部 SKU 的少量记录。
+
+### 内部 SKU 渠道表
+
+```sql
+SELECT channel, count(*)
+FROM internal_sku_channels
+WHERE enabled
+GROUP BY 1;
+```
+
+## 5. 数据清理（确认后执行）
+
+```bash
+# 先 dry-run
+npx tsx scripts/merge-duplicate-products.ts
+npx tsx scripts/cleanup-legacy-sku-code-rules.ts
+npx tsx scripts/reclassify-post-cutover.ts
+
+# 确认后 apply
+npx tsx scripts/merge-duplicate-products.ts --apply
+npx tsx scripts/cleanup-legacy-sku-code-rules.ts --apply
+npx tsx scripts/reclassify-post-cutover.ts --apply
+```
+
+顺序：
+
+1. 合并重复商品；
+2. 清理旧 sku_code 规则；
+3. 重判断代后订单；
+4. 跑一次完整管道；
+5. 核对成本/销售事实。
+
+## 6. 回滚
+
+- 代码：回退到部署前镜像/commit；
+- 数据库：
+  - 116–119 均为新增表/列/数据，不删旧表，可保留；
+  - 若必须回滚，恢复到部署前备份；
+- 迁移 118/119 不改旧字段语义，回滚代码后旧逻辑仍可运行。