import { loadEnvConfig } from '@next/env' loadEnvConfig(process.cwd()) // 以正式环境产品全量档案为准,对齐本地档案的「生产状态」和「创建时间」。 // 默认 dry-run;写库需要 --apply --confirm=reconcile-archive-from-prod。 import pg from 'pg' const CONFIRM_TOKEN = 'reconcile-archive-from-prod' function required(name: string) { const value = process.env[name]?.trim() if (!value) throw new Error(`${name} is required.`) return value } function normalizeTimestamp(value: string | null) { return (value ?? '').replace('T', ' ').slice(0, 19) } function flagValue(name: string) { const prefix = `--${name}=` return process.argv.slice(2).find((argument) => argument.startsWith(prefix))?.slice(prefix.length) ?? '' } async function main() { const apply = process.argv.includes('--apply') if (apply && flagValue('confirm') !== CONFIRM_TOKEN) { throw new Error(`写库需要 --apply --confirm=${CONFIRM_TOKEN}`) } const batchSize = Math.min(Math.max(Number(flagValue('batch-size') || '5000'), 500), 20_000) const prod = new pg.Client({ host: required('PROD_ARCHIVE_DB_HOST'), port: Number(process.env.PROD_ARCHIVE_DB_PORT?.trim() || '25432'), database: required('PROD_ARCHIVE_DB_NAME'), user: required('PROD_ARCHIVE_DB_USER'), password: required('PROD_ARCHIVE_DB_PASSWORD'), ssl: { rejectUnauthorized: false }, statement_timeout: 120_000, }) const local = new pg.Client({ connectionString: required('DATABASE_URL'), statement_timeout: 120_000, }) await prod.connect() await local.connect() const stats = { apply, scanned: 0, missingLocal: 0, statusChanged: 0, createdChanged: 0, updated: 0, personnelChanged: 0, samples: [] as Array<{ barcode: string; field: string; local: string; prod: string }>, } // 与产品全量档案保持一致的档案字段(人员/设计等,本地不再自行推导)。 const mirroredFields = ['opsDirector', 'opsDeptHead', 'opsHead', 'supplierName', 'purchaseOwner', 'designer'] as const try { let offset = 0 while (true) { const prodRows = (await prod.query( `select id, barcode, "productStatus", "businessCreatedAt"::text as created, "updatedAt"::text as updated, "opsDirector", "opsDeptHead", "opsHead", "supplierName", "purchaseOwner", "designer" from public."SupplyChainProductArchive" order by barcode offset $1 limit $2`, [offset, batchSize], )).rows if (prodRows.length === 0) break offset += prodRows.length const barcodes = prodRows.map((row) => row.barcode) const localRows = (await local.query( `select barcode, "productStatus", "businessCreatedAt"::text as created, "opsDirector", "opsDeptHead", "opsHead", "supplierName", "purchaseOwner", "designer" from "SupplyChainProductArchive" where barcode = any($1::text[])`, [barcodes], )).rows const localMap = new Map(localRows.map((row) => [row.barcode, row])) for (const prodRow of prodRows) { stats.scanned += 1 const localRow = localMap.get(prodRow.barcode) if (!localRow) { stats.missingLocal += 1 continue } const statusDiff = (localRow.productStatus ?? '') !== (prodRow.productStatus ?? '') const createdDiff = normalizeTimestamp(localRow.created) !== normalizeTimestamp(prodRow.created) && prodRow.created !== null const fieldDiffs = mirroredFields.filter((field) => (localRow[field] ?? '') !== (prodRow[field] ?? '')) if (!statusDiff && !createdDiff && fieldDiffs.length === 0) continue if (statusDiff) stats.statusChanged += 1 if (createdDiff) stats.createdChanged += 1 stats.personnelChanged += fieldDiffs.length if (stats.samples.length < 12) { if (statusDiff || createdDiff) { stats.samples.push({ barcode: prodRow.barcode, field: statusDiff ? 'productStatus' : 'businessCreatedAt', local: statusDiff ? String(localRow.productStatus ?? '') : String(localRow.created ?? ''), prod: statusDiff ? String(prodRow.productStatus ?? '') : String(prodRow.created ?? ''), }) } else { const field = fieldDiffs[0] stats.samples.push({ barcode: prodRow.barcode, field, local: String(localRow[field] ?? ''), prod: String(prodRow[field] ?? ''), }) } } if (!apply) continue const sourceUpdatedAt = new Date(String(prodRow.updated)).toISOString() const sourceFor = (column: string) => JSON.stringify({ sourceCode: 'PRODUCT_DEVELOPMENT', sourceSystem: '产品全量档案(正式库)', sourceSchema: 'public', sourceObject: 'SupplyChainProductArchive', sourceColumn: column, sourceRecordId: prodRow.id, sourceUpdatedAt, }) const assignments: string[] = [] const params: unknown[] = [prodRow.barcode] const sourceSets: Array<{ path: string; paramIndex: number }> = [] const pushValue = (column: string, value: unknown) => { params.push(value) assignments.push(`"${column}" = $${params.length}`) params.push(sourceFor(column)) sourceSets.push({ path: `'{${column}}'`, paramIndex: params.length }) } if (statusDiff) { pushValue('productStatus', prodRow.productStatus ?? '') if ((prodRow.productStatus ?? '') !== '销售') { assignments.push(`"sabcRating" = ''`) assignments.push(`"sabcUpdatedAt" = now()`) } } if (createdDiff) { params.push(prodRow.created) assignments.push(`"businessCreatedAt" = $${params.length}::timestamp`) params.push(sourceFor('businessCreatedAt')) sourceSets.push({ path: `'{businessCreatedAt}'`, paramIndex: params.length }) } for (const field of fieldDiffs) { pushValue(field, prodRow[field] ?? '') } if (sourceSets.length > 0) { let expression = '"fieldSourceMap"' for (const { path, paramIndex } of sourceSets) { expression = `jsonb_set(${expression}, ${path}, $${paramIndex}::jsonb, true)` } assignments.push(`"fieldSourceMap" = ${expression}`) } assignments.push(`"updatedAt" = now()`) await local.query( `update "SupplyChainProductArchive" set ${assignments.join(', ')} where barcode = $1`, params, ) stats.updated += 1 } } } finally { await prod.end() await local.end() } console.log(JSON.stringify({ ...stats, samples: undefined, sampleChanges: stats.samples }, null, 2)) if (!apply) console.log('dry-run:未写库。加 --apply --confirm=' + CONFIRM_TOKEN + ' 才会对齐。') } void main().catch((error) => { console.error(error instanceof Error ? error.message : String(error)) process.exitCode = 1 })