From 0606d5f5f74890a3f35042cdead6e781845ef1ef Mon Sep 17 00:00:00 2001 From: digua Date: Sun, 12 Jul 2026 16:39:44 +0800 Subject: [PATCH] feat(cli): enable automatic incremental imports --- apps/cli/src/cli.ts | 17 +- apps/cli/src/import/importer.test.ts | 49 ++++- apps/cli/src/import/index.ts | 2 + apps/cli/src/import/stream-import.test.ts | 24 ++ apps/cli/src/import/stream-import.ts | 123 +++++++---- .../src/import/incremental-importer.ts | 207 +++++++++--------- 6 files changed, 273 insertions(+), 149 deletions(-) diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts index 818d8b18..4a91c106 100644 --- a/apps/cli/src/cli.ts +++ b/apps/cli/src/cli.ts @@ -58,7 +58,7 @@ registerManifestCommand(program, getVersion()) program .command('import ') .description('Import a chat history file (14+ formats: QQ/WeChat/Telegram/WhatsApp/LINE/Discord/Instagram, etc.)') - .option('--session-id ', 'Specify session ID (auto-generated if omitted)') + .option('--session-id ', 'Force an existing session target, or create with this ID if missing') .option('--format ', 'Specify format ID (skip auto-detection)') .action(async (file, options) => { if (!fs.existsSync(file)) { @@ -66,7 +66,7 @@ program process.exit(1) } - const { streamImport, detectFormat } = await import('./import') + const { autoImport, detectFormat } = await import('./import') const { dbManager, pathProvider } = initRuntime() const nativeBinding = resolveNativeBinding() // 记录 Rust native parser 可用性（本次导入是否走 Rust 内核） @@ -83,8 +83,9 @@ program if (format) console.log(` Format: ${format.name} (${format.platform})`) try { - const result = await streamImport(dbManager, file, { + const result = await autoImport(dbManager, file, { formatId: options.format, + sessionId: options.sessionId, nativeBinding, onProgress: (p) => { process.stdout.write(`\r ${p.stage}: ${p.progress}%`) @@ -94,8 +95,10 @@ program if (result.success) { console.log(`\n\nImport succeeded!`) console.log(` Session ID: ${result.sessionId}`) - console.log(` Messages written: ${result.diagnostics?.messagesWritten ?? 0}`) - console.log(` Messages skipped: ${result.diagnostics?.messagesSkipped ?? 0}`) + console.log(` Mode: ${result.importMode === 'incremental' ? 'incremental' : 'created'}`) + if (result.matchedBy) console.log(` Matched by: ${result.matchedBy}`) + console.log(` New messages: ${result.newMessageCount ?? 0}`) + console.log(` Duplicates skipped: ${result.duplicateCount ?? 0}`) if (result.sessionId) { try { @@ -115,11 +118,11 @@ program } } else { console.error(`\n\nImport failed: ${result.error}`) - process.exit(1) + process.exitCode = 1 } } catch (err) { console.error(`\n\nImport error: ${err instanceof Error ? err.message : err}`) - process.exit(1) + process.exitCode = 1 } finally { dbManager.closeAll() } diff --git a/apps/cli/src/import/importer.test.ts b/apps/cli/src/import/importer.test.ts index 9f48ba00..6afd844d 100644 --- a/apps/cli/src/import/importer.test.ts +++ b/apps/cli/src/import/importer.test.ts @@ -5,7 +5,7 @@ import path from 'node:path' import test from 'node:test' import type { PathProvider } from '@openchatlab/core' import { DatabaseManager, raiseDataDirMinRuntimeVersion, readDataDirCompatibilityMeta } from '@openchatlab/node-runtime' -import { streamImport } from './stream-import' +import { autoImport, streamImport } from './stream-import' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -36,7 +36,7 @@ function writeTempChatFile(dir: string): string { filePath, JSON.stringify({ chatlab: { version: '0.0.2', exportedAt: 1711468800 }, - meta: { name: 'Test Chat', platform: 'qq', type: 'group' }, + meta: { name: 'Test Chat', platform: 'qq', type: 'group', groupId: 'group-42' }, members: [{ platformId: 'u1', accountName: 'Alice' }], // accountName is required by streaming-importer (skipped otherwise) messages: [{ sender: 'u1', accountName: 'Alice', timestamp: 1711468800, type: 0, content: 'hello' }], @@ -91,3 +91,48 @@ test('streamImport re-checks data directory compatibility before raw database wr assert.match(result.error ?? '', /requires runtime version 0\.26\.0 or newer/) assert.equal(fs.readdirSync(path.join(root, 'data', 'databases')).filter((name) => name.endsWith('.db')).length, 0) }) + +test('autoImport creates once and then incrementally imports the same stable chat', async () => { + const root = makeTempDir() + fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) + const manager = new DatabaseManager(createPathProvider(root), { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + const chatFile = writeTempChatFile(root) + + const created = await autoImport(manager, chatFile, { nativeBinding }) + const incremental = await autoImport(manager, chatFile, { nativeBinding }) + + assert.equal(created.success, true) + assert.equal(created.importMode, 'created') + assert.equal(incremental.success, true) + assert.equal(incremental.sessionId, created.sessionId) + assert.equal(incremental.importMode, 'incremental') + assert.equal(incremental.matchedBy, 'stable-id') + assert.equal(incremental.newMessageCount, 0) + assert.equal(incremental.duplicateCount, 1) + assert.equal(manager.listSessionIds().length, 1) +}) + +test('autoImport uses an explicit id for create and then forces incremental import', async () => { + const root = makeTempDir() + fs.mkdirSync(path.join(root, 'data', 'databases'), { recursive: true }) + const manager = new DatabaseManager(createPathProvider(root), { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + const chatFile = writeTempChatFile(root) + + const created = await autoImport(manager, chatFile, { sessionId: 'explicit', nativeBinding }) + const incremental = await autoImport(manager, chatFile, { sessionId: 'explicit', nativeBinding }) + + assert.equal(created.success, true) + assert.equal(created.sessionId, 'explicit') + assert.equal(created.importMode, 'created') + assert.equal(incremental.success, true) + assert.equal(incremental.sessionId, 'explicit') + assert.equal(incremental.importMode, 'incremental') + assert.equal(incremental.newMessageCount, 0) + assert.equal(manager.listSessionIds().length, 1) +}) diff --git a/apps/cli/src/import/index.ts b/apps/cli/src/import/index.ts index c33e3f6f..a59e7e63 100644 --- a/apps/cli/src/import/index.ts +++ b/apps/cli/src/import/index.ts @@ -1,6 +1,7 @@ // Full-format stream import via @openchatlab/parser + node-runtime streaming importer export { streamImport, + autoImport, incrementalImport, analyzeIncrementalImport, analyzeNewImport, @@ -20,4 +21,5 @@ export type { IncrementalImportResult, IncrementalAnalyzeResult, AnalyzeNewImportResult, + AutoImportResult, } from './stream-import' diff --git a/apps/cli/src/import/stream-import.test.ts b/apps/cli/src/import/stream-import.test.ts index 79748081..9e71051e 100644 --- a/apps/cli/src/import/stream-import.test.ts +++ b/apps/cli/src/import/stream-import.test.ts @@ -111,3 +111,27 @@ test('analyzeIncrementalImport propagates data directory compatibility errors', (error) => error instanceof DataDirCompatibilityError && error.code === 'DATA_DIR_REQUIRES_NEWER_RUNTIME' ) }) + +test('incrementalImport honors an explicitly selected parser format', async () => { + const root = makeTempDir() + const pathProvider = createPathProvider(root) + const manager = new DatabaseManager(pathProvider, { + nativeBinding, + runtime: { version: '0.25.1', kind: 'cli' }, + }) + const db = manager.openRawSessionDatabase('existing', { create: true, initializeChatTables: true }) + db.prepare('INSERT INTO meta (name, platform, type, imported_at) VALUES (?, ?, ?, ?)').run( + 'Existing Chat', + 'qq', + 'group', + 1000 + ) + db.close() + + const filePath = path.join(root, 'explicit-format.txt') + writeIncrementalJsonl(filePath) + const result = await incrementalImport(manager, 'existing', filePath, { formatId: 'chatlab-jsonl' }) + + assert.equal(result.success, true) + assert.equal(result.newMessageCount, 1) +}) diff --git a/apps/cli/src/import/stream-import.ts b/apps/cli/src/import/stream-import.ts index 3521fae7..34213673 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -9,6 +9,7 @@ import type { DatabaseManager } from '@openchatlab/node-runtime' import { DataDirCompatibilityError, + autoImportFile as sharedAutoImportFile, streamingImport, analyzeNewImport as sharedAnalyzeNewImport, analyzeIncrementalImport as sharedAnalyzeIncremental, @@ -23,6 +24,7 @@ import type { IncrementalImportDeps, ImportOptions, AnalyzeNewImportResult, + AutoImportResult, } from '@openchatlab/node-runtime' import { detectFormat as parserDetectFormat, @@ -76,6 +78,48 @@ function buildStreamImportDeps(dbManager: DatabaseManager, onProgress?: ImportPr } } +function createProgressAdapter(onProgress?: (progress: StreamImportProgress) => void): ImportProgressCallback { + if (!onProgress) return () => {} + return (progress) => { + let stage: StreamImportProgress['stage'] = 'parsing' + let pct = 0 + switch (progress.stage) { + case 'detecting': + stage = 'detecting' + pct = 5 + break + case 'parsing': + stage = 'parsing' + pct = Math.min(Math.round(progress.percentage * 0.7), 70) + break + case 'saving': + stage = 'saving' + pct = 80 + break + case 'indexing': + stage = 'indexing' + pct = 90 + break + case 'done': + stage = 'done' + pct = 100 + break + case 'error': + stage = 'error' + pct = 0 + break + } + onProgress({ + stage, + progress: pct, + message: progress.message || '', + bytesRead: progress.bytesRead, + totalBytes: progress.totalBytes, + messagesProcessed: progress.messagesProcessed, + }) + } +} + function deleteSessionDatabase(dbManager: DatabaseManager, sessionId: string): void { dbManager.deleteSessionDatabaseFiles(sessionId) } @@ -95,46 +139,7 @@ export async function streamImport( if (formatId) formatOptions.formatId = formatId if (chatIndex !== undefined) formatOptions.chatIndex = chatIndex - const progressAdapter: ImportProgressCallback = onProgress - ? (progress) => { - let stage: StreamImportProgress['stage'] = 'parsing' - let pct = 0 - switch (progress.stage) { - case 'detecting': - stage = 'detecting' - pct = 5 - break - case 'parsing': - stage = 'parsing' - pct = Math.min(Math.round(progress.percentage * 0.7), 70) - break - case 'saving': - stage = 'saving' - pct = 80 - break - case 'indexing': - stage = 'indexing' - pct = 90 - break - case 'done': - stage = 'done' - pct = 100 - break - case 'error': - stage = 'error' - pct = 0 - break - } - onProgress({ - stage, - progress: pct, - message: progress.message || '', - bytesRead: progress.bytesRead, - totalBytes: progress.totalBytes, - messagesProcessed: progress.messagesProcessed, - }) - } - : () => {} + const progressAdapter = createProgressAdapter(onProgress) const deps = buildStreamImportDeps(dbManager, progressAdapter) const result = await streamingImport(filePath, deps, formatOptions, sessionId) @@ -154,6 +159,43 @@ export async function streamImport( return result } +export async function autoImport( + dbManager: DatabaseManager, + filePath: string, + options?: StreamImportOptions +): Promise { + const { formatId, chatIndex, onProgress, sessionId } = options ?? {} + const formatOptions: Record = {} + if (formatId) formatOptions.formatId = formatId + if (chatIndex !== undefined) formatOptions.chatIndex = chatIndex + const progressAdapter = createProgressAdapter(onProgress) + + return sharedAutoImportFile( + filePath, + { + listSessionIds: () => dbManager.listSessionIds(), + openReadonly: (candidateSessionId) => dbManager.openRawSessionDatabase(candidateSessionId, { readonly: true }), + onProgress: progressAdapter, + sessionExists: (candidateSessionId) => dbManager.listSessionIds().includes(candidateSessionId), + createSession: (sourcePath, sourceFormatOptions, explicitSessionId) => + streamImport(dbManager, sourcePath, { + ...options, + ...sourceFormatOptions, + sessionId: explicitSessionId, + }), + appendSession: (targetSessionId, sourcePath, sourceFormatOptions) => + incrementalImport(dbManager, targetSessionId, sourcePath, { + ...sourceFormatOptions, + onProgress: progressAdapter, + }), + }, + { + explicitSessionId: sessionId, + formatOptions, + } + ) +} + // ==================== Incremental import ==================== function buildIncrementalDeps( @@ -249,4 +291,5 @@ export type { IncrementalAnalyzeResult, AnalyzeNewImportResult, ImportOptions, + AutoImportResult, } diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index 31161f99..d258f083 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -10,7 +10,7 @@ import type { DatabaseAdapter } from '@openchatlab/core' import { generateMessageKey, generateSessionIndex, generateIncrementalSessionIndex } from '@openchatlab/core' -import { streamParseFile, detectFormat, type ParseProgress } from '@openchatlab/parser' +import { streamParseFile, detectFormat, getFormatFeatureById, type ParseProgress } from '@openchatlab/parser' import { insertFtsEntries, hasFtsTable } from '../fts' import type { ImportProgressCallback } from './streaming-importer' @@ -19,6 +19,8 @@ import type { ImportProgressCallback } from './streaming-importer' export interface ImportOptions { metaUpdateMode?: 'patch' | 'none' memberUpdateMode?: 'upsert' | 'none' + formatId?: string + chatIndex?: number } export interface IncrementalAnalyzeResult { @@ -176,7 +178,7 @@ export async function incrementalImport( deps: IncrementalImportDeps, options?: ImportOptions ): Promise { - const formatFeature = detectFormat(filePath) + const formatFeature = options?.formatId ? getFormatFeatureById(options.formatId) : detectFormat(filePath) if (!formatFeature) { return { success: false, newMessageCount: 0, error: 'error.unrecognized_format' } } @@ -263,113 +265,118 @@ export async function incrementalImport( const newFtsEntries: Array<{ id: number; content: string | null }> = [] - await streamParseFile(filePath, { - onMeta: (meta) => { - if (metaUpdateMode === 'none') return - updateMeta.run( - meta.name || '', - meta.groupId || '', - meta.groupAvatar || '', - meta.ownerId || '', - Math.floor(Date.now() / 1000) - ) - metaUpdated = true - }, - onMembers: (members) => { - if (memberUpdateMode === 'none') return - for (const m of members) { - const existed = memberIdMap.has(m.platformId) - upsertMember.run( - m.platformId, - m.accountName || null, - m.groupNickname || null, - m.aliases ? JSON.stringify(m.aliases) : '[]', - m.avatar || null, - m.roles ? JSON.stringify(m.roles) : '[]' + await streamParseFile( + filePath, + { + formatOptions: options?.chatIndex === undefined ? undefined : { chatIndex: options.chatIndex }, + onMeta: (meta) => { + if (metaUpdateMode === 'none') return + updateMeta.run( + meta.name || '', + meta.groupId || '', + meta.groupAvatar || '', + meta.ownerId || '', + Math.floor(Date.now() / 1000) ) - if (!existed) { - const row = getMemberId.get(m.platformId) as { id: number } | undefined - if (row) memberIdMap.set(m.platformId, row.id) - membersAdded++ - } else { - membersUpdated++ - } - } - }, - onProgress: (progress: ParseProgress) => { - deps.onProgress(progress) - }, - onMessageBatch: (batch) => { - for (const msg of batch) { - processedCount++ - - if (!msg.senderPlatformId) { - trackError(processedCount, 'MISSING_SENDER', 'sender field is empty') - continue - } - if (msg.timestamp === undefined || msg.timestamp === null) { - trackError(processedCount, 'MISSING_TIMESTAMP', 'timestamp field is missing') - continue - } - const timestamp = normalizeImportTimestamp(msg.timestamp) - if (timestamp === null) { - trackError(processedCount, 'INVALID_TIMESTAMP', `timestamp value: ${msg.timestamp}`) - continue - } - - if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys)) { - duplicateCount++ - continue - } - - let memberId = memberIdMap.get(msg.senderPlatformId) - if (!memberId) { - insertMemberMinimal.run( - msg.senderPlatformId, - msg.senderAccountName || null, - msg.senderGroupNickname || null, - null + metaUpdated = true + }, + onMembers: (members) => { + if (memberUpdateMode === 'none') return + for (const m of members) { + const existed = memberIdMap.has(m.platformId) + upsertMember.run( + m.platformId, + m.accountName || null, + m.groupNickname || null, + m.aliases ? JSON.stringify(m.aliases) : '[]', + m.avatar || null, + m.roles ? JSON.stringify(m.roles) : '[]' ) - const row = getMemberId.get(msg.senderPlatformId) as { id: number } | undefined - if (row) { - memberId = row.id - memberIdMap.set(msg.senderPlatformId, memberId) + if (!existed) { + const row = getMemberId.get(m.platformId) as { id: number } | undefined + if (row) memberIdMap.set(m.platformId, row.id) membersAdded++ + } else { + membersUpdated++ } } - if (!memberId) continue + }, + onProgress: (progress: ParseProgress) => { + deps.onProgress(progress) + }, + onMessageBatch: (batch) => { + for (const msg of batch) { + processedCount++ - const msgResult = insertMessage.run( - memberId, - msg.senderAccountName || null, - msg.senderGroupNickname || null, - timestamp, - msg.type, - msg.content || null, - msg.replyToMessageId || null, - msg.platformMessageId || null - ) + if (!msg.senderPlatformId) { + trackError(processedCount, 'MISSING_SENDER', 'sender field is empty') + continue + } + if (msg.timestamp === undefined || msg.timestamp === null) { + trackError(processedCount, 'MISSING_TIMESTAMP', 'timestamp field is missing') + continue + } + const timestamp = normalizeImportTimestamp(msg.timestamp) + if (timestamp === null) { + trackError(processedCount, 'INVALID_TIMESTAMP', `timestamp value: ${msg.timestamp}`) + continue + } - newFtsEntries.push({ - id: Number((msgResult as any).lastInsertRowid ?? 0), - content: msg.content || null, - }) - if (timestamp < minWrittenTs) minWrittenTs = timestamp - newMessageCount++ - } + if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys)) { + duplicateCount++ + continue + } - if (processedCount % BATCH_SIZE === 0) { - deps.onProgress({ - stage: 'saving', - bytesRead: 0, - totalBytes: 0, - messagesProcessed: processedCount, - percentage: 50, - message: `Processed ${processedCount}, added ${newMessageCount}`, - }) - } + let memberId = memberIdMap.get(msg.senderPlatformId) + if (!memberId) { + insertMemberMinimal.run( + msg.senderPlatformId, + msg.senderAccountName || null, + msg.senderGroupNickname || null, + null + ) + const row = getMemberId.get(msg.senderPlatformId) as { id: number } | undefined + if (row) { + memberId = row.id + memberIdMap.set(msg.senderPlatformId, memberId) + membersAdded++ + } + } + if (!memberId) continue + + const msgResult = insertMessage.run( + memberId, + msg.senderAccountName || null, + msg.senderGroupNickname || null, + timestamp, + msg.type, + msg.content || null, + msg.replyToMessageId || null, + msg.platformMessageId || null + ) + + newFtsEntries.push({ + id: Number((msgResult as any).lastInsertRowid ?? 0), + content: msg.content || null, + }) + if (timestamp < minWrittenTs) minWrittenTs = timestamp + newMessageCount++ + } + + if (processedCount % BATCH_SIZE === 0) { + deps.onProgress({ + stage: 'saving', + bytesRead: 0, + totalBytes: 0, + messagesProcessed: processedCount, + percentage: 50, + message: `Processed ${processedCount}, added ${newMessageCount}`, + }) + } + }, }, - }) + options?.formatId + ) db.exec('COMMIT')