gongxiangp2/scripts/run-history-import-queue.ts

import path from 'path'
import { promises as fs } from 'fs'
import {
  importHistoryMaterials,
  resolveImportOperator,
  type HistoryImportRecord,
  type ImportHistoryMaterialsResultItem,
} from '@/lib/history-material-import'
import { prisma } from '@/lib/prisma'

type QueueCheckpoint = {
  input: string
  operator: string
  apply: boolean
  pauseMs: number
  updatedAt: string
  summary: {
    totalCandidates: number
    processed: number
    imported: number
    dryRun: number
    skipped: number
    failed: number
  }
  results: ImportHistoryMaterialsResultItem[]
}

function parseArgs() {
  const args = process.argv.slice(2)
  const getArg = (name: string) => {
    const match = args.find((arg) => arg.startsWith(`--${name}=`))
    return match ? match.slice(name.length + 3) : ''
  }

  const input = getArg('input')
  const operator = getArg('operator')
  const checkpoint = getArg('checkpoint')
  const maxItemsRaw = getArg('max-items')
  const pauseMsRaw = getArg('pause-ms')
  const apply = args.includes('--apply')
  const stopOnFailure = args.includes('--stop-on-failure')

  if (!input) {
    throw new Error('缺少 --input=扫描结果文件')
  }
  if (!operator) {
    throw new Error('缺少 --operator=用户名/邮箱/姓名')
  }

  return {
    input,
    operator,
    checkpoint:
      checkpoint || path.join(path.dirname(input), `${path.basename(input, path.extname(input))}.queue-progress.json`),
    maxItems: maxItemsRaw ? Number(maxItemsRaw) : 1,
    pauseMs: pauseMsRaw ? Number(pauseMsRaw) : 1500,
    apply,
    stopOnFailure,
  }
}

async function sleep(ms: number) {
  if (ms <= 0) {
    return
  }
  await new Promise((resolve) => setTimeout(resolve, ms))
}

async function readCheckpoint(checkpointPath: string): Promise<QueueCheckpoint | null> {
  try {
    const raw = await fs.readFile(checkpointPath, 'utf8')
    return JSON.parse(raw) as QueueCheckpoint
  } catch {
    return null
  }
}

async function writeCheckpoint(checkpointPath: string, payload: QueueCheckpoint) {
  await fs.mkdir(path.dirname(checkpointPath), { recursive: true })
  await fs.writeFile(checkpointPath, JSON.stringify(payload, null, 2), 'utf8')
}

function buildCheckpointSkeleton(input: string, operator: string, apply: boolean, pauseMs: number, totalCandidates: number) {
  return {
    input,
    operator,
    apply,
    pauseMs,
    updatedAt: new Date().toISOString(),
    summary: {
      totalCandidates,
      processed: 0,
      imported: 0,
      dryRun: 0,
      skipped: 0,
      failed: 0,
    },
    results: [],
  } satisfies QueueCheckpoint
}

async function main() {
  const options = parseArgs()
  const raw = await fs.readFile(options.input, 'utf8')
  const allRecords = JSON.parse(raw) as HistoryImportRecord[]
  const candidates = allRecords.filter((record) => {
    if (options.apply) {
      return record.status === 'ready'
    }
    return record.status !== 'skipped'
  })
  const operator = await resolveImportOperator(options.operator)

  const existingCheckpoint = await readCheckpoint(options.checkpoint)
  const checkpoint =
    existingCheckpoint ||
    buildCheckpointSkeleton(options.input, operator.realName || operator.username, options.apply, options.pauseMs, candidates.length)

  checkpoint.input = options.input
  checkpoint.operator = operator.realName || operator.username
  checkpoint.apply = options.apply
  checkpoint.pauseMs = options.pauseMs
  checkpoint.summary.totalCandidates = candidates.length

  const processedFingerprints = new Set(checkpoint.results.map((item) => item.fingerprint))
  const pendingRecords = candidates.filter((record) => !processedFingerprints.has(record.fingerprint))
  const batch = pendingRecords.slice(0, Math.max(1, options.maxItems))

  console.log('历史素材串行队列开始')
  console.log(`- 输入文件: ${options.input}`)
  console.log(`- 执行人: ${operator.realName || operator.username}`)
  console.log(`- 模式: ${options.apply ? '正式导入' : '仅演练'}`)
  console.log(`- 单次最多处理: ${options.maxItems}`)
  console.log(`- 条目间隔: ${options.pauseMs}ms`)
  console.log(`- 已处理: ${checkpoint.summary.processed}/${checkpoint.summary.totalCandidates}`)
  console.log(`- 本轮待处理: ${batch.length}`)
  console.log(`- 进度文件: ${options.checkpoint}`)

  for (let index = 0; index < batch.length; index += 1) {
    const record = batch[index]
    console.log(`\n[${index + 1}/${batch.length}] ${record.spu || '<无SPU>'} | ${record.guessedStore} | ${record.sourcePath}`)

    const [result] = await importHistoryMaterials({
      records: [record],
      operator,
      apply: options.apply,
      limit: 1,
    })

    checkpoint.results.push(result)
    checkpoint.summary.processed += 1
    checkpoint.summary.imported += result.status === 'imported' ? 1 : 0
    checkpoint.summary.dryRun += result.status === 'dry_run' ? 1 : 0
    checkpoint.summary.skipped += result.status === 'skipped' ? 1 : 0
    checkpoint.summary.failed += result.status === 'failed' ? 1 : 0
    checkpoint.updatedAt = new Date().toISOString()

    await writeCheckpoint(options.checkpoint, checkpoint)

    console.log(`- 结果: ${result.status}${result.reason ? ` | ${result.reason}` : ''}`)

    if (result.status === 'failed' && options.stopOnFailure) {
      console.log('- 命中失败即停止')
      break
    }

    if (index < batch.length - 1) {
      await sleep(options.pauseMs)
    }
  }

  console.log('\n串行队列完成')
  console.log(`- 已处理总数: ${checkpoint.summary.processed}/${checkpoint.summary.totalCandidates}`)
  console.log(`- 已导入: ${checkpoint.summary.imported}`)
  console.log(`- 演练记录: ${checkpoint.summary.dryRun}`)
  console.log(`- 已跳过: ${checkpoint.summary.skipped}`)
  console.log(`- 失败: ${checkpoint.summary.failed}`)
  console.log(`- 进度文件: ${options.checkpoint}`)
}

main()
  .catch((error) => {
    console.error('历史素材串行队列失败:', error)
    process.exitCode = 1
  })
  .finally(async () => {
    await prisma.$disconnect()
    process.exit(process.exitCode ?? 0)
  })