diff --git a/apps/backend/src/routes/admin/worker-platform/config.ts b/apps/backend/src/routes/admin/worker-platform/config.ts index 822c7577..fdacc2c2 100644 --- a/apps/backend/src/routes/admin/worker-platform/config.ts +++ b/apps/backend/src/routes/admin/worker-platform/config.ts @@ -1,6 +1,6 @@ import { Router } from 'express' import { setHttpCache } from '../../../middleware/http-cache.js' -import { syncAdminWorkOrderTemplateCostVersions } from '../../../services/cost-center/template-bootstrap-service.js' +import { syncAdminWorkOrderTemplateCostVersions } from '../../../services/cost-center/work-order-product-service.js' import { deleteAdminWorkCategory, diff --git a/apps/backend/src/services/cost-center/affiliate-product-service.ts b/apps/backend/src/services/cost-center/affiliate-product-service.ts new file mode 100644 index 00000000..ebae0c0f --- /dev/null +++ b/apps/backend/src/services/cost-center/affiliate-product-service.ts @@ -0,0 +1,199 @@ +/** + * 自动发货（affiliate_dash）商品同步与 SKU 回填。 + */ +import { query } from '../../db/client.js' +import { shiftDate } from '../../utils/finance-date-range.js' +import { nowIso } from '../../utils/time.js' +import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js' + +export async function syncAdminAffiliateDashProducts() { + const now = nowIso() + const result = await query<{ inserted: boolean }>( + ` + WITH auto_task AS ( + SELECT + btrim(context_json->>'skuCode') AS sku_code, + btrim(context_json->>'skuName') AS sku_name, + created_at + FROM fulfillment_tasks + WHERE executor_key = 'affiliate_dash' + AND btrim(COALESCE(context_json->>'skuCode', '')) <> '' + AND btrim(COALESCE(context_json->>'skuName', '')) <> '' + ), + agg AS ( + SELECT sku_code, count(*)::int AS task_count, max(created_at) AS last_fulfilled_at + FROM auto_task + GROUP BY sku_code + ), + ranked AS ( + SELECT + sku_code, + sku_name, + row_number() OVER ( + PARTITION BY sku_code + ORDER BY (sku_name <> sku_code) DESC, count(*) DESC, sku_name + ) AS rn + FROM auto_task + GROUP BY sku_code, sku_name + ) + INSERT INTO product_catalog ( + canonical_sku, canonical_name, category, unit_name, status, + attributes_json, created_at, updated_at + ) + SELECT + 'aff:' || ranked.sku_code, + ranked.sku_name, + '', + CASE WHEN ranked.sku_name LIKE '%个' THEN '个' ELSE '件' END, + 'active', + jsonb_build_object( + 'source', 'affiliate_dash_task', + 'skuCode', ranked.sku_code, + 'taskCount', agg.task_count, + 'lastFulfilledAt', agg.last_fulfilled_at + ), + $1, + $1 + FROM ranked + JOIN agg ON agg.sku_code = ranked.sku_code + WHERE ranked.rn = 1 + ON CONFLICT (canonical_sku) DO UPDATE + SET canonical_name = EXCLUDED.canonical_name, + unit_name = EXCLUDED.unit_name, + attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json, + updated_at = $1 + RETURNING (xmax = 0) AS inserted + `, + [now], + ) + const createdCount = result.rows.filter((row) => row.inserted).length + return { + createdCount, + updatedCount: result.rows.length - createdCount, + productCount: result.rows.length, + } +} + +type AffiliateSkuBackfillRow = { + affiliate_item_count: number + matched_count: number + updated_count: number + from_date: string | null + to_date: string | null +} + +/** 销售事实单次重算最多 400 天，长历史按窗口分批。 */ +export 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 } +} + +/** + * 按自动发货履约证据直接回填订单项标准商品。 + * + * 自动发货只有 affiliate-dash 一个上游渠道，SKU 固定归一为 `aff:`， + * 不写 channel_product_mappings。多个任务共用同一订单项时取最新 SKU。 + */ +export async function backfillAdminAffiliateDashSkus() { + const now = nowIso() + const result = await query( + ` + 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 affiliate_item + JOIN order_items oi ON oi.id = affiliate_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 affiliate_item + JOIN order_items oi ON oi.id = affiliate_item.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 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: Awaited> | null = null + if (matchedCount > 0 && row?.from_date && row?.to_date) { + factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date) + } + + return { + updatedCount, + skippedCount, + remainingCount, + affiliateItemCount, + matchedCount, + factsRebuild, + } +} diff --git a/apps/backend/src/services/cost-center/cost-center.test.ts b/apps/backend/src/services/cost-center/cost-center.test.ts index f6a91c7d..728cdbaa 100644 --- a/apps/backend/src/services/cost-center/cost-center.test.ts +++ b/apps/backend/src/services/cost-center/cost-center.test.ts @@ -6,7 +6,7 @@ import { normalizeProviderKey, resolveMappingNameKey, } from './cost-center-shared.js' -import { calculateCostTotal, validateCostVersionScope } from './cost-definition-service.js' +import { calculateCostTotal, validateCostVersionScope } from './cost-resolution-service.js' import { resolveMappingProductId } from './product-catalog-service.js' import { buildTemplateMappingPlan, diff --git a/apps/backend/src/services/cost-center/cost-definition-service.ts b/apps/backend/src/services/cost-center/cost-definition-service.ts index 38b10636..1d079a64 100644 --- a/apps/backend/src/services/cost-center/cost-definition-service.ts +++ b/apps/backend/src/services/cost-center/cost-definition-service.ts @@ -1,20 +1,14 @@ /** - * 成本项 / 成本版本 / 成本异常 / 成本预览 / 成本快照。 + * 成本定义 CRUD：成本项、成本版本、成本异常队列。 * - * 口径（2026-10-23 重写）： - * 1. 成本项用 cost_model 表达「成本怎么算」，不再有 cost_type + carrier_type； - * 2. 成本模型与履约方式正交：专属模型只对对应履约方式生效，通用模型全局生效； - * 3. 履约方式只从 order_items.fulfillment_mode 读，不允许临时推断； - * 4. 快照与异常表不再存履约载体； - * 5. 匹配不到成本不得默认 0，必须进入 cost_exceptions。 + * 金额只在 cost_versions 改；异常队列只做登记与处理，不做成本计算。 + * 成本解析与快照生成见 cost-resolution-service.ts。 */ -import { query, withTransaction } from '../../db/client.js' -import type { AdminSession } from '../admin/admin-auth-service.js' +import { query } from '../../db/client.js' import type { JsonObject } from '../../types/json.js' import { createHttpError } from '../../utils/http.js' import { nowIso } from '../../utils/time.js' -import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js' -import { resolveSkuNameQuantity } from '../product-match/pricing.js' +import type { AdminSession } from '../admin/admin-auth-service.js' import { COST_EXCEPTION_STATUSES, COST_EXCEPTION_SUBJECT_TYPES, @@ -28,13 +22,10 @@ import { PRODUCT_STATUSES, assertKnownProvider, assertOneOf, - costModelAppliesTo, costModelScopeSql, mapCostException, mapCostItem, mapCostVersion, - mapMapping, - mapProduct, normalizeId, normalizeIsoDate, normalizeNonNegativeInteger, @@ -48,19 +39,14 @@ import { requiredRow, text, toNumber, - type CostExceptionRow, - type CostItemRow, - type CostLine, - type CostPreview, - type CostResolution, - type CostVersionRow, - type FulfillmentMode, - type MappingRow, - type MissingCostItem, - type OrderItemCostContext, - type ProductRow, } from './cost-center-shared.js' -import { getProductRow } from './product-catalog-service.js' +import type { CostExceptionRow, CostItemRow, CostVersionRow } from './cost-center-shared.js' +import { + getCostItemRow, + getCostVersionById, + normalizeCostVersionScopeFields, + validateCostVersionScope, +} from './cost-resolution-service.js' export async function listAdminCostItems(payload: JsonObject = {}) { const page = normalizePage(payload.page) @@ -739,644 +725,3 @@ export async function resolveAdminCostException( ) return { exception: mapCostException(requiredRow(result.rows[0], '成本异常不存在')) } } - -export async function previewAdminCost(payload: JsonObject = {}) { - const context = await resolveCostContext(payload) - const mapping = await findMappingForContext(context) - const productId = context.canonicalProductId || mapping?.canonical_product_id || null - const product = productId ? mapProduct(await getProductRow(productId)) : null - const asOf = - normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || context.basisAt - const resolution: CostResolution = productId - ? await resolveCostLines({ - provider: context.provider, - shopId: context.shopId, - fulfillmentMode: context.fulfillmentMode, - canonicalProductId: productId, - workProductRuleId: context.workProductRuleId, - quantity: context.normalizedQuantity, - revenueFen: context.revenueFen, - asOf, - }) - : { lines: [], missingCostItems: [], applicableCostItemCount: 0 } - - const exceptions: CostPreview['exceptions'] = [] - if (!productId) { - exceptions.push({ - reason: 'product_mapping_missing', - message: `未找到 ${context.provider} / ${context.shopId || '全局'} / ${context.skuCode} 的标准商品映射`, - detail: { - provider: context.provider, - shopId: context.shopId, - skuCode: context.skuCode, - }, - }) - } else if (resolution.missingCostItems.length > 0) { - exceptions.push({ - reason: 'cost_missing', - message: `以下成本项没有匹配到有效成本版本：${resolution.missingCostItems - .map((item) => `${item.costItemName}（${item.costItemCode}）`) - .join('、')}`, - detail: { - fulfillmentMode: context.fulfillmentMode, - costItemIds: resolution.missingCostItems.map((item) => item.costItemId), - }, - }) - } else if (resolution.applicableCostItemCount === 0) { - exceptions.push({ - reason: 'cost_missing', - message: `没有配置适用于「${context.fulfillmentMode}」的成本项`, - detail: { fulfillmentMode: context.fulfillmentMode }, - }) - } - - // 人工履约的成本来自「人工 SKU 配置的消耗积分 × 平台换算率」，版本金额只是占位； - // 这里单独给出提示，并从成本行里剔掉 0 元占位，避免预览与实际扣款对不上。 - const manualCost = - context.fulfillmentMode === 'manual' ? await resolveManualCostHint(context.orderItemId) : null - const lines = manualCost - ? resolution.lines.filter((line) => line.costItemCode !== 'sys_manual_service') - : resolution.lines - - return { - context: { - ...context, - basisAt: asOf, - }, - mapping: mapping ? mapMapping(mapping) : null, - product, - lines, - missingCostItems: resolution.missingCostItems.filter( - (item) => !(manualCost && item.costItemCode === 'sys_manual_service'), - ), - manualCost, - totalCostFen: lines.reduce((sum, line) => sum + line.totalAmountFen, 0), - exceptions, - } -} - -/** 人工履约成本提示：直接读订单项绑定的履约配置。 */ -async function resolveManualCostHint(orderItemId: number | null) { - if (!orderItemId) return null - const result = await query<{ - platform_key: string - credit_quantity: string - cost_mode: string - unit_price_fen: string - }>( - ` - SELECT - cfg.platform_key, - cfg.credit_quantity, - cfg.cost_mode, - COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS unit_price_fen - FROM order_items oi - JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id - LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key - WHERE oi.id = $1 - LIMIT 1 - `, - [orderItemId], - ) - const row = result.rows[0] - if (!row) return null - const creditQuantity = Number(row.credit_quantity || 0) - const unitPriceFen = Number(row.unit_price_fen || 1) - return { - platformKey: row.platform_key, - creditQuantity, - unitPriceFen, - costMode: row.cost_mode === 'fixed' ? ('fixed' as const) : ('variable' as const), - costFen: creditQuantity > 0 ? creditQuantity * unitPriceFen : null, - } -} - -export async function createAdminCostSnapshot(payload: JsonObject = {}, _session: AdminSession) { - const preview = await previewAdminCost(payload) - const orderItemId = preview.context.orderItemId - if (!orderItemId) { - throw createHttpError('生成成本快照必须提供订单项 ID', { - statusCode: 400, - errorCode: 'cost_snapshot_order_item_required', - }) - } - const now = nowIso() - const stage = payload.snapshotStage === 'actual' ? 'actual' : 'estimated' - const snapshots: Array> = [] - let exceptionCreatedCount = 0 - let supersededCount = 0 - - await withTransaction(async (client) => { - for (const line of preview.lines) { - const currentResult = await client.query<{ - id: number - snapshot_stage: string - total_amount_fen: number - }>( - ` - SELECT id, snapshot_stage, total_amount_fen - FROM cost_snapshots - WHERE subject_type = 'order_item' - AND subject_id = $1 - AND cost_item_id = $2 - AND superseded_at IS NULL - FOR UPDATE - `, - [orderItemId, line.costItemId], - ) - const current = currentResult.rows[0] - if (current) { - // 预估快照一旦生成就代表支付时点的成本，不因后续改价重写； - // 只有实际成本允许替代当前有效行。 - if (stage === 'estimated') continue - if ( - current.snapshot_stage === 'actual' && - Number(current.total_amount_fen) === line.totalAmountFen - ) { - continue - } - await client.query( - `UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL`, - [now, current.id], - ) - supersededCount += 1 - } - - const result = await client.query<{ [column: string]: unknown }>( - ` - 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 - ) VALUES ( - 'order_item', $1, $2, $1, $3, $4, $5, $6, - $7, $8, $9, $10, 'resolved', $11::jsonb, $12, $13, $13 - ) - ON CONFLICT (subject_type, subject_id, cost_item_id) - WHERE superseded_at IS NULL - DO NOTHING - RETURNING * - `, - [ - orderItemId, - preview.context.orderId, - preview.context.workOrderId, - line.costItemId, - line.costVersionId, - preview.product?.id || null, - line.quantity, - line.unitAmountFen, - line.totalAmountFen, - stage, - JSON.stringify({ - skuCode: preview.context.skuCode, - skuName: preview.context.skuName, - provider: preview.context.provider, - shopId: preview.context.shopId, - normalizedQuantity: preview.context.normalizedQuantity, - fulfillmentMode: preview.context.fulfillmentMode, - basisAt: preview.context.basisAt, - }), - preview.context.basisAt, - now, - ], - ) - if (result.rows[0]) snapshots.push(result.rows[0]) - } - - for (const exception of preview.exceptions) { - const result = await client.query( - ` - INSERT INTO cost_exceptions ( - subject_type, subject_id, order_item_id, work_order_id, - reason, status, detail_json, created_at - ) VALUES ('order_item', $1, $1, $2, $3, 'open', $4::jsonb, $5) - ON CONFLICT (subject_type, subject_id, reason) DO NOTHING - `, - [ - orderItemId, - preview.context.workOrderId, - exception.reason, - JSON.stringify({ - ...exception.detail, - message: exception.message, - skuCode: preview.context.skuCode, - }), - now, - ], - ) - exceptionCreatedCount += result.rowCount || 0 - } - }) - - return { - preview, - snapshots, - createdCount: snapshots.length, - supersededCount, - exceptionCreatedCount, - } -} - -async function resolveCostContext(payload: JsonObject): Promise { - const orderItemId = normalizeOptionalId(payload.orderItemId ?? payload.order_item_id) - if (orderItemId) { - const result = await query<{ - order_item_id: number - order_id: number - provider: string - shop_id: string - sku_code: string - sku_name: string - normalized_sku_name: string | null - normalized_sku_series: string | null - quantity: number - normalized_quantity: number | null - canonical_product_id: number | null - revenue_fen: number - basis_at: string - work_order_id: number | null - work_product_rule_id: number | null - fulfillment_mode: string | null - }>( - ` - SELECT - oi.id AS order_item_id, - oi.order_id, - o.provider, - o.shop_id, - oi.sku_code, - oi.sku_name, - oi.normalized_sku_name, - oi.normalized_sku_series, - oi.quantity, - oi.normalized_quantity, - oi.canonical_product_id, - oi.revenue_amount_fen AS revenue_fen, - 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, - oi.fulfillment_mode - FROM order_items oi - JOIN orders o ON o.id = oi.order_id - LEFT JOIN LATERAL ( - SELECT id, product_rule_id - FROM work_orders - WHERE order_item_id = oi.id - ORDER BY id DESC - LIMIT 1 - ) wo ON TRUE - WHERE oi.id = $1 - LIMIT 1 - `, - [orderItemId], - ) - const row = result.rows[0] - if (!row) { - throw createHttpError('订单项不存在', { - statusCode: 404, - errorCode: 'cost_order_item_not_found', - }) - } - return { - orderItemId: Number(row.order_item_id), - orderId: Number(row.order_id), - provider: row.provider, - shopId: row.shop_id, - skuCode: row.sku_code, - skuName: row.sku_name, - normalizedSkuName: - row.normalized_sku_name || normalizeMatchText(row.sku_name || row.sku_code), - normalizedSkuSeries: - row.normalized_sku_series || normalizeSkuSeriesEquivalence(row.sku_name || row.sku_code), - quantity: Math.max(1, Number(row.quantity) || 1), - normalizedQuantity: - row.normalized_quantity && Number(row.normalized_quantity) > 0 - ? Number(row.normalized_quantity) - : Math.max(1, resolveSkuNameQuantity(row.sku_name || row.sku_code)), - canonicalProductId: row.canonical_product_id ? Number(row.canonical_product_id) : null, - revenueFen: Math.max(0, Number(row.revenue_fen) || 0), - basisAt: new Date(row.basis_at).toISOString(), - workOrderId: row.work_order_id ? Number(row.work_order_id) : null, - workProductRuleId: row.work_product_rule_id ? Number(row.work_product_rule_id) : null, - fulfillmentMode: resolveFulfillmentMode(row.fulfillment_mode), - } - } - - const provider = normalizeRequiredText(payload.provider, '来源', 64) - const skuCode = normalizeRequiredText(payload.skuCode ?? payload.sku_code, '平台 SKU', 120) - const quantity = Math.max(1, Math.trunc(Number(payload.quantity) || 1)) - const skuName = normalizeOptionalText(payload.skuName ?? payload.sku_name, 200) - const fulfillmentMode = resolveFulfillmentMode( - payload.fulfillmentMode ?? payload.fulfillment_mode, - ) - return { - orderItemId: 0, - orderId: 0, - provider, - shopId: normalizeOptionalText(payload.shopId ?? payload.shop_id, 120), - skuCode, - skuName, - normalizedSkuName: normalizeMatchText(skuName || skuCode), - normalizedSkuSeries: normalizeSkuSeriesEquivalence(skuName || skuCode), - quantity, - normalizedQuantity: quantity, - canonicalProductId: normalizeOptionalId( - payload.canonicalProductId ?? payload.canonical_product_id, - ), - revenueFen: Math.max(0, Math.trunc(Number(payload.revenueFen ?? payload.revenue_fen) || 0)), - basisAt: - normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || nowIso(), - workOrderId: null, - workProductRuleId: normalizeOptionalId( - payload.workProductRuleId ?? payload.work_product_rule_id, - ), - fulfillmentMode, - } -} - -/** - * 履约方式只读 order_items.fulfillment_mode。 - * 订单项还没判定时按人工处理，不允许用 provider 或核销码兜底。 - */ -function resolveFulfillmentMode(value: unknown): FulfillmentMode { - const normalized = text(value) - if (normalized && FULFILLMENT_MODES.has(normalized)) return normalized as FulfillmentMode - return 'manual' -} - -async function findMappingForContext(context: OrderItemCostContext) { - const result = await query( - ` - SELECT m.*, p.canonical_sku, p.canonical_name - FROM channel_product_mappings m - JOIN product_catalog p ON p.id = m.canonical_product_id AND p.status = 'active' - WHERE m.provider = $1 - AND m.enabled = TRUE - AND (m.shop_id = '' OR m.shop_id = $3) - AND ( - (m.match_type = 'sku_code' AND m.platform_sku_code = $2) - OR ( - m.match_type = 'sku_code_name' - AND m.platform_sku_code = $2 - AND m.platform_sku_name_key = $4 - ) - OR ( - m.match_type = 'sku_code_series' - AND m.platform_sku_code = $2 - AND m.platform_sku_name_key = $5 - ) - OR (m.match_type = 'sku_name_series' AND m.platform_sku_name_key = $5) - ) - ORDER BY - CASE - WHEN m.match_type = 'sku_code' AND m.shop_id = $3 THEN 1 - WHEN m.match_type = 'sku_code' AND m.shop_id = '' THEN 2 - WHEN m.match_type = 'sku_code_name' AND m.shop_id = $3 THEN 3 - WHEN m.match_type = 'sku_code_name' AND m.shop_id = '' THEN 4 - WHEN m.match_type = 'sku_code_series' AND m.shop_id = $3 THEN 5 - WHEN m.match_type = 'sku_code_series' AND m.shop_id = '' THEN 6 - WHEN m.match_type = 'sku_name_series' AND m.shop_id = $3 THEN 7 - ELSE 8 - END, - m.id DESC - LIMIT 1 - `, - [ - context.provider, - context.skuCode, - context.shopId, - context.normalizedSkuName, - context.normalizedSkuSeries, - ], - ) - return result.rows[0] || null -} - -async function resolveCostLines(input: { - provider: string - shopId: string - fulfillmentMode: FulfillmentMode - canonicalProductId: number - workProductRuleId: number | null - quantity: number - revenueFen: number - asOf?: string -}): Promise { - const result = await query< - CostItemRow & { - version_id: number | null - version_scope: string | null - version_amount_fen: number | null - version_rate_bps: number | null - } - >( - ` - SELECT - ci.*, - v.id AS version_id, - v.scope_type AS version_scope, - v.amount_fen AS version_amount_fen, - v.rate_bps AS version_rate_bps - FROM cost_items ci - LEFT JOIN LATERAL ( - SELECT v.* - FROM cost_versions v - WHERE v.cost_item_id = ci.id - AND v.status = 'active' - AND v.effective_from <= $1::timestamptz - AND (v.effective_to IS NULL OR v.effective_to > $1::timestamptz) - AND ( - (v.scope_type = 'rule' AND $5::bigint IS NOT NULL AND v.work_product_rule_id = $5) - OR (v.scope_type = 'product' AND v.canonical_product_id = $4) - OR (v.scope_type = 'shop' AND v.provider = $2 AND v.shop_id = $3) - OR (v.scope_type = 'provider' AND v.provider = $2) - OR (v.scope_type = 'global') - ) - ORDER BY - CASE v.scope_type - WHEN 'rule' THEN 1 - WHEN 'product' THEN 2 - WHEN 'shop' THEN 3 - WHEN 'provider' THEN 4 - ELSE 5 - END, - v.effective_from DESC, - v.id DESC - LIMIT 1 - ) v ON TRUE -WHERE ci.status = 'active' - AND ${costModelScopeSql('$6')} - ORDER BY ci.id - `, - [ - input.asOf || nowIso(), - input.provider, - input.shopId, - input.canonicalProductId, - input.workProductRuleId, - input.fulfillmentMode, - ], - ) - - const lines: CostLine[] = [] - const missingCostItems: MissingCostItem[] = [] - for (const row of result.rows) { - if (!row.version_id) { - missingCostItems.push({ - costItemId: Number(row.id), - costItemCode: row.code, - costItemName: row.name, - costModel: row.cost_model as MissingCostItem['costModel'], - }) - continue - } - const pricingMode = row.pricing_mode - const amountFen = Number(row.version_amount_fen || 0) - const rateBps = Number(row.version_rate_bps || 0) - const quantity = Math.max(0, Number(input.quantity) || 0) - const unitAmountFen = amountFen - const totalAmountFen = calculateCostTotal({ - pricingMode, - amountFen, - rateBps, - quantity, - revenueFen: input.revenueFen, - }) - lines.push({ - costItemId: Number(row.id), - costItemCode: row.code, - costItemName: row.name, - costModel: row.cost_model as CostLine['costModel'], - pricingMode, - costVersionId: Number(row.version_id), - scopeType: String(row.version_scope || ''), - amountFen, - rateBps, - quantity, - unitAmountFen, - totalAmountFen, - }) - } - - return { - lines, - missingCostItems, - applicableCostItemCount: result.rows.length, - } -} - -export function calculateCostTotal(input: { - pricingMode: string - amountFen: number - rateBps: number - quantity: number - revenueFen: number -}) { - const amountFen = Math.max(0, Math.trunc(Number(input.amountFen) || 0)) - const rateBps = Math.max(0, Math.min(10000, Math.trunc(Number(input.rateBps) || 0))) - const quantity = Math.max(0, Number(input.quantity) || 0) - const revenueFen = Math.max(0, Math.trunc(Number(input.revenueFen) || 0)) - let total = amountFen - if (input.pricingMode === 'per_item') { - total = Math.round(amountFen * quantity) - } else if (input.pricingMode === 'percent_of_revenue') { - total = Math.round((revenueFen * rateBps) / 10000) - } else if (input.pricingMode === 'per_order') { - total = amountFen - } - return Math.max(0, total) -} - -export function validateCostVersionScope( - scopeType: string, - input: { - provider: string - shopId: string - canonicalProductId: number | null - workProductRuleId: number | null - }, -) { - if (scopeType === 'provider' && !input.provider) { - throw createHttpError('来源范围必须填写来源', { - statusCode: 400, - errorCode: 'cost_version_scope_invalid', - }) - } - if (scopeType === 'shop' && (!input.provider || !input.shopId)) { - throw createHttpError('店铺范围必须填写来源和店铺', { - statusCode: 400, - errorCode: 'cost_version_scope_invalid', - }) - } - if (scopeType === 'product' && !input.canonicalProductId) { - throw createHttpError('商品范围必须选择标准商品', { - statusCode: 400, - errorCode: 'cost_version_scope_invalid', - }) - } - if (scopeType === 'rule' && !input.workProductRuleId) { - throw createHttpError('规则范围必须选择接单规则', { - statusCode: 400, - errorCode: 'cost_version_scope_invalid', - }) - } -} - -function normalizeCostVersionScopeFields( - scopeType: string, - input: { - provider: string - shopId: string - canonicalProductId: number | null - workProductRuleId: number | null - }, -) { - const empty = { - provider: '', - shopId: '', - canonicalProductId: null as number | null, - workProductRuleId: null as number | null, - } - if (scopeType === 'provider') { - return { ...empty, provider: input.provider } - } - if (scopeType === 'shop') { - return { ...empty, provider: input.provider, shopId: input.shopId } - } - if (scopeType === 'product') { - return { ...empty, canonicalProductId: input.canonicalProductId } - } - if (scopeType === 'rule') { - return { ...empty, workProductRuleId: input.workProductRuleId } - } - return empty -} - -async function getCostItemRow(id: number) { - const result = await query('SELECT * FROM cost_items WHERE id = $1 LIMIT 1', [id]) - return requiredRow(result.rows[0], '成本项不存在') -} - -async function getCostVersionById(id: number) { - const result = await query( - ` - SELECT - v.*, - ci.code AS cost_item_code, - ci.name AS cost_item_name, - ci.cost_model AS cost_item_cost_model, - ci.pricing_mode, - p.canonical_sku, - p.canonical_name - FROM cost_versions v - JOIN cost_items ci ON ci.id = v.cost_item_id - LEFT JOIN product_catalog p ON p.id = v.canonical_product_id - WHERE v.id = $1 - LIMIT 1 - `, - [id], - ) - return mapCostVersion(requiredRow(result.rows[0], '成本版本不存在')) -} diff --git a/apps/backend/src/services/cost-center/cost-resolution-service.ts b/apps/backend/src/services/cost-center/cost-resolution-service.ts new file mode 100644 index 00000000..7f2b7e11 --- /dev/null +++ b/apps/backend/src/services/cost-center/cost-resolution-service.ts @@ -0,0 +1,680 @@ +/** + * 成本解析与快照：预览、解析上下文、按成本版本计算并落快照。 + * + * 口径：金额只来自 cost_versions；匹配不到成本进异常队列，不默认 0； + * 历史快照永不回写（superseded_at 只标记被实际成本替代）。 + * 成本定义 CRUD 见 cost-definition-service.ts。 + */ +import { query, withTransaction } from '../../db/client.js' +import type { JsonObject } from '../../types/json.js' +import { createHttpError } from '../../utils/http.js' +import { getProductRow } from './product-catalog-service.js' +import { nowIso } from '../../utils/time.js' +import type { AdminSession } from '../admin/admin-auth-service.js' +import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js' +import { resolveSkuNameQuantity } from '../product-match/pricing.js' +import { + FULFILLMENT_MODES, + costModelScopeSql, + mapCostVersion, + mapMapping, + mapProduct, + normalizeIsoDate, + normalizeOptionalId, + normalizeOptionalText, + normalizeRequiredText, + requiredRow, + text, +} from './cost-center-shared.js' +import type { + CostItemRow, + CostLine, + CostPreview, + CostResolution, + CostVersionRow, + FulfillmentMode, + MappingRow, + MissingCostItem, + OrderItemCostContext, +} from './cost-center-shared.js' + +export async function previewAdminCost(payload: JsonObject = {}) { + const context = await resolveCostContext(payload) + const mapping = await findMappingForContext(context) + const productId = context.canonicalProductId || mapping?.canonical_product_id || null + const product = productId ? mapProduct(await getProductRow(productId)) : null + const asOf = + normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || context.basisAt + const resolution: CostResolution = productId + ? await resolveCostLines({ + provider: context.provider, + shopId: context.shopId, + fulfillmentMode: context.fulfillmentMode, + canonicalProductId: productId, + workProductRuleId: context.workProductRuleId, + quantity: context.normalizedQuantity, + revenueFen: context.revenueFen, + asOf, + }) + : { lines: [], missingCostItems: [], applicableCostItemCount: 0 } + + const exceptions: CostPreview['exceptions'] = [] + if (!productId) { + exceptions.push({ + reason: 'product_mapping_missing', + message: `未找到 ${context.provider} / ${context.shopId || '全局'} / ${context.skuCode} 的标准商品映射`, + detail: { + provider: context.provider, + shopId: context.shopId, + skuCode: context.skuCode, + }, + }) + } else if (resolution.missingCostItems.length > 0) { + exceptions.push({ + reason: 'cost_missing', + message: `以下成本项没有匹配到有效成本版本：${resolution.missingCostItems + .map((item) => `${item.costItemName}（${item.costItemCode}）`) + .join('、')}`, + detail: { + fulfillmentMode: context.fulfillmentMode, + costItemIds: resolution.missingCostItems.map((item) => item.costItemId), + }, + }) + } else if (resolution.applicableCostItemCount === 0) { + exceptions.push({ + reason: 'cost_missing', + message: `没有配置适用于「${context.fulfillmentMode}」的成本项`, + detail: { fulfillmentMode: context.fulfillmentMode }, + }) + } + + // 人工履约的成本来自「人工 SKU 配置的消耗积分 × 平台换算率」，版本金额只是占位； + // 这里单独给出提示，并从成本行里剔掉 0 元占位，避免预览与实际扣款对不上。 + const manualCost = + context.fulfillmentMode === 'manual' ? await resolveManualCostHint(context.orderItemId) : null + const lines = manualCost + ? resolution.lines.filter((line) => line.costItemCode !== 'sys_manual_service') + : resolution.lines + + return { + context: { + ...context, + basisAt: asOf, + }, + mapping: mapping ? mapMapping(mapping) : null, + product, + lines, + missingCostItems: resolution.missingCostItems.filter( + (item) => !(manualCost && item.costItemCode === 'sys_manual_service'), + ), + manualCost, + totalCostFen: lines.reduce((sum, line) => sum + line.totalAmountFen, 0), + exceptions, + } +} + +/** 人工履约成本提示：直接读订单项绑定的履约配置。 */ +async function resolveManualCostHint(orderItemId: number | null) { + if (!orderItemId) return null + const result = await query<{ + platform_key: string + credit_quantity: string + cost_mode: string + unit_price_fen: string + }>( + ` + SELECT + cfg.platform_key, + cfg.credit_quantity, + cfg.cost_mode, + COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS unit_price_fen + FROM order_items oi + JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id + LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key + WHERE oi.id = $1 + LIMIT 1 + `, + [orderItemId], + ) + const row = result.rows[0] + if (!row) return null + const creditQuantity = Number(row.credit_quantity || 0) + const unitPriceFen = Number(row.unit_price_fen || 1) + return { + platformKey: row.platform_key, + creditQuantity, + unitPriceFen, + costMode: row.cost_mode === 'fixed' ? ('fixed' as const) : ('variable' as const), + costFen: creditQuantity > 0 ? creditQuantity * unitPriceFen : null, + } +} + +export async function createAdminCostSnapshot(payload: JsonObject = {}, _session: AdminSession) { + const preview = await previewAdminCost(payload) + const orderItemId = preview.context.orderItemId + if (!orderItemId) { + throw createHttpError('生成成本快照必须提供订单项 ID', { + statusCode: 400, + errorCode: 'cost_snapshot_order_item_required', + }) + } + const now = nowIso() + const stage = payload.snapshotStage === 'actual' ? 'actual' : 'estimated' + const snapshots: Array> = [] + let exceptionCreatedCount = 0 + let supersededCount = 0 + + await withTransaction(async (client) => { + for (const line of preview.lines) { + const currentResult = await client.query<{ + id: number + snapshot_stage: string + total_amount_fen: number + }>( + ` + SELECT id, snapshot_stage, total_amount_fen + FROM cost_snapshots + WHERE subject_type = 'order_item' + AND subject_id = $1 + AND cost_item_id = $2 + AND superseded_at IS NULL + FOR UPDATE + `, + [orderItemId, line.costItemId], + ) + const current = currentResult.rows[0] + if (current) { + // 预估快照一旦生成就代表支付时点的成本，不因后续改价重写； + // 只有实际成本允许替代当前有效行。 + if (stage === 'estimated') continue + if ( + current.snapshot_stage === 'actual' && + Number(current.total_amount_fen) === line.totalAmountFen + ) { + continue + } + await client.query( + `UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL`, + [now, current.id], + ) + supersededCount += 1 + } + + const result = await client.query<{ [column: string]: unknown }>( + ` + 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 + ) VALUES ( + 'order_item', $1, $2, $1, $3, $4, $5, $6, + $7, $8, $9, $10, 'resolved', $11::jsonb, $12, $13, $13 + ) + ON CONFLICT (subject_type, subject_id, cost_item_id) + WHERE superseded_at IS NULL + DO NOTHING + RETURNING * + `, + [ + orderItemId, + preview.context.orderId, + preview.context.workOrderId, + line.costItemId, + line.costVersionId, + preview.product?.id || null, + line.quantity, + line.unitAmountFen, + line.totalAmountFen, + stage, + JSON.stringify({ + skuCode: preview.context.skuCode, + skuName: preview.context.skuName, + provider: preview.context.provider, + shopId: preview.context.shopId, + normalizedQuantity: preview.context.normalizedQuantity, + fulfillmentMode: preview.context.fulfillmentMode, + basisAt: preview.context.basisAt, + }), + preview.context.basisAt, + now, + ], + ) + if (result.rows[0]) snapshots.push(result.rows[0]) + } + + for (const exception of preview.exceptions) { + const result = await client.query( + ` + INSERT INTO cost_exceptions ( + subject_type, subject_id, order_item_id, work_order_id, + reason, status, detail_json, created_at + ) VALUES ('order_item', $1, $1, $2, $3, 'open', $4::jsonb, $5) + ON CONFLICT (subject_type, subject_id, reason) DO NOTHING + `, + [ + orderItemId, + preview.context.workOrderId, + exception.reason, + JSON.stringify({ + ...exception.detail, + message: exception.message, + skuCode: preview.context.skuCode, + }), + now, + ], + ) + exceptionCreatedCount += result.rowCount || 0 + } + }) + + return { + preview, + snapshots, + createdCount: snapshots.length, + supersededCount, + exceptionCreatedCount, + } +} + +async function resolveCostContext(payload: JsonObject): Promise { + const orderItemId = normalizeOptionalId(payload.orderItemId ?? payload.order_item_id) + if (orderItemId) { + const result = await query<{ + order_item_id: number + order_id: number + provider: string + shop_id: string + sku_code: string + sku_name: string + normalized_sku_name: string | null + normalized_sku_series: string | null + quantity: number + normalized_quantity: number | null + canonical_product_id: number | null + revenue_fen: number + basis_at: string + work_order_id: number | null + work_product_rule_id: number | null + fulfillment_mode: string | null + }>( + ` + SELECT + oi.id AS order_item_id, + oi.order_id, + o.provider, + o.shop_id, + oi.sku_code, + oi.sku_name, + oi.normalized_sku_name, + oi.normalized_sku_series, + oi.quantity, + oi.normalized_quantity, + oi.canonical_product_id, + oi.revenue_amount_fen AS revenue_fen, + 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, + oi.fulfillment_mode + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + LEFT JOIN LATERAL ( + SELECT id, product_rule_id + FROM work_orders + WHERE order_item_id = oi.id + ORDER BY id DESC + LIMIT 1 + ) wo ON TRUE + WHERE oi.id = $1 + LIMIT 1 + `, + [orderItemId], + ) + const row = result.rows[0] + if (!row) { + throw createHttpError('订单项不存在', { + statusCode: 404, + errorCode: 'cost_order_item_not_found', + }) + } + return { + orderItemId: Number(row.order_item_id), + orderId: Number(row.order_id), + provider: row.provider, + shopId: row.shop_id, + skuCode: row.sku_code, + skuName: row.sku_name, + normalizedSkuName: + row.normalized_sku_name || normalizeMatchText(row.sku_name || row.sku_code), + normalizedSkuSeries: + row.normalized_sku_series || normalizeSkuSeriesEquivalence(row.sku_name || row.sku_code), + quantity: Math.max(1, Number(row.quantity) || 1), + normalizedQuantity: + row.normalized_quantity && Number(row.normalized_quantity) > 0 + ? Number(row.normalized_quantity) + : Math.max(1, resolveSkuNameQuantity(row.sku_name || row.sku_code)), + canonicalProductId: row.canonical_product_id ? Number(row.canonical_product_id) : null, + revenueFen: Math.max(0, Number(row.revenue_fen) || 0), + basisAt: new Date(row.basis_at).toISOString(), + workOrderId: row.work_order_id ? Number(row.work_order_id) : null, + workProductRuleId: row.work_product_rule_id ? Number(row.work_product_rule_id) : null, + fulfillmentMode: resolveFulfillmentMode(row.fulfillment_mode), + } + } + + const provider = normalizeRequiredText(payload.provider, '来源', 64) + const skuCode = normalizeRequiredText(payload.skuCode ?? payload.sku_code, '平台 SKU', 120) + const quantity = Math.max(1, Math.trunc(Number(payload.quantity) || 1)) + const skuName = normalizeOptionalText(payload.skuName ?? payload.sku_name, 200) + const fulfillmentMode = resolveFulfillmentMode( + payload.fulfillmentMode ?? payload.fulfillment_mode, + ) + return { + orderItemId: 0, + orderId: 0, + provider, + shopId: normalizeOptionalText(payload.shopId ?? payload.shop_id, 120), + skuCode, + skuName, + normalizedSkuName: normalizeMatchText(skuName || skuCode), + normalizedSkuSeries: normalizeSkuSeriesEquivalence(skuName || skuCode), + quantity, + normalizedQuantity: quantity, + canonicalProductId: normalizeOptionalId( + payload.canonicalProductId ?? payload.canonical_product_id, + ), + revenueFen: Math.max(0, Math.trunc(Number(payload.revenueFen ?? payload.revenue_fen) || 0)), + basisAt: + normalizeIsoDate(payload.asOf ?? payload.costBasisAt, '成本基准时间', true) || nowIso(), + workOrderId: null, + workProductRuleId: normalizeOptionalId( + payload.workProductRuleId ?? payload.work_product_rule_id, + ), + fulfillmentMode, + } +} + +/** + * 履约方式只读 order_items.fulfillment_mode。 + * 订单项还没判定时按人工处理，不允许用 provider 或核销码兜底。 + */ +function resolveFulfillmentMode(value: unknown): FulfillmentMode { + const normalized = text(value) + if (normalized && FULFILLMENT_MODES.has(normalized)) return normalized as FulfillmentMode + return 'manual' +} + +async function findMappingForContext(context: OrderItemCostContext) { + const result = await query( + ` + SELECT m.*, p.canonical_sku, p.canonical_name + FROM channel_product_mappings m + JOIN product_catalog p ON p.id = m.canonical_product_id AND p.status = 'active' + WHERE m.provider = $1 + AND m.enabled = TRUE + AND (m.shop_id = '' OR m.shop_id = $3) + AND ( + (m.match_type = 'sku_code' AND m.platform_sku_code = $2) + OR ( + m.match_type = 'sku_code_name' + AND m.platform_sku_code = $2 + AND m.platform_sku_name_key = $4 + ) + OR ( + m.match_type = 'sku_code_series' + AND m.platform_sku_code = $2 + AND m.platform_sku_name_key = $5 + ) + OR (m.match_type = 'sku_name_series' AND m.platform_sku_name_key = $5) + ) + ORDER BY + CASE + WHEN m.match_type = 'sku_code' AND m.shop_id = $3 THEN 1 + WHEN m.match_type = 'sku_code' AND m.shop_id = '' THEN 2 + WHEN m.match_type = 'sku_code_name' AND m.shop_id = $3 THEN 3 + WHEN m.match_type = 'sku_code_name' AND m.shop_id = '' THEN 4 + WHEN m.match_type = 'sku_code_series' AND m.shop_id = $3 THEN 5 + WHEN m.match_type = 'sku_code_series' AND m.shop_id = '' THEN 6 + WHEN m.match_type = 'sku_name_series' AND m.shop_id = $3 THEN 7 + ELSE 8 + END, + m.id DESC + LIMIT 1 + `, + [ + context.provider, + context.skuCode, + context.shopId, + context.normalizedSkuName, + context.normalizedSkuSeries, + ], + ) + return result.rows[0] || null +} + +async function resolveCostLines(input: { + provider: string + shopId: string + fulfillmentMode: FulfillmentMode + canonicalProductId: number + workProductRuleId: number | null + quantity: number + revenueFen: number + asOf?: string +}): Promise { + const result = await query< + CostItemRow & { + version_id: number | null + version_scope: string | null + version_amount_fen: number | null + version_rate_bps: number | null + } + >( + ` + SELECT + ci.*, + v.id AS version_id, + v.scope_type AS version_scope, + v.amount_fen AS version_amount_fen, + v.rate_bps AS version_rate_bps + FROM cost_items ci + LEFT JOIN LATERAL ( + SELECT v.* + FROM cost_versions v + WHERE v.cost_item_id = ci.id + AND v.status = 'active' + AND v.effective_from <= $1::timestamptz + AND (v.effective_to IS NULL OR v.effective_to > $1::timestamptz) + AND ( + (v.scope_type = 'rule' AND $5::bigint IS NOT NULL AND v.work_product_rule_id = $5) + OR (v.scope_type = 'product' AND v.canonical_product_id = $4) + OR (v.scope_type = 'shop' AND v.provider = $2 AND v.shop_id = $3) + OR (v.scope_type = 'provider' AND v.provider = $2) + OR (v.scope_type = 'global') + ) + ORDER BY + CASE v.scope_type + WHEN 'rule' THEN 1 + WHEN 'product' THEN 2 + WHEN 'shop' THEN 3 + WHEN 'provider' THEN 4 + ELSE 5 + END, + v.effective_from DESC, + v.id DESC + LIMIT 1 + ) v ON TRUE +WHERE ci.status = 'active' + AND ${costModelScopeSql('$6')} + ORDER BY ci.id + `, + [ + input.asOf || nowIso(), + input.provider, + input.shopId, + input.canonicalProductId, + input.workProductRuleId, + input.fulfillmentMode, + ], + ) + + const lines: CostLine[] = [] + const missingCostItems: MissingCostItem[] = [] + for (const row of result.rows) { + if (!row.version_id) { + missingCostItems.push({ + costItemId: Number(row.id), + costItemCode: row.code, + costItemName: row.name, + costModel: row.cost_model as MissingCostItem['costModel'], + }) + continue + } + const pricingMode = row.pricing_mode + const amountFen = Number(row.version_amount_fen || 0) + const rateBps = Number(row.version_rate_bps || 0) + const quantity = Math.max(0, Number(input.quantity) || 0) + const unitAmountFen = amountFen + const totalAmountFen = calculateCostTotal({ + pricingMode, + amountFen, + rateBps, + quantity, + revenueFen: input.revenueFen, + }) + lines.push({ + costItemId: Number(row.id), + costItemCode: row.code, + costItemName: row.name, + costModel: row.cost_model as CostLine['costModel'], + pricingMode, + costVersionId: Number(row.version_id), + scopeType: String(row.version_scope || ''), + amountFen, + rateBps, + quantity, + unitAmountFen, + totalAmountFen, + }) + } + + return { + lines, + missingCostItems, + applicableCostItemCount: result.rows.length, + } +} + +export function calculateCostTotal(input: { + pricingMode: string + amountFen: number + rateBps: number + quantity: number + revenueFen: number +}) { + const amountFen = Math.max(0, Math.trunc(Number(input.amountFen) || 0)) + const rateBps = Math.max(0, Math.min(10000, Math.trunc(Number(input.rateBps) || 0))) + const quantity = Math.max(0, Number(input.quantity) || 0) + const revenueFen = Math.max(0, Math.trunc(Number(input.revenueFen) || 0)) + let total = amountFen + if (input.pricingMode === 'per_item') { + total = Math.round(amountFen * quantity) + } else if (input.pricingMode === 'percent_of_revenue') { + total = Math.round((revenueFen * rateBps) / 10000) + } else if (input.pricingMode === 'per_order') { + total = amountFen + } + return Math.max(0, total) +} + +export function validateCostVersionScope( + scopeType: string, + input: { + provider: string + shopId: string + canonicalProductId: number | null + workProductRuleId: number | null + }, +) { + if (scopeType === 'provider' && !input.provider) { + throw createHttpError('来源范围必须填写来源', { + statusCode: 400, + errorCode: 'cost_version_scope_invalid', + }) + } + if (scopeType === 'shop' && (!input.provider || !input.shopId)) { + throw createHttpError('店铺范围必须填写来源和店铺', { + statusCode: 400, + errorCode: 'cost_version_scope_invalid', + }) + } + if (scopeType === 'product' && !input.canonicalProductId) { + throw createHttpError('商品范围必须选择标准商品', { + statusCode: 400, + errorCode: 'cost_version_scope_invalid', + }) + } + if (scopeType === 'rule' && !input.workProductRuleId) { + throw createHttpError('规则范围必须选择接单规则', { + statusCode: 400, + errorCode: 'cost_version_scope_invalid', + }) + } +} + +export function normalizeCostVersionScopeFields( + scopeType: string, + input: { + provider: string + shopId: string + canonicalProductId: number | null + workProductRuleId: number | null + }, +) { + const empty = { + provider: '', + shopId: '', + canonicalProductId: null as number | null, + workProductRuleId: null as number | null, + } + if (scopeType === 'provider') { + return { ...empty, provider: input.provider } + } + if (scopeType === 'shop') { + return { ...empty, provider: input.provider, shopId: input.shopId } + } + if (scopeType === 'product') { + return { ...empty, canonicalProductId: input.canonicalProductId } + } + if (scopeType === 'rule') { + return { ...empty, workProductRuleId: input.workProductRuleId } + } + return empty +} + +export async function getCostItemRow(id: number) { + const result = await query('SELECT * FROM cost_items WHERE id = $1 LIMIT 1', [id]) + return requiredRow(result.rows[0], '成本项不存在') +} + +export async function getCostVersionById(id: number) { + const result = await query( + ` + SELECT + v.*, + ci.code AS cost_item_code, + ci.name AS cost_item_name, + ci.cost_model AS cost_item_cost_model, + ci.pricing_mode, + p.canonical_sku, + p.canonical_name + FROM cost_versions v + JOIN cost_items ci ON ci.id = v.cost_item_id + LEFT JOIN product_catalog p ON p.id = v.canonical_product_id + WHERE v.id = $1 + LIMIT 1 + `, + [id], + ) + return mapCostVersion(requiredRow(result.rows[0], '成本版本不存在')) +} diff --git a/apps/backend/src/services/cost-center/template-bootstrap-service.ts b/apps/backend/src/services/cost-center/template-bootstrap-service.ts index 44cb9b68..d1c22e5c 100644 --- a/apps/backend/src/services/cost-center/template-bootstrap-service.ts +++ b/apps/backend/src/services/cost-center/template-bootstrap-service.ts @@ -1,34 +1,23 @@ /** - * 接单模板引导、自动发货商品同步与工单证据回填。 + * 接单模板引导：预览与执行、映射计划构建、模板商品编码。 * - * 口径（2026-10-23 重写）： - * 1. 商品表不存履约载体，来源只由 canonical_sku 前缀表达： - * aff: 自动发货 / tpl: 接单模板 / wo: 工单证据； - * 2. 自动发货 SKU 固定归一为 aff:，不写渠道映射； - * 3. 匹配方式与名称键都参与映射唯一键，91 同店铺复用 sku_code 才能消歧。 + * 关联： + * - affiliate-product-service.ts 自动发货商品同步与 SKU 回填 + * - work-order-product-service.ts 接单平台商品回填与模板发单金额同步 */ import { query, withTransaction } from '../../db/client.js' -import type { AdminSession } from '../admin/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 type { AdminSession } from '../admin/admin-auth-service.js' import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-match/normalize.js' -import { - MAPPING_MATCH_TYPES, - WORK_ORDER_TEMPLATE_COST_BASELINE, - candidateKey, - mappingKey, - mappingMatchTypeLabel, - normalizeMappingMatchType, - resolveMappingNameKey, - type TemplateMappingMatchType, -} from './cost-center-shared.js' +import { candidateKey, mappingKey, mappingMatchTypeLabel } from './cost-center-shared.js' +import type { TemplateMappingMatchType } from './cost-center-shared.js' import { backfillAdminNormalizedQuantities, backfillAdminProductMappings, } from './product-catalog-service.js' +import { syncAdminWorkOrderTemplateCostVersions } from './work-order-product-service.js' export async function previewAdminTemplateProductBootstrap() { const data = await loadTemplateProductBootstrapData() @@ -225,657 +214,6 @@ export async function applyAdminTemplateProductBootstrap( * 平台 SKU 归一，不能作为自动发货依据。SKU 固定为 `aff:`， * 重复同步只更新名称/证据，不会重复建商品。 */ -export async function syncAdminAffiliateDashProducts() { - const now = nowIso() - const result = await query<{ inserted: boolean }>( - ` - WITH auto_task AS ( - SELECT - btrim(context_json->>'skuCode') AS sku_code, - btrim(context_json->>'skuName') AS sku_name, - created_at - FROM fulfillment_tasks - WHERE executor_key = 'affiliate_dash' - AND btrim(COALESCE(context_json->>'skuCode', '')) <> '' - AND btrim(COALESCE(context_json->>'skuName', '')) <> '' - ), - agg AS ( - SELECT sku_code, count(*)::int AS task_count, max(created_at) AS last_fulfilled_at - FROM auto_task - GROUP BY sku_code - ), - ranked AS ( - SELECT - sku_code, - sku_name, - row_number() OVER ( - PARTITION BY sku_code - ORDER BY (sku_name <> sku_code) DESC, count(*) DESC, sku_name - ) AS rn - FROM auto_task - GROUP BY sku_code, sku_name - ) - INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, - attributes_json, created_at, updated_at - ) - SELECT - 'aff:' || ranked.sku_code, - ranked.sku_name, - '', - CASE WHEN ranked.sku_name LIKE '%个' THEN '个' ELSE '件' END, - 'active', - jsonb_build_object( - 'source', 'affiliate_dash_task', - 'skuCode', ranked.sku_code, - 'taskCount', agg.task_count, - 'lastFulfilledAt', agg.last_fulfilled_at - ), - $1, - $1 - FROM ranked - JOIN agg ON agg.sku_code = ranked.sku_code - WHERE ranked.rn = 1 - ON CONFLICT (canonical_sku) DO UPDATE - SET canonical_name = EXCLUDED.canonical_name, - unit_name = EXCLUDED.unit_name, - attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json, - updated_at = $1 - RETURNING (xmax = 0) AS inserted - `, - [now], - ) - const createdCount = result.rows.filter((row) => row.inserted).length - return { - createdCount, - updatedCount: result.rows.length - createdCount, - productCount: result.rows.length, - } -} - -type AffiliateSkuBackfillRow = { - affiliate_item_count: number - matched_count: number - updated_count: number - from_date: string | null - 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 } -} - -/** - * 按自动发货履约证据直接回填订单项标准商品。 - * - * 自动发货只有 affiliate-dash 一个上游渠道，SKU 固定归一为 `aff:`， - * 不写 channel_product_mappings。多个任务共用同一订单项时取最新 SKU。 - */ -export async function backfillAdminAffiliateDashSkus() { - const now = nowIso() - const result = await query( - ` - 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 affiliate_item - JOIN order_items oi ON oi.id = affiliate_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 affiliate_item - JOIN order_items oi ON oi.id = affiliate_item.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 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: Awaited> | null = null - if (matchedCount > 0 && row?.from_date && row?.to_date) { - factsRebuild = await rebuildFactsForRange(row.from_date, row.to_date) - } - - return { - updatedCount, - skippedCount, - remainingCount, - affiliateItemCount, - matchedCount, - factsRebuild, - } -} - -type WorkOrderSkuBackfillRow = { - work_order_item_count: number - from_date: string | null - to_date: string | null -} - -const WORK_ORDER_BACKFILL_BATCH_SIZE = 20000 - -/** - * 回填接单平台订单项的标准商品。 - * - * 第一步用工单真实使用的 `product_rule_id` 指向 `tpl:`； - * 仍无法归一的长尾项，再按 `provider + shop_id + sku_code + normalized_sku_name` - * 补建 `wo:` 商品和 `sku_code_name` 渠道映射，保证后续同类订单能自动命中。 - */ -export async function backfillAdminWorkOrderProductSkus() { - const now = nowIso() - // order_item_id 有唯一索引，这里可以直接 count，避免 DISTINCT ON 被 - // 三个子查询反复物化后撞 15 秒语句超时。 - const result = await query( - ` - SELECT - COUNT(*)::int AS work_order_item_count, - to_char( - MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date), - 'YYYY-MM-DD' - ) AS from_date, - to_char( - MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date), - 'YYYY-MM-DD' - ) AS to_date - FROM work_orders wo - JOIN order_items oi ON oi.id = wo.order_item_id - JOIN orders o ON o.id = oi.order_id - WHERE wo.order_item_id IS NOT NULL - `, - ) - - // 单条 UPDATE 覆盖十几万行会撞 15 秒语句超时；按订单项 id 分批， - // 每批只选 2 万条工单证据，再按 id 数组更新。 - let lastRuleItemId = 0 - let ruleMatchedCount = 0 - let ruleUpdatedCount = 0 - for (let batch = 0; batch < 40; batch += 1) { - const targetRows = await query<{ id: number; canonical_product_id: number }>( - ` - 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 - AND wo.order_item_id > $1 - ORDER BY wo.order_item_id, wo.id DESC - LIMIT $2 - ) - 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 - ORDER BY woi.id - `, - [lastRuleItemId, WORK_ORDER_BACKFILL_BATCH_SIZE], - ) - if (targetRows.rows.length === 0) break - const lastSelectedId = Number(targetRows.rows[targetRows.rows.length - 1]?.id || 0) - if (lastSelectedId <= lastRuleItemId) break - lastRuleItemId = lastSelectedId - ruleMatchedCount += targetRows.rows.length - - const updated = await query( - ` - UPDATE order_items oi - SET canonical_product_id = input.canonical_product_id, - updated_at = $1 - FROM unnest($2::bigint[], $3::bigint[]) - AS input(id, canonical_product_id) - WHERE oi.id = input.id - AND oi.canonical_product_id IS DISTINCT FROM input.canonical_product_id - `, - [ - now, - targetRows.rows.map((row) => Number(row.id)), - targetRows.rows.map((row) => Number(row.canonical_product_id)), - ], - ) - ruleUpdatedCount += updated.rowCount || 0 - if (targetRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break - } - - 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, 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, - 'active', - jsonb_build_object('source', 'work_order_sku_name', 'normalizedName', m.name), - $1, - $1 - FROM missing m - ON CONFLICT (canonical_sku) DO UPDATE - SET 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], - ) - - let lastTailItemId = 0 - let tailUpdatedCount = 0 - for (let batch = 0; batch < 40; batch += 1) { - const tailRows = await query<{ id: number; canonical_product_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 oi.id > $1 - AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) - ORDER BY oi.id - LIMIT $2 - ) - 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 - ORDER BY t.id - `, - [lastTailItemId, WORK_ORDER_BACKFILL_BATCH_SIZE], - ) - if (tailRows.rows.length === 0) break - const lastSelectedId = Number(tailRows.rows[tailRows.rows.length - 1]?.id || 0) - if (lastSelectedId <= lastTailItemId) break - lastTailItemId = lastSelectedId - - const updated = await query( - ` - UPDATE order_items oi - SET canonical_product_id = input.canonical_product_id, - updated_at = $1 - FROM unnest($2::bigint[], $3::bigint[]) - AS input(id, canonical_product_id) - WHERE oi.id = input.id - AND oi.canonical_product_id IS NULL - `, - [ - now, - tailRows.rows.map((row) => Number(row.id)), - tailRows.rows.map((row) => Number(row.canonical_product_id)), - ], - ) - tailUpdatedCount += updated.rowCount || 0 - if (tailRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break - } - - const row = result.rows[0] - const workOrderItemCount = Number(row?.work_order_item_count || 0) - const tailProductCount = tailProducts.rows.filter((item) => item.inserted).length - const tailMappingCount = tailMappings.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> | 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 WorkOrderTemplateCostSyncRow = { - rule_count: number - unit_price_rule_count: number - active_version_count: number -} - -/** - * 把接单模板的发单金额同步成规则级成本版本。 - * - * 固定金额模板取 `reward_amount`；按件模板取 `unit_price_fen`，数量由订单 SKU - * 决定。全局 0 元版本只作为“按工单实际结算”的占位，不再代表模板价格。 - */ -export async function syncAdminWorkOrderTemplateCostVersions(ruleId?: number) { - const now = nowIso() - const scopedRuleId = ruleId && Number.isFinite(ruleId) ? Number(ruleId) : null - const retired = await query( - ` - WITH target AS ( - SELECT - wpr.id AS rule_id, - CASE - WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen - ELSE wpr.reward_amount - END AS amount_fen - FROM work_product_rules wpr - WHERE ($2::bigint IS NULL OR wpr.id = $2) - ), - sys AS ( - SELECT id FROM cost_items - WHERE code = 'sys_work_order_reward' AND status = 'active' - ), - changed AS ( - SELECT v.id - FROM cost_versions v - JOIN sys si ON si.id = v.cost_item_id - JOIN target t ON t.rule_id = v.work_product_rule_id - WHERE v.scope_type = 'rule' - AND v.status = 'active' - AND v.amount_fen <> t.amount_fen - ) - UPDATE cost_versions v - SET status = 'retired', - effective_to = $1, - change_reason = '模板发单金额变更', - updated_at = $1 - FROM changed c - WHERE v.id = c.id - RETURNING v.id - `, - [now, scopedRuleId], - ) - - const inserted = await query<{ id: number }>( - ` - WITH target AS ( - SELECT - wpr.id AS rule_id, - CASE - WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen - ELSE wpr.reward_amount - END AS amount_fen - FROM work_product_rules wpr - WHERE ($3::bigint IS NULL OR wpr.id = $3) - ), - sys AS ( - SELECT id FROM cost_items - WHERE code = 'sys_work_order_reward' AND status = 'active' - ), - missing AS ( - SELECT - t.rule_id, - t.amount_fen, - si.id AS cost_item_id, - EXISTS ( - SELECT 1 FROM cost_versions h - WHERE h.cost_item_id = si.id - AND h.scope_type = 'rule' - AND h.work_product_rule_id = t.rule_id - ) AS has_history - FROM target t - CROSS JOIN sys si - WHERE t.amount_fen > 0 - AND NOT EXISTS ( - SELECT 1 FROM cost_versions v - WHERE v.cost_item_id = si.id - AND v.scope_type = 'rule' - AND v.work_product_rule_id = t.rule_id - AND v.status = 'active' - ) - ) - INSERT INTO cost_versions ( - cost_item_id, scope_type, provider, shop_id, canonical_product_id, - work_product_rule_id, amount_fen, rate_bps, effective_from, effective_to, - source, status, change_reason, created_by_admin_user_id, - approved_by_admin_user_id, created_at, updated_at - ) - SELECT - m.cost_item_id, - 'rule', - '', - '', - NULL, - m.rule_id, - m.amount_fen, - 0, - CASE WHEN m.has_history THEN $1::timestamptz ELSE $2::timestamptz END, - NULL, - 'import', - 'active', - CASE - WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 - THEN '模板按件发单单价（数量取订单 SKU）' - ELSE '模板固定发单金额' - END, - NULL, - NULL, - $1, - $1 - FROM missing m - JOIN work_product_rules wpr ON wpr.id = m.rule_id - ON CONFLICT DO NOTHING - RETURNING id - `, - [now, WORK_ORDER_TEMPLATE_COST_BASELINE, scopedRuleId], - ) - - const linked = await query( - ` - WITH rule_version AS ( - SELECT DISTINCT ON (v.work_product_rule_id) - v.work_product_rule_id, - v.id AS cost_version_id, - v.cost_item_id - FROM cost_versions v - JOIN cost_items ci ON ci.id = v.cost_item_id - AND ci.code = 'sys_work_order_reward' - WHERE v.scope_type = 'rule' - AND v.status = 'active' - AND ($1::bigint IS NULL OR v.work_product_rule_id = $1) - ORDER BY v.work_product_rule_id, v.effective_from DESC, v.id DESC - ) - UPDATE cost_snapshots cs - SET cost_version_id = rv.cost_version_id - FROM work_orders wo - JOIN rule_version rv ON rv.work_product_rule_id = wo.product_rule_id - WHERE cs.order_item_id = wo.order_item_id - AND cs.cost_item_id = rv.cost_item_id - AND cs.superseded_at IS NULL - AND cs.cost_version_id <> rv.cost_version_id - AND EXISTS ( - SELECT 1 FROM cost_versions gv - WHERE gv.id = cs.cost_version_id AND gv.scope_type = 'global' - ) - AND ($1::bigint IS NULL OR wo.product_rule_id = $1) - `, - [scopedRuleId], - ) - - const summary = await query( - ` - SELECT - COUNT(*)::int AS rule_count, - COUNT(*) FILTER ( - WHERE wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 - )::int AS unit_price_rule_count, - COUNT(v.id) FILTER ( - WHERE v.scope_type = 'rule' AND v.status = 'active' - )::int AS active_version_count - FROM work_product_rules wpr - LEFT JOIN cost_versions v - ON v.cost_item_id = ( - SELECT id FROM cost_items WHERE code = 'sys_work_order_reward' LIMIT 1 - ) - AND v.work_product_rule_id = wpr.id - WHERE ($1::bigint IS NULL OR wpr.id = $1) - `, - [scopedRuleId], - ) - - const row = summary.rows[0] - return { - ruleId: scopedRuleId, - ruleCount: Number(row?.rule_count || 0), - unitPriceRuleCount: Number(row?.unit_price_rule_count || 0), - activeVersionCount: Number(row?.active_version_count || 0), - createdCount: inserted.rows.length, - retiredCount: retired.rowCount || 0, - linkedSnapshotCount: linked.rowCount || 0, - } -} type TemplateBootstrapTemplateRow = { id: number diff --git a/apps/backend/src/services/cost-center/work-order-product-service.ts b/apps/backend/src/services/cost-center/work-order-product-service.ts new file mode 100644 index 00000000..e8b42a30 --- /dev/null +++ b/apps/backend/src/services/cost-center/work-order-product-service.ts @@ -0,0 +1,467 @@ +/** + * 接单平台（work_order）商品回填与模板发单金额同步。 + */ +import { query } from '../../db/client.js' +import { nowIso } from '../../utils/time.js' +import { rebuildFactsForRange } from './affiliate-product-service.js' +import { WORK_ORDER_TEMPLATE_COST_BASELINE } from './cost-center-shared.js' + +type WorkOrderSkuBackfillRow = { + work_order_item_count: number + from_date: string | null + to_date: string | null +} + +const WORK_ORDER_BACKFILL_BATCH_SIZE = 20000 + +/** + * 回填接单平台订单项的标准商品。 + * + * 第一步用工单真实使用的 `product_rule_id` 指向 `tpl:`； + * 仍无法归一的长尾项，再按 `provider + shop_id + sku_code + normalized_sku_name` + * 补建 `wo:` 商品和 `sku_code_name` 渠道映射，保证后续同类订单能自动命中。 + */ +export async function backfillAdminWorkOrderProductSkus() { + const now = nowIso() + // order_item_id 有唯一索引，这里可以直接 count，避免 DISTINCT ON 被 + // 三个子查询反复物化后撞 15 秒语句超时。 + const result = await query( + ` + SELECT + COUNT(*)::int AS work_order_item_count, + to_char( + MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date), + 'YYYY-MM-DD' + ) AS from_date, + to_char( + MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date), + 'YYYY-MM-DD' + ) AS to_date + FROM work_orders wo + JOIN order_items oi ON oi.id = wo.order_item_id + JOIN orders o ON o.id = oi.order_id + WHERE wo.order_item_id IS NOT NULL + `, + ) + + // 单条 UPDATE 覆盖十几万行会撞 15 秒语句超时；按订单项 id 分批， + // 每批只选 2 万条工单证据，再按 id 数组更新。 + let lastRuleItemId = 0 + let ruleMatchedCount = 0 + let ruleUpdatedCount = 0 + for (let batch = 0; batch < 40; batch += 1) { + const targetRows = await query<{ id: number; canonical_product_id: number }>( + ` + 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 + AND wo.order_item_id > $1 + ORDER BY wo.order_item_id, wo.id DESC + LIMIT $2 + ) + 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 + ORDER BY woi.id + `, + [lastRuleItemId, WORK_ORDER_BACKFILL_BATCH_SIZE], + ) + if (targetRows.rows.length === 0) break + const lastSelectedId = Number(targetRows.rows[targetRows.rows.length - 1]?.id || 0) + if (lastSelectedId <= lastRuleItemId) break + lastRuleItemId = lastSelectedId + ruleMatchedCount += targetRows.rows.length + + const updated = await query( + ` + UPDATE order_items oi + SET canonical_product_id = input.canonical_product_id, + updated_at = $1 + FROM unnest($2::bigint[], $3::bigint[]) + AS input(id, canonical_product_id) + WHERE oi.id = input.id + AND oi.canonical_product_id IS DISTINCT FROM input.canonical_product_id + `, + [ + now, + targetRows.rows.map((row) => Number(row.id)), + targetRows.rows.map((row) => Number(row.canonical_product_id)), + ], + ) + ruleUpdatedCount += updated.rowCount || 0 + if (targetRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break + } + + 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, 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, + 'active', + jsonb_build_object('source', 'work_order_sku_name', 'normalizedName', m.name), + $1, + $1 + FROM missing m + ON CONFLICT (canonical_sku) DO UPDATE + SET 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], + ) + + let lastTailItemId = 0 + let tailUpdatedCount = 0 + for (let batch = 0; batch < 40; batch += 1) { + const tailRows = await query<{ id: number; canonical_product_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 oi.id > $1 + AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + ORDER BY oi.id + LIMIT $2 + ) + 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 + ORDER BY t.id + `, + [lastTailItemId, WORK_ORDER_BACKFILL_BATCH_SIZE], + ) + if (tailRows.rows.length === 0) break + const lastSelectedId = Number(tailRows.rows[tailRows.rows.length - 1]?.id || 0) + if (lastSelectedId <= lastTailItemId) break + lastTailItemId = lastSelectedId + + const updated = await query( + ` + UPDATE order_items oi + SET canonical_product_id = input.canonical_product_id, + updated_at = $1 + FROM unnest($2::bigint[], $3::bigint[]) + AS input(id, canonical_product_id) + WHERE oi.id = input.id + AND oi.canonical_product_id IS NULL + `, + [ + now, + tailRows.rows.map((row) => Number(row.id)), + tailRows.rows.map((row) => Number(row.canonical_product_id)), + ], + ) + tailUpdatedCount += updated.rowCount || 0 + if (tailRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break + } + + const row = result.rows[0] + const workOrderItemCount = Number(row?.work_order_item_count || 0) + const tailProductCount = tailProducts.rows.filter((item) => item.inserted).length + const tailMappingCount = tailMappings.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> | 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 WorkOrderTemplateCostSyncRow = { + rule_count: number + unit_price_rule_count: number + active_version_count: number +} + +/** + * 把接单模板的发单金额同步成规则级成本版本。 + * + * 固定金额模板取 `reward_amount`；按件模板取 `unit_price_fen`，数量由订单 SKU + * 决定。全局 0 元版本只作为“按工单实际结算”的占位，不再代表模板价格。 + */ +export async function syncAdminWorkOrderTemplateCostVersions(ruleId?: number) { + const now = nowIso() + const scopedRuleId = ruleId && Number.isFinite(ruleId) ? Number(ruleId) : null + const retired = await query( + ` + WITH target AS ( + SELECT + wpr.id AS rule_id, + CASE + WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen + ELSE wpr.reward_amount + END AS amount_fen + FROM work_product_rules wpr + WHERE ($2::bigint IS NULL OR wpr.id = $2) + ), + sys AS ( + SELECT id FROM cost_items + WHERE code = 'sys_work_order_reward' AND status = 'active' + ), + changed AS ( + SELECT v.id + FROM cost_versions v + JOIN sys si ON si.id = v.cost_item_id + JOIN target t ON t.rule_id = v.work_product_rule_id + WHERE v.scope_type = 'rule' + AND v.status = 'active' + AND v.amount_fen <> t.amount_fen + ) + UPDATE cost_versions v + SET status = 'retired', + effective_to = $1, + change_reason = '模板发单金额变更', + updated_at = $1 + FROM changed c + WHERE v.id = c.id + RETURNING v.id + `, + [now, scopedRuleId], + ) + + const inserted = await query<{ id: number }>( + ` + WITH target AS ( + SELECT + wpr.id AS rule_id, + CASE + WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 THEN wpr.unit_price_fen + ELSE wpr.reward_amount + END AS amount_fen + FROM work_product_rules wpr + WHERE ($3::bigint IS NULL OR wpr.id = $3) + ), + sys AS ( + SELECT id FROM cost_items + WHERE code = 'sys_work_order_reward' AND status = 'active' + ), + missing AS ( + SELECT + t.rule_id, + t.amount_fen, + si.id AS cost_item_id, + EXISTS ( + SELECT 1 FROM cost_versions h + WHERE h.cost_item_id = si.id + AND h.scope_type = 'rule' + AND h.work_product_rule_id = t.rule_id + ) AS has_history + FROM target t + CROSS JOIN sys si + WHERE t.amount_fen > 0 + AND NOT EXISTS ( + SELECT 1 FROM cost_versions v + WHERE v.cost_item_id = si.id + AND v.scope_type = 'rule' + AND v.work_product_rule_id = t.rule_id + AND v.status = 'active' + ) + ) + INSERT INTO cost_versions ( + cost_item_id, scope_type, provider, shop_id, canonical_product_id, + work_product_rule_id, amount_fen, rate_bps, effective_from, effective_to, + source, status, change_reason, created_by_admin_user_id, + approved_by_admin_user_id, created_at, updated_at + ) + SELECT + m.cost_item_id, + 'rule', + '', + '', + NULL, + m.rule_id, + m.amount_fen, + 0, + CASE WHEN m.has_history THEN $1::timestamptz ELSE $2::timestamptz END, + NULL, + 'import', + 'active', + CASE + WHEN wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 + THEN '模板按件发单单价（数量取订单 SKU）' + ELSE '模板固定发单金额' + END, + NULL, + NULL, + $1, + $1 + FROM missing m + JOIN work_product_rules wpr ON wpr.id = m.rule_id + ON CONFLICT DO NOTHING + RETURNING id + `, + [now, WORK_ORDER_TEMPLATE_COST_BASELINE, scopedRuleId], + ) + + const linked = await query( + ` + WITH rule_version AS ( + SELECT DISTINCT ON (v.work_product_rule_id) + v.work_product_rule_id, + v.id AS cost_version_id, + v.cost_item_id + FROM cost_versions v + JOIN cost_items ci ON ci.id = v.cost_item_id + AND ci.code = 'sys_work_order_reward' + WHERE v.scope_type = 'rule' + AND v.status = 'active' + AND ($1::bigint IS NULL OR v.work_product_rule_id = $1) + ORDER BY v.work_product_rule_id, v.effective_from DESC, v.id DESC + ) + UPDATE cost_snapshots cs + SET cost_version_id = rv.cost_version_id + FROM work_orders wo + JOIN rule_version rv ON rv.work_product_rule_id = wo.product_rule_id + WHERE cs.order_item_id = wo.order_item_id + AND cs.cost_item_id = rv.cost_item_id + AND cs.superseded_at IS NULL + AND cs.cost_version_id <> rv.cost_version_id + AND EXISTS ( + SELECT 1 FROM cost_versions gv + WHERE gv.id = cs.cost_version_id AND gv.scope_type = 'global' + ) + AND ($1::bigint IS NULL OR wo.product_rule_id = $1) + `, + [scopedRuleId], + ) + + const summary = await query( + ` + SELECT + COUNT(*)::int AS rule_count, + COUNT(*) FILTER ( + WHERE wpr.unit_price_fen > 0 AND wpr.reward_amount = 0 + )::int AS unit_price_rule_count, + COUNT(v.id) FILTER ( + WHERE v.scope_type = 'rule' AND v.status = 'active' + )::int AS active_version_count + FROM work_product_rules wpr + LEFT JOIN cost_versions v + ON v.cost_item_id = ( + SELECT id FROM cost_items WHERE code = 'sys_work_order_reward' LIMIT 1 + ) + AND v.work_product_rule_id = wpr.id + WHERE ($1::bigint IS NULL OR wpr.id = $1) + `, + [scopedRuleId], + ) + + const row = summary.rows[0] + return { + ruleId: scopedRuleId, + ruleCount: Number(row?.rule_count || 0), + unitPriceRuleCount: Number(row?.unit_price_rule_count || 0), + activeVersionCount: Number(row?.active_version_count || 0), + createdCount: inserted.rows.length, + retiredCount: retired.rowCount || 0, + linkedSnapshotCount: linked.rowCount || 0, + } +}