diff --git a/apps/backend/package.json b/apps/backend/package.json index bcedc9f2..e451f6d0 100644 --- a/apps/backend/package.json +++ b/apps/backend/package.json @@ -9,8 +9,6 @@ "db:migrate": "tsx src/db/migrate.ts", "db:migrate:create": "tsx scripts/create-migration.ts", "db:migrate:status": "tsx scripts/migration-status.ts", - "finance:sync": "tsx scripts/sync-cost-finance.ts", - "finance:rebuild-cutover": "tsx scripts/rebuild-after-cutover.ts", "finance:rebuild-derived": "tsx scripts/rebuild-finance-derived.ts", "audit:finance": "tsx scripts/audit-finance-derived.ts", "revenue:backfill": "tsx scripts/backfill-revenue-basis.ts", @@ -39,8 +37,6 @@ "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", - "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": { diff --git a/apps/backend/scripts/merge-duplicate-products.ts b/apps/backend/scripts/merge-duplicate-products.ts deleted file mode 100644 index 343bfecf..00000000 --- a/apps/backend/scripts/merge-duplicate-products.ts +++ /dev/null @@ -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 { - 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()) diff --git a/apps/backend/scripts/rebuild-after-cutover.ts b/apps/backend/scripts/rebuild-after-cutover.ts deleted file mode 100644 index bf55fff6..00000000 --- a/apps/backend/scripts/rebuild-after-cutover.ts +++ /dev/null @@ -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 = {} - 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)) -} diff --git a/apps/backend/scripts/reclassify-post-cutover.ts b/apps/backend/scripts/reclassify-post-cutover.ts deleted file mode 100644 index 2c98f059..00000000 --- a/apps/backend/scripts/reclassify-post-cutover.ts +++ /dev/null @@ -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 = { - 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 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 { - 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/src/db/migrations/168_product_source_kind_constraint.sql b/apps/backend/src/db/migrations/168_product_source_kind_constraint.sql new file mode 100644 index 00000000..32fce05c --- /dev/null +++ b/apps/backend/src/db/migrations/168_product_source_kind_constraint.sql @@ -0,0 +1,29 @@ +-- 168_product_source_kind_constraint.sql +-- +-- 商品来源统一由 product_catalog.source_kind 表达。 +-- 运行期不再解析 canonical_sku 前缀;数据库同步收紧允许值。 + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'chk_product_catalog_source_kind' + ) THEN + ALTER TABLE product_catalog + ADD CONSTRAINT chk_product_catalog_source_kind + CHECK ( + source_kind IN ( + 'affiliate_dash', + 'work_order', + 'third_party_voucher', + 'manual', + 'unclassified' + ) + ) + NOT VALID; + END IF; +END $$; + +ALTER TABLE product_catalog + VALIDATE CONSTRAINT chk_product_catalog_source_kind; diff --git a/apps/backend/src/services/cost-center/affiliate-product-service.ts b/apps/backend/src/services/cost-center/affiliate-product-service.ts index 381277fb..ee8f74d2 100644 --- a/apps/backend/src/services/cost-center/affiliate-product-service.ts +++ b/apps/backend/src/services/cost-center/affiliate-product-service.ts @@ -110,7 +110,7 @@ export async function syncAdminAffiliateDashProducts() { GROUP BY sku_code, sku_name ) INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, + canonical_sku, canonical_name, category, unit_name, source_kind, status, attributes_json, created_at, updated_at ) SELECT @@ -118,6 +118,7 @@ export async function syncAdminAffiliateDashProducts() { ranked.sku_name, '', CASE WHEN ranked.sku_name LIKE '%个' THEN '个' ELSE '件' END, + 'affiliate_dash', 'active', jsonb_build_object( 'source', 'affiliate_dash_task', @@ -133,6 +134,7 @@ export async function syncAdminAffiliateDashProducts() { ON CONFLICT (canonical_sku) DO UPDATE SET canonical_name = CASE WHEN product_catalog.canonical_name = '' THEN EXCLUDED.canonical_name ELSE product_catalog.canonical_name END, unit_name = EXCLUDED.unit_name, + source_kind = EXCLUDED.source_kind, attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json, updated_at = $1 RETURNING (xmax = 0) AS inserted @@ -319,12 +321,15 @@ export async function setAdminAffiliateProductPrice( id: number canonical_sku: string canonical_name: string - }>('SELECT id, canonical_sku, canonical_name FROM product_catalog WHERE id = $1', [productId]) + source_kind: string + }>('SELECT id, canonical_sku, canonical_name, source_kind FROM product_catalog WHERE id = $1', [ + productId, + ]) const product = productResult.rows[0] if (!product) { throw createHttpError('商品不存在', { statusCode: 404 }) } - if (!product.canonical_sku.startsWith('aff:')) { + if (product.source_kind !== 'affiliate_dash') { throw createHttpError('该商品不是自动发货商品,无法设置自动发货采购成本', { statusCode: 400 }) } @@ -501,16 +506,17 @@ export async function importAdminAffiliateDashPricesBaseline(session: AdminSessi const productInsert = await query<{ id: number; inserted: boolean }>( ` INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, + canonical_sku, canonical_name, category, unit_name, source_kind, status, attributes_json, created_at, updated_at ) VALUES ( - $1, $2, $3, $4, 'active', + $1, $2, $3, $4, 'affiliate_dash', 'active', jsonb_build_object('source', 'affiliate_dash_price_import', 'skuCode', $5::text), $6, $6 ) ON CONFLICT (canonical_sku) DO UPDATE SET canonical_name = CASE WHEN product_catalog.canonical_name = '' THEN EXCLUDED.canonical_name ELSE product_catalog.canonical_name END, category = CASE WHEN product_catalog.category = '' THEN EXCLUDED.category ELSE product_catalog.category END, + source_kind = EXCLUDED.source_kind, updated_at = $6 RETURNING id, (xmax = 0) AS inserted `, diff --git a/apps/backend/src/services/cost-center/cost-center-constants.ts b/apps/backend/src/services/cost-center/cost-center-constants.ts index 192529fb..6e18acdb 100644 --- a/apps/backend/src/services/cost-center/cost-center-constants.ts +++ b/apps/backend/src/services/cost-center/cost-center-constants.ts @@ -103,8 +103,8 @@ export const PROVIDERS = ['91kaquan', 'pdd', 'douyin_mashida'] export const WORK_ORDER_TEMPLATE_COST_BASELINE = '2026-07-10T00:00:00+08:00' /** - * 商品来源前缀。商品表不再存履约载体,来源只由 canonical_sku 前缀表达。 - * 前缀之外的 SKU 归入「未分类」,不允许反推履约方式。 + * 商品创建时使用的编码前缀。运行期来源统一读取 product_catalog.source_kind, + * 不能再用 canonical_sku 前缀反推商品来源或履约方式。 */ export const PRODUCT_SOURCE_PREFIXES: Array<{ prefix: string diff --git a/apps/backend/src/services/cost-center/cost-center-mappers.ts b/apps/backend/src/services/cost-center/cost-center-mappers.ts index d376a5ff..b64c39d6 100644 --- a/apps/backend/src/services/cost-center/cost-center-mappers.ts +++ b/apps/backend/src/services/cost-center/cost-center-mappers.ts @@ -3,7 +3,7 @@ * * 把数据库行映射成接口 DTO,并提供「成本模型是否适用于履约方式」这类口径函数。 */ -import { GENERIC_COST_MODELS, PRODUCT_SOURCE_PREFIXES } from './cost-center-constants.js' +import { GENERIC_COST_MODELS } from './cost-center-constants.js' import type { FulfillmentChannel } from './cost-center-constants.js' import { parseJsonObject } from './cost-center-normalizers.js' import type { @@ -33,26 +33,9 @@ export const PRODUCT_CURRENT_CHANNEL_FILTERS = [ export type ProductCurrentChannelFilter = (typeof PRODUCT_CURRENT_CHANNEL_FILTERS)[number] -/** - * 由 canonical_sku 前缀推导商品来源。 - * 返回空串表示未分类;未分类商品不允许反推履约方式。 - */ -export function productSourceOf(canonicalSku: string): string { - const sku = String(canonicalSku || '').trim() - for (const item of PRODUCT_SOURCE_PREFIXES) { - if (sku.startsWith(item.prefix)) return item.source - } - return '' -} - -/** 商品来源筛选的 SQL 片段:按 canonical_sku 前缀过滤。 */ -export function productSourceFilterSql(paramIndex: number, column = 'p.canonical_sku'): string { - return `( - ($${paramIndex} = 'affiliate_dash' AND ${column} LIKE 'aff:%') - OR ($${paramIndex} = 'work_order' AND (${column} LIKE 'tpl:%' OR ${column} LIKE 'wo:%')) - OR ($${paramIndex} = 'third_party_voucher' AND ${column} LIKE 'tpv:%') - OR ($${paramIndex} = 'manual' AND ${column} LIKE 'man:%') - )` +/** 商品来源筛选的唯一 SQL 口径:读取 source_kind,不解析 canonical_sku。 */ +export function productSourceFilterSql(paramIndex: number, column = 'p.source_kind'): string { + return `${column} = $${paramIndex}` } /** 当前渠道筛选:标准商品页签的唯一筛选口径。 */ @@ -121,8 +104,8 @@ export function mapProduct(row: ProductRow) { canonicalSku: row.canonical_sku, canonicalName: row.canonical_name, category: row.category || '', - source: String(row.source_kind || productSourceOf(row.canonical_sku)), - sourceKind: String(row.source_kind || productSourceOf(row.canonical_sku)), + source: String(row.source_kind || 'unclassified'), + sourceKind: String(row.source_kind || 'unclassified'), rootCategoryId: row.root_category_id ? Number(row.root_category_id) : null, rootCategoryName: row.root_category_name || row.category || '', subCategoryId: row.sub_category_id ? Number(row.sub_category_id) : null, diff --git a/apps/backend/src/services/cost-center/cost-center-shared.ts b/apps/backend/src/services/cost-center/cost-center-shared.ts index 5d341ab9..03401da1 100644 --- a/apps/backend/src/services/cost-center/cost-center-shared.ts +++ b/apps/backend/src/services/cost-center/cost-center-shared.ts @@ -4,7 +4,7 @@ * 口径(2026-10-23 重写): * 1. 履约渠道只有 fulfillments.channel/status 一个真源; * 2. 成本项用 cost_model 表达「成本怎么算」,与履约方式正交; - * 3. 商品来源由 canonical_sku 前缀推导,商品表不再存履约载体; + * 3. 商品来源读取 product_catalog.source_kind,商品表不再存履约载体; * 4. 快照与异常表都不再存履约载体,join order_items 即可拿到。 * * 2026-10-25 拆分:804 行的共享层按职责拆成四个模块,本文件只做统一出口, diff --git a/apps/backend/src/services/cost-center/cost-center.test.ts b/apps/backend/src/services/cost-center/cost-center.test.ts index 0538b07b..012b10e8 100644 --- a/apps/backend/src/services/cost-center/cost-center.test.ts +++ b/apps/backend/src/services/cost-center/cost-center.test.ts @@ -2,7 +2,7 @@ import assert from 'node:assert/strict' import { readFile } from 'node:fs/promises' import test from 'node:test' -import { costModelAppliesTo } from './cost-center-mappers.js' +import { costModelAppliesTo, mapProduct } from './cost-center-mappers.js' import { normalizeMappingMatchType, normalizeProviderKey, @@ -288,8 +288,9 @@ test('接单模板改价必须在同一事务中同步成本版本与模板投 test('成本模型与商品来源必须覆盖第三方卡券渠道', () => { assert.equal(costModelAppliesTo('third_party_voucher_cost', 'third_party_voucher'), true) assert.equal(costModelAppliesTo('third_party_voucher_cost', 'manual'), false) - assert.ok(productSourceFilterSql(1).includes("'third_party_voucher'")) - assert.ok(productSourceFilterSql(1).includes("'tpv:%'")) + const sourceSql = productSourceFilterSql(1) + assert.equal(sourceSql, 'p.source_kind = $1') + assert.ok(!sourceSql.includes('canonical_sku')) const channelSql = productChannelFilterSql(1) assert.ok(channelSql.includes('active_primary_channel')) assert.ok(channelSql.includes("$1 = 'unrouted'")) @@ -351,11 +352,35 @@ test('运行期停止写入已废弃的 product_channel_capabilities', async () } }) -test('tpv: 商品来源归为第三方卡券', async () => { - const { productSourceOf } = await import('./cost-center-mappers.js') - const assert = (await import('node:assert/strict')).default - assert.equal(productSourceOf('tpv:模拟卡券'), 'third_party_voucher') - assert.equal(productSourceOf('man:模拟人工'), 'manual') +test('商品来源只读 source_kind,不再按 canonical_sku 前缀回退', () => { + const product = mapProduct({ + id: 1, + canonical_sku: 'man:编码与来源故意不一致', + canonical_name: '测试商品', + category: '', + unit_name: '件', + status: 'active', + root_category_id: null, + root_category_name: '', + sub_category_id: null, + sub_category_name: '', + attributes_json: {}, + created_at: '', + updated_at: '', + source_kind: 'third_party_voucher', + }) + assert.equal(product.source, 'third_party_voucher') + assert.equal(product.sourceKind, 'third_party_voucher') +}) + +test('阶段 6 不再保留会写旧履约列的运维脚本', async () => { + const packageJson = JSON.parse( + await readFile(new URL('../../../package.json', import.meta.url), 'utf8'), + ) as { scripts?: Record } + assert.equal(packageJson.scripts?.['finance:sync'], undefined) + assert.equal(packageJson.scripts?.['finance:rebuild-cutover'], undefined) + assert.equal(packageJson.scripts?.['products:merge-duplicates'], undefined) + assert.equal(packageJson.scripts?.['products:reclassify'], undefined) }) test('人工标准商品禁止绕过人工 SKU 配置手工创建', async () => { diff --git a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts index 6891b2c3..dc4bdf72 100644 --- a/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts +++ b/apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts @@ -247,7 +247,7 @@ export async function listManualSkus(input: Record = {}) { const platformKey = String(input.platformKey || input.platform_key || '').trim() const provider = String(input.provider || '').trim() - const conditions: string[] = [`p.status = 'active'`, `p.canonical_sku LIKE 'man:%'`] + const conditions: string[] = [`p.status = 'active'`] const params: unknown[] = ['manual'] conditions.push(productSourceFilterSql(params.length)) @@ -277,8 +277,14 @@ export async function listManualSkus(input: Record = {}) { JOIN orders o ON o.id = oi.order_id WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual' AND o.provider = $${params.length} - AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) - = canonicalize_internal_sku(substring(p.canonical_sku from 5)) + AND p.internal_sku_id = ( + SELECT sku.id + FROM internal_skus sku + WHERE sku.internal_sku_key = resolve_internal_sku( + oi.manual_sku_key, oi.sku_name, oi.sku_code + ) + LIMIT 1 + ) )`) } @@ -287,11 +293,11 @@ export async function listManualSkus(input: Record = {}) { ` SELECT COUNT(*)::int AS total FROM product_catalog p + JOIN internal_skus product_sku ON product_sku.id = p.internal_sku_id LEFT JOIN LATERAL ( SELECT * FROM manual_fulfillment_configs cfg - WHERE canonicalize_internal_sku(cfg.sku_key) - = canonicalize_internal_sku(substring(p.canonical_sku from 5)) + WHERE canonicalize_internal_sku(cfg.sku_key) = product_sku.internal_sku_key ORDER BY cfg.id LIMIT 1 ) cfg ON TRUE @@ -322,7 +328,7 @@ export async function listManualSkus(input: Record = {}) { ) SELECT p.id, - substring(p.canonical_sku from 5) AS sku_key, + product_sku.internal_sku_key AS sku_key, p.canonical_name AS display_name, COALESCE(manual_stats.shop_count, 0)::int AS shop_count, COALESCE(manual_stats.order_count, 0)::int AS order_count, @@ -340,16 +346,16 @@ export async function listManualSkus(input: Record = {}) { cfg.enabled AS config_enabled, cfg.channels_json AS config_channels_json FROM product_catalog p + JOIN internal_skus product_sku ON product_sku.id = p.internal_sku_id LEFT JOIN LATERAL ( SELECT * FROM manual_fulfillment_configs cfg - WHERE canonicalize_internal_sku(cfg.sku_key) - = canonicalize_internal_sku(substring(p.canonical_sku from 5)) + WHERE canonicalize_internal_sku(cfg.sku_key) = product_sku.internal_sku_key ORDER BY cfg.id LIMIT 1 ) cfg ON TRUE LEFT JOIN manual_stats - ON manual_stats.sku_key = canonicalize_internal_sku(substring(p.canonical_sku from 5)) + ON manual_stats.sku_key = product_sku.internal_sku_key WHERE ${where} ORDER BY COALESCE(manual_stats.item_count, 0) DESC, p.canonical_name ASC, p.id ASC LIMIT $${listParams.length - 1} OFFSET $${listParams.length} @@ -388,18 +394,22 @@ export async function getManualSkuOverview() { GROUP BY 1 ), manual_scope AS ( - SELECT p.id, p.canonical_sku, p.canonical_name, cfg.id AS config_id + SELECT + p.id, + p.canonical_sku, + p.canonical_name, + product_sku.internal_sku_key, + cfg.id AS config_id FROM product_catalog p + JOIN internal_skus product_sku ON product_sku.id = p.internal_sku_id LEFT JOIN LATERAL ( SELECT cfg.id FROM manual_fulfillment_configs cfg - WHERE canonicalize_internal_sku(cfg.sku_key) - = canonicalize_internal_sku(substring(p.canonical_sku from 5)) + WHERE canonicalize_internal_sku(cfg.sku_key) = product_sku.internal_sku_key ORDER BY cfg.id LIMIT 1 ) cfg ON TRUE WHERE p.status = 'active' - AND p.canonical_sku LIKE 'man:%' AND ${productSourceFilterSql(1)} ) SELECT @@ -414,9 +424,7 @@ export async function getManualSkuOverview() { MAX(manual_stats.last_seen_date)::text AS last_scan_at FROM manual_scope LEFT JOIN manual_stats - ON manual_stats.sku_key = canonicalize_internal_sku( - substring(manual_scope.canonical_sku from 5) - ) + ON manual_stats.sku_key = manual_scope.internal_sku_key `, ['manual'], ) @@ -469,8 +477,12 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) { let resolvedProductId = canonicalProductId if (resolvedProductId) { - const product = await client.query<{ canonical_sku: string; status: string }>( - `SELECT canonical_sku, status + const product = await client.query<{ + canonical_sku: string + source_kind: string + status: string + }>( + `SELECT canonical_sku, source_kind, status FROM product_catalog WHERE id = $1 LIMIT 1`, @@ -479,9 +491,9 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) { if ( !product.rows[0] || product.rows[0].status !== 'active' || - !String(product.rows[0].canonical_sku || '').startsWith('man:') + product.rows[0].source_kind !== 'manual' ) { - throw createHttpError('人工履约配置只能关联有效的 man: 标准商品', { + throw createHttpError('人工履约配置只能关联来源为人工的有效标准商品', { statusCode: 400, errorCode: 'manual_config_product_source_invalid', }) @@ -510,10 +522,10 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) { .slice(0, 120) || skuKey.slice(0, 120) const product = await client.query<{ id: number }>( `INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, + canonical_sku, canonical_name, category, unit_name, source_kind, status, attributes_json, created_at, updated_at ) VALUES ( - $1, $2, '人工', '件', 'active', + $1, $2, '人工', '件', 'manual', 'active', '{"source":"manual_config","auto_created":true}'::jsonb, $3, $3 ) ON CONFLICT (canonical_sku) DO UPDATE SET @@ -789,8 +801,8 @@ export async function deleteManualSkuClassification( ), updated_at = $2::timestamptz WHERE status = 'active' - AND canonical_sku LIKE 'man:%' - AND canonicalize_internal_sku(substring(canonical_sku from 5)) = $1 + AND source_kind = 'manual' + AND internal_sku_id = (SELECT id FROM internal_skus WHERE internal_sku_key = $1 LIMIT 1) RETURNING id `, [key, nowIso()], diff --git a/apps/backend/src/services/cost-center/product-catalog-service.ts b/apps/backend/src/services/cost-center/product-catalog-service.ts index fda4738a..7336850a 100644 --- a/apps/backend/src/services/cost-center/product-catalog-service.ts +++ b/apps/backend/src/services/cost-center/product-catalog-service.ts @@ -67,13 +67,10 @@ import { /** 标准商品页签的当前渠道筛选取值。 */ const PRODUCT_CHANNEL_FILTERS = new Set(PRODUCT_CURRENT_CHANNEL_FILTERS) -/** - * 校验新建商品的 SKU 前缀。 - * 前缀是商品来源的唯一表达,不允许出现无法归类的商品。 - */ -function assertKnownSourcePrefix(canonicalSku: string) { +/** 创建商品时把编码前缀固化为 source_kind,运行期不再反推来源。 */ +function resolveProductSourceKind(canonicalSku: string) { for (const item of PRODUCT_SOURCE_PREFIXES) { - if (canonicalSku.startsWith(item.prefix)) return + if (canonicalSku.startsWith(item.prefix)) return item.source } throw createHttpError( `标准 SKU 必须以下列前缀开头:${PRODUCT_SOURCE_PREFIXES.map((item) => item.prefix).join(' / ')}`, @@ -225,9 +222,9 @@ export async function listAdminProducts(payload: JsonObject = {}) { ORDER BY p.status ASC, CASE - WHEN p.canonical_sku LIKE 'aff:%' THEN 1 - WHEN p.canonical_sku LIKE 'tpl:%' OR p.canonical_sku LIKE 'wo:%' THEN 2 - WHEN p.canonical_sku LIKE 'man:%' THEN 3 + WHEN p.source_kind = 'affiliate_dash' THEN 1 + WHEN p.source_kind = 'work_order' THEN 2 + WHEN p.source_kind = 'manual' THEN 3 ELSE 4 END, p.canonical_name ASC, @@ -323,7 +320,7 @@ export async function createAdminProduct(payload: JsonObject, _session: AdminSes errorCode: 'cost_product_sku_invalid', }) } - assertKnownSourcePrefix(canonicalSku) + const sourceKind = resolveProductSourceKind(canonicalSku) const canonicalName = normalizeRequiredText( payload.canonicalName ?? payload.canonical_name, '商品名称', @@ -338,13 +335,22 @@ export async function createAdminProduct(payload: JsonObject, _session: AdminSes const result = await query( ` INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, + canonical_sku, canonical_name, category, unit_name, source_kind, status, attributes_json, created_at, updated_at - ) VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $7) + ) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $8) ON CONFLICT (canonical_sku) DO NOTHING RETURNING * `, - [canonicalSku, canonicalName, category, unitName, status, JSON.stringify(attributes), now], + [ + canonicalSku, + canonicalName, + category, + unitName, + sourceKind, + status, + JSON.stringify(attributes), + now, + ], ) const inserted = result.rows[0] if (!inserted) { @@ -449,7 +455,7 @@ export async function deleteAdminVoucherProduct(productId: number | string) { [id], ) const current = requiredRow(currentResult.rows[0], '商品不存在') - if (!current.canonical_sku.startsWith('tpv:')) { + if (current.source_kind !== 'third_party_voucher') { throw createHttpError('该商品不是第三方卡券商品', { statusCode: 400, errorCode: 'voucher_product_type_invalid', @@ -1498,9 +1504,10 @@ export async function setAdminProductPrice( id: number canonical_sku: string canonical_name: string + source_kind: string attributes_json: unknown }>( - 'SELECT id, canonical_sku, canonical_name, attributes_json FROM product_catalog WHERE id = $1', + 'SELECT id, canonical_sku, canonical_name, source_kind, attributes_json FROM product_catalog WHERE id = $1', [productId], ) const product = productResult.rows[0] @@ -1509,7 +1516,7 @@ export async function setAdminProductPrice( } // 1. 如果是自动发货商品,委托给 setAdminAffiliateProductPrice - if (product.canonical_sku.startsWith('aff:')) { + if (product.source_kind === 'affiliate_dash') { return setAdminAffiliateProductPrice(productId, payload, session) } @@ -1523,7 +1530,7 @@ export async function setAdminProductPrice( const changeReason = String(payload.changeReason || '在商品与成本中心统一调整价格').trim() // 2. 如果是接单模板商品 (tpl:) - if (product.canonical_sku.startsWith('tpl:')) { + if (product.source_kind === 'work_order' && product.canonical_sku.startsWith('tpl:')) { const ruleKey = product.canonical_sku.slice(4) const ruleResult = await query<{ id: number @@ -1729,12 +1736,13 @@ export async function setAdminProductPrice( } } - // 3. 其他标准商品 (wo: / tpv: / man:) - const costItemCode = product.canonical_sku.startsWith('wo:') - ? 'sys_work_order_reward' - : product.canonical_sku.startsWith('tpv:') - ? 'sys_third_party_voucher_cost' - : 'sys_manual_service' + // 3. 其他标准商品:成本项由 source_kind 决定,canonical_sku 只保留载体键。 + const costItemCode = + product.source_kind === 'work_order' + ? 'sys_work_order_reward' + : product.source_kind === 'third_party_voucher' + ? 'sys_third_party_voucher_cost' + : 'sys_manual_service' const costItemResult = await query<{ id: number }>( 'SELECT id FROM cost_items WHERE code = $1 LIMIT 1', [costItemCode], @@ -1911,13 +1919,16 @@ export async function syncAdminProductPrice(productIdInput: unknown, session: Ad id: number canonical_sku: string canonical_name: string - }>('SELECT id, canonical_sku, canonical_name FROM product_catalog WHERE id = $1', [productId]) + source_kind: string + }>('SELECT id, canonical_sku, canonical_name, source_kind FROM product_catalog WHERE id = $1', [ + productId, + ]) const product = productResult.rows[0] if (!product) { throw createHttpError('商品不存在', { statusCode: 404 }) } - if (product.canonical_sku.startsWith('tpl:')) { + if (product.source_kind === 'work_order' && product.canonical_sku.startsWith('tpl:')) { const ruleKey = product.canonical_sku.slice(4) const ruleResult = await query<{ id: number }>( 'SELECT id FROM work_product_rules WHERE rule_key = $1 LIMIT 1', @@ -1937,7 +1948,7 @@ export async function syncAdminProductPrice(productIdInput: unknown, session: Ad } } - if (product.canonical_sku.startsWith('aff:')) { + if (product.source_kind === 'affiliate_dash') { // 成本快照由成本快照管道按缺补算,这里不再同步触发。 return { success: true, diff --git a/apps/backend/src/services/cost-center/product-normalize-service.test.ts b/apps/backend/src/services/cost-center/product-normalize-service.test.ts index 4ac1cd97..4bd99f2c 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.test.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.test.ts @@ -68,7 +68,7 @@ function item(patch: Record = {}) { canonical_product_id: null as number | null, work_rule_key: null as string | null, manual_config_product_id: null as number | null, - manual_config_product_sku: null as string | null, + manual_config_product_source_kind: null as string | null, ...patch, } } @@ -126,7 +126,7 @@ test('商品归一:人工渠道直接使用配置里指定的商品', () => { fulfillment_channel: 'manual', sku_code: 'unknown', manual_config_product_id: 44, - manual_config_product_sku: 'man:配置商品', + manual_config_product_source_kind: 'manual', }), index, ), @@ -177,13 +177,13 @@ test('商品归一:其他已确定渠道也不能借用同名其他来源商 } }) -test('商品归一:人工配置也只能挂载 man: 商品', () => { +test('商品归一:人工配置只能挂载来源为人工的标准商品', () => { assert.equal( resolveProductId( item({ fulfillment_channel: 'manual', manual_config_product_id: 99, - manual_config_product_sku: 'aff:错误商品', + manual_config_product_source_kind: 'affiliate_dash', }), index, ), @@ -194,7 +194,7 @@ test('商品归一:人工配置也只能挂载 man: 商品', () => { item({ fulfillment_channel: 'on_account', manual_config_product_id: 99, - manual_config_product_sku: 'man:正确商品', + manual_config_product_source_kind: 'manual', }), index, ), @@ -240,10 +240,11 @@ test('三方卡券自动建商品时优先使用内部中文 SKU', () => { test('商品归一扫描会纠正所有渠道的错误来源商品', async () => { const source = await readFile(new URL('./product-normalize-service.ts', import.meta.url), 'utf8') - assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'aff:%'")) - assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'man:%'")) - assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'tpl:%'")) - assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'tpv:%'")) + assert.ok(source.includes("current_product.source_kind IS DISTINCT FROM 'affiliate_dash'")) + assert.ok(source.includes("current_product.source_kind IS DISTINCT FROM 'manual'")) + assert.ok(source.includes("current_product.source_kind IS DISTINCT FROM 'work_order'")) + assert.ok(source.includes("current_product.source_kind IS DISTINCT FROM 'third_party_voucher'")) + assert.ok(!source.includes("current_product.canonical_sku NOT LIKE 'man:%'")) assert.ok(source.includes('canonicalProductId: null')) assert.ok(source.includes('onlyWhenCanonicalProductDifferent: true')) }) diff --git a/apps/backend/src/services/cost-center/product-normalize-service.ts b/apps/backend/src/services/cost-center/product-normalize-service.ts index 6cbbeffd..e99ac9fd 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -57,7 +57,15 @@ type ItemRow = { canonical_product_id?: number | null work_rule_key: string | null manual_config_product_id: number | null - manual_config_product_sku?: string | null + manual_config_product_source_kind?: string | null +} + +const EXPECTED_PRODUCT_SOURCE: Record = { + affiliate_dash: 'affiliate_dash', + work_order: 'work_order', + third_party_voucher: 'third_party_voucher', + manual: 'manual', + on_account: 'manual', } function key(...parts: string[]): string { @@ -179,10 +187,13 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise( ` INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, + canonical_sku, canonical_name, category, unit_name, source_kind, status, attributes_json, created_at, updated_at ) - VALUES ($1, $2, '卡券', '张', 'active', '{"source":"third_party_voucher","auto_created":true}'::jsonb, NOW(), NOW()) + VALUES ( + $1, $2, '卡券', '张', 'third_party_voucher', 'active', + '{"source":"third_party_voucher","auto_created":true}'::jsonb, NOW(), NOW() + ) ON CONFLICT (canonical_sku) DO NOTHING RETURNING id `, @@ -235,7 +246,7 @@ export function resolveProductId(item: ItemRow, index: MappingIndex): number | n // 已确定渠道必须命中本渠道商品,不能借用同名的其他来源商品。 if (item.fulfillment_channel === 'manual' || item.fulfillment_channel === 'on_account') { return item.manual_config_product_id && - String(item.manual_config_product_sku || '').startsWith('man:') + item.manual_config_product_source_kind === EXPECTED_PRODUCT_SOURCE.manual ? Number(item.manual_config_product_id) : null } @@ -338,7 +349,7 @@ export async function normalizeOrderItemProducts( oi.canonical_product_id, wpr.rule_key AS work_rule_key, cfg.canonical_product_id AS manual_config_product_id, - cfg_product.canonical_sku AS manual_config_product_sku + cfg_product.source_kind AS manual_config_product_source_kind FROM v_order_item_current oi JOIN orders o ON o.id = oi.order_id LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled' @@ -349,36 +360,22 @@ export async function normalizeOrderItemProducts( WHERE oi.fulfillment_status = 'assigned' AND ( oi.canonical_product_id IS NULL + OR current_product.id IS NULL OR ( oi.fulfillment_channel = 'third_party_voucher' - AND ( - current_product.id IS NULL - OR current_product.canonical_sku NOT LIKE 'tpv:%' - ) + AND current_product.source_kind IS DISTINCT FROM 'third_party_voucher' ) OR ( oi.fulfillment_channel = 'affiliate_dash' - AND ( - current_product.id IS NULL - OR current_product.canonical_sku NOT LIKE 'aff:%' - ) + AND current_product.source_kind IS DISTINCT FROM 'affiliate_dash' ) OR ( oi.fulfillment_channel IN ('manual', 'on_account') - AND ( - current_product.id IS NULL - OR current_product.canonical_sku NOT LIKE 'man:%' - ) + AND current_product.source_kind IS DISTINCT FROM 'manual' ) OR ( oi.fulfillment_channel = 'work_order' - AND ( - current_product.id IS NULL - OR ( - current_product.canonical_sku NOT LIKE 'tpl:%' - AND current_product.canonical_sku NOT LIKE 'wo:%' - ) - ) + AND current_product.source_kind IS DISTINCT FROM 'work_order' ) ) AND oi.id > $1 @@ -450,7 +447,7 @@ async function loadMappingIndex(): Promise { ` SELECT id, canonical_sku FROM product_catalog - WHERE status = 'active' AND canonical_sku LIKE 'aff:%' + WHERE status = 'active' AND source_kind = 'affiliate_dash' `, ) @@ -458,7 +455,9 @@ async function loadMappingIndex(): Promise { ` SELECT id, canonical_sku FROM product_catalog - WHERE status = 'active' AND canonical_sku LIKE 'tpl:%' + WHERE status = 'active' + AND source_kind = 'work_order' + AND canonical_sku LIKE 'tpl:%' `, ) @@ -466,7 +465,7 @@ async function loadMappingIndex(): Promise { ` SELECT id, canonical_sku FROM product_catalog - WHERE status = 'active' AND canonical_sku LIKE 'tpv:%' + WHERE status = 'active' AND source_kind = 'third_party_voucher' `, ) @@ -528,7 +527,7 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{ oi.canonical_product_id, wpr.rule_key AS work_rule_key, cfg.canonical_product_id AS manual_config_product_id, - cfg_product.canonical_sku AS manual_config_product_sku + cfg_product.source_kind AS manual_config_product_source_kind FROM v_order_item_current oi JOIN orders o ON o.id = oi.order_id LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled' @@ -539,36 +538,22 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{ WHERE oi.fulfillment_status = 'assigned' AND ( oi.canonical_product_id IS NULL + OR current_product.id IS NULL OR ( oi.fulfillment_channel = 'third_party_voucher' - AND ( - current_product.id IS NULL - OR current_product.canonical_sku NOT LIKE 'tpv:%' - ) + AND current_product.source_kind IS DISTINCT FROM 'third_party_voucher' ) OR ( oi.fulfillment_channel = 'affiliate_dash' - AND ( - current_product.id IS NULL - OR current_product.canonical_sku NOT LIKE 'aff:%' - ) + AND current_product.source_kind IS DISTINCT FROM 'affiliate_dash' ) OR ( oi.fulfillment_channel IN ('manual', 'on_account') - AND ( - current_product.id IS NULL - OR current_product.canonical_sku NOT LIKE 'man:%' - ) + AND current_product.source_kind IS DISTINCT FROM 'manual' ) OR ( oi.fulfillment_channel = 'work_order' - AND ( - current_product.id IS NULL - OR ( - current_product.canonical_sku NOT LIKE 'tpl:%' - AND current_product.canonical_sku NOT LIKE 'wo:%' - ) - ) + AND current_product.source_kind IS DISTINCT FROM 'work_order' ) ) AND oi.id = ANY($1::bigint[]) diff --git a/apps/backend/src/services/cost-center/template-bootstrap-service.ts b/apps/backend/src/services/cost-center/template-bootstrap-service.ts index 4c549259..03137925 100644 --- a/apps/backend/src/services/cost-center/template-bootstrap-service.ts +++ b/apps/backend/src/services/cost-center/template-bootstrap-service.ts @@ -168,9 +168,9 @@ export async function applyAdminTemplateProductBootstrap( const inserted = await client.query<{ id: number }>( ` INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, + canonical_sku, canonical_name, category, unit_name, source_kind, status, attributes_json, created_at, updated_at - ) VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $7) + ) VALUES ($1, $2, $3, $4, 'work_order', $5, $6::jsonb, $7, $7) ON CONFLICT (canonical_sku) DO NOTHING RETURNING id `, diff --git a/apps/backend/src/services/cost-center/voucher-product-service.ts b/apps/backend/src/services/cost-center/voucher-product-service.ts index c2f397bb..01e06074 100644 --- a/apps/backend/src/services/cost-center/voucher-product-service.ts +++ b/apps/backend/src/services/cost-center/voucher-product-service.ts @@ -60,14 +60,15 @@ export async function upsertAdminVoucherProduct(payload: JsonObject, session: Ad const productResult = await client.query( ` INSERT INTO product_catalog ( - canonical_sku, canonical_name, category, unit_name, status, + canonical_sku, canonical_name, category, unit_name, source_kind, status, attributes_json, created_at, updated_at ) VALUES ( - $1, $2, '卡券', '张', 'active', + $1, $2, '卡券', '张', 'third_party_voucher', 'active', $3::jsonb, $4, $4 ) ON CONFLICT (canonical_sku) DO UPDATE SET status = 'active', + source_kind = EXCLUDED.source_kind, canonical_name = EXCLUDED.canonical_name, attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json, updated_at = EXCLUDED.updated_at diff --git a/apps/backend/src/services/cost-center/work-order-product-service.ts b/apps/backend/src/services/cost-center/work-order-product-service.ts index 7440acb2..4fe319f7 100644 --- a/apps/backend/src/services/cost-center/work-order-product-service.ts +++ b/apps/backend/src/services/cost-center/work-order-product-service.ts @@ -141,7 +141,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record 64) return null @@ -2129,16 +2139,18 @@ async function ensureDraftProductForChannel( canonical_name, category, unit_name, + source_kind, status, attributes_json, created_at, updated_at ) VALUES ( - $1, $2, $3, $4, 'active', - $5::jsonb, $6, $6 + $1, $2, $3, $4, $5, 'active', + $6::jsonb, $7, $7 ) ON CONFLICT (canonical_sku) DO UPDATE SET status = 'active', + source_kind = EXCLUDED.source_kind, canonical_name = CASE WHEN BTRIM(product_catalog.canonical_name) = '' THEN EXCLUDED.canonical_name ELSE product_catalog.canonical_name @@ -2157,6 +2169,7 @@ async function ensureDraftProductForChannel( key.slice(0, 120), definition.category, definition.unit, + definition.sourceKind, JSON.stringify({ source: channel, auto_created: true, diff --git a/deploy/post-deploy.sh b/deploy/post-deploy.sh index b9441f25..f3c9771e 100755 --- a/deploy/post-deploy.sh +++ b/deploy/post-deploy.sh @@ -8,7 +8,7 @@ set -Eeuo pipefail # 它把以前散落在文档里的脚本步骤串成一条可重复执行的流水线: # ① 存量订单项收入回填 revenue:backfill # ② 渠道/商品由管道处理(无需回填) -# ③ 财务历史重建(分窗口) finance:rebuild-cutover +# ③ 财务历史重建(分窗口) finance:rebuild-derived # ④ 上线验收(只读 SQL) scripts/verify-finance-launch.sql # # 设计原则: @@ -388,17 +388,18 @@ step_sync() { if [[ "$products" == "0" ]]; then warn "商品目录为空:财务重建会生成大量 product_mapping_missing 异常;建议先完成商品初始化再跑 --only=sync" fi - log "③ 财务历史重建(finance:rebuild-cutover),区间 ${from} → ${to},每窗 ${WINDOW_DAYS} 天" + log "③ 财务派生数据重建(finance:rebuild-derived),区间 ${from} → ${to},每窗 ${WINDOW_DAYS} 天" while :; do win_to="$(date_add_days "$from" $((WINDOW_DAYS - 1)))" if [[ "$win_to" > "$to" ]]; then win_to="$to" fi if [[ "$DRY_RUN" == true ]]; then - log " [dry-run] npm run finance:rebuild-cutover -- --date-from=${from} --date-to=${win_to}" + log " [dry-run] npm run finance:rebuild-derived -- --date-from=${from} --date-to=${win_to} --execute --confirm=REBUILD_DERIVED" else log " 窗口 ${from} → ${win_to}" - run_tool finance:rebuild-cutover --date-from="$from" --date-to="$win_to" + run_tool finance:rebuild-derived --date-from="$from" --date-to="$win_to" \ + --execute --confirm=REBUILD_DERIVED fi [[ "$win_to" == "$to" ]] && break from="$(date_add_days "$win_to" 1)" diff --git a/docs/统一订单履约与财务模型-权威版.md b/docs/统一订单履约与财务模型-权威版.md index 9ec22b7e..8976d4fb 100644 --- a/docs/统一订单履约与财务模型-权威版.md +++ b/docs/统一订单履约与财务模型-权威版.md @@ -3,7 +3,7 @@ type: 权威架构与实施计划 description: 订单、内部 SKU、履约路由、履约执行、商品、成本与财务的唯一权威口径及重构计划。 tags: [订单, 内部SKU, 履约, 商品, 成本, 财务, 迁移] status: authoritative -version: 2.1 +version: 2.2 updated_at: 2026-11-08 --- @@ -12,7 +12,7 @@ updated_at: 2026-11-08 > 状态:**唯一权威文档**。 > 最后校准:2026-11-08。 > 生产基线:迁移 `146`。 -> 开发基线:迁移 `167`(仅本地,尚未部署生产)。 +> 开发基线:迁移 `168`(仅本地,尚未部署生产)。 > > 从本文件生效起,订单、内部 SKU、履约渠道、履约载体、标准商品、成本和财务相关实现, > 只能以本文件为业务口径。其他旧分析只作为历史资料,不再具备设计或验收效力。 @@ -332,7 +332,7 @@ finance_journal_* -- 总账凭证,可重建 2026-11-08 生产只读核对结果: -- 生产迁移 `146`,本地开发基线迁移 `167`。 +- 生产迁移 `146`,本地开发基线迁移 `168`。 - 启用标准商品 `1127`:`aff:` 30、`tpl:` 903、`tpv:` 46、`man:` 148。 - `man:` 商品中只有 105 个当前主渠道为人工;41 个已走其他主渠道,2 个无主渠道。 - 人工页头部显示 `107`,商品列表显示 `148`,相差的 41 个就是跨渠道旧商品。 @@ -612,6 +612,8 @@ assigned 且 carrier_status=missing 必须进入异常队列 并进入 `missing_internal_sku` 异常,而不是触发非空约束。 - 迁移 `167`:建立 `v_finance_order_item_scope`,统一销售事实、重建与审计的收入、 退款、商品、渠道和业务日口径;增加销售凭证按日查询索引。 +- 迁移 `168`:商品来源 `source_kind` 增加数据库约束;商品筛选、人工 SKU 清单和成本 + 分支统一读取 `source_kind` / `internal_sku_id`,运行期不再按 `canonical_sku` 前缀判断来源。 - 商品列表、总览、渠道映射和商品下拉统一读取商品当前渠道。 - 运行期代码停止写入 `product_channel_capabilities`,该表仅保留到兼容期结束。 - 只读审计:`npm run audit:product-routing -- --strict`。 @@ -619,7 +621,7 @@ assigned 且 carrier_status=missing 必须进入异常队列 - 自动/卡券/接单主渠道商品的人工配置均保留为显式 `fallback`,没有被误升为人工主渠道。 - 1 个无主路由、无人工配置、无启用映射的 `man:` 商品已归档,未删除历史引用。 - 旧 `internal_sku_channels` 已冻结:运行时不再读、不再写,也不再有视图或触发器依赖。 -- 空库从迁移 `001` 顺序执行到 `167` 已通过;新库不再保留任何旧履约影子同步触发器。 +- 空库从迁移 `001` 顺序执行到 `168` 已通过;新库不再保留任何旧履约影子同步触发器。 - 无主路由的活动商品会进入“未路由”,不自动猜测渠道。 当前本地仍需业务处理的数据: @@ -751,7 +753,21 @@ npm run audit:finance -- --strict ### 阶段 6:停止旧写路径并删除兼容层 -状态:履约与财务派生写入路径已在本地收敛;商品、成本、人工配置的全局兼容清理与生产观察期仍需后续完成。 +状态:本地运行期旧写路径已收敛;兼容数据表和冻结字段保留到生产观察期结束。 + +本地已完成: + +- 删除会直接写已改名订单项列的 `reclassify-post-cutover.ts` 和 + `merge-duplicate-products.ts`。 +- 删除重复的 `finance:rebuild-cutover` 入口,部署脚本统一调用 + `finance:rebuild-derived --execute --confirm=REBUILD_DERIVED`。 +- 删除失效的 `finance:sync` 脚本别名。 +- 商品来源筛选、归一纠正、成本分支、人工 SKU 清单统一读取 + `product_catalog.source_kind` 与 `internal_sku_id`。 +- 商品创建写入口显式写入 `source_kind`,数据库约束只允许 + `affiliate_dash / work_order / third_party_voucher / manual / unclassified`。 +- `product_channel_capabilities`、`manual_sku_registry`、`internal_sku_channels` 和历史 + `legacy_*` 字段继续保留为审计留存,不在生产观察期前删除。 只有前五个阶段持续稳定后才允许: @@ -845,14 +861,13 @@ docs/archive/ ## 十三、当前下一步 -本轮只完成文档和计划,不修改业务代码、不执行生产数据清理。 +本地阶段 0–6 已按硬切方案完成,生产仍停留在迁移 `146`。后续顺序固定为: -后续按以下顺序启动: - -1. 完成阶段 1 的只读审计脚本与冲突清单。 -2. 确定 `internal_skus`、`sku_channel_routes`、`product_catalog.source_kind` - 的最终字段和迁移顺序。 -3. 先建立投影,不直接切换写路径。 -4. 清理派生数据前先做可恢复备份。 -5. 每个阶段单独验收,不合并成一次性大迁移。 +1. 生成生产升级清单并做全量备份,不在本地继续扩展新模型。 +2. 在生产依次执行迁移 `147 → 168`,每阶段运行对应严格审计; + 无法确定的数据进入未路由或待分类,不做猜测。 +3. 执行一次 `finance:rebuild-derived` 的 dry-run,核对 A 级数据守恒后, + 再按窗口执行 `--execute --confirm=REBUILD_DERIVED`。 +4. 观察一个完整财务期间,确认页面、履约、事实和总账持续对平。 +5. 观察期结束后,再单独发布兼容表、旧脚本别名和 `legacy_*` 字段的删除迁移。 diff --git a/docs/部署启动与数据库重建.md b/docs/部署启动与数据库重建.md index 7964893b..d64fce9e 100644 --- a/docs/部署启动与数据库重建.md +++ b/docs/部署启动与数据库重建.md @@ -219,7 +219,7 @@ docker compose ps docker compose logs -f backend ``` -## 财务/成本上线后的初始化(迁移 103–105) +## 财务/成本上线后的初始化(当前开发基线 168) 代码与迁移只负责建表;**存量订单回填、备用金账户、积分余额、商品目录、人工 SKU 不会自动创建**。 部署成功后按「回填 → 商品初始化 → 财务同步 → 验收」四步走: @@ -234,7 +234,7 @@ bash deploy/post-deploy.sh --only=backfill # ① 收入回填 + ② 履约 > 不要在商品目录为空时直接跑 `--only=sync`:财务同步会为每个未归一订单项写一条 > `product_mapping_missing` 成本异常(生产规模 90 天实测约 36 万条),后台异常队列会被淹没。 -脚本包含 `revenue:backfill`、渠道/成本管道重建(`finance:rebuild-cutover`)与只读验收;任何一步都能重跑, +脚本包含 `revenue:backfill`、财务派生数据重建(`finance:rebuild-derived`)与只读验收;任何一步都能重跑, `--only=backfill|sync|verify` 可单独执行。 > 生产 backend 镜像没有 `tsx` 与 `scripts/`,脚本通过 `tools` 镜像执行 @@ -255,14 +255,15 @@ bash deploy/post-deploy.sh --only=backfill # ① 收入回填 + ② 履约 商品归一完成后执行: ```bash -bash deploy/post-deploy.sh --only=sync # ③ 成本快照 / 销售事实 +bash deploy/post-deploy.sh --only=sync # ③ 成本快照 / 销售事实 / 总账调整 ``` > 工具镜像每次运行都会重建(层缓存下只需几秒),确保 `git pull` 后执行的是当前脚本。 ### 4. 总账与验收 -5. 财务 → 总账与对账 → 同步销售事实到总账 → 检查试算平衡;按需关账/导入平台账单。 +5. 财务 → 总账与对账 → 检查试算平衡;`post-deploy.sh` 已按日追加销售事实调整凭证, + 之后按需关账/导入平台账单。 ```bash bash deploy/post-deploy.sh --only=verify # ④ 只读验收 diff --git a/scripts/verify-finance-launch.sql b/scripts/verify-finance-launch.sql index 1df0af8c..854ee9fa 100644 --- a/scripts/verify-finance-launch.sql +++ b/scripts/verify-finance-launch.sql @@ -249,7 +249,7 @@ INSERT INTO _launch_checks SELECT 14, '14', '销售事实已生成(sales_daily_facts)', v::text, '> 0', CASE WHEN v > 0 THEN 'PASS' ELSE 'FAIL' END, - '跑 finance:rebuild-cutover(post-deploy.sh 会按窗口执行);定时任务只回看水位窗口' + '跑 finance:rebuild-derived(post-deploy.sh 会按窗口执行);定时任务只回看水位窗口' FROM (SELECT count(*) AS v FROM sales_daily_facts) s; -- 15 近 7 天成本覆盖(仅提示,不阻断上线) ---------------------------------