Function parseTelegram

packages/parser/src/formats/telegram-native.ts:119–270

(options: ParseOptions)

// ==================== 解析器实现 ====================

async function* parseTelegram(options: ParseOptions): AsyncGenerator<ParseEvent, void, unknown> {
  const { filePath, batchSize = 5000, formatOptions, onProgress, onLog } = options

  // 获取目标聊天索引
  const chatIndex = (formatOptions?.chatIndex as number) ?? 0

  const totalBytes = getFileSize(filePath)
  let bytesRead = 0
  let messagesProcessed = 0

  // 发送初始进度
  const initialProgress = createProgress('parsing', 0, totalBytes, 0, '')
  yield { type: 'progress', data: initialProgress }
  onProgress?.(initialProgress)

  onLog?.('info', `开始解析 Telegram JSON 文件，大小: ${(totalBytes / 1024 / 1024).toFixed(2)} MB`)
  onLog?.('info', `目标聊天索引: ${chatIndex}`)

  // 使用 stream-json 流式解析目标聊天
  // 定位到 chats.list[chatIndex] 对象
  const chatPathFilter = `chats.list.${chatIndex}`

  const chatData = await new Promise<TelegramChat | null>((resolve, reject) => {
    const readStream = fs.createReadStream(filePath, { encoding: 'utf-8' })

    readStream.on('data', (chunk: string | Buffer) => {
      bytesRead += typeof chunk === 'string' ? Buffer.byteLength(chunk) : chunk.length
    })

    const pipeline = chain([
      readStream,
      parser(),
      pick({ filter: new RegExp(`^${chatPathFilter.replace('.', '\\.')}$`) }),
      streamValues(),
    ])

    let found = false

    pipeline.on('data', ({ value }: { value: TelegramChat }) => {
      found = true
      resolve(value)
    })

    pipeline.on('end', () => {
      if (!found) resolve(null)
    })

    pipeline.on('error', reject)
  })

  if (!chatData) {
    onLog?.('error', `未找到索引 ${chatIndex} 对应的聊天`)
    yield { type: 'error', data: new Error(`未找到索引 ${chatIndex} 对应的聊天`) }
    return
  }

  onLog?.('info', `找到聊天: "${chatData.name}", 类型: ${chatData.type}, 消息数: ${chatData.messages?.length || 0}`)