fix(import): stabilize large and multi-file imports

This commit is contained in:
digua committed 2026-07-18 17:38:12 +08:00
1 parent 4518a3aaa9
commit c0942a72f0
6 files changed
+201 -17

No files matched your search

@@ -0,0 +1,83 @@
import assert from 'node:assert/strict'
import fs from 'node:fs'
import os from 'node:os'
import path from 'node:path'
import test from 'node:test'
import { MessageType } from '@openchatlab/shared-types'
import { findEntryFileInDirectory, parseFileWithFormat } from '../index'
function makeTempDir(): string {
const baseDir = process.env.CHATLAB_TEST_TMPDIR ?? (fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir())
return fs.mkdtempSync(path.join(baseDir, 'chatlab-jsonl-parser-'))
}
function writeChatLabJsonl(filePath: string, messageCount: number): void {
const lines = [
JSON.stringify({
_type: 'header',
chatlab: { version: '0.0.2', exportedAt: 1711468800 },
meta: { name: 'JSONL Test', platform: 'wechat', type: 'group', groupId: 'group-1' },
}),
JSON.stringify({ _type: 'member', platformId: 'member-1', accountName: 'Alice' }),
]
for (let index = 0; index < messageCount; index++) {
lines.push(
JSON.stringify({
_type: 'message',
sender: 'member-1',
accountName: 'Alice',
timestamp: 1711468800 + index,
type: MessageType.TEXT,
content: `message-${index}`,
})
)
}
fs.writeFileSync(filePath, `${lines.join('\n')}\n`, 'utf-8')
}
test('ChatLab JSONL emits message batches and progress while reading', async () => {
const root = makeTempDir()
const filePath = path.join(root, 'large-chat.jsonl')
writeChatLabJsonl(filePath, 12)
try {
const batchSizes: number[] = []
let progressCalls = 0
let progressCallsAtFirstBatch: number | null = null
for await (const event of parseFileWithFormat('chatlab-jsonl', {
filePath,
batchSize: 5,
onProgress: () => {
progressCalls++
},
})) {
if (event.type === 'messages') {
progressCallsAtFirstBatch ??= progressCalls
batchSizes.push(event.data.length)
}
}
assert.deepEqual(batchSizes, [5, 5, 2])
assert.equal(progressCallsAtFirstBatch, 1)
assert.equal(progressCalls, 4)
} finally {
fs.rmSync(root, { recursive: true, force: true })
}
})
test('directory entry detection accepts a top-level ChatLab JSONL file', () => {
const root = makeTempDir()
const filePath = path.join(root, 'chat.jsonl')
fs.mkdirSync(path.join(root, 'media'))
writeChatLabJsonl(filePath, 1)
try {
assert.equal(findEntryFileInDirectory(root), filePath)
} finally {
fs.rmSync(root, { recursive: true, force: true })
}
})
+7 -5
View File
@@ -225,8 +225,10 @@ async function* parseChatLabJsonl(options: ParseOptions): AsyncGenerator<ParseEv
})
messagesProcessed++
// 达到批次大小,发送进度
// Flush each batch immediately so large files do not retain every message or flood progress IPC.
if (messageBatch.length >= batchSize) {
yield { type: 'messages', data: messageBatch.splice(0) }
const progress = createProgress(
'parsing',
bytesRead,
@@ -234,6 +236,7 @@ async function* parseChatLabJsonl(options: ParseOptions): AsyncGenerator<ParseEv
messagesProcessed,
`已处理 ${messagesProcessed.toLocaleString()} 条消息...`
)
yield { type: 'progress', data: progress }
onProgress?.(progress)
}
break
@@ -257,10 +260,9 @@ async function* parseChatLabJsonl(options: ParseOptions): AsyncGenerator<ParseEv
yield { type: 'members', data: Array.from(memberMap.values()) }
}
// 分批发送消息
for (let i = 0; i < messageBatch.length; i += batchSize) {
const batch = messageBatch.slice(i, i + batchSize)
yield { type: 'messages', data: batch }
// 发送不足一个批次的剩余消息
if (messageBatch.length > 0) {
yield { type: 'messages', data: messageBatch }
}
// 完成
+8 -1
View File
@@ -260,6 +260,13 @@ export async function parseFileInfo(
* @returns path to the entry file, or null if none found
*/
export function findEntryFileInDirectory(dirPath: string, maxDepth = 1): string | null {
const supportedExtensions = new Set(
sniffer
.getSupportedFormats()
.flatMap((feature) => feature.extensions)
.map((extension) => extension.toLowerCase())
)
function walk(current: string, depth: number): string | null {
if (depth > maxDepth) return null
let entries: fs.Dirent[]
@@ -270,7 +277,7 @@ export function findEntryFileInDirectory(dirPath: string, maxDepth = 1): string
}
for (const entry of entries) {
const full = path.join(current, entry.name)
if (entry.isFile() && entry.name.endsWith('.json')) {
if (entry.isFile() && supportedExtensions.has(path.extname(entry.name).toLowerCase())) {
const format = detectFormat(full)
if (format) return full
}
@@ -300,9 +300,7 @@ async function processWebFiles(files: File[]) {
importError.value = translateError('error.archive_single_file_required')
return
}
for (const file of files) {
await importSingleWebFile(file)
}
await sessionStore.importFiles(files)
}
}
@@ -465,7 +463,7 @@ async function processFilePaths(paths: string[]) {
}
} else {
// 多文件批量导入(未启用合并)
await sessionStore.importFilesFromPaths(paths)
await sessionStore.importFiles(paths)
}
return
}
+78
View File
@@ -0,0 +1,78 @@
import assert from 'node:assert/strict'
import { setImmediate as waitForImmediate } from 'node:timers/promises'
import test from 'node:test'
import { createPinia, setActivePinia } from 'pinia'
import { registerAdapter } from '@/services/registry'
import type { DataAdapter, ImportAdapter, SessionIndexAdapter } from '@/services'
function createMemoryStorage(): Storage {
const values = new Map<string, string>()
return {
get length() {
return values.size
},
clear: () => values.clear(),
getItem: (key) => values.get(key) ?? null,
key: (index) => [...values.keys()][index] ?? null,
removeItem: (key) => values.delete(key),
setItem: (key, value) => values.set(key, value),
}
}
test('batch import accepts browser File objects and processes them serially', async () => {
Object.defineProperties(globalThis, {
localStorage: { configurable: true, value: createMemoryStorage() },
sessionStorage: { configurable: true, value: createMemoryStorage() },
})
const importCalls: File[] = []
let activeImports = 0
let maxActiveImports = 0
registerAdapter('import', {
async importFile(source: File | string) {
assert.ok(source instanceof File)
importCalls.push(source)
activeImports++
maxActiveImports = Math.max(maxActiveImports, activeImports)
await waitForImmediate()
activeImports--
return {
success: true,
sessionId: `session-${importCalls.length}`,
importMode: 'incremental' as const,
newMessageCount: 1,
duplicateCount: 0,
}
},
} as ImportAdapter)
registerAdapter('data', {
getSessions: async () => [],
tryApplyOwnerProfile: async () => ({ applied: false }),
} as unknown as DataAdapter)
registerAdapter('session-index', {
generate: async () => undefined,
} as unknown as SessionIndexAdapter)
setActivePinia(createPinia())
const { useSessionStore } = await import('./session')
const store = useSessionStore()
const files = [new File(['a'], 'first.json'), new File(['b'], 'second.jsonl')]
const result = await store.importFiles(files)
assert.deepEqual(
importCalls.map((file) => file.name),
['first.json', 'second.jsonl']
)
assert.equal(maxActiveImports, 1)
assert.equal(result.success, 2)
assert.equal(result.failed, 0)
assert.deepEqual(
result.files.map((file) => ({ name: file.name, status: file.status })),
[
{ name: 'first.json', status: 'success' },
{ name: 'second.jsonl', status: 'success' },
]
)
})
+23 -7
View File
@@ -340,20 +340,36 @@ export const useSessionStore = defineStore(
/**
* 批量导入多个文件(串行执行)
*/
async function importFilesFromPaths(filePaths: string[]): Promise<BatchImportResult> {
if (filePaths.length === 0) {
async function importFiles(sources: Array<File | string>): Promise<BatchImportResult> {
if (sources.length === 0) {
return { total: 0, success: 0, failed: 0, cancelled: 0, files: [] }
}
const importSources = sources.map((source) => {
if (typeof source === 'string') {
return {
source,
path: source,
name: source.split('/').pop() || source.split('\\').pop() || source,
}
}
return {
source,
path: source.webkitRelativePath || source.name,
name: source.name,
}
})
// 初始化批量导入状态
isBatchImporting.value = true
batchImportCancelled.value = false
batchImportResult.value = null
// 初始化文件列表
batchFiles.value = filePaths.map((path) => ({
batchFiles.value = importSources.map(({ path, name }) => ({
path,
name: path.split('/').pop() || path.split('\\').pop() || path,
name,
status: 'pending' as BatchFileStatus,
}))
@@ -422,7 +438,7 @@ export const useSessionStore = defineStore(
isProcessing = false
}
const importResult = await useImportService().importFile(file.path, undefined, (progress) => {
const importResult = await useImportService().importFile(importSources[i].source, undefined, (progress) => {
if (progress.stage === 'done') return
queue.push(progress)
processQueue()
@@ -503,7 +519,7 @@ export const useSessionStore = defineStore(
// 生成结果
const result: BatchImportResult = {
total: filePaths.length,
total: sources.length,
success: successCount,
failed: failedCount,
cancelled: cancelledCount,
@@ -880,7 +896,7 @@ export const useSessionStore = defineStore(
batchFiles,
batchImportCancelled,
batchImportResult,
importFilesFromPaths,
importFiles,
cancelBatchImport,
clearBatchImportResult,
// 合并导入