feat(finance): 补齐自动发货与接单平台成本和对账

自动发货：价格导入支持 --effective-from，把首次基线版本回填到历史起点；41,676 个订单项全部归一并生成采购成本快照，缺失 0。\n\n接单平台：新增按工单 product_rule_id 回填商品的接口与按钮，长尾项按 provider+shop_id+sku_code+normalized_sku_name 补建 wo: 商品和 sku_code_name 渠道映射；118,053 个订单项全部归一。\n\n成本：已验收工单按 reward_amount 写 actual；未验收但已确定报酬的工单写 estimated，验收后升级。全量 95,870 actual + 21,217 estimated，actual 与 reward 0 差异。\n\n剩余 966 条取消/零报酬工单保留在成本异常队列，不自动记 0；人工块未改动。\n\n实测自动发货与接单平台未归一收入均为 0；整体未归一 79.67% 降至 57.54%，全部来自人工块。

apps/backend/src/services/admin/admin-cost-center-service.ts

@@ -1557,6 +1557,24 @@ type AffiliateSkuBackfillRow = {
   to_date: string | null
 }
 
+/** 销售事实单次重算最多 400 天，长历史按窗口分批。 */
+async function rebuildFactsForRange(fromDate: string, toDate: string) {
+  let cursor = fromDate
+  let deletedCount = 0
+  let insertedCount = 0
+  let batchCount = 0
+  while (cursor <= toDate) {
+    const windowEnd = shiftDate(cursor, 399)
+    const chunkEnd = windowEnd > toDate ? toDate : windowEnd
+    const rebuilt = await rebuildSalesDailyFacts({ dateFrom: cursor, dateTo: chunkEnd })
+    deletedCount += rebuilt.deletedCount
+    insertedCount += rebuilt.insertedCount
+    batchCount += 1
+    cursor = shiftDate(chunkEnd, 1)
+  }
+  return { fromDate, toDate, deletedCount, insertedCount, batchCount }
+}
+
 /**
  * 按自动发货履约证据直接回填订单项标准商品。
  *
@@ -1604,8 +1622,8 @@ export async function backfillAdminAffiliateDashSkus() {
             MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
             'YYYY-MM-DD'
           )
-          FROM target
-          JOIN order_items oi ON oi.id = target.id
+          FROM affiliate_item
+          JOIN order_items oi ON oi.id = affiliate_item.id
           JOIN orders o ON o.id = oi.order_id
         ) AS from_date,
         (
@@ -1613,8 +1631,8 @@ export async function backfillAdminAffiliateDashSkus() {
             MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
             'YYYY-MM-DD'
           )
-          FROM target
-          JOIN order_items oi ON oi.id = target.id
+          FROM affiliate_item
+          JOIN order_items oi ON oi.id = affiliate_item.id
           JOIN orders o ON o.id = oi.order_id
         ) AS to_date
     `,
@@ -1626,32 +1644,23 @@ export async function backfillAdminAffiliateDashSkus() {
   const matchedCount = Number(row?.matched_count || 0)
   const updatedCount = Number(row?.updated_count || 0)
   const skippedCount = Math.max(0, matchedCount - updatedCount)
-  const remainingCount = Math.max(0, affiliateItemCount - matchedCount)
+  const remainingResult = await query<{ total: number }>(
+    `
+      SELECT COUNT(*)::int AS total
+      FROM order_items oi
+      WHERE oi.canonical_product_id IS NULL
+        AND EXISTS (
+          SELECT 1 FROM fulfillment_tasks ft
+          WHERE ft.order_item_id = oi.id
+            AND ft.executor_key = 'affiliate_dash'
+        )
+    `,
+  )
+  const remainingCount = Number(remainingResult.rows[0]?.total || 0)
 
-  let factsRebuild: {
-    fromDate: string
-    toDate: string
-    deletedCount: number
-    insertedCount: number
-    batchCount: number
-  } | null = null
+  let factsRebuild: Awaited<ReturnType<typeof rebuildFactsForRange>> | null = null
   if (matchedCount > 0 && row?.from_date && row?.to_date) {
-    const fromDate = row.from_date
-    const toDate = row.to_date
-    let cursor = fromDate
-    let deletedCount = 0
-    let insertedCount = 0
-    let batchCount = 0
-    while (cursor <= toDate) {
-      const windowEnd = shiftDate(cursor, 399)
-      const chunkEnd = windowEnd > toDate ? toDate : windowEnd
-      const rebuilt = await rebuildSalesDailyFacts({ dateFrom: cursor, dateTo: chunkEnd })
-      deletedCount += rebuilt.deletedCount
-      insertedCount += rebuilt.insertedCount
-      batchCount += 1
-      cursor = shiftDate(chunkEnd, 1)
-    }
-    factsRebuild = { fromDate, toDate, deletedCount, insertedCount, batchCount }
+    factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date)
   }
 
   return {
@@ -1664,6 +1673,240 @@ export async function backfillAdminAffiliateDashSkus() {
   }
 }
 
+type WorkOrderSkuBackfillRow = {
+  work_order_item_count: number
+  matched_count: number
+  updated_count: number
+  from_date: string | null
+  to_date: string | null
+}
+
+/**
+ * 回填接单平台订单项的标准商品。
+ *
+ * 第一步用工单真实使用的 `product_rule_id` 指向 `tpl:<rule_key>`；
+ * 仍无法归一的长尾项，再按 `provider + shop_id + sku_code + normalized_sku_name`
+ * 补建 `wo:` 商品和 `sku_code_name` 渠道映射，保证后续同类订单能自动命中。
+ */
+export async function backfillAdminWorkOrderProductSkus() {
+  const now = nowIso()
+  const result = await query<WorkOrderSkuBackfillRow>(
+    `
+      WITH work_order_item AS (
+        SELECT DISTINCT ON (wo.order_item_id)
+          wo.order_item_id AS id,
+          wo.product_rule_id
+        FROM work_orders wo
+        WHERE wo.order_item_id IS NOT NULL
+        ORDER BY wo.order_item_id, wo.id DESC
+      ),
+      target AS (
+        SELECT woi.id, p.id AS canonical_product_id
+        FROM work_order_item woi
+        JOIN work_product_rules wpr ON wpr.id = woi.product_rule_id
+        JOIN product_catalog p ON p.canonical_sku = 'tpl:' || wpr.rule_key
+      ),
+      updated AS (
+        UPDATE order_items oi
+        SET canonical_product_id = target.canonical_product_id,
+            updated_at = $1
+        FROM target
+        WHERE oi.id = target.id
+          AND oi.canonical_product_id IS DISTINCT FROM target.canonical_product_id
+        RETURNING oi.id
+      )
+      SELECT
+        (
+          SELECT COUNT(DISTINCT wo.order_item_id)::int
+          FROM work_orders wo
+          WHERE wo.order_item_id IS NOT NULL
+        ) AS work_order_item_count,
+        (SELECT COUNT(*)::int FROM target) AS matched_count,
+        (SELECT COUNT(*)::int FROM updated) AS updated_count,
+        (
+          SELECT to_char(
+            MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
+            'YYYY-MM-DD'
+          )
+          FROM work_order_item
+          JOIN order_items oi ON oi.id = work_order_item.id
+          JOIN orders o ON o.id = oi.order_id
+        ) AS from_date,
+        (
+          SELECT to_char(
+            MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date),
+            'YYYY-MM-DD'
+          )
+          FROM work_order_item
+          JOIN order_items oi ON oi.id = work_order_item.id
+          JOIN orders o ON o.id = oi.order_id
+        ) AS to_date
+    `,
+    [now],
+  )
+
+  const tailProducts = await query<{ inserted: boolean }>(
+    `
+      WITH tail AS (
+        SELECT
+          oi.normalized_sku_name AS name,
+          MIN(oi.sku_name) AS sample_name
+        FROM order_items oi
+        WHERE oi.canonical_product_id IS NULL
+          AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
+          AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+        GROUP BY 1
+      ),
+      missing AS (
+        SELECT t.name, t.sample_name
+        FROM tail t
+        WHERE NOT EXISTS (
+          SELECT 1 FROM product_catalog p
+          WHERE p.status = 'active' AND p.canonical_name = t.name
+        )
+      )
+      INSERT INTO product_catalog (
+        canonical_sku, canonical_name, category, unit_name, carrier_type, status,
+        attributes_json, created_at, updated_at
+      )
+      SELECT
+        'wo:' || substr(md5(m.name), 1, 12),
+        m.name,
+        '',
+        CASE
+          WHEN m.sample_name LIKE '%颗%' THEN '颗'
+          WHEN m.sample_name LIKE '%个%' THEN '个'
+          ELSE '件'
+        END,
+        'work_order',
+        'active',
+        jsonb_build_object('source', 'work_order_sku_name', 'normalizedName', m.name),
+        $1,
+        $1
+      FROM missing m
+      ON CONFLICT (canonical_sku) DO UPDATE
+      SET carrier_type = 'work_order',
+          updated_at = $1
+      RETURNING (xmax = 0) AS inserted
+    `,
+    [now],
+  )
+
+  const tailMappings = await query<{ id: number }>(
+    `
+      WITH tail AS (
+        SELECT
+          o.provider,
+          o.shop_id,
+          oi.sku_code,
+          MIN(oi.sku_name) AS sku_name,
+          oi.normalized_sku_name
+        FROM order_items oi
+        JOIN orders o ON o.id = oi.order_id
+        WHERE oi.canonical_product_id IS NULL
+          AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
+          AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+        GROUP BY 1, 2, 3, 5
+      ),
+      resolved AS (
+        SELECT t.*, p.id AS canonical_product_id
+        FROM tail t
+        JOIN LATERAL (
+          SELECT id
+          FROM product_catalog
+          WHERE status = 'active' AND canonical_name = t.normalized_sku_name
+          ORDER BY (canonical_sku LIKE 'tpl:%') DESC, id
+          LIMIT 1
+        ) p ON TRUE
+      )
+      INSERT INTO channel_product_mappings (
+        provider, shop_id, platform_sku_code, platform_sku_name,
+        platform_sku_name_key, match_type, canonical_product_id,
+        quantity_multiplier, confidence, source, enabled, note, created_at, updated_at
+      )
+      SELECT
+        r.provider, r.shop_id, r.sku_code, r.sku_name,
+        r.normalized_sku_name, 'sku_code_name', r.canonical_product_id,
+        1, 'exact', 'import', TRUE, '按历史工单 SKU 名称自动补齐', $1, $1
+      FROM resolved r
+      ON CONFLICT (
+        provider, shop_id, platform_sku_code, match_type, platform_sku_name_key
+      ) DO NOTHING
+      RETURNING id
+    `,
+    [now],
+  )
+
+  const tailUpdated = await query<{ id: number }>(
+    `
+      WITH tail AS (
+        SELECT oi.id, oi.normalized_sku_name
+        FROM order_items oi
+        WHERE oi.canonical_product_id IS NULL
+          AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
+          AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+      ),
+      resolved AS (
+        SELECT t.id, p.id AS canonical_product_id
+        FROM tail t
+        JOIN LATERAL (
+          SELECT id
+          FROM product_catalog
+          WHERE status = 'active' AND canonical_name = t.normalized_sku_name
+          ORDER BY (canonical_sku LIKE 'tpl:%') DESC, id
+          LIMIT 1
+        ) p ON TRUE
+      )
+      UPDATE order_items oi
+      SET canonical_product_id = r.canonical_product_id,
+          updated_at = $1
+      FROM resolved r
+      WHERE oi.id = r.id
+        AND oi.canonical_product_id IS NULL
+      RETURNING oi.id
+    `,
+    [now],
+  )
+
+  const row = result.rows[0]
+  const workOrderItemCount = Number(row?.work_order_item_count || 0)
+  const ruleMatchedCount = Number(row?.matched_count || 0)
+  const ruleUpdatedCount = Number(row?.updated_count || 0)
+  const tailProductCount = tailProducts.rows.filter((item) => item.inserted).length
+  const tailMappingCount = tailMappings.rows.length
+  const tailUpdatedCount = tailUpdated.rows.length
+  const updatedCount = ruleUpdatedCount + tailUpdatedCount
+  const remainingResult = await query<{ total: number }>(
+    `
+      SELECT COUNT(*)::int AS total
+      FROM order_items oi
+      WHERE oi.canonical_product_id IS NULL
+        AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+    `,
+  )
+  const remainingCount = Number(remainingResult.rows[0]?.total || 0)
+  const matchedCount = Math.max(0, workOrderItemCount - remainingCount)
+  const skippedCount = Math.max(0, matchedCount - updatedCount)
+
+  let factsRebuild: Awaited<ReturnType<typeof rebuildFactsForRange>> | null = null
+  if (matchedCount > 0 && row?.from_date && row?.to_date) {
+    factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date)
+  }
+
+  return {
+    updatedCount,
+    skippedCount,
+    remainingCount,
+    workOrderItemCount,
+    matchedCount,
+    ruleMatchedCount,
+    tailUpdatedCount,
+    tailProductCount,
+    tailMappingCount,
+    factsRebuild,
+  }
+}
+
 type TemplateBootstrapTemplateRow = {
   id: number
   rule_key: string

apps/backend/src/services/cost-center/cost-snapshot-sync-service.ts

@@ -22,6 +22,7 @@ export type ActualCostSnapshotSyncResult = {
   createdCount: number
   upgradedCount: number
   directRewardCount: number
+  estimatedRewardCount: number
 }
 
 /**
@@ -198,6 +199,11 @@ export async function syncActualWorkOrderCostSnapshots(
       range.fromDate,
       range.toDate,
     )
+    const estimatedReward = await syncCommittedWorkOrderRewardSnapshots(
+      client,
+      range.fromDate,
+      range.toDate,
+    )
     const upgraded = await client.query(
       `
         WITH accepted AS (
@@ -283,9 +289,10 @@ export async function syncActualWorkOrderCostSnapshots(
 
     return {
       range,
-      createdCount: directReward.inserted + (upgraded.rowCount || 0),
+      createdCount: directReward.inserted + estimatedReward.inserted + (upgraded.rowCount || 0),
       upgradedCount: upgraded.rowCount || 0,
       directRewardCount: directReward.inserted,
+      estimatedRewardCount: estimatedReward.inserted,
     }
   })
 }
@@ -327,7 +334,7 @@ async function syncDirectWorkOrderRewardSnapshots(
         JOIN orders o ON o.id = oi.order_id
         WHERE wo.status = 'accepted'
           AND wo.order_item_id IS NOT NULL
-          AND COALESCE(wo.reward_amount, 0) > 0
+          AND COALESCE(wo.reward_amount, 0) >= 0
           AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at)
             AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date
         ORDER BY wo.order_item_id, wo.id DESC
@@ -382,7 +389,7 @@ async function syncDirectWorkOrderRewardSnapshots(
         JOIN orders o ON o.id = oi.order_id
         WHERE wo.status = 'accepted'
           AND wo.order_item_id IS NOT NULL
-          AND COALESCE(wo.reward_amount, 0) > 0
+          AND COALESCE(wo.reward_amount, 0) >= 0
           AND (COALESCE(wo.accepted_at, wo.updated_at, o.paid_at, o.created_at)
             AT TIME ZONE 'Asia/Shanghai')::date BETWEEN $1::date AND $2::date
         ORDER BY wo.order_item_id, wo.id DESC
@@ -429,6 +436,138 @@ async function syncDirectWorkOrderRewardSnapshots(
   }
 }
 
+/**
+ * 未验收工单的报酬在创建时已确定，先按 reward_amount 写 estimated 快照。
+ * 验收后由 syncDirectWorkOrderRewardSnapshots 升级为 actual；取消工单不计成本。
+ */
+async function syncCommittedWorkOrderRewardSnapshots(
+  client: PoolClient,
+  fromDate: string,
+  toDate: string,
+) {
+  const superseded = await client.query(
+    `
+      WITH system_item AS (
+        SELECT ci.id AS cost_item_id
+        FROM cost_items ci
+        WHERE ci.code = 'sys_work_order_reward'
+          AND ci.status = 'active'
+      ),
+      latest AS (
+        SELECT DISTINCT ON (wo.order_item_id)
+          wo.id AS work_order_id,
+          wo.order_item_id,
+          GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount
+        FROM work_orders wo
+        JOIN orders o ON o.id = wo.order_id
+        WHERE wo.order_item_id IS NOT NULL
+          AND wo.status IN (
+            'pending_material', 'in_progress', 'unassigned', 'open',
+            'pending_acceptance', 'problem'
+          )
+          AND COALESCE(wo.reward_amount, 0) > 0
+          AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
+            BETWEEN $1::date AND $2::date
+        ORDER BY wo.order_item_id, wo.id DESC
+      )
+      UPDATE cost_snapshots cs
+      SET superseded_at = NOW()
+      FROM latest l
+      JOIN system_item si ON TRUE
+      WHERE cs.subject_type = 'order_item'
+        AND cs.subject_id = l.order_item_id
+        AND cs.cost_item_id = si.cost_item_id
+        AND cs.superseded_at IS NULL
+        AND cs.snapshot_stage = 'estimated'
+        AND (
+          cs.work_order_id IS DISTINCT FROM l.work_order_id
+          OR cs.total_amount_fen <> l.reward_amount
+        )
+    `,
+    [fromDate, toDate],
+  )
+
+  const inserted = await client.query(
+    `
+      WITH system_item AS (
+        SELECT ci.id AS cost_item_id, v.id AS cost_version_id
+        FROM cost_items ci
+        JOIN LATERAL (
+          SELECT id
+          FROM cost_versions
+          WHERE cost_item_id = ci.id
+            AND status = 'active'
+            AND effective_from <= NOW()
+          ORDER BY effective_from DESC, id DESC
+          LIMIT 1
+        ) v ON TRUE
+        WHERE ci.code = 'sys_work_order_reward'
+          AND ci.status = 'active'
+      ),
+      latest AS (
+        SELECT DISTINCT ON (wo.order_item_id)
+          wo.id AS work_order_id,
+          wo.order_item_id,
+          wo.status,
+          GREATEST(COALESCE(wo.reward_amount, 0), 0)::bigint AS reward_amount,
+          COALESCE(wo.created_at, o.paid_at, o.created_at, NOW()) AS basis_at
+        FROM work_orders wo
+        JOIN orders o ON o.id = wo.order_id
+        WHERE wo.order_item_id IS NOT NULL
+          AND wo.status IN (
+            'pending_material', 'in_progress', 'unassigned', 'open',
+            'pending_acceptance', 'problem'
+          )
+          AND COALESCE(wo.reward_amount, 0) > 0
+          AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
+            BETWEEN $1::date AND $2::date
+        ORDER BY wo.order_item_id, wo.id DESC
+      )
+      INSERT INTO cost_snapshots (
+        subject_type, subject_id, order_id, order_item_id, work_order_id,
+        carrier_type, cost_item_id, cost_version_id, canonical_product_id,
+        quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source,
+        metadata_json, cost_basis_at, resolved_at, created_at
+      )
+      SELECT
+        'order_item', l.order_item_id, oi.order_id, l.order_item_id, l.work_order_id,
+        'work_order', si.cost_item_id, si.cost_version_id, oi.canonical_product_id,
+        1, l.reward_amount, l.reward_amount, 'estimated', 'resolved',
+        jsonb_build_object(
+          'source', 'work_order_reward_committed',
+          'workOrderStatus', l.status,
+          'estimatedRewardFen', l.reward_amount,
+          'basisAt', l.basis_at
+        ),
+        l.basis_at, NOW(), NOW()
+      FROM latest l
+      CROSS JOIN system_item si
+      JOIN order_items oi ON oi.id = l.order_item_id
+      WHERE NOT EXISTS (
+        SELECT 1 FROM work_orders accepted
+        WHERE accepted.order_item_id = l.order_item_id
+          AND accepted.status = 'accepted'
+      )
+        AND NOT EXISTS (
+          SELECT 1 FROM cost_snapshots cs
+          WHERE cs.subject_type = 'order_item'
+            AND cs.subject_id = l.order_item_id
+            AND cs.cost_item_id = si.cost_item_id
+            AND cs.superseded_at IS NULL
+        )
+      ON CONFLICT (subject_type, subject_id, cost_item_id)
+        WHERE superseded_at IS NULL
+      DO NOTHING
+    `,
+    [fromDate, toDate],
+  )
+
+  return {
+    inserted: inserted.rowCount || 0,
+    superseded: superseded.rowCount || 0,
+  }
+}
+
 async function insertAndResolveProductMappingExceptions(
   client: PoolClient,
   fromDate: string,

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

@@ -28,10 +28,14 @@ export async function runCostFinanceSyncJob(job: JsonObject = {}) {
 
   return {
     status: 'ok',
-    message: `成本快照 ${estimated.createdCount} 预估 / ${actual.createdCount} 实际，销售事实 ${facts.insertedCount} 行`,
+    message:
+      `成本快照 ${estimated.createdCount} 商品预估 / ` +
+      `${actual.directRewardCount + actual.upgradedCount} 工单实际 / ` +
+      `${actual.estimatedRewardCount} 工单在途，销售事实 ${facts.insertedCount} 行`,
     days,
     estimatedCreated: estimated.createdCount,
-    actualCreated: actual.createdCount,
+    actualCreated: actual.directRewardCount + actual.upgradedCount,
+    estimatedRewardCreated: actual.estimatedRewardCount,
     exceptionsCreated: estimated.exceptionCreatedCount,
     factsInserted: facts.insertedCount,
     dateFrom: facts.range.fromDate,