From 5fe33b0b461ac9a29bc3b7fa464fffae6dc27eff Mon Sep 17 00:00:00 2001 From: yml2213 Date: Mon, 2 Nov 2026 14:17:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(finance):=20=E7=BB=9F=E4=B8=80=E5=86=85?= =?UTF-8?q?=E9=83=A8=20SKU=20=E9=A9=B1=E5=8A=A8=E7=9A=84=E5=B1=A5=E7=BA=A6?= =?UTF-8?q?=E4=B8=8E=E8=B4=A2=E5=8A=A1=E5=88=86=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 internal_sku_mappings / internal_sku_channels，平台字段统一映射到内部中文 SKU - 下单入口按内部 SKU 渠道决策修正默认人工路由，自动发货补全 dash sku - 财务 channel_assign 读取统一渠道表；卡券/人工能力兜底，自动发货/接单要求真实载体 - carrier reconciliation 自动补自动发货任务和接单工单 - pending 落明确原因：无内部 SKU / 无渠道能力 / 无履约载体 - 合并、重判、旧 sku_code 规则清理、部署后 preflight 核对脚本 - 待分类队列按内部 SKU 聚合并展示原因 --- .../scripts/cleanup-legacy-sku-code-rules.ts | 133 ++++++++++++ .../scripts/merge-duplicate-products.ts | 16 ++ .../scripts/preflight-fulfillment-routing.ts | 99 +++++++++ .../scripts/reclassify-post-cutover.ts | 17 +- .../migrations/118_internal_sku_channels.sql | 67 ++++++ .../119_fulfillment_pending_reason.sql | 14 ++ .../cost-center/product-normalize-service.ts | 10 + .../carrier-reconciliation-service.ts | 195 ++++++++++++++++++ .../channel-assign-service.test.ts | 32 ++- .../fulfillment/channel-assign-service.ts | 61 +++++- .../intake-channel-decision-service.ts | 125 +++++++++++ .../internal-sku-channel-service.ts | 83 ++++++++ .../fulfillment/product-resolution-service.ts | 9 + .../src/services/order/order-service.ts | 14 +- .../scheduler/finance-pipeline-jobs.ts | 5 + .../admin/AdminFulfillmentChannelPage.tsx | 13 ++ .../src/services/admin/fulfillment-channel.ts | 1 + docs/生产部署核对清单-2026-11-02.md | 141 +++++++++++++ 18 files changed, 1020 insertions(+), 15 deletions(-) create mode 100644 apps/backend/scripts/cleanup-legacy-sku-code-rules.ts create mode 100644 apps/backend/scripts/preflight-fulfillment-routing.ts create mode 100644 apps/backend/src/db/migrations/118_internal_sku_channels.sql create mode 100644 apps/backend/src/db/migrations/119_fulfillment_pending_reason.sql create mode 100644 apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts create mode 100644 apps/backend/src/services/fulfillment/intake-channel-decision-service.ts create mode 100644 apps/backend/src/services/fulfillment/internal-sku-channel-service.ts create mode 100644 docs/生产部署核对清单-2026-11-02.md diff --git a/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts b/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts new file mode 100644 index 00000000..918e645e --- /dev/null +++ b/apps/backend/scripts/cleanup-legacy-sku-code-rules.ts @@ -0,0 +1,133 @@ +/** + * 把旧的 sku_code 渠道规则转换为 internal_sku 全局规则，并停用旧规则。 + * + * 用法： + * npx tsx scripts/cleanup-legacy-sku-code-rules.ts + * npx tsx scripts/cleanup-legacy-sku-code-rules.ts --apply + */ +import { closeDb, query } from '../src/db/client.js' + +type LegacyRule = { + id: number + provider: string + shop_id: string + match_key: string + channel: string + priority: number + source: string +} + +async function resolveInternalSku(rule: LegacyRule): Promise { + if (rule.channel === 'affiliate_dash') { + const product = await query<{ canonical_name: string }>( + `SELECT canonical_name FROM product_catalog + WHERE canonical_sku = $1 AND status = 'active' AND btrim(canonical_name) <> '' + LIMIT 1`, + [`aff:${rule.match_key}`], + ) + if (product.rows[0]?.canonical_name) return String(product.rows[0].canonical_name).trim() + } + + const params: unknown[] = [rule.match_key] + let scope = '' + if (rule.provider) { + params.push(rule.provider) + scope += ` AND o.provider = $${params.length}` + } + if (rule.shop_id) { + params.push(rule.shop_id) + scope += ` AND o.shop_id = $${params.length}` + } + const items = await query<{ internal_sku: string }>( + ` + SELECT DISTINCT btrim(oi.manual_sku_key) AS internal_sku + FROM order_items oi + JOIN orders o ON o.id = oi.order_id + WHERE oi.sku_code = $1 + AND btrim(COALESCE(oi.manual_sku_key, '')) ~ '[一-龥]' + ${scope} + LIMIT 2 + `, + params, + ) + const values = items.rows.map((row) => String(row.internal_sku || '').trim()).filter(Boolean) + return values.length === 1 ? values[0]! : '' +} + +async function main() { + const apply = process.argv.includes('--apply') + const rules = await query( + ` + SELECT id, provider, shop_id, match_key, channel, priority, source + FROM sku_channel_rules + WHERE match_key_type = 'sku_code' AND enabled = TRUE + ORDER BY id + `, + ) + + const converted: Array<{ id: number; matchKey: string; internalSku: string; channel: string }> = + [] + const unresolved: LegacyRule[] = [] + for (const rule of rules.rows) { + const internalSku = await resolveInternalSku(rule) + if (!internalSku) { + unresolved.push(rule) + continue + } + converted.push({ id: rule.id, matchKey: rule.match_key, internalSku, channel: rule.channel }) + if (!apply) continue + + 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, $2, $3, $4, TRUE, + '由旧 sku_code 规则迁移', NOW(), NOW()) + ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING + `, + [internalSku, rule.channel, rule.priority, rule.source || 'operator'], + ) + await query( + ` + INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, source + ) VALUES ($1, $2, $3, FALSE, TRUE, 'legacy-rule') + ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), + enabled = TRUE, + updated_at = NOW() + `, + [internalSku, rule.channel, rule.priority], + ) + await query( + `UPDATE sku_channel_rules + SET enabled = FALSE, + note = CONCAT(COALESCE(note, ''), ' [已迁移 internal_sku:', $2::text, ']'), + updated_at = NOW() + WHERE id = $1`, + [rule.id, internalSku], + ) + } + + console.log( + JSON.stringify( + { + apply, + legacyRules: rules.rows.length, + convertible: converted.length, + converted, + unresolved, + }, + null, + 2, + ), + ) +} + +main() + .catch((error) => { + console.error(error) + process.exitCode = 1 + }) + .finally(() => closeDb()) diff --git a/apps/backend/scripts/merge-duplicate-products.ts b/apps/backend/scripts/merge-duplicate-products.ts index fe5f4ea3..ab7ab141 100644 --- a/apps/backend/scripts/merge-duplicate-products.ts +++ b/apps/backend/scripts/merge-duplicate-products.ts @@ -147,6 +147,22 @@ async function mergeGroup( [group.keeper_id], ) + await query( + ` + INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source + ) + VALUES ($1, 'affiliate_dash', 10, FALSE, TRUE, $2, 'merged'), + ($1, 'manual', 20, TRUE, TRUE, $2, 'merged') + ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), + is_fallback = internal_sku_channels.is_fallback OR EXCLUDED.is_fallback, + canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id), + updated_at = NOW() + `, + [group.canonical_name, group.keeper_id], + ) + await query( ` INSERT INTO sku_channel_rules ( diff --git a/apps/backend/scripts/preflight-fulfillment-routing.ts b/apps/backend/scripts/preflight-fulfillment-routing.ts new file mode 100644 index 00000000..10ed333a --- /dev/null +++ b/apps/backend/scripts/preflight-fulfillment-routing.ts @@ -0,0 +1,99 @@ +/** + * 履约/财务统一分流部署后核对（只读）。 + * + * 用法： + * npx tsx scripts/preflight-fulfillment-routing.ts + * npx tsx scripts/preflight-fulfillment-routing.ts --strict + */ +import { closeDb, query } from '../src/db/client.js' + +type CountRow = { label: string; total: number } + +async function main() { + const strict = process.argv.includes('--strict') + const checks: Record = {} + + checks.migrations = ( + await query<{ name: string }>(`SELECT name FROM schema_migrations ORDER BY id DESC LIMIT 5`) + ).rows.map((row) => row.name) + + checks.pendingReasons = ( + await query( + `SELECT COALESCE(fulfillment_pending_reason, '(null)') AS label, count(*)::int AS total + FROM order_items WHERE fulfillment_status = 'pending' + GROUP BY 1 ORDER BY 2 DESC`, + ) + ).rows + + checks.renameCardChannels = ( + await query( + `SELECT COALESCE(fulfillment_channel, '(pending)') AS label, count(*)::int AS total + FROM order_items + WHERE sku_name = '11月改名卡' + GROUP BY 1 ORDER BY 2 DESC`, + ) + ).rows + + checks.rules = ( + await query( + `SELECT match_key_type || '/' || channel || '/' || + CASE WHEN enabled THEN 'on' ELSE 'off' END AS label, count(*)::int AS total + FROM sku_channel_rules GROUP BY 1 ORDER BY 2 DESC`, + ) + ).rows + + checks.internalChannels = ( + await query( + `SELECT channel AS label, count(*)::int AS total + FROM internal_sku_channels WHERE enabled GROUP BY 1 ORDER BY 2 DESC`, + ) + ).rows + + checks.pendingWithoutChannel = ( + await query( + `SELECT 'pending_no_internal_channel' AS label, count(*)::int AS total + FROM order_items oi + WHERE oi.fulfillment_status = 'pending' + AND btrim(COALESCE(oi.manual_sku_key, '')) <> '' + AND NOT EXISTS ( + SELECT 1 FROM internal_sku_channels c + WHERE c.enabled AND btrim(c.internal_sku) = btrim(oi.manual_sku_key) + )`, + ) + ).rows[0] + + checks.channelCarrier = ( + await query<{ + channel: string + total: number + work_orders: number + affiliate_tasks: number + manual_tasks: number + }>( + `SELECT COALESCE(fulfillment_channel, '(pending)') AS channel, + count(*)::int AS total, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id))::int AS work_orders, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'))::int AS affiliate_tasks, + count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'))::int AS manual_tasks + FROM order_items oi + WHERE oi.fulfillment_status = 'assigned' + GROUP BY 1 ORDER BY 2 DESC`, + ) + ).rows + + const failed: string[] = [] + const rename = checks.renameCardChannels as CountRow[] + if (rename.some((row) => row.label === 'pending')) failed.push('11月改名卡仍有 pending') + if ((checks.pendingWithoutChannel as CountRow).total > 0) + failed.push('存在有内部 SKU 但无渠道能力的 pending') + + console.log(JSON.stringify({ strict, failed, ...checks }, null, 2)) + if (strict && failed.length > 0) process.exitCode = 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 index 5af914b5..2c32f19c 100644 --- a/apps/backend/scripts/reclassify-post-cutover.ts +++ b/apps/backend/scripts/reclassify-post-cutover.ts @@ -33,11 +33,20 @@ async function loadCandidates(limit: number): Promise { 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 + FROM internal_sku_channels c + WHERE c.enabled AND c.channel IN ('affiliate_dash', 'manual', 'third_party_voucher') - ORDER BY c.priority ASC + AND btrim(c.internal_sku) = COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), btrim(oi.sku_name)) + ORDER BY + c.priority ASC, + CASE c.channel + WHEN 'affiliate_dash' THEN 1 + WHEN 'work_order' THEN 2 + WHEN 'third_party_voucher' THEN 3 + WHEN 'manual' THEN 4 + ELSE 9 + END, + c.channel ASC LIMIT 1 ) cap ON TRUE WHERE oi.fulfillment_status = 'assigned' diff --git a/apps/backend/src/db/migrations/118_internal_sku_channels.sql b/apps/backend/src/db/migrations/118_internal_sku_channels.sql new file mode 100644 index 00000000..1a69325d --- /dev/null +++ b/apps/backend/src/db/migrations/118_internal_sku_channels.sql @@ -0,0 +1,67 @@ +-- 118_internal_sku_channels.sql +-- +-- 内部中文 SKU 的渠道决策单一真源。 +-- 从商品渠道能力 + 现有 internal_sku 全局规则回填；后续 intake/finance 都读这张表。 + +CREATE TABLE IF NOT EXISTS internal_sku_channels ( + internal_sku TEXT NOT NULL, + channel TEXT NOT NULL CHECK (channel IN ('work_order', 'affiliate_dash', 'third_party_voucher', 'manual')), + priority INTEGER NOT NULL DEFAULT 100, + is_fallback BOOLEAN NOT NULL DEFAULT FALSE, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + canonical_product_id BIGINT REFERENCES product_catalog(id) ON DELETE SET NULL, + config_json JSONB NOT NULL DEFAULT '{}'::jsonb, + source TEXT NOT NULL DEFAULT 'backfill', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (internal_sku, channel) +); + +CREATE INDEX IF NOT EXISTS idx_internal_sku_channels_priority + ON internal_sku_channels (internal_sku, priority, channel) + WHERE enabled = TRUE; + +COMMENT ON TABLE internal_sku_channels IS + '内部中文 SKU 渠道能力：priority 越小越优先，is_fallback 表示兜底渠道'; + +WITH capability_source AS ( + SELECT + btrim(p.canonical_name) AS internal_sku, + c.channel, + MIN(c.priority) AS priority, + BOOL_AND(c.is_fallback) AS is_fallback, + MIN(p.id) AS canonical_product_id + FROM product_catalog p + JOIN product_channel_capabilities c ON c.product_id = p.id AND c.enabled = TRUE + WHERE p.status = 'active' AND btrim(p.canonical_name) <> '' + GROUP BY 1, 2 +) +INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, canonical_product_id, source +) +SELECT internal_sku, channel, priority, is_fallback, canonical_product_id, 'capability' +FROM capability_source +ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), + is_fallback = internal_sku_channels.is_fallback AND EXCLUDED.is_fallback, + canonical_product_id = COALESCE(internal_sku_channels.canonical_product_id, EXCLUDED.canonical_product_id), + updated_at = NOW(); + +WITH rule_source AS ( + SELECT btrim(match_key) AS internal_sku, channel, MIN(priority) AS priority + FROM sku_channel_rules + WHERE match_key_type = 'internal_sku' + AND provider = '' AND shop_id = '' + AND enabled = TRUE + AND btrim(match_key) <> '' + GROUP BY 1, 2 +) +INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, canonical_product_id, source +) +SELECT internal_sku, channel, priority, FALSE, NULL, 'rule' +FROM rule_source +ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), + is_fallback = FALSE, + updated_at = NOW(); diff --git a/apps/backend/src/db/migrations/119_fulfillment_pending_reason.sql b/apps/backend/src/db/migrations/119_fulfillment_pending_reason.sql new file mode 100644 index 00000000..985f924c --- /dev/null +++ b/apps/backend/src/db/migrations/119_fulfillment_pending_reason.sql @@ -0,0 +1,14 @@ +-- 119_fulfillment_pending_reason.sql +-- +-- 待分类不再是一个模糊状态：落明确原因，供队列/审计区分。 +-- no_internal_sku 没有内部中文 SKU +-- no_channel_capability 有内部 SKU，但没有渠道能力/规则 +-- no_carrier 有渠道能力，但缺任务/工单载体 + +ALTER TABLE order_items + ADD COLUMN IF NOT EXISTS fulfillment_pending_reason TEXT, + ADD COLUMN IF NOT EXISTS fulfillment_pending_detail JSONB; + +CREATE INDEX IF NOT EXISTS idx_order_items_pending_reason + ON order_items (fulfillment_pending_reason) + WHERE fulfillment_status = 'pending'; 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 de89759c..47f1d870 100644 --- a/apps/backend/src/services/cost-center/product-normalize-service.ts +++ b/apps/backend/src/services/cost-center/product-normalize-service.ts @@ -1,6 +1,7 @@ import { query } from '../../db/client.js' import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exception-service.js' import { readWatermarkItemId } from '../watermark-store.js' +import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js' /** * 管道 B：商品归一。 @@ -191,6 +192,15 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise undefined) + // 同一内部中文 SKU 跨平台码统一分流到卡券渠道。 await query( ` diff --git a/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts new file mode 100644 index 00000000..794cb6cb --- /dev/null +++ b/apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts @@ -0,0 +1,195 @@ +import { query } from '../../db/client.js' +import { getOrderById } from '../../repositories/order-repo.js' +import { syncDeliveryTasksForOrder } from '../order/delivery-task-service.js' +import { syncWorkerOrdersForSourceOrder } from '../worker-platform/sync-worker-orders-service.js' +import { applyIntakeChannelDecision } from './intake-channel-decision-service.js' +import type { ResolvedFulfillmentItem } from './product-resolution-service.js' +import { logWarn } from '../../utils/logger.js' + +type PendingCarrierItem = { + id: number + order_id: number + sku_code: string + sku_name: string + quantity: number + item_snapshot_json: unknown + spec_json: unknown + channel: 'affiliate_dash' | 'work_order' +} + +function parseObject(value: unknown): Record { + if (value && typeof value === 'object' && !Array.isArray(value)) { + return value as Record + } + try { + const parsed = JSON.parse(String(value || '{}')) + return parsed && typeof parsed === 'object' && !Array.isArray(parsed) + ? (parsed as Record) + : {} + } catch { + return {} + } +} + +/** + * 为“内部 SKU 有渠道能力、但还没有载体”的 pending 单补载体。 + * + * - 接单平台：走现有工单匹配器，能匹配模板就建工单； + * - 自动发货：补 fulfillmentRoute/affiliateDash 后建任务； + * - 补不出来的继续保留 no_carrier，等运营补模板/映射。 + * + * 财务渠道判定不在本服务内做；调用方在补完载体后再跑一次 channel_assign。 + */ +export async function reconcilePendingCarriers(input: { limit?: number } = {}): Promise<{ + scanned: number + workOrdersCreated: number + affiliateTasksCreated: number + remaining: number +}> { + const limit = Math.min(500, Math.max(1, Math.trunc(Number(input.limit) || 200))) + const rows = await query( + ` + SELECT oi.id, oi.order_id, oi.sku_code, oi.sku_name, oi.quantity, + oi.item_snapshot_json, oi.spec_json, + cap.channel + FROM order_items oi + JOIN LATERAL ( + SELECT c.channel + FROM internal_sku_channels c + WHERE c.enabled = TRUE + AND c.channel IN ('affiliate_dash', 'work_order') + AND btrim(c.internal_sku) = COALESCE(NULLIF(btrim(oi.manual_sku_key), ''), btrim(oi.sku_name)) + ORDER BY + c.priority ASC, + CASE c.channel + WHEN 'affiliate_dash' THEN 1 + WHEN 'work_order' THEN 2 + ELSE 9 + END, + c.channel ASC + LIMIT 1 + ) cap ON TRUE + WHERE oi.fulfillment_status = 'pending' + AND oi.fulfillment_locked = FALSE + AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id) + AND NOT EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + ) + ORDER BY oi.id + LIMIT $1 + `, + [limit], + ) + + let workOrdersCreated = 0 + let affiliateTasksCreated = 0 + const byOrder = new Map() + for (const row of rows.rows) { + const list = byOrder.get(Number(row.order_id)) || [] + list.push(row) + byOrder.set(Number(row.order_id), list) + } + + for (const [orderId, items] of byOrder) { + const order = await getOrderById(orderId) + if (!order) continue + const itemRows = await query<{ + id: number + order_id: number + sku_code: string + sku_name: string + quantity: number + item_snapshot_json: unknown + spec_json: unknown + }>( + `SELECT id, order_id, sku_code, sku_name, quantity, item_snapshot_json, spec_json + FROM order_items WHERE order_id = $1`, + [orderId], + ) + const itemById = new Map(itemRows.rows.map((item) => [Number(item.id), item])) + + for (const target of items) { + const item = itemById.get(Number(target.id)) + if (!item) continue + + try { + if (target.channel === 'work_order') { + const result = await syncWorkerOrdersForSourceOrder(order, [item as never], { + source: 'carrier_reconcile', + autoOnly: true, + }) + workOrdersCreated += result.created.length + continue + } + + const resolved = { + ...item, + itemId: String(item.id), + externalItemId: String(item.id), + externalSkuCode: item.sku_code, + externalSkuName: item.sku_name, + skuCode: item.sku_code, + skuName: item.sku_name, + spec: parseObject(item.spec_json), + snapshot: parseObject(item.item_snapshot_json), + isConfigured: false, + } as unknown as ResolvedFulfillmentItem + const enriched = await applyIntakeChannelDecision(resolved, { + provider: order.provider, + platform: order.platform, + shopId: order.shop_id, + }) + const snapshot = parseObject(enriched.snapshot) + const route = parseObject(snapshot.fulfillmentRoute) + if (route.selectedExecutorKey !== 'affiliate_dash') continue + + await query( + `UPDATE order_items + SET sku_code = $2, + spec_json = $3, + item_snapshot_json = $4, + updated_at = NOW() + WHERE id = $1`, + [ + item.id, + String(enriched.skuCode || item.sku_code), + JSON.stringify(enriched.spec || {}), + JSON.stringify(snapshot), + ], + ) + const tasks = await syncDeliveryTasksForOrder(order, [ + { + ...item, + sku_code: String(enriched.skuCode || item.sku_code), + spec_json: enriched.spec || {}, + item_snapshot_json: snapshot, + } as never, + ]) + affiliateTasksCreated += tasks.length + } catch (error) { + logWarn('[carrier-reconciliation]', '补履约载体失败', { + orderItemId: target.id, + channel: target.channel, + error: error instanceof Error ? error.message : String(error), + }) + } + } + } + + const remainingResult = await query<{ count: number }>( + ` + SELECT count(*)::int AS count + FROM order_items oi + WHERE oi.fulfillment_status = 'pending' + AND oi.fulfillment_pending_reason = 'no_carrier' + `, + ) + + return { + scanned: rows.rows.length, + workOrdersCreated, + affiliateTasksCreated, + remaining: Number(remainingResult.rows[0]?.count || 0), + } +} diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts index 111b2a00..98de6e62 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.test.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.test.ts @@ -15,9 +15,12 @@ function buildItem(patch: Partial = {}): ChannelCandidateI sku_code: 'suit_xxx', sku_name: '套装-测试', manual_sku_key: 'suit_xxx', + internal_sku: '套装-测试', normalized_sku_name: '套装-测试', has_work_order: false, has_affiliate_task: false, + has_manual_task: false, + capability_channel: null, basis_date: '2026-10-31', ...patch, } @@ -98,7 +101,9 @@ test('渠道判定：内部 SKU 与 SKU 名称两种键都能命中', () => { channel: 'manual', }) assert.equal( - resolveChannelDecision(buildItem({ manual_sku_key: 'nick-001' }), [byInternal]).channel, + resolveChannelDecision(buildItem({ manual_sku_key: 'nick-001', internal_sku: 'nick-001' }), [ + byInternal, + ]).channel, 'manual', ) @@ -117,3 +122,28 @@ test('渠道判定：停用的规则不参与', () => { const decision = resolveChannelDecision(buildItem(), [buildRule({ enabled: false })]) assert.equal(decision.channel, null) }) + +test('渠道判定：卡券/人工能力可兜底，自动发货/接单必须靠载体', () => { + const voucher = resolveChannelDecision( + buildItem({ capability_channel: 'third_party_voucher' }), + [], + ) + assert.equal(voucher.channel, 'third_party_voucher') + assert.equal(voucher.source, 'auto_mapping') + + const manual = resolveChannelDecision(buildItem({ capability_channel: 'manual' }), []) + assert.equal(manual.channel, 'manual') + assert.equal(manual.source, 'auto_mapping') + + const affiliate = resolveChannelDecision(buildItem({ capability_channel: 'affiliate_dash' }), []) + assert.equal(affiliate.channel, null, '自动发货能力没有任务时不能只改财务口径') + + const workOrder = resolveChannelDecision(buildItem({ capability_channel: 'work_order' }), []) + assert.equal(workOrder.channel, null, '接单能力没有工单时不能只改财务口径') +}) + +test('渠道判定：人工任务作为人工载体证据', () => { + const decision = resolveChannelDecision(buildItem({ has_manual_task: true }), []) + assert.equal(decision.channel, 'manual') + assert.equal(decision.source, 'carrier_evidence') +}) diff --git a/apps/backend/src/services/fulfillment/channel-assign-service.ts b/apps/backend/src/services/fulfillment/channel-assign-service.ts index 82163509..4fe3ee6c 100644 --- a/apps/backend/src/services/fulfillment/channel-assign-service.ts +++ b/apps/backend/src/services/fulfillment/channel-assign-service.ts @@ -63,6 +63,7 @@ export type ChannelCandidateItem = { normalized_sku_name: string | null has_work_order: boolean has_affiliate_task: boolean + has_manual_task: boolean capability_channel: FulfillmentChannel | null basis_date: string } @@ -91,6 +92,25 @@ function itemKey(item: ChannelCandidateItem, keyType: MatchKeyType): string { return normalizeKey(item.normalized_sku_name) || normalizeKey(item.sku_name).toLowerCase() } +function resolvePendingReason(item: ChannelCandidateItem): { + reason: string + detail: Record +} { + if (!String(item.internal_sku || '').trim()) { + return { reason: 'no_internal_sku', detail: { skuCode: item.sku_code } } + } + if (!item.capability_channel) { + return { + reason: 'no_channel_capability', + detail: { internalSku: item.internal_sku }, + } + } + return { + reason: 'no_carrier', + detail: { internalSku: item.internal_sku, channel: item.capability_channel }, + } +} + /** 纯函数：给定规则与订单项，得出渠道决策（便于单测与文档对照）。 */ export function resolveChannelDecision( item: ChannelCandidateItem, @@ -144,6 +164,15 @@ export function resolveChannelDecision( matchedRuleIds: [], } } + if (item.has_manual_task) { + return { + channel: 'manual', + source: 'carrier_evidence', + conflict: false, + conflictChannels: [], + matchedRuleIds: [], + } + } // 卡券/人工没有必须创建的履约任务，可直接用商品渠道能力兜底； // 自动发货/接单平台仍要求真实载体，避免只改口径不派单。 @@ -173,6 +202,8 @@ export type AssignChannelsResult = { affectedDates: string[] /** 本轮新判定的订单项（供立即计算成本，最多 2000 条）。 */ assignedItemIds: number[] + /** 待分类原因计数。 */ + pendingReasons: Record } /** 批量判定：默认只扫 pending 且未锁定的订单项。 */ @@ -198,6 +229,7 @@ export async function assignPendingOrderItemChannels( byChannel: {}, affectedDates: [], assignedItemIds: [], + pendingReasons: {}, } const affectedDates = new Set() @@ -222,17 +254,20 @@ export async function assignPendingOrderItemChannels( SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' ) AS has_affiliate_task, + EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + ) AS has_manual_task, cap.channel AS capability_channel FROM order_items oi JOIN orders o ON o.id = oi.order_id LEFT JOIN LATERAL ( SELECT c.channel - FROM product_catalog p - JOIN product_channel_capabilities c - ON c.product_id = p.id AND c.enabled = TRUE - WHERE p.status = 'active' - AND btrim(p.canonical_name) = ${INTERNAL_SKU_SQL} + FROM internal_sku_channels c + WHERE c.enabled = TRUE + AND btrim(c.internal_sku) = ${INTERNAL_SKU_SQL} ORDER BY + c.priority ASC, CASE c.channel WHEN 'affiliate_dash' THEN 1 WHEN 'work_order' THEN 2 @@ -240,7 +275,7 @@ export async function assignPendingOrderItemChannels( WHEN 'manual' THEN 4 ELSE 9 END, - c.priority ASC + c.channel ASC LIMIT 1 ) cap ON TRUE WHERE oi.fulfillment_status = 'pending' @@ -282,6 +317,16 @@ export async function assignPendingOrderItemChannels( } if (!decision.channel) { result.stillPending += 1 + const pending = resolvePendingReason(item) + result.pendingReasons[pending.reason] = (result.pendingReasons[pending.reason] || 0) + 1 + await query( + `UPDATE order_items + SET fulfillment_pending_reason = $2, + fulfillment_pending_detail = $3, + updated_at = NOW() + WHERE id = $1 AND fulfillment_status = 'pending'`, + [item.id, pending.reason, JSON.stringify(pending.detail)], + ) continue } assignments.push({ id: item.id, channel: decision.channel, source: decision.source }) @@ -297,6 +342,8 @@ export async function assignPendingOrderItemChannels( fulfillment_status = 'assigned', fulfillment_source = input.source, fulfillment_decided_at = NOW(), + fulfillment_pending_reason = NULL, + fulfillment_pending_detail = NULL, updated_at = NOW() FROM unnest($1::bigint[], $2::text[], $3::text[]) AS input(id, channel, source) @@ -385,6 +432,7 @@ export type PendingGroupRow = { internal_sku: string sku_name: string sku_codes: string[] + pending_reasons: string[] item_count: number revenue_fen: number first_seen_at: string @@ -432,6 +480,7 @@ export async function listPendingChannelGroups(input: { ${INTERNAL_SKU_SQL} AS internal_sku, MIN(oi.sku_name) AS sku_name, array_agg(DISTINCT oi.sku_code) AS sku_codes, + array_remove(array_agg(DISTINCT oi.fulfillment_pending_reason), NULL) AS pending_reasons, COUNT(*)::int AS item_count, COALESCE(SUM(oi.revenue_amount_fen), 0)::bigint AS revenue_fen, MIN(oi.created_at) AS first_seen_at, diff --git a/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts b/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts new file mode 100644 index 00000000..0210b7c2 --- /dev/null +++ b/apps/backend/src/services/fulfillment/intake-channel-decision-service.ts @@ -0,0 +1,125 @@ +import { + listInternalSkuChannels, + type InternalSkuChannelRow, +} from './internal-sku-channel-service.js' +import type { ResolvedFulfillmentItem } from './product-resolution-service.js' +import { matchAffiliateDashSkuByInternalSku } from './product-resolution-service.js' +import { logWarn } from '../../utils/logger.js' + +type IntakeChannelDecisionInput = { + provider?: string + platform?: string + shopId?: string +} + +function parseObject(value: unknown): Record { + return value && typeof value === 'object' && !Array.isArray(value) + ? (value as Record) + : {} +} + +function resolveInternalSku(item: ResolvedFulfillmentItem): string { + const snapshot = parseObject(item.snapshot) + const spec = parseObject(item.spec) + return String( + snapshot.internalSku || spec.skuNick || spec.outSkuSn || item.skuCode || item.skuName || '', + ).trim() +} + +/** + * 下单入口读取内部 SKU 渠道决策。 + * + * 目前只补“自动发货”载体：若内部 SKU 能力是 affiliate_dash 且能解析出 dash sku， + * 直接写 fulfillmentRoute + affiliateDash，让 delivery-task-service 建任务。 + * 接单平台仍由 work_order 匹配器创建；卡券/人工无任务，由财务侧兜底。 + */ +export async function applyIntakeChannelDecision( + item: ResolvedFulfillmentItem, + _input: IntakeChannelDecisionInput, +): Promise { + const internalSku = resolveInternalSku(item) + if (!internalSku) return item + + let channels: InternalSkuChannelRow[] = [] + try { + channels = await listInternalSkuChannels(internalSku) + } catch (error) { + // 单测/旧库没有 118 迁移时不影响下单主流程，生产由迁移保证。 + logWarn('[intake-channel-decision]', '读取内部 SKU 渠道失败', { + internalSku, + error: error instanceof Error ? error.message : String(error), + }) + return item + } + if (channels.length === 0) return item + + const snapshot = parseObject(item.snapshot) + const route = parseObject(snapshot.fulfillmentRoute) + const selected = String(route.selectedExecutorKey || '').trim() + // 路由配置未命中时会默认落 manual_dispatch；这不是运营显式规则，不能覆盖内部 SKU 能力。 + const isDefaultManualFallback = selected === 'manual_dispatch' && !route.matchedRule + const primary = channels[0]! + const nextSnapshot: Record = { + ...snapshot, + internalSku: String(snapshot.internalSku || internalSku), + internalSkuChannel: primary.channel, + internalSkuChannelSource: primary.source, + } + + const canOverrideDefaultManual = !selected || isDefaultManualFallback + const existingAffiliateSku = String(parseObject(snapshot.affiliateDash).sku || '').trim() + + if (primary.channel === 'affiliate_dash' || selected === 'affiliate_dash') { + const dashSku = existingAffiliateSku || matchAffiliateDashSkuByInternalSku(internalSku) + if (dashSku) { + if (canOverrideDefaultManual || selected === 'affiliate_dash') { + nextSnapshot.fulfillmentRoute = { + ...route, + selectedExecutorKey: 'affiliate_dash', + selectedRuleId: String(route.selectedRuleId || ''), + reason: canOverrideDefaultManual + ? '内部 SKU 渠道能力指定自动发货' + : String(route.reason || '自动发货路由'), + matchedRule: route.matchedRule ?? null, + candidates: Array.isArray(route.candidates) ? route.candidates : [], + skipped: Array.isArray(route.skipped) ? route.skipped : [], + } + nextSnapshot.affiliateDash = { + matchMode: 'affiliate_dash_sku_mapping', + productNo: internalSku, + sku: dashSku, + skuName: internalSku, + } + // 商品归一按 aff: 找商品；内部中文 SKU 固定落 spec.skuNick。 + return { + ...item, + skuCode: dashSku, + spec: { + ...(item.spec && typeof item.spec === 'object' ? item.spec : {}), + skuNick: internalSku, + }, + isConfigured: true, + snapshot: nextSnapshot, + } + } + } + nextSnapshot.internalSkuChannelReason = 'missing_affiliate_dash_mapping' + return { ...item, snapshot: nextSnapshot } + } + + // 内部能力是接单/卡券，但路由只是默认人工兜底：清掉人工任务，交给工单或财务能力兜底。 + if ( + (primary.channel === 'work_order' || primary.channel === 'third_party_voucher') && + isDefaultManualFallback + ) { + nextSnapshot.fulfillmentRoute = { + ...route, + selectedExecutorKey: '', + reason: `内部 SKU 渠道能力为 ${primary.channel}，不使用默认人工兜底`, + matchedRule: null, + } + return { ...item, isConfigured: false, snapshot: nextSnapshot } + } + + return { ...item, snapshot: nextSnapshot } +} diff --git a/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts b/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts new file mode 100644 index 00000000..e55de975 --- /dev/null +++ b/apps/backend/src/services/fulfillment/internal-sku-channel-service.ts @@ -0,0 +1,83 @@ +import { query } from '../../db/client.js' +import type { FulfillmentChannel } from './channel-types.js' + +export type InternalSkuChannelRow = { + internal_sku: string + channel: FulfillmentChannel + priority: number + is_fallback: boolean + enabled: boolean + canonical_product_id: number | null + source: string +} + +export async function listInternalSkuChannels( + internalSku: string, +): Promise { + const key = String(internalSku || '').trim() + if (!key) return [] + const result = await query( + ` + SELECT internal_sku, channel, priority, is_fallback, enabled, canonical_product_id, source + FROM internal_sku_channels + WHERE btrim(internal_sku) = $1 AND enabled = TRUE + ORDER BY + priority ASC, + CASE channel + WHEN 'affiliate_dash' THEN 1 + WHEN 'work_order' THEN 2 + WHEN 'third_party_voucher' THEN 3 + WHEN 'manual' THEN 4 + ELSE 9 + END, + channel ASC + `, + [key], + ) + return result.rows +} + +/** 财务侧可直接兜底的渠道：卡券/人工不要求新建履约任务。 */ +export async function resolveInternalSkuFallbackChannel( + internalSku: string, +): Promise { + const channels = await listInternalSkuChannels(internalSku) + const safe = channels.find( + (row) => row.channel === 'third_party_voucher' || row.channel === 'manual', + ) + return safe?.channel || null +} + +export async function upsertInternalSkuChannel(input: { + internalSku: string + channel: FulfillmentChannel + priority?: number + isFallback?: boolean + canonicalProductId?: number | null + source?: string +}): Promise { + const internalSku = String(input.internalSku || '').trim() + if (!internalSku) return + await query( + ` + INSERT INTO internal_sku_channels ( + internal_sku, channel, priority, is_fallback, enabled, + canonical_product_id, source, created_at, updated_at + ) VALUES ($1, $2, $3, $4, TRUE, $5, $6, NOW(), NOW()) + ON CONFLICT (internal_sku, channel) DO UPDATE SET + priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority), + is_fallback = internal_sku_channels.is_fallback AND EXCLUDED.is_fallback, + canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id), + enabled = TRUE, + updated_at = NOW() + `, + [ + internalSku, + input.channel, + Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100, + input.isFallback === true, + input.canonicalProductId ?? null, + String(input.source || 'auto').trim() || 'auto', + ], + ) +} diff --git a/apps/backend/src/services/fulfillment/product-resolution-service.ts b/apps/backend/src/services/fulfillment/product-resolution-service.ts index 87923372..b87ec2cb 100644 --- a/apps/backend/src/services/fulfillment/product-resolution-service.ts +++ b/apps/backend/src/services/fulfillment/product-resolution-service.ts @@ -377,6 +377,15 @@ function resolveDouyinMashidaItemCandidate({ * 映射来源 = **权威常量表 + 后台增量覆盖**（`affiliate-dash` 配置面板的 `skuMapping`）， * 与 91 分支同一套来源：新增抖音商品在后台加一行即可，不必改代码发版。 */ +export function matchAffiliateDashSkuByInternalSku(internalSku: unknown): string { + const raw = String(internalSku ?? '').trim() + if (!raw) return '' + return matchAffiliateDashSku( + mergeAffiliateDashSkuMapping(getAffiliateDashConfig().skuMapping), + raw, + ) +} + export function matchDouyinMashidaAffiliateDashSku(internalSku: unknown): string { const raw = String(internalSku ?? '').trim() if (!raw) { diff --git a/apps/backend/src/services/order/order-service.ts b/apps/backend/src/services/order/order-service.ts index 29b5653a..3ef5a81a 100644 --- a/apps/backend/src/services/order/order-service.ts +++ b/apps/backend/src/services/order/order-service.ts @@ -10,6 +10,7 @@ import { import { withTransaction } from '../../db/client.js' import { syncDeliveryTasksForOrder } from './delivery-task-service.js' import { resolveOrderItemForFulfillment } from '../fulfillment/product-resolution-service.js' +import { applyIntakeChannelDecision } from '../fulfillment/intake-channel-decision-service.js' import { resolveOrderFulfillmentReadiness, syncOrderFulfillmentAttachments, @@ -113,14 +114,19 @@ export async function upsertOrderFromSource( }) const resolvedItems = await Promise.all( - event.items.map((item) => - resolveOrderItemForFulfillment({ + event.items.map(async (item) => { + const resolved = await resolveOrderItemForFulfillment({ provider: event.provider, platform: event.platform, shopId: event.shopId, item, - }), - ), + }) + return applyIntakeChannelDecision(resolved, { + provider: event.provider, + platform: event.platform, + shopId: event.shopId, + }) + }), ) const resolvedFulfillmentItems = resolvedItems as FulfillmentOrderItem[] const configuredItems = resolvedFulfillmentItems.filter((item) => item.isConfigured) diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 303e3dfd..bafc610d 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -2,6 +2,7 @@ import type { JsonObject } from '../../types/json.js' import { query } from '../../db/client.js' import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js' import { runAffiliateFallback } from '../fulfillment/affiliate-fallback-service.js' +import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js' import { resolveExceptionsForItemsByReason, resolveOrphanCostExceptions, @@ -84,6 +85,10 @@ export async function runChannelAssignJob(job: JsonObject = {}) { export async function runCarrierDispatchJob(job: JsonObject = {}) { const config = jobConfig(job) const limit = clampInt(config.limit, 200, 1, 2000) + const reconciled = await reconcilePendingCarriers({ limit: Math.min(limit, 200) }) + if (reconciled.affiliateTasksCreated > 0) { + await assignPendingOrderItemChannels({ batchSize: 200, maxBatches: 1 }).catch(() => undefined) + } const cursor = config.afterId != null ? Math.max(0, Math.trunc(Number(config.afterId))) diff --git a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx index 5e4d7de7..430eea91 100644 --- a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx +++ b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx @@ -126,6 +126,19 @@ function PendingPanel() { render: (value: string[] | undefined) => (value?.length ? value.join(' / ') : '-'), }, { title: '商品名', dataIndex: 'sku_name', ellipsis: true }, + { + title: '待分类原因', + dataIndex: 'pending_reasons', + width: 160, + render: (value: string[] | undefined) => { + const labels: Record = { + no_internal_sku: '无内部SKU', + no_channel_capability: '无渠道能力', + no_carrier: '无履约载体', + } + return value?.length ? value.map((reason) => labels[reason] || reason).join(' / ') : '-' + }, + }, { title: '订单数', dataIndex: 'item_count', width: 90 }, { title: '收入', diff --git a/apps/frontend/src/services/admin/fulfillment-channel.ts b/apps/frontend/src/services/admin/fulfillment-channel.ts index 28e20d9e..3ae7be53 100644 --- a/apps/frontend/src/services/admin/fulfillment-channel.ts +++ b/apps/frontend/src/services/admin/fulfillment-channel.ts @@ -7,6 +7,7 @@ export type PendingChannelGroup = { internal_sku: string sku_name: string sku_codes: string[] + pending_reasons: string[] item_count: number revenue_fen: number first_seen_at: string diff --git a/docs/生产部署核对清单-2026-11-02.md b/docs/生产部署核对清单-2026-11-02.md new file mode 100644 index 00000000..350dc0f7 --- /dev/null +++ b/docs/生产部署核对清单-2026-11-02.md @@ -0,0 +1,141 @@ +# 生产部署核对清单（履约与财务分流统一） + +日期：2026-11-02 +适用：生产当前在迁移 115，本次新增 116–119 + backend/frontend。 + +## 1. 部署前 + +1. 备份生产数据库（pg_dump，报告文件大小和 sha256）。 +2. 记录当前： + - `schema_migrations` 最大版本； + - pending 数量及原因； + - `order_items` 渠道分布； + - `sku_channel_rules` 匹配键分布。 +3. 确认没有正在跑的财务管道（或暂停定时任务）。 +4. 本地/预发已通过： + - 后端测试、前端测试、build、lint、typecheck、SQL guard； + - 三平台 E2E（自动发货 / 卡券 / 接单 / 抖音自动发货）； + - 成本、销售事实金额正确。 + +## 2. 迁移顺序 + +```text +116_voucher_internal_sku_backfill +117_internal_sku_mappings +118_internal_sku_channels +119_fulfillment_pending_reason +``` + +重点检查： + +- 116：给已有 `tpv:` 商品补 capability 和 internal_sku 规则； +- 118：按商品能力/规则回填 `internal_sku_channels`； + - SQL 已做聚合，避免同名商品同渠道 `ON CONFLICT` 二次更新报错； +- 119：只加列和索引，风险低。 + +## 3. 部署顺序 + +1. 执行迁移 116–119； +2. 部署 backend； +3. 部署 frontend； +4. 启动/恢复定时任务。 + +## 4. 部署后核对 + +### 待分类 + +```sql +SELECT fulfillment_pending_reason, count(*) +FROM order_items +WHERE fulfillment_status = 'pending' +GROUP BY 1; +``` + +预期只剩： + +- `no_internal_sku` +- `no_channel_capability` +- `no_carrier` + +### 11月改名卡 + +```sql +SELECT oi.fulfillment_channel, count(*) +FROM order_items oi +WHERE oi.sku_name = '11月改名卡' + AND oi.fulfillment_status = 'assigned' +GROUP BY 1; +``` + +预期：`third_party_voucher`，不再 manual / pending。 + +### 财务与载体一致 + +```sql +SELECT + oi.fulfillment_channel, + count(*) FILTER (WHERE EXISTS ( + SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id + )) AS work_orders, + count(*) FILTER (WHERE EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash' + )) AS affiliate_tasks, + count(*) FILTER (WHERE EXISTS ( + SELECT 1 FROM fulfillment_tasks ft + WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' + )) AS manual_tasks +FROM order_items oi +WHERE oi.fulfillment_status = 'assigned' +GROUP BY 1; +``` + +### 规则 + +```sql +SELECT match_key_type, channel, enabled, count(*) +FROM sku_channel_rules +GROUP BY 1,2,3 +ORDER BY 1,2,3; +``` + +旧 `sku_code` 规则应只剩无法定位内部 SKU 的少量记录。 + +### 内部 SKU 渠道表 + +```sql +SELECT channel, count(*) +FROM internal_sku_channels +WHERE enabled +GROUP BY 1; +``` + +## 5. 数据清理（确认后执行） + +```bash +# 先 dry-run +npx tsx scripts/merge-duplicate-products.ts +npx tsx scripts/cleanup-legacy-sku-code-rules.ts +npx tsx scripts/reclassify-post-cutover.ts + +# 确认后 apply +npx tsx scripts/merge-duplicate-products.ts --apply +npx tsx scripts/cleanup-legacy-sku-code-rules.ts --apply +npx tsx scripts/reclassify-post-cutover.ts --apply +``` + +顺序： + +1. 合并重复商品； +2. 清理旧 sku_code 规则； +3. 重判断代后订单； +4. 跑一次完整管道； +5. 核对成本/销售事实。 + +## 6. 回滚 + +- 代码：回退到部署前镜像/commit； +- 数据库： + - 116–119 均为新增表/列/数据，不删旧表，可保留； + - 若必须回滚，恢复到部署前备份； +- 迁移 118/119 不改旧字段语义，回滚代码后旧逻辑仍可运行。