feat(cost-center): 渠道映射改为诊断优先，新增未归一队列

- 顶部统计卡：标准商品 / 映射 / 未归一订单项 / 未归一 SKU 组 - 新增「未归一」队列：按来源+店铺+平台SKU+履约方式聚合，行内建映射并预填 - 目录表从 7 列精简到 4 列，匹配方式移入展开明细 - 回填订单项后自动把已归一的 product_mapping_missing 异常收口 - 新增 /cost-center/mappings/diagnostics 与 /mappings/unmapped，按履约方式拆分聚合避免全表扫描
2026-10-24 16:36:16 +08:00
parent 956ef5d695
commit e5122d116a
5 changed files with 787 additions and 159 deletions
@@ -27,8 +27,10 @@ import {
  createAdminProduct,
  deleteAdminChannelMapping,
  deleteAdminProduct,
+  getAdminChannelMappingDiagnostics,
  listAdminChannelMappings,
  listAdminProducts,
+  listAdminUnmappedMappingGroups,
  updateAdminChannelMapping,
  updateAdminProduct,
 } from '../../services/cost-center/product-catalog-service.js'
@@ -273,6 +275,26 @@ router.get(
  }),
 )

+router.get(
+  '/cost-center/mappings/diagnostics',
+  requireAdminPermission('finance.view'),
+  createJsonHandler((req) => getAdminChannelMappingDiagnostics(req.query as JsonObject), {
+    successMessage: 'ok',
+    errorMessage: '读取渠道映射诊断失败',
+    scope: '[admin/cost-center/mappings/diagnostics]',
+  }),
+)
+
+router.get(
+  '/cost-center/mappings/unmapped',
+  requireAdminPermission('finance.view'),
+  createJsonHandler((req) => listAdminUnmappedMappingGroups(req.query as JsonObject), {
+    successMessage: 'ok',
+    errorMessage: '读取未归一队列失败',
+    scope: '[admin/cost-center/mappings/unmapped]',
+  }),
+)
+
 router.post(
  '/cost-center/mappings',
  requireAdminPermission('finance.config'),
@@ -15,6 +15,7 @@ import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-ma
 import { resolveSkuNameQuantity } from '../product-match/pricing.js'
 import {
  AFFILIATE_TASK_AGG_JOIN,
+  FULFILLMENT_MODES,
  MAPPING_CONFIDENCES,
  MAPPING_SOURCES,
  PRODUCT_CATEGORY_JOINS,
@@ -500,6 +501,227 @@ export async function listAdminChannelMappings(payload: JsonObject = {}) {
  }
 }

+/**
+ * 渠道映射诊断：商品/映射数量 + 未归一订单项与 SKU 组数。
+ *
+ * 未归一数据直接读成本异常队列（145k 行），不去扫 316k 订单项。
+ * 按履约方式分组统计：带模式过滤时走单次聚合（约 80ms），
+ * 「全部」时并行跑四种模式再求和，避免一次全表 hash join。
+ */
+export async function getAdminChannelMappingDiagnostics(payload: JsonObject = {}) {
+  const source = text(payload.carrierType ?? payload.carrier_type ?? payload.source)
+  if (source) assertOneOf(source, PRODUCT_SOURCE_FILTERS, '商品来源无效')
+  const fulfillmentMode = text(payload.fulfillmentMode ?? payload.fulfillment_mode)
+  if (fulfillmentMode) assertOneOf(fulfillmentMode, FULFILLMENT_MODES, '履约方式无效')
+
+  const productParams: unknown[] = []
+  let sourceSql = 'TRUE'
+  if (source) {
+    productParams.push(source)
+    sourceSql = productSourceFilterSql(productParams.length)
+  }
+  const counts = await query<{
+    product_total: number
+    mapping_total: number
+    mapping_product_total: number
+  }>(
+    `
+      SELECT
+        (
+          SELECT COUNT(*)::int FROM product_catalog p
+          WHERE p.status = 'active' AND ${sourceSql}
+        ) AS product_total,
+        (
+          SELECT COUNT(*)::int
+          FROM channel_product_mappings m
+          JOIN product_catalog p ON p.id = m.canonical_product_id
+          WHERE m.enabled = TRUE AND ${sourceSql}
+        ) AS mapping_total,
+        (
+          SELECT COUNT(DISTINCT m.canonical_product_id)::int
+          FROM channel_product_mappings m
+          JOIN product_catalog p ON p.id = m.canonical_product_id
+          WHERE m.enabled = TRUE AND ${sourceSql}
+        ) AS mapping_product_total
+    `,
+    productParams,
+  )
+
+  async function aggregateUnmapped(mode: string) {
+    const result = await query<{ group_total: number; item_total: number }>(
+      `
+        WITH grouped AS (
+          SELECT
+            ce.detail_json->>'provider' AS provider,
+            ce.detail_json->>'shopId' AS shop_id,
+            COALESCE(ce.detail_json->>'skuCode', '') AS sku_code,
+            oi.fulfillment_mode AS fulfillment_mode,
+            COUNT(*)::int AS item_count
+          FROM cost_exceptions ce
+          JOIN order_items oi ON oi.id = ce.order_item_id
+          WHERE ce.reason = 'product_mapping_missing'
+            AND ce.status = 'open'
+            AND oi.fulfillment_mode = $1
+          GROUP BY 1, 2, 3, 4
+        )
+        SELECT
+          COUNT(*)::int AS group_total,
+          COALESCE(SUM(item_count), 0)::int AS item_total
+        FROM grouped
+      `,
+      [mode],
+    )
+    return {
+      groupTotal: toNumber(result.rows[0]?.group_total),
+      itemTotal: toNumber(result.rows[0]?.item_total),
+    }
+  }
+
+  const modes = fulfillmentMode
+    ? [fulfillmentMode]
+    : ['work_order', 'affiliate_dash', 'legacy_platform', 'manual']
+  // 顺序跑：并行会同时扫 cost_exceptions，实测 4 路并发比串行更慢（I/O 争抢）。
+  const aggregates: Array<{ groupTotal: number; itemTotal: number }> = []
+  for (const mode of modes) {
+    aggregates.push(await aggregateUnmapped(mode))
+  }
+  const unmappedGroupTotal = aggregates.reduce((sum, item) => sum + item.groupTotal, 0)
+  const unmappedItemTotal = aggregates.reduce((sum, item) => sum + item.itemTotal, 0)
+
+  return {
+    fulfillmentMode: fulfillmentMode || null,
+    productTotal: toNumber(counts.rows[0]?.product_total),
+    mappingTotal: toNumber(counts.rows[0]?.mapping_total),
+    mappedProductTotal: toNumber(counts.rows[0]?.mapping_product_total),
+    unmappedGroupTotal,
+    unmappedItemTotal,
+  }
+}
+
+/**
+ * 未归一订单项队列：按「来源 + 店铺 + 平台 SKU + 履约方式」聚合。
+ *
+ * 每行代表一个可以直接建映射的平台 SKU，而不是 145k 条订单项流水。
+ */
+export async function listAdminUnmappedMappingGroups(payload: JsonObject = {}) {
+  const page = normalizePage(payload.page)
+  const pageSize = normalizePageSize(payload.pageSize)
+  const offset = (page - 1) * pageSize
+  const params: unknown[] = []
+  const clauses = [`ce.reason = 'product_mapping_missing'`, `ce.status = 'open'`]
+
+  const fulfillmentMode = text(payload.fulfillmentMode ?? payload.fulfillment_mode)
+  if (fulfillmentMode) assertOneOf(fulfillmentMode, FULFILLMENT_MODES, '履约方式无效')
+  const provider = text(payload.provider)
+  if (provider) {
+    params.push(normalizeProviderKey(provider))
+    clauses.push(`ce.detail_json->>'provider' = $${params.length}`)
+  }
+  const keyword = text(payload.keyword)
+  if (keyword) {
+    params.push(`%${keyword}%`)
+    clauses.push(
+      `(ce.detail_json->>'skuCode' ILIKE $${params.length}
+        OR ce.detail_json->>'skuName' ILIKE $${params.length}
+        OR ce.detail_json->>'shopId' ILIKE $${params.length})`,
+    )
+  }
+
+  // 分组总数单独算：SQL 守卫禁止窗口函数分页（会在 LIMIT 前消费全部候选行）。
+  // 按履约方式拆开统计：命中索引，单模式约 90ms；「全部」时顺序跑四种模式。
+  const modes = fulfillmentMode
+    ? [fulfillmentMode]
+    : ['work_order', 'affiliate_dash', 'legacy_platform', 'manual']
+  const totals: Array<{ groupTotal: number; itemTotal: number }> = []
+  for (const mode of modes) {
+    const totalParams = [...params, mode]
+    const result = await query<{ group_total: number; item_total: number }>(
+      `
+        WITH grouped AS (
+          SELECT
+            ce.detail_json->>'provider' AS provider,
+            ce.detail_json->>'shopId' AS shop_id,
+            COALESCE(ce.detail_json->>'skuCode', '') AS sku_code,
+            COUNT(*)::int AS item_count
+          FROM cost_exceptions ce
+          JOIN order_items oi ON oi.id = ce.order_item_id
+          WHERE ${clauses.join(' AND ')}
+            AND oi.fulfillment_mode = $${totalParams.length}
+          GROUP BY 1, 2, 3
+        )
+        SELECT
+          COUNT(*)::int AS group_total,
+          COALESCE(SUM(item_count), 0)::int AS item_total
+        FROM grouped
+      `,
+      totalParams,
+    )
+    totals.push({
+      groupTotal: toNumber(result.rows[0]?.group_total),
+      itemTotal: toNumber(result.rows[0]?.item_total),
+    })
+  }
+  const groupTotal = totals.reduce((sum, item) => sum + item.groupTotal, 0)
+  const itemTotal = totals.reduce((sum, item) => sum + item.itemTotal, 0)
+  if (groupTotal === 0) {
+    return {
+      items: [],
+      pagination: { page, pageSize, total: 0, itemTotal: 0 },
+    }
+  }
+
+  const modeIndex = fulfillmentMode ? params.length + 1 : null
+  const pageIndex = params.length + (modeIndex ? 2 : 1)
+  const offsetIndex = pageIndex + 1
+  const listParams = [...params, ...(fulfillmentMode ? [fulfillmentMode] : []), pageSize, offset]
+  const modeClause = modeIndex ? `AND oi.fulfillment_mode = $${modeIndex}` : ''
+  const result = await query<{
+    provider: string
+    shop_id: string
+    sku_code: string
+    sku_name: string
+    fulfillment_mode: string
+    item_count: number
+    last_seen_at: string | null
+  }>(
+    `
+      WITH grouped AS (
+        SELECT
+          ce.detail_json->>'provider' AS provider,
+          ce.detail_json->>'shopId' AS shop_id,
+          COALESCE(ce.detail_json->>'skuCode', '') AS sku_code,
+          oi.fulfillment_mode AS fulfillment_mode,
+          MAX(ce.detail_json->>'skuName') AS sku_name,
+          MAX(NULLIF(ce.detail_json->>'basisAt', '')) AS last_seen_at,
+          COUNT(*)::int AS item_count
+        FROM cost_exceptions ce
+        JOIN order_items oi ON oi.id = ce.order_item_id
+        WHERE ${clauses.join(' AND ')}
+          ${modeClause}
+        GROUP BY 1, 2, 3, 4
+      )
+      SELECT *
+      FROM grouped g
+      ORDER BY g.item_count DESC, g.provider ASC, g.shop_id ASC, g.sku_code ASC
+      LIMIT $${pageIndex} OFFSET $${offsetIndex}
+    `,
+    listParams,
+  )
+
+  return {
+    items: result.rows.map((row) => ({
+      provider: row.provider || '',
+      shopId: row.shop_id || '',
+      skuCode: row.sku_code || '',
+      skuName: row.sku_name || '',
+      fulfillmentMode: row.fulfillment_mode || '',
+      itemCount: toNumber(row.item_count),
+      lastSeenAt: row.last_seen_at || null,
+    })),
+    pagination: { page, pageSize, total: groupTotal, itemTotal },
+  }
+}
+
 export async function createAdminChannelMapping(payload: JsonObject, _session: AdminSession) {
  const provider = normalizeProviderKey(normalizeRequiredText(payload.provider, '来源', 64))
  assertKnownProvider(provider)
@@ -774,6 +996,26 @@ export async function backfillAdminProductMappings(payload: JsonObject = {}) {
    if (!all || itemsResult.rows.length < limit) break
  }

+  // 回填后把已经归一的「商品未归一」异常一起收口，避免映射目录改完了
+  // 未归一队列还挂着一批已处理完的旧异常。
+  let resolvedExceptionCount = 0
+  if (updatedCount > 0) {
+    const resolved = await query(
+      `
+        UPDATE cost_exceptions ce
+        SET status = 'resolved', resolved_at = NOW()
+        WHERE ce.status = 'open'
+          AND ce.subject_type = 'order_item'
+          AND ce.reason = 'product_mapping_missing'
+          AND EXISTS (
+            SELECT 1 FROM order_items oi
+            WHERE oi.id = ce.subject_id AND oi.canonical_product_id IS NOT NULL
+          )
+      `,
+    )
+    resolvedExceptionCount = resolved.rowCount || 0
+  }
+
  const remainingResult = await query<{ total: number }>(
    `
      SELECT COUNT(*)::int AS total
@@ -786,6 +1028,7 @@ export async function backfillAdminProductMappings(payload: JsonObject = {}) {
  )
  return {
    updatedCount,
+    resolvedExceptionCount,
    limit,
    batches,
    remainingCount: Number(remainingResult.rows[0]?.total || 0),