feat(electron): enable automatic incremental imports

This commit is contained in:
digua committed 2026-07-13 01:01:42 +08:00
1 parent 0606d5f5f7
commit 3e6f036826
8 files changed
+139 -97

No files matched your search

+28 -76
View File
@@ -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<ReturnType<typeof worker.autoImport>>) => {
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) }
+2
View File
@@ -76,6 +76,7 @@ import {
} from './query'
import {
streamImport,
autoImport,
streamParseFileInfo,
analyzeIncrementalImport,
incrementalImport,
@@ -246,6 +247,7 @@ const syncHandlers: Record<string, (payload: any) => any> = {
const asyncHandlers: Record<string, (payload: any, requestId: string) => Promise<any>> = {
// 流式导入
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),
// 增量导入
+2
View File
@@ -6,9 +6,11 @@
// 流式导入(核心导入功能)
export {
streamImport,
autoImport,
streamParseFileInfo,
analyzeNewImport,
type StreamImportResult,
type AutoImportResult,
type StreamParseFileInfoResult,
type AnalyzeNewImportResult,
} from './streamImport'
@@ -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<string, unknown>,
explicitSessionId?: string
): Promise<AutoImportResult> {
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.
*/
+33 -1
View File
@@ -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<string, unknown>,
explicitSessionId?: string
): Promise<AutoImportResult> {
assertDataDirCompatibleNow()
const result = await sendToWorkerWithProgress<AutoImportResult>(
'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']) {
@@ -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)
+22 -9
View File
@@ -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<ChatImportResult> => ipcRenderer.invoke('chat:import', filePath),
importDirectory: (dirPath: string): Promise<{ success: boolean; sessionId?: string; error?: string }> =>
ipcRenderer.invoke('chat:importDirectory', dirPath),
importDirectory: (dirPath: string): Promise<ChatImportResult> => 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<string, unknown>
): Promise<{ success: boolean; sessionId?: string; error?: string }> =>
importWithOptions: (filePath: string, formatOptions: Record<string, unknown>): Promise<ChatImportResult> =>
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<ChatImportResult> =>
ipcRenderer.invoke('chat:importPreparedChat', sourceId, chatId),
releaseImportSource: (sourceId: string) => ipcRenderer.invoke('chat:releaseImportSource', sourceId),
+15 -10
View File
@@ -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<string, unknown>
) => Promise<{ success: boolean; sessionId?: string; error?: string }>
import: (filePath: string) => Promise<ChatImportResult>
importDirectory: (dirPath: string) => Promise<ChatImportResult>
importWithOptions: (filePath: string, formatOptions: Record<string, unknown>) => Promise<ChatImportResult>
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<ChatImportResult>
releaseImportSource: (sourceId: string) => Promise<{ success: boolean }>
checkMigration: () => Promise<MigrationCheckResult>
runMigration: () => Promise<{ success: boolean; error?: string }>