mirror of
https://github.com/hellodigua/ChatLab.git
synced 2026-10-25 17:54:34 +08:00
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.
This commit is contained in:
1 parent
1271c16f17
commit
53406f4c6d
14 files changed
+677
-186
No files matched your search
+3
-88
@@ -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 <file>')
|
||||
.description('Import a chat history file (14+ formats: QQ/WeChat/Telegram/WhatsApp/LINE/Discord/Instagram, etc.)')
|
||||
.option('--session-id <id>', 'Force an existing session target, or create with this ID if missing')
|
||||
.option('--format <id>', '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 <id> 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')
|
||||
|
||||
@@ -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'])
|
||||
})
|
||||
@@ -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<ImportCommandEnvelope['error']> {
|
||||
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 <id>.',
|
||||
}
|
||||
}
|
||||
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<T>(enabled: boolean, run: () => Promise<T>): Promise<T> {
|
||||
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<typeof initRuntime>['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 <file>')
|
||||
.description('Import a chat history file (14+ formats: QQ/WeChat/Telegram/WhatsApp/LINE/Discord, etc.)')
|
||||
.option('--session-id <id>', 'Force an existing session target, or create with this ID if missing')
|
||||
.option('--format <id>', '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<typeof initRuntime>['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
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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 }))
|
||||
|
||||
@@ -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'
|
||||
@@ -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)
|
||||
})
|
||||
@@ -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<AutoImportAnalysisResult> {
|
||||
const { formatId, chatIndex, onProgress, sessionId } = options ?? {}
|
||||
const formatOptions: Record<string, unknown> = {}
|
||||
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<IncrementalAnalyzeResult> {
|
||||
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<ImportOptions, 'formatId' | 'chatIndex'>
|
||||
): Promise<AnalyzeNewImportResult> {
|
||||
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,
|
||||
}
|
||||
@@ -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<string, unknown> }>,
|
||||
append: [] as Array<{ sessionId: string; filePath: string; formatOptions?: Record<string, unknown> }>,
|
||||
}
|
||||
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, [])
|
||||
})
|
||||
@@ -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<string, unknown>): Promise<AnalyzeNewImportResult>
|
||||
analyzeAppendSession(
|
||||
sessionId: string,
|
||||
filePath: string,
|
||||
formatOptions?: Record<string, unknown>
|
||||
): Promise<IncrementalAnalyzeResult>
|
||||
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<AutoImportDeps, 'listSessionIds' | 'openReadonly' | 'onProgress' | 'sessionExists' | 'resolveTarget'>,
|
||||
options: AutoImportOptions
|
||||
): Promise<AutoImportPlan> {
|
||||
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<AutoImportResult> {
|
||||
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<AutoImportAnalysisResult> {
|
||||
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) }
|
||||
}
|
||||
}
|
||||
@@ -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<IncrementalAnalyzeResult> {
|
||||
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 }
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
@@ -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<AnalyzeNewImportResult> {
|
||||
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 }
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user