feat(finance): P1 人工商品归一，按内部 SKU 生成 man 商品并回填

- 新服务 manual-product-normalization-service：skuNick → outSkuSn → sku_code 兜底，按内部 SKU 生成 man:<sku> 商品
- 订单项按 id 分批（2 万/批）回填，避免 15s 语句超时；同步补齐人工成本快照的商品维度
- 新增 POST /cost-center/manual-products/normalize（支持 dryRun/日期窗口），成本中心「标准商品」加「归一人工商品」按钮
- 每日 cost_finance_sync 接入最近 N 天的增量归一
- 渠道映射未归一队列增加人工单提示：人工单不能用平台 sku_code 建映射
- 文档同步：P1 标记完成，补充归一接口契约与销售事实重算结果

apps/backend/src/routes/admin/cost-center.ts

@@ -59,6 +59,7 @@ import {
   syncAdminAffiliateDashProducts,
   syncAdminWorkOrderTemplateCostVersions,
 } from '../../services/cost-center/template-bootstrap-service.js'
+import { normalizeManualProducts } from '../../services/cost-center/manual-product-normalization-service.js'
 import type { AdminSession } from '../../services/admin/admin-auth-service.js'
 import {
   syncActualWorkOrderCostSnapshots,
@@ -198,6 +199,30 @@ router.post(
   }),
 )
 
+router.post(
+  '/cost-center/manual-products/normalize',
+  requireAdminPermission('finance.config'),
+  createJsonHandler(
+    (req) =>
+      normalizeManualProducts({
+        dateFrom: bodyOf(req).dateFrom as string | undefined,
+        dateTo: bodyOf(req).dateTo as string | undefined,
+        dryRun: bodyOf(req).dryRun === true,
+      }),
+    {
+      successMessage: '人工商品归一完成',
+      errorMessage: '归一人工商品失败',
+      scope: '[admin/cost-center/manual-products/normalize]',
+      audit: (_req, data) => ({
+        action: 'cost_manual_products_normalized',
+        targetType: 'product_catalog',
+        targetId: '',
+        data: data as Record<string, unknown>,
+      }),
+    },
+  ),
+)
+
 router.get(
   '/cost-center/products',
   requireAdminPermission('finance.view'),

apps/backend/src/services/cost-center/manual-product-normalization-service.ts

@@ -0,0 +1,273 @@
+/**
+ * P1 人工商品归一：把人工履约订单项归到 `man:<内部SKU>` 标准商品。
+ *
+ * 为什么不用渠道映射：人工单的「内部 SKU」是快手 spec_json.skuNick / 拼多多
+ * spec_json.outSkuSn，与平台 sku_code 不是同一个值（91kaquan 实测 0 条相同），
+ * 所以按 provider + shop + sku_code 建映射抓不到这些单。这里直接按内部 SKU 归一。
+ *
+ * 幂等：商品按 canonical_sku 去重；订单项只更新 canonical_product_id IS NULL 的行，
+ * 并按 id 分批（每批 2 万）避免撞 DATABASE_STATEMENT_TIMEOUT_MS。
+ */
+import { query } from '../../db/client.js'
+import { nowIso } from '../../utils/time.js'
+
+const MANUAL_PRODUCT_CATEGORY = '人工客服'
+const MANUAL_PRODUCT_UNIT = '件'
+const NORMALIZE_BATCH_SIZE = 20000
+
+export type NormalizeManualProductsInput = {
+  dateFrom?: string | undefined
+  dateTo?: string | undefined
+  dryRun?: boolean
+  resolveExceptions?: boolean
+}
+
+export type NormalizeManualProductsResult = {
+  dryRun: boolean
+  range: { fromDate: string; toDate: string } | null
+  distinctSkuCount: number
+  pendingItemCount: number
+  createdProductCount: number
+  updatedItemCount: number
+  updatedSnapshotCount: number
+  resolvedExceptionCount: number
+  remainingItemCount: number
+}
+
+/** 内部 SKU：快手 skuNick / 拼多多 outSkuSn / 平台 sku_code 兜底。 */
+const MANUAL_SKU_KEY_SQL = `
+  COALESCE(
+    NULLIF(TRIM(oi.spec_json->>'skuNick'), ''),
+    NULLIF(TRIM(oi.spec_json->>'outSkuSn'), ''),
+    NULLIF(TRIM(oi.sku_code), '')
+  )
+`
+
+export async function normalizeManualProducts(
+  input: NormalizeManualProductsInput = {},
+): Promise<NormalizeManualProductsResult> {
+  const fromDate = normalizeDate(input.dateFrom)
+  const toDate = normalizeDate(input.dateTo)
+  const hasRange = Boolean(fromDate || toDate)
+  const dryRun = input.dryRun === true
+  const now = nowIso()
+  const joinSql = hasRange ? 'JOIN orders o ON o.id = oi.order_id' : ''
+
+  // 日期条件按传入的起始占位符编号生成，便于不同语句复用。
+  function buildRangeClause(startIndex: number) {
+    const params: unknown[] = []
+    const parts: string[] = []
+    if (fromDate) {
+      parts.push(
+        `(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date >= $${startIndex + params.length}::date`,
+      )
+      params.push(fromDate)
+    }
+    if (toDate) {
+      parts.push(
+        `(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date <= $${startIndex + params.length}::date`,
+      )
+      params.push(toDate)
+    }
+    return { sql: parts.length ? `AND ${parts.join(' AND ')}` : '', params }
+  }
+
+  const candidateRange = buildRangeClause(1)
+  const pendingWhere = `
+    oi.fulfillment_mode = 'manual'
+    AND oi.canonical_product_id IS NULL
+    AND ${MANUAL_SKU_KEY_SQL} IS NOT NULL
+    ${candidateRange.sql}
+  `
+
+  const candidatesResult = await query<{
+    sku_key: string
+    canonical_name: string
+    item_count: number
+  }>(
+    `
+      WITH item_keys AS (
+        SELECT
+          ${MANUAL_SKU_KEY_SQL} AS sku_key,
+          NULLIF(TRIM(oi.sku_name), '') AS sku_name
+        FROM order_items oi
+        ${joinSql}
+        WHERE ${pendingWhere}
+      )
+      SELECT
+        k.sku_key,
+        LEFT(
+          COALESCE(NULLIF(TRIM(r.display_name), ''), MIN(k.sku_name), k.sku_key),
+          120
+        ) AS canonical_name,
+        COUNT(*)::int AS item_count
+      FROM item_keys k
+      LEFT JOIN manual_sku_registry r ON r.sku_key = k.sku_key
+      GROUP BY k.sku_key, r.display_name
+      ORDER BY COUNT(*) DESC, k.sku_key ASC
+    `,
+    candidateRange.params,
+  )
+
+  const candidates = candidatesResult.rows.map((row) => ({
+    skuKey: row.sku_key,
+    canonicalSku: `man:${row.sku_key}`,
+    canonicalName: row.canonical_name || row.sku_key,
+    itemCount: Number(row.item_count || 0),
+  }))
+  const pendingItemCount = candidates.reduce((sum, item) => sum + item.itemCount, 0)
+
+  if (dryRun) {
+    return {
+      dryRun: true,
+      range: hasRange ? { fromDate: fromDate || '', toDate: toDate || '' } : null,
+      distinctSkuCount: candidates.length,
+      pendingItemCount,
+      createdProductCount: 0,
+      updatedItemCount: 0,
+      updatedSnapshotCount: 0,
+      resolvedExceptionCount: 0,
+      remainingItemCount: pendingItemCount,
+    }
+  }
+
+  let createdProductCount = 0
+  if (candidates.length > 0) {
+    const inserted = await query(
+      `
+        INSERT INTO product_catalog (
+          canonical_sku, canonical_name, category, unit_name, status,
+          attributes_json, created_at, updated_at
+        )
+        SELECT
+          input.canonical_sku,
+          input.canonical_name,
+          $2,
+          $3,
+          'active',
+          jsonb_build_object('source', 'manual_normalization', 'internalSku', input.sku_key),
+          $4::timestamptz,
+          $4::timestamptz
+        FROM unnest($1::text[], $5::text[], $6::text[])
+          AS input(canonical_sku, canonical_name, sku_key)
+        ON CONFLICT (canonical_sku) DO NOTHING
+        RETURNING id
+      `,
+      [
+        candidates.map((item) => item.canonicalSku),
+        MANUAL_PRODUCT_CATEGORY,
+        MANUAL_PRODUCT_UNIT,
+        now,
+        candidates.map((item) => item.canonicalName),
+        candidates.map((item) => item.skuKey),
+      ],
+    )
+    createdProductCount = inserted.rowCount || 0
+  }
+
+  // 回填订单项：按 id 分批，每批重新取日期参数，避免单条语句超过 15s。
+  let updatedItemCount = 0
+  let lastId = 0
+  for (let batch = 0; batch < 40; batch += 1) {
+    const range = buildRangeClause(3)
+    const updated = await query(
+      `
+        WITH target AS (
+          SELECT oi.id, p.id AS product_id
+          FROM order_items oi
+          ${joinSql}
+          JOIN product_catalog p
+            ON p.canonical_sku = 'man:' || ${MANUAL_SKU_KEY_SQL}
+          WHERE oi.fulfillment_mode = 'manual'
+            AND oi.canonical_product_id IS NULL
+            AND oi.id > $1
+            ${range.sql}
+          ORDER BY oi.id
+          LIMIT $2
+        )
+        UPDATE order_items oi
+        SET canonical_product_id = t.product_id,
+            updated_at = NOW()
+        FROM target t
+        WHERE oi.id = t.id
+          AND oi.canonical_product_id IS NULL
+        RETURNING oi.id
+      `,
+      [lastId, NORMALIZE_BATCH_SIZE, ...range.params],
+    )
+    const rowCount = updated.rowCount || 0
+    if (rowCount === 0) break
+    updatedItemCount += rowCount
+    const ids = updated.rows as Array<{ id: number }>
+    lastId = Number(ids[ids.length - 1]?.id || lastId)
+    if (rowCount < NORMALIZE_BATCH_SIZE) break
+  }
+
+  // 已有实际成本快照补上商品维度，让毛利能按商品归集。
+  let updatedSnapshotCount = 0
+  if (updatedItemCount > 0) {
+    const updatedSnapshots = await query(
+      `
+        UPDATE cost_snapshots cs
+        SET canonical_product_id = oi.canonical_product_id
+        FROM order_items oi
+        WHERE cs.order_item_id = oi.id
+          AND cs.canonical_product_id IS NULL
+          AND oi.fulfillment_mode = 'manual'
+          AND oi.canonical_product_id IS NOT NULL
+      `,
+    )
+    updatedSnapshotCount = updatedSnapshots.rowCount || 0
+  }
+
+  let resolvedExceptionCount = 0
+  if (input.resolveExceptions !== false && updatedItemCount > 0) {
+    const resolved = await query(
+      `
+        UPDATE cost_exceptions ce
+        SET status = 'resolved', resolved_at = $1::timestamptz
+        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
+          )
+      `,
+      [now],
+    )
+    resolvedExceptionCount = resolved.rowCount || 0
+  }
+
+  const remainingRange = buildRangeClause(1)
+  const remaining = await query<{ total: number }>(
+    `
+      SELECT COUNT(*)::int AS total
+      FROM order_items oi
+      ${joinSql}
+      WHERE oi.fulfillment_mode = 'manual'
+        AND oi.canonical_product_id IS NULL
+        ${remainingRange.sql}
+    `,
+    remainingRange.params,
+  )
+
+  return {
+    dryRun: false,
+    range: hasRange ? { fromDate: fromDate || '', toDate: toDate || '' } : null,
+    distinctSkuCount: candidates.length,
+    pendingItemCount,
+    createdProductCount,
+    updatedItemCount,
+    updatedSnapshotCount,
+    resolvedExceptionCount,
+    remainingItemCount: Number(remaining.rows[0]?.total || 0),
+  }
+}
+
+function normalizeDate(value: unknown): string {
+  const text = String(value || '').trim()
+  if (!text) return ''
+  if (!/^\d{4}-\d{2}-\d{2}$/.test(text)) return ''
+  return text
+}

apps/backend/src/services/scheduler/cost-finance-sync-job.ts

@@ -3,11 +3,12 @@ import {
   syncActualWorkOrderCostSnapshots,
   syncEstimatedOrderItemCostSnapshots,
 } from '../cost-center/cost-snapshot-sync-service.js'
+import { normalizeManualProducts } from '../cost-center/manual-product-normalization-service.js'
 import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js'
 import { reconcilePendingFulfillmentModes } from '../fulfillment/fulfillment-mode-service.js'
 
 /**
- * 财务闭环同步：履约方式兜底 -> 预估成本快照 -> 工单实际快照 -> 销售事实重算。
+ * 财务闭环同步：履约方式兜底 -> 人工商品归一 -> 预估成本快照 -> 工单实际快照 -> 销售事实重算。
  * 只处理最近 N 天，避免定时任务扫全量历史。
  */
 export async function runCostFinanceSyncJob(job: JsonObject = {}) {
@@ -19,6 +20,12 @@ export async function runCostFinanceSyncJob(job: JsonObject = {}) {
     maxBatches: 10,
     reason: 'cost_finance_sync',
   })
+  // 人工单按内部 SKU 归一；异常收口交给预估同步统一处理，这里只补商品维度。
+  const manualProducts = await normalizeManualProducts({
+    dateFrom: shiftDays(-(days - 1)),
+    dateTo: today(),
+    resolveExceptions: false,
+  })
 
   const estimated = await syncEstimatedOrderItemCostSnapshots({
     dateFrom: shiftDays(-(days - 1)),
@@ -37,12 +44,15 @@ export async function runCostFinanceSyncJob(job: JsonObject = {}) {
     status: 'ok',
     message:
       `履约方式 ${fulfillment.updatedCount} 项 / ` +
+      `人工商品 ${manualProducts.updatedItemCount} 项 / ` +
       `成本快照 ${estimated.createdCount} 商品预估 / ` +
       `${actual.directRewardCount + actual.upgradedCount} 工单实际 / ` +
       `${actual.estimatedRewardCount} 工单在途，销售事实 ${facts.insertedCount} 行`,
     days,
     fulfillmentModeUpdated: fulfillment.updatedCount,
     fulfillmentModeRemaining: fulfillment.remainingCount,
+    manualItemsNormalized: manualProducts.updatedItemCount,
+    manualProductsCreated: manualProducts.createdProductCount,
     estimatedCreated: estimated.createdCount,
     actualCreated: actual.directRewardCount + actual.upgradedCount,
     estimatedRewardCreated: actual.estimatedRewardCount,