mirror of
https://github.com/hellodigua/ChatLab.git
synced 2026-10-26 02:04:37 +08:00
feat(import): add Rust native parsing kernels for WeFlow and ChatLab JSON
Add @openchatlab/parser-native (napi-rs v3) with a shared byte-level JSON scanner and two format kernels. Parsing large exports is 4-10x faster in real-data benchmarks. The native module is an optional local build; the pure-TS parsers remain as automatic fallback when it is unavailable, disabled via CHATLAB_DISABLE_NATIVE_PERF=1, or rejects off-spec files. The ChatLab kernel also parses the members array structurally, fixing the TS head-regex heuristic that silently dropped avatars/roles for files with nested arrays or long member lists. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
1 parent
bbf89332b2
commit
1e7aefe08b
29 files changed
+4953
-6
No files matched your search
@@ -7,6 +7,7 @@
|
||||
"main": "./src/index.ts",
|
||||
"types": "./src/index.ts",
|
||||
"dependencies": {
|
||||
"@openchatlab/parser-native": "workspace:*",
|
||||
"@openchatlab/shared-types": "workspace:*",
|
||||
"stream-json": "^1.9.1"
|
||||
},
|
||||
|
||||
@@ -276,9 +276,15 @@ async function* parseChatLab(options: ParseOptions): AsyncGenerator<ParseEvent,
|
||||
|
||||
// ==================== 导出解析器 ====================
|
||||
|
||||
import { withNativeChatlab } from '../native/chatlab-native'
|
||||
|
||||
// parseChatLabAccelerated:优先走 Rust 内核,native 不可用/失败时自动回退本文件的 TS 实现
|
||||
export { parseChatLab }
|
||||
export const parseChatLabAccelerated = withNativeChatlab(parseChatLab)
|
||||
|
||||
export const parser_: Parser = {
|
||||
feature,
|
||||
parse: parseChatLab,
|
||||
parse: parseChatLabAccelerated,
|
||||
}
|
||||
|
||||
// ==================== 导出格式模块 ====================
|
||||
|
||||
@@ -606,12 +606,16 @@ async function* parseWeFlow(options: ParseOptions): AsyncGenerator<ParseEvent, v
|
||||
|
||||
// ==================== 导出解析器 ====================
|
||||
|
||||
import { withNativeWeflow } from '../native/weflow-native'
|
||||
|
||||
// 导出解析函数供其他格式复用(如 ycccccccy-echotrace)
|
||||
// parseWeFlowAccelerated:优先走 Rust 内核,native 不可用/失败时自动回退本文件的 TS 实现
|
||||
export { parseWeFlow }
|
||||
export const parseWeFlowAccelerated = withNativeWeflow(parseWeFlow)
|
||||
|
||||
export const parser_: Parser = {
|
||||
feature,
|
||||
parse: parseWeFlow,
|
||||
parse: parseWeFlowAccelerated,
|
||||
}
|
||||
|
||||
// ==================== 预处理器(预留) ====================
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
import { KNOWN_PLATFORMS } from '@openchatlab/shared-types'
|
||||
import type { FormatFeature, FormatModule, Parser } from '../types'
|
||||
import { parseWeFlow } from './weflow'
|
||||
import { parseWeFlowAccelerated } from './weflow'
|
||||
import { weflowPreprocessor } from './weflow-preprocessor'
|
||||
|
||||
// ==================== 特征定义 ====================
|
||||
@@ -37,7 +37,7 @@ export const feature: FormatFeature = {
|
||||
// 复用 WeFlow 的解析逻辑(两种格式的数据结构相同)
|
||||
export const parser_: Parser = {
|
||||
feature,
|
||||
parse: parseWeFlow,
|
||||
parse: parseWeFlowAccelerated,
|
||||
}
|
||||
|
||||
// ==================== 预处理器(复用 WeFlow) ====================
|
||||
|
||||
@@ -0,0 +1,220 @@
|
||||
/**
|
||||
* Parity tests: the Rust ChatLab kernel must produce identical ParseResult
|
||||
* output to the pure-TS parser for spec-compliant files, and must fall back
|
||||
* to the TS parser for off-spec files.
|
||||
*
|
||||
* Skipped automatically when the native module has not been built locally
|
||||
* (pnpm build:native).
|
||||
*/
|
||||
|
||||
import assert from 'node:assert/strict'
|
||||
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { describe, it } from 'node:test'
|
||||
|
||||
import { detectFormat, parseFile, parseFileSync } from '../index'
|
||||
import type { ParseResult } from '../types'
|
||||
import { loadNativeParser } from './loader'
|
||||
|
||||
/** Collect a full ParseResult while capturing onLog messages. */
|
||||
async function parseCollectingLogs(filePath: string, logs: string[]): Promise<ParseResult> {
|
||||
const result: ParseResult = {
|
||||
meta: { name: '', platform: '', type: 'group' as ParseResult['meta']['type'] },
|
||||
members: [],
|
||||
messages: [],
|
||||
}
|
||||
for await (const event of parseFile({ filePath, onLog: (_level, message) => logs.push(message) })) {
|
||||
if (event.type === 'meta') result.meta = event.data
|
||||
else if (event.type === 'members') result.members.push(...event.data)
|
||||
else if (event.type === 'messages') result.messages.push(...event.data)
|
||||
else if (event.type === 'error') throw event.data
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
const ENV_KEY = 'CHATLAB_DISABLE_NATIVE_PERF'
|
||||
|
||||
function nativeAvailable(): boolean {
|
||||
const saved = process.env[ENV_KEY]
|
||||
delete process.env[ENV_KEY]
|
||||
try {
|
||||
return loadNativeParser() !== null
|
||||
} finally {
|
||||
if (saved !== undefined) process.env[ENV_KEY] = saved
|
||||
}
|
||||
}
|
||||
|
||||
async function parseBothWays(filename: string, content: string, nativeLogs?: string[]) {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'chatlab-json-parity-'))
|
||||
const filePath = join(dir, filename)
|
||||
try {
|
||||
writeFileSync(filePath, content, 'utf-8')
|
||||
assert.equal(detectFormat(filePath)?.id, 'chatlab')
|
||||
|
||||
delete process.env[ENV_KEY]
|
||||
const nativeResult = nativeLogs ? await parseCollectingLogs(filePath, nativeLogs) : await parseFileSync(filePath)
|
||||
|
||||
process.env[ENV_KEY] = '1'
|
||||
const tsResult = await parseFileSync(filePath)
|
||||
|
||||
return { nativeResult, tsResult }
|
||||
} finally {
|
||||
delete process.env[ENV_KEY]
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
|
||||
function assertParity(nativeResult: ParseResult, tsResult: ParseResult) {
|
||||
assert.deepEqual(nativeResult.meta, tsResult.meta)
|
||||
assert.deepEqual(nativeResult.members, tsResult.members)
|
||||
assert.equal(nativeResult.messages.length, tsResult.messages.length)
|
||||
for (let i = 0; i < tsResult.messages.length; i++) {
|
||||
assert.deepEqual(nativeResult.messages[i], tsResult.messages[i], `message #${i} differs`)
|
||||
}
|
||||
}
|
||||
|
||||
// Canonical fixture without member roles (the TS head-regex parses this
|
||||
// members array successfully, so both implementations use the head members).
|
||||
const CANONICAL_FIXTURE = {
|
||||
chatlab: { version: '0.0.2', exportedAt: 1700000000, generator: 'ChatLab' },
|
||||
meta: {
|
||||
name: '测试群',
|
||||
platform: 'weixin',
|
||||
type: 'group',
|
||||
groupId: 'g1@chatroom',
|
||||
groupAvatar: 'https://example.com/g.jpg',
|
||||
},
|
||||
members: [
|
||||
{ platformId: 'u1', accountName: 'Alice', groupNickname: '小A', avatar: 'https://example.com/a.jpg' },
|
||||
{ platformId: 'u2', accountName: 'Bob' },
|
||||
],
|
||||
messages: [
|
||||
{
|
||||
platformMessageId: 'm1',
|
||||
sender: 'u1',
|
||||
accountName: 'Alice',
|
||||
groupNickname: '小A',
|
||||
timestamp: 1700000100,
|
||||
type: 0,
|
||||
content: '你好',
|
||||
},
|
||||
{
|
||||
platformMessageId: 'm2',
|
||||
sender: 'u2',
|
||||
accountName: 'Bob',
|
||||
timestamp: 1700000200,
|
||||
type: 25,
|
||||
content: '回复内容',
|
||||
replyToMessageId: 'm1',
|
||||
},
|
||||
// null content (e.g. image message) and no optional ids
|
||||
{ sender: 'u1', accountName: 'Alice', groupNickname: '小A', timestamp: 1700000300, type: 1, content: null },
|
||||
// empty-string content passes through as-is
|
||||
{ sender: 'u2', accountName: 'Bob', timestamp: 1700000400, type: 0, content: '' },
|
||||
],
|
||||
}
|
||||
|
||||
describe('chatlab native parser parity', { skip: !nativeAvailable() && 'native module not built' }, () => {
|
||||
it('produces identical output to the TS parser on a canonical file', async () => {
|
||||
const { nativeResult, tsResult } = await parseBothWays('测试群.json', JSON.stringify(CANONICAL_FIXTURE, null, 2))
|
||||
assertParity(nativeResult, tsResult)
|
||||
|
||||
assert.equal(nativeResult.meta.name, '测试群')
|
||||
assert.equal(nativeResult.meta.platform, 'weixin')
|
||||
assert.equal(nativeResult.meta.groupId, 'g1@chatroom')
|
||||
assert.equal(nativeResult.members.length, 2)
|
||||
assert.equal(nativeResult.members[0].avatar, 'https://example.com/a.jpg')
|
||||
assert.equal(nativeResult.messages.length, 4)
|
||||
assert.equal(nativeResult.messages[1].replyToMessageId, 'm1')
|
||||
assert.equal(nativeResult.messages[2].content, null)
|
||||
assert.equal(nativeResult.messages[3].content, '')
|
||||
})
|
||||
|
||||
it('collects members from messages when the members array is absent', async () => {
|
||||
const noMembers = {
|
||||
chatlab: { version: '0.0.2' },
|
||||
meta: { name: '无成员表', platform: 'qq', type: 'group' },
|
||||
messages: [
|
||||
{ sender: 'u1', accountName: '旧名', groupNickname: '昵称', timestamp: 1, type: 0, content: 'a' },
|
||||
// Later message replaces the whole member entry (JS Map.set semantics)
|
||||
{ sender: 'u1', accountName: '新名', timestamp: 2, type: 0, content: 'b' },
|
||||
{ sender: 'u2', accountName: 'B', timestamp: 3, type: 0, content: 'c' },
|
||||
],
|
||||
}
|
||||
const { nativeResult, tsResult } = await parseBothWays('无成员表.json', JSON.stringify(noMembers))
|
||||
assertParity(nativeResult, tsResult)
|
||||
assert.equal(nativeResult.members.length, 2)
|
||||
assert.equal(nativeResult.members[0].accountName, '新名')
|
||||
assert.equal(nativeResult.members[0].groupNickname, undefined)
|
||||
})
|
||||
|
||||
it('handles missing meta name and empty messages identically', async () => {
|
||||
const minimal = {
|
||||
chatlab: { version: '0.0.2' },
|
||||
meta: { platform: 'weixin', type: 'private' },
|
||||
messages: [],
|
||||
}
|
||||
const { nativeResult, tsResult } = await parseBothWays('私聊记录.json', JSON.stringify(minimal))
|
||||
assertParity(nativeResult, tsResult)
|
||||
// Default name falls back to the filename.
|
||||
assert.equal(nativeResult.meta.name, '私聊记录')
|
||||
assert.equal(nativeResult.meta.type, 'private')
|
||||
assert.deepEqual(nativeResult.messages, [])
|
||||
})
|
||||
|
||||
it('keeps member roles and avatars where the TS head-regex heuristic loses them', async () => {
|
||||
// The TS parser extracts head members with a non-greedy regex that breaks
|
||||
// on nested arrays (roles), silently degrading to message-collected
|
||||
// members. The Rust kernel parses the members array structurally, so this
|
||||
// is a deliberate improvement over TS behavior, not a parity case.
|
||||
const withRoles = {
|
||||
chatlab: { version: '0.0.2' },
|
||||
meta: { name: '带角色群', platform: 'qq', type: 'group' },
|
||||
members: [
|
||||
{
|
||||
platformId: 'u1',
|
||||
accountName: 'Alice',
|
||||
avatar: 'https://example.com/a.jpg',
|
||||
roles: [{ id: 'owner' }, { id: 'admin', name: '管理员' }],
|
||||
},
|
||||
{ platformId: 'u2', accountName: 'Bob' },
|
||||
],
|
||||
messages: [{ sender: 'u1', accountName: 'Alice', timestamp: 1, type: 0, content: 'x' }],
|
||||
}
|
||||
const { nativeResult, tsResult } = await parseBothWays('带角色群.json', JSON.stringify(withRoles))
|
||||
|
||||
// TS loses the head members entirely (only u1 appears, from messages).
|
||||
assert.equal(tsResult.members.length, 1)
|
||||
assert.equal(tsResult.members[0].avatar, undefined)
|
||||
|
||||
// Native keeps the full member list including roles and avatars.
|
||||
assert.equal(nativeResult.members.length, 2)
|
||||
assert.equal(nativeResult.members[0].avatar, 'https://example.com/a.jpg')
|
||||
assert.deepEqual(nativeResult.members[0].roles, [{ id: 'owner' }, { id: 'admin', name: '管理员' }])
|
||||
|
||||
// Meta and messages still match exactly.
|
||||
assert.deepEqual(nativeResult.meta, tsResult.meta)
|
||||
assert.deepEqual(nativeResult.messages, tsResult.messages)
|
||||
})
|
||||
|
||||
it('falls back to the TS parser for off-spec files', async () => {
|
||||
const offSpec = {
|
||||
chatlab: { version: '0.0.2' },
|
||||
meta: { name: '异常文件', platform: 'qq', type: 'group' },
|
||||
messages: [
|
||||
// content must be string|null per spec; a number forces the fallback.
|
||||
{ sender: 'u1', accountName: 'A', timestamp: 1, type: 0, content: 42 },
|
||||
],
|
||||
}
|
||||
const logs: string[] = []
|
||||
const { nativeResult, tsResult } = await parseBothWays('异常文件.json', JSON.stringify(offSpec), logs)
|
||||
assert.ok(
|
||||
logs.some((m) => m.includes('[NativeParser]') && m.includes('falling back')),
|
||||
'expected a fallback log entry'
|
||||
)
|
||||
// Fallback means the native path output IS the TS output, quirks included.
|
||||
assertParity(nativeResult, tsResult)
|
||||
assert.equal(nativeResult.messages[0].content, 42 as unknown as string)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,153 @@
|
||||
/**
|
||||
* Rust-accelerated ChatLab JSON parsing with transparent TS fallback.
|
||||
*
|
||||
* Event order and payload shapes are identical to the pure-TS parser in
|
||||
* formats/chatlab.ts for spec-compliant files (verified by parity tests).
|
||||
* The Rust kernel is strict about the format spec: off-spec files (wrong
|
||||
* field types, missing required fields) make parse() fail, and we fall back
|
||||
* to the TS parser which replicates all JS passthrough quirks. Fallback also
|
||||
* happens when the native module is unavailable or disabled via
|
||||
* CHATLAB_DISABLE_NATIVE_PERF=1.
|
||||
*/
|
||||
|
||||
import type { ChatType, MessageType } from '@openchatlab/shared-types'
|
||||
import type { ChatlabParser, NativeChatlabMember, NativeChatlabMessage } from '@openchatlab/parser-native'
|
||||
import type { ParseEvent, ParseOptions, ParsedMember, ParsedMessage, ParsedMeta } from '../types'
|
||||
import { getFileSize, createProgress } from '../utils'
|
||||
import { loadNativeParser } from './loader'
|
||||
|
||||
const PROGRESS_POLL_INTERVAL_MS = 200
|
||||
|
||||
type ParseGenerator = (options: ParseOptions) => AsyncGenerator<ParseEvent, void, unknown>
|
||||
|
||||
function toParsedMessage(message: NativeChatlabMessage): ParsedMessage {
|
||||
return {
|
||||
senderPlatformId: message.senderPlatformId,
|
||||
senderAccountName: message.senderAccountName,
|
||||
senderGroupNickname: message.senderGroupNickname,
|
||||
timestamp: message.timestamp,
|
||||
type: message.messageType as MessageType,
|
||||
// The kernel rejects a missing content key, so undefined here always
|
||||
// means the source value was JSON null.
|
||||
content: message.content ?? null,
|
||||
platformMessageId: message.platformMessageId,
|
||||
replyToMessageId: message.replyToMessageId,
|
||||
}
|
||||
}
|
||||
|
||||
function toParsedMember(member: NativeChatlabMember, fromHead: boolean): ParsedMember {
|
||||
if (!fromHead) {
|
||||
// Members collected from messages carry only these three keys in the TS
|
||||
// parser; keep the object shape identical.
|
||||
return {
|
||||
platformId: member.platformId,
|
||||
accountName: member.accountName,
|
||||
groupNickname: member.groupNickname,
|
||||
}
|
||||
}
|
||||
return {
|
||||
platformId: member.platformId,
|
||||
accountName: member.accountName,
|
||||
groupNickname: member.groupNickname,
|
||||
avatar: member.avatar,
|
||||
// Role objects pass through with their original key set ({id} or {id, name}).
|
||||
roles: member.roles?.map((role) => (role.name !== undefined ? { id: role.id, name: role.name } : { id: role.id })),
|
||||
}
|
||||
}
|
||||
|
||||
async function* parseChatLabNative(
|
||||
options: ParseOptions,
|
||||
parser: ChatlabParser,
|
||||
fallback: ParseGenerator
|
||||
): AsyncGenerator<ParseEvent, void, unknown> {
|
||||
const { filePath, batchSize = 5000, onProgress, onLog } = options
|
||||
|
||||
const totalBytes = getFileSize(filePath)
|
||||
const initialProgress = createProgress('parsing', 0, totalBytes, 0, '')
|
||||
yield { type: 'progress', data: initialProgress }
|
||||
onProgress?.(initialProgress)
|
||||
onLog?.(
|
||||
'info',
|
||||
`[NativeParser] Parsing ChatLab JSON with Rust kernel, size: ${(totalBytes / 1024 / 1024).toFixed(2)} MB`
|
||||
)
|
||||
|
||||
const progressTimer = setInterval(() => {
|
||||
const progress = parser.progress()
|
||||
onProgress?.(
|
||||
createProgress(
|
||||
'parsing',
|
||||
progress.bytesRead,
|
||||
totalBytes,
|
||||
progress.messagesProcessed,
|
||||
`已处理 ${progress.messagesProcessed} 条消息...`
|
||||
)
|
||||
)
|
||||
}, PROGRESS_POLL_INTERVAL_MS)
|
||||
|
||||
try {
|
||||
await parser.parse()
|
||||
} catch (error) {
|
||||
onLog?.(
|
||||
'warn',
|
||||
`[NativeParser] Rust parse failed, falling back to TS parser: ${error instanceof Error ? error.message : String(error)}`
|
||||
)
|
||||
// Only progress events have been emitted so far — safe to restart with TS.
|
||||
yield* fallback(options)
|
||||
return
|
||||
} finally {
|
||||
clearInterval(progressTimer)
|
||||
}
|
||||
|
||||
const nativeMeta = parser.getMeta()
|
||||
const meta: ParsedMeta = {
|
||||
name: nativeMeta.name,
|
||||
platform: nativeMeta.platform,
|
||||
type: nativeMeta.chatType as ChatType,
|
||||
groupId: nativeMeta.groupId,
|
||||
groupAvatar: nativeMeta.groupAvatar,
|
||||
}
|
||||
yield { type: 'meta', data: meta }
|
||||
|
||||
const members = parser.takeMembers().map((member) => toParsedMember(member, nativeMeta.membersFromHead))
|
||||
yield { type: 'members', data: members }
|
||||
|
||||
let messagesProcessed = 0
|
||||
while (true) {
|
||||
const batch = parser.takeBatch(batchSize)
|
||||
if (!batch) break
|
||||
messagesProcessed += batch.length
|
||||
yield { type: 'messages', data: batch.map(toParsedMessage) }
|
||||
}
|
||||
|
||||
const doneProgress = createProgress('done', totalBytes, totalBytes, messagesProcessed, '')
|
||||
yield { type: 'progress', data: doneProgress }
|
||||
onProgress?.(doneProgress)
|
||||
onLog?.('info', `解析完成: ${messagesProcessed} 条消息, ${members.length} 个成员`)
|
||||
|
||||
yield { type: 'done', data: { messageCount: messagesProcessed, memberCount: members.length } }
|
||||
}
|
||||
|
||||
/** Wrap the TS ChatLab parse generator with native acceleration. */
|
||||
export function withNativeChatlab(fallback: ParseGenerator): ParseGenerator {
|
||||
return async function* (options: ParseOptions): AsyncGenerator<ParseEvent, void, unknown> {
|
||||
const native = loadNativeParser()
|
||||
if (!native) {
|
||||
yield* fallback(options)
|
||||
return
|
||||
}
|
||||
|
||||
let parser: ChatlabParser
|
||||
try {
|
||||
parser = new native.ChatlabParser(options.filePath)
|
||||
} catch (error) {
|
||||
options.onLog?.(
|
||||
'warn',
|
||||
`[NativeParser] Failed to create Rust parser, falling back to TS parser: ${error instanceof Error ? error.message : String(error)}`
|
||||
)
|
||||
yield* fallback(options)
|
||||
return
|
||||
}
|
||||
|
||||
yield* parseChatLabNative(options, parser, fallback)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
/**
|
||||
* Loader for the optional Rust parsing kernels (@openchatlab/parser-native).
|
||||
*
|
||||
* The native module is a local-build artifact: contributors without a Rust
|
||||
* toolchain, published CLI installs and unsupported platforms simply fall
|
||||
* back to the pure-TS parsers. Loading uses a runtime require so bundlers
|
||||
* (electron-vite / tsup) never try to inline the .node binary.
|
||||
*/
|
||||
|
||||
import { createRequire } from 'node:module'
|
||||
|
||||
import type { ChatlabParser, WeflowParser } from '@openchatlab/parser-native'
|
||||
|
||||
export interface NativeParserModule {
|
||||
WeflowParser: typeof WeflowParser
|
||||
ChatlabParser: typeof ChatlabParser
|
||||
}
|
||||
|
||||
let cachedModule: NativeParserModule | null | undefined
|
||||
|
||||
function isNativeDisabled(): boolean {
|
||||
return process.env.CHATLAB_DISABLE_NATIVE_PERF === '1'
|
||||
}
|
||||
|
||||
function requireNativeModule(): NativeParserModule {
|
||||
const requireFn = createRequire(import.meta.url)
|
||||
return requireFn('@openchatlab/parser-native') as NativeParserModule
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the native module, or null when disabled/unavailable.
|
||||
* The module itself is cached; the env switch is evaluated on every call so
|
||||
* tests can toggle it at runtime.
|
||||
*/
|
||||
export function loadNativeParser(): NativeParserModule | null {
|
||||
if (isNativeDisabled()) return null
|
||||
if (cachedModule !== undefined) return cachedModule
|
||||
try {
|
||||
cachedModule = requireNativeModule()
|
||||
} catch {
|
||||
cachedModule = null
|
||||
}
|
||||
return cachedModule
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
/**
|
||||
* Parity tests: the Rust WeFlow kernel must produce byte-for-byte identical
|
||||
* ParseResult output to the pure-TS stream-json parser.
|
||||
*
|
||||
* Skipped automatically when the native module has not been built locally
|
||||
* (pnpm build:native).
|
||||
*/
|
||||
|
||||
import assert from 'node:assert/strict'
|
||||
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { describe, it } from 'node:test'
|
||||
|
||||
import { detectFormat, parseFileSync } from '../index'
|
||||
import type { ParseResult } from '../types'
|
||||
import { loadNativeParser } from './loader'
|
||||
|
||||
const ENV_KEY = 'CHATLAB_DISABLE_NATIVE_PERF'
|
||||
|
||||
function nativeAvailable(): boolean {
|
||||
const saved = process.env[ENV_KEY]
|
||||
delete process.env[ENV_KEY]
|
||||
try {
|
||||
return loadNativeParser() !== null
|
||||
} finally {
|
||||
if (saved !== undefined) process.env[ENV_KEY] = saved
|
||||
}
|
||||
}
|
||||
|
||||
async function parseBothWays(filename: string, content: string, expectedFormatId: string) {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'chatlab-weflow-parity-'))
|
||||
const filePath = join(dir, filename)
|
||||
try {
|
||||
writeFileSync(filePath, content, 'utf-8')
|
||||
assert.equal(detectFormat(filePath)?.id, expectedFormatId)
|
||||
|
||||
delete process.env[ENV_KEY]
|
||||
const nativeResult = await parseFileSync(filePath)
|
||||
|
||||
process.env[ENV_KEY] = '1'
|
||||
const tsResult = await parseFileSync(filePath)
|
||||
|
||||
return { nativeResult, tsResult }
|
||||
} finally {
|
||||
delete process.env[ENV_KEY]
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
|
||||
function assertParity(nativeResult: ParseResult, tsResult: ParseResult) {
|
||||
assert.deepEqual(nativeResult.meta, tsResult.meta)
|
||||
assert.deepEqual(nativeResult.members, tsResult.members)
|
||||
assert.equal(nativeResult.messages.length, tsResult.messages.length)
|
||||
for (let i = 0; i < tsResult.messages.length; i++) {
|
||||
assert.deepEqual(nativeResult.messages[i], tsResult.messages[i], `message #${i} differs`)
|
||||
}
|
||||
}
|
||||
|
||||
// Fixture covering every branch of processMessage / meta extraction:
|
||||
// avatars lookup, @chatroom skip, missing/empty sender, missing/null
|
||||
// createTime, non-string content, whitespace-only content, renamed member,
|
||||
// senderAvatarKey fallback, unknown message type, ownerId inference.
|
||||
const FULL_FIXTURE = {
|
||||
weflow: { version: '1.2.3', exportedAt: 1704164645 },
|
||||
session: {
|
||||
wxid: 'room123@chatroom',
|
||||
nickname: '昵称群',
|
||||
remark: '',
|
||||
displayName: '显示名群',
|
||||
type: '群聊',
|
||||
lastTimestamp: 1704200000,
|
||||
messageCount: 10,
|
||||
avatar: '',
|
||||
},
|
||||
avatars: {
|
||||
wxid_alice: 'data:image/jpeg;base64,ALICE',
|
||||
'room123@chatroom': 'data:image/jpeg;base64,ROOM',
|
||||
wxid_keyed: 'data:image/jpeg;base64,KEYED',
|
||||
wxid_empty: '',
|
||||
},
|
||||
messages: [
|
||||
{
|
||||
localId: 1,
|
||||
createTime: 1704164645,
|
||||
formattedTime: '',
|
||||
type: '文本消息',
|
||||
localType: 1,
|
||||
content: ' hello world ',
|
||||
isSend: 0,
|
||||
senderUsername: 'wxid_alice',
|
||||
senderDisplayName: 'Alice',
|
||||
senderAvatarKey: 'wxid_alice',
|
||||
source: '',
|
||||
},
|
||||
// Owner message (isSend=1)
|
||||
{
|
||||
localId: 2,
|
||||
createTime: 1704164700,
|
||||
type: '图片消息',
|
||||
content: '[图片]',
|
||||
isSend: 1,
|
||||
senderUsername: 'wxid_owner',
|
||||
senderDisplayName: 'Me',
|
||||
senderAvatarKey: '',
|
||||
source: '',
|
||||
},
|
||||
// System message from the room itself -> skipped
|
||||
{
|
||||
localId: 3,
|
||||
createTime: 1704164710,
|
||||
type: '系统消息',
|
||||
content: 'joined',
|
||||
isSend: null,
|
||||
senderUsername: 'room123@chatroom',
|
||||
senderDisplayName: '群',
|
||||
},
|
||||
// Missing senderUsername -> skipped
|
||||
{ localId: 4, createTime: 1704164720, type: '文本消息', content: 'no sender' },
|
||||
// Empty senderUsername -> skipped
|
||||
{ localId: 5, createTime: 1704164725, type: '文本消息', content: 'x', senderUsername: '' },
|
||||
// Missing createTime -> skipped
|
||||
{ localId: 6, type: '文本消息', content: 'no time', senderUsername: 'wxid_alice' },
|
||||
// Null createTime -> kept with null timestamp (importer will skip)
|
||||
{
|
||||
localId: 7,
|
||||
createTime: null,
|
||||
type: '文本消息',
|
||||
content: 'null time',
|
||||
senderUsername: 'wxid_alice',
|
||||
senderDisplayName: 'Alice',
|
||||
},
|
||||
// Non-string content -> JSON.stringify
|
||||
{
|
||||
localId: 8,
|
||||
createTime: 1704164730,
|
||||
type: '引用消息',
|
||||
content: { title: '引用', items: [1, true, null], nested: { a: 'b' } },
|
||||
isSend: 0,
|
||||
senderUsername: 'wxid_alice',
|
||||
senderDisplayName: 'Alice Renamed',
|
||||
},
|
||||
// Whitespace-only content -> null; avatar via senderAvatarKey pointing elsewhere
|
||||
{
|
||||
localId: 9,
|
||||
createTime: 1704164740,
|
||||
type: '未知类型(42)',
|
||||
content: ' ',
|
||||
isSend: 0,
|
||||
senderUsername: 'wxid_bob',
|
||||
senderDisplayName: '',
|
||||
senderAvatarKey: 'wxid_keyed',
|
||||
},
|
||||
// Numeric localId with fraction, emoji content, unicode whitespace trim
|
||||
{
|
||||
localId: 10.5,
|
||||
createTime: 1704164750,
|
||||
type: '动画表情',
|
||||
content: '\u00a0😀 表情 \u3000',
|
||||
isSend: 0,
|
||||
senderUsername: 'wxid_owner',
|
||||
senderDisplayName: 'Me2',
|
||||
},
|
||||
// Missing localId -> String(undefined); missing content -> null
|
||||
{
|
||||
createTime: 1704164760,
|
||||
type: '红包卡片',
|
||||
isSend: 0,
|
||||
senderUsername: 'wxid_alice',
|
||||
senderDisplayName: 'Alice Renamed',
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
describe('weflow native parser parity', { skip: !nativeAvailable() && 'native module not built' }, () => {
|
||||
it('produces identical output to the TS parser on the full fixture', async () => {
|
||||
const { nativeResult, tsResult } = await parseBothWays(
|
||||
'显示名群.json',
|
||||
JSON.stringify(FULL_FIXTURE, null, 2),
|
||||
'weflow'
|
||||
)
|
||||
assertParity(nativeResult, tsResult)
|
||||
|
||||
// Sanity-check key semantics directly (guards against both parsers
|
||||
// being wrong in the same way on the highest-risk branches).
|
||||
assert.equal(nativeResult.meta.name, '显示名群')
|
||||
assert.equal(nativeResult.meta.groupId, 'room123@chatroom')
|
||||
assert.equal(nativeResult.meta.groupAvatar, 'data:image/jpeg;base64,ROOM')
|
||||
assert.equal(nativeResult.meta.ownerId, 'wxid_owner')
|
||||
assert.equal(nativeResult.messages.length, 7)
|
||||
assert.equal(nativeResult.messages[0].content, 'hello world')
|
||||
assert.equal(nativeResult.messages[2].timestamp, null)
|
||||
assert.equal(nativeResult.messages[3].content, '{"title":"引用","items":[1,true,null],"nested":{"a":"b"}}')
|
||||
assert.equal(nativeResult.messages[4].content, null)
|
||||
assert.equal(nativeResult.messages[5].platformMessageId, '10.5')
|
||||
assert.equal(nativeResult.messages[6].platformMessageId, 'undefined')
|
||||
const bob = nativeResult.members.find((m) => m.platformId === 'wxid_bob')
|
||||
assert.equal(bob?.avatar, 'data:image/jpeg;base64,KEYED')
|
||||
assert.equal(bob?.accountName, 'wxid_bob')
|
||||
const alice = nativeResult.members.find((m) => m.platformId === 'wxid_alice')
|
||||
assert.equal(alice?.accountName, 'Alice Renamed')
|
||||
})
|
||||
|
||||
it('handles private chats and missing session identically', async () => {
|
||||
const privateChat = {
|
||||
weflow: { version: '1.0.0' },
|
||||
session: { wxid: 'wxid_friend', type: '私聊', nickname: '好友', displayName: '' },
|
||||
avatars: {},
|
||||
messages: [
|
||||
{
|
||||
localId: 1,
|
||||
createTime: 100,
|
||||
type: '文本消息',
|
||||
content: 'hi',
|
||||
isSend: 1,
|
||||
senderUsername: 'wxid_me',
|
||||
senderDisplayName: '我',
|
||||
},
|
||||
],
|
||||
}
|
||||
const { nativeResult, tsResult } = await parseBothWays('好友聊天.json', JSON.stringify(privateChat), 'weflow')
|
||||
assertParity(nativeResult, tsResult)
|
||||
assert.equal(nativeResult.meta.type, 'private')
|
||||
assert.equal(nativeResult.meta.groupId, undefined)
|
||||
assert.equal(nativeResult.meta.name, '好友')
|
||||
})
|
||||
|
||||
it('handles the echotrace variant (shared parser) identically', async () => {
|
||||
const echotrace = {
|
||||
session: { wxid: 'room9@chatroom', type: '群聊', displayName: 'Echo 群' },
|
||||
messages: [
|
||||
{
|
||||
localId: 1,
|
||||
createTime: 200,
|
||||
type: '文本消息',
|
||||
content: 'echo',
|
||||
isSend: 0,
|
||||
senderUsername: 'wxid_e',
|
||||
senderDisplayName: 'E',
|
||||
},
|
||||
],
|
||||
}
|
||||
const { nativeResult, tsResult } = await parseBothWays(
|
||||
'echo.json',
|
||||
JSON.stringify(echotrace),
|
||||
'ycccccccy-echotrace'
|
||||
)
|
||||
assertParity(nativeResult, tsResult)
|
||||
assert.equal(nativeResult.messages.length, 1)
|
||||
})
|
||||
|
||||
it('handles empty message arrays identically', async () => {
|
||||
const empty = {
|
||||
weflow: { version: '1.0.0' },
|
||||
session: { wxid: 'r@chatroom', type: '群聊', displayName: '空群' },
|
||||
avatars: {},
|
||||
messages: [],
|
||||
}
|
||||
const { nativeResult, tsResult } = await parseBothWays('空群.json', JSON.stringify(empty), 'weflow')
|
||||
assert.deepEqual(nativeResult.meta, tsResult.meta)
|
||||
assert.deepEqual(nativeResult.members, tsResult.members)
|
||||
assert.deepEqual(nativeResult.messages, tsResult.messages)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,138 @@
|
||||
/**
|
||||
* Rust-accelerated WeFlow parsing with transparent TS fallback.
|
||||
*
|
||||
* Event order and payload shapes are identical to the pure-TS parser in
|
||||
* formats/weflow.ts (verified by parity tests): progress → meta → members →
|
||||
* message batches → done. Fallback happens when the native module is
|
||||
* unavailable, disabled via CHATLAB_DISABLE_NATIVE_PERF=1, or fails before
|
||||
* any data event has been emitted.
|
||||
*/
|
||||
|
||||
import { KNOWN_PLATFORMS, ChatType, type MessageType } from '@openchatlab/shared-types'
|
||||
import type { NativeParsedMessage, WeflowParser } from '@openchatlab/parser-native'
|
||||
import type { ParseEvent, ParseOptions, ParsedMessage, ParsedMeta } from '../types'
|
||||
import { getFileSize, createProgress } from '../utils'
|
||||
import { loadNativeParser } from './loader'
|
||||
|
||||
const PROGRESS_POLL_INTERVAL_MS = 200
|
||||
|
||||
type ParseGenerator = (options: ParseOptions) => AsyncGenerator<ParseEvent, void, unknown>
|
||||
|
||||
function toParsedMessage(message: NativeParsedMessage): ParsedMessage {
|
||||
return {
|
||||
platformMessageId: message.platformMessageId,
|
||||
senderPlatformId: message.senderPlatformId,
|
||||
senderAccountName: message.senderAccountName,
|
||||
// WeFlow has no separate group nickname field (same as the TS parser).
|
||||
senderGroupNickname: undefined,
|
||||
// The TS parser passes createTime through as-is, including null.
|
||||
timestamp: (message.timestamp ?? null) as unknown as number,
|
||||
type: message.messageType as MessageType,
|
||||
content: message.content ?? null,
|
||||
}
|
||||
}
|
||||
|
||||
async function* parseWeFlowNative(
|
||||
options: ParseOptions,
|
||||
parser: WeflowParser,
|
||||
fallback: ParseGenerator
|
||||
): AsyncGenerator<ParseEvent, void, unknown> {
|
||||
const { filePath, batchSize = 5000, onProgress, onLog } = options
|
||||
|
||||
const totalBytes = getFileSize(filePath)
|
||||
const initialProgress = createProgress('parsing', 0, totalBytes, 0, '')
|
||||
yield { type: 'progress', data: initialProgress }
|
||||
onProgress?.(initialProgress)
|
||||
onLog?.(
|
||||
'info',
|
||||
`[NativeParser] Parsing WeFlow export with Rust kernel, size: ${(totalBytes / 1024 / 1024).toFixed(2)} MB`
|
||||
)
|
||||
|
||||
const progressTimer = setInterval(() => {
|
||||
const progress = parser.progress()
|
||||
onProgress?.(
|
||||
createProgress(
|
||||
'parsing',
|
||||
progress.bytesRead,
|
||||
totalBytes,
|
||||
progress.messagesProcessed,
|
||||
`已处理 ${progress.messagesProcessed} 条消息...`
|
||||
)
|
||||
)
|
||||
}, PROGRESS_POLL_INTERVAL_MS)
|
||||
|
||||
try {
|
||||
await parser.parse()
|
||||
} catch (error) {
|
||||
onLog?.(
|
||||
'warn',
|
||||
`[NativeParser] Rust parse failed, falling back to TS parser: ${error instanceof Error ? error.message : String(error)}`
|
||||
)
|
||||
// Only progress events have been emitted so far — safe to restart with TS.
|
||||
yield* fallback(options)
|
||||
return
|
||||
} finally {
|
||||
clearInterval(progressTimer)
|
||||
}
|
||||
|
||||
const nativeMeta = parser.getMeta()
|
||||
const meta: ParsedMeta = {
|
||||
name: nativeMeta.name,
|
||||
platform: KNOWN_PLATFORMS.WECHAT,
|
||||
type: nativeMeta.chatType === 'private' ? ChatType.PRIVATE : ChatType.GROUP,
|
||||
groupId: nativeMeta.groupId,
|
||||
groupAvatar: nativeMeta.groupAvatar,
|
||||
ownerId: nativeMeta.ownerId,
|
||||
}
|
||||
yield { type: 'meta', data: meta }
|
||||
|
||||
const members = parser.takeMembers().map((member) => ({
|
||||
platformId: member.platformId,
|
||||
accountName: member.accountName,
|
||||
avatar: member.avatar,
|
||||
}))
|
||||
yield { type: 'members', data: members }
|
||||
|
||||
let messagesProcessed = 0
|
||||
while (true) {
|
||||
const batch = parser.takeBatch(batchSize)
|
||||
if (!batch) break
|
||||
messagesProcessed += batch.length
|
||||
yield { type: 'messages', data: batch.map(toParsedMessage) }
|
||||
}
|
||||
|
||||
const doneProgress = createProgress('done', totalBytes, totalBytes, messagesProcessed, '')
|
||||
yield { type: 'progress', data: doneProgress }
|
||||
onProgress?.(doneProgress)
|
||||
onLog?.('info', `解析完成: ${messagesProcessed} 条消息, ${members.length} 个成员`)
|
||||
|
||||
yield { type: 'done', data: { messageCount: messagesProcessed, memberCount: members.length } }
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap a TS parse generator with native acceleration. Used by the WeFlow and
|
||||
* ycccccccy-echotrace format modules (they share the same data structure).
|
||||
*/
|
||||
export function withNativeWeflow(fallback: ParseGenerator): ParseGenerator {
|
||||
return async function* (options: ParseOptions): AsyncGenerator<ParseEvent, void, unknown> {
|
||||
const native = loadNativeParser()
|
||||
if (!native) {
|
||||
yield* fallback(options)
|
||||
return
|
||||
}
|
||||
|
||||
let parser: WeflowParser
|
||||
try {
|
||||
parser = new native.WeflowParser(options.filePath)
|
||||
} catch (error) {
|
||||
options.onLog?.(
|
||||
'warn',
|
||||
`[NativeParser] Failed to create Rust parser, falling back to TS parser: ${error instanceof Error ? error.message : String(error)}`
|
||||
)
|
||||
yield* fallback(options)
|
||||
return
|
||||
}
|
||||
|
||||
yield* parseWeFlowNative(options, parser, fallback)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user