diff --git a/apps/backend/package.json b/apps/backend/package.json index 46a33a59..fb25d538 100644 --- a/apps/backend/package.json +++ b/apps/backend/package.json @@ -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", diff --git a/apps/backend/scripts/merge-duplicate-products.ts b/apps/backend/scripts/merge-duplicate-products.ts new file mode 100644 index 00000000..fe5f4ea3 --- /dev/null +++ b/apps/backend/scripts/merge-duplicate-products.ts @@ -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 { + 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()) diff --git a/apps/backend/scripts/reclassify-post-cutover.ts b/apps/backend/scripts/reclassify-post-cutover.ts new file mode 100644 index 00000000..5af914b5 --- /dev/null +++ b/apps/backend/scripts/reclassify-post-cutover.ts @@ -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 = { + 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 { + const result = await query( + ` + 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 { + 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() + 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()) diff --git a/apps/backend/scripts/seed-local-finance-demo.ts b/apps/backend/scripts/seed-local-finance-demo.ts new file mode 100644 index 00000000..288f45a5 --- /dev/null +++ b/apps/backend/scripts/seed-local-finance-demo.ts @@ -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())