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 PLINK_PATH = process.env.FIXED_ASSET_PLINK_PATH || 'C:\\Program Files\\PuTTY\\plink.exe' const SSH_HOST = process.env.FIXED_ASSET_SSH_HOST || 'root@1.13.22.133' const SSH_HOSTKEY = process.env.FIXED_ASSET_SSH_HOSTKEY || 'SHA256:huTZZAYXh4yd5AvE4jo7drXaF/WSFF0SkzYHoddbMLA' const SSH_PASSWORD = process.env.FIXED_ASSET_SSH_PASSWORD || '' 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 parseArgs = () => { const args = process.argv.slice(2) const options = { inputDir: '' } for (let index = 0; index < args.length; index += 1) { const arg = args[index] if (arg === '--input-dir') { options.inputDir = args[index + 1] || '' index += 1 } } return options } 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) => { if (!fs.existsSync(PLINK_PATH)) { throw new Error(`未找到 plink.exe: ${PLINK_PATH}`) } if (!SSH_PASSWORD) { throw new Error('缺少 FIXED_ASSET_SSH_PASSWORD 环境变量') } const result = spawnSync( PLINK_PATH, ['-batch', '-ssh', '-hostkey', SSH_HOSTKEY, '-pw', SSH_PASSWORD, SSH_HOST, REMOTE_NODE_COMMAND], { encoding: 'utf8', input: script, maxBuffer: 64 * 1024 * 1024, windowsHide: true } ) if (result.error) { throw result.error } if (result.status !== 0) { throw new Error(`远程脚本执行失败: ${result.stderr || result.stdout}`) } const stdout = normalizeText(result.stdout) if (!stdout) { throw new Error('远程脚本返回为空') } return JSON.parse(stdout) } const buildRemoteScript = (payload) => { 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')) ;(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 = [] 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({ 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 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)) writeJson(path.join(outputDir, 'result.json'), result) console.log( JSON.stringify( { outputDir, inputDir, payloadCount: payload.length, result }, null, 2 ) ) } try { run() } catch (error) { console.error(error && error.stack ? error.stack : String(error)) process.exit(1) }