feat(finance): 商品归一、成本快照与渠道销售事实三条管道
- 商品归一统一为一个匹配器，禁止自动建商品，未命中进未归一队列 - 成本快照收敛为唯一入口：商品/规则/全局版本解析，工单验收自动升级 actual - 异常改为 SKU 级聚合，补齐后自动关闭 - 销售事实按渠道汇总，pending 收入进 unclassified 桶，归档不进报表
This commit is contained in:
yml2213
@@ -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<string, unknown>
|
||||
}
|
||||
|
||||
/** 幂等上浮一条聚合问题：已存在则累加明细并刷新 last_seen。 */
|
||||
export async function upsertAggregatedException(input: UpsertExceptionInput): Promise<void> {
|
||||
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<number> {
|
||||
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<number> {
|
||||
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<ExceptionQueueRow>(
|
||||
`
|
||||
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<number> {
|
||||
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
|
||||
}
|
||||
@@ -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<string, CostItemRow>([
|
||||
['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<string, unknown> = {}) {
|
||||
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)
|
||||
})
|
||||
@@ -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<string, string> = {
|
||||
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<string, CostItemRow>,
|
||||
versions: CostVersionRow[],
|
||||
pricingModeByItemId: Map<number, string>,
|
||||
): 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<SyncCostSnapshotsResult> {
|
||||
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<number, string>()
|
||||
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<CostCandidateItem>(
|
||||
`
|
||||
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<CostResolution>
|
||||
}> = []
|
||||
|
||||
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<CostResolution> }>,
|
||||
): Promise<number> {
|
||||
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<Map<string, CostItemRow>> {
|
||||
const rows = await query<CostItemRow>(
|
||||
`SELECT id, code, pricing_mode FROM cost_items WHERE status = 'active'`,
|
||||
)
|
||||
const map = new Map<string, CostItemRow>()
|
||||
for (const row of rows.rows) map.set(row.code, row)
|
||||
return map
|
||||
}
|
||||
|
||||
async function loadActiveVersions(): Promise<CostVersionRow[]> {
|
||||
const rows = await query<CostVersionRow>(
|
||||
`
|
||||
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
|
||||
}
|
||||
@@ -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<string, unknown> = {}) {
|
||||
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)
|
||||
})
|
||||
@@ -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_code>` 商品直接命中（该渠道的商品就是 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<string, number>
|
||||
bySkuName: Map<string, number>
|
||||
bySkuSeries: Map<string, number>
|
||||
byNameSeries: Map<string, number>
|
||||
affSkus: Map<string, number>
|
||||
}
|
||||
|
||||
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<NormalizeProductsResult> {
|
||||
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<ItemRow>(
|
||||
`
|
||||
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<MappingIndex> {
|
||||
const mappings = await query<MappingRow>(
|
||||
`
|
||||
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),
|
||||
})),
|
||||
)
|
||||
}
|
||||
@@ -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<ChannelFactsResult> {
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user