feat(server): add chatlab import command with ChatLab JSON/JSONL parser

Implement self-contained ChatLab format reader (JSON and JSONL) and
database importer in packages/server/src/import/. Supports full import
(new session) and incremental import (dedup via platform_message_id or
content hash). Includes FTS index creation and member upsert.

CLI: chatlab import <file> [--session-id <id>] [--dry-run]
This commit is contained in:
digua committed 2026-05-11 20:37:39 +08:00
1 parent 95e4a1b5ee
commit 847d75e4b6
4 files changed
+679

No files matched your search

+59
View File
@@ -200,6 +200,65 @@ program
dbManager.closeAll()
})
// chatlab import <file> - 导入聊天记录
program
.command('import <file>')
.description('导入聊天记录文件(支持 ChatLab JSON / JSONL 格式)')
.option('--session-id <id>', '指定会话 ID(不指定则自动生成,已存在则增量导入)')
.option('--dry-run', '试运行,不实际写入', false)
.action(async (file, options) => {
if (!fs.existsSync(file)) {
console.error(`文件不存在: ${file}`)
process.exit(1)
}
const { parseFile, importData } = await import('./import')
const { dbManager } = initRuntime()
console.log(`解析文件: ${file}`)
try {
const data = await parseFile(file, (processed, total) => {
const pct = total > 0 ? Math.round((processed / total) * 100) : 0
process.stdout.write(`\r 解析进度: ${pct}%`)
})
console.log(`\n 解析完成: ${data.messages.length} 条消息, ${data.members.length} 个成员`)
if (options.dryRun) {
console.log(`\n[试运行] 将${options.sessionId ? '增量导入到 ' + options.sessionId : '创建新会话'}`)
console.log(` 消息数: ${data.messages.length}`)
console.log(` 成员数: ${data.members.length}`)
dbManager.closeAll()
return
}
const nativeBinding = resolveNativeBinding()
const result = await importData(dbManager, data, {
sessionId: options.sessionId,
nativeBinding,
onProgress: (msg) => console.log(` ${msg}`),
})
if (result.success) {
console.log(`\n导入成功!`)
console.log(` 会话 ID: ${result.sessionId}`)
console.log(` ${result.created ? '新建' : '增量'}导入`)
console.log(` 写入消息: ${result.messageCount}`)
console.log(` 成员数: ${result.memberCount}`)
if (result.duplicateCount > 0) {
console.log(` 重复跳过: ${result.duplicateCount}`)
}
} else {
console.error(`\n导入失败: ${result.error}`)
process.exit(1)
}
} catch (err) {
console.error(`\n导入错误: ${err instanceof Error ? err.message : err}`)
process.exit(1)
} finally {
dbManager.closeAll()
}
})
// chatlab mcp - MCP Server
program
.command('mcp')
@@ -0,0 +1,244 @@
/**
* ChatLab 格式读取器
*
* 支持 ChatLab JSON (.json) 和 ChatLab JSONL (.jsonl) 两种格式。
* 不依赖 stream-json,使用原生 readline 和 JSON.parse 实现。
*/
import * as fs from 'fs'
import * as readline from 'readline'
export interface ImportMeta {
name: string
platform: string
type: string
groupId?: string
groupAvatar?: string
ownerId?: string
}
export interface ImportMember {
platformId: string
accountName: string
groupNickname?: string
avatar?: string
roles?: Array<{ id: string; name?: string }>
}
export interface ImportMessage {
platformMessageId?: string
senderPlatformId: string
senderAccountName: string
senderGroupNickname?: string
timestamp: number
type: number
content: string | null
replyToMessageId?: string
}
export interface ParsedData {
meta: ImportMeta
members: ImportMember[]
messages: ImportMessage[]
}
export type ProgressCallback = (processed: number, total: number) => void
/**
* 检测文件是否为 ChatLab 格式
*/
export function detectFormat(filePath: string): 'chatlab-json' | 'chatlab-jsonl' | null {
const ext = filePath.toLowerCase()
if (ext.endsWith('.jsonl')) {
const head = fs.readFileSync(filePath, { encoding: 'utf-8', flag: 'r' }).slice(0, 1024)
if (head.includes('"_type"') && head.includes('"chatlab"')) {
return 'chatlab-jsonl'
}
return null
}
if (ext.endsWith('.json')) {
const head = fs.readFileSync(filePath, { encoding: 'utf-8', flag: 'r' }).slice(0, 4096)
if (/"chatlab"\s*:\s*\{/.test(head)) {
return 'chatlab-json'
}
return null
}
return null
}
/**
* 解析 ChatLab JSON 格式文件
*/
export async function parseChatLabJson(
filePath: string,
onProgress?: ProgressCallback
): Promise<ParsedData> {
const raw = fs.readFileSync(filePath, 'utf-8')
const data = JSON.parse(raw)
if (!data.chatlab || typeof data.chatlab !== 'object') {
throw new Error('Invalid ChatLab JSON: missing "chatlab" field')
}
const meta: ImportMeta = {
name: data.meta?.name || '未知',
platform: data.meta?.platform || 'unknown',
type: data.meta?.type || 'group',
groupId: data.meta?.groupId,
groupAvatar: data.meta?.groupAvatar,
ownerId: data.meta?.ownerId,
}
const members: ImportMember[] = (data.members || []).map((m: Record<string, unknown>) => ({
platformId: String(m.platformId || ''),
accountName: String(m.accountName || m.platformId || ''),
groupNickname: m.groupNickname ? String(m.groupNickname) : undefined,
avatar: m.avatar ? String(m.avatar) : undefined,
roles: m.roles as ImportMember['roles'],
}))
const rawMessages = data.messages || []
const totalMessages = rawMessages.length
const messages: ImportMessage[] = rawMessages.map(
(msg: Record<string, unknown>, i: number) => {
if (onProgress && i % 10000 === 0) {
onProgress(i, totalMessages)
}
return {
platformMessageId: msg.platformMessageId ? String(msg.platformMessageId) : undefined,
senderPlatformId: String(msg.sender || ''),
senderAccountName: String(msg.accountName || msg.sender || ''),
senderGroupNickname: msg.groupNickname ? String(msg.groupNickname) : undefined,
timestamp: Number(msg.timestamp),
type: Number(msg.type ?? 0),
content: msg.content != null ? String(msg.content) : null,
replyToMessageId: msg.replyToMessageId ? String(msg.replyToMessageId) : undefined,
}
}
)
if (onProgress) onProgress(totalMessages, totalMessages)
if (members.length === 0) {
collectMembersFromMessages(messages, members)
}
return { meta, members, messages }
}
/**
* 解析 ChatLab JSONL 格式文件
*/
export async function parseChatLabJsonl(
filePath: string,
onProgress?: ProgressCallback
): Promise<ParsedData> {
const meta: ImportMeta = { name: '未知', platform: 'unknown', type: 'group' }
const members: ImportMember[] = []
const messages: ImportMessage[] = []
const stat = fs.statSync(filePath)
let bytesRead = 0
const rl = readline.createInterface({
input: fs.createReadStream(filePath, 'utf-8'),
crlfDelay: Infinity,
})
for await (const line of rl) {
bytesRead += Buffer.byteLength(line, 'utf-8') + 1
const trimmed = line.trim()
if (!trimmed) continue
let obj: Record<string, unknown>
try {
obj = JSON.parse(trimmed)
} catch {
continue
}
const type = obj._type as string
if (type === 'header' || (!type && obj.chatlab)) {
const m = obj.meta as Record<string, unknown> | undefined
if (m) {
meta.name = String(m.name || '未知')
meta.platform = String(m.platform || 'unknown')
meta.type = String(m.type || 'group')
if (m.groupId) meta.groupId = String(m.groupId)
if (m.groupAvatar) meta.groupAvatar = String(m.groupAvatar)
if (m.ownerId) meta.ownerId = String(m.ownerId)
}
} else if (type === 'member') {
members.push({
platformId: String(obj.platformId || ''),
accountName: String(obj.accountName || obj.platformId || ''),
groupNickname: obj.groupNickname ? String(obj.groupNickname) : undefined,
avatar: obj.avatar ? String(obj.avatar) : undefined,
roles: obj.roles as ImportMember['roles'],
})
} else if (type === 'message') {
messages.push({
platformMessageId: obj.platformMessageId ? String(obj.platformMessageId) : undefined,
senderPlatformId: String(obj.sender || obj.senderPlatformId || ''),
senderAccountName: String(obj.accountName || obj.senderAccountName || obj.sender || ''),
senderGroupNickname: obj.groupNickname
? String(obj.groupNickname)
: obj.senderGroupNickname
? String(obj.senderGroupNickname)
: undefined,
timestamp: Number(obj.timestamp),
type: Number(obj.type ?? 0),
content: obj.content != null ? String(obj.content) : null,
replyToMessageId: obj.replyToMessageId ? String(obj.replyToMessageId) : undefined,
})
if (onProgress && messages.length % 10000 === 0) {
onProgress(bytesRead, stat.size)
}
}
}
if (onProgress) onProgress(stat.size, stat.size)
if (members.length === 0) {
collectMembersFromMessages(messages, members)
}
return { meta, members, messages }
}
function collectMembersFromMessages(messages: ImportMessage[], members: ImportMember[]): void {
const seen = new Set<string>()
for (const msg of messages) {
if (!seen.has(msg.senderPlatformId)) {
seen.add(msg.senderPlatformId)
members.push({
platformId: msg.senderPlatformId,
accountName: msg.senderAccountName || msg.senderPlatformId,
groupNickname: msg.senderGroupNickname,
})
}
}
}
/**
* 统一入口:自动检测格式并解析
*/
export async function parseFile(
filePath: string,
onProgress?: ProgressCallback
): Promise<ParsedData> {
const format = detectFormat(filePath)
if (!format) {
throw new Error(`Unsupported file format: ${filePath}. Only ChatLab JSON (.json) and JSONL (.jsonl) are supported.`)
}
if (format === 'chatlab-json') {
return parseChatLabJson(filePath, onProgress)
} else {
return parseChatLabJsonl(filePath, onProgress)
}
}
+372
View File
@@ -0,0 +1,372 @@
/**
* ChatLab 数据导入器
*
* 将解析后的数据写入 SQLite 数据库。
* 支持新建导入和增量导入(基于 platform_message_id 或内容哈希去重)。
*/
import * as fs from 'fs'
import * as crypto from 'crypto'
import type { DatabaseAdapter } from '@openchatlab/core'
import { CHAT_DB_SCHEMA, FTS_TABLE_SCHEMA } from '@openchatlab/core'
import type { DatabaseManager } from '@openchatlab/node-runtime'
import { openBetterSqliteDatabase } from '@openchatlab/node-runtime'
import type { ParsedData, ImportMessage } from './chatlab-reader'
export interface ImportResult {
success: boolean
sessionId: string
created: boolean
messageCount: number
memberCount: number
duplicateCount: number
error?: string
}
export interface ImportOptions {
sessionId?: string
dryRun?: boolean
nativeBinding?: string
}
function generateSessionId(): string {
const ts = Date.now()
const rand = crypto.randomBytes(4).toString('hex')
return `chat_${ts}_${rand}`
}
function generateMessageKey(ts: number, senderPlatformId: string, content: string | null): string {
const raw = `${ts}|${senderPlatformId}|${content ?? ''}`
return crypto.createHash('sha256').update(raw).digest('hex').slice(0, 16)
}
/**
* 新建导入:创建新的会话数据库并导入全部数据
*/
function fullImport(
db: DatabaseAdapter,
data: ParsedData,
onProgress?: (msg: string) => void
): { messageCount: number; memberCount: number; duplicateCount: number } {
db.exec(CHAT_DB_SCHEMA)
onProgress?.('写入元信息...')
db.prepare(
`INSERT INTO meta (name, platform, type, imported_at, group_id, group_avatar, owner_id, schema_version)
VALUES (?, ?, ?, ?, ?, ?, ?, 4)`
).run(
data.meta.name,
data.meta.platform,
data.meta.type,
Math.floor(Date.now() / 1000),
data.meta.groupId || null,
data.meta.groupAvatar || null,
data.meta.ownerId || null
)
onProgress?.(`写入 ${data.members.length} 个成员...`)
const insertMember = db.prepare(
`INSERT OR IGNORE INTO member (platform_id, account_name, group_nickname, avatar, roles)
VALUES (?, ?, ?, ?, ?)`
)
for (const m of data.members) {
insertMember.run(
m.platformId,
m.accountName || m.platformId,
m.groupNickname || null,
m.avatar || null,
m.roles ? JSON.stringify(m.roles) : '[]'
)
}
const memberIdMap = buildMemberIdMap(db)
onProgress?.(`写入 ${data.messages.length} 条消息...`)
const insertMsg = db.prepare(
`INSERT INTO message (sender_id, sender_account_name, sender_group_nickname, ts, type, content, reply_to_message_id, platform_message_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
)
let written = 0
let skipped = 0
const BATCH = 5000
for (let i = 0; i < data.messages.length; i += BATCH) {
const batch = data.messages.slice(i, i + BATCH)
db.transaction(() => {
for (const msg of batch) {
let senderId = memberIdMap.get(msg.senderPlatformId)
if (!senderId) {
insertMember.run(msg.senderPlatformId, msg.senderAccountName || msg.senderPlatformId, msg.senderGroupNickname || null, null, '[]')
senderId = (db.prepare('SELECT id FROM member WHERE platform_id = ?').get(msg.senderPlatformId) as { id: number })?.id
if (senderId) memberIdMap.set(msg.senderPlatformId, senderId)
}
if (!senderId) {
skipped++
continue
}
insertMsg.run(
senderId,
msg.senderAccountName || null,
msg.senderGroupNickname || null,
msg.timestamp,
msg.type,
msg.content,
msg.replyToMessageId || null,
msg.platformMessageId || null
)
written++
}
})
if (onProgress && (i + BATCH) % 50000 < BATCH) {
onProgress(`已写入 ${Math.min(i + BATCH, data.messages.length)} / ${data.messages.length} 条消息`)
}
}
buildFts(db, onProgress)
return { messageCount: written, memberCount: data.members.length, duplicateCount: skipped }
}
/**
* 增量导入:向已有数据库追加新消息,自动去重
*/
function incrementalImport(
db: DatabaseAdapter,
data: ParsedData,
onProgress?: (msg: string) => void
): { messageCount: number; memberCount: number; duplicateCount: number } {
onProgress?.('加载去重索引...')
const existingKeys = new Set<string>()
const existingPmids = db
.prepare('SELECT platform_message_id FROM message WHERE platform_message_id IS NOT NULL')
.all() as Array<{ platform_message_id: string }>
for (const row of existingPmids) {
existingKeys.add(`pmid:${row.platform_message_id}`)
}
const existingHashes = db
.prepare(
`SELECT msg.ts, m.platform_id, msg.content
FROM message msg
JOIN member m ON msg.sender_id = m.id
WHERE msg.platform_message_id IS NULL`
)
.all() as Array<{ ts: number; platform_id: string; content: string | null }>
for (const row of existingHashes) {
existingKeys.add(`hash:${generateMessageKey(row.ts, row.platform_id, row.content)}`)
}
onProgress?.(`去重索引: ${existingKeys.size} 条已有消息`)
const upsertMember = db.prepare(
`INSERT INTO member (platform_id, account_name, group_nickname, avatar, roles)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(platform_id) DO UPDATE SET
account_name = COALESCE(NULLIF(excluded.account_name, ''), account_name),
group_nickname = COALESCE(excluded.group_nickname, group_nickname)`
)
let newMembers = 0
for (const m of data.members) {
const result = upsertMember.run(
m.platformId,
m.accountName || m.platformId,
m.groupNickname || null,
m.avatar || null,
m.roles ? JSON.stringify(m.roles) : '[]'
)
if (result.changes > 0) newMembers++
}
const memberIdMap = buildMemberIdMap(db)
const insertMsg = db.prepare(
`INSERT INTO message (sender_id, sender_account_name, sender_group_nickname, ts, type, content, reply_to_message_id, platform_message_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
)
const insertMemberMinimal = db.prepare(
`INSERT OR IGNORE INTO member (platform_id, account_name) VALUES (?, ?)`
)
let written = 0
let duplicates = 0
const BATCH = 5000
for (let i = 0; i < data.messages.length; i += BATCH) {
const batch = data.messages.slice(i, i + BATCH)
db.transaction(() => {
for (const msg of batch) {
if (isDuplicate(msg, existingKeys)) {
duplicates++
continue
}
let senderId = memberIdMap.get(msg.senderPlatformId)
if (!senderId) {
insertMemberMinimal.run(msg.senderPlatformId, msg.senderAccountName || msg.senderPlatformId)
senderId = (db.prepare('SELECT id FROM member WHERE platform_id = ?').get(msg.senderPlatformId) as { id: number })?.id
if (senderId) memberIdMap.set(msg.senderPlatformId, senderId)
}
if (!senderId) continue
insertMsg.run(
senderId,
msg.senderAccountName || null,
msg.senderGroupNickname || null,
msg.timestamp,
msg.type,
msg.content,
msg.replyToMessageId || null,
msg.platformMessageId || null
)
written++
}
})
if (onProgress && (i + BATCH) % 50000 < BATCH) {
onProgress(`已处理 ${Math.min(i + BATCH, data.messages.length)} / ${data.messages.length} 条消息 (新增 ${written}, 重复 ${duplicates})`)
}
}
if (written > 0) {
insertFtsEntries(db, onProgress)
db.prepare('UPDATE meta SET imported_at = ?').run(Math.floor(Date.now() / 1000))
}
return { messageCount: written, memberCount: newMembers, duplicateCount: duplicates }
}
function isDuplicate(msg: ImportMessage, existingKeys: Set<string>): boolean {
if (msg.platformMessageId) {
return existingKeys.has(`pmid:${msg.platformMessageId}`)
}
const key = generateMessageKey(msg.timestamp, msg.senderPlatformId, msg.content)
return existingKeys.has(`hash:${key}`)
}
function buildMemberIdMap(db: DatabaseAdapter): Map<string, number> {
const rows = db.prepare('SELECT id, platform_id FROM member').all() as Array<{ id: number; platform_id: string }>
const map = new Map<string, number>()
for (const row of rows) {
map.set(row.platform_id, row.id)
}
return map
}
function buildFts(db: DatabaseAdapter, onProgress?: (msg: string) => void): void {
onProgress?.('创建全文搜索索引...')
db.exec(FTS_TABLE_SCHEMA)
const textMessages = db
.prepare("SELECT id, content FROM message WHERE type = 0 AND content IS NOT NULL AND content != ''")
.all() as Array<{ id: number; content: string }>
const insertFts = db.prepare('INSERT INTO message_fts(rowid, content) VALUES (?, ?)')
const BATCH = 5000
for (let i = 0; i < textMessages.length; i += BATCH) {
const batch = textMessages.slice(i, i + BATCH)
db.transaction(() => {
for (const row of batch) {
insertFts.run(row.id, row.content)
}
})
}
}
function insertFtsEntries(db: DatabaseAdapter, onProgress?: (msg: string) => void): void {
try {
db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='message_fts'").get()
} catch {
return
}
const hasFts = db.prepare("SELECT COUNT(*) as cnt FROM sqlite_master WHERE type='table' AND name='message_fts'").get() as { cnt: number }
if (!hasFts || hasFts.cnt === 0) return
onProgress?.('更新全文搜索索引...')
const maxRowid = (db.prepare('SELECT MAX(rowid) as m FROM message_fts').get() as { m: number | null })?.m ?? 0
const newRows = db
.prepare("SELECT id, content FROM message WHERE type = 0 AND content IS NOT NULL AND content != '' AND id > ?")
.all(maxRowid) as Array<{ id: number; content: string }>
if (newRows.length === 0) return
const insertFts = db.prepare('INSERT INTO message_fts(rowid, content) VALUES (?, ?)')
db.transaction(() => {
for (const row of newRows) {
insertFts.run(row.id, row.content)
}
})
}
/**
* 执行导入(新建或增量)
*/
export async function importData(
dbManager: DatabaseManager,
data: ParsedData,
options?: ImportOptions & { onProgress?: (msg: string) => void }
): Promise<ImportResult> {
const sessionId = options?.sessionId || generateSessionId()
const dbPath = dbManager.getDbPath(sessionId)
const exists = fs.existsSync(dbPath)
if (options?.dryRun) {
if (exists) {
const db = dbManager.open(sessionId)
if (!db) return { success: false, sessionId, created: false, messageCount: 0, memberCount: 0, duplicateCount: 0, error: 'Failed to open database' }
const existingCount = (db.prepare('SELECT COUNT(*) as cnt FROM message').get() as { cnt: number }).cnt
return {
success: true,
sessionId,
created: false,
messageCount: data.messages.length,
memberCount: data.members.length,
duplicateCount: 0,
error: `Dry run: ${data.messages.length} messages to import into existing session (${existingCount} existing messages)`,
}
}
return {
success: true,
sessionId,
created: true,
messageCount: data.messages.length,
memberCount: data.members.length,
duplicateCount: 0,
error: `Dry run: would create new session with ${data.messages.length} messages`,
}
}
try {
if (exists) {
dbManager.close(sessionId)
const db = openBetterSqliteDatabase(dbPath, { readonly: false, nativeBinding: options?.nativeBinding })
try {
const result = incrementalImport(db, data, options?.onProgress)
db.close()
return { success: true, sessionId, created: false, ...result }
} catch (err) {
db.close()
throw err
}
}
const db = openBetterSqliteDatabase(dbPath, { nativeBinding: options?.nativeBinding })
const result = fullImport(db, data, options?.onProgress)
db.close()
return { success: true, sessionId, created: true, ...result }
} catch (err: unknown) {
const message = err instanceof Error ? err.message : String(err)
if (!exists) {
try { fs.unlinkSync(dbPath) } catch { /* ignore */ }
try { fs.unlinkSync(dbPath + '-wal') } catch { /* ignore */ }
try { fs.unlinkSync(dbPath + '-shm') } catch { /* ignore */ }
}
return { success: false, sessionId, created: false, messageCount: 0, memberCount: 0, duplicateCount: 0, error: message }
}
}
+4
View File
@@ -0,0 +1,4 @@
export { parseFile, detectFormat } from './chatlab-reader'
export type { ParsedData, ImportMeta, ImportMember, ImportMessage, ProgressCallback } from './chatlab-reader'
export { importData } from './importer'
export type { ImportResult, ImportOptions } from './importer'