let PrismaClient try { ;({ PrismaClient } = require('@prisma/client')) } catch { ;({ PrismaClient } = require('/app/node_modules/@prisma/client')) } const fs = require('fs') const path = require('path') const { spawnSync } = require('child_process') const { resolveStorageLocationName } = require('./locationNormalization') const OUTPUT_ROOT = path.resolve(__dirname, '../../temp_migration_output') const REMOTE_NODE_COMMAND = 'docker exec -i equipment-backend node' const normalizeText = (value) => String(value == null ? '' : value) .replace(/\r/g, ' ') .replace(/\n/g, ' ') .replace(/\s+/g, ' ') .trim() const writeJson = (filePath, value) => { fs.writeFileSync(filePath, `${JSON.stringify(value, null, 2)}\n`) } const readJsonl = (filePath) => fs .readFileSync(filePath, 'utf8') .split(/\r?\n/) .filter(Boolean) .map((line) => JSON.parse(line)) const ensureOutputDir = () => { const timestamp = new Date().toISOString().replace(/[:.]/g, '-').replace('T', '_').slice(0, 19) const outputDir = path.join(OUTPUT_ROOT, `location_backfill_${timestamp}`) fs.mkdirSync(outputDir, { recursive: true }) return outputDir } const resolveLatestMigrationDir = () => { const candidates = fs .readdirSync(OUTPUT_ROOT, { withFileTypes: true }) .filter((entry) => entry.isDirectory() && entry.name.startsWith('migration_dry_run_')) .map((entry) => entry.name) .sort() if (!candidates.length) { throw new Error(`未找到 dry-run 输出目录: ${OUTPUT_ROOT}`) } return path.join(OUTPUT_ROOT, candidates[candidates.length - 1]) } const requireArgValue = (args, index, argumentName) => { const value = args[index + 1] if (!value || String(value).startsWith('--')) { throw new Error(`${argumentName} 缺少参数值`) } return value } const parseArgs = (args = process.argv.slice(2)) => { const options = { inputDir: '', apply: false, dryRun: true, plinkPath: '', sshHost: '', sshHostkey: '', sshPassword: '' } let explicitDryRun = false for (let index = 0; index < args.length; index += 1) { const arg = args[index] if (arg === '--input-dir') { options.inputDir = requireArgValue(args, index, arg) index += 1 } else if (arg === '--dry-run') { explicitDryRun = true } else if (arg === '--apply') { options.apply = true } else if (arg === '--plink-path') { options.plinkPath = requireArgValue(args, index, arg) index += 1 } else if (arg === '--ssh-host') { options.sshHost = requireArgValue(args, index, arg) index += 1 } else if (arg === '--ssh-hostkey') { options.sshHostkey = requireArgValue(args, index, arg) index += 1 } else if (arg === '--ssh-password') { options.sshPassword = requireArgValue(args, index, arg) index += 1 } else { throw new Error(`不支持的参数: ${arg}`) } } if (options.apply && explicitDryRun) { throw new Error('--apply 与 --dry-run 不能同时使用') } options.dryRun = !options.apply return options } const resolveRemoteConnection = (options, env = process.env) => { const connection = { plinkPath: normalizeText(options.plinkPath || env.FIXED_ASSET_PLINK_PATH), sshHost: normalizeText(options.sshHost || env.FIXED_ASSET_SSH_HOST), sshHostkey: normalizeText(options.sshHostkey || env.FIXED_ASSET_SSH_HOSTKEY), sshPassword: options.sshPassword || env.FIXED_ASSET_SSH_PASSWORD || '' } const missing = [] if (!connection.plinkPath) missing.push('FIXED_ASSET_PLINK_PATH / --plink-path') if (!connection.sshHost) missing.push('FIXED_ASSET_SSH_HOST / --ssh-host') if (!connection.sshHostkey) missing.push('FIXED_ASSET_SSH_HOSTKEY / --ssh-hostkey') if (!connection.sshPassword) missing.push('FIXED_ASSET_SSH_PASSWORD / --ssh-password') if (missing.length > 0) { throw new Error(`缺少固定资产远程连接参数:${missing.join('、')}。未发起网络连接。`) } return connection } const resolveInputDir = (inputDirArg) => { if (!inputDirArg) { return resolveLatestMigrationDir() } const resolved = path.isAbsolute(inputDirArg) ? inputDirArg : path.resolve(process.cwd(), inputDirArg) if (!fs.existsSync(resolved)) { throw new Error(`指定的输入目录不存在: ${resolved}`) } return resolved } const buildPayload = (records) => { const importableRecords = records.filter( (record) => record.bucket === 'import-ready' || record.bucket === 'import-without-owner' ) const grouped = new Map() for (const record of importableRecords) { const legacyCode = normalizeText(record.assetCodeRaw).toUpperCase() if (!legacyCode) { continue } if (!grouped.has(legacyCode)) { grouped.set(legacyCode, []) } grouped.get(legacyCode).push(record) } const duplicateCodes = new Set( [...grouped.entries()].filter(([, items]) => items.length > 1).map(([legacyCode]) => legacyCode) ) const updates = [] for (const record of importableRecords) { const legacyCode = normalizeText(record.assetCodeRaw).toUpperCase() if (!legacyCode || duplicateCodes.has(legacyCode)) { continue } const resolvedLocationName = normalizeText( record.locationResolvedName || resolveStorageLocationName(record.addressRaw).normalizedValue ) if (!resolvedLocationName || !record.locationMatch) { continue } updates.push({ legacyCode, rawLocation: normalizeText(record.addressRaw), resolvedLocationName, sourceRow: record.sourceRow }) } return updates } const runRemoteNodeScript = (script, connection) => { const result = spawnSync( connection.plinkPath, [ '-batch', '-ssh', '-hostkey', connection.sshHostkey, '-pw', connection.sshPassword, connection.sshHost, REMOTE_NODE_COMMAND ], { encoding: 'utf8', input: script, maxBuffer: 64 * 1024 * 1024, windowsHide: true } ) if (result.error) { throw new Error('无法启动远程连接工具,请检查 FIXED_ASSET_PLINK_PATH / --plink-path。') } if (result.status !== 0) { throw new Error('远程脚本执行失败,请检查连接参数、远程服务状态和服务器日志。') } const stdout = normalizeText(result.stdout) if (!stdout) { throw new Error('远程脚本返回为空') } try { return JSON.parse(stdout) } catch { throw new Error('远程脚本返回格式无效') } } const buildRemoteScript = (payload, apply) => { const payloadBase64 = Buffer.from(JSON.stringify(payload), 'utf8').toString('base64') return ` let PrismaClient try { ({ PrismaClient } = require('@prisma/client')) } catch { ({ PrismaClient } = require('/app/node_modules/@prisma/client')) } const prisma = new PrismaClient() const payload = JSON.parse(Buffer.from('${payloadBase64}', 'base64').toString('utf8')) const shouldApply = ${apply ? 'true' : 'false'} ;(async () => { const locationNames = [...new Set(payload.map((item) => item.resolvedLocationName))] const [locations, equipmentsBefore] = await Promise.all([ prisma.storageLocation.findMany({ where: { status: 'active', name: { in: locationNames } }, select: { id: true, name: true, code: true } }), prisma.equipment.findMany({ where: { equipment_code: { in: payload.map((item) => item.legacyCode) } }, select: { id: true, equipment_code: true, equipment_name: true, storage_location_id: true, storageLocation: { select: { id: true, name: true } } } }) ]) const locationByName = new Map(locations.map((item) => [item.name, item])) const equipmentByCode = new Map(equipmentsBefore.map((item) => [item.equipment_code, item])) const skipped = [] const planned = [] for (const item of payload) { const equipment = equipmentByCode.get(item.legacyCode) if (!equipment) { skipped.push({ ...item, reason: 'equipment_not_found' }) continue } if (equipment.storage_location_id) { skipped.push({ ...item, reason: 'already_has_storage_location', current_storage_location: equipment.storageLocation?.name || '' }) continue } const location = locationByName.get(item.resolvedLocationName) if (!location) { skipped.push({ ...item, reason: 'storage_location_not_found' }) continue } planned.push({ equipmentId: equipment.id, legacyCode: item.legacyCode, equipmentName: equipment.equipment_name, rawLocation: item.rawLocation, resolvedLocationName: item.resolvedLocationName, storageLocationId: location.id }) } const updated = [] if (shouldApply) { for (const item of planned) { const result = await prisma.equipment.update({ where: { id: item.equipmentId }, data: { storage_location_id: item.storageLocationId }, select: { id: true, equipment_code: true, storage_location_id: true, storageLocation: { select: { id: true, name: true } } } }) updated.push(result) } } const verification = { nullStorageLocationCount: await prisma.equipment.count({ where: { storage_location_id: null } }), updatedCount: updated.length } console.log(JSON.stringify({ applied: shouldApply, payloadCount: payload.length, plannedCount: planned.length, updatedCount: updated.length, skippedCount: skipped.length, planned, skipped, verification })) await prisma.$disconnect() })().catch(async (error) => { console.error(error && error.stack ? error.stack : String(error)) try { await prisma.$disconnect() } catch {} process.exit(1) }) ` } const run = () => { const options = parseArgs() const connection = resolveRemoteConnection(options) const inputDir = resolveInputDir(options.inputDir) const normalizedPath = path.join(inputDir, 'normalized_assets.jsonl') if (!fs.existsSync(normalizedPath)) { throw new Error(`未找到 normalized_assets.jsonl: ${normalizedPath}`) } const normalizedRecords = readJsonl(normalizedPath) const payload = buildPayload(normalizedRecords) if (!payload.length) { throw new Error('没有可回填的地址记录') } const outputDir = ensureOutputDir() writeJson(path.join(outputDir, 'planned_payload.json'), { inputDir, payloadCount: payload.length, payload }) const result = runRemoteNodeScript(buildRemoteScript(payload, options.apply), connection) writeJson(path.join(outputDir, 'result.json'), result) console.log( JSON.stringify( { outputDir, inputDir, applied: options.apply, payloadCount: payload.length, result }, null, 2 ) ) } if (require.main === module) { try { run() } catch (error) { console.error(error && error.stack ? error.stack : String(error)) process.exit(1) } } module.exports = { parseArgs, resolveRemoteConnection }