dachui/order_site
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构：硬切商品来源并移除旧运维入口

Browse Source
This commit is contained in:
yml2213
2026-11-08 19:44:27 +08:00
parent f8cbb645fe
commit 2ebc4a0ec7
24 changed files with 273 additions and 734 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/scripts/merge-duplicate-products.ts
-253
View File
@@ -1,253 +0,0 @@
/**
* 合并同一内部 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],
)
await client.query(
'UPDATE sku_channel_routes 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(
`UPDATE sku_channel_routes
SET route_role = 'fallback',
priority = GREATEST(priority, 90),
updated_at = NOW()
WHERE internal_sku_id = (
SELECT internal_sku_id FROM product_catalog WHERE id = $1
)
AND channel <> 'affiliate_dash'
AND enabled = TRUE
AND route_role = 'primary'`,
[group.keeper_id],
)
await query(
`
INSERT INTO sku_channel_routes (
internal_sku_id, channel, route_role, priority, enabled, canonical_product_id, source
)
SELECT product.internal_sku_id, merged.channel, merged.route_role,
merged.priority, TRUE, product.id, 'merged'
FROM product_catalog product
CROSS JOIN (
VALUES
('affiliate_dash', 'primary', 10),
('manual', 'fallback', 20)
) merged(channel, route_role, priority)
WHERE product.id = $1
AND product.internal_sku_id IS NOT NULL
ON CONFLICT (internal_sku_id, channel) DO UPDATE SET
priority = LEAST(sku_channel_routes.priority, EXCLUDED.priority),
route_role = EXCLUDED.route_role,
enabled = TRUE,
canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, sku_channel_routes.canonical_product_id),
updated_at = NOW()
`,
[group.keeper_id],
)
// 只删除已经没有引用的重复商品，仍被引用就归档。
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())
apps/backend/scripts/rebuild-after-cutover.ts
-113
View File
@@ -1,113 +0,0 @@
import process from 'node:process'
import { closeDb, query } from '../src/db/client.js'
import { assignPendingOrderItemChannels } from '../src/services/fulfillment/channel-assign-service.js'
import { normalizeOrderItemProducts } from '../src/services/cost-center/product-normalize-service.js'
import { syncChannelCostSnapshots } from '../src/services/cost-center/cost-snapshot-service.js'
import { rebuildChannelSalesDailyFacts } from '../src/services/finance/sales-facts-channel-service.js'
import { syncFinanceLedgerFromFacts } from '../src/services/finance/finance-ledger-service.js'
import { formatShanghaiDate } from '../src/utils/finance-date-range.js'
/**
* 断代后重建：从 cutover 日起重跑三条管道，生成成本快照、销售事实与期初分录。
*
* 幂等：渠道判定/商品归一/成本快照都按「缺什么补什么」推进，事实按日先删后插，
* 总账按「event_type + 日期」幂等。可以反复执行。
*
* 用法：
* npm run finance:rebuild-cutover -- --date-from=2026-10-31 --date-to=2026-10-31
* npm run finance:rebuild-cutover -- --max-batches=200
*/
const options = parseOptions(process.argv.slice(2))
const dateFrom = String(options.dateFrom || '2026-10-31').slice(0, 10)
const dateTo = String(options.dateTo || formatShanghaiDate(new Date())).slice(0, 10)
const maxBatches = clampInt(options.maxBatches, 200, 1, 2000)
const batchSize = clampInt(options.batchSize, 2000, 1, 5000)
async function main() {
await assertSchema()
const startedAt = Date.now()
const channel = await assignPendingOrderItemChannels({ batchSize, maxBatches })
const product = await normalizeOrderItemProducts({ batchSize, maxBatches })
const cost = await syncChannelCostSnapshots({ batchSize, maxBatches })
const facts = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo })
const ledger = await syncFinanceLedgerFromFacts({ dateFrom, dateTo })
const summary = await query<{
assigned: number
pending: number
archived: number
facts: number
snapshots: number
}>(
`
SELECT
COUNT(*) FILTER (WHERE status = 'assigned')::int AS assigned,
COUNT(*) FILTER (WHERE status = 'pending')::int AS pending,
COUNT(*) FILTER (WHERE status = 'archived')::int AS archived,
(SELECT COUNT(*)::int FROM sales_daily_facts) AS facts,
(SELECT COUNT(*)::int FROM cost_snapshots WHERE superseded_at IS NULL) AS snapshots
FROM fulfillments
`,
)
console.log(
JSON.stringify(
{
dateFrom,
dateTo,
channel,
product,
cost,
facts,
ledger,
totals: summary.rows[0],
durationMs: Date.now() - startedAt,
},
null,
2,
),
)
}
main()
.catch((error) => {
console.error(error instanceof Error ? error.message : error)
process.exitCode = 1
})
.finally(async () => {
await closeDb()
})
async function assertSchema() {
const ready = await query<{ ready: boolean }>(
`SELECT
to_regclass('public.internal_skus') IS NOT NULL
AND to_regclass('public.sku_channel_routes') IS NOT NULL AS ready`,
)
if (!ready.rows[0]?.ready) {
throw new Error('缺少统一 SKU 路由迁移产物；请先执行 npm run db:migrate')
}
}
function parseOptions(args: string[]) {
const result: Record<string, string> = {}
for (const arg of args) {
if (!arg.startsWith('--')) continue
const body = arg.slice(2)
const index = body.indexOf('=')
const rawKey = index === -1 ? body : body.slice(0, index)
const value = index === -1 ? 'true' : body.slice(index + 1)
result[rawKey] = value
result[rawKey.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())] = value
}
return result
}
function clampInt(value: unknown, fallback: number, min: number, max: number) {
const parsed = Number(value)
if (!Number.isInteger(parsed)) return fallback
return Math.min(max, Math.max(min, parsed))
}
apps/backend/scripts/reclassify-post-cutover.ts
-181
View File
@@ -1,181 +0,0 @@
/**
* 断代后订单按新的商品渠道能力重判。
*
* 仅处理：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 route.channel
FROM sku_channel_routes route
JOIN internal_skus sku ON sku.id = route.internal_sku_id
WHERE route.enabled
AND route.route_role = 'primary'
AND route.channel IN ('affiliate_dash', 'work_order', 'manual', 'third_party_voucher')
AND sku.internal_sku_key = resolve_internal_sku(
oi.manual_sku_key, oi.sku_name, oi.sku_code
)
ORDER BY
route.priority ASC,
CASE route.channel
WHEN 'affiliate_dash' THEN 1
WHEN 'work_order' THEN 2
WHEN 'third_party_voucher' THEN 3
WHEN 'manual' THEN 4
ELSE 9
END,
route.channel 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 ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
)
)
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 <> 'cancelled'
)
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())