diff --git a/apps/backend/src/services/cost-center/cost-exception-service.ts b/apps/backend/src/services/cost-center/cost-exception-service.ts new file mode 100644 index 00000000..e4044631 --- /dev/null +++ b/apps/backend/src/services/cost-center/cost-exception-service.ts @@ -0,0 +1,199 @@ +import { query } from '../../db/client.js' +import { nowIso } from '../../utils/time.js' + +/** + * 成本/分类问题队列(SKU 级)。 + * + * 与旧模型的区别: + * · 旧:一个订单项一条异常(生产曾堆到 36 万行); + * · 新:同一 scope 聚合为一条问题头,明细放 cost_exception_items, + * 新单进来只加计数,解决后自动关闭。 + * + * scope_key 约定:`reason|provider|shop_id|sku|channel`,由调用方拼好。 + */ + +export const COST_EXCEPTION_REASONS = [ + 'classification_missing', + 'routing_conflict', + 'product_mapping_missing', + 'cost_missing', + 'manual_config_missing', +] as const + +export type CostExceptionReason = (typeof COST_EXCEPTION_REASONS)[number] + +export type UpsertExceptionInput = { + reason: CostExceptionReason + scopeKey: string + orderItemId: number + detail?: Record +} + +/** 幂等上浮一条聚合问题:已存在则累加明细并刷新 last_seen。 */ +export async function upsertAggregatedException(input: UpsertExceptionInput): Promise { + const now = nowIso() + const detail = JSON.stringify(input.detail || {}) + + const result = await query<{ id: number }>( + ` + INSERT INTO cost_exceptions ( + subject_type, subject_id, order_item_id, reason, status, detail_json, + scope_key, item_count, first_seen_at, last_seen_at, created_at + ) + VALUES ('order_item', $1, $1, $2, 'open', $3::jsonb, $4, 1, $5, $5, $5) + ON CONFLICT (scope_key) WHERE status <> 'resolved' + DO UPDATE SET + last_seen_at = EXCLUDED.last_seen_at, + detail_json = EXCLUDED.detail_json + RETURNING id + `, + [input.orderItemId, input.reason, detail, input.scopeKey, now], + ) + + const exceptionId = result.rows[0]?.id + if (!exceptionId) return + + await query( + ` + INSERT INTO cost_exception_items (exception_id, order_item_id, created_at) + VALUES ($1, $2, $3) + ON CONFLICT (exception_id, order_item_id) DO NOTHING + `, + [exceptionId, input.orderItemId, now], + ) + + await query( + ` + UPDATE cost_exceptions ce + SET item_count = ( + SELECT COUNT(*)::int FROM cost_exception_items i WHERE i.exception_id = ce.id + ) + WHERE ce.id = $1 + `, + [exceptionId], + ) +} + +/** 问题已解决:关闭问题头并清空明细。 */ +export async function resolveAggregatedException(scopeKey: string): Promise { + const now = nowIso() + const result = await query<{ id: number }>( + ` + UPDATE cost_exceptions + SET status = 'resolved', resolved_at = $2, item_count = 0, last_seen_at = $2 + WHERE scope_key = $1 AND status <> 'resolved' + RETURNING id + `, + [scopeKey, now], + ) + const ids = result.rows.map((row) => row.id) + if (ids.length > 0) { + await query(`DELETE FROM cost_exception_items WHERE exception_id = ANY($1::bigint[])`, [ids]) + } + return ids.length +} + +/** 订单项已补齐配置时,关闭它关联的全部未解决问题。 */ +export async function resolveExceptionsForOrderItem(orderItemId: number): Promise { + const now = nowIso() + const result = await query<{ id: number }>( + ` + UPDATE cost_exceptions ce + SET status = 'resolved', resolved_at = $2, item_count = 0, last_seen_at = $2 + WHERE ce.status <> 'resolved' + AND EXISTS ( + SELECT 1 FROM cost_exception_items i + WHERE i.exception_id = ce.id AND i.order_item_id = $1 + ) + RETURNING id + `, + [orderItemId, now], + ) + const ids = result.rows.map((row) => row.id) + if (ids.length > 0) { + await query(`DELETE FROM cost_exception_items WHERE exception_id = ANY($1::bigint[])`, [ids]) + } + return ids.length +} + +export type ExceptionQueueRow = { + id: number + reason: string + status: string + scope_key: string + item_count: number + first_seen_at: string + last_seen_at: string + detail_json: unknown +} + +export async function listExceptionQueue(input: { + reason?: string + status?: string + page?: number + pageSize?: number +}): Promise<{ items: ExceptionQueueRow[]; total: number; page: number; pageSize: number }> { + const page = Math.max(1, Math.trunc(Number(input.page) || 1)) + const pageSize = Math.min(200, Math.max(1, Math.trunc(Number(input.pageSize) || 30))) + const params: unknown[] = [] + const clauses: string[] = [] + + if (input.reason) { + params.push(String(input.reason).trim()) + clauses.push(`reason = $${params.length}`) + } + if (input.status) { + params.push(String(input.status).trim()) + clauses.push(`status = $${params.length}`) + } + const where = clauses.length > 0 ? `WHERE ${clauses.join(' AND ')}` : '' + + const total = await query<{ count: number }>( + `SELECT COUNT(*)::int AS count FROM cost_exceptions ${where}`, + params, + ) + + params.push(pageSize, (page - 1) * pageSize) + const rows = await query( + ` + SELECT id, reason, status, scope_key, item_count, first_seen_at, last_seen_at, detail_json + FROM cost_exceptions + ${where} + ORDER BY last_seen_at DESC NULLS LAST, id DESC + LIMIT $${params.length - 1} OFFSET $${params.length} + `, + params, + ) + + return { items: rows.rows, total: total.rows[0]?.count || 0, page, pageSize } +} + +/** + * 订单项已补齐后,关闭它关联的未解决问题。 + * + * 批量版本:一次处理一批订单项,避免逐条 UPDATE。 + */ +export async function resolveExceptionsForItems(orderItemIds: number[]): Promise { + const ids = orderItemIds.filter((id) => Number.isInteger(id) && id > 0) + if (ids.length === 0) return 0 + const now = nowIso() + const result = await query<{ id: number }>( + ` + WITH removed AS ( + DELETE FROM cost_exception_items + WHERE order_item_id = ANY($1::bigint[]) + RETURNING exception_id + ) + UPDATE cost_exceptions ce + SET status = 'resolved', resolved_at = $2, item_count = 0, last_seen_at = $2 + WHERE ce.id IN (SELECT DISTINCT exception_id FROM removed) + AND ce.status <> 'resolved' + AND NOT EXISTS ( + SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id + ) + RETURNING ce.id + `, + [ids, now], + ) + return result.rowCount || 0 +} 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 new file mode 100644 index 00000000..e54aa193 --- /dev/null +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.test.ts @@ -0,0 +1,174 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + resolveCost, + selectCostVersion, + type CostItemRow, + type CostVersionRow, +} from './cost-snapshot-service.js' + +const costItems = new Map([ + ['sys_affiliate_purchase', { id: 3, code: 'sys_affiliate_purchase', pricing_mode: 'per_item' }], + ['sys_work_order_reward', { id: 1, code: 'sys_work_order_reward', pricing_mode: 'per_order' }], + [ + 'sys_third_party_voucher_cost', + { id: 4, code: 'sys_third_party_voucher_cost', pricing_mode: 'per_item' }, + ], +]) + +const pricingModes = new Map([ + [1, 'per_order'], + [3, 'per_item'], + [4, 'per_item'], +]) + +const versions: CostVersionRow[] = [ + { + id: 100, + cost_item_id: 3, + scope_type: 'product', + canonical_product_id: 11, + work_product_rule_id: null, + amount_fen: 1200, + effective_from: '2026-09-29T00:00:00+08:00', + }, + { + id: 101, + cost_item_id: 1, + scope_type: 'rule', + canonical_product_id: null, + work_product_rule_id: 9, + amount_fen: 840, + effective_from: '2026-09-29T00:00:00+08:00', + }, + { + id: 102, + cost_item_id: 4, + scope_type: 'product', + canonical_product_id: 33, + work_product_rule_id: null, + amount_fen: 5000, + effective_from: '2026-10-20T00:00:00+08:00', + }, + { + id: 103, + cost_item_id: 4, + scope_type: 'product', + canonical_product_id: 33, + work_product_rule_id: null, + amount_fen: 6000, + effective_from: '2026-11-01T00:00:00+08:00', + }, +] + +function item(patch: Record = {}) { + return { + id: 1, + order_id: 1, + fulfillment_channel: 'affiliate_dash', + canonical_product_id: 11, + quantity: 2, + basis_at: '2026-10-31T10:00:00+08:00', + work_order_id: null as number | null, + work_product_rule_id: null as number | null, + work_order_status: null as string | null, + reward_amount: null as number | null, + ...patch, + } +} + +test('成本解析:自动发货按件计,单价乘数量', () => { + const resolution = resolveCost(item(), costItems, versions, pricingModes) + assert.ok(resolution) + assert.equal(resolution.costVersionId, 100) + assert.equal(resolution.totalAmountFen, 2400) + assert.equal(resolution.stage, 'estimated') +}) + +test('成本解析:卡券渠道按件计,取业务日期前的最新版本', () => { + const resolution = resolveCost( + item({ fulfillment_channel: 'third_party_voucher', canonical_product_id: 33, quantity: 1 }), + costItems, + versions, + pricingModes, + ) + assert.ok(resolution) + assert.equal(resolution.costVersionId, 102) + assert.equal(resolution.unitAmountFen, 5000) +}) + +test('成本解析:生效日期晚于业务日期的版本不参与', () => { + const version = selectCostVersion(versions, { + costItemId: 4, + productId: 33, + ruleId: null, + basisAt: '2026-10-31T10:00:00+08:00', + }) + assert.equal(version?.id, 102) + + const future = selectCostVersion(versions, { + costItemId: 4, + productId: 33, + ruleId: null, + basisAt: '2026-11-02T10:00:00+08:00', + }) + assert.equal(future?.id, 103) +}) + +test('成本解析:接单平台优先按规则版本', () => { + const resolution = resolveCost( + item({ + fulfillment_channel: 'work_order', + canonical_product_id: 55, + work_product_rule_id: 9, + quantity: 1, + }), + costItems, + versions, + pricingModes, + ) + assert.ok(resolution) + assert.equal(resolution.costVersionId, 101) + assert.equal(resolution.totalAmountFen, 840) +}) + +test('成本解析:工单已验收时用实际奖励金额,阶段为 actual', () => { + const resolution = resolveCost( + item({ + fulfillment_channel: 'work_order', + canonical_product_id: 55, + work_product_rule_id: 9, + work_order_id: 77, + work_order_status: 'accepted', + reward_amount: 3300, + quantity: 1, + }), + costItems, + versions, + pricingModes, + ) + assert.ok(resolution) + assert.equal(resolution.totalAmountFen, 3300) + assert.equal(resolution.stage, 'actual') +}) + +test('成本解析:没有可用版本时返回 null(进成本缺失队列,不记 0)', () => { + const resolution = resolveCost( + item({ canonical_product_id: 999 }), + costItems, + versions, + pricingModes, + ) + assert.equal(resolution, null) +}) + +test('成本解析:人工渠道不在这里算成本', () => { + const resolution = resolveCost( + item({ fulfillment_channel: 'manual' }), + costItems, + versions, + pricingModes, + ) + assert.equal(resolution, null) +}) diff --git a/apps/backend/src/services/cost-center/cost-snapshot-service.ts b/apps/backend/src/services/cost-center/cost-snapshot-service.ts new file mode 100644 index 00000000..12322944 --- /dev/null +++ b/apps/backend/src/services/cost-center/cost-snapshot-service.ts @@ -0,0 +1,322 @@ +import { query } from '../../db/client.js' +import { nowIso } from '../../utils/time.js' +import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js' + +/** + * 管道 C1:成本快照(唯一入口)。 + * + * 旧实现有三条路径(预估 / 工单实际 / 自动发货专用),互相叠加; + * 这里统一成一条:按渠道取成本项 → 解析版本 → 写不可变快照。 + * + * 渠道 → 成本项: + * affiliate_dash → sys_affiliate_purchase(按商品) + * third_party_voucher → sys_third_party_voucher_cost(按商品) + * work_order → sys_work_order_reward(优先按接单规则,其次按商品) + * manual → 跳过(确认履约时按实际录入) + */ + +const CHANNEL_COST_CODE: Record = { + affiliate_dash: 'sys_affiliate_purchase', + third_party_voucher: 'sys_third_party_voucher_cost', + work_order: 'sys_work_order_reward', +} + +export type CostItemRow = { + id: number + code: string + pricing_mode: string +} + +export type CostVersionRow = { + id: number + cost_item_id: number + scope_type: string + canonical_product_id: number | null + work_product_rule_id: number | null + amount_fen: number + effective_from: string +} + +export type CostCandidateItem = { + id: number + order_id: number + fulfillment_channel: string | null + canonical_product_id: number | null + quantity: number + basis_at: string + work_order_id: number | null + work_product_rule_id: number | null + work_order_status: string | null + reward_amount: number | null +} + +export type CostResolution = { + costItemId: number + costVersionId: number + unitAmountFen: number + totalAmountFen: number + stage: 'estimated' | 'actual' +} | null + +function toTime(value: unknown): number { + const parsed = new Date(String(value)).getTime() + return Number.isFinite(parsed) ? parsed : 0 +} + +function isEffective(version: CostVersionRow, basisAt: string): boolean { + return toTime(version.effective_from) <= toTime(basisAt) +} + +/** 纯函数:按「规则 > 商品 > 全局」优先级取最新有效版本。 */ +export function selectCostVersion( + versions: CostVersionRow[], + input: { costItemId: number; productId: number | null; ruleId: number | null; basisAt: string }, +): CostVersionRow | null { + const candidates = versions.filter( + (version) => + Number(version.cost_item_id) === Number(input.costItemId) && + isEffective(version, input.basisAt) && + (version.scope_type === 'global' || + (version.scope_type === 'product' && + input.productId != null && + Number(version.canonical_product_id) === Number(input.productId)) || + (version.scope_type === 'rule' && + input.ruleId != null && + Number(version.work_product_rule_id) === Number(input.ruleId))), + ) + if (candidates.length === 0) return null + + const rank = (version: CostVersionRow) => + version.scope_type === 'rule' ? 3 : version.scope_type === 'product' ? 2 : 1 + + return candidates.sort((left, right) => { + const byRank = rank(right) - rank(left) + if (byRank !== 0) return byRank + return toTime(right.effective_from) - toTime(left.effective_from) + })[0]! +} + +/** 纯函数:结合订单项与版本得到快照金额。 */ +export function resolveCost( + item: CostCandidateItem, + costItems: Map, + versions: CostVersionRow[], + pricingModeByItemId: Map, +): CostResolution { + const code = CHANNEL_COST_CODE[String(item.fulfillment_channel || '')] + if (!code) return null + const costItem = costItems.get(code) + if (!costItem) return null + + // 工单已结算:直接按实际奖励金额记 actual,不依赖版本价格。 + if ( + String(item.fulfillment_channel) === 'work_order' && + String(item.work_order_status || '') === 'accepted' && + Number(item.reward_amount || 0) > 0 + ) { + const version = selectCostVersion(versions, { + costItemId: costItem.id, + productId: item.canonical_product_id, + ruleId: item.work_product_rule_id, + basisAt: item.basis_at, + }) + if (!version) return null + return { + costItemId: costItem.id, + costVersionId: version.id, + unitAmountFen: Number(item.reward_amount), + totalAmountFen: Number(item.reward_amount), + stage: 'actual', + } + } + + const version = selectCostVersion(versions, { + costItemId: costItem.id, + productId: item.canonical_product_id, + ruleId: item.work_product_rule_id, + basisAt: item.basis_at, + }) + if (!version) return null + + 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 + + return { + costItemId: costItem.id, + costVersionId: version.id, + unitAmountFen: amount, + totalAmountFen: total, + stage: 'estimated', + } +} + +export type SyncCostSnapshotsResult = { + scanned: number + created: number + missing: number + lastId: number +} + +export async function syncChannelCostSnapshots( + options: { batchSize?: number; maxBatches?: number; afterId?: number } = {}, +): Promise { + const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500))) + const maxBatches = Math.min(200, Math.max(1, Math.trunc(options.maxBatches || 20))) + let cursor = Math.max(0, Math.trunc(options.afterId || 0)) + + const costItems = await loadCostItems() + const versions = await loadActiveVersions() + const pricingModeByItemId = new Map() + for (const item of costItems.values()) pricingModeByItemId.set(item.id, item.pricing_mode) + + const result: SyncCostSnapshotsResult = { scanned: 0, created: 0, missing: 0, lastId: cursor } + + for (let batch = 0; batch < maxBatches; batch += 1) { + const rows = await query( + ` + SELECT + oi.id, + oi.order_id, + oi.fulfillment_channel, + 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, + wo.id AS work_order_id, + wo.product_rule_id AS work_product_rule_id, + wo.status AS work_order_status, + wo.reward_amount + FROM order_items oi + 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 oi.id > $1 + ORDER BY oi.id + LIMIT $2 + `, + [cursor, batchSize], + ) + + if (rows.rows.length === 0) break + result.scanned += rows.rows.length + cursor = Number(rows.rows[rows.rows.length - 1]!.id) + result.lastId = cursor + + const inserts: Array<{ + item: CostCandidateItem + resolution: NonNullable + }> = [] + + for (const item of rows.rows) { + const resolution = resolveCost(item, costItems, versions, pricingModeByItemId) + if (!resolution) { + result.missing += 1 + await upsertAggregatedException({ + reason: 'cost_missing', + scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`, + orderItemId: item.id, + detail: { channel: item.fulfillment_channel }, + }) + continue + } + inserts.push({ item, resolution }) + } + + if (inserts.length === 0) continue + const created = await insertSnapshots(inserts) + result.created += created + await resolveExceptionsForItems(inserts.map((row) => row.item.id)) + } + + return result +} + +async function insertSnapshots( + rows: Array<{ item: CostCandidateItem; resolution: NonNullable }>, +): Promise { + const now = nowIso() + const result = await query( + ` + INSERT INTO cost_snapshots ( + subject_type, subject_id, order_id, order_item_id, work_order_id, + 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', + input.item_id, + input.order_id, + input.item_id, + NULLIF(input.work_order_id, 0), + input.cost_item_id, + input.cost_version_id, + NULLIF(input.product_id, 0), + input.quantity, + input.unit_amount_fen, + input.total_amount_fen, + input.stage, + 'resolved', + jsonb_build_object('channel', input.channel, 'pipeline', 'cost_snapshot'), + input.basis_at, + $1, + $1 + FROM unnest( + $2::bigint[], $3::bigint[], $4::bigint[], $5::bigint[], $6::bigint[], + $7::bigint[], $8::numeric[], $9::bigint[], $10::bigint[], + $11::text[], $12::text[], $13::timestamptz[] + ) AS input( + item_id, order_id, work_order_id, cost_item_id, cost_version_id, + product_id, quantity, unit_amount_fen, total_amount_fen, stage, + channel, basis_at + ) + ON CONFLICT (subject_type, subject_id, cost_item_id) WHERE superseded_at IS NULL + DO NOTHING + `, + [ + now, + rows.map((row) => row.item.id), + rows.map((row) => row.item.order_id), + rows.map((row) => row.item.work_order_id ?? 0), + rows.map((row) => row.resolution.costItemId), + rows.map((row) => row.resolution.costVersionId), + rows.map((row) => row.item.canonical_product_id ?? 0), + rows.map((row) => row.item.quantity), + rows.map((row) => row.resolution.unitAmountFen), + rows.map((row) => row.resolution.totalAmountFen), + rows.map((row) => row.resolution.stage), + rows.map((row) => String(row.item.fulfillment_channel || '')), + rows.map((row) => row.item.basis_at), + ], + ) + return result.rowCount || 0 +} + +async function loadCostItems(): Promise> { + const rows = await query( + `SELECT id, code, pricing_mode FROM cost_items WHERE status = 'active'`, + ) + const map = new Map() + for (const row of rows.rows) map.set(row.code, row) + return map +} + +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 + FROM cost_versions + WHERE status = 'active' + `, + ) + return rows.rows +} diff --git a/apps/backend/src/services/cost-center/product-normalize-service.test.ts b/apps/backend/src/services/cost-center/product-normalize-service.test.ts new file mode 100644 index 00000000..2f785537 --- /dev/null +++ b/apps/backend/src/services/cost-center/product-normalize-service.test.ts @@ -0,0 +1,95 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { buildMappingIndex, resolveProductId } from './product-normalize-service.js' + +const mappings = [ + { + provider: '91kaquan', + shop_id: 'shop-a', + platform_sku_code: 'sku-1', + platform_sku_name_key: '', + match_type: 'sku_code', + canonical_product_id: 11, + enabled: true, + }, + { + provider: '91kaquan', + shop_id: '', + platform_sku_code: 'sku-2', + platform_sku_name_key: '', + match_type: 'sku_code', + canonical_product_id: 12, + enabled: true, + }, + { + provider: '91kaquan', + shop_id: '', + platform_sku_code: '', + platform_sku_name_key: '套装-红', + match_type: 'sku_code_name', + canonical_product_id: 13, + enabled: true, + }, + { + provider: 'pdd', + shop_id: '', + platform_sku_code: '', + platform_sku_name_key: '系列-蓝', + match_type: 'sku_name_series', + canonical_product_id: 14, + enabled: true, + }, +] + +const index = buildMappingIndex(mappings, [{ sku: 'aff-sku', productId: 21 }]) + +function item(patch: Record = {}) { + return { + id: 1, + provider: '91kaquan', + shop_id: 'shop-a', + sku_code: 'sku-1', + sku_name: '任意', + normalized_sku_name: null as string | null, + normalized_sku_series: null as string | null, + fulfillment_channel: 'work_order' as string | null, + ...patch, + } +} + +test('商品归一:店铺精确映射优先', () => { + assert.equal(resolveProductId(item(), index), 11) +}) + +test('商品归一:店铺不匹配回落到全局映射', () => { + assert.equal(resolveProductId(item({ shop_id: 'shop-b', sku_code: 'sku-2' }), index), 12) +}) + +test('商品归一:自动发货渠道先按 aff:SKU 命中', () => { + assert.equal( + resolveProductId(item({ fulfillment_channel: 'affiliate_dash', sku_code: 'aff-sku' }), index), + 21, + ) +}) + +test('商品归一:编码未命中时按归一名称匹配', () => { + assert.equal( + resolveProductId(item({ sku_code: 'unknown', normalized_sku_name: '套装-红' }), index), + 13, + ) +}) + +test('商品归一:名称未命中时按名称系列匹配', () => { + assert.equal( + resolveProductId( + item({ provider: 'pdd', sku_code: 'unknown', normalized_sku_series: '系列-蓝' }), + index, + ), + 14, + ) +}) + +test('商品归一:全部未命中返回 null,不自动建商品', () => { + assert.equal(resolveProductId(item({ sku_code: 'nope' }), index), null) +}) diff --git a/apps/backend/src/services/cost-center/product-normalize-service.ts b/apps/backend/src/services/cost-center/product-normalize-service.ts new file mode 100644 index 00000000..932e7a56 --- /dev/null +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -0,0 +1,242 @@ +import { query } from '../../db/client.js' +import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js' + +/** + * 管道 B:商品归一。 + * + * 只写 `order_items.canonical_product_id`,不建商品、不改渠道、不碰成本。 + * + * 解析顺序(与旧的四条回填路径统一): + * 1. 自动发货渠道:`aff:` 商品直接命中(该渠道的商品就是 SKU 本身); + * 2. 精确映射:provider + shop + sku_code; + * 3. 精确映射(店铺无关); + * 4. 编码+名称映射:provider + shop + normalized_sku_name; + * 5. 编码系列 → 名称系列。 + * + * 全部未命中 → 保持为空,写一条 SKU 级 `product_mapping_missing`。 + */ + +type MappingRow = { + provider: string + shop_id: string + platform_sku_code: string + platform_sku_name_key: string + match_type: string + canonical_product_id: number + enabled: boolean +} + +export type MappingIndex = { + bySku: Map + bySkuName: Map + bySkuSeries: Map + byNameSeries: Map + affSkus: Map +} + +type ItemRow = { + id: number + provider: string + shop_id: string + sku_code: string + sku_name: string + normalized_sku_name: string | null + normalized_sku_series: string | null + fulfillment_channel: string | null +} + +function key(...parts: string[]): string { + return parts.map((part) => String(part ?? '').trim()).join('|') +} + +export function buildMappingIndex( + mappings: MappingRow[], + affiliateProducts: Array<{ sku: string; productId: number }>, +): MappingIndex { + const index: MappingIndex = { + bySku: new Map(), + bySkuName: new Map(), + bySkuSeries: new Map(), + byNameSeries: new Map(), + affSkus: new Map(), + } + + for (const mapping of mappings) { + if (!mapping.enabled) continue + const productId = Number(mapping.canonical_product_id) || 0 + if (productId <= 0) continue + + if (mapping.match_type === 'sku_code' && mapping.platform_sku_code) { + index.bySku.set(key(mapping.provider, mapping.shop_id, mapping.platform_sku_code), productId) + } else if (mapping.match_type === 'sku_code_name' && mapping.platform_sku_name_key) { + index.bySkuName.set( + key(mapping.provider, mapping.shop_id, mapping.platform_sku_name_key), + productId, + ) + } else if (mapping.match_type === 'sku_code_series' && mapping.platform_sku_name_key) { + index.bySkuSeries.set( + key(mapping.provider, mapping.shop_id, mapping.platform_sku_name_key), + productId, + ) + } else if (mapping.match_type === 'sku_name_series' && mapping.platform_sku_name_key) { + index.byNameSeries.set( + key(mapping.provider, mapping.shop_id, mapping.platform_sku_name_key), + productId, + ) + } + } + + for (const product of affiliateProducts) { + index.affSkus.set(String(product.sku).trim(), product.productId) + } + + return index +} + +/** 纯函数:按索引解析商品,便于单测。 */ +export function resolveProductId(item: ItemRow, index: MappingIndex): number | null { + const provider = String(item.provider || '').trim() + const shopId = String(item.shop_id || '').trim() + const skuCode = String(item.sku_code || '').trim() + const nameKey = String(item.normalized_sku_name || '').trim() + const seriesKey = String(item.normalized_sku_series || '').trim() + + if (item.fulfillment_channel === 'affiliate_dash' && skuCode) { + const aff = index.affSkus.get(skuCode) + if (aff) return aff + } + + const exactWithShop = index.bySku.get(key(provider, shopId, skuCode)) + if (exactWithShop) return exactWithShop + const exactGlobal = index.bySku.get(key(provider, '', skuCode)) + if (exactGlobal) return exactGlobal + + if (nameKey) { + const nameWithShop = index.bySkuName.get(key(provider, shopId, nameKey)) + if (nameWithShop) return nameWithShop + const nameGlobal = index.bySkuName.get(key(provider, '', nameKey)) + if (nameGlobal) return nameGlobal + } + + if (seriesKey) { + const nameSeriesWithShop = index.byNameSeries.get(key(provider, shopId, seriesKey)) + if (nameSeriesWithShop) return nameSeriesWithShop + const nameSeriesGlobal = index.byNameSeries.get(key(provider, '', seriesKey)) + if (nameSeriesGlobal) return nameSeriesGlobal + + const seriesWithShop = index.bySkuSeries.get(key(provider, shopId, seriesKey)) + if (seriesWithShop) return seriesWithShop + const seriesGlobal = index.bySkuSeries.get(key(provider, '', seriesKey)) + if (seriesGlobal) return seriesGlobal + } + + return null +} + +export type NormalizeProductsResult = { + scanned: number + matched: number + unmatched: number + lastId: number +} + +export async function normalizeOrderItemProducts( + options: { batchSize?: number; maxBatches?: number; afterId?: number } = {}, +): Promise { + const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500))) + const maxBatches = Math.min(200, Math.max(1, Math.trunc(options.maxBatches || 20))) + let cursor = Math.max(0, Math.trunc(options.afterId || 0)) + + const index = await loadMappingIndex() + const result: NormalizeProductsResult = { scanned: 0, matched: 0, unmatched: 0, lastId: cursor } + + for (let batch = 0; batch < maxBatches; batch += 1) { + const rows = await query( + ` + SELECT + oi.id, + o.provider, + o.shop_id, + oi.sku_code, + oi.sku_name, + oi.normalized_sku_name, + oi.normalized_sku_series, + oi.fulfillment_channel + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + WHERE oi.fulfillment_status = 'assigned' + AND oi.canonical_product_id IS NULL + AND oi.id > $1 + ORDER BY oi.id + LIMIT $2 + `, + [cursor, batchSize], + ) + + if (rows.rows.length === 0) break + result.scanned += rows.rows.length + cursor = Number(rows.rows[rows.rows.length - 1]!.id) + result.lastId = cursor + + const matched: Array<{ id: number; productId: number }> = [] + + for (const item of rows.rows) { + const productId = resolveProductId(item, index) + if (!productId) { + result.unmatched += 1 + await upsertAggregatedException({ + reason: 'product_mapping_missing', + scopeKey: `product_mapping_missing|${item.provider}|${item.shop_id}|${item.sku_code}|${item.fulfillment_channel || '-'}`, + orderItemId: item.id, + detail: { skuName: item.sku_name, channel: item.fulfillment_channel }, + }) + continue + } + result.matched += 1 + matched.push({ id: item.id, productId }) + } + + if (matched.length > 0) { + await query( + ` + UPDATE order_items oi + SET canonical_product_id = input.product_id, updated_at = NOW() + FROM unnest($1::bigint[], $2::bigint[]) AS input(id, product_id) + WHERE oi.id = input.id + AND oi.canonical_product_id IS NULL + `, + [matched.map((row) => row.id), matched.map((row) => row.productId)], + ) + await resolveExceptionsForItems(matched.map((row) => row.id)) + } + } + + return result +} + +async function loadMappingIndex(): Promise { + const mappings = await query( + ` + SELECT provider, shop_id, platform_sku_code, platform_sku_name_key, + match_type, canonical_product_id, enabled + FROM channel_product_mappings + WHERE enabled = TRUE + `, + ) + + const affiliateProducts = await query<{ canonical_sku: string; id: number }>( + ` + SELECT id, canonical_sku + FROM product_catalog + WHERE status = 'active' AND canonical_sku LIKE 'aff:%' + `, + ) + + return buildMappingIndex( + mappings.rows, + affiliateProducts.rows.map((row) => ({ + sku: String(row.canonical_sku).slice(4), + productId: Number(row.id), + })), + ) +} diff --git a/apps/backend/src/services/finance/sales-facts-channel-service.ts b/apps/backend/src/services/finance/sales-facts-channel-service.ts new file mode 100644 index 00000000..3954bb10 --- /dev/null +++ b/apps/backend/src/services/finance/sales-facts-channel-service.ts @@ -0,0 +1,161 @@ +import { withTransaction } from '../../db/client.js' +import { nowIso } from '../../utils/time.js' + +/** + * 管道 C2:销售事实(渠道版)。 + * + * 与旧实现的关键差别: + * 1. 维度用 `fulfillment_channel`,未分类订单进 `unclassified` 桶, + * 收入不会因为「还没分类」而消失; + * 2. 归档订单(archived)不进事实; + * 3. 成本来自 `cost_snapshots`,缺失就记 0 并计数,不把 0 当真实成本。 + * + * 事实表暂时沿用 `fulfillment_mode` 列承载渠道值(113 再改列名)。 + */ + +export type ChannelFactsResult = { + dateFrom: string + dateTo: string + deleted: number + inserted: number + skippedClosedPeriods: string[] +} + +export async function rebuildChannelSalesDailyFacts(input: { + dateFrom: string + dateTo: string +}): Promise { + const dateFrom = normalizeDate(input.dateFrom) + const dateTo = normalizeDate(input.dateTo) + if (dateFrom > dateTo) { + throw new Error(`开始日期不能晚于结束日期:${dateFrom} > ${dateTo}`) + } + const computedAt = nowIso() + + const closed = await withTransaction(async (client) => { + const periods = await client.query<{ period_key: string }>( + ` + SELECT period_key FROM finance_periods + WHERE status = 'closed' + AND period_key BETWEEN $1 AND $2 + ORDER BY period_key + `, + [dateFrom.slice(0, 7), dateTo.slice(0, 7)], + ) + const skipped = periods.rows.map((row) => row.period_key) + + const deleted = await client.query( + ` + DELETE FROM sales_daily_facts f + WHERE f.stat_date BETWEEN $1::date AND $2::date + AND NOT EXISTS ( + SELECT 1 FROM finance_periods p + WHERE p.status = 'closed' + AND to_char(f.stat_date, 'YYYY-MM') = p.period_key + ) + `, + [dateFrom, dateTo], + ) + + const inserted = await client.query( + ` + WITH scope AS ( + SELECT + (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date AS stat_date, + o.provider, + o.shop_id, + oi.canonical_product_id, + CASE + WHEN oi.fulfillment_status = 'pending' THEN 'unclassified' + WHEN oi.fulfillment_channel IS NULL THEN 'unclassified' + ELSE oi.fulfillment_channel + END AS channel, + oi.id AS order_item_id, + o.id AS order_id, + COALESCE(oi.normalized_quantity, oi.quantity, 1)::numeric AS quantity, + COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + WHERE oi.fulfillment_status IN ('assigned', 'pending') + AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date + BETWEEN $1::date AND $2::date + ), + costed AS ( + SELECT + s.*, + COALESCE(c.cost_fen, 0)::bigint AS cost_fen, + (c.order_item_id IS NULL) AS cost_missing, + COALESCE(c.has_actual, FALSE) AS has_actual + FROM scope s + LEFT JOIN ( + SELECT + cs.order_item_id, + SUM(cs.total_amount_fen)::bigint AS cost_fen, + BOOL_AND(cs.snapshot_stage = 'actual') AS has_actual + FROM cost_snapshots cs + WHERE cs.superseded_at IS NULL + GROUP BY cs.order_item_id + ) c ON c.order_item_id = s.order_item_id + ) + INSERT INTO sales_daily_facts ( + stat_date, provider, shop_id, canonical_product_id, fulfillment_mode, + order_count, item_count, quantity, + gross_revenue_fen, refund_amount_fen, net_revenue_fen, + cost_amount_fen, cost_missing_count, cost_stage, + gross_profit_fen, computed_at, created_at, updated_at + ) + SELECT + stat_date, + provider, + shop_id, + canonical_product_id, + channel, + COUNT(DISTINCT order_id)::int, + COUNT(*)::int, + SUM(quantity), + SUM(revenue_fen), + 0, + SUM(revenue_fen), + SUM(cost_fen), + COUNT(*) FILTER (WHERE cost_missing)::int, + CASE + WHEN COUNT(*) FILTER (WHERE cost_missing) = COUNT(*) THEN 'missing' + WHEN COUNT(*) FILTER (WHERE cost_missing) > 0 THEN 'partial' + WHEN BOOL_AND(has_actual) THEN 'actual' + ELSE 'estimated' + END, + SUM(revenue_fen) - SUM(cost_fen), + $3::timestamptz, + $3::timestamptz, + $3::timestamptz + FROM costed + GROUP BY stat_date, provider, shop_id, canonical_product_id, channel + `, + [dateFrom, dateTo, computedAt], + ) + + return { + deleted: deleted.rowCount || 0, + inserted: inserted.rowCount || 0, + skipped, + } + }) + + return { + dateFrom, + dateTo, + deleted: closed.deleted, + inserted: closed.inserted, + skippedClosedPeriods: closed.skipped, + } +} + +function normalizeDate(value: unknown): string { + const text = String(value || '') + .trim() + .slice(0, 10) + if (!/^\d{4}-\d{2}-\d{2}$/.test(text)) { + throw new Error(`日期格式无效:${String(value)}`) + } + return text +}