feat(finance): 自动发货订单项按上游 SKU 直接归一

自动发货只有 affiliate-dash 单一上游渠道，订单项直接按 fulfillment_tasks.context_json.skuCode 归一为 aff:<skuCode>，不写 channel_product_mappings。\n\n新增 /cost-center/affiliate-products/backfill：按 order_item_id 取最新任务 SKU 回填 canonical_product_id，只更新有变化的订单项，并自动重算受影响日期的销售事实；返回 updatedCount/skippedCount/remainingCount。\n\n自动发货 SKU 列表改为实时统计履约任务数与最近履约时间，不读会被价格导入覆盖的 attributes_json。\n\n本地实测 41,676 个订单项全部归一，自动发货未归一收入 5,089,712.26 元降至 0；整体未归一占比 79.67% 降至 60.47%。
2026-10-22 18:41:31 +08:00
parent 0889235870
commit 54d2eb2f59
7 changed files with 227 additions and 10 deletions
@@ -3,6 +3,7 @@ import { Router } from 'express'

 import {
  applyAdminTemplateProductBootstrap,
+  backfillAdminAffiliateDashSkus,
  backfillAdminNormalizedQuantities,
  backfillAdminProductMappings,
  createAdminChannelMapping,
@@ -116,6 +117,22 @@ router.post(
  }),
 )

+router.post(
+  '/cost-center/affiliate-products/backfill',
+  requireAdminPermission('finance.config'),
+  createJsonHandler(() => backfillAdminAffiliateDashSkus(), {
+    successMessage: '自动发货 SKU 已回填',
+    errorMessage: '回填自动发货 SKU 失败',
+    scope: '[admin/cost-center/affiliate-products/backfill]',
+    audit: (_req, data) => ({
+      action: 'cost_affiliate_skus_backfilled',
+      targetType: 'order_items',
+      targetId: '',
+      data: data as Record<string, unknown>,
+    }),
+  }),
+)
+
 router.get(
  '/cost-center/products',
  requireAdminPermission('finance.view'),
@@ -2,7 +2,9 @@ import { query, withTransaction } from '../../db/client.js'
 import type { AdminSession } from './admin-auth-service.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
+import { shiftDate } from '../../utils/finance-date-range.js'
 import { nowIso } from '../../utils/time.js'
+import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js'
 import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js'
 import { resolveSkuNameQuantity } from '../product-match/pricing.js'

@@ -59,6 +61,24 @@ const PRODUCT_CATEGORY_JOINS = `
  ) c ON TRUE
  LEFT JOIN work_categories root ON root.id = COALESCE(c.parent_id, c.id)
 `
+/**
+ * 自动发货的履约证据按上游 SKU 聚合一次，再挂到 aff:<skuCode> 商品上。
+ * 商品表里的 attributes_json 会被价格导入覆盖，不能作为统计源。
+ */
+const AFFILIATE_TASK_AGG_JOIN = `
+  LEFT JOIN (
+    SELECT
+      btrim(context_json->>'skuCode') AS sku_code,
+      COUNT(*)::int AS task_count,
+      MAX(created_at) AS last_fulfilled_at
+    FROM fulfillment_tasks
+    WHERE executor_key = 'affiliate_dash'
+      AND btrim(COALESCE(context_json->>'skuCode', '')) <> ''
+    GROUP BY 1
+  ) aff ON p.carrier_type = 'affiliate_dash'
+    AND p.canonical_sku LIKE 'aff:%'
+    AND aff.sku_code = substring(p.canonical_sku from 5)
+`

 type ProductRow = {
  [column: string]: unknown
@@ -78,6 +98,8 @@ type ProductRow = {
  updated_at: string
  mapping_count?: number
  cost_version_count?: number
+  affiliate_task_count?: number
+  affiliate_last_fulfilled_at?: string | null
 }

 type ProductCategoryStatRow = {
@@ -444,9 +466,12 @@ export async function listAdminProducts(payload: JsonObject = {}) {
          root.id AS root_category_id,
          root.name AS root_category_name,
          (SELECT COUNT(*)::int FROM channel_product_mappings m WHERE m.canonical_product_id = p.id) AS mapping_count,
-          (SELECT COUNT(*)::int FROM cost_versions v WHERE v.canonical_product_id = p.id AND v.status = 'active') AS cost_version_count
+          (SELECT COUNT(*)::int FROM cost_versions v WHERE v.canonical_product_id = p.id AND v.status = 'active') AS cost_version_count,
+          COALESCE(aff.task_count, 0) AS affiliate_task_count,
+          aff.last_fulfilled_at AS affiliate_last_fulfilled_at
        FROM product_catalog p
        ${PRODUCT_CATEGORY_JOINS}
+        ${AFFILIATE_TASK_AGG_JOIN}
        ${where}
        ORDER BY
          p.status ASC,
@@ -1524,6 +1549,121 @@ export async function syncAdminAffiliateDashProducts() {
  }
 }

+type AffiliateSkuBackfillRow = {
+  affiliate_item_count: number
+  matched_count: number
+  updated_count: number
+  from_date: string | null
+  to_date: string | null
+}
+
+/**
+ * 按自动发货履约证据直接回填订单项标准商品。
+ *
+ * 自动发货只有 affiliate-dash 一个上游渠道，SKU 固定归一为 `aff:<skuCode>`，
+ * 不写 channel_product_mappings。多个任务共用同一订单项时取最新 SKU。
+ */
+export async function backfillAdminAffiliateDashSkus() {
+  const now = nowIso()
+  const result = await query<AffiliateSkuBackfillRow>(
+    `
+      WITH affiliate_item AS (
+        SELECT DISTINCT ON (ft.order_item_id)
+          ft.order_item_id AS id,
+          btrim(ft.context_json->>'skuCode') AS sku_code
+        FROM fulfillment_tasks ft
+        WHERE ft.executor_key = 'affiliate_dash'
+          AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> ''
+        ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC
+      ),
+      target AS (
+        SELECT ai.id, p.id AS canonical_product_id
+        FROM affiliate_item ai
+        JOIN product_catalog p
+          ON p.canonical_sku = 'aff:' || ai.sku_code
+      ),
+      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 ft.order_item_id)::int
+          FROM fulfillment_tasks ft
+          WHERE ft.executor_key = 'affiliate_dash'
+        ) AS affiliate_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 target
+          JOIN order_items oi ON oi.id = target.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 target
+          JOIN order_items oi ON oi.id = target.id
+          JOIN orders o ON o.id = oi.order_id
+        ) AS to_date
+    `,
+    [now],
+  )
+
+  const row = result.rows[0]
+  const affiliateItemCount = Number(row?.affiliate_item_count || 0)
+  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)
+
+  let factsRebuild: {
+    fromDate: string
+    toDate: string
+    deletedCount: number
+    insertedCount: number
+    batchCount: number
+  } | 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 }
+  }
+
+  return {
+    updatedCount,
+    skippedCount,
+    remainingCount,
+    affiliateItemCount,
+    matchedCount,
+    factsRebuild,
+  }
+}
+
 type TemplateBootstrapTemplateRow = {
  id: number
  rule_key: string
@@ -3297,6 +3437,8 @@ function mapProduct(row: ProductRow) {
    attributes: parseJsonObject(row.attributes_json),
    mappingCount: Number(row.mapping_count || 0),
    costVersionCount: Number(row.cost_version_count || 0),
+    affiliateTaskCount: Number(row.affiliate_task_count || 0),
+    affiliateLastFulfilledAt: row.affiliate_last_fulfilled_at || null,
    createdAt: row.created_at,
    updatedAt: row.updated_at,
  }
@@ -40,6 +40,7 @@ import { usePersistedState } from '@/hooks/usePersistedState'
 import dayjs from '@/lib/dayjs'
 import {
  applyAdminTemplateProductBootstrap,
+  backfillAdminAffiliateDashSkus,
  backfillAdminNormalizedQuantities,
  backfillAdminProductMappings,
  createAdminChannelProductMapping,
@@ -366,6 +367,7 @@ export default function AdminCostCenterPage() {
  const [bootstrapPreviewing, setBootstrapPreviewing] = useState(false)
  const [bootstrapApplying, setBootstrapApplying] = useState(false)
  const [syncingAffiliate, setSyncingAffiliate] = useState(false)
+  const [backfillingAffiliateSkus, setBackfillingAffiliateSkus] = useState(false)
  const versionScopeType = Form.useWatch('scopeType', versionForm)
  const versionCostItemId = Form.useWatch('costItemId', versionForm)

@@ -669,6 +671,29 @@ export default function AdminCostCenterPage() {
    }
  }

+  async function backfillAffiliateSkus() {
+    setBackfillingAffiliateSkus(true)
+    try {
+      const response = await backfillAdminAffiliateDashSkus()
+      const result = response.data
+      const factsText = result.factsRebuild
+        ? `；已重算 ${result.factsRebuild.fromDate} ~ ${result.factsRebuild.toDate} 的销售事实`
+        : ''
+      const remainingText =
+        result.remainingCount > 0
+          ? `；仍有 ${result.remainingCount} 个订单项缺少可匹配的自动发货 SKU`
+          : ''
+      message.success(
+        `已回填 ${result.updatedCount} 个订单项，${result.skippedCount} 个已是归一状态${factsText}${remainingText}`,
+      )
+      refreshAll()
+    } catch (error) {
+      message.error(error instanceof Error ? error.message : '回填自动发货 SKU 失败')
+    } finally {
+      setBackfillingAffiliateSkus(false)
+    }
+  }
+
  function openEditProduct(item: ProductCatalogItem) {
    setEditingProduct(item)
    productForm.setFieldsValue({
@@ -1970,6 +1995,13 @@ export default function AdminCostCenterPage() {
                      >
                        同步自动发货商品
                      </Button>
+                      <Button
+                        icon={<SyncOutlined />}
+                        loading={backfillingAffiliateSkus}
+                        onClick={() => void backfillAffiliateSkus()}
+                      >
+                        回填自动发货 SKU
+                      </Button>
                      <Button
                        icon={<ReloadOutlined />}
                        loading={productsQuery.isFetching}
@@ -1986,7 +2018,7 @@ export default function AdminCostCenterPage() {
                  type="info"
                  showIcon
                  style={{ marginBottom: 12 }}
-                  message="自动发货只按上游货源 SKU 配采购成本：这里不维护游戏分类、渠道映射和商品上下架，那些只属于接单平台。"
+                  message="自动发货只按上游货源 SKU 配采购成本：订单项直接按 aff:<skuCode> 归一，不走渠道映射；这里不维护游戏分类和商品上下架，那些只属于接单平台。"
                />
                <Table
                  tableLayout="fixed"
@@ -2031,14 +2063,14 @@ export default function AdminCostCenterPage() {
                    {
                      title: '履约任务',
                      width: 110,
-                      render: (_, row) => `${Number(row.attributes?.taskCount || 0)} 单`,
+                      render: (_, row) => `${row.affiliateTaskCount} 单`,
                    },
                    {
                      title: '最近履约',
                      width: 180,
                      render: (_, row) =>
-                        row.attributes?.lastFulfilledAt
-                          ? formatAdminDateTime(String(row.attributes.lastFulfilledAt))
+                        row.affiliateLastFulfilledAt
+                          ? formatAdminDateTime(row.affiliateLastFulfilledAt)
                          : '-',
                    },
                    {
@@ -52,6 +52,23 @@ export function syncAdminAffiliateDashProducts() {
  )
 }

+export function backfillAdminAffiliateDashSkus() {
+  return apiPost<{
+    updatedCount: number
+    skippedCount: number
+    remainingCount: number
+    affiliateItemCount: number
+    matchedCount: number
+    factsRebuild: {
+      fromDate: string
+      toDate: string
+      deletedCount: number
+      insertedCount: number
+      batchCount: number
+    } | null
+  }>('/api/v1/admin/cost-center/affiliate-products/backfill')
+}
+
 export function updateAdminProductCatalog(productId: number, payload: Record<string, unknown>) {
  return apiPatch<{ product: ProductCatalogItem }>(
    `/api/v1/admin/cost-center/products/${productId}`,
@@ -27,6 +27,9 @@ export interface ProductCatalogItem {
  attributes: Record<string, unknown>
  mappingCount: number
  costVersionCount: number
+  /** 自动发货履约任务数，实时取自 fulfillment_tasks。 */
+  affiliateTaskCount: number
+  affiliateLastFulfilledAt: string | null
  createdAt: string
  updatedAt: string
 }
@@ -55,7 +55,7 @@
 | 客服备用金台账 | **已实施 P0** | `fund_accounts` / `fund_entries` / `manual_expenses` / `fund_reconciliations`；后台页面、权限、审计已完成 |
 | 后台用户中文显示名 | **已实施** | `admin_users.display_name`，备用金姓名触发器同步；审计、登录记录、任务处理人优先显示中文名 |
 | 商品主数据 + 渠道 SKU 映射 | **已实施 P0（dev 分支）** | 迁移 `103`（映射消歧在 `106`）；`product_catalog` / `channel_product_mappings`，后台「商品与成本」页支持维护、历史订单项回填与按载体分区 |
-| 成本中心 | **已实施 P0（dev 分支）** | 迁移 `104/105/107/108/110`；`cost_items` / `cost_versions` / `cost_snapshots` / `cost_exceptions` 支持版本、预览、快照、异常队列、自动发货价格导入；接单平台按工单实际报酬、人工按支出登记写入实际快照 |
+| 成本中心 | **已实施 P0（dev 分支）** | 迁移 `104/105/107/108/110`；`cost_items` / `cost_versions` / `cost_snapshots` / `cost_exceptions` 支持版本、预览、快照、异常队列、自动发货价格导入；自动发货订单项按履约任务 SKU 直接归一为 `aff:<skuCode>`（不建渠道映射），接单平台按工单实际报酬、人工按支出登记写入实际快照 |
 | 销售事实 / 订单毛利 | **已实施 P0（dev 分支）** | 迁移 `109`；`sales_daily_facts` 按店铺×日期×标准商品×载体汇总，未归一商品单独成行；后台新增「销售毛利」页，成本缺失会显式告警 |
 | 公司人工支出登记 | **已实施 P0（dev 分支）** | 迁移 `111`；`manual_expense_records` / `manual_expense_events`，后台「人工支出」页支持登记、审批、驳回、报销；与客服备用金 `manual_expenses` 通过 `fund_expense_id` 关联 |
 | 影子总账 | 未实施 | 现有财务页仍是 `UNION ALL` 业务事件拼装，不是复式总账 |
@@ -138,7 +138,7 @@ product_catalog
  attributes_json
 ```

-`carrier_type`（迁移 `107_product_catalog_carrier.sql`）是后台「商品与成本中心」的分区维度：自动发货 / 接单平台 / 人工 / 共用。回填口径（2026-10-22 修正）：只把**真正走 affiliate-dash 执行器**的商品算作 `affiliate_dash`——以 `fulfillment_tasks.executor_key='affiliate_dash'` 的履约证据为准，按 `context_json.skuCode` 建标准商品（`aff:<skuCode>`，生产快照 41,676 条任务 / 28 个 SKU）；接单模板初始化出来的 `tpl:*` 商品全部是 `work_order`（662 个）。**渠道映射只代表平台 SKU 能归一到标准商品，不代表自动发货**。人工/共用由后台按需手工调整；新增自动发货 SKU 可在页面点「同步自动发货商品」补建。
+`carrier_type`（迁移 `107_product_catalog_carrier.sql`）是后台「商品与成本中心」的分区维度：自动发货 / 接单平台 / 人工 / 共用。回填口径（2026-10-22 修正）：只把**真正走 affiliate-dash 执行器**的商品算作 `affiliate_dash`——以 `fulfillment_tasks.executor_key='affiliate_dash'` 的履约证据为准，按 `context_json.skuCode` 建标准商品（`aff:<skuCode>`，生产快照 41,676 条任务 / 28 个 SKU）；接单模板初始化出来的 `tpl:*` 商品全部是 `work_order`（662 个）。**渠道映射只代表平台 SKU 能归一到标准商品，不代表自动发货**。自动发货是单一上游渠道，订单项直接取最新 `fulfillment_tasks.context_json.skuCode` 归一为 `aff:<skuCode>`，**不写 `channel_product_mappings`**；页面「回填自动发货 SKU」会回填 `order_items.canonical_product_id` 并自动重算受影响日期的销售事实。人工/共用由后台按需手工调整；新增自动发货 SKU 可在页面点「同步自动发货商品」补建。

 上游自动发货商品价格用 `apps/backend/scripts/sync-affiliate-dash-prices.ts` 批量导入（价格清单见 `apps/backend/scripts/data/affiliate-dash-prices.json`）：按 `aff:<skuCode>` 补建/更新商品，并给指定成本项生成 `scope=product` 的按件版本；金额取上游 `cost_amount`（分），价格变化时先停用旧版本再插入新版本，历史快照不变。示例：`npx tsx scripts/sync-affiliate-dash-prices.ts --cost-item-id=5`。

@@ -364,26 +364,32 @@ orders + order_items + channel_product_mappings + cost_snapshots
 后台入口：`/admin/sales-profit`。回填脚本：`npm run finance:sync -- --days=14 --mode=all`；定时任务 `cost-finance-sync` 每 5 分钟处理最近 7 天。成本缺失会写入 `cost_missing_count`，毛利在补齐成本前只作参考。

 数据校验：本地 2026-07-10 ~ 2026-10-14 共 316,597 个已支付订单，其中 12 单没有订单项（¥2,480.60），无法归集到商品/成本，销售事实会排除并在页面单独提示；其余 316,585 个订单项收入与订单总额完全对平。
-最近 30 天回填后，接单平台实际成本快照 38,023 条、金额 1,633,917.74 元；其中落在销售事实日期范围内的成本 1,563,232.98 元，与工单 reward 逐条一致；自动发货与人工客服仍因缺少映射/正式单价显示成本缺失。
+最近 30 天回填后，接单平台实际成本快照 38,023 条、金额 1,633,917.74 元；其中落在销售事实日期范围内的成本 1,563,232.98 元，与工单 reward 逐条一致。自动发货收入已于 2026-10-22 全量归一（41,676 个订单项），但自动发货采购成本快照仍为 0 条，需要按 `aff:<skuCode>` 的成本版本补跑成本同步；人工客服仍因缺少映射/正式单价显示成本缺失。

 ---

 ## 六、自动发货块

 ```text
-订单项 -> 命中 affiliate_dash SKU 映射 -> fulfillment_task
+订单项 -> 最新 fulfillment_task.context_json.skuCode
+       -> product_catalog.canonical_sku = 'aff:<skuCode>'
       -> 上游商品成本 cost_amount -> cost_snapshot
       -> 履约完成 -> sales_daily_facts
 ```

 要点：

- 上游商品接口已有 `price_amount` / `cost_amount`，但当前只映射、不落库；需要同步到成本中心。
+- **不建渠道映射**：自动发货只有 affiliate-dash 一个上游渠道，订单项直接从履约任务 SKU 归一；`channel_product_mappings` 只服务接单平台。
+- 回填入口：后台「自动发货 SKU」→「回填自动发货 SKU」，接口 `POST /api/v1/admin/cost-center/affiliate-products/backfill`。按 `order_item_id` 取最新任务 SKU，只更新有变化的订单项，完成后自动重算受影响日期的销售事实；返回 `updatedCount / skippedCount / remainingCount`。
+- 列表的「履约任务 / 最近履约」实时取自 `fulfillment_tasks`，不读会被价格导入覆盖的 `product_catalog.attributes_json`。
+- 上游商品接口已有 `price_amount` / `cost_amount`，价格由 `sync-affiliate-dash-prices.ts` 落到成本中心。
 - `price_amount` 是售价/积分，`cost_amount` 是成本，不能混用。
 - 成本变化生成新版本；变动超阈值告警/审批。
 - 店铺/日期/皮肤统计必须走 `canonical_product_id`，不能按原始 `sku_code`。
 - 统计件数要解析 SKU 文本中的数量。

+实测（本地 2026-10-22）：41,676 个自动发货订单项全部回填，自动发货未归一收入 ¥5,089,712.26 → ¥0.00（未归一占比 100% → 0%）。整体未归一收入占比 79.67% → 60.47%，剩余几乎全部来自人工块（¥15,253,660.86，94.84%），需要业务先确定人工 SKU/商品口径。
+
 ---

 ## 七、接单平台块