From 3e6f0368262f8fdbe898cedbb5fded82011e6ac5 Mon Sep 17 00:00:00 2001 From: digua Date: Sun, 12 Jul 2026 16:44:58 +0800 Subject: [PATCH] feat(electron): enable automatic incremental imports --- apps/desktop/main/ipc/chat.ts | 104 +++++------------- apps/desktop/main/worker/dbWorker.ts | 2 + apps/desktop/main/worker/import/index.ts | 2 + .../main/worker/import/streamImport.ts | 37 ++++++- apps/desktop/main/worker/workerManager.ts | 34 +++++- .../main/worker/workerTimeoutPolicy.test.ts | 1 + apps/desktop/preload/apis/chat.ts | 31 ++++-- apps/desktop/preload/index.d.ts | 25 +++-- 8 files changed, 139 insertions(+), 97 deletions(-) diff --git a/apps/desktop/main/ipc/chat.ts b/apps/desktop/main/ipc/chat.ts index 7d6d536d..ef344ade 100644 --- a/apps/desktop/main/ipc/chat.ts +++ b/apps/desktop/main/ipc/chat.ts @@ -19,6 +19,26 @@ import { getArchiveImportSourceManager, importPreparedChatWithSource } from '../ export function registerChatHandlers(ctx: IpcContext): void { const { win } = ctx + const forwardImportProgress = (progress: ParseProgress) => { + win.webContents.send('chat:importProgress', { + stage: progress.stage, + progress: progress.percentage, + message: progress.message, + bytesRead: progress.bytesRead, + totalBytes: progress.totalBytes, + messagesProcessed: progress.messagesProcessed, + }) + } + + const finishAutoImport = (result: Awaited>) => { + if (result.success) { + if (result.importMode === 'incremental') win.webContents.send('api:importCompleted') + } else { + win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: result.error }) + } + return result + } + // ==================== 数据库迁移 ==================== ipcMain.handle('chat:checkMigration', async () => { @@ -127,28 +147,9 @@ export function registerChatHandlers(ctx: IpcContext): void { getArchiveImportSourceManager(), sourceId, chatId, - (manifestPath) => - worker.streamImport( - manifestPath, - (progress: ParseProgress) => { - win.webContents.send('chat:importProgress', { - stage: progress.stage, - progress: progress.percentage, - message: progress.message, - bytesRead: progress.bytesRead, - totalBytes: progress.totalBytes, - messagesProcessed: progress.messagesProcessed, - }) - }, - { formatId: 'google-chat-takeout' } - ) + (manifestPath) => worker.autoImport(manifestPath, forwardImportProgress, { formatId: 'google-chat-takeout' }) ) - return { - success: result.success, - sessionId: result.sessionId, - error: result.error, - diagnostics: result.diagnostics, - } + return finishAutoImport(result) } catch (error) { return { success: false, @@ -173,23 +174,8 @@ export function registerChatHandlers(ctx: IpcContext): void { try { win.webContents.send('chat:importProgress', { stage: 'detecting', progress: 5, message: '' }) - const result = await worker.streamImport(filePath, (progress: ParseProgress) => { - win.webContents.send('chat:importProgress', { - stage: progress.stage, - progress: progress.percentage, - message: progress.message, - bytesRead: progress.bytesRead, - totalBytes: progress.totalBytes, - messagesProcessed: progress.messagesProcessed, - }) - }) - - if (result.success) { - return { success: true, sessionId: result.sessionId, diagnostics: result.diagnostics } - } else { - win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: result.error }) - return { success: false, error: result.error, diagnostics: result.diagnostics } - } + const result = await worker.autoImport(filePath, forwardImportProgress) + return finishAutoImport(result) } catch (error) { win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: String(error) }) return { success: false, error: String(error) } @@ -203,23 +189,8 @@ export function registerChatHandlers(ctx: IpcContext): void { win.webContents.send('chat:importProgress', { stage: 'detecting', progress: 5, message: '' }) - const result = await worker.streamImport(entryPath, (progress: ParseProgress) => { - win.webContents.send('chat:importProgress', { - stage: progress.stage, - progress: progress.percentage, - message: progress.message, - bytesRead: progress.bytesRead, - totalBytes: progress.totalBytes, - messagesProcessed: progress.messagesProcessed, - }) - }) - - if (result.success) { - return { success: true, sessionId: result.sessionId, diagnostics: result.diagnostics } - } else { - win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: result.error }) - return { success: false, error: result.error, diagnostics: result.diagnostics } - } + const result = await worker.autoImport(entryPath, forwardImportProgress) + return finishAutoImport(result) } catch (error) { win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: String(error) }) return { success: false, error: String(error) } @@ -230,27 +201,8 @@ export function registerChatHandlers(ctx: IpcContext): void { try { win.webContents.send('chat:importProgress', { stage: 'detecting', progress: 5, message: '' }) - const result = await worker.streamImport( - filePath, - (progress: ParseProgress) => { - win.webContents.send('chat:importProgress', { - stage: progress.stage, - progress: progress.percentage, - message: progress.message, - bytesRead: progress.bytesRead, - totalBytes: progress.totalBytes, - messagesProcessed: progress.messagesProcessed, - }) - }, - formatOptions - ) - - if (result.success) { - return { success: true, sessionId: result.sessionId, diagnostics: result.diagnostics } - } else { - win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: result.error }) - return { success: false, error: result.error, diagnostics: result.diagnostics } - } + const result = await worker.autoImport(filePath, forwardImportProgress, formatOptions) + return finishAutoImport(result) } catch (error) { win.webContents.send('chat:importProgress', { stage: 'error', progress: 0, message: String(error) }) return { success: false, error: String(error) } diff --git a/apps/desktop/main/worker/dbWorker.ts b/apps/desktop/main/worker/dbWorker.ts index cbbff4aa..730f19be 100644 --- a/apps/desktop/main/worker/dbWorker.ts +++ b/apps/desktop/main/worker/dbWorker.ts @@ -76,6 +76,7 @@ import { } from './query' import { streamImport, + autoImport, streamParseFileInfo, analyzeIncrementalImport, incrementalImport, @@ -246,6 +247,7 @@ 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), // 流式解析文件信息（用于合并预览） streamParseFileInfo: (p, id) => streamParseFileInfo(p.filePath, id), // 增量导入 diff --git a/apps/desktop/main/worker/import/index.ts b/apps/desktop/main/worker/import/index.ts index b266133a..f0ab1a3d 100644 --- a/apps/desktop/main/worker/import/index.ts +++ b/apps/desktop/main/worker/import/index.ts @@ -6,9 +6,11 @@ // 流式导入（核心导入功能） export { streamImport, + autoImport, streamParseFileInfo, analyzeNewImport, type StreamImportResult, + type AutoImportResult, type StreamParseFileInfoResult, type AnalyzeNewImportResult, } from './streamImport' diff --git a/apps/desktop/main/worker/import/streamImport.ts b/apps/desktop/main/worker/import/streamImport.ts index 67de6d26..7225a70a 100644 --- a/apps/desktop/main/worker/import/streamImport.ts +++ b/apps/desktop/main/worker/import/streamImport.ts @@ -11,6 +11,7 @@ import * as path from 'path' import type Database from 'better-sqlite3' import { BetterSqliteAdapter, + autoImportFile as sharedAutoImportFile, streamingImport, analyzeNewImport as sharedAnalyzeNewImport, streamParseFileInfo as sharedStreamParseFileInfo, @@ -18,10 +19,12 @@ import { computeAndSetOverviewCache, deleteSessionCache, } from '@openchatlab/node-runtime' -import type { StreamImportDeps, StreamImportResult, ImportLogger } from '@openchatlab/node-runtime' +import type { AutoImportResult, StreamImportDeps, StreamImportResult, ImportLogger } from '@openchatlab/node-runtime' import { sendProgress, generateSessionId, getDbPath, createDatabaseWithoutIndexes } from './utils' +import { incrementalImport } from './incrementalImport' import { getCacheDir, + getDbDir, getTempDir, openRawDatabase, initPerfLog, @@ -35,6 +38,7 @@ import { } from '../core' export type { StreamImportResult } +export type { AutoImportResult } export type { AnalyzeNewImportResult, StreamParseFileInfoResult } from '@openchatlab/node-runtime' export type { SkipReasons, ImportDiagnostics } from '@openchatlab/node-runtime' @@ -118,6 +122,37 @@ export async function streamImport( return streamingImport(filePath, buildStreamImportDeps(requestId), formatOptions, externalSessionId) } +export async function autoImport( + filePath: string, + requestId: string, + formatOptions?: Record, + explicitSessionId?: string +): Promise { + return sharedAutoImportFile( + filePath, + { + listSessionIds: () => + fs.existsSync(getDbDir()) + ? fs + .readdirSync(getDbDir()) + .filter((name) => name.endsWith('.db')) + .map((name) => name.slice(0, -3)) + : [], + openReadonly: (sessionId) => new BetterSqliteAdapter(openRawDatabase(getDbPath(sessionId), { readonly: true })), + onProgress: (progress) => sendProgress(requestId, progress), + sessionExists: (sessionId) => fs.existsSync(getDbPath(sessionId)), + createSession: (sourcePath, sourceFormatOptions, sessionId) => + streamImport(sourcePath, requestId, sourceFormatOptions, sessionId), + appendSession: (sessionId, sourcePath, sourceFormatOptions) => + incrementalImport(sessionId, sourcePath, requestId, { + formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, + chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, + }), + }, + { explicitSessionId, formatOptions } + ) +} + /** * Dry-run analysis: parse without writing to DB. */ diff --git a/apps/desktop/main/worker/workerManager.ts b/apps/desktop/main/worker/workerManager.ts index 1744e023..760af638 100644 --- a/apps/desktop/main/worker/workerManager.ts +++ b/apps/desktop/main/worker/workerManager.ts @@ -8,7 +8,7 @@ import { app } from 'electron' import * as path from 'path' import * as fs from 'fs' import type { ParseProgress } from '../parser' -import type { StreamImportResult } from './import' +import type { AutoImportResult, StreamImportResult } from './import' import { getDatabaseDir, getCacheDir, getTempDir, getLogsDir, ensureDir } from '../paths' import { getNlpDir } from '../nlp/dictManager' @@ -551,6 +551,38 @@ export async function streamImport( return result } +export async function autoImport( + filePath: string, + onProgress?: (progress: ParseProgress) => void, + formatOptions?: Record, + explicitSessionId?: string +): Promise { + assertDataDirCompatibleNow() + + const result = await sendToWorkerWithProgress( + 'autoImport', + { filePath, formatOptions, explicitSessionId }, + onProgress + ) + if (!result.success || !result.sessionId || result.importMode !== 'created') return result + + try { + raiseChatDbCompatibilityGate(getPathProvider(), { + version: getDesktopAppVersion(app.getVersion()), + kind: 'desktop', + }) + } catch (error) { + deleteImportedSessionFiles(result.sessionId) + return { + success: false, + error: error instanceof Error ? error.message : String(error), + diagnostics: result.diagnostics, + } + } + + return result +} + function deleteImportedSessionFiles(sessionId: string): void { const dbPath = path.join(getDbDir(), `${sessionId}.db`) for (const suffix of ['', '-wal', '-shm']) { diff --git a/apps/desktop/main/worker/workerTimeoutPolicy.test.ts b/apps/desktop/main/worker/workerTimeoutPolicy.test.ts index c9aea90c..e742ff81 100644 --- a/apps/desktop/main/worker/workerTimeoutPolicy.test.ts +++ b/apps/desktop/main/worker/workerTimeoutPolicy.test.ts @@ -20,6 +20,7 @@ describe('worker timeout policy', () => { assert.equal(isRestartableReadOnlyRequestType('mergeMembers'), false) assert.equal(isRestartableReadOnlyRequestType('deleteMember'), false) assert.equal(isRestartableReadOnlyRequestType('streamImport'), false) + assert.equal(isRestartableReadOnlyRequestType('autoImport'), false) assert.equal(isRestartableReadOnlyRequestType('incrementalImport'), false) assert.equal(isRestartableReadOnlyRequestType('exportFilterResultToFile'), false) assert.equal(isRestartableReadOnlyRequestType('unknownFutureWorkerRequest'), false) diff --git a/apps/desktop/preload/apis/chat.ts b/apps/desktop/preload/apis/chat.ts index ba7b3669..24a78b1a 100644 --- a/apps/desktop/preload/apis/chat.ts +++ b/apps/desktop/preload/apis/chat.ts @@ -7,6 +7,24 @@ import { ipcRenderer } from 'electron' import type { ImportProgress } from '../../../../src/types/base' +interface ChatImportResult { + success: boolean + sessionId?: string + error?: string + importMode?: 'created' | 'incremental' + matchedBy?: 'stable-id' | 'trailing-messages' + newMessageCount?: number + duplicateCount?: number + diagnostics?: { + logFile: string | null + detectedFormat: string | null + messagesReceived: number + messagesWritten: number + messagesSkipped: number + skipReasons: { noSenderId: number; noAccountName: number; invalidTimestamp: number; noType: number } + } +} + export const chatApi = { // ==================== 数据库迁移 ==================== @@ -25,21 +43,16 @@ export const chatApi = { selectFile: (): Promise<{ filePath?: string; format?: string; error?: string } | null> => ipcRenderer.invoke('chat:selectFile'), - import: (filePath: string): Promise<{ success: boolean; sessionId?: string; error?: string }> => - ipcRenderer.invoke('chat:import', filePath), + import: (filePath: string): Promise => ipcRenderer.invoke('chat:import', filePath), - importDirectory: (dirPath: string): Promise<{ success: boolean; sessionId?: string; error?: string }> => - ipcRenderer.invoke('chat:importDirectory', dirPath), + importDirectory: (dirPath: string): Promise => ipcRenderer.invoke('chat:importDirectory', dirPath), detectFormat: ( filePath: string ): Promise<{ id: string; name: string; platform: string; multiChat: boolean } | null> => ipcRenderer.invoke('chat:detectFormat', filePath), - importWithOptions: ( - filePath: string, - formatOptions: Record - ): Promise<{ success: boolean; sessionId?: string; error?: string }> => + importWithOptions: (filePath: string, formatOptions: Record): Promise => ipcRenderer.invoke('chat:importWithOptions', filePath, formatOptions), scanMultiChatFile: ( @@ -52,7 +65,7 @@ export const chatApi = { prepareImportSource: (filePath: string) => ipcRenderer.invoke('chat:prepareImportSource', filePath), - importPreparedChat: (sourceId: string, chatId: string) => + importPreparedChat: (sourceId: string, chatId: string): Promise => ipcRenderer.invoke('chat:importPreparedChat', sourceId, chatId), releaseImportSource: (sourceId: string) => ipcRenderer.invoke('chat:releaseImportSource', sourceId), diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index a7354bef..6f961641 100644 --- a/apps/desktop/preload/index.d.ts +++ b/apps/desktop/preload/index.d.ts @@ -38,18 +38,26 @@ interface ImportDiagnostics { } } +interface ChatImportResult { + success: boolean + sessionId?: string + error?: string + importMode?: 'created' | 'incremental' + matchedBy?: 'stable-id' | 'trailing-messages' + newMessageCount?: number + duplicateCount?: number + diagnostics?: ImportDiagnostics +} + /** * ChatApi — 导入、迁移、Demo（数据查询/分析/成员/SQL 已迁移到 HTTP） */ 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<{ success: boolean; sessionId?: string; error?: string }> - importDirectory: (dirPath: string) => Promise<{ success: boolean; sessionId?: string; error?: string }> - importWithOptions: ( - filePath: string, - formatOptions: Record - ) => Promise<{ success: boolean; sessionId?: string; error?: string }> + import: (filePath: string) => Promise + importDirectory: (dirPath: string) => Promise + importWithOptions: (filePath: string, formatOptions: Record) => Promise scanMultiChatFile: (filePath: string) => Promise<{ success: boolean chats: Array<{ index: number; name: string; type: string; id: number; messageCount: number }> @@ -72,10 +80,7 @@ interface ChatApi { } error?: string }> - importPreparedChat: ( - sourceId: string, - chatId: string - ) => Promise<{ success: boolean; sessionId?: string; error?: string; diagnostics?: ImportDiagnosticsInfo }> + importPreparedChat: (sourceId: string, chatId: string) => Promise releaseImportSource: (sourceId: string) => Promise<{ success: boolean }> checkMigration: () => Promise runMigration: () => Promise<{ success: boolean; error?: string }>