#!/usr/bin/env tsx /** * 从统一履约/商品真源重建财务派生数据。 * * 只删除可重建数据: * - estimated 成本快照; * - 指定期间的销售事实; * - 商品/成本类异常明细。 * * 永久保留: * - 订单、订单项、退款、真实资金流水、人工支出; * - actual 成本快照; * - 全部已过账凭证;销售凭证只追加反向/补差调整,不删除; * - 手工凭证与已导入的平台对账原始批次。 * * 默认 dry-run;执行需要显式确认。 * * npm run finance:rebuild-derived * npm run finance:rebuild-derived -- --execute --confirm=REBUILD_DERIVED */ import process from 'node:process' import { closeDb, query } from '../src/db/client.js' import { syncCostSnapshotsForItems } from '../src/services/cost-center/cost-snapshot-service.js' import { resolveExceptionsForItemsByReason, upsertAggregatedExceptionsBatch, type AggregatedExceptionEntry, type CostExceptionReason, } from '../src/services/cost-center/cost-exception-service.js' import { normalizeProductsForItems } from '../src/services/cost-center/product-normalize-service.js' import { syncFinanceLedgerFromFacts } from '../src/services/finance/finance-ledger-service.js' import { rebuildChannelSalesDailyFacts } from '../src/services/finance/sales-facts-channel-service.js' const CONFIRMATION = 'REBUILD_DERIVED' const REBUILD_EXCEPTION_REASONS = [ 'product_mapping_missing', 'cost_missing', 'manual_cost_missing', 'on_account_cost_pending', ] as const satisfies readonly CostExceptionReason[] const options = parseOptions(process.argv.slice(2)) const execute = options.execute === 'true' const confirmation = String(options.confirm || '').trim() async function main() { await assertSchema() const bounds = await resolveDateRange() const closedPeriods = await loadClosedPeriods(bounds.dateFrom, bounds.dateTo) if (closedPeriods.length > 0) { throw new Error( `区间包含已关账期间 ${closedPeriods.join(', ')};请先反关账,禁止直接重建已关账数据`, ) } const before = await loadProtectedTotals() const scopeItemIds = await loadScopeItemIds(bounds.dateFrom, bounds.dateTo) const plan = await loadPlan(bounds.dateFrom, bounds.dateTo) const startedAt = Date.now() if (!execute) { console.log( JSON.stringify( { mode: 'dry-run', ...bounds, scopeItems: scopeItemIds.length, closedPeriods, plan, protectedTotals: before, hint: `确认后执行:npm run finance:rebuild-derived -- --execute --confirm=${CONFIRMATION}`, }, null, 2, ), ) return } if (confirmation !== CONFIRMATION) { throw new Error(`--confirm 必须是 ${CONFIRMATION}`) } const clearedExceptions = await clearFinanceExceptions(scopeItemIds) const clearedSnapshots = await clearEstimatedSnapshots(bounds.dateFrom, bounds.dateTo) const clearedFacts = await clearSalesFacts(bounds.dateFrom, bounds.dateTo) const normalization = { matched: 0, unmatched: 0, batches: 0 } const cost = { created: 0, superseded: 0, missing: 0, affectedDates: new Set() } for (const ids of chunk(scopeItemIds, 2000)) { const normalized = await normalizeProductsForItems(ids) normalization.matched += normalized.matched normalization.unmatched += normalized.unmatched normalization.batches += 1 const snapshots = await syncCostSnapshotsForItems(ids) cost.created += snapshots.created cost.superseded += snapshots.superseded cost.missing += snapshots.missing for (const date of snapshots.affectedDates) cost.affectedDates.add(date) } const facts = await rebuildChannelSalesDailyFacts(bounds) const ledger = await syncFinanceLedgerFromFacts(bounds) const regeneratedExceptions = await regenerateCostExceptions(scopeItemIds) const after = await loadProtectedTotals() assertProtectedTotals(before, after) console.log( JSON.stringify( { mode: 'execute', ...bounds, scopeItems: scopeItemIds.length, cleared: { estimatedSnapshots: clearedSnapshots, salesFacts: clearedFacts, financeExceptionItems: clearedExceptions, }, normalization, cost: { ...cost, affectedDates: [...cost.affectedDates].sort(), }, facts, ledger, regeneratedExceptions, protectedTotals: after, durationMs: Date.now() - startedAt, auditCommand: 'npm run audit:finance -- --strict', }, null, 2, ), ) } async function resolveDateRange() { const bounds = await query<{ date_from: string | null; date_to: string | null }>( `WITH dates AS ( SELECT stat_date AS stat_date FROM v_finance_order_item_scope UNION ALL SELECT stat_date FROM sales_daily_facts UNION ALL SELECT (occurred_at AT TIME ZONE 'Asia/Shanghai')::date FROM finance_journal_entries WHERE ref_type = 'sales_fact' AND status = 'posted' ) SELECT MIN(stat_date)::text AS date_from, MAX(stat_date)::text AS date_to FROM dates`, ) const dateFrom = normalizeDate(options.dateFrom) || bounds.rows[0]?.date_from const dateTo = normalizeDate(options.dateTo) || bounds.rows[0]?.date_to || dateFrom if (!dateFrom || !dateTo) throw new Error('财务销售范围没有可用业务日') if (dateFrom > dateTo) throw new Error('date-from 不能晚于 date-to') return { dateFrom, dateTo } } async function loadClosedPeriods(dateFrom: string, dateTo: string): Promise { const result = await query<{ period_key: string }>( `SELECT period_key FROM finance_periods WHERE status = 'closed' AND period_key BETWEEN to_char($1::date, 'YYYY-MM') AND to_char($2::date, 'YYYY-MM') ORDER BY period_key`, [dateFrom, dateTo], ) return result.rows.map((row) => row.period_key) } async function loadScopeItemIds(dateFrom: string, dateTo: string): Promise { const result = await query<{ order_item_id: number }>( `SELECT order_item_id FROM v_finance_order_item_scope WHERE stat_date BETWEEN $1::date AND $2::date ORDER BY order_item_id`, [dateFrom, dateTo], ) return result.rows.map((row) => Number(row.order_item_id)) } async function loadPlan(dateFrom: string, dateTo: string) { const result = await query<{ estimated_snapshots: number sales_facts: number sales_journal_entries_to_adjust: number finance_exception_items: number }>( `SELECT ( SELECT COUNT(*)::int FROM cost_snapshots snapshot JOIN v_finance_order_item_scope scope ON scope.order_item_id = snapshot.order_item_id WHERE scope.stat_date BETWEEN $1::date AND $2::date AND snapshot.snapshot_stage = 'estimated' ) AS estimated_snapshots, ( SELECT COUNT(*)::int FROM sales_daily_facts WHERE stat_date BETWEEN $1::date AND $2::date ) AS sales_facts, ( SELECT COUNT(*)::int FROM finance_journal_entries WHERE ref_type = 'sales_fact' AND occurred_at >= ($1::date::timestamp AT TIME ZONE 'Asia/Shanghai') AND occurred_at < (($2::date + interval '1 day')::timestamp AT TIME ZONE 'Asia/Shanghai') ) AS sales_journal_entries_to_adjust, ( SELECT COUNT(*)::int FROM cost_exception_items item JOIN cost_exceptions exception ON exception.id = item.exception_id JOIN v_finance_order_item_scope scope ON scope.order_item_id = item.order_item_id WHERE exception.reason = ANY($3::text[]) AND scope.stat_date BETWEEN $1::date AND $2::date ) AS finance_exception_items`, [dateFrom, dateTo, REBUILD_EXCEPTION_REASONS], ) const row = result.rows[0] return { estimatedSnapshots: Number(row?.estimated_snapshots || 0), salesFacts: Number(row?.sales_facts || 0), salesJournalEntriesToAdjust: Number(row?.sales_journal_entries_to_adjust || 0), financeExceptionItems: Number(row?.finance_exception_items || 0), } } async function clearFinanceExceptions(itemIds: number[]): Promise { let resolved = 0 for (const ids of chunk(itemIds, 2000)) { for (const reason of REBUILD_EXCEPTION_REASONS) { resolved += await resolveExceptionsForItemsByReason(ids, reason) } } return resolved } async function clearEstimatedSnapshots(dateFrom: string, dateTo: string): Promise { const result = await query( `DELETE FROM cost_snapshots snapshot USING v_finance_order_item_scope scope WHERE scope.order_item_id = snapshot.order_item_id AND scope.stat_date BETWEEN $1::date AND $2::date AND snapshot.snapshot_stage = 'estimated'`, [dateFrom, dateTo], ) return result.rowCount || 0 } async function clearSalesFacts(dateFrom: string, dateTo: string): Promise { const result = await query( `DELETE FROM sales_daily_facts WHERE stat_date BETWEEN $1::date AND $2::date`, [dateFrom, dateTo], ) return result.rowCount || 0 } async function regenerateCostExceptions(itemIds: number[]) { const manualRows = itemIds.length ? await query<{ order_item_id: number; sku_key: string }>( `SELECT DISTINCT ON (scope.order_item_id) scope.order_item_id, resolve_internal_sku(item.manual_sku_key, item.sku_name, item.sku_code) AS sku_key FROM v_finance_order_item_scope scope JOIN order_items item ON item.id = scope.order_item_id JOIN fulfillments fulfillment ON fulfillment.order_item_id = scope.order_item_id WHERE scope.order_item_id = ANY($1::bigint[]) AND fulfillment.channel = 'manual' AND fulfillment.fulfillment_config_id IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM cost_snapshots snapshot WHERE snapshot.subject_type = 'manual_case' AND snapshot.order_item_id = scope.order_item_id AND snapshot.superseded_at IS NULL AND snapshot.reversed_at IS NULL ) ORDER BY scope.order_item_id`, [itemIds], ) : { rows: [] } const onAccountRows = itemIds.length ? await query<{ order_item_id: number; sku_key: string; bill_id: number }>( `SELECT bill.order_item_id, bill.sku_key, bill.id AS bill_id FROM on_account_bills bill WHERE bill.status = 'pending' AND bill.order_item_id = ANY($1::bigint[]) ORDER BY bill.order_item_id`, [itemIds], ) : { rows: [] } const entries: AggregatedExceptionEntry[] = [ ...manualRows.rows.map((row) => ({ reason: 'manual_cost_missing', scopeKey: `manual_cost_missing|-|-|${row.sku_key}|manual`, orderItemId: Number(row.order_item_id), detail: { skuKey: row.sku_key }, })), ...onAccountRows.rows.map((row) => ({ reason: 'on_account_cost_pending', scopeKey: `on_account_cost_pending|-|-|${row.sku_key}|on_account`, orderItemId: Number(row.order_item_id), detail: { billId: Number(row.bill_id), internalSku: row.sku_key }, })), ] await upsertAggregatedExceptionsBatch(entries) return { manualCostMissing: manualRows.rows.length, onAccountCostPending: onAccountRows.rows.length, linked: entries.length, } } async function loadProtectedTotals() { const result = await query<{ orders: number order_items: number order_revenue_fen: string refunds: number refund_fen: string manual_expenses: number manual_expense_fen: string fund_entries: number }>( `SELECT (SELECT COUNT(*)::int FROM orders) AS orders, (SELECT COUNT(*)::int FROM order_items) AS order_items, (SELECT COALESCE(SUM(revenue_amount_fen), 0)::bigint FROM order_items) AS order_revenue_fen, (SELECT COUNT(*)::int FROM order_refunds WHERE status = 'confirmed') AS refunds, (SELECT COALESCE(SUM(amount_fen), 0)::bigint FROM order_refunds WHERE status = 'confirmed') AS refund_fen, (SELECT COUNT(*)::int FROM manual_expenses) AS manual_expenses, (SELECT COALESCE(SUM(amount_fen), 0)::bigint FROM manual_expenses) AS manual_expense_fen, (SELECT COUNT(*)::int FROM fund_entries) AS fund_entries`, ) const row = result.rows[0] return { orders: Number(row?.orders || 0), orderItems: Number(row?.order_items || 0), orderRevenueFen: Number(row?.order_revenue_fen || 0), refunds: Number(row?.refunds || 0), refundFen: Number(row?.refund_fen || 0), manualExpenses: Number(row?.manual_expenses || 0), manualExpenseFen: Number(row?.manual_expense_fen || 0), fundEntries: Number(row?.fund_entries || 0), } } function assertProtectedTotals( before: Awaited>, after: Awaited>, ) { if (JSON.stringify(before) !== JSON.stringify(after)) { throw new Error(`原始订单/退款/资金数据在重建前后不守恒:${JSON.stringify({ before, after })}`) } } async function assertSchema() { const result = await query<{ ready: boolean }>( `SELECT to_regclass('public.v_finance_order_item_scope') IS NOT NULL AND to_regclass('public.fulfillments') IS NOT NULL AND to_regclass('public.v_order_item_current') IS NOT NULL AS ready`, ) if (!result.rows[0]?.ready) { throw new Error('财务真源视图缺失;请先执行 npm run db:migrate') } } function chunk(items: T[], size: number): T[][] { const result: T[][] = [] for (let index = 0; index < items.length; index += size) { result.push(items.slice(index, index + size)) } return result } 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 normalizeDate(value: unknown): string | null { const text = String(value || '').trim() return /^\d{4}-\d{2}-\d{2}$/.test(text) ? text : null } main() .catch((error) => { console.error(error instanceof Error ? error.message : error) process.exitCode = 1 }) .finally(() => closeDb())