diff --git a/apps/cli/src/http/routes/web/import-source.test.ts b/apps/cli/src/http/routes/web/import-source.test.ts index 5f40cd1e..fa9c06a7 100644 --- a/apps/cli/src/http/routes/web/import-source.test.ts +++ b/apps/cli/src/http/routes/web/import-source.test.ts @@ -63,10 +63,12 @@ describe('CLI Web archive import source routes', () => { const manager = new ArchiveImportSourceManager({ tempRoot: join(dir, 'sources') }) const importedManifests: string[] = [] + const receivedThresholds: Array = [] registerImportRoutes(app, {} as any, { sourceManager: manager, - runPreparedImport: async (manifestPath) => { + runPreparedImport: async (manifestPath, _onProgress, sessionGapThreshold) => { importedManifests.push(readFileSync(manifestPath, 'utf8')) + receivedThresholds.push(sessionGapThreshold) return { success: true, sessionId: `session-${importedManifests.length}` } }, }) @@ -88,13 +90,14 @@ describe('CLI Web archive import source routes', () => { const response = await app.inject({ method: 'POST', url: `/_web/import-sources/${prepared.source.sourceId}/import`, - payload: { chatId: 'Groups/DM sample' }, + payload: { chatId: 'Groups/DM sample', sessionGapThreshold: 7200 }, }) assert.equal(response.statusCode, 200) assert.match(response.body, /event: done/) assert.match(response.body, new RegExp(`session-${index + 1}`)) } assert.equal(importedManifests.length, 2) + assert.deepEqual(receivedThresholds, [7200, 7200]) for (let index = 0; index < 2; index++) { const response = await app.inject({ diff --git a/apps/cli/src/http/routes/web/import.test.ts b/apps/cli/src/http/routes/web/import.test.ts index c0ab8795..e417bf25 100644 --- a/apps/cli/src/http/routes/web/import.test.ts +++ b/apps/cli/src/http/routes/web/import.test.ts @@ -19,7 +19,14 @@ function multipartPayload(): { payload: Buffer; contentType: string } { return { payload: Buffer.from( - [field('formatId', 'telegram-json'), field('chatIndex', '2'), file, `--${boundary}--`, ''].join('\r\n') + [ + field('formatId', 'telegram-json'), + field('chatIndex', '2'), + field('sessionGapThreshold', '7200'), + file, + `--${boundary}--`, + '', + ].join('\r\n') ), contentType: `multipart/form-data; boundary=${boundary}`, } @@ -56,7 +63,7 @@ function directoryMultipartPayload(): { payload: Buffer; contentType: string } { describe('CLI Web automatic import route', () => { it('forwards parser options and preserves a zero-new incremental result in the done event', async () => { const app = Fastify() - const calls: Array<{ formatId?: string; chatIndex?: number }> = [] + const calls: Array<{ formatId?: string; chatIndex?: number; sessionGapThreshold?: number }> = [] try { await app.register(multipart) registerImportRoutes( @@ -67,6 +74,7 @@ describe('CLI Web automatic import route', () => { calls.push({ formatId: options.formatId as string | undefined, chatIndex: options.chatIndex as number | undefined, + sessionGapThreshold: options.sessionGapThreshold as number | undefined, }) return { success: true, @@ -89,7 +97,7 @@ describe('CLI Web automatic import route', () => { }) assert.equal(response.statusCode, 200) - assert.deepEqual(calls, [{ formatId: 'telegram-json', chatIndex: 2 }]) + assert.deepEqual(calls, [{ formatId: 'telegram-json', chatIndex: 2, sessionGapThreshold: 7200 }]) assert.match(response.body, /event: done/) assert.doesNotMatch(response.body, /event: error/) assert.match(response.body, /"importMode":"incremental"/) diff --git a/apps/cli/src/http/routes/web/import.ts b/apps/cli/src/http/routes/web/import.ts index 02ce45e7..b474b74f 100644 --- a/apps/cli/src/http/routes/web/import.ts +++ b/apps/cli/src/http/routes/web/import.ts @@ -33,10 +33,18 @@ interface ImportRouteOptions { runAutoImport?: (filePath: string, options: StreamImportOptions) => Promise runPreparedImport?: ( manifestPath: string, - onProgress: (progress: unknown) => void + onProgress: (progress: unknown) => void, + sessionGapThreshold?: number ) => Promise<{ success: boolean; sessionId?: string; error?: string; messageCount?: number; memberCount?: number }> } +function parseOptionalInteger(value: unknown): number | undefined { + if (typeof value === 'number') return Number.isInteger(value) ? value : undefined + if (typeof value !== 'string' || value.trim() === '') return undefined + const parsed = Number(value) + return Number.isInteger(parsed) ? parsed : undefined +} + function cleanupTemp(...paths: string[]) { for (const p of paths) { try { @@ -61,9 +69,10 @@ export function registerImportRoutes( const runAutoImport = options.runAutoImport ?? autoImport.bind(null, dbManager) const runPreparedImport = options.runPreparedImport ?? - (async (manifestPath: string, onProgress: (progress: unknown) => void) => { + (async (manifestPath: string, onProgress: (progress: unknown) => void, sessionGapThreshold?: number) => { const result = await runAutoImport(manifestPath, { formatId: 'google-chat-takeout', + sessionGapThreshold, nativeBinding: resolveNativeBinding(), onProgress: onProgress as any, }) @@ -102,7 +111,7 @@ export function registerImportRoutes( } }) - server.post<{ Params: { sourceId: string }; Body: { chatId?: string } }>( + server.post<{ Params: { sourceId: string }; Body: { chatId?: string; sessionGapThreshold?: number } }>( '/_web/import-sources/:sourceId/import', async (request, reply) => { const chatId = request.body?.chatId @@ -119,8 +128,9 @@ export function registerImportRoutes( } try { + const sessionGapThreshold = parseOptionalInteger(request.body?.sessionGapThreshold) const result = await sourceManager.withMaterializedChat(request.params.sourceId, chatId, (manifestPath) => - runPreparedImport(manifestPath, (progress) => sendEvent('progress', progress)) + runPreparedImport(manifestPath, (progress) => sendEvent('progress', progress), sessionGapThreshold) ) if (result.success) { sendEvent('done', result) @@ -207,6 +217,7 @@ export function registerImportRoutes( const formatId = (data.fields?.formatId as any)?.value as string | undefined const chatIndexStr = (data.fields?.chatIndex as any)?.value as string | undefined const chatIndex = chatIndexStr !== undefined ? parseInt(chatIndexStr, 10) : undefined + const sessionGapThreshold = parseOptionalInteger((data.fields?.sessionGapThreshold as any)?.value) reply.raw.writeHead(200, { 'Content-Type': 'text/event-stream', @@ -223,6 +234,7 @@ export function registerImportRoutes( const result = await runAutoImport(tmpPath, { formatId, chatIndex, + sessionGapThreshold, nativeBinding, onProgress: (p) => sendEvent('progress', p), }) @@ -249,11 +261,14 @@ export function registerImportRoutes( const tmpDir = createChatLabTempDir('imports', 'directory-') const relativePaths: string[] = [] const fileBuffers: { data: Buffer; filename: string }[] = [] + let sessionGapThreshold: number | undefined try { for await (const part of parts) { if (part.type === 'field' && part.fieldname === 'relativePaths') { relativePaths.push(String(part.value)) + } else if (part.type === 'field' && part.fieldname === 'sessionGapThreshold') { + sessionGapThreshold = parseOptionalInteger(part.value) } else if (part.type === 'file') { const chunks: Buffer[] = [] for await (const chunk of part.file) { @@ -292,6 +307,7 @@ export function registerImportRoutes( const nativeBinding = resolveNativeBinding() const result = await runAutoImport(entryPath, { + sessionGapThreshold, nativeBinding, onProgress: (p) => sendEvent('progress', p), }) @@ -407,32 +423,36 @@ export function registerImportRoutes( // ==================== Demo Import ==================== - server.post<{ Body: { locale?: string; timeZone?: string } }>('/_web/demo/import', async (request, reply) => { - const locale = request.body?.locale === 'cn' ? 'cn' : 'en' - const targetTimeZone = typeof request.body?.timeZone === 'string' ? request.body.timeZone : undefined - const nativeBinding = resolveNativeBinding() + server.post<{ Body: { locale?: string; timeZone?: string; sessionGapThreshold?: number } }>( + '/_web/demo/import', + async (request, reply) => { + const locale = request.body?.locale === 'cn' ? 'cn' : 'en' + const targetTimeZone = typeof request.body?.timeZone === 'string' ? request.body.timeZone : undefined + const sessionGapThreshold = parseOptionalInteger(request.body?.sessionGapThreshold) + const nativeBinding = resolveNativeBinding() - reply.raw.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - }) + reply.raw.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }) - function sendEvent(event: string, eventData: unknown) { - reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`) + function sendEvent(event: string, eventData: unknown) { + reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(eventData)}\n\n`) + } + + const result = await importDemoSessions({ + locale, + tempPrefix: 'cli-demo-', + targetTimeZone, + importFile: (filePath) => streamImport(dbManager, filePath, { nativeBinding, sessionGapThreshold }), + deleteSession: (sessionId) => { + dbManager.deleteSessionDatabaseFiles(sessionId) + }, + onProgress: (progress) => sendEvent('progress', progress), + }) + sendEvent('result', result) + reply.raw.end() } - - const result = await importDemoSessions({ - locale, - tempPrefix: 'cli-demo-', - targetTimeZone, - importFile: (filePath) => streamImport(dbManager, filePath, { nativeBinding }), - deleteSession: (sessionId) => { - dbManager.deleteSessionDatabaseFiles(sessionId) - }, - onProgress: (progress) => sendEvent('progress', progress), - }) - sendEvent('result', result) - reply.raw.end() - }) + ) } diff --git a/apps/cli/src/import/command.ts b/apps/cli/src/import/command.ts index 3d84b9c1..99f13413 100644 --- a/apps/cli/src/import/command.ts +++ b/apps/cli/src/import/command.ts @@ -174,6 +174,7 @@ export function registerImportCommand(program: Command): void { const importOptions = { formatId: options.format, sessionId: options.sessionId, + sessionGapThreshold: runtime.config.ui.session_gap_threshold, nativeBinding, onProgress: options.json ? undefined diff --git a/apps/cli/src/import/stream-import-flow.test.ts b/apps/cli/src/import/stream-import-flow.test.ts index 8fa0cdd4..c9872423 100644 --- a/apps/cli/src/import/stream-import-flow.test.ts +++ b/apps/cli/src/import/stream-import-flow.test.ts @@ -76,13 +76,27 @@ test('streamImport raises the data directory gate after creating a current-schem }) const chatFile = writeTempChatFile(root) - const result = await streamImport(manager, chatFile, { sessionId: 'test-session', nativeBinding }) + const result = await streamImport(manager, chatFile, { + sessionId: 'test-session', + sessionGapThreshold: 7200, + nativeBinding, + }) assert.equal(result.success, true) const meta = readDataDirCompatibilityMeta(path.join(root, 'data')) assert.equal(meta?.minRuntimeVersion, '0.25.1') assert.equal(meta?.dataCompatibilityVersion, 1) assert.deepEqual(meta?.reasons, ['segment-schema']) + + const db = new Database(path.join(root, 'data', 'databases', 'test-session.db'), { + readonly: true, + nativeBinding, + }) + const sessionMeta = db.prepare('SELECT session_gap_threshold FROM meta LIMIT 1').get() as { + session_gap_threshold: number | null + } + db.close() + assert.equal(sessionMeta.session_gap_threshold, 7200) }) test('streamImport re-checks data directory compatibility before raw database writes', async () => { diff --git a/apps/cli/src/import/stream-import.ts b/apps/cli/src/import/stream-import.ts index a7a6fce1..3f9f882d 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -58,6 +58,7 @@ export interface StreamImportProgress { export interface StreamImportOptions { formatId?: string chatIndex?: number + sessionGapThreshold?: number nativeBinding?: string onProgress?: (progress: StreamImportProgress) => void /** Fix the target session ID instead of auto-generating one. Used by sync/pull adapters. */ @@ -70,7 +71,11 @@ function generateSessionId(): string { return `chat_${ts}_${rand}` } -function buildStreamImportDeps(dbManager: DatabaseManager, onProgress?: ImportProgressCallback): StreamImportDeps { +function buildStreamImportDeps( + dbManager: DatabaseManager, + onProgress?: ImportProgressCallback, + sessionGapThreshold?: number +): StreamImportDeps { return { openDatabase(sessionId: string) { return dbManager.openRawSessionDatabase(sessionId, { create: true, initializeChatTables: true }) @@ -80,6 +85,7 @@ function buildStreamImportDeps(dbManager: DatabaseManager, onProgress?: ImportPr }, onProgress: onProgress ?? (() => {}), generateSessionId, + sessionGapThreshold, } } @@ -138,7 +144,7 @@ async function streamImportUnlocked( filePath: string, options?: StreamImportOptions ): Promise { - const { formatId, chatIndex, onProgress, sessionId } = options || {} + const { formatId, chatIndex, sessionGapThreshold, onProgress, sessionId } = options || {} const formatOptions: Record = {} if (formatId) formatOptions.formatId = formatId @@ -146,7 +152,7 @@ async function streamImportUnlocked( const progressAdapter = createProgressAdapter(onProgress) - const deps = buildStreamImportDeps(dbManager, progressAdapter) + const deps = buildStreamImportDeps(dbManager, progressAdapter, sessionGapThreshold) const result = await streamingImport(filePath, deps, formatOptions, sessionId) if (!result.success || !result.sessionId) return result diff --git a/apps/desktop/main/internal-api/server.ts b/apps/desktop/main/internal-api/server.ts index 8c2490cc..c1df5089 100644 --- a/apps/desktop/main/internal-api/server.ts +++ b/apps/desktop/main/internal-api/server.ts @@ -111,7 +111,11 @@ export async function startInternalServer( newSemanticIndexService = null } - const electronStreamImport = async (dm: DatabaseManager, filePath: string) => { + const electronStreamImport = async ( + dm: DatabaseManager, + filePath: string, + options?: { sessionGapThreshold?: number } + ) => { return withDataDirImportLock(dm.getUserDataDir(), async () => { const deps: StreamImportDeps = { openDatabase(sessionId: string) { @@ -123,6 +127,7 @@ export async function startInternalServer( onProgress() { /* no progress for merge-triggered import */ }, + sessionGapThreshold: options?.sessionGapThreshold, } const result = await streamingImport(filePath, deps) if (!result.success) throw new Error(result.error || 'Import failed') diff --git a/apps/desktop/main/ipc/chat.ts b/apps/desktop/main/ipc/chat.ts index 717d814f..999ffca2 100644 --- a/apps/desktop/main/ipc/chat.ts +++ b/apps/desktop/main/ipc/chat.ts @@ -146,24 +146,33 @@ export function registerChatHandlers(ctx: IpcContext): void { } }) - ipcMain.handle('chat:importPreparedChat', async (_, sourceId: string, chatId: string) => { - try { - const result = await getArchiveImportSourceManager().withMaterializedChat(sourceId, chatId, (manifestPath) => - worker.autoImport(manifestPath, forwardImportProgress, { formatId: 'google-chat-takeout' }) - ) - return finishAutoImport(result) - } catch (error) { - return { - success: false, - error: - error && typeof error === 'object' && 'code' in error - ? String(error.code) - : error instanceof Error - ? error.message - : String(error), + ipcMain.handle( + 'chat:importPreparedChat', + async (_, sourceId: string, chatId: string, options?: { sessionGapThreshold?: number }) => { + try { + const result = await getArchiveImportSourceManager().withMaterializedChat(sourceId, chatId, (manifestPath) => + worker.autoImport( + manifestPath, + forwardImportProgress, + { formatId: 'google-chat-takeout' }, + undefined, + options?.sessionGapThreshold + ) + ) + return finishAutoImport(result) + } catch (error) { + return { + success: false, + error: + error && typeof error === 'object' && 'code' in error + ? String(error.code) + : error instanceof Error + ? error.message + : String(error), + } } } - }) + ) ipcMain.handle('chat:releaseImportSource', async (_, sourceId: string) => { await getArchiveImportSourceManager().release(sourceId) @@ -184,14 +193,20 @@ export function registerChatHandlers(ctx: IpcContext): void { } }) - ipcMain.handle('chat:importDirectory', async (_, dirPath: string) => { + ipcMain.handle('chat:importDirectory', async (_, dirPath: string, options?: { sessionGapThreshold?: number }) => { try { const entryPath = findEntryFileInDirectory(dirPath) if (!entryPath) return { success: false, error: 'No recognizable import format found in directory' } win.webContents.send('chat:importProgress', { stage: 'detecting', progress: 5, message: '' }) - const result = await worker.autoImport(entryPath, forwardImportProgress) + const result = await worker.autoImport( + entryPath, + forwardImportProgress, + undefined, + undefined, + options?.sessionGapThreshold + ) return finishAutoImport(result) } catch (error) { win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: String(error) }) @@ -199,11 +214,18 @@ export function registerChatHandlers(ctx: IpcContext): void { } }) - ipcMain.handle('chat:importWithOptions', async (_, filePath: string, formatOptions: Record) => { + ipcMain.handle('chat:importWithOptions', async (_, filePath: string, options: Record) => { try { win.webContents.send('chat:importProgress', { stage: 'detecting', progress: 5, message: '' }) - const result = await worker.autoImport(filePath, forwardImportProgress, formatOptions) + const { sessionGapThreshold, ...formatOptions } = options + const result = await worker.autoImport( + filePath, + forwardImportProgress, + formatOptions, + undefined, + typeof sessionGapThreshold === 'number' ? sessionGapThreshold : undefined + ) return finishAutoImport(result) } catch (error) { win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: String(error) }) diff --git a/apps/desktop/main/ipc/demo.ts b/apps/desktop/main/ipc/demo.ts index a25a7303..111c87c9 100644 --- a/apps/desktop/main/ipc/demo.ts +++ b/apps/desktop/main/ipc/demo.ts @@ -18,7 +18,8 @@ export function registerDemoHandlers(ctx: IpcContext): void { 'demo:downloadAndImport', async ( _, - locale: string + locale: string, + options?: { sessionGapThreshold?: number } ): Promise<{ success: boolean groupSessionId?: string @@ -32,7 +33,8 @@ export function registerDemoHandlers(ctx: IpcContext): void { return importDemoSessions({ locale: locale === 'cn' ? 'cn' : 'en', tempPrefix: 'desktop-demo-', - importFile: (filePath) => worker.streamImport(filePath), + importFile: (filePath) => + worker.streamImport(filePath, undefined, undefined, undefined, options?.sessionGapThreshold), deleteSession: (sessionId) => worker.deleteImportedSession(sessionId), onProgress: sendProgress, }) diff --git a/apps/desktop/main/worker/dbWorker.ts b/apps/desktop/main/worker/dbWorker.ts index 867c8570..2cce168e 100644 --- a/apps/desktop/main/worker/dbWorker.ts +++ b/apps/desktop/main/worker/dbWorker.ts @@ -251,8 +251,8 @@ const syncHandlers: Record any> = { // 异步消息处理器(流式操作) const asyncHandlers: Record Promise> = { // 流式导入 - streamImport: (p, id) => streamImport(p.filePath, id, p.formatOptions, p.externalSessionId), - autoImport: (p, id) => autoImport(p.filePath, id, p.formatOptions, p.explicitSessionId), + streamImport: (p, id) => streamImport(p.filePath, id, p.formatOptions, p.externalSessionId, p.sessionGapThreshold), + autoImport: (p, id) => autoImport(p.filePath, id, p.formatOptions, p.explicitSessionId, p.sessionGapThreshold), analyzePushImport: (p) => analyzePushImport(p.sessionId, p.payload), pushImport: (p) => pushImport(p.sessionId, p.payload), // 流式解析文件信息(用于合并预览) diff --git a/apps/desktop/main/worker/import/streamImport.ts b/apps/desktop/main/worker/import/streamImport.ts index a6e8d869..5f2fb0cc 100644 --- a/apps/desktop/main/worker/import/streamImport.ts +++ b/apps/desktop/main/worker/import/streamImport.ts @@ -66,7 +66,7 @@ function buildElectronLogger(): ImportLogger { } } -function buildStreamImportDeps(requestId: string): StreamImportDeps { +function buildStreamImportDeps(requestId: string, sessionGapThreshold?: number): StreamImportDeps { return { openDatabase(sessionId: string) { const db = createDatabaseWithoutIndexes(sessionId) @@ -101,6 +101,7 @@ function buildStreamImportDeps(requestId: string): StreamImportDeps { } }, generateSessionId, + sessionGapThreshold, } } @@ -111,16 +112,23 @@ export async function streamImport( filePath: string, requestId: string, formatOptions?: Record, - externalSessionId?: string + externalSessionId?: string, + sessionGapThreshold?: number ): Promise { - return streamingImport(filePath, buildStreamImportDeps(requestId), formatOptions, externalSessionId) + return streamingImport( + filePath, + buildStreamImportDeps(requestId, sessionGapThreshold), + formatOptions, + externalSessionId + ) } export async function autoImport( filePath: string, requestId: string, formatOptions?: Record, - explicitSessionId?: string + explicitSessionId?: string, + sessionGapThreshold?: number ): Promise { return sharedAutoImportFile( filePath, @@ -136,7 +144,7 @@ export async function autoImport( onProgress: (progress) => sendProgress(requestId, progress), sessionExists: (sessionId) => fs.existsSync(getDbPath(sessionId)), createSession: (sourcePath, sourceFormatOptions, sessionId) => - streamImport(sourcePath, requestId, sourceFormatOptions, sessionId), + streamImport(sourcePath, requestId, sourceFormatOptions, sessionId, sessionGapThreshold), appendSession: (sessionId, sourcePath, sourceFormatOptions) => incrementalImport(sessionId, sourcePath, requestId, { formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index 1105accc..04ab6b6d 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/apps/desktop/main/worker/workerManager.ts @@ -529,13 +529,14 @@ async function streamImportUnlocked( filePath: string, onProgress?: (progress: ParseProgress) => void, formatOptions?: Record, - externalSessionId?: string + externalSessionId?: string, + sessionGapThreshold?: number ): Promise { assertDataDirCompatibleNow() const result = await sendToWorkerWithProgress( 'streamImport', - { filePath, formatOptions, externalSessionId }, + { filePath, formatOptions, externalSessionId, sessionGapThreshold }, onProgress ) if (!result.success || !result.sessionId) return result @@ -561,11 +562,12 @@ export async function streamImport( filePath: string, onProgress?: (progress: ParseProgress) => void, formatOptions?: Record, - externalSessionId?: string + externalSessionId?: string, + sessionGapThreshold?: number ): Promise { try { return await withDataDirImportLock(getPathProvider().getUserDataDir(), () => - streamImportUnlocked(filePath, onProgress, formatOptions, externalSessionId) + streamImportUnlocked(filePath, onProgress, formatOptions, externalSessionId, sessionGapThreshold) ) } catch (error) { if (error instanceof ImportInProgressError) { @@ -579,7 +581,8 @@ export async function autoImport( filePath: string, onProgress?: (progress: ParseProgress) => void, formatOptions?: Record, - explicitSessionId?: string + explicitSessionId?: string, + sessionGapThreshold?: number ): Promise { try { return await withDataDirImportLock(getPathProvider().getUserDataDir(), async () => { @@ -587,7 +590,7 @@ export async function autoImport( const result = await sendToWorkerWithProgress( 'autoImport', - { filePath, formatOptions, explicitSessionId }, + { filePath, formatOptions, explicitSessionId, sessionGapThreshold }, onProgress ) if (!result.success || !result.sessionId || result.importMode !== 'created') return result diff --git a/apps/desktop/preload/apis/chat.ts b/apps/desktop/preload/apis/chat.ts index 389eddd2..c5f10663 100644 --- a/apps/desktop/preload/apis/chat.ts +++ b/apps/desktop/preload/apis/chat.ts @@ -48,7 +48,8 @@ export const chatApi = { import: (filePath: string): Promise => ipcRenderer.invoke('chat:import', filePath), - importDirectory: (dirPath: string): Promise => ipcRenderer.invoke('chat:importDirectory', dirPath), + importDirectory: (dirPath: string, options?: Record): Promise => + ipcRenderer.invoke('chat:importDirectory', dirPath, options), detectFormat: ( filePath: string @@ -68,8 +69,11 @@ export const chatApi = { prepareImportSource: (filePath: string) => ipcRenderer.invoke('chat:prepareImportSource', filePath), - importPreparedChat: (sourceId: string, chatId: string): Promise => - ipcRenderer.invoke('chat:importPreparedChat', sourceId, chatId), + importPreparedChat: ( + sourceId: string, + chatId: string, + options?: Record + ): Promise => ipcRenderer.invoke('chat:importPreparedChat', sourceId, chatId, options), releaseImportSource: (sourceId: string) => ipcRenderer.invoke('chat:releaseImportSource', sourceId), @@ -105,9 +109,10 @@ export const chatApi = { // ==================== Demo ==================== importDemo: ( - locale: string + locale: string, + options?: Record ): Promise<{ success: boolean; groupSessionId?: string; privateSessionIds?: string[]; error?: string }> => - ipcRenderer.invoke('demo:downloadAndImport', locale), + ipcRenderer.invoke('demo:downloadAndImport', locale, options), onDemoProgress: ( callback: (progress: { stage: string; current: number; total: number; message?: string }) => void diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index 1cfc6d78..9374f3da 100644 --- a/apps/desktop/preload/index.d.ts +++ b/apps/desktop/preload/index.d.ts @@ -86,7 +86,7 @@ interface ChatApi { selectFile: () => Promise<{ filePath?: string; format?: string; error?: string } | null> detectFormat: (filePath: string) => Promise<{ id: string; name: string; platform: string; multiChat: boolean } | null> import: (filePath: string) => Promise - importDirectory: (dirPath: string) => Promise + importDirectory: (dirPath: string, options?: Record) => Promise importWithOptions: (filePath: string, formatOptions: Record) => Promise scanMultiChatFile: (filePath: string) => Promise<{ success: boolean @@ -110,7 +110,7 @@ interface ChatApi { } error?: string }> - importPreparedChat: (sourceId: string, chatId: string) => Promise + importPreparedChat: (sourceId: string, chatId: string, options?: Record) => Promise releaseImportSource: (sourceId: string) => Promise<{ success: boolean }> checkMigration: () => Promise runMigration: () => Promise<{ success: boolean; error?: string }> @@ -131,7 +131,10 @@ interface ChatApi { sessionId: string, filePath: string ) => Promise<{ success: boolean; newMessageCount: number; error?: string }> - importDemo: (locale: string) => Promise<{ + importDemo: ( + locale: string, + options?: Record + ) => Promise<{ success: boolean groupSessionId?: string privateSessionIds?: string[] diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 6a4e4eb6..db902301 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -26,6 +26,7 @@ export { getSummaryCount, getLastPlatformMessageId, DEFAULT_SESSION_GAP_THRESHOLD, + normalizeSessionGapThreshold, hasSessionIndex, getSessionIndexStats, getChatSessionList, diff --git a/packages/core/src/query/__tests__/session-index.test.ts b/packages/core/src/query/__tests__/session-index.test.ts index 0785e425..758182e0 100644 --- a/packages/core/src/query/__tests__/session-index.test.ts +++ b/packages/core/src/query/__tests__/session-index.test.ts @@ -12,6 +12,7 @@ import assert from 'node:assert/strict' import Database from 'better-sqlite3' import { DEFAULT_SESSION_GAP_THRESHOLD, + normalizeSessionGapThreshold, hasSessionIndex, getSessionIndexStats, getChatSessionList, @@ -124,6 +125,15 @@ describe('DEFAULT_SESSION_GAP_THRESHOLD', () => { it('equals 1800 (30 minutes)', () => { assert.equal(DEFAULT_SESSION_GAP_THRESHOLD, 1800) }) + + it('normalizes transport values to the supported range', () => { + assert.equal(normalizeSessionGapThreshold(60), 60) + assert.equal(normalizeSessionGapThreshold(86400), 86400) + assert.equal(normalizeSessionGapThreshold(59), DEFAULT_SESSION_GAP_THRESHOLD) + assert.equal(normalizeSessionGapThreshold(86401), DEFAULT_SESSION_GAP_THRESHOLD) + assert.equal(normalizeSessionGapThreshold(60.5), DEFAULT_SESSION_GAP_THRESHOLD) + assert.equal(normalizeSessionGapThreshold('7200'), DEFAULT_SESSION_GAP_THRESHOLD) + }) }) describe('hasSessionIndex', () => { @@ -179,6 +189,7 @@ describe('generateSessionIndex', () => { assert.equal(count, 2) assert.equal(countRows(db, 'segment'), 2) assert.equal(countRows(db, 'message_context'), 4) + assert.equal(getSessionIndexStats(db).gapThreshold, 2000) }) it('puts all messages in one session when gap is large enough', () => { @@ -373,8 +384,9 @@ describe('generateSessionIndex atomicity', () => { END; `) - assert.throws(() => generateSessionIndex(db, 2000), /blocked/) + assert.throws(() => generateSessionIndex(db, 900), /blocked/) assert.equal(countRows(db, 'segment'), 2) assert.equal(countRows(db, 'message_context'), 2) + assert.equal(getSessionIndexStats(db).gapThreshold, 2000) }) }) diff --git a/packages/core/src/query/index.ts b/packages/core/src/query/index.ts index 1496b350..f1265d7f 100644 --- a/packages/core/src/query/index.ts +++ b/packages/core/src/query/index.ts @@ -14,6 +14,7 @@ export { getLastPlatformMessageId, // Session index (segment) helpers DEFAULT_SESSION_GAP_THRESHOLD, + normalizeSessionGapThreshold, hasSessionIndex, getSessionIndexStats, getChatSessionList, diff --git a/packages/core/src/query/session-queries.ts b/packages/core/src/query/session-queries.ts index 55df2207..ed3482c5 100644 --- a/packages/core/src/query/session-queries.ts +++ b/packages/core/src/query/session-queries.ts @@ -369,6 +369,17 @@ export function getSegmentSummaries( /** Default gap threshold for session segmentation: 30 minutes (seconds) */ export const DEFAULT_SESSION_GAP_THRESHOLD = 1800 +const MIN_SESSION_GAP_THRESHOLD = 60 +const MAX_SESSION_GAP_THRESHOLD = 86400 + +export function normalizeSessionGapThreshold(value: unknown): number { + return typeof value === 'number' && + Number.isInteger(value) && + value >= MIN_SESSION_GAP_THRESHOLD && + value <= MAX_SESSION_GAP_THRESHOLD + ? value + : DEFAULT_SESSION_GAP_THRESHOLD +} export interface ChatSessionItem { id: number @@ -650,6 +661,7 @@ export function generateSessionIndex( processed++ if (onProgress && processed % 100 === 0) onProgress(processed, total) } + updateSessionGapThreshold(db, gapThreshold) if (onProgress) onProgress(total, total) return total }) diff --git a/packages/http-routes/src/context/merge.ts b/packages/http-routes/src/context/merge.ts index 84290bc3..1962f1c7 100644 --- a/packages/http-routes/src/context/merge.ts +++ b/packages/http-routes/src/context/merge.ts @@ -4,5 +4,9 @@ import type { DatabaseManager, MergeSessionCache } from '@openchatlab/node-runti export interface MergeRouteContext { mergeSessionCache?: MergeSessionCache /** Platform-specific import function for the merge "andImport" flow. */ - streamImport?: (dbManager: DatabaseManager, filePath: string) => Promise<{ sessionId: string }> + streamImport?: ( + dbManager: DatabaseManager, + filePath: string, + options?: { sessionGapThreshold?: number } + ) => Promise<{ sessionId: string }> } diff --git a/packages/http-routes/src/routes/web/merge.test.ts b/packages/http-routes/src/routes/web/merge.test.ts new file mode 100644 index 00000000..9d05bfc3 --- /dev/null +++ b/packages/http-routes/src/routes/web/merge.test.ts @@ -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 Fastify from 'fastify' +import type { PathProvider } from '@openchatlab/core' +import { DatabaseManager, MergeSessionCache, TempDbWriter } from '@openchatlab/node-runtime' +import { registerMergeRoutes } from './merge' + +const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') + +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-merge-route-')) +} + +function createPathProvider(rootDir: string): PathProvider { + return { + getSystemDir: () => rootDir, + getUserDataDir: () => rootDir, + getDatabaseDir: () => path.join(rootDir, 'databases'), + getVectorDir: () => path.join(rootDir, 'vector'), + getAiDataDir: () => path.join(rootDir, 'ai'), + getSettingsDir: () => path.join(rootDir, 'settings'), + getCacheDir: () => path.join(rootDir, 'cache'), + getTempDir: () => path.join(rootDir, 'temp'), + getLogsDir: () => path.join(rootDir, 'logs'), + getDownloadsDir: () => rootDir, + } +} + +test('forwards the session gap threshold when importing merged output', async (t) => { + const rootDir = makeTempDir() + t.after(() => fs.rmSync(rootDir, { recursive: true, force: true })) + + const pathProvider = createPathProvider(rootDir) + const dbManager = new DatabaseManager(pathProvider, { nativeBinding, allowMissingRuntimeForTests: true }) + const mergeSessionCache = new MergeSessionCache(pathProvider, { nativeBinding }) + const { db, tempDbPath } = mergeSessionCache.createTempDatabase('source.json') + const writer = new TempDbWriter(db) + writer.writeMeta({ name: 'Source', platform: 'wechat', type: 'private' }) + writer.writeMembers([{ platformId: 'wxid_alice', accountName: 'Alice' }]) + writer.writeMessages([ + { + senderPlatformId: 'wxid_alice', + senderAccountName: 'Alice', + timestamp: 100, + type: 0, + content: 'hello', + }, + ]) + writer.finish() + const handle = mergeSessionCache.store('source.json', tempDbPath) + + let receivedOptions: { sessionGapThreshold?: number } | undefined + const app = Fastify() + registerMergeRoutes(app, { + dbManager, + mergeSessionCache, + async streamImport(_manager, _filePath, options) { + receivedOptions = options + return { sessionId: 'merged-session' } + }, + }) + await app.ready() + t.after(() => app.close()) + + const response = await app.inject({ + method: 'POST', + url: '/_web/merge/execute', + payload: { + handles: [handle], + outputName: 'Merged', + andImport: true, + sessionGapThreshold: 7200, + }, + }) + + assert.equal(response.statusCode, 200) + assert.equal(response.json().sessionId, 'merged-session') + assert.deepEqual(receivedOptions, { sessionGapThreshold: 7200 }) +}) diff --git a/packages/http-routes/src/routes/web/merge.ts b/packages/http-routes/src/routes/web/merge.ts index fa106706..c67adef2 100644 --- a/packages/http-routes/src/routes/web/merge.ts +++ b/packages/http-routes/src/routes/web/merge.ts @@ -131,9 +131,15 @@ export function registerMergeRoutes(server: FastifyInstance, ctx: MergeRoutesCon // ── execute ──────────────────────────────────────────────────────── server.post<{ - Body: { handles: string[]; outputName: string; format?: 'json' | 'jsonl'; andImport?: boolean } + Body: { + handles: string[] + outputName: string + format?: 'json' | 'jsonl' + andImport?: boolean + sessionGapThreshold?: number + } }>('/_web/merge/execute', async (request, reply) => { - const { handles, outputName, format = 'json', andImport } = request.body as any + const { handles, outputName, format = 'json', andImport, sessionGapThreshold } = request.body if (!handles || !Array.isArray(handles) || handles.length === 0) { return reply.code(400).send({ error: 'Missing handles' }) } @@ -165,7 +171,7 @@ export function registerMergeRoutes(server: FastifyInstance, ctx: MergeRoutesCon fs.writeFileSync(tmpPath, jsonData, 'utf-8') try { - const importResult = await streamImport(dbManager, tmpPath) + const importResult = await streamImport(dbManager, tmpPath, { sessionGapThreshold }) sessionId = importResult.sessionId } finally { try { diff --git a/packages/node-runtime/src/import/incremental-importer.test.ts b/packages/node-runtime/src/import/incremental-importer.test.ts index bc696bf0..8ded1098 100644 --- a/packages/node-runtime/src/import/incremental-importer.test.ts +++ b/packages/node-runtime/src/import/incremental-importer.test.ts @@ -3,7 +3,7 @@ import fs from 'node:fs' import os from 'node:os' import path from 'node:path' import test from 'node:test' -import { CHAT_DB_SCHEMA } from '@openchatlab/core' +import { CHAT_DB_SCHEMA, generateSessionIndex } from '@openchatlab/core' import { openBetterSqliteDatabase } from '../better-sqlite3-adapter' import { analyzeIncrementalImport, incrementalImport, type IncrementalImportDeps } from './incremental-importer' @@ -145,6 +145,58 @@ test('imports ChatLab JSONL messages with numeric string timestamps consistently }) }) +test('continues incremental session indexing with the stored gap threshold', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const dbPath = path.join(tempDir, 'session.db') + const filePath = path.join(tempDir, 'incremental-threshold.json') + seedSessionDb(dbPath) + + const seededDb = openBetterSqliteDatabase(dbPath, { nativeBinding }) + seededDb.prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)').run('wxid_alice', 'Alice') + const member = seededDb.prepare('SELECT id FROM member WHERE platform_id = ?').get('wxid_alice') as { id: number } + seededDb + .prepare('INSERT INTO message (sender_id, ts, type, content) VALUES (?, ?, ?, ?)') + .run(member.id, 100, 0, 'before') + generateSessionIndex(seededDb, 60) + seededDb.close() + + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 200 }, + meta: { name: 'Threshold', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [ + { + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: 200, + type: 0, + content: 'after', + }, + ], + }), + 'utf8' + ) + + const result = await incrementalImport('session', filePath, createDeps(dbPath)) + assert.equal(result.success, true) + + const db = openBetterSqliteDatabase(dbPath, { readonly: true, nativeBinding }) + const counts = db + .prepare( + `SELECT + (SELECT COUNT(*) FROM segment) AS segments, + (SELECT session_gap_threshold FROM meta LIMIT 1) AS gapThreshold` + ) + .get() as { segments: number; gapThreshold: number | null } + db.close() + + assert.deepEqual(counts, { segments: 2, gapThreshold: 60 }) +}) + test('preserves ChatLab JSON member aliases during incremental import', async (t) => { const tempDir = makeTempDir() t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index 9e2dc3aa..1a364202 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -9,7 +9,7 @@ */ import type { DatabaseAdapter } from '@openchatlab/core' -import { generateSessionIndex, generateIncrementalSessionIndex } from '@openchatlab/core' +import { generateSessionIndex, generateIncrementalSessionIndex, getSessionIndexStats } from '@openchatlab/core' import { streamParseFile, detectFormat, getFormatFeatureById, type ParseProgress } from '@openchatlab/parser' import { insertFtsEntries, hasFtsTable } from '../fts' import { generateFallbackMessageKey, registerMessageAndCheckDuplicate } from './message-deduplicator' @@ -444,10 +444,11 @@ export async function incrementalImport( // older backfilled message would be incorrectly attached to the newest segment. if (newMessageCount > 0) { try { + const sessionGapThreshold = getSessionIndexStats(db).gapThreshold if (minWrittenTs < preWriteMaxTs) { - generateSessionIndex(db) + generateSessionIndex(db, sessionGapThreshold) } else { - generateIncrementalSessionIndex(db) + generateIncrementalSessionIndex(db, sessionGapThreshold) } } catch { /* non-fatal */ diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index 51eee213..3b69b6fd 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/packages/node-runtime/src/import/streaming-importer.test.ts @@ -4,7 +4,7 @@ import os from 'node:os' import path from 'node:path' import test from 'node:test' import Database from 'better-sqlite3' -import { CHAT_DB_TABLES, type DatabaseAdapter } from '@openchatlab/core' +import { CHAT_DB_TABLES, generateSessionIndex, type DatabaseAdapter } from '@openchatlab/core' import { isNativeFormatAvailable } from '@openchatlab/parser' import { BetterSqliteAdapter } from '../better-sqlite3-adapter' import { computeAndSetOverviewCache } from '../cache/session-cache' @@ -416,6 +416,36 @@ test('streamingImport updates avatars for members first created from message bat assert.equal(row?.avatar, 'data:image/png;base64,AAAA') }) +test('streamingImport builds the session index exactly once with the requested gap threshold', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + const manifestPath = writeChunkedShuakamiQqExport(root) + const dbPath = path.join(root, 'session-index-threshold.db') + const deps = createImportDeps(dbPath) + let generateCalls = 0 + let receivedThreshold: number | undefined + + deps.sessionGapThreshold = 60 + deps.generateSessionIndex = (db, gapThreshold) => { + generateCalls++ + receivedThreshold = gapThreshold + return generateSessionIndex(db, gapThreshold) + } + + const result = await streamingImport(manifestPath, deps, undefined, 'session-index-threshold') + + assert.equal(result.success, true) + assert.equal(generateCalls, 1) + assert.equal(receivedThreshold, 60) + + const db = new Database(dbPath, { readonly: true, nativeBinding }) + const meta = db.prepare('SELECT session_gap_threshold FROM meta LIMIT 1').get() as { + session_gap_threshold: number | null + } + db.close() + assert.equal(meta.session_gap_threshold, 60) +}) + test( 'streamingImport persists shuakami/qq-chat-exporter V4 native-first output with deduplication, reply and FTS', { diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index ea0b3484..c32913e8 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -10,7 +10,11 @@ */ import type { DatabaseAdapter } from '@openchatlab/core' -import { CHAT_DB_INDEXES, generateSessionIndex } from '@openchatlab/core' +import { + CHAT_DB_INDEXES, + generateSessionIndex as generateCoreSessionIndex, + normalizeSessionGapThreshold, +} from '@openchatlab/core' import type { ParsedMember, ParsedMessage } from '@openchatlab/shared-types' import { streamParseFile, @@ -112,6 +116,10 @@ export interface StreamImportDeps { postImportHook?: (db: DatabaseAdapter, sessionId: string) => void | Promise /** Generate a session ID. Defaults to timestamp + random. */ generateSessionId?: () => string + /** Session gap threshold used by the single post-import session-index build. */ + sessionGapThreshold?: number + /** Optional test/platform override for the session-index builder. */ + generateSessionIndex?: typeof generateCoreSessionIndex } // ==================== Core streaming import ==================== @@ -764,11 +772,14 @@ async function streamImportSingle( // Build session index (segment / message_context tables) const sessionIndexStartedAt = performance.now() + const sessionGapThreshold = normalizeSessionGapThreshold(deps.sessionGapThreshold) try { - generateSessionIndex(db) + const generateSessionIndex = deps.generateSessionIndex ?? generateCoreSessionIndex + generateSessionIndex(db, sessionGapThreshold) + logger?.perfDetail(`[SessionIndex] Built with gap threshold ${sessionGapThreshold}s`) logger?.perf('Session index built', totalMessageCount) - } catch { - /* non-fatal */ + } catch (error) { + logger?.error('Session index build failed (non-fatal)', error instanceof Error ? error : undefined) } finally { timings.sessionIndexMs = elapsedMs(sessionIndexStartedAt) sampleRss() diff --git a/packages/node-runtime/src/services/push-importer.test.ts b/packages/node-runtime/src/services/push-importer.test.ts index 78ab0a37..fbf49742 100644 --- a/packages/node-runtime/src/services/push-importer.test.ts +++ b/packages/node-runtime/src/services/push-importer.test.ts @@ -3,6 +3,7 @@ import fs from 'node:fs' import os from 'node:os' import path from 'node:path' import test from 'node:test' +import { generateSessionIndex } from '@openchatlab/core' import { DatabaseManager } from '../database-manager' import { raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta, type RuntimeIdentity } from '../data-dir-compat' import { withDataDirImportLock } from '../import/import-lock' @@ -142,6 +143,47 @@ test('analyzes incremental push deduplication without writing', async (t) => { assert.equal(outcome.result.batch.duplicateCount, analysis.result.duplicateCount) }) +test('continues incremental push indexing with the stored gap threshold', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const manager = createDatabaseManager(tempDir) + const sessionId = 'incremental-index-threshold' + const initialOutcome = await pushImport(manager, sessionId, { + chatlab: { version: '0.0.2', exportedAt: 100 }, + meta: { name: 'Incremental Threshold', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: [{ sender: 'wxid_alice', timestamp: 100, type: 0, content: 'before' }], + }) + assert.equal(initialOutcome.ok, true) + + const writeDb = manager.openRawSessionDatabase(sessionId) + try { + generateSessionIndex(writeDb, 60) + } finally { + writeDb.close() + } + + const incrementalOutcome = await pushImport(manager, sessionId, { + messages: [{ sender: 'wxid_alice', timestamp: 200, type: 0, content: 'after' }], + }) + assert.equal(incrementalOutcome.ok, true) + + const readDb = manager.openRawSessionDatabase(sessionId, { readonly: true }) + try { + const counts = readDb + .prepare( + `SELECT + (SELECT COUNT(*) FROM segment) AS segments, + (SELECT session_gap_threshold FROM meta LIMIT 1) AS gapThreshold` + ) + .get() as { segments: number; gapThreshold: number | null } + assert.deepEqual(counts, { segments: 2, gapThreshold: 60 }) + } finally { + readDb.close() + } +}) + test('rejects push imports while any writer holds the data-directory import lock', async (t) => { const root = makeTempDir() t.after(() => fs.rmSync(root, { recursive: true, force: true })) diff --git a/packages/node-runtime/src/services/push-importer.ts b/packages/node-runtime/src/services/push-importer.ts index b83db53c..91125507 100644 --- a/packages/node-runtime/src/services/push-importer.ts +++ b/packages/node-runtime/src/services/push-importer.ts @@ -7,7 +7,12 @@ import * as fs from 'fs' import { DataDirCompatibilityError } from '../data-dir-compat' -import { CHAT_DB_SCHEMA, generateSessionIndex, generateIncrementalSessionIndex } from '@openchatlab/core' +import { + CHAT_DB_SCHEMA, + generateSessionIndex, + generateIncrementalSessionIndex, + getSessionIndexStats, +} from '@openchatlab/core' import type { DatabaseAdapter } from '@openchatlab/core' import type { DatabaseManager } from '../database-manager' import { buildFtsIndex, insertFtsEntries } from '../fts' @@ -510,12 +515,13 @@ function writeIncrementalImport(db: DatabaseAdapter, payload: PushImportPayload) if (writtenCount > 0) { try { + const sessionGapThreshold = getSessionIndexStats(db).gapThreshold // Use minWrittenTs (not payload min) to avoid false-positive backfill detection // when overlap duplicates in the batch have older timestamps than written rows. if (minWrittenTs < preWriteMaxTs) { - generateSessionIndex(db) + generateSessionIndex(db, sessionGapThreshold) } else { - generateIncrementalSessionIndex(db) + generateIncrementalSessionIndex(db, sessionGapThreshold) } } catch { /* non-fatal */ diff --git a/packages/web-runtime/src/import/session-runtime.test.ts b/packages/web-runtime/src/import/session-runtime.test.ts index 187cd8a7..81b29024 100644 --- a/packages/web-runtime/src/import/session-runtime.test.ts +++ b/packages/web-runtime/src/import/session-runtime.test.ts @@ -228,6 +228,32 @@ describe('BrowserSessionRuntime', () => { database.dispose() }) + it('builds the imported session index with the requested gap threshold', async () => { + const sqlite3 = await sqlite3InitModule() + const database = new MemoryWorkspaceDatabase(sqlite3) + const runtime = new BrowserSessionRuntime(database, { + createSessionId: () => 'threshold-session', + now: () => 100, + }) + + try { + await runtime.importSource(source('threshold.json', chat('Threshold', 'alice', 1)), { + formatId: 'chatlab', + sessionGapThreshold: 60, + }) + + const db = database.getDatabase(sessionDatabaseFilename('threshold-session')) + assert.ok(db) + const meta = db.prepare('SELECT session_gap_threshold FROM meta LIMIT 1').get() as { + session_gap_threshold: number | null + } + assert.equal(meta.session_gap_threshold, 60) + assert.equal((db.prepare('SELECT COUNT(*) AS count FROM segment').get() as { count: number }).count, 1) + } finally { + database.dispose() + } + }) + it('canonicalizes reserved SYSTEM senders and excludes them from catalog member counts', async () => { const sqlite3 = await sqlite3InitModule() const database = new MemoryWorkspaceDatabase(sqlite3) diff --git a/packages/web-runtime/src/import/session-runtime.ts b/packages/web-runtime/src/import/session-runtime.ts index 0f03985a..72de4d29 100644 --- a/packages/web-runtime/src/import/session-runtime.ts +++ b/packages/web-runtime/src/import/session-runtime.ts @@ -2,6 +2,7 @@ import { CHAT_DB_INDEXES, CHAT_DB_TABLES, generateSessionIndex, + normalizeSessionGapThreshold, getBrowserWordFrequency as queryBrowserWordFrequency, getClusterGraph as queryClusterGraph, getLanguagePreferenceAnalysis as queryLanguagePreferenceAnalysis, @@ -72,6 +73,7 @@ export interface BrowserImportProgress { export interface BrowserSessionImportOptions { formatId?: BrowserImportFormatId chatIndex?: number + sessionGapThreshold?: number signal?: AbortSignal checkCancelled?: () => void onDatabaseStage?: (stage: WorkspaceDatabaseStage) => void @@ -247,6 +249,7 @@ export class BrowserSessionRuntime { const stats = await this.database.withDatabase(filename, CHAT_DB_TABLES, (db) => { const result = writeParseResultToDb(db, parsed.meta, members, messages) db.exec(CHAT_DB_INDEXES) + generateSessionIndex(db, normalizeSessionGapThreshold(options.sessionGapThreshold)) return { ...result, memberCount, diff --git a/packages/web-runtime/src/rpc/protocol.ts b/packages/web-runtime/src/rpc/protocol.ts index aa719088..639ee5fb 100644 --- a/packages/web-runtime/src/rpc/protocol.ts +++ b/packages/web-runtime/src/rpc/protocol.ts @@ -81,7 +81,12 @@ export interface WebRuntimeTaskMap { result: BrowserMultiChatEntry[] } 'import.start': { - payload: { source: BrowserParseSource; formatId?: BrowserImportFormatId; chatIndex?: number } + payload: { + source: BrowserParseSource + formatId?: BrowserImportFormatId + chatIndex?: number + sessionGapThreshold?: number + } result: BrowserSessionImportResult } 'session.list': { diff --git a/packages/web-runtime/src/worker/controller.ts b/packages/web-runtime/src/worker/controller.ts index 7c2acb1e..2e35475c 100644 --- a/packages/web-runtime/src/worker/controller.ts +++ b/packages/web-runtime/src/worker/controller.ts @@ -215,11 +215,13 @@ export class WebRuntimeWorkerController { size: request.payload.source.size, requestedFormat: request.payload.formatId ?? 'auto', chatIndex: request.payload.chatIndex, + sessionGapThreshold: request.payload.sessionGapThreshold, }, }) const result = await this.sessionRuntime.importSource(request.payload.source, { formatId: request.payload.formatId, chatIndex: request.payload.chatIndex, + sessionGapThreshold: request.payload.sessionGapThreshold, signal, checkCancelled: () => this.throwIfCancelled(request.id), onDatabaseStage: (stage) => this.handleDatabaseStage(request, stage), diff --git a/src/components/common/Settings/BatchManageTab.vue b/src/components/common/Settings/BatchManageTab.vue index fd323d26..acacc1a6 100644 --- a/src/components/common/Settings/BatchManageTab.vue +++ b/src/components/common/Settings/BatchManageTab.vue @@ -4,6 +4,7 @@ import { useI18n } from 'vue-i18n' import { storeToRefs } from 'pinia' import { useVirtualizer } from '@tanstack/vue-virtual' import { useToast } from '@/composables/useToast' +import { getSessionGapThreshold } from '@/composables/useUiConfig' import { useTableRowSelection, useTableSort, type TableSortDirection } from '@/composables/useTable' import { useSessionStore } from '@/stores/session' import type { AnalysisSession } from '@/types/base' @@ -455,6 +456,7 @@ async function executeMerge() { resolution: 'keep1' as const, })), andAnalyze: true, // 直接导入分析 + sessionGapThreshold: getSessionGapThreshold(), }) if (!mergeResult.success) { diff --git a/src/components/home/DemoImportButton.vue b/src/components/home/DemoImportButton.vue index f2f2c20d..5137e5e9 100644 --- a/src/components/home/DemoImportButton.vue +++ b/src/components/home/DemoImportButton.vue @@ -5,17 +5,13 @@ import { useI18n } from 'vue-i18n' import { useSessionStore } from '@/stores/session' import { useSettingsStore } from '@/stores/settings' import { getChatlabSiteLocalePath } from '@/utils/chatlabSiteLocale' -import { useDataService, useImportService, useSessionIndexService } from '@/services' +import { useDataService, useImportService } from '@/services' import { getSessionGapThreshold } from '@/composables/useUiConfig' const { t } = useI18n() const router = useRouter() const sessionStore = useSessionStore() const settingsStore = useSettingsStore() -const props = withDefaults(defineProps<{ generateSessionIndexes?: boolean }>(), { - generateSessionIndexes: true, -}) - const isImporting = ref(false) const stage = ref<'downloading' | 'importing'>('downloading') const error = ref(null) @@ -31,9 +27,13 @@ async function navigateToSession(sessionId: string) { async function handleImportViaService() { // Demo assets are available in Chinese and English; other interface locales use the English version. const demoLocale = getChatlabSiteLocalePath(settingsStore.locale) === 'cn' ? 'cn' : 'en' - return useImportService().importDemo(demoLocale, (progress) => { - stage.value = progress.stage - }) + return useImportService().importDemo( + demoLocale, + (progress) => { + stage.value = progress.stage + }, + { sessionGapThreshold: getSessionGapThreshold() } + ) } async function handleImport() { @@ -48,19 +48,6 @@ async function handleImport() { await sessionStore.loadSessions() sessionStore.selectSession(result.groupSessionId) - if (props.generateSessionIndexes) { - try { - const gapThreshold = getSessionGapThreshold() - const sessionIndexService = useSessionIndexService() - await sessionIndexService.generate(result.groupSessionId, gapThreshold) - if (result.privateSessionIds?.length) { - await Promise.all(result.privateSessionIds.map((id) => sessionIndexService.generate(id, gapThreshold))) - } - } catch (e) { - console.error('自动生成会话索引失败:', e) - } - } - await navigateToSession(result.groupSessionId) } else { error.value = result.error || t('home.demo.failed') diff --git a/src/components/import/ImportArea.vue b/src/components/import/ImportArea.vue index d4abeab2..c62ae618 100644 --- a/src/components/import/ImportArea.vue +++ b/src/components/import/ImportArea.vue @@ -12,9 +12,9 @@ import { useSessionStore, type BatchFileInfo, type MergeFileInfo } from '@/store import { useDataService, useImportService, - useSessionIndexService, usePlatformService, type PreparedImportSource, + type ImportOptions, type ImportResult, } from '@/services' import { IS_ELECTRON } from '@/utils/platform' @@ -59,14 +59,8 @@ const preparedImportSource = ref(null) const showFormatSelector = ref(false) const formatSelectorFilePath = ref('') -async function autoGenerateSessionIndex(sessionId: string, importMode?: ImportResult['importMode']) { - if (!props.backendFeatures || importMode !== 'created') return - try { - const gapThreshold = getSessionGapThreshold() - await useSessionIndexService().generate(sessionId, gapThreshold) - } catch (error) { - console.error('自动生成会话索引失败:', error) - } +function withSessionGapThreshold(options: ImportOptions = {}): ImportOptions { + return { ...options, sessionGapThreshold: getSessionGapThreshold() } } function showImportOutcome(result: ImportResult) { @@ -313,12 +307,12 @@ async function processWebFiles(files: File[]) { const pendingWebFile = ref(null) -async function importSingleWebFile(file: File, options?: { formatId?: string; chatIndex?: number }) { +async function importSingleWebFile(file: File, options?: ImportOptions) { isImporting.value = true importProgress.value = { stage: 'detecting', progress: 0, message: '' } try { - const result = await useImportService().importFile(file, options, (p) => { + const result = await useImportService().importFile(file, withSessionGapThreshold(options), (p) => { if (p.stage === 'done') return importProgress.value = p }) @@ -387,7 +381,7 @@ async function importDirectory(source: File[] | string) { importProgress.value = { stage: 'detecting', progress: 0, message: '' } try { - const result = await useImportService().importDirectory(source, undefined, (p) => { + const result = await useImportService().importDirectory(source, withSessionGapThreshold(), (p) => { if (p.stage === 'done') return importProgress.value = p }) @@ -500,7 +494,7 @@ async function handleFormatSelect(formatId: string) { importProgress.value = { stage: 'detecting', progress: 0, message: '' } try { - const result = await useImportService().importFile(filePath, { formatId }, (progress) => { + const result = await useImportService().importFile(filePath, withSessionGapThreshold({ formatId }), (progress) => { if (progress.stage === 'done') return importProgress.value = progress }) @@ -513,7 +507,6 @@ async function handleFormatSelect(formatId: string) { if (result.success && result.sessionId) { await sessionStore.loadSessions() sessionStore.selectSession(result.sessionId) - await autoGenerateSessionIndex(result.sessionId, result.importMode) showImportOutcome(result) await navigateToSession(result.sessionId) } else { @@ -547,12 +540,12 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { selectedChats[0].chatId, (progress) => { if (progress.stage !== 'done') importProgress.value = progress - } + }, + withSessionGapThreshold() ) if (result.success && result.sessionId) { await sessionStore.loadSessions() sessionStore.selectSession(result.sessionId) - await autoGenerateSessionIndex(result.sessionId, result.importMode) showImportOutcome(result) await navigateToSession(result.sessionId) } else { @@ -591,12 +584,14 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { }, importChat: async (sourceId, chatId) => { const index = selectedChats.findIndex((chat) => chat.chatId === chatId) - const result = await useImportService().importPreparedChat(sourceId, chatId, (progress) => { - if (progress.stage !== 'done' && index >= 0) batchFiles.value[index].progress = progress - }) - if (result.success && result.sessionId) { - await autoGenerateSessionIndex(result.sessionId, result.importMode) - } + const result = await useImportService().importPreparedChat( + sourceId, + chatId, + (progress) => { + if (progress.stage !== 'done' && index >= 0) batchFiles.value[index].progress = progress + }, + withSessionGapThreshold() + ) return result }, onItemComplete: (_chat, index, result) => { @@ -655,7 +650,7 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { try { const result = await useImportService().importFile( filePath, - { chatIndex: selectedChats[0].index! }, + withSessionGapThreshold({ chatIndex: selectedChats[0].index! }), (progress) => { if (progress.stage === 'done') return importProgress.value = progress @@ -670,7 +665,6 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { if (result.success && result.sessionId) { await sessionStore.loadSessions() sessionStore.selectSession(result.sessionId) - await autoGenerateSessionIndex(result.sessionId, result.importMode) showImportOutcome(result) await navigateToSession(result.sessionId) } else { @@ -700,10 +694,14 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { batchFiles.value[i].status = 'importing' try { - const result = await useImportService().importFile(filePath, { chatIndex: chat.index! }, (progress) => { - if (progress.stage === 'done') return - batchFiles.value[i].progress = progress - }) + const result = await useImportService().importFile( + filePath, + withSessionGapThreshold({ chatIndex: chat.index! }), + (progress) => { + if (progress.stage === 'done') return + batchFiles.value[i].progress = progress + } + ) if (result.success && result.sessionId) { batchFiles.value[i].status = 'success' @@ -714,7 +712,6 @@ async function handleChatSelect(selectedChats: ChatInfo[]) { batchFiles.value[i].newMessageCount = result.newMessageCount batchFiles.value[i].duplicateCount = result.duplicateCount successCount++ - await autoGenerateSessionIndex(result.sessionId, result.importMode) } else { batchFiles.value[i].status = 'failed' batchFiles.value[i].error = result.error diff --git a/src/services/import/browser.test.ts b/src/services/import/browser.test.ts index 5be10be4..49fe5f12 100644 --- a/src/services/import/browser.test.ts +++ b/src/services/import/browser.test.ts @@ -385,9 +385,19 @@ describe('BrowserImportAdapter', () => { const file = createFile('result.json') assert.equal((await adapter.scanMultiChatFile(file)).length, 2) - assert.equal((await adapter.importFile(file, { formatId: 'telegram-native', chatIndex: 1 })).success, true) + assert.equal( + ( + await adapter.importFile(file, { + formatId: 'telegram-native', + chatIndex: 1, + sessionGapThreshold: 7200, + }) + ).success, + true + ) assert.equal(requests[0].type, 'import.scanChats') assert.equal(requests[1].type, 'import.start') assert.equal((requests[1].payload as WebRuntimeTaskPayload<'import.start'>).chatIndex, 1) + assert.equal((requests[1].payload as WebRuntimeTaskPayload<'import.start'>).sessionGapThreshold, 7200) }) }) diff --git a/src/services/import/browser.ts b/src/services/import/browser.ts index cc13c5a1..f159f71b 100644 --- a/src/services/import/browser.ts +++ b/src/services/import/browser.ts @@ -46,7 +46,14 @@ export class BrowserImportAdapter implements ImportAdapter { const controller = new AbortController() this.activeImport = controller try { - return await this.importBrowserFile(file, formatId, options?.chatIndex, controller.signal, onProgress) + return await this.importBrowserFile( + file, + formatId, + options?.chatIndex, + options?.sessionGapThreshold, + controller.signal, + onProgress + ) } catch (error) { return { success: false, error: error instanceof Error ? error.message : String(error) } } finally { @@ -81,7 +88,8 @@ export class BrowserImportAdapter implements ImportAdapter { importPreparedChat( _sourceId: string, _chatId: string, - _onProgress?: (progress: ImportProgress) => void + _onProgress?: (progress: ImportProgress) => void, + _options?: ImportOptions ): Promise { return unsupported('Prepared import sources') } @@ -90,7 +98,11 @@ export class BrowserImportAdapter implements ImportAdapter { return unsupported('Prepared import sources') } - async importDemo(locale: string, onProgress?: (progress: DemoProgress) => void): Promise { + async importDemo( + locale: string, + onProgress?: (progress: DemoProgress) => void, + options?: ImportOptions + ): Promise { if (this.activeImport) return { success: false, error: 'Another Web WASM import is already running' } const controller = new AbortController() @@ -113,7 +125,13 @@ export class BrowserImportAdapter implements ImportAdapter { for (const file of files) { onProgress?.({ stage: 'importing' }) - const result = await this.importBrowserFile(file, 'chatlab', undefined, controller.signal) + const result = await this.importBrowserFile( + file, + 'chatlab', + undefined, + options?.sessionGapThreshold, + controller.signal + ) if (!result.success || !result.sessionId) { throw new Error(result.error || `Failed to import demo: ${file.name}`) } @@ -201,12 +219,13 @@ export class BrowserImportAdapter implements ImportAdapter { file: File, formatId: BrowserImportFormatId | undefined, chatIndex: number | undefined, + sessionGapThreshold: number | undefined, signal: AbortSignal, onProgress?: (progress: ImportProgress) => void ): Promise { const result = await this.rpc.request( 'import.start', - { source: file as BrowserParseSource, formatId, chatIndex }, + { source: file as BrowserParseSource, formatId, chatIndex, sessionGapThreshold }, { signal, onProgress: (progress) => onProgress?.(mapImportProgress(progress)), diff --git a/src/services/import/electron.ts b/src/services/import/electron.ts index d87c9629..d12ec1ed 100644 --- a/src/services/import/electron.ts +++ b/src/services/import/electron.ts @@ -46,7 +46,7 @@ export class ElectronImportAdapter implements ImportAdapter { }) const importPromise = - options && (options.formatId || options.chatIndex !== undefined) + options && (options.formatId || options.chatIndex !== undefined || options.sessionGapThreshold !== undefined) ? window.chatApi.importWithOptions(filePath, options as Record) : window.chatApi.import(filePath) @@ -100,7 +100,8 @@ export class ElectronImportAdapter implements ImportAdapter { async importPreparedChat( sourceId: string, chatId: string, - onProgress?: (p: ImportProgress) => void + onProgress?: (p: ImportProgress) => void, + options?: ImportOptions ): Promise { const unlisten = window.chatApi.onImportProgress((progress: any) => { onProgress?.({ @@ -113,7 +114,11 @@ export class ElectronImportAdapter implements ImportAdapter { }) }) try { - return normalizeImportResult(await window.chatApi.importPreparedChat(sourceId, chatId)) + return normalizeImportResult( + await (options + ? window.chatApi.importPreparedChat(sourceId, chatId, options as Record) + : window.chatApi.importPreparedChat(sourceId, chatId)) + ) } finally { unlisten() } @@ -127,7 +132,11 @@ export class ElectronImportAdapter implements ImportAdapter { return window.chatApi.getSupportedFormats() } - async importDemo(locale: string, onProgress?: (p: DemoProgress) => void): Promise { + async importDemo( + locale: string, + onProgress?: (p: DemoProgress) => void, + options?: ImportOptions + ): Promise { const unlisten = window.chatApi.onDemoProgress((progress: any) => { if (progress.stage === 'downloading' || progress.stage === 'importing') { onProgress?.({ stage: progress.stage }) @@ -135,7 +144,9 @@ export class ElectronImportAdapter implements ImportAdapter { }) try { - const result = await window.chatApi.importDemo(locale) + const result = await (options + ? window.chatApi.importDemo(locale, options as Record) + : window.chatApi.importDemo(locale)) return result } finally { unlisten() @@ -170,7 +181,7 @@ export class ElectronImportAdapter implements ImportAdapter { async importDirectory( source: File[] | string, - _options?: ImportOptions, + options?: ImportOptions, onProgress?: (p: ImportProgress) => void ): Promise { if (typeof source !== 'string') { @@ -189,8 +200,11 @@ export class ElectronImportAdapter implements ImportAdapter { }) }) - window.chatApi - .importDirectory(source) + const importPromise = options + ? window.chatApi.importDirectory(source, options as Record) + : window.chatApi.importDirectory(source) + + importPromise .then((result) => { unlisten() resolve( diff --git a/src/services/import/fetch.ts b/src/services/import/fetch.ts index d5ef44d3..4bf60003 100644 --- a/src/services/import/fetch.ts +++ b/src/services/import/fetch.ts @@ -69,6 +69,9 @@ export class FetchImportAdapter implements ImportAdapter { form.append('file', file) if (options?.formatId) form.append('formatId', options.formatId) if (options?.chatIndex !== undefined) form.append('chatIndex', String(options.chatIndex)) + if (options?.sessionGapThreshold !== undefined) { + form.append('sessionGapThreshold', String(options.sessionGapThreshold)) + } const res = await fetchWithAuth(`${getBaseUrl()}/import`, { method: 'POST', body: form }) @@ -120,12 +123,13 @@ export class FetchImportAdapter implements ImportAdapter { async importPreparedChat( sourceId: string, chatId: string, - onProgress?: (p: ImportProgress) => void + onProgress?: (p: ImportProgress) => void, + options?: ImportOptions ): Promise { const res = await fetchWithAuth(`${getBaseUrl()}/import-sources/${encodeURIComponent(sourceId)}/import`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ chatId }), + body: JSON.stringify({ chatId, sessionGapThreshold: options?.sessionGapThreshold }), }) if (!res.ok) { const text = await res.text() @@ -146,7 +150,11 @@ export class FetchImportAdapter implements ImportAdapter { return get('/supported-formats') } - async importDemo(locale: string, onProgress?: (p: DemoProgress) => void): Promise { + async importDemo( + locale: string, + onProgress?: (p: DemoProgress) => void, + options?: ImportOptions + ): Promise { return new Promise((resolve) => { fetchWithAuth(`${getBaseUrl()}/demo/import`, { method: 'POST', @@ -154,6 +162,7 @@ export class FetchImportAdapter implements ImportAdapter { body: JSON.stringify({ locale, timeZone: Intl.DateTimeFormat().resolvedOptions().timeZone, + sessionGapThreshold: options?.sessionGapThreshold, }), }) .then(async (resp) => { @@ -250,7 +259,7 @@ export class FetchImportAdapter implements ImportAdapter { async importDirectory( source: File[] | string, - _options?: ImportOptions, + options?: ImportOptions, onProgress?: (p: ImportProgress) => void ): Promise { if (typeof source === 'string') { @@ -266,6 +275,9 @@ export class FetchImportAdapter implements ImportAdapter { form.append('files', file) form.append('relativePaths', file.webkitRelativePath || file.name) } + if (options?.sessionGapThreshold !== undefined) { + form.append('sessionGapThreshold', String(options.sessionGapThreshold)) + } const res = await fetchWithAuth(`${getBaseUrl()}/import-directory`, { method: 'POST', body: form }) diff --git a/src/services/import/importSourceAdapters.test.ts b/src/services/import/importSourceAdapters.test.ts index 74fc083e..6bf360e7 100644 --- a/src/services/import/importSourceAdapters.test.ts +++ b/src/services/import/importSourceAdapters.test.ts @@ -17,6 +17,29 @@ afterEach(() => { }) describe('archive import source adapters', () => { + it('forwards the session gap threshold through Electron file import', async () => { + const calls: unknown[][] = [] + Object.defineProperty(globalThis, 'window', { + configurable: true, + value: { + chatApi: { + importWithOptions: async (...args: unknown[]) => { + calls.push(args) + return { success: true, sessionId: 'session-1' } + }, + onImportProgress: () => () => {}, + }, + }, + }) + + const result = await new ElectronImportAdapter().importFile('/tmp/chat.json', { + sessionGapThreshold: 7200, + }) + + assert.equal(result.sessionId, 'session-1') + assert.deepEqual(calls, [['/tmp/chat.json', { sessionGapThreshold: 7200 }]]) + }) + it('forwards Electron source lifecycle calls through preload', async () => { const calls: unknown[][] = [] Object.defineProperty(globalThis, 'window', { @@ -51,12 +74,19 @@ describe('archive import source adapters', () => { const adapter = new ElectronImportAdapter() assert.equal((await adapter.prepareImportSource('/tmp/takeout.zip')).source?.sourceId, 'source-1') - assert.equal((await adapter.importPreparedChat('source-1', 'Groups/DM sample')).sessionId, 'session-1') + assert.equal( + ( + await adapter.importPreparedChat('source-1', 'Groups/DM sample', undefined, { + sessionGapThreshold: 7200, + }) + ).sessionId, + 'session-1' + ) await adapter.releaseImportSource('source-1') assert.deepEqual(calls, [ ['prepare', '/tmp/takeout.zip'], - ['import', 'source-1', 'Groups/DM sample'], + ['import', 'source-1', 'Groups/DM sample', { sessionGapThreshold: 7200 }], ['release', 'source-1'], ]) }) @@ -91,7 +121,14 @@ describe('archive import source adapters', () => { const adapter = new FetchImportAdapter() const file = new File(['zip'], 'takeout.zip', { type: 'application/zip' }) assert.equal((await adapter.prepareImportSource(file)).source?.sourceId, 'source-2') - assert.equal((await adapter.importPreparedChat('source-2', 'Groups/DM sample')).sessionId, 'session-2') + assert.equal( + ( + await adapter.importPreparedChat('source-2', 'Groups/DM sample', undefined, { + sessionGapThreshold: 7200, + }) + ).sessionId, + 'session-2' + ) await adapter.releaseImportSource('source-2') assert.equal(requests.length, 3) @@ -101,7 +138,7 @@ describe('archive import source adapters', () => { requests[1].init?.headers && new Headers(requests[1].init?.headers).get('Content-Type'), 'application/json' ) - assert.equal(requests[1].init?.body, JSON.stringify({ chatId: 'Groups/DM sample' })) + assert.equal(requests[1].init?.body, JSON.stringify({ chatId: 'Groups/DM sample', sessionGapThreshold: 7200 })) assert.equal(requests[2].init?.method, 'DELETE') }) @@ -115,12 +152,13 @@ describe('archive import source adapters', () => { }) as typeof fetch const adapter = new FetchImportAdapter() - const result = await adapter.importDemo('cn') + const result = await adapter.importDemo('cn', undefined, { sessionGapThreshold: 7200 }) assert.equal(result.success, true) assert.deepEqual(JSON.parse(requestBody ?? ''), { locale: 'cn', timeZone: Intl.DateTimeFormat().resolvedOptions().timeZone, + sessionGapThreshold: 7200, }) }) diff --git a/src/services/import/telemetry.ts b/src/services/import/telemetry.ts index 5e0df858..f4f45d63 100644 --- a/src/services/import/telemetry.ts +++ b/src/services/import/telemetry.ts @@ -152,10 +152,11 @@ export class TelemetryImportAdapter implements ImportAdapter { importPreparedChat( sourceId: string, chatId: string, - onProgress?: (progress: ImportProgress) => void + onProgress?: (progress: ImportProgress) => void, + options?: ImportOptions ): Promise { return this.runTrackedImport(this.sourcePlatforms.get(sourceId) ?? 'unknown', () => - this.delegate.importPreparedChat(sourceId, chatId, onProgress) + this.delegate.importPreparedChat(sourceId, chatId, onProgress, options) ) } @@ -171,8 +172,12 @@ export class TelemetryImportAdapter implements ImportAdapter { return this.delegate.getSupportedFormats() } - importDemo(locale: string, onProgress?: (progress: DemoProgress) => void): Promise { - return this.delegate.importDemo(locale, onProgress) + importDemo( + locale: string, + onProgress?: (progress: DemoProgress) => void, + options?: ImportOptions + ): Promise { + return this.delegate.importDemo(locale, onProgress, options) } async analyzeIncrementalImport(sessionId: string, file: File | string): Promise { diff --git a/src/services/import/types.ts b/src/services/import/types.ts index 002ff278..2aa08c45 100644 --- a/src/services/import/types.ts +++ b/src/services/import/types.ts @@ -12,6 +12,8 @@ import type { ImportProgress } from '@/types/base' export interface ImportOptions { formatId?: string chatIndex?: number + /** Session split threshold used when the created session index is built. */ + sessionGapThreshold?: number } // ==================== 导入结果 ==================== @@ -188,7 +190,12 @@ export interface ImportAdapter { prepareImportSource(file: File | string): Promise /** 从已注册导入源中导入一个聊天。 */ - importPreparedChat(sourceId: string, chatId: string, onProgress?: (p: ImportProgress) => void): Promise + importPreparedChat( + sourceId: string, + chatId: string, + onProgress?: (p: ImportProgress) => void, + options?: ImportOptions + ): Promise /** 释放导入源及其临时文件;重复调用应保持幂等。 */ releaseImportSource(sourceId: string): Promise @@ -197,7 +204,7 @@ export interface ImportAdapter { getSupportedFormats(): Promise /** 导入 Demo 数据 */ - importDemo(locale: string, onProgress?: (p: DemoProgress) => void): Promise + importDemo(locale: string, onProgress?: (p: DemoProgress) => void, options?: ImportOptions): Promise /** 分析增量导入(预览去重后可新增多少消息) */ analyzeIncrementalImport(sessionId: string, file: File | string): Promise diff --git a/src/services/registry.ts b/src/services/registry.ts index 40577616..3acb359e 100644 --- a/src/services/registry.ts +++ b/src/services/registry.ts @@ -265,6 +265,7 @@ function installMergeShims(platform: 'electron' | 'cli-web'): void { outputName: string outputFormat?: string andAnalyze?: boolean + sessionGapThreshold?: number }) => { try { const handles = params.filePaths.map(resolveHandle) @@ -276,6 +277,7 @@ function installMergeShims(platform: 'electron' | 'cli-web'): void { outputName: params.outputName, format: params.outputFormat || 'json', andImport: params.andAnalyze ?? false, + sessionGapThreshold: params.sessionGapThreshold, }), }) await ensureOk(resp, 'mergeFiles') diff --git a/src/stores/session.ts b/src/stores/session.ts index 87d226f9..e9576e23 100644 --- a/src/stores/session.ts +++ b/src/stores/session.ts @@ -1,7 +1,7 @@ import { defineStore } from 'pinia' import { ref, computed } from 'vue' import type { AnalysisSession, ImportProgress, ChatType } from '@/types/base' -import { useDataService, useImportService, useSessionIndexService, usePlatformService } from '@/services' +import { useDataService, useImportService, usePlatformService } from '@/services' import type { AutoImportCreateReason, AutoImportMatchMethod, AutoImportMode, ImportDiagnosticsInfo } from '@/services' import { IS_ELECTRON } from '@/utils/platform' @@ -265,11 +265,15 @@ export const useSessionStore = defineStore( isProcessing = false } - const importResult = await useImportService().importFile(filePath, undefined, (progress) => { - if (progress.stage === 'done') return - queue.push(progress) - processQueue() - }) + const importResult = await useImportService().importFile( + filePath, + { sessionGapThreshold: getSessionGapThreshold() }, + (progress) => { + if (progress.stage === 'done') return + queue.push(progress) + processQueue() + } + ) while (queue.length > 0 || isProcessing) { await new Promise((resolve) => setTimeout(resolve, 100)) @@ -289,14 +293,6 @@ export const useSessionStore = defineStore( await loadSessions() currentSessionId.value = importResult.sessionId - if (importResult.importMode === 'created') { - try { - await useSessionIndexService().generate(importResult.sessionId, getSessionGapThreshold()) - } catch (error) { - console.error('自动生成会话索引失败:', error) - } - } - await applyOwnerProfileAfterImport(importResult.sessionId) return { @@ -426,11 +422,15 @@ export const useSessionStore = defineStore( isProcessing = false } - const importResult = await useImportService().importFile(importSources[i].source, undefined, (progress) => { - if (progress.stage === 'done') return - queue.push(progress) - processQueue() - }) + const importResult = await useImportService().importFile( + importSources[i].source, + { sessionGapThreshold: getSessionGapThreshold() }, + (progress) => { + if (progress.stage === 'done') return + queue.push(progress) + processQueue() + } + ) // 等待进度队列处理完成(但如果已取消则快速跳过) let waitCount = 0 @@ -452,14 +452,6 @@ export const useSessionStore = defineStore( file.duplicateCount = importResult.duplicateCount successCount++ - // 即使取消了也要为已导入成功的文件生成会话索引 - if (importResult.importMode === 'created') { - try { - await useSessionIndexService().generate(importResult.sessionId, getSessionGapThreshold()) - } catch (error) { - console.error('自动生成会话索引失败:', error) - } - } await applyOwnerProfileAfterImport(importResult.sessionId) } else { file.status = 'failed' @@ -481,14 +473,6 @@ export const useSessionStore = defineStore( file.duplicateCount = importResult.duplicateCount successCount++ - // 自动生成会话索引(跳过如果已取消) - if (!batchImportCancelled.value && importResult.importMode === 'created') { - try { - await useSessionIndexService().generate(importResult.sessionId, getSessionGapThreshold()) - } catch (error) { - console.error('自动生成会话索引失败:', error) - } - } await applyOwnerProfileAfterImport(importResult.sessionId) } else { file.status = 'failed' @@ -604,6 +588,7 @@ export const useSessionStore = defineStore( outputName, conflictResolutions: [], // 默认 keepBoth(保留所有消息) andAnalyze: true, // 合并后创建会话 + sessionGapThreshold: getSessionGapThreshold(), }) if (!result.success) { @@ -625,15 +610,6 @@ export const useSessionStore = defineStore( // 刷新会话列表 await loadSessions() - // 自动生成会话索引 - if (result.sessionId) { - try { - await useSessionIndexService().generate(result.sessionId, getSessionGapThreshold()) - } catch (error) { - console.error('自动生成会话索引失败:', error) - } - } - return { success: true, sessionId: result.sessionId } } catch (err) { mergeStage.value = 'error' diff --git a/src/types/format.ts b/src/types/format.ts index 89d4a4d8..94ba83db 100644 --- a/src/types/format.ts +++ b/src/types/format.ts @@ -132,6 +132,7 @@ export interface MergeParams { outputFormat?: OutputFormat // 输出格式,默认 'json' conflictResolutions: ConflictResolution[] andAnalyze: boolean + sessionGapThreshold?: number } /**