From 42d587441168bbdb5f7270b8f65c45fa726f65d4 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sat, 31 Oct 2026 20:41:38 +0800 Subject: [PATCH] =?UTF-8?q?fix(cost):=20=E5=BF=AB=E7=85=A7=E5=8F=98?= =?UTF-8?q?=E6=9B=B4=E5=8D=B3=E9=87=8D=E7=AE=97=EF=BC=8C0=20=E5=85=83?= =?UTF-8?q?=E5=8D=A0=E4=BD=8D=E8=A7=86=E4=B8=BA=E7=BC=BA=E5=A4=B1=EF=BC=8C?= =?UTF-8?q?=E9=80=9A=E7=94=A8=E6=88=90=E6=9C=AC=E5=85=A5=E8=B4=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 预估→实际、补配价格时 supersede 旧快照再写新行，不再一次写死 - 全局 0 元版本不产生快照，转入 cost_missing - 比例成本按 rate_bps 计算，费率为 0 视为未配置 - 平台费/支付费/售后等通用成本模型按渠道+通用各写一条快照 --- .../cost-center/cost-snapshot-service.test.ts | 102 +++++++++ .../cost-center/cost-snapshot-service.ts | 214 +++++++++++++++--- 2 files changed, 286 insertions(+), 30 deletions(-) diff --git a/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts b/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts index e54aa193..c72f57b3 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts @@ -3,6 +3,7 @@ import test from 'node:test' import { resolveCost, + resolveGenericCosts, selectCostVersion, type CostItemRow, type CostVersionRow, @@ -74,6 +75,8 @@ function item(patch: Record = {}) { work_product_rule_id: null as number | null, work_order_status: null as string | null, reward_amount: null as number | null, + revenue_fen: 10000, + basis_date: '2026-10-31', ...patch, } } @@ -172,3 +175,102 @@ test('成本解析：人工渠道不在这里算成本', () => { ) assert.equal(resolution, null) }) + +test('全局 0 元占位版本不能当有效成本', () => { + const zeroVersions: CostVersionRow[] = [ + { + id: 900, + cost_item_id: 1, + scope_type: 'global', + canonical_product_id: null, + work_product_rule_id: null, + amount_fen: 0, + rate_bps: 0, + effective_from: '2026-09-01T00:00:00+08:00', + }, + ] + assert.equal( + resolveCost( + item({ fulfillment_channel: 'work_order', canonical_product_id: null }), + costItems, + zeroVersions, + pricingModes, + ), + null, + ) +}) + +test('比例成本按费率计算，费率为 0 视为未配置', () => { + const percentItems = new Map([ + [ + 'sys_work_order_reward', + { + id: 1, + code: 'sys_work_order_reward', + pricing_mode: 'percent_of_revenue', + cost_model: 'work_order_reward', + }, + ], + ]) + const percentVersions: CostVersionRow[] = [ + { + id: 901, + cost_item_id: 1, + scope_type: 'rule', + canonical_product_id: null, + work_product_rule_id: 9, + amount_fen: 0, + rate_bps: 500, + effective_from: '2026-09-01T00:00:00+08:00', + }, + ] + const resolution = resolveCost( + item({ fulfillment_channel: 'work_order', work_product_rule_id: 9, revenue_fen: 10000 }), + percentItems, + percentVersions, + new Map([[1, 'percent_of_revenue']]), + ) + assert.equal(resolution?.totalAmountFen, 500) + + const zeroRate = percentVersions.map((row) => ({ ...row, rate_bps: 0 })) + assert.equal( + resolveCost( + item({ fulfillment_channel: 'work_order', work_product_rule_id: 9 }), + percentItems, + zeroRate, + new Map([[1, 'percent_of_revenue']]), + ), + null, + ) +}) + +test('通用成本模型按商品/全局版本解析且能与渠道成本共存', () => { + const genericItems: CostItemRow[] = [ + { + id: 20, + code: 'sys_platform_fee', + pricing_mode: 'percent_of_revenue', + cost_model: 'platform_fee', + }, + ] + const versionsWithGeneric: CostVersionRow[] = [ + { + id: 902, + cost_item_id: 20, + scope_type: 'global', + canonical_product_id: null, + work_product_rule_id: null, + amount_fen: 0, + rate_bps: 300, + effective_from: '2026-09-01T00:00:00+08:00', + }, + ] + const generic = resolveGenericCosts( + item({ revenue_fen: 20000 }), + genericItems, + versionsWithGeneric, + new Map([[20, 'percent_of_revenue']]), + ) + assert.equal(generic.length, 1) + assert.equal(generic[0]?.totalAmountFen, 600) +}) diff --git a/apps/backend/src/services/cost-center/cost-snapshot-service.ts b/apps/backend/src/services/cost-center/cost-snapshot-service.ts index 8059733d..369477f3 100644 --- a/apps/backend/src/services/cost-center/cost-snapshot-service.ts +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.ts @@ -1,5 +1,6 @@ import { query } from '../../db/client.js' import { nowIso } from '../../utils/time.js' +import { GENERIC_COST_MODELS } from './cost-center-constants.js' import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js' import { readWatermarkItemId } from '../watermark-store.js' @@ -12,8 +13,14 @@ import { readWatermarkItemId } from '../watermark-store.js' * 渠道 → 成本项： * affiliate_dash → sys_affiliate_purchase（按商品） * third_party_voucher → sys_third_party_voucher_cost（按商品） - * work_order → sys_work_order_reward（优先按接单规则，其次按商品） - * manual → 跳过（确认履约时按实际录入） + * work_order → sys_work_order_reward（优先按接单规则，其次按商品；已接单按实际奖励） + * manual → 不写渠道成本（确认履约时按实际录入） + * + * 通用成本模型（platform_fee / payment_fee / after_sales_loss）对所有渠道生效， + * 与渠道成本项各写一条快照；解析不到版本时静默跳过，不进异常队列。 + * + * 快照不是「一次性」：每轮对比当前解析结果与已有有效快照，金额/版本/阶段变化时 + * 先 supersede 旧行再写新行，保证「预估 → 实际」「补配价格」都能被纠正。 */ const CHANNEL_COST_CODE: Record = { @@ -26,6 +33,7 @@ export type CostItemRow = { id: number code: string pricing_mode: string + cost_model: string } export type CostVersionRow = { @@ -35,6 +43,7 @@ export type CostVersionRow = { canonical_product_id: number | null work_product_rule_id: number | null amount_fen: number + rate_bps: number effective_from: string } @@ -49,6 +58,8 @@ export type CostCandidateItem = { work_product_rule_id: number | null work_order_status: string | null reward_amount: number | null + revenue_fen: number + basis_date: string } export type CostResolution = { @@ -142,8 +153,24 @@ export function resolveCost( const pricingMode = pricingModeByItemId.get(costItem.id) || costItem.pricing_mode const amount = Number(version.amount_fen) || 0 const quantity = Math.max(1, Number(item.quantity) || 1) - const total = pricingMode === 'per_item' ? amount * quantity : amount + if (pricingMode === 'percent_of_revenue') { + const rate = Number(version.rate_bps) || 0 + // 比例成本只看费率：费率为 0 说明还没配价，不能当 0 元成本入账。 + if (rate <= 0) return null + const total = Math.round((Number(item.revenue_fen) || 0) * (rate / 10000)) + return { + costItemId: costItem.id, + costVersionId: version.id, + unitAmountFen: total, + totalAmountFen: total, + stage: 'estimated', + } + } + + // 全局 0 元是占位版本，不是真实成本；否则毛利会被静默抬高。 + if (amount <= 0) return null + const total = pricingMode === 'per_item' ? amount * quantity : amount return { costItemId: costItem.id, costVersionId: version.id, @@ -153,12 +180,64 @@ export function resolveCost( } } +/** + * 通用成本模型：平台费/支付费/售后损失等，对所有渠道生效。 + * 只按商品或全局版本解析，不挂接单规则。 + */ +export function resolveGenericCosts( + item: CostCandidateItem, + genericItems: CostItemRow[], + versions: CostVersionRow[], + pricingModeByItemId: Map, +): Array> { + const result: Array> = [] + for (const costItem of genericItems) { + const version = selectCostVersion(versions, { + costItemId: costItem.id, + productId: item.canonical_product_id, + ruleId: null, + basisAt: item.basis_at, + }) + if (!version) continue + const pricingMode = pricingModeByItemId.get(costItem.id) || costItem.pricing_mode + const amount = Number(version.amount_fen) || 0 + + if (pricingMode === 'percent_of_revenue') { + const rate = Number(version.rate_bps) || 0 + if (rate <= 0) continue + const total = Math.round((Number(item.revenue_fen) || 0) * (rate / 10000)) + result.push({ + costItemId: costItem.id, + costVersionId: version.id, + unitAmountFen: total, + totalAmountFen: total, + stage: 'estimated', + }) + continue + } + if (amount <= 0) continue + const quantity = Math.max(1, Number(item.quantity) || 1) + const total = pricingMode === 'per_item' ? amount * quantity : amount + result.push({ + costItemId: costItem.id, + costVersionId: version.id, + unitAmountFen: amount, + totalAmountFen: total, + stage: 'estimated', + }) + } + return result +} + export type SyncCostSnapshotsResult = { scanned: number created: number + superseded: number missing: number lastId: number nextCursor: number + /** 发生过快照新增/替换的业务日，供事实与总账按日回填。 */ + affectedDates: string[] } export async function syncChannelCostSnapshots( @@ -172,18 +251,28 @@ export async function syncChannelCostSnapshots( : await readWatermarkItemId('cost_snapshot') let reachedEnd = false - const costItems = await loadCostItems() + const allCostItems = await loadCostItems() + const channelCostItems = new Map() + const genericCostItems: CostItemRow[] = [] + for (const item of allCostItems) { + if ((GENERIC_COST_MODELS as readonly string[]).includes(item.cost_model)) { + genericCostItems.push(item) + } else if (item.code) channelCostItems.set(item.code, item) + } const versions = await loadActiveVersions() const pricingModeByItemId = new Map() - for (const item of costItems.values()) pricingModeByItemId.set(item.id, item.pricing_mode) + for (const item of allCostItems) pricingModeByItemId.set(item.id, item.pricing_mode) const result: SyncCostSnapshotsResult = { scanned: 0, created: 0, + superseded: 0, missing: 0, lastId: cursor, nextCursor: cursor, + affectedDates: [], } + const affectedDates = new Set() for (let batch = 0; batch < maxBatches; batch += 1) { const rows = await query( @@ -195,6 +284,9 @@ export async function syncChannelCostSnapshots( oi.canonical_product_id, GREATEST(COALESCE(oi.normalized_quantity, oi.quantity, 1), 1)::int AS quantity, COALESCE(o.paid_at, o.created_at) AS basis_at, + (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text + AS basis_date, + COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen, wo.id AS work_order_id, wo.product_rule_id AS work_product_rule_id, wo.status AS work_order_status, @@ -203,17 +295,8 @@ export async function syncChannelCostSnapshots( JOIN orders o ON o.id = oi.order_id LEFT JOIN work_orders wo ON wo.order_item_id = oi.id WHERE oi.fulfillment_status = 'assigned' - AND oi.fulfillment_channel IN ('affiliate_dash', 'third_party_voucher', 'work_order') - -- 接单平台的成本可以只靠规则版本，不要求商品已归一； - -- 其余渠道必须有商品。 - AND ( - oi.canonical_product_id IS NOT NULL - OR (oi.fulfillment_channel = 'work_order' AND wo.product_rule_id IS NOT NULL) - ) - -- 只处理还没有有效快照的订单项，避免每轮重复解析整张表。 - AND NOT EXISTS ( - SELECT 1 FROM cost_snapshots cs - WHERE cs.order_item_id = oi.id AND cs.superseded_at IS NULL + AND oi.fulfillment_channel IN ( + 'affiliate_dash', 'third_party_voucher', 'work_order', 'manual' ) AND oi.id > $1 ORDER BY oi.id @@ -231,14 +314,20 @@ export async function syncChannelCostSnapshots( cursor = Number(rows.rows[rows.rows.length - 1]!.id) result.lastId = cursor - const inserts: Array<{ - item: CostCandidateItem - resolution: NonNullable - }> = [] + const existingMap = await loadActiveSnapshots(rows.rows.map((row) => row.id)) + const inserts: Array<{ item: CostCandidateItem; resolution: NonNullable }> = [] + const supersedeIds: number[] = [] for (const item of rows.rows) { - const resolution = resolveCost(item, costItems, versions, pricingModeByItemId) - if (!resolution) { + const desired: Array> = [] + const code = CHANNEL_COST_CODE[String(item.fulfillment_channel || '')] + const channelResolution = code + ? resolveCost(item, channelCostItems, versions, pricingModeByItemId) + : null + if (channelResolution) { + desired.push(channelResolution) + } else if (code) { + // 只有渠道成本项缺失才算异常；人工渠道本来就不写渠道成本。 result.missing += 1 await upsertAggregatedException({ reason: 'cost_missing', @@ -246,11 +335,35 @@ export async function syncChannelCostSnapshots( orderItemId: item.id, detail: { channel: item.fulfillment_channel }, }) - continue } - inserts.push({ item, resolution }) + + if (genericCostItems.length > 0) { + desired.push(...resolveGenericCosts(item, genericCostItems, versions, pricingModeByItemId)) + } + + const existing = existingMap.get(Number(item.id)) || new Map() + for (const resolution of desired) { + const current = existing.get(Number(resolution.costItemId)) + if ( + current && + Number(current.cost_version_id) === Number(resolution.costVersionId) && + Number(current.total_amount_fen) === Number(resolution.totalAmountFen) && + current.snapshot_stage === resolution.stage + ) { + continue + } + if (current) { + supersedeIds.push(Number(current.id)) + result.superseded += 1 + } + inserts.push({ item, resolution }) + affectedDates.add(item.basis_date) + } } + if (supersedeIds.length > 0) { + await supersedeSnapshots(supersedeIds) + } if (inserts.length === 0) continue const created = await insertSnapshots(inserts) result.created += created @@ -258,9 +371,52 @@ export async function syncChannelCostSnapshots( } result.nextCursor = reachedEnd ? 0 : cursor + result.affectedDates = [...affectedDates].sort() return result } +type ExistingSnapshot = { + id: number + cost_item_id: number + cost_version_id: number + total_amount_fen: number + snapshot_stage: string +} + +/** 批量读取候选订单项当前有效快照，按订单项分组。 */ +async function loadActiveSnapshots( + itemIds: number[], +): Promise>> { + const map = new Map>() + if (itemIds.length === 0) return map + const rows = await query( + ` + SELECT id, order_item_id, cost_item_id, cost_version_id, total_amount_fen, snapshot_stage + FROM cost_snapshots + WHERE superseded_at IS NULL + AND order_item_id = ANY($1::bigint[]) + `, + [itemIds], + ) + for (const row of rows.rows) { + let perItem = map.get(Number(row.order_item_id)) + if (!perItem) { + perItem = new Map() + map.set(Number(row.order_item_id), perItem) + } + perItem.set(Number(row.cost_item_id), row) + } + return map +} + +async function supersedeSnapshots(ids: number[]): Promise { + if (ids.length === 0) return + await query('UPDATE cost_snapshots SET superseded_at = $1 WHERE id = ANY($2::bigint[])', [ + nowIso(), + ids, + ]) +} + async function insertSnapshots( rows: Array<{ item: CostCandidateItem; resolution: NonNullable }>, ): Promise { @@ -322,20 +478,18 @@ async function insertSnapshots( return result.rowCount || 0 } -async function loadCostItems(): Promise> { +async function loadCostItems(): Promise { const rows = await query( - `SELECT id, code, pricing_mode FROM cost_items WHERE status = 'active'`, + `SELECT id, code, pricing_mode, cost_model FROM cost_items WHERE status = 'active'`, ) - const map = new Map() - for (const row of rows.rows) map.set(row.code, row) - return map + return rows.rows } async function loadActiveVersions(): Promise { const rows = await query( ` SELECT id, cost_item_id, scope_type, canonical_product_id, - work_product_rule_id, amount_fen, effective_from + work_product_rule_id, amount_fen, rate_bps, effective_from FROM cost_versions WHERE status = 'active' `,