From 53406f4c6d2c0ad3a1c89e44f06911e076bf20cc Mon Sep 17 00:00:00 2001 From: digua Date: Tue, 14 Jul 2026 11:00:17 +0800 Subject: [PATCH] feat(import): add dry-run preview Share auto-import planning between preview and write paths, expose a machine-readable CLI result envelope, and preserve explicit parser options during analysis. --- apps/cli/src/cli.ts | 91 +------- apps/cli/src/import/command.test.ts | 68 ++++++ apps/cli/src/import/command.ts | 210 ++++++++++++++++++ apps/cli/src/import/importer.test.ts | 27 ++- apps/cli/src/import/index.ts | 2 + apps/cli/src/import/stream-import.test.ts | 6 +- apps/cli/src/import/stream-import.ts | 50 ++++- .../src/import/auto-importer.test.ts | 105 ++++++++- .../node-runtime/src/import/auto-importer.ts | 172 ++++++++++---- .../src/import/incremental-importer.ts | 44 ++-- packages/node-runtime/src/import/index.ts | 11 +- .../src/import/streaming-importer.test.ts | 15 ++ .../src/import/streaming-importer.ts | 59 +++-- packages/node-runtime/src/index.ts | 3 + 14 files changed, 677 insertions(+), 186 deletions(-) create mode 100644 apps/cli/src/import/command.test.ts create mode 100644 apps/cli/src/import/command.ts diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts index ce4b0518..bd14fc1d 100644 --- a/apps/cli/src/cli.ts +++ b/apps/cli/src/cli.ts @@ -8,20 +8,14 @@ import * as fs from 'fs' import { execSync } from 'child_process' import { Command } from 'commander' import { DEFAULT_API_PORT, loadConfig, getConfigPath, setConfigField, ConfigSetError } from '@openchatlab/config' -import { - AIChatManager, - initAppLogger, - appLogger, - getSystemLogsDir, - IMPORT_IN_PROGRESS_ERROR_KEY, - logNativeParserStatus, -} from '@openchatlab/node-runtime' +import { AIChatManager, initAppLogger, appLogger, getSystemLogsDir } from '@openchatlab/node-runtime' import { getVersion } from './version' import { resolveCliPath } from './paths' import { isPortAvailable, formatPortInUseError } from './http/port' import { initRuntime, resolveNativeBinding } from './runtime' import { registerQueryCommands } from './query/register' import { registerManifestCommand } from './query/manifest' +import { registerImportCommand } from './import/command' const program = new Command() @@ -56,86 +50,7 @@ program registerQueryCommands(program) 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 ', '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)) { - console.error(`File not found: ${file}`) - process.exit(1) - } - - const { autoImport, detectFormat } = await import('./import') - const { dbManager, pathProvider } = initRuntime() - const nativeBinding = resolveNativeBinding() - // 记录 Rust native parser 可用性（本次导入是否走 Rust 内核） - logNativeParserStatus() - - const format = detectFormat(file) - if (!format && !options.format) { - console.error(`Unrecognized file format: ${file}`) - console.error('Use --format to specify manually, or run "chatlab formats" to see supported formats') - process.exit(1) - } - - console.log(`Importing: ${file}`) - if (format) console.log(` Format: ${format.name} (${format.platform})`) - - try { - const result = await autoImport(dbManager, file, { - formatId: options.format, - sessionId: options.sessionId, - nativeBinding, - onProgress: (p) => { - process.stdout.write(`\r ${p.stage}: ${p.progress}%`) - }, - }) - - if (result.success) { - console.log(`\n\nImport succeeded!`) - console.log(` Session ID: ${result.sessionId}`) - console.log(` Mode: ${result.importMode ?? 'unknown'}`) - if (result.matchedBy) console.log(` Matched by: ${result.matchedBy}`) - if (result.createReason) console.log(` Create reason: ${result.createReason}`) - if (result.createReason === 'ambiguous') { - console.warn(' Warning: multiple existing sessions matched, so a new session was created.') - } - console.log(` New messages: ${result.newMessageCount ?? 0}`) - console.log(` Duplicates skipped: ${result.duplicateCount ?? 0}`) - - if (result.sessionId) { - try { - const { PreferencesManager, createDatabaseManagerAdapter, ownerProfileService } = - await import('@openchatlab/node-runtime') - const applied = ownerProfileService.tryApplyOwnerProfile( - createDatabaseManagerAdapter(dbManager), - new PreferencesManager(pathProvider.getSystemDir()), - result.sessionId - ) - if (applied.applied) { - console.log(` Owner auto-detected: ${applied.ownerId}`) - } - } catch (ownerErr) { - console.warn(` Owner profile apply skipped: ${ownerErr instanceof Error ? ownerErr.message : ownerErr}`) - } - } - } else { - const errorMessage = - result.error === IMPORT_IN_PROGRESS_ERROR_KEY - ? 'Another import is already in progress. Please retry later.' - : result.error - console.error(`\n\nImport failed: ${errorMessage}`) - process.exitCode = 1 - } - } catch (err) { - console.error(`\n\nImport error: ${err instanceof Error ? err.message : err}`) - process.exitCode = 1 - } finally { - dbManager.closeAll() - } - }) +registerImportCommand(program) program .command('formats') diff --git a/apps/cli/src/import/command.test.ts b/apps/cli/src/import/command.test.ts new file mode 100644 index 00000000..72eebedb --- /dev/null +++ b/apps/cli/src/import/command.test.ts @@ -0,0 +1,68 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { buildImportCommandEnvelope, withJsonDiagnosticsOnStderr } from './command' + +test('buildImportCommandEnvelope exposes a stable dry-run success contract', () => { + assert.deepEqual( + buildImportCommandEnvelope( + { + success: true, + importMode: 'incremental', + sessionId: 'existing', + matchedBy: 'stable-id', + totalMessageCount: 12, + newMessageCount: 4, + duplicateCount: 8, + }, + true + ), + { + ok: true, + command: 'import', + data: { + success: true, + importMode: 'incremental', + sessionId: 'existing', + matchedBy: 'stable-id', + totalMessageCount: 12, + newMessageCount: 4, + duplicateCount: 8, + }, + meta: { dryRun: true, apiVersion: 1 }, + } + ) +}) + +test('buildImportCommandEnvelope maps import-lock failures for agents', () => { + assert.deepEqual(buildImportCommandEnvelope({ success: false, error: 'error.import_in_progress' }, false), { + ok: false, + command: 'import', + error: { + code: 'IMPORT_IN_PROGRESS', + message: 'Another import is already in progress.', + hint: 'Wait for the active import to finish, then retry.', + }, + meta: { dryRun: false, apiVersion: 1 }, + }) +}) + +test('withJsonDiagnosticsOnStderr keeps runtime diagnostics out of machine-readable stdout', async () => { + const stdout: string[] = [] + const stderr: string[] = [] + const originalLog = console.log + const originalError = console.error + console.log = (...args: unknown[]) => stdout.push(args.join(' ')) + console.error = (...args: unknown[]) => stderr.push(args.join(' ')) + + try { + await withJsonDiagnosticsOnStderr(true, async () => { + console.log('[Parser V2] Using parser: ChatLab JSON') + }) + } finally { + console.log = originalLog + console.error = originalError + } + + assert.deepEqual(stdout, []) + assert.deepEqual(stderr, ['[Parser V2] Using parser: ChatLab JSON']) +}) diff --git a/apps/cli/src/import/command.ts b/apps/cli/src/import/command.ts new file mode 100644 index 00000000..3d84b9c1 --- /dev/null +++ b/apps/cli/src/import/command.ts @@ -0,0 +1,210 @@ +import fs from 'node:fs' +import type { Command } from 'commander' +import { + IMPORT_IN_PROGRESS_ERROR_KEY, + PreferencesManager, + createDatabaseManagerAdapter, + logNativeParserStatus, + ownerProfileService, + type AutoImportAnalysisResult, + type AutoImportResult, +} from '@openchatlab/node-runtime' +import { initRuntime, resolveNativeBinding } from '../runtime' +import { analyzeAutoImport, autoImport, detectFormat } from './index' + +interface ImportCommandOptions { + sessionId?: string + format?: string + dryRun?: boolean + json?: boolean +} + +interface ImportCommandEnvelope { + ok: boolean + command: 'import' + data?: AutoImportResult | AutoImportAnalysisResult + error?: { + code: string + message: string + hint?: string + } + meta: { + dryRun: boolean + apiVersion: 1 + } +} + +function normalizeImportError(message: string): NonNullable { + if (message.startsWith('File not found:')) { + return { + code: 'FILE_NOT_FOUND', + message, + hint: 'Pass an absolute path or a path relative to the current working directory.', + } + } + if (message === IMPORT_IN_PROGRESS_ERROR_KEY) { + return { + code: 'IMPORT_IN_PROGRESS', + message: 'Another import is already in progress.', + hint: 'Wait for the active import to finish, then retry.', + } + } + if (message === 'error.unrecognized_format' || message.toLowerCase().includes('unrecognized')) { + return { + code: 'UNRECOGNIZED_FORMAT', + message: 'The chat export format could not be recognized.', + hint: 'Run "chatlab formats" and retry with --format .', + } + } + if (message.includes('sessionId contains invalid characters')) { + return { + code: 'INVALID_SESSION_ID', + message, + hint: 'Use only letters, numbers, underscores, and hyphens (maximum 128 characters).', + } + } + return { code: 'IMPORT_FAILED', message } +} + +export function buildImportCommandEnvelope( + result: AutoImportResult | AutoImportAnalysisResult, + dryRun: boolean +): ImportCommandEnvelope { + return result.success + ? { ok: true, command: 'import', data: result, meta: { dryRun, apiVersion: 1 } } + : { + ok: false, + command: 'import', + error: normalizeImportError(result.error || 'Import failed'), + meta: { dryRun, apiVersion: 1 }, + } +} + +function printHumanResult(result: AutoImportResult | AutoImportAnalysisResult, dryRun: boolean): void { + if (!result.success) { + const error = normalizeImportError(result.error || 'Import failed') + console.error(`\n\nImport failed: ${error.message}`) + if (error.hint) console.error(` Hint: ${error.hint}`) + return + } + + console.log(dryRun ? '\n\nDry run completed.' : '\n\nImport succeeded!') + console.log(` Mode: ${result.importMode ?? 'unknown'}`) + if (result.sessionId) console.log(` Session ID: ${result.sessionId}`) + if (result.matchedBy) console.log(` Matched by: ${result.matchedBy}`) + if (result.createReason) console.log(` Create reason: ${result.createReason}`) + if (dryRun && 'totalMessageCount' in result && result.totalMessageCount !== undefined) { + console.log(` Messages scanned: ${result.totalMessageCount}`) + } + console.log(` New messages: ${result.newMessageCount ?? 0}`) + console.log(` Duplicates skipped: ${result.duplicateCount ?? 0}`) +} + +function printJsonResult(result: AutoImportResult | AutoImportAnalysisResult, dryRun: boolean): void { + process.stdout.write(`${JSON.stringify(buildImportCommandEnvelope(result, dryRun))}\n`) +} + +export async function withJsonDiagnosticsOnStderr(enabled: boolean, run: () => Promise): Promise { + if (!enabled) return run() + + const originalConsoleLog = console.log + console.log = (...args: unknown[]) => console.error(...args) + try { + return await run() + } finally { + console.log = originalConsoleLog + } +} + +async function applyOwnerProfile( + result: AutoImportResult, + systemDir: string, + dbManager: ReturnType['dbManager'] +) { + if (!result.sessionId) return + try { + const applied = ownerProfileService.tryApplyOwnerProfile( + createDatabaseManagerAdapter(dbManager), + new PreferencesManager(systemDir), + result.sessionId + ) + if (applied.applied) console.error(`Owner auto-detected: ${applied.ownerId}`) + } catch (error) { + console.error(`Owner profile apply skipped: ${error instanceof Error ? error.message : error}`) + } +} + +export function registerImportCommand(program: Command): void { + program + .command('import ') + .description('Import a chat history file (14+ formats: QQ/WeChat/Telegram/WhatsApp/LINE/Discord, etc.)') + .option('--session-id ', 'Force an existing session target, or create with this ID if missing') + .option('--format ', 'Specify the input format ID (skip auto-detection)') + .option('--dry-run', 'Analyze the target and message counts without writing data') + .option('--json', 'Output one machine-readable JSON envelope') + .action(async (file: string, options: ImportCommandOptions) => { + if (!fs.existsSync(file)) { + const result: AutoImportAnalysisResult = { success: false, error: `File not found: ${file}` } + if (options.json) printJsonResult(result, Boolean(options.dryRun)) + else printHumanResult(result, Boolean(options.dryRun)) + process.exitCode = 3 + return + } + + const result = await withJsonDiagnosticsOnStderr(Boolean(options.json), async () => { + let dbManager: ReturnType['dbManager'] | undefined + + try { + const runtime = initRuntime() + dbManager = runtime.dbManager + const nativeBinding = resolveNativeBinding() + logNativeParserStatus() + + const detectedFormat = options.format ? undefined : detectFormat(file) + if (!detectedFormat && !options.format) { + return { success: false, error: 'error.unrecognized_format' } satisfies AutoImportAnalysisResult + } + + if (!options.json) { + console.log(`${options.dryRun ? 'Analyzing' : 'Importing'}: ${file}`) + if (detectedFormat) console.log(` Format: ${detectedFormat.name} (${detectedFormat.platform})`) + else if (options.format) console.log(` Format ID: ${options.format} (explicit)`) + } + + const importOptions = { + formatId: options.format, + sessionId: options.sessionId, + nativeBinding, + onProgress: options.json + ? undefined + : (progress: { stage: string; progress: number }) => { + process.stdout.write(`\r ${progress.stage}: ${progress.progress}%`) + }, + } + const importResult = options.dryRun + ? await analyzeAutoImport(dbManager, file, importOptions) + : await autoImport(dbManager, file, importOptions) + + if (importResult.success && !options.dryRun) { + await applyOwnerProfile(importResult as AutoImportResult, runtime.pathProvider.getSystemDir(), dbManager) + } + return importResult + } catch (error) { + return { + success: false, + error: error instanceof Error ? error.message : String(error), + } satisfies AutoImportAnalysisResult + } finally { + dbManager?.closeAll() + } + }) + + if (options.json) printJsonResult(result, Boolean(options.dryRun)) + else printHumanResult(result, Boolean(options.dryRun)) + + if (!result.success) { + if (result.error === 'error.unrecognized_format') process.exitCode = 2 + else process.exitCode = result.error === IMPORT_IN_PROGRESS_ERROR_KEY ? 4 : 1 + } + }) +} diff --git a/apps/cli/src/import/importer.test.ts b/apps/cli/src/import/importer.test.ts index 1b07ef00..04351363 100644 --- a/apps/cli/src/import/importer.test.ts +++ b/apps/cli/src/import/importer.test.ts @@ -11,7 +11,7 @@ import { readDataDirCompatibilityMeta, withDataDirImportLock, } from '@openchatlab/node-runtime' -import { autoImport, streamImport } from './stream-import' +import { analyzeAutoImport, autoImport, streamImport } from './stream-import' const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') @@ -136,6 +136,31 @@ test('autoImport creates once and then incrementally imports the same stable cha assert.equal(manager.listSessionIds().length, 1) }) +test('analyzeAutoImport previews a new session without writing a database', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + 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 result = await analyzeAutoImport(manager, chatFile, { nativeBinding }) + + assert.deepEqual(result, { + success: true, + importMode: 'created', + createReason: 'no-match', + totalMessageCount: 1, + newMessageCount: 1, + duplicateCount: 0, + totalMemberCount: 1, + meta: { name: 'Test Chat', platform: 'qq', type: 'group' }, + }) + assert.deepEqual(manager.listSessionIds(), []) +}) + test('autoImport reports the same exact-message deduplication on create and incremental paths', async (t) => { const root = makeTempDir() t.after(() => fs.rmSync(root, { recursive: true, force: true })) diff --git a/apps/cli/src/import/index.ts b/apps/cli/src/import/index.ts index a59e7e63..803c2f9f 100644 --- a/apps/cli/src/import/index.ts +++ b/apps/cli/src/import/index.ts @@ -2,6 +2,7 @@ export { streamImport, autoImport, + analyzeAutoImport, incrementalImport, analyzeIncrementalImport, analyzeNewImport, @@ -21,5 +22,6 @@ export type { IncrementalImportResult, IncrementalAnalyzeResult, AnalyzeNewImportResult, + AutoImportAnalysisResult, 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 9e71051e..bad4ef96 100644 --- a/apps/cli/src/import/stream-import.test.ts +++ b/apps/cli/src/import/stream-import.test.ts @@ -112,7 +112,7 @@ test('analyzeIncrementalImport propagates data directory compatibility errors', ) }) -test('incrementalImport honors an explicitly selected parser format', async () => { +test('incremental analysis and import honor an explicitly selected parser format', async () => { const root = makeTempDir() const pathProvider = createPathProvider(root) const manager = new DatabaseManager(pathProvider, { @@ -130,8 +130,12 @@ test('incrementalImport honors an explicitly selected parser format', async () = const filePath = path.join(root, 'explicit-format.txt') writeIncrementalJsonl(filePath) + const analysis = await analyzeIncrementalImport(manager, 'existing', filePath, undefined, { + formatId: 'chatlab-jsonl', + }) const result = await incrementalImport(manager, 'existing', filePath, { formatId: 'chatlab-jsonl' }) + assert.deepEqual(analysis, { newMessageCount: 1, duplicateCount: 0, totalInFile: 1 }) 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 5d813167..da0087bf 100644 --- a/apps/cli/src/import/stream-import.ts +++ b/apps/cli/src/import/stream-import.ts @@ -11,6 +11,7 @@ import { DataDirCompatibilityError, IMPORT_IN_PROGRESS_ERROR_KEY, ImportInProgressError, + analyzeAutoImportFile as sharedAnalyzeAutoImportFile, autoImportFile as sharedAutoImportFile, streamingImport, analyzeNewImport as sharedAnalyzeNewImport, @@ -27,6 +28,7 @@ import type { IncrementalImportDeps, ImportOptions, AnalyzeNewImportResult, + AutoImportAnalysisResult, AutoImportResult, } from '@openchatlab/node-runtime' import { @@ -226,6 +228,42 @@ export async function autoImport( } } +export async function analyzeAutoImport( + 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 sharedAnalyzeAutoImportFile( + filePath, + { + listSessionIds: () => dbManager.listSessionIds(), + openReadonly: (candidateSessionId) => dbManager.openRawSessionDatabase(candidateSessionId, { readonly: true }), + onProgress: progressAdapter, + sessionExists: (candidateSessionId) => dbManager.listSessionIds().includes(candidateSessionId), + analyzeCreateSession: (sourcePath, sourceFormatOptions) => + sharedAnalyzeNewImport(sourcePath, progressAdapter, { + formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, + chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, + }), + analyzeAppendSession: (targetSessionId, sourcePath, sourceFormatOptions) => + sharedAnalyzeIncremental(targetSessionId, sourcePath, buildIncrementalDeps(dbManager, progressAdapter), { + formatId: typeof sourceFormatOptions?.formatId === 'string' ? sourceFormatOptions.formatId : undefined, + chatIndex: typeof sourceFormatOptions?.chatIndex === 'number' ? sourceFormatOptions.chatIndex : undefined, + }), + }, + { + explicitSessionId: sessionId, + formatOptions, + } + ) +} + // ==================== Incremental import ==================== function buildIncrementalDeps( @@ -301,7 +339,8 @@ export async function analyzeIncrementalImport( dbManager: DatabaseManager, sessionId: string, filePath: string, - onProgress?: ImportProgressCallback + onProgress?: ImportProgressCallback, + options?: ImportOptions ): Promise { let compatibilityError: DataDirCompatibilityError | null = null const result = await sharedAnalyzeIncremental( @@ -309,7 +348,8 @@ export async function analyzeIncrementalImport( filePath, buildIncrementalDeps(dbManager, onProgress, (error) => { compatibilityError = error - }) + }), + options ) if (compatibilityError) throw compatibilityError return result @@ -317,9 +357,10 @@ export async function analyzeIncrementalImport( export async function analyzeNewImport( filePath: string, - onProgress?: ImportProgressCallback + onProgress?: ImportProgressCallback, + options?: Pick ): Promise { - return sharedAnalyzeNewImport(filePath, onProgress ?? (() => {})) + return sharedAnalyzeNewImport(filePath, onProgress ?? (() => {}), options) } // ==================== Re-exports from parser ==================== @@ -338,6 +379,7 @@ export type { IncrementalImportResult, IncrementalAnalyzeResult, AnalyzeNewImportResult, + AutoImportAnalysisResult, ImportOptions, AutoImportResult, } diff --git a/packages/node-runtime/src/import/auto-importer.test.ts b/packages/node-runtime/src/import/auto-importer.test.ts index d9c126c1..aea47448 100644 --- a/packages/node-runtime/src/import/auto-importer.test.ts +++ b/packages/node-runtime/src/import/auto-importer.test.ts @@ -1,6 +1,11 @@ import assert from 'node:assert/strict' import test from 'node:test' -import { autoImportFile, type AutoImportDeps } from './auto-importer' +import { + analyzeAutoImportFile, + autoImportFile, + type AutoImportAnalysisDeps, + type AutoImportDeps, +} from './auto-importer' import type { AutoImportDecision } from './auto-import-matcher' function createDeps(options?: { @@ -209,3 +214,101 @@ test('matcher failure stops import instead of silently creating a session', asyn assert.deepEqual(calls.create, []) assert.deepEqual(calls.append, []) }) + +function createAnalysisDeps(options?: { + existingSessionIds?: string[] + decision?: AutoImportDecision + incrementalNewCount?: number + incrementalDuplicateCount?: number +}) { + const existing = new Set(options?.existingSessionIds ?? []) + const calls = { + create: [] as Array<{ filePath: string; formatOptions?: Record }>, + append: [] as Array<{ sessionId: string; filePath: string; formatOptions?: Record }>, + } + const deps: AutoImportAnalysisDeps = { + listSessionIds: () => [...existing], + openReadonly: () => { + throw new Error('not used by fake matcher') + }, + sessionExists: (sessionId) => existing.has(sessionId), + resolveTarget: async () => options?.decision ?? { action: 'create', reason: 'no-match' }, + analyzeCreateSession: async (filePath, formatOptions) => { + calls.create.push({ filePath, formatOptions }) + return { + totalMessages: 12, + newMessageCount: 10, + duplicateCount: 2, + totalMembers: 3, + meta: { name: 'Fixture', platform: 'qq', type: 'group' }, + } + }, + analyzeAppendSession: async (sessionId, filePath, formatOptions) => { + calls.append.push({ sessionId, filePath, formatOptions }) + return { + totalInFile: 12, + newMessageCount: options?.incrementalNewCount ?? 4, + duplicateCount: options?.incrementalDuplicateCount ?? 8, + } + }, + } + return { deps, calls } +} + +test('dry-run analysis reuses the automatic incremental target without writing', async () => { + const { deps, calls } = createAnalysisDeps({ + existingSessionIds: ['existing'], + decision: { action: 'incremental', sessionId: 'existing', matchedBy: 'stable-id' }, + }) + + const result = await analyzeAutoImportFile('source.json', deps, { + formatOptions: { formatId: 'chatlab', chatIndex: 1 }, + }) + + assert.deepEqual(result, { + success: true, + importMode: 'incremental', + sessionId: 'existing', + matchedBy: 'stable-id', + totalMessageCount: 12, + newMessageCount: 4, + duplicateCount: 8, + }) + assert.deepEqual(calls.create, []) + assert.deepEqual(calls.append, [ + { + sessionId: 'existing', + filePath: 'source.json', + formatOptions: { formatId: 'chatlab', chatIndex: 1 }, + }, + ]) +}) + +test('dry-run analysis reports a new-session plan without creating a database', async () => { + const { deps, calls } = createAnalysisDeps({ decision: { action: 'create', reason: 'ambiguous' } }) + + const result = await analyzeAutoImportFile('source.json', deps) + + assert.deepEqual(result, { + success: true, + importMode: 'created', + createReason: 'ambiguous', + totalMessageCount: 12, + newMessageCount: 10, + duplicateCount: 2, + totalMemberCount: 3, + meta: { name: 'Fixture', platform: 'qq', type: 'group' }, + }) + assert.deepEqual(calls.append, []) + assert.deepEqual(calls.create, [{ filePath: 'source.json', formatOptions: undefined }]) +}) + +test('dry-run analysis rejects an unsafe explicit session id before parsing', async () => { + const { deps, calls } = createAnalysisDeps() + + const result = await analyzeAutoImportFile('source.json', deps, { explicitSessionId: '../outside' }) + + assert.deepEqual(result, { success: false, error: 'sessionId contains invalid characters' }) + assert.deepEqual(calls.create, []) + assert.deepEqual(calls.append, []) +}) diff --git a/packages/node-runtime/src/import/auto-importer.ts b/packages/node-runtime/src/import/auto-importer.ts index 30b91ec7..6151f8fd 100644 --- a/packages/node-runtime/src/import/auto-importer.ts +++ b/packages/node-runtime/src/import/auto-importer.ts @@ -7,7 +7,8 @@ import { type AutoImportMatchMethod, } from './auto-import-matcher' import type { IncrementalImportResult } from './incremental-importer' -import type { ImportDiagnostics, StreamImportResult } from './streaming-importer' +import type { IncrementalAnalyzeResult } from './incremental-importer' +import type { AnalyzeNewImportResult, ImportDiagnostics, StreamImportResult } from './streaming-importer' import { isValidImportSessionId } from './session-id' export interface AutoImportOptions { @@ -30,6 +31,17 @@ export interface AutoImportDeps extends AutoImportMatcherDeps { resolveTarget?: typeof resolveAutoImportTarget } +export interface AutoImportAnalysisDeps extends AutoImportMatcherDeps { + sessionExists(sessionId: string): boolean + analyzeCreateSession(filePath: string, formatOptions?: Record): Promise + analyzeAppendSession( + sessionId: string, + filePath: string, + formatOptions?: Record + ): Promise + resolveTarget?: typeof resolveAutoImportTarget +} + export interface AutoImportResult { success: boolean sessionId?: string @@ -47,6 +59,48 @@ export interface AutoImportResult { error?: string } +export interface AutoImportAnalysisResult { + success: boolean + importMode?: 'created' | 'incremental' + sessionId?: string + matchedBy?: AutoImportMatchMethod + createReason?: AutoImportCreateReason + totalMessageCount?: number + newMessageCount?: number + duplicateCount?: number + totalMemberCount?: number + meta?: AnalyzeNewImportResult['meta'] + error?: string +} + +type AutoImportPlan = + | { action: 'incremental'; sessionId: string; matchedBy?: AutoImportMatchMethod } + | { action: 'create'; sessionId?: string; reason?: AutoImportCreateReason } + +async function planAutoImport( + filePath: string, + deps: Pick, + options: AutoImportOptions +): Promise { + if (options.explicitSessionId) { + if (!isValidImportSessionId(options.explicitSessionId)) { + throw new Error('sessionId contains invalid characters') + } + return deps.sessionExists(options.explicitSessionId) + ? { action: 'incremental', sessionId: options.explicitSessionId } + : { action: 'create', sessionId: options.explicitSessionId } + } + + const decision: AutoImportDecision = await (deps.resolveTarget ?? resolveAutoImportTarget)( + filePath, + deps, + options.formatOptions + ) + return decision.action === 'incremental' + ? { action: 'incremental', sessionId: decision.sessionId, matchedBy: decision.matchedBy } + : { action: 'create', reason: decision.reason } +} + function mapCreateResult(result: StreamImportResult, createReason?: AutoImportCreateReason): AutoImportResult { if (!result.success || !result.sessionId) { return { success: false, error: result.error, diagnostics: result.diagnostics } @@ -94,53 +148,22 @@ export async function autoImportFile( options: AutoImportOptions = {} ): Promise { try { - if (options.explicitSessionId) { - if (!isValidImportSessionId(options.explicitSessionId)) { - return { success: false, error: 'sessionId contains invalid characters' } - } - if (deps.sessionExists(options.explicitSessionId)) { - const result = mapIncrementalResult( - options.explicitSessionId, - await deps.appendSession(options.explicitSessionId, filePath, options.formatOptions) - ) - appLogger.info('import', 'Explicit incremental import completed', { - sessionId: options.explicitSessionId, - success: result.success, - newMessageCount: result.newMessageCount, - duplicateCount: result.duplicateCount, - }) - return result - } - - const result = mapCreateResult( - await deps.createSession(filePath, options.formatOptions, options.explicitSessionId) - ) - appLogger.info('import', 'Explicit session import completed', { - sessionId: result.sessionId, - success: result.success, - importMode: result.importMode, + if (!options.explicitSessionId) { + appLogger.info('import', 'Automatic session matching started', { + candidateCount: deps.listSessionIds().length, }) - return result } + const plan = await planAutoImport(filePath, deps, options) - appLogger.info('import', 'Automatic session matching started', { - candidateCount: deps.listSessionIds().length, - }) - const decision: AutoImportDecision = await (deps.resolveTarget ?? resolveAutoImportTarget)( - filePath, - deps, - options.formatOptions - ) - - if (decision.action === 'incremental') { + if (plan.action === 'incremental') { const result = mapIncrementalResult( - decision.sessionId, - await deps.appendSession(decision.sessionId, filePath, options.formatOptions), - decision.matchedBy + plan.sessionId, + await deps.appendSession(plan.sessionId, filePath, options.formatOptions), + plan.matchedBy ) - appLogger.info('import', 'Automatic incremental import completed', { - sessionId: decision.sessionId, - matchedBy: decision.matchedBy, + appLogger.info('import', 'Incremental import completed', { + sessionId: plan.sessionId, + matchedBy: plan.matchedBy, success: result.success, newMessageCount: result.newMessageCount, duplicateCount: result.duplicateCount, @@ -148,9 +171,12 @@ export async function autoImportFile( return result } - const result = mapCreateResult(await deps.createSession(filePath, options.formatOptions), decision.reason) - appLogger.info('import', 'Automatic import created a new session', { - reason: decision.reason, + const result = mapCreateResult( + await deps.createSession(filePath, options.formatOptions, plan.sessionId), + plan.reason + ) + appLogger.info('import', 'Import created a new session', { + reason: plan.reason, sessionId: result.sessionId, success: result.success, }) @@ -160,3 +186,57 @@ export async function autoImportFile( return { success: false, error: error instanceof Error ? error.message : String(error) } } } + +export async function analyzeAutoImportFile( + filePath: string, + deps: AutoImportAnalysisDeps, + options: AutoImportOptions = {} +): Promise { + try { + const plan = await planAutoImport(filePath, deps, options) + + if (plan.action === 'incremental') { + const analysis = await deps.analyzeAppendSession(plan.sessionId, filePath, options.formatOptions) + if (analysis.error) return { success: false, error: analysis.error } + const result: AutoImportAnalysisResult = { + success: true, + importMode: 'incremental', + sessionId: plan.sessionId, + ...(plan.matchedBy ? { matchedBy: plan.matchedBy } : {}), + totalMessageCount: analysis.totalInFile, + newMessageCount: analysis.newMessageCount, + duplicateCount: analysis.duplicateCount, + } + appLogger.info('import', 'Incremental import analysis completed', { + sessionId: plan.sessionId, + matchedBy: plan.matchedBy, + newMessageCount: analysis.newMessageCount, + duplicateCount: analysis.duplicateCount, + }) + return result + } + + const analysis = await deps.analyzeCreateSession(filePath, options.formatOptions) + if (analysis.error) return { success: false, error: analysis.error } + const result: AutoImportAnalysisResult = { + success: true, + importMode: 'created', + ...(plan.sessionId ? { sessionId: plan.sessionId } : {}), + ...(plan.reason ? { createReason: plan.reason } : {}), + totalMessageCount: analysis.totalMessages, + newMessageCount: analysis.newMessageCount, + duplicateCount: analysis.duplicateCount, + totalMemberCount: analysis.totalMembers, + meta: analysis.meta, + } + appLogger.info('import', 'New session import analysis completed', { + reason: plan.reason, + messageCount: analysis.totalMessages, + memberCount: analysis.totalMembers, + }) + return result + } catch (error) { + appLogger.error('import', 'Automatic import analysis failed', error) + return { success: false, error: error instanceof Error ? error.message : String(error) } + } +} diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index 9afc0ff3..52e5cbd7 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -148,9 +148,10 @@ export function normalizeImportTimestamp(timestamp: unknown): number | null { export async function analyzeIncrementalImport( sessionId: string, filePath: string, - deps: IncrementalImportDeps + deps: IncrementalImportDeps, + options?: ImportOptions ): Promise { - const formatFeature = detectFormat(filePath) + const formatFeature = options?.formatId ? getFormatFeatureById(options.formatId) : detectFormat(filePath) if (!formatFeature) { return { error: 'error.unrecognized_format', newMessageCount: 0, duplicateCount: 0, totalInFile: 0 } } @@ -169,26 +170,31 @@ export async function analyzeIncrementalImport( let newMessageCount = 0 let duplicateCount = 0 - await streamParseFile(filePath, { - onMeta: () => {}, - onMembers: () => {}, - onProgress: (progress: ParseProgress) => { - deps.onProgress(progress) - }, - onMessageBatch: (batch) => { - for (const msg of batch) { - totalInFile++ - const timestamp = normalizeImportTimestamp(msg.timestamp) - if (timestamp === null) continue + await streamParseFile( + filePath, + { + formatOptions: options?.chatIndex === undefined ? undefined : { chatIndex: options.chatIndex }, + onMeta: () => {}, + onMembers: () => {}, + onProgress: (progress: ParseProgress) => { + deps.onProgress(progress) + }, + onMessageBatch: (batch) => { + for (const msg of batch) { + totalInFile++ + const timestamp = normalizeImportTimestamp(msg.timestamp) + if (timestamp === null) continue - if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys)) { - duplicateCount++ - } else { - newMessageCount++ + if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys)) { + duplicateCount++ + } else { + newMessageCount++ + } } - } + }, }, - }) + options?.formatId + ) return { newMessageCount, duplicateCount, totalInFile } } diff --git a/packages/node-runtime/src/import/index.ts b/packages/node-runtime/src/import/index.ts index f102437b..84f61bd0 100644 --- a/packages/node-runtime/src/import/index.ts +++ b/packages/node-runtime/src/import/index.ts @@ -21,6 +21,7 @@ export type { ImportProgressCallback, ImportLogger, StreamImportDeps, + AnalyzeNewImportOptions, AnalyzeNewImportResult, StreamParseFileInfoResult, StreamParseFileInfoDeps, @@ -41,8 +42,14 @@ export type { AutoImportMatcherDeps, AutoImportMatchMethod, } from './auto-import-matcher' -export { autoImportFile } from './auto-importer' -export type { AutoImportDeps, AutoImportOptions, AutoImportResult } from './auto-importer' +export { analyzeAutoImportFile, autoImportFile } from './auto-importer' +export type { + AutoImportAnalysisDeps, + AutoImportAnalysisResult, + AutoImportDeps, + AutoImportOptions, + AutoImportResult, +} from './auto-importer' export { IMPORT_IN_PROGRESS_ERROR_KEY, IMPORT_LOCK_FILENAME, diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index b73cfdc7..1b9c42f9 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/packages/node-runtime/src/import/streaming-importer.test.ts @@ -215,3 +215,18 @@ test('streamingImport applies incremental-equivalent deduplication on first impo db.close() assert.equal(row.count, 3) }) + +test('analyzeNewImport honors an explicitly selected parser format', async (t) => { + const root = makeTempDir() + t.after(() => fs.rmSync(root, { recursive: true, force: true })) + const detectedPath = writeDuplicateChatLabExport(root) + const filePath = path.join(root, 'explicit-format.txt') + fs.renameSync(detectedPath, filePath) + + const analysis = await analyzeNewImport(filePath, () => {}, { formatId: 'chatlab' }) + + assert.equal(analysis.error, undefined) + assert.equal(analysis.totalMessages, 5) + assert.equal(analysis.newMessageCount, 3) + assert.equal(analysis.duplicateCount, 2) +}) diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index 715ab235..d566eb7e 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -679,11 +679,17 @@ export interface AnalyzeNewImportResult { error?: string } +export interface AnalyzeNewImportOptions { + formatId?: string + chatIndex?: number +} + export async function analyzeNewImport( filePath: string, - onProgress: ImportProgressCallback + onProgress: ImportProgressCallback, + options?: AnalyzeNewImportOptions ): Promise { - const formatFeature = detectFormat(filePath) + const formatFeature = options?.formatId ? getFormatFeatureById(options.formatId) : detectFormat(filePath) if (!formatFeature) { return { totalMessages: 0, @@ -702,31 +708,36 @@ export async function analyzeNewImport( let newMessageCount = 0 let duplicateCount = 0 - await streamParseFile(filePath, { - onMeta: (parsedMeta: ParsedMeta) => { - meta = { name: parsedMeta.name, platform: parsedMeta.platform, type: parsedMeta.type } - }, - onMembers: (members: ParsedMember[]) => { - for (const m of members) memberSet.add(m.platformId) - }, - onProgress: (progress: ParseProgress) => { - onProgress(progress) - }, - onMessageBatch: (batch: ParsedMessage[]) => { - for (const msg of batch) { - totalMessages++ - if (!memberSet.has(msg.senderPlatformId)) memberSet.add(msg.senderPlatformId) + await streamParseFile( + filePath, + { + formatOptions: options?.chatIndex === undefined ? undefined : { chatIndex: options.chatIndex }, + onMeta: (parsedMeta: ParsedMeta) => { + meta = { name: parsedMeta.name, platform: parsedMeta.platform, type: parsedMeta.type } + }, + onMembers: (members: ParsedMember[]) => { + for (const m of members) memberSet.add(m.platformId) + }, + onProgress: (progress: ParseProgress) => { + onProgress(progress) + }, + onMessageBatch: (batch: ParsedMessage[]) => { + for (const msg of batch) { + totalMessages++ + if (!memberSet.has(msg.senderPlatformId)) memberSet.add(msg.senderPlatformId) - const prepared = prepareMessageForCreate(msg) - if ('skipCounter' in prepared) continue - if (registerMessageAndCheckDuplicate(prepared.message, dedupState)) { - duplicateCount++ - } else { - newMessageCount++ + const prepared = prepareMessageForCreate(msg) + if ('skipCounter' in prepared) continue + if (registerMessageAndCheckDuplicate(prepared.message, dedupState)) { + duplicateCount++ + } else { + newMessageCount++ + } } - } + }, }, - }) + options?.formatId + ) return { totalMessages, newMessageCount, duplicateCount, totalMembers: memberSet.size, meta } } diff --git a/packages/node-runtime/src/index.ts b/packages/node-runtime/src/index.ts index 7643339a..ef714074 100644 --- a/packages/node-runtime/src/index.ts +++ b/packages/node-runtime/src/index.ts @@ -16,6 +16,7 @@ export { analyzeIncrementalImport, incrementalImport, resolveAutoImportTarget, + analyzeAutoImportFile, autoImportFile, isValidImportSessionId, IMPORT_IN_PROGRESS_ERROR_KEY, @@ -49,6 +50,8 @@ export type { AutoImportCreateReason, AutoImportMatcherDeps, AutoImportMatchMethod, + AutoImportAnalysisDeps, + AutoImportAnalysisResult, AutoImportDeps, AutoImportOptions, AutoImportResult,