feat(scripts): 重复商品合并、断代重判与本地演示数据

This commit is contained in:
yml2213
2026-11-02 13:17:15 +08:00
parent 3c4c81003f
commit c79a7e0790
4 changed files with 573 additions and 1 deletions
+4 -1
View File
@@ -34,7 +34,10 @@
"typecheck": "tsc -p tsconfig.json --noEmit",
"check": "npm run format:check && npm run check:sql && npm run lint:check && npm run typecheck && npm test && npm run build",
"start": "node dist/index.js",
"start:src": "tsx src/index.ts"
"start:src": "tsx src/index.ts",
"products:merge-duplicates": "tsx scripts/merge-duplicate-products.ts",
"products:reclassify": "tsx scripts/reclassify-post-cutover.ts",
"demo:seed-finance": "tsx scripts/seed-local-finance-demo.ts"
},
"dependencies": {
"@alicloud/dysmsapi20170525": "^4.6.0",
@@ -0,0 +1,235 @@
/**
* 合并同一内部 SKU 的重复商品(aff: / man:)。
*
* 默认 dry-run;--apply 才写库。
* 保留 aff: 商品作为主商品,把 man: 商品的引用全部迁过去,
* 再把人工/自动两种渠道能力挂到主商品上,并生成中文内部 SKU 的自动发货规则。
*
* 用法:
* npm run products:merge-duplicates -- --dry-run
* npm run products:merge-duplicates -- --apply
*/
import process from 'node:process'
import { closeDb, query, withTransaction } from '../src/db/client.js'
import { refreshFactsForDates } from '../src/services/finance/refresh-facts.js'
type GroupRow = {
canonical_name: string
keeper_id: number
keeper_sku: string
duplicate_ids: number[]
duplicate_skus: string[]
}
async function loadGroups(): Promise<GroupRow[]> {
const result = await query<{
canonical_name: string
keeper_id: number
keeper_sku: string
duplicate_ids: number[]
duplicate_skus: string[]
}>(
`
SELECT
canonical_name,
(array_agg(id ORDER BY (canonical_sku LIKE 'aff:%') DESC, id))[1] AS keeper_id,
(array_agg(canonical_sku ORDER BY (canonical_sku LIKE 'aff:%') DESC, id))[1] AS keeper_sku,
array_remove(array_agg(id ORDER BY id), (array_agg(id ORDER BY (canonical_sku LIKE 'aff:%') DESC, id))[1]) AS duplicate_ids,
array_remove(array_agg(canonical_sku ORDER BY id), (array_agg(canonical_sku ORDER BY (canonical_sku LIKE 'aff:%') DESC, id))[1]) AS duplicate_skus
FROM product_catalog
WHERE status = 'active'
GROUP BY canonical_name
HAVING count(*) > 1
AND bool_or(canonical_sku LIKE 'aff:%')
AND bool_or(canonical_sku LIKE 'man:%')
ORDER BY canonical_name
`,
)
return result.rows
}
async function mergeGroup(
group: GroupRow,
): Promise<{ mappings: number; versions: number; facts: number }> {
let mappings = 0
let versions = 0
let facts = 0
for (const duplicateId of group.duplicate_ids) {
await withTransaction(async (client) => {
await client.query(
'UPDATE order_items SET canonical_product_id = $2, updated_at = NOW() WHERE canonical_product_id = $1',
[duplicateId, group.keeper_id],
)
await client.query(
'UPDATE cost_snapshots SET canonical_product_id = $2 WHERE canonical_product_id = $1',
[duplicateId, group.keeper_id],
)
await client.query(
'UPDATE manual_fulfillment_configs SET canonical_product_id = $2, updated_at = NOW() WHERE canonical_product_id = $1',
[duplicateId, group.keeper_id],
)
// 映射:先删掉会与主商品冲突的重复映射,再迁移其余。
const conflict = await client.query(
`
DELETE FROM channel_product_mappings d
USING channel_product_mappings k
WHERE d.canonical_product_id = $1 AND k.canonical_product_id = $2
AND d.provider = k.provider AND d.shop_id = k.shop_id
AND d.platform_sku_code = k.platform_sku_code AND d.match_type = k.match_type
AND d.platform_sku_name_key = k.platform_sku_name_key
`,
[duplicateId, group.keeper_id],
)
mappings += conflict.rowCount || 0
const movedMappings = await client.query(
'UPDATE channel_product_mappings SET canonical_product_id = $2 WHERE canonical_product_id = $1',
[duplicateId, group.keeper_id],
)
mappings += movedMappings.rowCount || 0
// 成本版本:有唯一键冲突就中止该组,避免静默丢价格。
const versionConflict = await client.query(
`
SELECT 1 FROM cost_versions d
JOIN cost_versions k
ON k.cost_item_id = d.cost_item_id
AND k.scope_type = d.scope_type
AND k.provider = d.provider
AND k.shop_id = d.shop_id
AND COALESCE(k.work_product_rule_id, 0) = COALESCE(d.work_product_rule_id, 0)
AND k.effective_from = d.effective_from
WHERE d.canonical_product_id = $1 AND k.canonical_product_id = $2
LIMIT 1
`,
[duplicateId, group.keeper_id],
)
if (versionConflict.rowCount) {
throw new Error(`商品 ${duplicateId} 的成本版本与主商品冲突,已中止`)
}
const movedVersions = await client.query(
'UPDATE cost_versions SET canonical_product_id = $2 WHERE canonical_product_id = $1',
[duplicateId, group.keeper_id],
)
versions += movedVersions.rowCount || 0
// 事实:先删冲突行,其余迁移;迁移后统一重算受影响日期。
const conflictFacts = await client.query(
`
DELETE FROM sales_daily_facts d
USING sales_daily_facts k
WHERE d.canonical_product_id = $1 AND k.canonical_product_id = $2
AND d.stat_date = k.stat_date AND d.provider = k.provider
AND d.shop_id = k.shop_id AND d.channel = k.channel
`,
[duplicateId, group.keeper_id],
)
facts += conflictFacts.rowCount || 0
const movedFacts = await client.query(
'UPDATE sales_daily_facts SET canonical_product_id = $2 WHERE canonical_product_id = $1',
[duplicateId, group.keeper_id],
)
facts += movedFacts.rowCount || 0
})
}
await query(
`
INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback, enabled, source)
VALUES ($1, 'affiliate_dash', 10, FALSE, TRUE, 'merged'), ($1, 'manual', 20, TRUE, TRUE, 'merged')
ON CONFLICT (product_id, channel) DO UPDATE
SET priority = LEAST(product_channel_capabilities.priority, EXCLUDED.priority),
is_fallback = product_channel_capabilities.is_fallback OR EXCLUDED.is_fallback,
updated_at = NOW()
`,
[group.keeper_id],
)
await query(
`
INSERT INTO sku_channel_rules (
provider, shop_id, match_key_type, match_key, channel, priority,
source, enabled, note, created_at, updated_at
)
VALUES ('', '', 'internal_sku', $1, 'affiliate_dash', 10, 'affiliate', TRUE,
'合并重复商品后按内部SKU生成', NOW(), NOW())
ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING
`,
[group.canonical_name],
)
// 只删除已经没有引用的重复商品,仍被引用就归档。
for (const duplicateId of group.duplicate_ids) {
const refs = await query<{ total: number }>(
`
SELECT (
(SELECT count(*) FROM order_items WHERE canonical_product_id = $1) +
(SELECT count(*) FROM cost_snapshots WHERE canonical_product_id = $1) +
(SELECT count(*) FROM sales_daily_facts WHERE canonical_product_id = $1) +
(SELECT count(*) FROM cost_versions WHERE canonical_product_id = $1) +
(SELECT count(*) FROM channel_product_mappings WHERE canonical_product_id = $1) +
(SELECT count(*) FROM manual_fulfillment_configs WHERE canonical_product_id = $1)
)::int AS total
`,
[duplicateId],
)
if (Number(refs.rows[0]?.total || 0) === 0) {
await query('DELETE FROM product_catalog WHERE id = $1', [duplicateId])
} else {
await query(
"UPDATE product_catalog SET status = 'archived', updated_at = NOW() WHERE id = $1",
[duplicateId],
)
}
}
return { mappings, versions, facts }
}
async function main() {
const apply = process.argv.includes('--apply')
const groups = await loadGroups()
console.log(
JSON.stringify(
{
mode: apply ? 'apply' : 'dry-run',
groups: groups.map((g) => ({
name: g.canonical_name,
keeper: g.keeper_sku,
duplicates: g.duplicate_skus,
})),
},
null,
2,
),
)
if (!apply || groups.length === 0) return
const dates = await query<{ stat_date: string }>(
`SELECT DISTINCT stat_date::text FROM sales_daily_facts WHERE stat_date >= '2026-10-31'`,
)
const affectedDates = dates.rows.map((row) => row.stat_date)
let totals = { mappings: 0, versions: 0, facts: 0 }
for (const group of groups) {
const result = await mergeGroup(group)
totals = {
mappings: totals.mappings + result.mappings,
versions: totals.versions + result.versions,
facts: totals.facts + result.facts,
}
console.log(
`merged ${group.canonical_name}: keeper=${group.keeper_sku} dup=${group.duplicate_skus.join(',')}`,
)
}
await refreshFactsForDates(affectedDates)
console.log(JSON.stringify({ totals, affectedDates }, null, 2))
}
main()
.catch((error) => {
console.error(error)
process.exitCode = 1
})
.finally(() => closeDb())
@@ -0,0 +1,167 @@
/**
* 断代后订单按新的商品渠道能力重判。
*
* 仅处理:assigned、未锁定、无成功履约证据、有商品能力的订单项。
* 默认 dry-run;--apply 才写库。
*/
import process from 'node:process'
import { closeDb, query, withTransaction } from '../src/db/client.js'
import { syncCostSnapshotsForItems } from '../src/services/cost-center/cost-snapshot-service.js'
import { refreshFactsForDates } from '../src/services/finance/refresh-facts.js'
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',
}
type Candidate = {
id: number
current_channel: string
desired_channel: string
basis_date: string
}
async function loadCandidates(limit: number): Promise<Candidate[]> {
const result = await query<Candidate>(
`
SELECT oi.id, oi.fulfillment_channel AS current_channel,
cap.channel AS desired_channel,
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
JOIN LATERAL (
SELECT c.channel
FROM product_channel_capabilities c
WHERE c.product_id = oi.canonical_product_id
AND c.enabled
AND c.channel IN ('affiliate_dash', 'manual', 'third_party_voucher')
ORDER BY c.priority ASC
LIMIT 1
) cap ON TRUE
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_locked = FALSE
AND COALESCE(o.paid_at, o.created_at) >= '2026-10-31 00:00:00+08'
AND oi.fulfillment_source NOT IN ('operator_assigned', 'manual_confirm')
AND (
cap.channel <> 'affiliate_dash'
OR EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
)
)
AND NOT EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id
AND ft.executor_key = 'affiliate_dash'
AND ft.task_status IN ('redeemed', 'paid')
)
AND NOT EXISTS (
SELECT 1 FROM work_orders wo
WHERE wo.order_item_id = oi.id AND wo.status = 'accepted'
)
AND NOT EXISTS (
SELECT 1 FROM cost_snapshots cs
JOIN cost_items ci ON ci.id = cs.cost_item_id
WHERE cs.order_item_id = oi.id
AND ci.code = 'sys_manual_service'
AND cs.superseded_at IS NULL
AND cs.reversed_at IS NULL
)
ORDER BY oi.id
LIMIT $1
`,
[limit],
)
return result.rows
}
async function applyOne(row: Candidate): Promise<boolean> {
const previous = row.current_channel
if (!previous || previous === row.desired_channel) return false
const changed = await withTransaction(async (client) => {
const result = await client.query(
`
UPDATE order_items
SET original_channel = COALESCE(original_channel, $2),
fulfillment_channel = $3,
fulfillment_source = 'auto_rule',
channel_switch_reason = 'reclassify',
channel_switched_at = NOW(),
updated_at = NOW()
WHERE id = $1
AND fulfillment_status = 'assigned'
AND fulfillment_locked = FALSE
AND fulfillment_channel = $2
RETURNING id
`,
[row.id, previous, row.desired_channel],
)
if ((result.rowCount || 0) === 0) return false
const code = CHANNEL_COST_CODE[previous]
if (code) {
await client.query(
`
UPDATE cost_snapshots
SET reversed_at = NOW(), reversal_reason = 'reclassify'
WHERE order_item_id = $1
AND cost_item_id = (SELECT id FROM cost_items WHERE code = $2)
AND superseded_at IS NULL
AND reversed_at IS NULL
`,
[row.id, code],
)
}
return true
})
if (!changed) return false
await syncCostSnapshotsForItems([row.id]).catch(() => undefined)
return true
}
async function main() {
const apply = process.argv.includes('--apply')
const limit = 5000
const rows = await loadCandidates(limit)
const changes = rows.filter((row) => row.current_channel !== row.desired_channel)
console.log(
JSON.stringify(
{
mode: apply ? 'apply' : 'dry-run',
scanned: rows.length,
toChange: changes.length,
sample: changes.slice(0, 20).map((row) => ({
id: row.id,
from: row.current_channel,
to: row.desired_channel,
date: row.basis_date,
})),
},
null,
2,
),
)
if (!apply || changes.length === 0) return
let switched = 0
const dates = new Set<string>()
for (const row of changes) {
if (await applyOne(row)) {
switched += 1
dates.add(row.basis_date)
}
}
await refreshFactsForDates([...dates])
console.log(JSON.stringify({ switched, dates: [...dates].sort() }, null, 2))
}
main()
.catch((error) => {
console.error(error)
process.exitCode = 1
})
.finally(() => closeDb())
@@ -0,0 +1,167 @@
/**
* 本地财务流程演示数据:待分类、自动发货、人工、兜底、未分类。
* 用法:npx tsx scripts/seed-local-finance-demo.ts
*/
import { closeDb, query } from '../src/db/client.js'
const SHOPS = ['sim-voucher', 'sim-aff', 'sim-manual', 'sim-fallback', 'sim-unmapped']
const TODAY = '2026-11-02 10:00:00+08'
async function cleanup() {
await query(
`DELETE FROM fulfillment_tasks WHERE order_item_id IN (SELECT id FROM order_items WHERE order_id IN (SELECT id FROM orders WHERE shop_id = ANY($1)))`,
[SHOPS],
)
await query(
`DELETE FROM cost_snapshots WHERE order_item_id IN (SELECT id FROM order_items WHERE order_id IN (SELECT id FROM orders WHERE shop_id = ANY($1)))`,
[SHOPS],
)
await query(`DELETE FROM sales_daily_facts WHERE shop_id = ANY($1)`, [SHOPS])
await query(
`DELETE FROM order_items WHERE order_id IN (SELECT id FROM orders WHERE shop_id = ANY($1))`,
[SHOPS],
)
await query(`DELETE FROM orders WHERE shop_id = ANY($1)`, [SHOPS])
await query(
`DELETE FROM sku_channel_rules WHERE match_key IN ('SIM_AFF_1', 'SIMV001', 'SIMM001')`,
)
await query(
`DELETE FROM product_channel_capabilities WHERE product_id IN (SELECT id FROM product_catalog WHERE canonical_sku = 'aff:SIM_AFF_1')`,
)
await query(
`DELETE FROM cost_versions WHERE canonical_product_id IN (SELECT id FROM product_catalog WHERE canonical_sku = 'aff:SIM_AFF_1')`,
)
await query(`DELETE FROM product_catalog WHERE canonical_sku = 'aff:SIM_AFF_1'`)
await query(`DELETE FROM manual_fulfillment_configs WHERE sku_key = '模拟人工商品'`)
await query(
`DELETE FROM product_catalog WHERE canonical_sku IN ('tpv:模拟三方卡券', 'tpv:模拟三方卡券B')`,
)
}
async function createOrder(shop: string, orderNo: string, amountFen = 1000) {
const result = await query<{ id: number }>(
`INSERT INTO orders (provider, platform, shop_id, platform_order_id, order_status, pay_status, total_amount, paid_at, created_at, updated_at)
VALUES ('91kaquan','91kaquan',$1,$2,'paid','paid',$3,$4,NOW(),NOW()) RETURNING id`,
[shop, orderNo, amountFen, TODAY],
)
return Number(result.rows[0]!.id)
}
async function createItem(input: {
orderId: number
skuCode: string
skuName: string
channel?: string
status?: string
productId?: number | null
}) {
const result = await query<{ id: number }>(
`INSERT INTO order_items (
order_id, sku_code, sku_name, quantity, revenue_amount_fen, revenue_basis,
fulfillment_channel, fulfillment_status, fulfillment_source, canonical_product_id,
created_at, updated_at
) VALUES ($1,$2,$3,1,1000,'order_total',$4,$5,$6,$7,NOW(),NOW()) RETURNING id`,
[
input.orderId,
input.skuCode,
input.skuName,
input.channel || null,
input.status || 'pending',
input.channel ? 'auto_rule' : '',
input.productId || null,
],
)
return Number(result.rows[0]!.id)
}
async function main() {
await cleanup()
// 自动发货演示商品(带采购价)
const affProduct = await query<{ id: number }>(
`INSERT INTO product_catalog (canonical_sku, canonical_name, category, unit_name, status, attributes_json, created_at, updated_at)
VALUES ('aff:SIM_AFF_1','模拟自动发货商品','试玩','件','active','{}',NOW(),NOW()) RETURNING id`,
)
const affProductId = Number(affProduct.rows[0]!.id)
const affCostItem = await query<{ id: number }>(
`SELECT id FROM cost_items WHERE code='sys_affiliate_purchase'`,
)
await query(
`INSERT INTO cost_versions (cost_item_id, scope_type, canonical_product_id, amount_fen, rate_bps, effective_from, status, change_reason, created_at, updated_at)
VALUES ($1,'product',$2,500,0,'2026-10-01T00:00:00+08:00','active','本地演示采购价',NOW(),NOW())`,
[affCostItem.rows[0]!.id, affProductId],
)
await query(
`INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback, enabled, source)
VALUES ($1,'affiliate_dash',10,FALSE,TRUE,'demo'), ($1,'manual',20,TRUE,TRUE,'demo')`,
[affProductId],
)
await query(
`INSERT INTO sku_channel_rules (provider, shop_id, match_key_type, match_key, channel, priority, source, enabled, note, created_at, updated_at)
VALUES ('','','sku_code','SIM_AFF_1','affiliate_dash',60,'operator',TRUE,'本地演示',NOW(),NOW())`,
)
// 人工演示配置
await query(
`INSERT INTO manual_fulfillment_configs (sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at)
VALUES ('模拟人工商品','nanhang',100,'fixed',TRUE,'本地演示',NOW(),NOW())`,
)
// 1. 三方卡券:待分类,指定后自动建商品
const voucherOrder = await createOrder('sim-voucher', 'SIM-VOUCHER-1', 1950)
await createItem({ orderId: voucherOrder, skuCode: 'SIMV001', skuName: '模拟三方卡券' })
// 2. 自动发货:规则命中
const affOrder = await createOrder('sim-aff', 'SIM-AFF-1', 1000)
await createItem({ orderId: affOrder, skuCode: 'SIM_AFF_1', skuName: '模拟自动发货商品' })
// 3. 人工:待分类,指定人工后挂配置
const manualOrder = await createOrder('sim-manual', 'SIM-MANUAL-1', 1000)
await createItem({ orderId: manualOrder, skuCode: 'SIMM001', skuName: '模拟人工商品' })
// 4. 自动发货失败:应转人工并冲销采购成本
const fallbackOrder = await createOrder('sim-fallback', 'SIM-FALLBACK-1', 1000)
const fallbackProduct = await query<{ id: number }>(
`SELECT id FROM product_catalog WHERE canonical_sku='aff:rock_black_tshirt' LIMIT 1`,
)
const fallbackItem = await createItem({
orderId: fallbackOrder,
skuCode: 'SIMFB001',
skuName: '黑色ROCK长袖T恤',
channel: 'affiliate_dash',
status: 'assigned',
productId: fallbackProduct.rows[0]?.id,
})
await query(
`INSERT INTO fulfillment_tasks (order_id, order_item_id, unit_index, task_no, provider, platform, shop_id, platform_order_id, profile_id, executor_key, task_status, delivery_status, created_at, updated_at)
VALUES ($1,$2,1,'SIM-FB-TASK','91kaquan','kuaishou','sim-fallback','SIM-FALLBACK-1',3,'affiliate_dash','closed','failed',NOW(),NOW())`,
[fallbackOrder, fallbackItem],
)
// 5. 未分类:无规则、无配置
const unmappedOrder = await createOrder('sim-unmapped', 'SIM-UNMAPPED-1', 1000)
await createItem({ orderId: unmappedOrder, skuCode: 'SIMUNMAPPED001', skuName: '模拟未分类商品' })
console.log(
JSON.stringify(
{
shops: SHOPS,
voucherOrder,
affOrder,
manualOrder,
fallbackOrder,
unmappedOrder,
note: '等待调度器处理;可在后台履约渠道-待分类队列、商品与成本中心观察',
},
null,
2,
),
)
}
main()
.catch((error) => {
console.error(error)
process.exitCode = 1
})
.finally(() => closeDb())