重构：统一财务派生数据真源与重建
This commit is contained in:
yml2213
@@ -0,0 +1,339 @@
|
||||
#!/usr/bin/env tsx
|
||||
/**
|
||||
* 财务销售/成本/总账一致性只读审计。
|
||||
*
|
||||
* 用法：
|
||||
* npm run audit:finance
|
||||
* npm run audit:finance -- --strict
|
||||
* npm run audit:finance -- --date-from=2026-11-01 --date-to=2026-11-30
|
||||
*
|
||||
* 唯一口径：
|
||||
* v_finance_order_item_scope 是销售范围的唯一来源；
|
||||
* sales_daily_facts 必须与其收入、退款守恒；
|
||||
* finance_journal_entries(ref_type=sales_fact) 必须与事实守恒。
|
||||
*/
|
||||
import process from 'node:process'
|
||||
|
||||
import { closeDb, query } from '../src/db/client.js'
|
||||
|
||||
const strict = process.argv.includes('--strict')
|
||||
const options = parseOptions(process.argv.slice(2))
|
||||
|
||||
type Totals = {
|
||||
revenueFen: number
|
||||
refundFen: number
|
||||
costFen: number
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const viewReady = await query<{ ready: boolean }>(
|
||||
`SELECT to_regclass('public.v_finance_order_item_scope') IS NOT NULL AS ready`,
|
||||
)
|
||||
if (!viewReady.rows[0]?.ready) {
|
||||
throw new Error('缺少 v_finance_order_item_scope；请先执行 npm run db:migrate')
|
||||
}
|
||||
|
||||
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 || today()
|
||||
const dateTo = normalizeDate(options.dateTo) || bounds.rows[0]?.date_to || dateFrom
|
||||
if (dateFrom > dateTo) throw new Error('date-from 不能晚于 date-to')
|
||||
|
||||
const scope = await loadScopeTotals(dateFrom, dateTo)
|
||||
const facts = await loadFactTotals(dateFrom, dateTo)
|
||||
const ledger = await loadLedgerTotals(dateFrom, dateTo)
|
||||
const snapshots = await loadSnapshotTotals(dateFrom, dateTo)
|
||||
const quality = await loadQualityChecks(dateFrom, dateTo)
|
||||
|
||||
const critical: string[] = []
|
||||
compare('销售事实收入', scope.revenueFen, facts.revenueFen, critical)
|
||||
compare('销售事实退款', scope.refundFen, facts.refundFen, critical)
|
||||
compare('销售事实成本', snapshots.costFen, facts.costFen, critical)
|
||||
compare('总账收入', facts.revenueFen, ledger.revenueFen, critical)
|
||||
compare('总账退款', facts.refundFen, ledger.refundFen, critical)
|
||||
compare('总账成本', facts.costFen, ledger.costFen, critical)
|
||||
if (quality.duplicateActiveSnapshots > 0) critical.push('存在重复有效成本快照')
|
||||
if (quality.negativeFactRows > 0) critical.push('销售事实存在负数金额')
|
||||
if (quality.unresolvedReconciliations > 0) critical.push('存在未解决平台对账差异')
|
||||
|
||||
const result = {
|
||||
dateFrom,
|
||||
dateTo,
|
||||
scope,
|
||||
facts,
|
||||
ledger,
|
||||
snapshots,
|
||||
quality,
|
||||
critical,
|
||||
}
|
||||
console.log(JSON.stringify(result, null, 2))
|
||||
|
||||
if (strict && critical.length > 0) {
|
||||
process.exitCode = 1
|
||||
}
|
||||
}
|
||||
|
||||
async function loadScopeTotals(
|
||||
dateFrom: string,
|
||||
dateTo: string,
|
||||
): Promise<Totals & { items: number }> {
|
||||
const result = await query<{
|
||||
items: number
|
||||
revenue_fen: string
|
||||
refund_fen: string
|
||||
}>(
|
||||
`SELECT
|
||||
COUNT(*)::int AS items,
|
||||
COALESCE(SUM(revenue_fen), 0)::bigint AS revenue_fen,
|
||||
COALESCE(SUM(refund_fen), 0)::bigint AS refund_fen
|
||||
FROM v_finance_order_item_scope
|
||||
WHERE stat_date BETWEEN $1::date AND $2::date`,
|
||||
[dateFrom, dateTo],
|
||||
)
|
||||
const row = result.rows[0]
|
||||
return {
|
||||
items: Number(row?.items || 0),
|
||||
revenueFen: Number(row?.revenue_fen || 0),
|
||||
refundFen: Number(row?.refund_fen || 0),
|
||||
costFen: 0,
|
||||
}
|
||||
}
|
||||
|
||||
async function loadFactTotals(
|
||||
dateFrom: string,
|
||||
dateTo: string,
|
||||
): Promise<Totals & { rows: number }> {
|
||||
const result = await query<{
|
||||
rows: number
|
||||
revenue_fen: string
|
||||
refund_fen: string
|
||||
cost_fen: string
|
||||
}>(
|
||||
`SELECT
|
||||
COUNT(*)::int AS rows,
|
||||
COALESCE(SUM(gross_revenue_fen), 0)::bigint AS revenue_fen,
|
||||
COALESCE(SUM(refund_amount_fen), 0)::bigint AS refund_fen,
|
||||
COALESCE(SUM(cost_amount_fen), 0)::bigint AS cost_fen
|
||||
FROM sales_daily_facts
|
||||
WHERE stat_date BETWEEN $1::date AND $2::date`,
|
||||
[dateFrom, dateTo],
|
||||
)
|
||||
const row = result.rows[0]
|
||||
return {
|
||||
rows: Number(row?.rows || 0),
|
||||
revenueFen: Number(row?.revenue_fen || 0),
|
||||
refundFen: Number(row?.refund_fen || 0),
|
||||
costFen: Number(row?.cost_fen || 0),
|
||||
}
|
||||
}
|
||||
|
||||
async function loadLedgerTotals(
|
||||
dateFrom: string,
|
||||
dateTo: string,
|
||||
): Promise<Totals & { entries: number }> {
|
||||
const result = await query<{
|
||||
entries: number
|
||||
revenue_fen: string
|
||||
refund_fen: string
|
||||
cost_fen: string
|
||||
}>(
|
||||
`WITH entries AS (
|
||||
SELECT id
|
||||
FROM finance_journal_entries
|
||||
WHERE ref_type = 'sales_fact'
|
||||
AND status = 'posted'
|
||||
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')
|
||||
)
|
||||
SELECT
|
||||
(SELECT COUNT(*)::int FROM entries) AS entries,
|
||||
COALESCE(SUM(line.debit_fen - line.credit_fen) FILTER (
|
||||
WHERE line.account_code = '1101'
|
||||
AND entry.event_type IN ('sales_revenue', 'sales_revenue_adjustment')
|
||||
), 0)::bigint AS revenue_fen,
|
||||
COALESCE(SUM(line.debit_fen - line.credit_fen) FILTER (
|
||||
WHERE line.account_code = '4101'
|
||||
AND entry.event_type IN ('sales_refund', 'sales_refund_adjustment')
|
||||
), 0)::bigint AS refund_fen,
|
||||
COALESCE(SUM(line.debit_fen - line.credit_fen) FILTER (
|
||||
WHERE line.account_code = '5001'
|
||||
AND entry.event_type IN ('sales_cost', 'sales_cost_adjustment')
|
||||
), 0)::bigint AS cost_fen
|
||||
FROM entries
|
||||
JOIN finance_journal_entries entry ON entry.id = entries.id
|
||||
JOIN finance_journal_lines line ON line.entry_id = entry.id`,
|
||||
[dateFrom, dateTo],
|
||||
)
|
||||
const row = result.rows[0]
|
||||
return {
|
||||
entries: Number(row?.entries || 0),
|
||||
revenueFen: Number(row?.revenue_fen || 0),
|
||||
refundFen: Number(row?.refund_fen || 0),
|
||||
costFen: Number(row?.cost_fen || 0),
|
||||
}
|
||||
}
|
||||
|
||||
async function loadSnapshotTotals(
|
||||
dateFrom: string,
|
||||
dateTo: string,
|
||||
): Promise<Totals & { estimated: number; actual: number }> {
|
||||
const result = await query<{
|
||||
estimated: number
|
||||
actual: number
|
||||
cost_fen: string
|
||||
}>(
|
||||
`SELECT
|
||||
COUNT(*) FILTER (WHERE snapshot.snapshot_stage = 'estimated')::int AS estimated,
|
||||
COUNT(*) FILTER (WHERE snapshot.snapshot_stage = 'actual')::int AS actual,
|
||||
COALESCE(SUM(snapshot.total_amount_fen), 0)::bigint AS cost_fen
|
||||
FROM cost_snapshots snapshot
|
||||
JOIN v_finance_order_item_scope scope ON scope.order_item_id = snapshot.order_item_id
|
||||
WHERE snapshot.superseded_at IS NULL
|
||||
AND snapshot.reversed_at IS NULL
|
||||
AND scope.stat_date BETWEEN $1::date AND $2::date`,
|
||||
[dateFrom, dateTo],
|
||||
)
|
||||
const row = result.rows[0]
|
||||
return {
|
||||
estimated: Number(row?.estimated || 0),
|
||||
actual: Number(row?.actual || 0),
|
||||
revenueFen: 0,
|
||||
refundFen: 0,
|
||||
costFen: Number(row?.cost_fen || 0),
|
||||
}
|
||||
}
|
||||
|
||||
async function loadQualityChecks(dateFrom: string, dateTo: string) {
|
||||
const result = await query<{
|
||||
pending_normalization: number
|
||||
duplicate_active_snapshots: number
|
||||
open_cost_exceptions: number
|
||||
open_manual_cost_exceptions: number
|
||||
open_on_account_cost_exceptions: number
|
||||
negative_fact_rows: number
|
||||
unresolved_reconciliations: number
|
||||
closed_periods: number
|
||||
}>(
|
||||
`SELECT
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM v_finance_order_item_scope
|
||||
WHERE stat_date BETWEEN $1::date AND $2::date
|
||||
AND channel <> 'unclassified'
|
||||
AND canonical_product_id IS NULL
|
||||
) AS pending_normalization,
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM (
|
||||
SELECT subject_type, subject_id, cost_item_id
|
||||
FROM cost_snapshots
|
||||
WHERE superseded_at IS NULL AND reversed_at IS NULL
|
||||
GROUP BY subject_type, subject_id, cost_item_id
|
||||
HAVING COUNT(*) > 1
|
||||
) duplicate
|
||||
) AS duplicate_active_snapshots,
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM cost_exceptions
|
||||
WHERE reason = 'cost_missing' AND status <> 'resolved'
|
||||
) AS open_cost_exceptions,
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM cost_exceptions
|
||||
WHERE reason = 'manual_cost_missing' AND status <> 'resolved'
|
||||
) AS open_manual_cost_exceptions,
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM cost_exceptions
|
||||
WHERE reason = 'on_account_cost_pending' AND status <> 'resolved'
|
||||
) AS open_on_account_cost_exceptions,
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM sales_daily_facts
|
||||
WHERE stat_date BETWEEN $1::date AND $2::date
|
||||
AND (
|
||||
gross_revenue_fen < 0
|
||||
OR refund_amount_fen < 0
|
||||
OR cost_amount_fen < 0
|
||||
)
|
||||
) AS negative_fact_rows,
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM finance_reconciliations
|
||||
WHERE status = 'differences'
|
||||
AND period_from <= $2::date
|
||||
AND period_to >= $1::date
|
||||
) AS unresolved_reconciliations,
|
||||
(
|
||||
SELECT COUNT(*)::int
|
||||
FROM finance_periods
|
||||
WHERE status = 'closed'
|
||||
AND period_key BETWEEN to_char($1::date, 'YYYY-MM') AND to_char($2::date, 'YYYY-MM')
|
||||
) AS closed_periods`,
|
||||
[dateFrom, dateTo],
|
||||
)
|
||||
const row = result.rows[0]
|
||||
return {
|
||||
pendingNormalization: Number(row?.pending_normalization || 0),
|
||||
duplicateActiveSnapshots: Number(row?.duplicate_active_snapshots || 0),
|
||||
openCostExceptions: Number(row?.open_cost_exceptions || 0),
|
||||
openManualCostExceptions: Number(row?.open_manual_cost_exceptions || 0),
|
||||
openOnAccountCostExceptions: Number(row?.open_on_account_cost_exceptions || 0),
|
||||
negativeFactRows: Number(row?.negative_fact_rows || 0),
|
||||
unresolvedReconciliations: Number(row?.unresolved_reconciliations || 0),
|
||||
closedPeriods: Number(row?.closed_periods || 0),
|
||||
}
|
||||
}
|
||||
|
||||
function compare(label: string, expected: number, actual: number, critical: string[]) {
|
||||
if (expected !== actual) {
|
||||
critical.push(`${label}不一致：应为 ${expected}，实际 ${actual}`)
|
||||
}
|
||||
}
|
||||
|
||||
function parseOptions(args: string[]) {
|
||||
const result: Record<string, string> = {}
|
||||
for (const arg of args) {
|
||||
if (!arg.startsWith('--')) continue
|
||||
const body = arg.slice(2)
|
||||
const index = body.indexOf('=')
|
||||
const key = index === -1 ? body : body.slice(0, index)
|
||||
result[key.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())] =
|
||||
index === -1 ? 'true' : body.slice(index + 1)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
function normalizeDate(value: unknown): string | null {
|
||||
const text = String(value || '').trim()
|
||||
return /^\d{4}-\d{2}-\d{2}$/.test(text) ? text : null
|
||||
}
|
||||
|
||||
function today() {
|
||||
return new Intl.DateTimeFormat('en-CA', {
|
||||
timeZone: 'Asia/Shanghai',
|
||||
year: 'numeric',
|
||||
month: '2-digit',
|
||||
day: '2-digit',
|
||||
}).format(new Date())
|
||||
}
|
||||
|
||||
main()
|
||||
.catch((error) => {
|
||||
console.error(error instanceof Error ? error.message : error)
|
||||
process.exitCode = 1
|
||||
})
|
||||
.finally(() => closeDb())
|
||||
@@ -1,210 +0,0 @@
|
||||
/**
|
||||
* 履约载体 ↔ 财务渠道 对齐审计（只读）。
|
||||
*
|
||||
* 用法：npx tsx scripts/audit-fulfillment-finance-alignment.ts
|
||||
*/
|
||||
import { closeDb, query } from '../src/db/client.js'
|
||||
|
||||
async function main() {
|
||||
const pending = await query<{
|
||||
total: number
|
||||
with_work_order: number
|
||||
with_affiliate_task: number
|
||||
with_manual_task: number
|
||||
no_carrier: number
|
||||
}>(
|
||||
`
|
||||
SELECT
|
||||
count(*)::int AS total,
|
||||
count(*) FILTER (WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int AS with_work_order,
|
||||
count(*) FILTER (WHERE 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')))::int AS with_affiliate_task,
|
||||
count(*) FILTER (WHERE EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS with_manual_task,
|
||||
count(*) FILTER (
|
||||
WHERE 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 fulfillment_tasks ft WHERE ft.order_item_id = oi.id)
|
||||
)::int AS no_carrier
|
||||
FROM order_items oi
|
||||
WHERE oi.fulfillment_status = 'pending'
|
||||
`,
|
||||
)
|
||||
|
||||
const channelCarrier = await query<{
|
||||
channel: string
|
||||
total: number
|
||||
work_orders: number
|
||||
affiliate_tasks: number
|
||||
manual_tasks: number
|
||||
config_missing: number
|
||||
no_carrier: number
|
||||
}>(
|
||||
`
|
||||
SELECT
|
||||
COALESCE(NULLIF(oi.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 AND wo.status <> 'cancelled'))::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' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::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' AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')))::int AS manual_tasks,
|
||||
count(*) FILTER (
|
||||
WHERE oi.fulfillment_channel = 'manual' AND oi.fulfillment_config_id IS NULL
|
||||
)::int AS config_missing,
|
||||
count(*) FILTER (
|
||||
WHERE (
|
||||
oi.fulfillment_channel = 'affiliate_dash'
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM fulfillment_tasks ft
|
||||
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
|
||||
AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
|
||||
)
|
||||
) OR (
|
||||
oi.fulfillment_channel = 'work_order'
|
||||
AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled')
|
||||
)
|
||||
)::int AS no_carrier
|
||||
FROM order_items oi
|
||||
WHERE oi.fulfillment_status = 'assigned'
|
||||
GROUP BY 1
|
||||
ORDER BY 2 DESC
|
||||
`,
|
||||
)
|
||||
|
||||
const internalSku = await query<{
|
||||
total: number
|
||||
null_or_blank: number
|
||||
numeric_only: number
|
||||
with_cjk: number
|
||||
}>(
|
||||
`
|
||||
SELECT
|
||||
count(*)::int AS total,
|
||||
count(*) FILTER (WHERE btrim(COALESCE(resolve_internal_sku(
|
||||
manual_sku_key, sku_name, sku_code
|
||||
), '')) = '')::int AS null_or_blank,
|
||||
count(*) FILTER (WHERE btrim(COALESCE(resolve_internal_sku(
|
||||
manual_sku_key, sku_name, sku_code
|
||||
), '')) ~ '^[0-9]+$')::int AS numeric_only,
|
||||
count(*) FILTER (WHERE btrim(COALESCE(resolve_internal_sku(
|
||||
manual_sku_key, sku_name, sku_code
|
||||
), '')) ~ '[㐀-䶿一-鿿豈-𠀀-𰀀-𲎯]')::int AS with_cjk
|
||||
FROM order_items
|
||||
WHERE fulfillment_status = 'assigned'
|
||||
`,
|
||||
)
|
||||
|
||||
const duplicateProducts = await query<{ duplicate_names: number; duplicate_rows: number }>(
|
||||
`
|
||||
SELECT count(*)::int AS duplicate_names, COALESCE(sum(total), 0)::int AS duplicate_rows
|
||||
FROM (
|
||||
SELECT canonical_name, count(*) AS total
|
||||
FROM product_catalog
|
||||
WHERE status = 'active'
|
||||
GROUP BY canonical_name
|
||||
HAVING count(*) > 1
|
||||
) grouped
|
||||
`,
|
||||
)
|
||||
|
||||
const routes = await query<{
|
||||
channel: string
|
||||
route_role: string
|
||||
source: string
|
||||
total: number
|
||||
}>(
|
||||
`SELECT channel, route_role, source, count(*)::int AS total
|
||||
FROM sku_channel_routes
|
||||
WHERE enabled = TRUE
|
||||
GROUP BY channel, route_role, source
|
||||
ORDER BY channel, route_role, source`,
|
||||
)
|
||||
|
||||
const internalRouteConflicts = await query<{
|
||||
internal_sku_id: number
|
||||
total: number
|
||||
}>(
|
||||
`SELECT internal_sku_id, count(*)::int AS total
|
||||
FROM sku_channel_routes
|
||||
WHERE enabled = TRUE AND route_role = 'primary'
|
||||
GROUP BY internal_sku_id
|
||||
HAVING count(*) > 1
|
||||
ORDER BY count(*) DESC, internal_sku_id`,
|
||||
)
|
||||
|
||||
const assignedCarrierConflicts = await query<{
|
||||
order_item_id: number
|
||||
channel: string
|
||||
carrier_count: number
|
||||
}>(
|
||||
`SELECT oi.id AS order_item_id, oi.fulfillment_channel AS channel,
|
||||
(EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int
|
||||
+ (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')
|
||||
))::int
|
||||
+ (EXISTS (
|
||||
SELECT 1 FROM fulfillment_tasks ft
|
||||
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'
|
||||
AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
|
||||
))::int AS carrier_count
|
||||
FROM order_items oi
|
||||
WHERE oi.fulfillment_status = 'assigned'
|
||||
AND (EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'))::int
|
||||
+ (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')
|
||||
))::int
|
||||
+ (EXISTS (
|
||||
SELECT 1 FROM fulfillment_tasks ft
|
||||
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'
|
||||
AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
|
||||
))::int > 1
|
||||
ORDER BY oi.id
|
||||
LIMIT 200`,
|
||||
)
|
||||
|
||||
const byProvider = await query<{
|
||||
provider: string
|
||||
total: number
|
||||
pending: number
|
||||
with_carrier: number
|
||||
}>(
|
||||
`
|
||||
SELECT o.provider,
|
||||
count(*)::int AS total,
|
||||
count(*) FILTER (WHERE oi.fulfillment_status = 'pending')::int AS pending,
|
||||
count(*) FILTER (
|
||||
WHERE EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled')
|
||||
OR EXISTS (SELECT 1 FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id)
|
||||
)::int AS with_carrier
|
||||
FROM order_items oi
|
||||
JOIN orders o ON o.id = oi.order_id
|
||||
WHERE COALESCE(o.paid_at, o.created_at) >= '2026-10-31 00:00:00+08'
|
||||
GROUP BY 1
|
||||
ORDER BY 2 DESC
|
||||
`,
|
||||
)
|
||||
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
pending: pending.rows[0],
|
||||
channelCarrier: channelCarrier.rows,
|
||||
internalSku: internalSku.rows[0],
|
||||
duplicateProducts: duplicateProducts.rows[0],
|
||||
routes: routes.rows,
|
||||
internalRouteConflicts: internalRouteConflicts.rows,
|
||||
assignedCarrierConflicts: assignedCarrierConflicts.rows,
|
||||
byProvider: byProvider.rows,
|
||||
},
|
||||
null,
|
||||
2,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
main()
|
||||
.catch((error) => {
|
||||
console.error(error)
|
||||
process.exitCode = 1
|
||||
})
|
||||
.finally(() => closeDb())
|
||||
@@ -1,259 +0,0 @@
|
||||
import process from 'node:process'
|
||||
|
||||
import { closeDb, query } from '../src/db/client.js'
|
||||
|
||||
/**
|
||||
* 财务断代 v2：清理断代前的派生数据，为 2026-10-31 起点重建让路。
|
||||
*
|
||||
* 只删「派生数据」，不删业务原始票据：
|
||||
* 删除：cost_snapshots / cost_exceptions(+items) / sales_daily_facts /
|
||||
* finance_journal_lines(+entries) / finance_reconciliations(+items) /
|
||||
* finance_refund_intents / order_refunds / manual_expenses /
|
||||
* fund_entries / platform_credit_entries / manual_sku_registry
|
||||
* 保留：orders / order_items / work_orders / fulfillment_tasks（审计只读）
|
||||
* product_catalog / channel_product_mappings / cost_items /
|
||||
* cost_versions / work_product_rules / manual_fulfillment_configs /
|
||||
* sku_channel_routes（配置）
|
||||
*
|
||||
* 前置：
|
||||
* 1. 迁移 112/113 已执行；
|
||||
* 2. 已做数据库备份。
|
||||
*
|
||||
* 断代本身包含两步：先把断代前的订单项批量置为 archived，再清理派生数据。
|
||||
*
|
||||
* 默认 dry-run，只打印将删除的行数；加 `--execute --confirm=2026-10-31` 才真正执行。
|
||||
* 不可重复执行：成功后会写入 finance_cutovers，第二次会被拦下。
|
||||
*/
|
||||
|
||||
const CUTOVER_DATE = '2026-10-31'
|
||||
|
||||
const options = parseOptions(process.argv.slice(2))
|
||||
const execute = options.execute === 'true'
|
||||
const confirm = String(options.confirm || '').trim()
|
||||
|
||||
/** 派生表：按外键顺序删除。 */
|
||||
const DERIVED_TABLES: Array<{ table: string; order: number; where?: string }> = [
|
||||
{ table: 'cost_exception_items', order: 10 },
|
||||
{ table: 'cost_exceptions', order: 20 },
|
||||
{ table: 'cost_snapshots', order: 30 },
|
||||
{ table: 'sales_daily_facts', order: 40 },
|
||||
{ table: 'finance_journal_lines', order: 50 },
|
||||
{ table: 'finance_journal_entries', order: 60 },
|
||||
{ table: 'finance_reconciliation_items', order: 70 },
|
||||
{ table: 'finance_reconciliations', order: 80 },
|
||||
{ table: 'finance_refund_intents', order: 90 },
|
||||
{ table: 'order_refunds', order: 100 },
|
||||
{ table: 'manual_expenses', order: 110 },
|
||||
{ table: 'fund_entries', order: 120 },
|
||||
{ table: 'platform_credit_entries', order: 130 },
|
||||
{ table: 'manual_sku_registry', order: 140 },
|
||||
]
|
||||
|
||||
async function main() {
|
||||
await assertPreconditions()
|
||||
|
||||
const startedAt = Date.now()
|
||||
const plan = await buildPlan()
|
||||
|
||||
if (!execute) {
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
mode: 'dry-run',
|
||||
cutoverDate: CUTOVER_DATE,
|
||||
deletePlan: plan,
|
||||
hint: '确认无误后追加 --execute --confirm=2026-10-31',
|
||||
durationMs: Date.now() - startedAt,
|
||||
},
|
||||
null,
|
||||
2,
|
||||
),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
if (confirm !== CUTOVER_DATE) {
|
||||
throw new Error(`--confirm 必须是 ${CUTOVER_DATE}，当前为 "${confirm}"`)
|
||||
}
|
||||
|
||||
await assertNotExecuted()
|
||||
|
||||
const archived = await archiveHistoricalItems()
|
||||
|
||||
const deleted: Record<string, number> = {}
|
||||
for (const item of [...DERIVED_TABLES].sort((a, b) => a.order - b.order)) {
|
||||
const result = await query(
|
||||
`DELETE FROM ${item.table}${item.where ? ` WHERE ${item.where}` : ''}`,
|
||||
)
|
||||
deleted[item.table] = result.rowCount || 0
|
||||
process.stderr.write(`删除 ${item.table}: ${result.rowCount || 0}\n`)
|
||||
}
|
||||
|
||||
// 旧归档表不再保留：生产库不留可查询的历史异常副本。
|
||||
await query(`DROP TABLE IF EXISTS cost_exceptions_archive`)
|
||||
|
||||
// 水位归零：管道从断代日起重新扫描。
|
||||
await query(
|
||||
`
|
||||
UPDATE finance_watermarks
|
||||
SET last_item_id = 0, last_stat_date = $1::date, last_success_at = NULL,
|
||||
last_error = '', updated_at = NOW()
|
||||
`,
|
||||
[CUTOVER_DATE],
|
||||
)
|
||||
|
||||
await query(
|
||||
`
|
||||
INSERT INTO finance_cutovers (cutover_date, executed_at, detail_json)
|
||||
VALUES ($1::date, NOW(), $2::jsonb)
|
||||
ON CONFLICT (cutover_date) DO NOTHING
|
||||
`,
|
||||
[CUTOVER_DATE, JSON.stringify({ deleted, tables: DERIVED_TABLES.map((t) => t.table) })],
|
||||
)
|
||||
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
mode: 'execute',
|
||||
cutoverDate: CUTOVER_DATE,
|
||||
archivedItems: archived.updated,
|
||||
deleted,
|
||||
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 buildPlan() {
|
||||
const plan: Array<{ table: string; rows: number }> = []
|
||||
for (const item of [...DERIVED_TABLES].sort((a, b) => a.order - b.order)) {
|
||||
const result = await query<{ count: number }>(
|
||||
`SELECT COUNT(*)::int AS count FROM ${item.table}${item.where ? ` WHERE ${item.where}` : ''}`,
|
||||
)
|
||||
plan.push({ table: item.table, rows: result.rows[0]?.count || 0 })
|
||||
}
|
||||
return plan
|
||||
}
|
||||
|
||||
async function assertPreconditions() {
|
||||
const columns = await query<{ column_name: string }>(
|
||||
`
|
||||
SELECT column_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_schema = 'public'
|
||||
AND table_name = 'order_items'
|
||||
AND column_name IN ('fulfillment_channel', 'fulfillment_status', 'fulfillment_source')
|
||||
`,
|
||||
)
|
||||
if (columns.rows.length < 3) {
|
||||
throw new Error('缺少 112 的三列；请先执行迁移 112')
|
||||
}
|
||||
|
||||
const watermark = await query<{ exists: boolean }>(
|
||||
`SELECT to_regclass('public.finance_watermarks') IS NOT NULL AS exists`,
|
||||
)
|
||||
if (!watermark.rows[0]?.exists) {
|
||||
throw new Error('finance_watermarks 不存在；请先执行迁移 112')
|
||||
}
|
||||
|
||||
await query(
|
||||
`
|
||||
CREATE TABLE IF NOT EXISTS finance_cutovers (
|
||||
cutover_date DATE PRIMARY KEY,
|
||||
executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
detail_json JSONB NOT NULL DEFAULT '{}'::jsonb
|
||||
)
|
||||
`,
|
||||
)
|
||||
|
||||
const inconsistent = await query<{ count: number }>(
|
||||
`
|
||||
SELECT COUNT(*)::int AS count
|
||||
FROM order_items
|
||||
WHERE fulfillment_status = 'assigned' AND fulfillment_channel IS NULL
|
||||
`,
|
||||
)
|
||||
const bad = inconsistent.rows[0]?.count || 0
|
||||
if (bad > 0) {
|
||||
throw new Error(
|
||||
`有 ${bad} 行 status=assigned 但 channel 为空；请先执行 fulfillment:backfill-channel`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 断代前的订单项统一置为 archived（分批，避免 42 万行整表更新撞超时）。
|
||||
*/
|
||||
async function archiveHistoricalItems(): Promise<{ updated: number }> {
|
||||
let cursor = 0
|
||||
let updated = 0
|
||||
for (let batch = 0; batch < 500; batch += 1) {
|
||||
const rows = await query<{ id: string }>(
|
||||
`
|
||||
SELECT oi.id::text
|
||||
FROM order_items oi
|
||||
JOIN orders o ON o.id = oi.order_id
|
||||
WHERE (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date < $1::date
|
||||
AND oi.fulfillment_status <> 'archived'
|
||||
AND oi.id > $2
|
||||
ORDER BY oi.id
|
||||
LIMIT 20000
|
||||
`,
|
||||
[CUTOVER_DATE, cursor],
|
||||
)
|
||||
if (rows.rows.length === 0) break
|
||||
cursor = Number(rows.rows[rows.rows.length - 1]!.id)
|
||||
|
||||
const result = await query(
|
||||
`
|
||||
UPDATE order_items
|
||||
SET fulfillment_status = 'archived',
|
||||
fulfillment_channel = NULL,
|
||||
fulfillment_source = 'cutover_archive',
|
||||
fulfillment_decided_at = NOW(),
|
||||
updated_at = NOW()
|
||||
WHERE id = ANY($1::bigint[])
|
||||
AND fulfillment_status <> 'archived'
|
||||
`,
|
||||
[rows.rows.map((row) => Number(row.id))],
|
||||
)
|
||||
updated += result.rowCount || 0
|
||||
process.stderr.write(`归档 ${result.rowCount || 0} 行（游标 ${cursor}）\n`)
|
||||
}
|
||||
return { updated }
|
||||
}
|
||||
|
||||
async function assertNotExecuted() {
|
||||
const done = await query<{ count: number }>(
|
||||
`SELECT COUNT(*)::int AS count FROM finance_cutovers WHERE cutover_date = $1::date`,
|
||||
[CUTOVER_DATE],
|
||||
)
|
||||
if ((done.rows[0]?.count || 0) > 0) {
|
||||
throw new Error(`${CUTOVER_DATE} 的断代已执行过，禁止重复执行`)
|
||||
}
|
||||
}
|
||||
|
||||
function parseOptions(args: string[]) {
|
||||
const result: Record<string, string> = {}
|
||||
for (const arg of args) {
|
||||
if (!arg.startsWith('--')) continue
|
||||
const body = arg.slice(2)
|
||||
const index = body.indexOf('=')
|
||||
const rawKey = index === -1 ? body : body.slice(0, index)
|
||||
const value = index === -1 ? 'true' : body.slice(index + 1)
|
||||
result[rawKey] = value
|
||||
result[rawKey.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())] = value
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -44,12 +44,12 @@ async function main() {
|
||||
}>(
|
||||
`
|
||||
SELECT
|
||||
COUNT(*) FILTER (WHERE fulfillment_status = 'assigned')::int AS assigned,
|
||||
COUNT(*) FILTER (WHERE fulfillment_status = 'pending')::int AS pending,
|
||||
COUNT(*) FILTER (WHERE fulfillment_status = 'archived')::int AS archived,
|
||||
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 order_items
|
||||
FROM fulfillments
|
||||
`,
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,408 @@
|
||||
#!/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<string>() }
|
||||
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<string[]> {
|
||||
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<number[]> {
|
||||
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<number> {
|
||||
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<number> {
|
||||
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<number> {
|
||||
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<ReturnType<typeof loadProtectedTotals>>,
|
||||
after: Awaited<ReturnType<typeof loadProtectedTotals>>,
|
||||
) {
|
||||
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<T>(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<string, string> = {}
|
||||
for (const arg of args) {
|
||||
if (!arg.startsWith('--')) continue
|
||||
const body = arg.slice(2)
|
||||
const index = body.indexOf('=')
|
||||
const rawKey = index === -1 ? body : body.slice(0, index)
|
||||
const value = index === -1 ? 'true' : body.slice(index + 1)
|
||||
result[rawKey] = value
|
||||
result[rawKey.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())] = value
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
function 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())
|
||||
Reference in New Issue
Block a user