hicccc77/WeFlow
1
0
Fork 0
mirror of https://github.com/hicccc77/WeFlow.git synced 2026-03-24 23:06:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(export): improve batch text progress and precheck interruptibility

Browse Source
This commit is contained in:
tisonhuang
2026-03-02 18:52:02 +08:00
parent 698d2c96d7
commit 81bc5aefff
1 changed files with 230 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
electron/services/exportService.ts
View File

@@ -255,6 +255,27 @@ class ExportService {
return { mode }
}
private createCollectProgressReporter(
sessionName: string,
onProgress?: (progress: ExportProgress) => void,
progressCurrent = 5
): ((payload: { fetched: number }) => void) | undefined {
if (!onProgress) return undefined
let lastReportAt = 0
return ({ fetched }) => {
const now = Date.now()
if (now - lastReportAt < 350) return
lastReportAt = now
onProgress({
current: progressCurrent,
total: 100,
currentSession: sessionName,
phase: 'preparing',
phaseLabel: `收集消息 ${fetched.toLocaleString()} 条`
})
}
}
private shouldDecodeMessageContentInFastMode(localType: number): boolean {
// 这些类型在文本导出里只需要占位符，无需解码完整 XML / 压缩内容
if (localType === 3 || localType === 34 || localType === 42 || localType === 43 || localType === 47) {
@@ -2853,6 +2874,7 @@ class ExportService {
})
const collectParams = this.resolveCollectParams(options)
const collectProgressReporter = this.createCollectProgressReporter(sessionInfo.displayName, onProgress, 5)
const collected = await this.collectMessages(
sessionId,
cleanedMyWxid,
@@ -2860,7 +2882,8 @@ class ExportService {
options.senderUsername,
collectParams.mode,
collectParams.targetMediaTypes,
control
control,
collectProgressReporter
)
const allMessages = collected.rows
@@ -3273,7 +3296,7 @@ class ExportService {
})
const collectParams = this.resolveCollectParams(options)
let collectProgressLastReportAt = 0
const collectProgressReporter = this.createCollectProgressReporter(sessionInfo.displayName, onProgress, 5)
const collected = await this.collectMessages(
sessionId,
cleanedMyWxid,
@@ -3282,18 +3305,7 @@ class ExportService {
collectParams.mode,
collectParams.targetMediaTypes,
control,
({ fetched }) => {
const now = Date.now()
if (now - collectProgressLastReportAt < 350) return
collectProgressLastReportAt = now
onProgress?.({
current: 5,
total: 100,
currentSession: sessionInfo.displayName,
phase: 'preparing',
phaseLabel: `收集消息 ${fetched.toLocaleString()} 条`
})
}
collectProgressReporter
)
// 如果没有消息,不创建文件
@@ -3890,6 +3902,7 @@ class ExportService {
})
const collectParams = this.resolveCollectParams(options)
const collectProgressReporter = this.createCollectProgressReporter(sessionInfo.displayName, onProgress, 5)
const collected = await this.collectMessages(
sessionId,
cleanedMyWxid,
@@ -3897,7 +3910,8 @@ class ExportService {
options.senderUsername,
collectParams.mode,
collectParams.targetMediaTypes,
control
control,
collectProgressReporter
)
// 如果没有消息,不创建文件
@@ -4401,6 +4415,7 @@ class ExportService {
})
const collectParams = this.resolveCollectParams(options)
const collectProgressReporter = this.createCollectProgressReporter(sessionInfo.displayName, onProgress, 5)
const collected = await this.collectMessages(
sessionId,
cleanedMyWxid,
@@ -4408,7 +4423,8 @@ class ExportService {
options.senderUsername,
collectParams.mode,
collectParams.targetMediaTypes,
control
control,
collectProgressReporter
)
// 如果没有消息,不创建文件
@@ -4705,6 +4721,7 @@ class ExportService {
})
const collectParams = this.resolveCollectParams(options)
const collectProgressReporter = this.createCollectProgressReporter(sessionInfo.displayName, onProgress, 5)
const collected = await this.collectMessages(
sessionId,
cleanedMyWxid,
@@ -4712,7 +4729,8 @@ class ExportService {
options.senderUsername,
collectParams.mode,
collectParams.targetMediaTypes,
control
control,
collectProgressReporter
)
if (collected.rows.length === 0) {
return { success: false, error: '该会话在指定时间范围内没有消息' }
@@ -5071,6 +5089,7 @@ class ExportService {
}
const collectParams = this.resolveCollectParams(options)
const collectProgressReporter = this.createCollectProgressReporter(sessionInfo.displayName, onProgress, 5)
const collected = await this.collectMessages(
sessionId,
cleanedMyWxid,
@@ -5078,7 +5097,8 @@ class ExportService {
options.senderUsername,
collectParams.mode,
collectParams.targetMediaTypes,
control
control,
collectProgressReporter
)
// 如果没有消息,不创建文件
@@ -5671,40 +5691,122 @@ class ExportService {
? (effectiveOptions.sessionLayout ?? 'per-session')
: 'shared'
let completedCount = 0
const activeSessionRatios = new Map<string, number>()
const computeAggregateCurrent = () => {
let activeRatioSum = 0
for (const ratio of activeSessionRatios.values()) {
activeRatioSum += Math.max(0, Math.min(1, ratio))
}
return Math.min(sessionIds.length, completedCount + activeRatioSum)
}
const defaultConcurrency = exportMediaEnabled ? 2 : 4
const rawConcurrency = typeof effectiveOptions.exportConcurrency === 'number'
? Math.floor(effectiveOptions.exportConcurrency)
: defaultConcurrency
const clampedConcurrency = Math.max(1, Math.min(rawConcurrency, 6))
const sessionConcurrency = clampedConcurrency
const queue = [...sessionIds]
let pauseRequested = false
let stopRequested = false
const emptySessionIds = new Set<string>()
const canFastSkipEmptySessions = this.isUnboundedDateRange(effectiveOptions.dateRange) &&
!String(effectiveOptions.senderUsername || '').trim()
if (canFastSkipEmptySessions && sessionIds.length > 0) {
const countsResult = await wcdbService.getMessageCounts(sessionIds)
const EMPTY_SESSION_PRECHECK_LIMIT = 1200
if (sessionIds.length <= EMPTY_SESSION_PRECHECK_LIMIT) {
let checkedCount = 0
onProgress?.({
current: computeAggregateCurrent(),
total: sessionIds.length,
currentSession: '',
currentSessionId: '',
phase: 'preparing',
phaseProgress: 0,
phaseTotal: sessionIds.length,
phaseLabel: `预检查空会话 0/${sessionIds.length}`
})
const PRECHECK_BATCH_SIZE = 160
for (let i = 0; i < sessionIds.length; i += PRECHECK_BATCH_SIZE) {
if (control?.shouldStop?.()) {
stopRequested = true
break
}
if (control?.shouldPause?.()) {
pauseRequested = true
break
}
const batchSessionIds = sessionIds.slice(i, i + PRECHECK_BATCH_SIZE)
const countsResult = await wcdbService.getMessageCounts(batchSessionIds)
if (countsResult.success && countsResult.counts) {
for (const sessionId of sessionIds) {
const count = countsResult.counts[sessionId]
for (const batchSessionId of batchSessionIds) {
const count = countsResult.counts[batchSessionId]
if (typeof count === 'number' && Number.isFinite(count) && count <= 0) {
emptySessionIds.add(sessionId)
emptySessionIds.add(batchSessionId)
}
}
}
checkedCount = Math.min(sessionIds.length, checkedCount + batchSessionIds.length)
onProgress?.({
current: computeAggregateCurrent(),
total: sessionIds.length,
currentSession: '',
currentSessionId: '',
phase: 'preparing',
phaseProgress: checkedCount,
phaseTotal: sessionIds.length,
phaseLabel: `预检查空会话 ${checkedCount}/${sessionIds.length}`
})
}
} else {
onProgress?.({
current: computeAggregateCurrent(),
total: sessionIds.length,
currentSession: '',
currentSessionId: '',
phase: 'preparing',
phaseLabel: `会话较多，已跳过空会话预检查（${sessionIds.length} 个）`
})
}
}
if (stopRequested) {
return {
success: true,
successCount,
failCount,
stopped: true,
pendingSessionIds: [...queue],
successSessionIds,
failedSessionIds
}
}
if (pauseRequested) {
return {
success: true,
successCount,
failCount,
paused: true,
pendingSessionIds: [...queue],
successSessionIds,
failedSessionIds
}
}
const queue = [...sessionIds]
let pauseRequested = false
let stopRequested = false
const runOne = async (sessionId: string): Promise<'done' | 'stopped'> => {
try {
this.throwIfStopRequested(control)
const sessionInfo = await this.getContactInfo(sessionId)
if (emptySessionIds.has(sessionId)) {
failCount++
failedSessionIds.push(sessionId)
activeSessionRatios.delete(sessionId)
completedCount++
onProgress?.({
current: completedCount,
current: computeAggregateCurrent(),
total: sessionIds.length,
currentSession: sessionInfo.displayName,
currentSessionId: sessionId,
@@ -5717,11 +5819,13 @@ class ExportService {
const sessionProgress = (progress: ExportProgress) => {
const phaseTotal = Number.isFinite(progress.total) && progress.total > 0 ? progress.total : 100
const phaseCurrent = Number.isFinite(progress.current) ? progress.current : 0
const ratio = Math.max(0, Math.min(1, phaseCurrent / phaseTotal))
const aggregateCurrent = Math.min(sessionIds.length, completedCount + ratio)
const ratio = progress.phase === 'complete'
? 1
: Math.max(0, Math.min(1, phaseCurrent / phaseTotal))
activeSessionRatios.set(sessionId, ratio)
onProgress?.({
...progress,
current: aggregateCurrent,
current: computeAggregateCurrent(),
total: sessionIds.length,
currentSession: sessionInfo.displayName,
currentSessionId: sessionId
@@ -5729,10 +5833,11 @@ class ExportService {
}
sessionProgress({
current: completedCount,
total: sessionIds.length,
current: 0,
total: 100,
currentSession: sessionInfo.displayName,
phase: 'exporting'
phase: 'preparing',
phaseLabel: '准备导出'
})
const sanitizeName = (value: string) => value.replace(/[<>:"\/\\|?*]/g, '_').replace(/\.+$/, '').trim()
@@ -5773,6 +5878,7 @@ class ExportService {
}
if (!result.success && this.isStopError(result.error)) {
activeSessionRatios.delete(sessionId)
return 'stopped'
}
@@ -5785,14 +5891,23 @@ class ExportService {
console.error(`导出 ${sessionId} 失败:`, result.error)
}
activeSessionRatios.delete(sessionId)
completedCount++
onProgress?.({
current: completedCount,
current: computeAggregateCurrent(),
total: sessionIds.length,
currentSession: sessionInfo.displayName,
currentSessionId: sessionId,
phase: 'exporting'
})
return 'done'
} catch (error) {
if (this.isStopError(error)) {
activeSessionRatios.delete(sessionId)
return 'stopped'
}
throw error
}
}
const workers = Array.from({ length: Math.min(sessionConcurrency, queue.length) }, async () => {