diff --git a/frontend/composables/useApi.js b/frontend/composables/useApi.js
index 8263de95..f7ae1ece 100644
--- a/frontend/composables/useApi.js
+++ b/frontend/composables/useApi.js
@@ -425,6 +425,28 @@ export const useApi = () => {
return await request(url)
}
+ const startSnsFullSync = async (params = {}) => {
+ const query = new URLSearchParams()
+ if (params && params.account) query.set('account', params.account)
+ const url = '/sns/realtime/full_sync' + (query.toString() ? `?${query.toString()}` : '')
+ return await request(url, { method: 'POST' })
+ }
+
+ const getSnsFullSyncStatus = async (params = {}) => {
+ const query = new URLSearchParams()
+ if (params && params.account) query.set('account', params.account)
+ const url = '/sns/realtime/full_sync/status' + (query.toString() ? `?${query.toString()}` : '')
+ return await request(url)
+ }
+
+ const cancelSnsFullSync = async (params = {}) => {
+ const query = new URLSearchParams()
+ if (params && params.account) query.set('account', params.account)
+ if (params && params.sync_id) query.set('sync_id', String(params.sync_id))
+ const url = '/sns/realtime/full_sync' + (query.toString() ? `?${query.toString()}` : '')
+ return await request(url, { method: 'DELETE' })
+ }
+
const openChatMediaFolder = async (params = {}) => {
const query = new URLSearchParams()
if (params && params.account) query.set('account', params.account)
@@ -1117,6 +1139,9 @@ export const useApi = () => {
listSnsUsers,
syncSnsRealtimeLatest,
getSnsSnapshotStatus,
+ startSnsFullSync,
+ getSnsFullSyncStatus,
+ cancelSnsFullSync,
openChatMediaFolder,
downloadChatEmoji,
saveMediaKeys,
diff --git a/frontend/pages/sns.vue b/frontend/pages/sns.vue
index 039c1649..90aaa930 100644
--- a/frontend/pages/sns.vue
+++ b/frontend/pages/sns.vue
@@ -13,10 +13,25 @@
:disabled="!selectedAccount || isRefreshing || isLoading"
@click="refreshSnsData"
>
- {{ isRefreshing ? '刷新中…' : '刷新' }}
+ {{ snsFullSyncButtonLabel }}
+
+ {{ snsFullSyncStatusText }}
+
+
{
+ const status = String(snsFullSyncJob.value?.status || '')
+ return status === 'queued' || status === 'running'
+})
+const snsFullSyncButtonLabel = computed(() => {
+ if (isRefreshing.value) return '启动中…'
+ return isSnsFullSyncActive.value ? '同步中' : '刷新'
+})
+const snsFullSyncStatusText = computed(() => {
+ const job = snsFullSyncJob.value
+ const status = String(job?.status || '')
+ const progress = job?.progress || {}
+ const changed = Math.max(0, Number(progress?.changed || 0))
+ const percent = Math.max(0, Math.min(100, Number(progress?.percent || 0)))
+ if (status === 'queued') return `等待同步 · 已变化 ${changed}`
+ if (status === 'running') return `${percent}% · 已变化 ${changed}`
+ if (status === 'done') return `同步完成 · 已变化 ${changed}`
+ if (status === 'cancelled') return `已取消 · 已保留变化 ${changed}`
+ if (status === 'error') return `同步失败 · 已保留变化 ${changed}`
+ return ''
+})
// 首次水合时保持按钮禁用，挂载后再按账号状态启用，避免服务端 disabled 残留。
const isSnsPageMounted = ref(false)
const error = ref('')
@@ -3360,11 +3398,12 @@ const loadAccounts = async () => {
}
}
-let refreshQueued = false
const SNS_REALTIME_SYNC_TIMEOUT_MS = 10000
const SNS_VISIBLE_RECONCILE_BUFFER_MIN = 20
const SNS_VISIBLE_RECONCILE_WINDOW_MAX = 200
-const SNS_MANUAL_REFRESH_SCAN_LIMIT = 200
+const SNS_INCREMENTAL_DEFAULT_SCAN_LIMIT = 200
+const SNS_FULL_SYNC_MERGE_THROTTLE_MS = 400
+const SNS_FULL_SYNC_USER_REFRESH_BATCHES = 5
const SNS_EVENT_RECONNECT_DELAYS_MS = [1000, 2000, 5000, 10000, 30000]
let snsSnapshotVersion = ''
let snsRealtimeSyncInFlight = null
@@ -3375,6 +3414,10 @@ let snsEventReconnectTimer = null
let snsEventReconnectAttempt = 0
let snsLastEventSequence = 0
let snsQueuedRealtimeEvent = null
+let snsQueuedFullSyncMerge = null
+let snsFullSyncMergePromise = null
+let snsFullSyncMergeTimer = null
+let snsFullSyncLastUserRefreshBatch = 0
let snsPageUnmounted = false
let snsVisiblePostStart = 0
let snsVisiblePostEnd = -1
@@ -3429,7 +3472,7 @@ const beginSnsRealtimeSync = (
const syncLatestSnsWithTimeout = async (
account,
{
- maxScan = SNS_MANUAL_REFRESH_SCAN_LIMIT,
+ maxScan = SNS_INCREMENTAL_DEFAULT_SCAN_LIMIT,
scanOffset = null,
usernames = [],
waitForCurrent = false
@@ -3506,80 +3549,58 @@ const describeSnsSyncFailure = (failure) => {
}
const refreshSnsData = async () => {
- if (!String(selectedAccount.value || '').trim()) return
- if (isRefreshing.value) {
- refreshQueued = true
- return
- }
-
+ const account = String(selectedAccount.value || '').trim()
+ if (!account || isRefreshing.value) return
isRefreshing.value = true
+ syncWarning.value = ''
try {
- do {
- refreshQueued = false
- const account = String(selectedAccount.value || '').trim()
- if (!account) break
- const reconcileWindow = getSnsVisibleReconcileWindow()
- const selectedUsername = String(selectedSnsUser.value || '').trim()
- let shouldMergeTimeline = false
-
- // 按钮本身显示刷新状态，避免插入提示行导致联系人列表上下跳动。
- syncWarning.value = ''
- const activeReconcile = snsVisibleReconcilePromise
- if (activeReconcile) {
- try {
- await activeReconcile
- } catch {}
+ const response = await api.startSnsFullSync({ account })
+ if (account !== String(selectedAccount.value || '').trim()) return
+ const job = response?.job || null
+ applySnsFullSyncJob(job)
+ isSnsFullSyncCancelling.value = false
+ if (job) {
+ const status = String(job?.status || '')
+ const final = status === 'done' || status === 'error' || status === 'cancelled'
+ const version = String(job?.snapshotVersion || '').trim()
+ if (final || (version && version !== snsSnapshotVersion)) {
+ queueSnsFullSyncMerge(job, { final })
}
- try {
- const syncResult = await syncLatestSnsWithTimeout(account, {
- maxScan: SNS_MANUAL_REFRESH_SCAN_LIMIT,
- scanOffset: reconcileWindow.scanOffset,
- usernames: selectedUsername ? [selectedUsername] : [],
- waitForCurrent: true
- })
- const syncStatus = String(syncResult?.status || '').trim().toLowerCase()
- if (syncStatus === 'ok' || syncStatus === 'noop') {
- syncWarning.value = ''
- const responseVersion = String(syncResult?.snapshotVersion || '').trim()
- shouldMergeTimeline = !!(
- Number(syncResult?.changed ?? syncResult?.upserted ?? 0) > 0
- || syncResult?.snapshotChanged === true
- || (responseVersion && snsSnapshotVersion && responseVersion !== snsSnapshotVersion)
- )
- } else {
- syncWarning.value = describeSnsSyncFailure(syncResult)
- console.warn('同步最新朋友圈未成功，继续读取已解密快照', syncResult)
- }
- } catch (e) {
- syncWarning.value = describeSnsSyncFailure(e)
- console.warn('同步最新朋友圈失败，继续读取已解密快照', e)
- }
- if (!shouldMergeTimeline) {
- try {
- const localVersion = await readSnsSnapshotVersion(account)
- shouldMergeTimeline = !!(
- localVersion
- && snsSnapshotVersion
- && localVersion !== snsSnapshotVersion
- )
- } catch {}
- }
- if (account !== String(selectedAccount.value || '').trim()) break
- const refreshTasks = [loadSelfInfo()]
- if (shouldMergeTimeline) {
- refreshTasks.push(
- loadSnsUsers({ preserveExisting: true }),
- mergeVisiblePostsWindow(reconcileWindow)
- )
- }
- await Promise.all(refreshTasks)
- await updateSnsSnapshotBaseline(account)
- } while (refreshQueued)
+ }
+ } catch (e) {
+ if (account === String(selectedAccount.value || '').trim()) {
+ syncWarning.value = describeSnsSyncFailure(e)
+ }
} finally {
isRefreshing.value = false
}
}
+const cancelSnsFullSync = async () => {
+ const account = String(selectedAccount.value || '').trim()
+ const syncId = String(snsFullSyncJob.value?.syncId || '').trim()
+ if (!account || !syncId || !isSnsFullSyncActive.value || isSnsFullSyncCancelling.value) return
+ isSnsFullSyncCancelling.value = true
+ try {
+ const response = await api.cancelSnsFullSync({ account, sync_id: syncId })
+ if (
+ account === String(selectedAccount.value || '').trim()
+ && syncId === String(snsFullSyncJob.value?.syncId || '')
+ && response?.job
+ ) {
+ snsFullSyncJob.value = response.job
+ }
+ } catch (e) {
+ if (account === String(selectedAccount.value || '').trim()) {
+ syncWarning.value = describeSnsSyncFailure(e)
+ }
+ } finally {
+ if (account === String(selectedAccount.value || '').trim()) {
+ isSnsFullSyncCancelling.value = false
+ }
+ }
+}
+
let postsRequestGeneration = 0
const isCurrentPostsRequest = (generation, account) => {
@@ -3885,6 +3906,145 @@ const mergeVisiblePostsWindow = async (windowRange = getSnsVisibleReconcileWindo
const mergeLatestPosts = async () => mergeVisiblePostsWindow(getSnsVisibleReconcileWindow())
+const clearSnsFullSyncMergeTimer = () => {
+ if (!process.client || snsFullSyncMergeTimer === null) return
+ window.clearTimeout(snsFullSyncMergeTimer)
+ snsFullSyncMergeTimer = null
+}
+
+const drainSnsFullSyncMerge = () => {
+ clearSnsFullSyncMergeTimer()
+ if (snsFullSyncMergePromise) return snsFullSyncMergePromise
+
+ let trackedPromise = null
+ const task = (async () => {
+ let merged = false
+ while (snsQueuedFullSyncMerge) {
+ const pending = snsQueuedFullSyncMerge
+ snsQueuedFullSyncMerge = null
+ const account = String(pending?.account || '')
+ if (
+ !process.client
+ || snsPageUnmounted
+ || document.visibilityState !== 'visible'
+ || !account
+ || account !== String(selectedAccount.value || '').trim()
+ ) continue
+
+ const job = pending?.job || {}
+ const progress = job?.progress || {}
+ const snapshotVersion = String(job?.snapshotVersion || pending?.snapshotVersion || '').trim()
+ const changed = Math.max(0, Number(progress?.changed || 0))
+ const batch = Math.max(0, Number(progress?.batchesCompleted || 0))
+ const finalMerge = !!pending?.final
+ const snapshotChanged = !!(
+ snapshotVersion
+ && snapshotVersion !== snsSnapshotVersion
+ && (changed > 0 || finalMerge)
+ )
+ if (!snapshotChanged && !finalMerge) continue
+
+ const activeReconcile = snsVisibleReconcilePromise
+ if (activeReconcile) {
+ try {
+ await activeReconcile
+ } catch {}
+ }
+
+ const shouldRefreshUsers = finalMerge
+ || batch - snsFullSyncLastUserRefreshBatch >= SNS_FULL_SYNC_USER_REFRESH_BATCHES
+ const tasks = [mergeVisiblePostsWindow(getSnsVisibleReconcileWindow())]
+ if (shouldRefreshUsers) tasks.push(loadSnsUsers({ preserveExisting: true }))
+ const results = await Promise.all(tasks)
+ const timelineMerged = results[0] === true
+ if (!timelineMerged) continue
+
+ merged = true
+ if (shouldRefreshUsers) snsFullSyncLastUserRefreshBatch = batch
+ if (snapshotVersion) {
+ snsSnapshotVersion = snapshotVersion
+ } else {
+ await updateSnsSnapshotBaseline(account)
+ }
+ }
+ return merged
+ })()
+
+ trackedPromise = task.finally(() => {
+ if (snsFullSyncMergePromise === trackedPromise) snsFullSyncMergePromise = null
+ if (snsQueuedFullSyncMerge) void drainSnsFullSyncMerge()
+ })
+ snsFullSyncMergePromise = trackedPromise
+ return trackedPromise
+}
+
+// 全量同步事件使用累计进度；中间事件即使被合并，下一次事件仍能恢复正确状态。
+const queueSnsFullSyncMerge = (job, { final = false } = {}) => {
+ const account = String(selectedAccount.value || '').trim()
+ if (!account || !job) return null
+ const previous = snsQueuedFullSyncMerge
+ snsQueuedFullSyncMerge = {
+ account,
+ job,
+ snapshotVersion: String(job?.snapshotVersion || ''),
+ final: !!(final || previous?.final)
+ }
+
+ if (final) {
+ clearSnsFullSyncMergeTimer()
+ return drainSnsFullSyncMerge()
+ }
+ if (!process.client || snsFullSyncMergePromise || snsFullSyncMergeTimer !== null) {
+ return snsFullSyncMergePromise
+ }
+ snsFullSyncMergeTimer = window.setTimeout(() => {
+ snsFullSyncMergeTimer = null
+ void drainSnsFullSyncMerge()
+ }, SNS_FULL_SYNC_MERGE_THROTTLE_MS)
+ return null
+}
+
+const applySnsFullSyncJob = (job) => {
+ const previousSyncId = String(snsFullSyncJob.value?.syncId || '')
+ const nextSyncId = String(job?.syncId || '')
+ if (nextSyncId && nextSyncId !== previousSyncId) {
+ snsFullSyncLastUserRefreshBatch = 0
+ }
+ snsFullSyncJob.value = job || null
+ const status = String(job?.status || '')
+ if (status !== 'queued' && status !== 'running') {
+ isSnsFullSyncCancelling.value = false
+ }
+ if (status === 'error') {
+ syncWarning.value = String(job?.error?.message || '朋友圈全量同步失败，请稍后重试')
+ } else if (status === 'done' || status === 'cancelled') {
+ syncWarning.value = ''
+ }
+}
+
+const restoreSnsFullSyncStatus = async (account) => {
+ const requestedAccount = String(account || '').trim()
+ if (!requestedAccount) return null
+ try {
+ const response = await api.getSnsFullSyncStatus({ account: requestedAccount })
+ if (requestedAccount !== String(selectedAccount.value || '').trim()) return null
+ const job = response?.job || null
+ applySnsFullSyncJob(job)
+ if (job) {
+ const status = String(job?.status || '')
+ const final = status === 'done' || status === 'error' || status === 'cancelled'
+ const version = String(job?.snapshotVersion || '').trim()
+ if (final || (version && version !== snsSnapshotVersion)) {
+ queueSnsFullSyncMerge(job, { final })
+ }
+ }
+ return job
+ } catch {
+ // 状态恢复失败不影响本地快照浏览，SSE 重连后还会再次核对。
+ return null
+ }
+}
+
// 首屏三路并行读取本地快照，不等待实时同步。
const loadLocalSnsData = async () => {
const account = String(selectedAccount.value || '').trim()
@@ -4065,6 +4225,7 @@ const onSnsRealtimeReady = async (event) => {
snsEventReconnectAttempt = 0
snsLastEventSequence = Math.max(snsLastEventSequence, Number(payload?.sequence || 0))
+ await restoreSnsFullSyncStatus(account)
if (payload?.watcherAvailable === false) {
syncWarning.value = String(payload?.message || '系统文件通知不可用，请使用手动刷新')
return
@@ -4112,6 +4273,24 @@ const onSnsRealtimeSyncError = (event) => {
syncWarning.value = String(payload?.message || '朋友圈实时同步失败，请使用手动刷新')
}
+const onSnsFullSyncEvent = (event) => {
+ const payload = parseSnsRealtimeEvent(event)
+ const account = String(selectedAccount.value || '').trim()
+ if (!payload?.job || String(payload?.account || '') !== account) return
+ const sequence = Number(payload?.sequence || 0)
+ if (sequence > 0 && sequence <= snsLastEventSequence) return
+ snsLastEventSequence = Math.max(snsLastEventSequence, sequence)
+
+ const job = payload.job
+ applySnsFullSyncJob(job)
+ const status = String(job?.status || '')
+ const final = status === 'done' || status === 'error' || status === 'cancelled'
+ const snapshotVersion = String(job?.snapshotVersion || payload?.snapshotVersion || '').trim()
+ if (final || (snapshotVersion && snapshotVersion !== snsSnapshotVersion)) {
+ queueSnsFullSyncMerge(job, { final })
+ }
+}
+
function connectSnsEventStream() {
if (!process.client || snsPageUnmounted || document.visibilityState !== 'visible') return
const account = String(selectedAccount.value || '').trim()
@@ -4130,6 +4309,10 @@ function connectSnsEventStream() {
source.addEventListener('ready', onSnsRealtimeReady)
source.addEventListener('change', onSnsRealtimeChange)
source.addEventListener('sync_error', onSnsRealtimeSyncError)
+ source.addEventListener('full_sync_progress', onSnsFullSyncEvent)
+ source.addEventListener('full_sync_done', onSnsFullSyncEvent)
+ source.addEventListener('full_sync_error', onSnsFullSyncEvent)
+ source.addEventListener('full_sync_cancelled', onSnsFullSyncEvent)
source.onerror = () => {
if (source !== snsEventSource) return
closeSnsEventStream()
@@ -4146,8 +4329,13 @@ watch(
async (v, oldV) => {
if (v !== oldV) {
closeSnsEventStream({ resetAttempt: true })
+ clearSnsFullSyncMergeTimer()
snsLastEventSequence = 0
snsQueuedRealtimeEvent = null
+ snsQueuedFullSyncMerge = null
+ snsFullSyncJob.value = null
+ isSnsFullSyncCancelling.value = false
+ snsFullSyncLastUserRefreshBatch = 0
snsSnapshotVersion = ''
}
if (v && v !== oldV) {
@@ -4172,6 +4360,7 @@ watch(
resetSnsMediaErrors()
if (previewCtx.value) closeImagePreview()
await loadLocalSnsData()
+ await restoreSnsFullSyncStatus(String(v || ''))
// 首屏就绪后建立事件连接；后端启动同步或重连差异由 ready 事件补齐。
connectSnsEventStream()
}
@@ -4248,6 +4437,7 @@ const runPassiveSnsRefresh = async () => {
if (!String(selectedAccount.value || '').trim()) return
// 窗口重新可见时只核对一次本地版本，然后恢复 SSE。
await reconcileSnsSnapshotOnce()
+ await restoreSnsFullSyncStatus(String(selectedAccount.value || ''))
connectSnsEventStream()
}
@@ -4290,7 +4480,9 @@ onUnmounted(() => {
passiveRefreshTimer = null
}
closeSnsEventStream({ resetAttempt: true })
+ clearSnsFullSyncMergeTimer()
snsQueuedRealtimeEvent = null
+ snsQueuedFullSyncMerge = null
if (snsVisibleWindowRaf !== null) {
window.cancelAnimationFrame(snsVisibleWindowRaf)
snsVisibleWindowRaf = null
diff --git a/frontend/tests/sns-page-initialization.test.mjs b/frontend/tests/sns-page-initialization.test.mjs
index df0c314e..b7807ad5 100644
--- a/frontend/tests/sns-page-initialization.test.mjs
+++ b/frontend/tests/sns-page-initialization.test.mjs
@@ -54,10 +54,11 @@ test('朋友圈使用 SSE 事件单飞核对随视口浮动的上下窗口', asy
assert.match(source, /const SNS_VISIBLE_RECONCILE_BUFFER_MIN = 20/)
assert.match(source, /const SNS_VISIBLE_RECONCILE_WINDOW_MAX = 200/)
- assert.match(source, /const SNS_MANUAL_REFRESH_SCAN_LIMIT = 200/)
+ assert.match(source, /const SNS_INCREMENTAL_DEFAULT_SCAN_LIMIT = 200/)
assert.match(source, /const SNS_EVENT_RECONNECT_DELAYS_MS = \[1000, 2000, 5000, 10000, 30000\]/)
assert.match(source, /new EventSource\([\s\S]*?\/sns\/realtime\/events\?account=/)
assert.match(source, /source\.addEventListener\('change', onSnsRealtimeChange\)/)
+ assert.match(source, /source\.addEventListener\('full_sync_progress', onSnsFullSyncEvent\)/)
assert.match(source, /const versionChanged = !!\(version && version !== snsSnapshotVersion\)/)
assert.match(source, /api\.syncSnsRealtimeLatest\(\{[\s\S]*?force: 1,[\s\S]*?max_scan: maxScan/)
assert.match(source, /if \(snsVisibleReconcilePromise\) return snsVisibleReconcilePromise/)
@@ -74,6 +75,25 @@ test('朋友圈使用 SSE 事件单飞核对随视口浮动的上下窗口', asy
})
+test('朋友圈手动刷新启动全账号任务并可恢复、取消和无感合并', async () => {
+ const source = await readFile(new URL('../pages/sns.vue', import.meta.url), 'utf8')
+ const apiSource = await readFile(new URL('../composables/useApi.js', import.meta.url), 'utf8')
+ const refresh = source.split('const refreshSnsData = async () => {', 2)[1]
+ .split(/\r?\n\r?\nconst cancelSnsFullSync/, 1)[0]
+
+ assert.match(apiSource, /const startSnsFullSync = async \(params = \{\}\) => \{[\s\S]*?\/sns\/realtime\/full_sync/)
+ assert.match(apiSource, /const getSnsFullSyncStatus = async/)
+ assert.match(apiSource, /const cancelSnsFullSync = async/)
+ assert.match(refresh, /api\.startSnsFullSync\(\{ account \}\)/)
+ assert.doesNotMatch(refresh, /selectedSnsUser|scanOffset|usernames|syncLatestSnsWithTimeout/)
+ assert.match(source, /const restoreSnsFullSyncStatus = async \(account\) =>/)
+ assert.match(source, /await restoreSnsFullSyncStatus\(String\(v \|\| ''\)\)/)
+ assert.match(source, /const SNS_FULL_SYNC_MERGE_THROTTLE_MS = 400/)
+ assert.match(source, /mergeVisiblePostsWindow\(getSnsVisibleReconcileWindow\(\)\)/)
+ assert.match(source, /restoreSnsScrollAnchor\(anchor\)/)
+})
+
+
test('朋友圈导出按钮在客户端挂载后再解除禁用，避免水合残留', async () => {
const source = await readFile(new URL('../pages/sns.vue', import.meta.url), 'utf8')
diff --git a/src/wechat_decrypt_tool/api.py b/src/wechat_decrypt_tool/api.py
index 3c95f155..e4372a08 100644
--- a/src/wechat_decrypt_tool/api.py
+++ b/src/wechat_decrypt_tool/api.py
@@ -281,8 +281,12 @@ async def _startup_background_jobs() -> None:
logger.exception("Failed to start realtime autosync service")
try:
SNS_REALTIME_AUTOSYNC.start()
- except Exception:
+ except Exception as exc:
logger.exception("Failed to start SNS realtime autosync service")
+ logger.error(
+ "[sns.incremental-sync] status=error phase=service-start error_type=%s",
+ type(exc).__name__,
+ )
@app.on_event("shutdown")
diff --git a/src/wechat_decrypt_tool/routers/sns.py b/src/wechat_decrypt_tool/routers/sns.py
index f676a640..48064363 100644
--- a/src/wechat_decrypt_tool/routers/sns.py
+++ b/src/wechat_decrypt_tool/routers/sns.py
@@ -13,6 +13,7 @@
import subprocess
import threading
import time
+import uuid
import xml.etree.ElementTree as ET
from typing import Any, Optional
from urllib.parse import urlparse
@@ -31,6 +32,7 @@
from ..path_fix import PathFixRoute
from ..perf_trace import create_perf_trace
from ..sns_realtime_autosync import SNS_REALTIME_AUTOSYNC
+from ..sns_full_sync import SNS_FULL_SYNC
from .. import sns_media as _sns_media
from ..wcdb_realtime import (
WCDBRealtimeError,
@@ -535,6 +537,11 @@ def _pack_blob(value: Any) -> Optional[bytes]:
len(rows),
error_text,
)
+ logger.warning(
+ "[sns.incremental-sync] status=error phase=writing prepared=%s error_type=%s",
+ len(rows),
+ type(e).__name__,
+ )
try:
conn.rollback()
except Exception:
@@ -1796,6 +1803,28 @@ async def event_stream():
)
+@router.post("/api/sns/realtime/full_sync", summary="启动朋友圈全量缓存同步")
+def start_sns_realtime_full_sync(account: Optional[str] = None):
+ account_dir = _resolve_account_dir(account)
+ job, reused = SNS_FULL_SYNC.start(account_dir)
+ return {"status": "ok", "reused": reused, "job": job}
+
+
+@router.get("/api/sns/realtime/full_sync/status", summary="获取朋友圈全量同步状态")
+def get_sns_realtime_full_sync_status(account: Optional[str] = None):
+ account_dir = _resolve_account_dir(account)
+ return {"status": "ok", "job": SNS_FULL_SYNC.get(account_dir)}
+
+
+@router.delete("/api/sns/realtime/full_sync", summary="取消朋友圈全量缓存同步")
+def cancel_sns_realtime_full_sync(account: Optional[str] = None, sync_id: str = ""):
+ account_dir = _resolve_account_dir(account)
+ job, accepted = SNS_FULL_SYNC.cancel(account_dir, sync_id)
+ if not accepted:
+ raise HTTPException(status_code=409, detail="同步任务已结束或任务标识不匹配")
+ return {"status": "ok", "cancelled": True, "job": job}
+
+
@router.post("/api/sns/realtime/sync_latest", summary="实时朋友圈同步到解密库（增量）")
def sync_sns_realtime_timeline_latest(
account: Optional[str] = None,
@@ -1809,6 +1838,12 @@ def sync_sns_realtime_timeline_latest(
This is best-effort and intentionally **append-only**: we never delete rows from the decrypted snapshot
even if the post is deleted/hidden later, so users can still browse/export historical cached content.
"""
+ sync_request_id = uuid.uuid4().hex
+ sync_started = time.perf_counter()
+ logger.info(
+ "[sns.incremental-sync] status=running request_id=%s phase=connecting",
+ sync_request_id,
+ )
try:
lim = int(max_scan or 200)
except Exception:
@@ -1855,6 +1890,21 @@ def _sync_response(
result["highwaterAdvanced"] = bool(highwater_advanced)
result["scanOffset"] = int(requested_scan_offset)
result["scanLimit"] = int(lim)
+ status = str(result.get("status") or "error").strip().lower()
+ raw_code = str(result.get("error") or result.get("reason") or "").strip().lower()
+ code = raw_code if re.fullmatch(r"[a-z0-9_.-]{1,80}", raw_code) else ""
+ log_method = logger.error if status == "error" else logger.info
+ log_method(
+ "[sns.incremental-sync] status=%s request_id=%s phase=finalizing code=%s scanned=%s prepared=%s changed=%s unchanged=%s elapsed_ms=%s",
+ status,
+ sync_request_id,
+ code,
+ int(result.get("scanned") or 0),
+ int(prepared),
+ int(changed),
+ int(unchanged),
+ int((time.perf_counter() - sync_started) * 1000),
+ )
return result
# If there is no local decrypted sns.db yet, force a first-time materialization.
@@ -1867,6 +1917,11 @@ def _sync_response(
info = WCDB_REALTIME.get_status(account_dir)
available = bool(info.get("dll_present") and info.get("key_present") and info.get("db_storage_dir"))
if not available:
+ logger.error(
+ "[sns.incremental-sync] status=error request_id=%s phase=connecting code=realtime_not_available error_type=AvailabilityError elapsed_ms=%s",
+ sync_request_id,
+ int((time.perf_counter() - sync_started) * 1000),
+ )
raise HTTPException(status_code=404, detail="WCDB realtime not available.")
st = _read_sns_realtime_sync_state(account_dir)
@@ -1879,9 +1934,18 @@ def _sync_response(
if last_max_id_u <= 0:
last_max_id_u = _max_sns_timeline_tid_unsigned_in_decrypted_sqlite(account_dir / "sns.db")
- conn = WCDB_REALTIME.ensure_connected(account_dir)
+ try:
+ conn = WCDB_REALTIME.ensure_connected(account_dir)
+ except Exception as exc:
+ logger.error(
+ "[sns.incremental-sync] status=error request_id=%s phase=connecting code=connection_failed error_type=%s elapsed_ms=%s",
+ sync_request_id,
+ type(exc).__name__,
+ int((time.perf_counter() - sync_started) * 1000),
+ )
+ raise
- t0 = time.perf_counter()
+ t0 = sync_started
rows: list[dict[str, Any]] = []
max_id_u = 0
upsert_rows: list[tuple[int, str, str, Optional[Any]]] = []
@@ -2107,6 +2171,16 @@ def _sync_response(
write_success = changed_count == prepared_count
write_error = ""
+ logger.info(
+ "[sns.incremental-sync] status=running request_id=%s phase=scanning batches=1 scanned=%s prepared=%s changed=%s unchanged=%s elapsed_ms=%s",
+ sync_request_id,
+ len(rows),
+ prepared_count,
+ changed_count,
+ unchanged_count,
+ int((time.perf_counter() - sync_started) * 1000),
+ )
+
prepared_tids = {int(row[0]) for row in upsert_rows}
missing_required_tids = required_tids - prepared_tids
snapshot_complete = bool(upsert_rows) and all((
@@ -2125,6 +2199,15 @@ def _sync_response(
unchanged_count,
len(missing_required_tids),
)
+ logger.warning(
+ "[sns.incremental-sync] status=error request_id=%s phase=writing code=snapshot_write_incomplete scanned=%s prepared=%s changed=%s unchanged=%s skipped=%s",
+ sync_request_id,
+ len(rows),
+ prepared_count,
+ changed_count,
+ unchanged_count,
+ len(missing_required_tids),
+ )
return _sync_response({
"status": "error",
"error": "decrypted_snapshot_write_incomplete",
@@ -2143,6 +2226,11 @@ def _sync_response(
len(rows),
last_max_id_u,
)
+ logger.warning(
+ "[sns.incremental-sync] status=skipped request_id=%s phase=scanning code=scan_cap_reached scanned=%s",
+ sync_request_id,
+ len(rows),
+ )
return _sync_response({
"status": "skipped",
"reason": "backlog exceeds scan cap",
@@ -2161,6 +2249,10 @@ def _sync_response(
st2["updatedAt"] = int(time.time())
if _write_sns_realtime_sync_state(account_dir, st2) is False:
logger.warning("[sns-sync] state write failed account=%s", account_dir.name)
+ logger.warning(
+ "[sns.incremental-sync] status=error request_id=%s phase=finalizing code=sync_state_write_failed",
+ sync_request_id,
+ )
return _sync_response({
"status": "error",
"error": "sync_state_write_failed",
diff --git a/src/wechat_decrypt_tool/sns_full_sync.py b/src/wechat_decrypt_tool/sns_full_sync.py
new file mode 100644
index 00000000..622f12ca
--- /dev/null
+++ b/src/wechat_decrypt_tool/sns_full_sync.py
@@ -0,0 +1,557 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+import re
+import threading
+import time
+import uuid
+from typing import Any, Optional
+
+from .logging_config import get_logger
+from .sns_realtime_autosync import SNS_REALTIME_AUTOSYNC
+from .wcdb_realtime import WCDB_REALTIME, exec_query as _wcdb_exec_query
+
+
+logger = get_logger(__name__)
+
+_BATCH_SIZE = 200
+_ACTIVE_STATUSES = {"queued", "running"}
+_SAFE_ERROR_TYPE_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,79}$")
+
+
+@dataclass
+class _FullSyncProgress:
+ phase: str = "connecting"
+ source_rows_total: int = 0
+ source_rows_scanned: int = 0
+ batches_completed: int = 0
+ prepared: int = 0
+ changed: int = 0
+ unchanged: int = 0
+ skipped: int = 0
+
+
+@dataclass
+class _FullSyncJob:
+ account_dir: Path
+ sync_id: str = field(default_factory=lambda: uuid.uuid4().hex)
+ status: str = "queued"
+ created_at: int = field(default_factory=lambda: int(time.time() * 1000))
+ started_at: Optional[int] = None
+ finished_at: Optional[int] = None
+ cancel_requested: bool = False
+ snapshot_version: str = ""
+ progress: _FullSyncProgress = field(default_factory=_FullSyncProgress)
+ error: Optional[dict[str, str]] = None
+ cancel_event: threading.Event = field(default_factory=threading.Event, repr=False)
+
+
+class SnsFullSyncManager:
+ """朋友圈全量缓存同步任务管理器。
+
+ 每个账号只保留一个活动任务，同时通过全局信号量保证任意时刻只扫描一个账号。
+ """
+
+ def __init__(self) -> None:
+ self._mu = threading.RLock()
+ self._global_slot = threading.BoundedSemaphore(1)
+ self._latest_by_account: dict[str, _FullSyncJob] = {}
+
+ @staticmethod
+ def _account_key(account_dir: Path) -> str:
+ # 账号仅作为内存索引，不进入日志或公开任务结构。
+ return str(Path(account_dir).resolve())
+
+ @staticmethod
+ def _safe_error_type(exc: BaseException) -> str:
+ name = type(exc).__name__
+ return name if _SAFE_ERROR_TYPE_RE.fullmatch(name) else "Exception"
+
+ def _public_job_locked(self, job: _FullSyncJob) -> dict[str, Any]:
+ progress = job.progress
+ total = max(0, int(progress.source_rows_total))
+ scanned = max(0, int(progress.source_rows_scanned))
+ if job.status == "done":
+ percent = 100
+ elif total <= 0:
+ percent = 0
+ else:
+ percent = min(99, int(scanned * 100 / total))
+
+ payload: dict[str, Any] = {
+ "syncId": job.sync_id,
+ "status": job.status,
+ "createdAt": job.created_at,
+ "startedAt": job.started_at,
+ "finishedAt": job.finished_at,
+ "cancelRequested": bool(job.cancel_requested),
+ "snapshotVersion": job.snapshot_version,
+ "progress": {
+ "phase": progress.phase,
+ "sourceRowsTotal": total,
+ "sourceRowsScanned": scanned,
+ "batchesCompleted": int(progress.batches_completed),
+ "prepared": int(progress.prepared),
+ "changed": int(progress.changed),
+ "unchanged": int(progress.unchanged),
+ "skipped": int(progress.skipped),
+ "percent": percent,
+ },
+ }
+ if job.error is not None:
+ payload["error"] = dict(job.error)
+ return payload
+
+ def get(self, account_dir: Path) -> Optional[dict[str, Any]]:
+ key = self._account_key(account_dir)
+ with self._mu:
+ job = self._latest_by_account.get(key)
+ return self._public_job_locked(job) if job is not None else None
+
+ def start(self, account_dir: Path) -> tuple[dict[str, Any], bool]:
+ resolved = Path(account_dir).resolve()
+ key = self._account_key(resolved)
+ with self._mu:
+ current = self._latest_by_account.get(key)
+ if current is not None and current.status in _ACTIVE_STATUSES:
+ return self._public_job_locked(current), True
+
+ job = _FullSyncJob(account_dir=resolved)
+ self._latest_by_account[key] = job
+ public = self._public_job_locked(job)
+
+ worker = threading.Thread(
+ target=self._run_job,
+ args=(key, job),
+ name=f"sns-full-sync-{job.sync_id[:8]}",
+ daemon=True,
+ )
+ logger.info(
+ "[sns.full-sync] status=queued sync_id=%s phase=connecting",
+ job.sync_id,
+ )
+ self._publish(job, "full_sync_progress")
+ try:
+ worker.start()
+ except Exception as exc:
+ self._finish_error(
+ job,
+ code="sync_worker_unavailable",
+ message="朋友圈同步线程不可用，请稍后重试",
+ exc=exc,
+ started_monotonic=time.monotonic(),
+ )
+ # 极小数据集可能在线程启动后立即完成，返回最新状态避免旧 queued 覆盖 SSE 终态。
+ return self.get(resolved) or public, False
+
+ def cancel(self, account_dir: Path, sync_id: str) -> tuple[Optional[dict[str, Any]], bool]:
+ key = self._account_key(account_dir)
+ requested_id = str(sync_id or "").strip()
+ with self._mu:
+ job = self._latest_by_account.get(key)
+ if (
+ job is None
+ or job.sync_id != requested_id
+ or job.status not in _ACTIVE_STATUSES
+ ):
+ return (self._public_job_locked(job) if job is not None else None), False
+ job.cancel_requested = True
+ job.cancel_event.set()
+ return self._public_job_locked(job), True
+
+ def _publish(self, job: _FullSyncJob, event_type: str) -> None:
+ with self._mu:
+ public = self._public_job_locked(job)
+ SNS_REALTIME_AUTOSYNC.publish_external_event(
+ Path(job.account_dir).name,
+ {
+ "type": event_type,
+ "account": Path(job.account_dir).name,
+ "job": public,
+ "snapshotVersion": public.get("snapshotVersion") or "",
+ "timestamp": int(time.time() * 1000),
+ },
+ )
+
+ def _finish_cancelled(self, job: _FullSyncJob, started_monotonic: float) -> None:
+ with self._mu:
+ job.status = "cancelled"
+ job.finished_at = int(time.time() * 1000)
+ logger.info(
+ "[sns.full-sync] status=cancelled sync_id=%s phase=%s batches=%s scanned=%s prepared=%s changed=%s unchanged=%s skipped=%s elapsed_ms=%s",
+ job.sync_id,
+ job.progress.phase,
+ job.progress.batches_completed,
+ job.progress.source_rows_scanned,
+ job.progress.prepared,
+ job.progress.changed,
+ job.progress.unchanged,
+ job.progress.skipped,
+ int((time.monotonic() - started_monotonic) * 1000),
+ )
+ self._publish(job, "full_sync_cancelled")
+
+ def _finish_error(
+ self,
+ job: _FullSyncJob,
+ *,
+ code: str,
+ message: str,
+ exc: Optional[BaseException],
+ started_monotonic: float,
+ ) -> None:
+ error_type = self._safe_error_type(exc) if exc is not None else "SyncError"
+ with self._mu:
+ job.status = "error"
+ job.finished_at = int(time.time() * 1000)
+ job.error = {"code": code, "message": message}
+ logger.error(
+ "[sns.full-sync] status=error sync_id=%s phase=%s code=%s error_type=%s batches=%s scanned=%s prepared=%s changed=%s unchanged=%s skipped=%s elapsed_ms=%s",
+ job.sync_id,
+ job.progress.phase,
+ code,
+ error_type,
+ job.progress.batches_completed,
+ job.progress.source_rows_scanned,
+ job.progress.prepared,
+ job.progress.changed,
+ job.progress.unchanged,
+ job.progress.skipped,
+ int((time.monotonic() - started_monotonic) * 1000),
+ )
+ self._publish(job, "full_sync_error")
+
+ @staticmethod
+ def _row_value(row: dict[str, Any], name: str, default: Any = None) -> Any:
+ if name in row:
+ return row.get(name)
+ lowered = name.lower()
+ for key, value in row.items():
+ if str(key).lower() == lowered:
+ return value
+ return default
+
+ @staticmethod
+ def _source_db_path(connection: Any) -> Optional[Path]:
+ try:
+ root = Path(connection.db_storage_dir)
+ candidates = (root / "sns" / "sns.db", root / "sns.db")
+ for candidate in candidates:
+ if candidate.is_file():
+ return candidate
+ except Exception:
+ return None
+ return None
+
+ def _query(self, connection: Any, source_path: Path, sql: str) -> list[dict[str, Any]]:
+ with connection.lock:
+ rows = _wcdb_exec_query(
+ connection.handle,
+ kind="media",
+ path=str(source_path),
+ sql=sql,
+ )
+ return [row for row in (rows or []) if isinstance(row, dict)]
+
+ def _count_and_bounds(
+ self,
+ connection: Any,
+ source_path: Path,
+ ) -> tuple[str, int, Optional[int], Optional[int]]:
+ valid_where = (
+ "tid IS NOT NULL AND user_name IS NOT NULL AND user_name != '' "
+ "AND content IS NOT NULL AND content != ''"
+ )
+ last_exc: Optional[BaseException] = None
+ for cursor_column in ("rowid", "tid"):
+ sql = (
+ "SELECT COUNT(*) AS source_rows_total, "
+ f"MIN({cursor_column}) AS min_cursor, MAX({cursor_column}) AS max_cursor "
+ f"FROM SnsTimeLine WHERE {valid_where}"
+ )
+ try:
+ rows = self._query(connection, source_path, sql)
+ row = rows[0] if rows else {}
+ total = int(self._row_value(row, "source_rows_total", 0) or 0)
+ min_raw = self._row_value(row, "min_cursor")
+ max_raw = self._row_value(row, "max_cursor")
+ min_cursor = int(min_raw) if min_raw is not None else None
+ max_cursor = int(max_raw) if max_raw is not None else None
+ return cursor_column, total, min_cursor, max_cursor
+ except Exception as exc:
+ last_exc = exc
+ if last_exc is not None:
+ raise last_exc
+ raise RuntimeError("SnsTimeLine cursor is unavailable")
+
+ def _read_batch(
+ self,
+ connection: Any,
+ source_path: Path,
+ *,
+ cursor_column: str,
+ min_cursor: int,
+ max_cursor: int,
+ after_cursor: Optional[int],
+ include_pack: bool,
+ ) -> tuple[list[dict[str, Any]], bool]:
+ lower = (
+ f"{cursor_column} >= {int(min_cursor)}"
+ if after_cursor is None
+ else f"{cursor_column} > {int(after_cursor)}"
+ )
+ where_sql = (
+ f"{lower} AND {cursor_column} <= {int(max_cursor)} "
+ "AND tid IS NOT NULL AND user_name IS NOT NULL AND user_name != '' "
+ "AND content IS NOT NULL AND content != ''"
+ )
+ select_pack = ", pack_info_buf" if include_pack else ""
+ sql = (
+ f"SELECT {cursor_column} AS source_cursor, tid, user_name, content{select_pack} "
+ f"FROM SnsTimeLine WHERE {where_sql} "
+ f"ORDER BY {cursor_column} ASC LIMIT {_BATCH_SIZE}"
+ )
+ try:
+ return self._query(connection, source_path, sql), include_pack
+ except Exception:
+ if not include_pack:
+ raise
+ # 老版本源表没有 pack_info_buf，保持主记录同步能力。
+ return self._read_batch(
+ connection,
+ source_path,
+ cursor_column=cursor_column,
+ min_cursor=min_cursor,
+ max_cursor=max_cursor,
+ after_cursor=after_cursor,
+ include_pack=False,
+ )
+
+ def _run_job(self, _key: str, job: _FullSyncJob) -> None:
+ started_monotonic = time.monotonic()
+ slot_acquired = False
+ try:
+ while not job.cancel_event.is_set():
+ if self._global_slot.acquire(timeout=0.1):
+ slot_acquired = True
+ break
+ if not slot_acquired:
+ self._finish_cancelled(job, started_monotonic)
+ return
+
+ with self._mu:
+ job.status = "running"
+ job.started_at = int(time.time() * 1000)
+ job.progress.phase = "connecting"
+ logger.info(
+ "[sns.full-sync] status=running sync_id=%s phase=connecting",
+ job.sync_id,
+ )
+ self._publish(job, "full_sync_progress")
+
+ if job.cancel_event.is_set():
+ self._finish_cancelled(job, started_monotonic)
+ return
+
+ try:
+ connection = WCDB_REALTIME.ensure_connected(job.account_dir, timeout=15.0)
+ except Exception as exc:
+ self._finish_error(
+ job,
+ code="realtime_not_available",
+ message="朋友圈实时组件未连接，请确认微信已登录且数据库密钥有效",
+ exc=exc,
+ started_monotonic=started_monotonic,
+ )
+ return
+ source_path = self._source_db_path(connection)
+ if source_path is None:
+ self._finish_error(
+ job,
+ code="sns_source_not_found",
+ message="未找到微信本地朋友圈数据库",
+ exc=None,
+ started_monotonic=started_monotonic,
+ )
+ return
+
+ with self._mu:
+ job.progress.phase = "counting"
+ self._publish(job, "full_sync_progress")
+
+ try:
+ cursor_column, total, min_cursor, max_cursor = self._count_and_bounds(
+ connection,
+ source_path,
+ )
+ except Exception as exc:
+ self._finish_error(
+ job,
+ code="sns_source_schema_unsupported",
+ message="当前朋友圈数据库结构暂不支持全量同步",
+ exc=exc,
+ started_monotonic=started_monotonic,
+ )
+ return
+
+ with self._mu:
+ job.progress.source_rows_total = total
+ job.progress.phase = "scanning"
+ self._publish(job, "full_sync_progress")
+
+ # 延迟导入路由辅助函数，避免模块加载时形成循环依赖。
+ from .routers.sns import (
+ _build_sns_snapshot_status,
+ _decode_sns_text_blob,
+ _looks_like_xml_text,
+ _read_sns_realtime_sync_state,
+ _upsert_sns_timeline_rows_to_decrypted_db,
+ _write_sns_realtime_sync_state,
+ )
+
+ after_cursor: Optional[int] = None
+ include_pack = True
+ max_tid_unsigned = 0
+
+ while min_cursor is not None and max_cursor is not None:
+ if job.cancel_event.is_set():
+ self._finish_cancelled(job, started_monotonic)
+ return
+
+ rows, include_pack = self._read_batch(
+ connection,
+ source_path,
+ cursor_column=cursor_column,
+ min_cursor=min_cursor,
+ max_cursor=max_cursor,
+ after_cursor=after_cursor,
+ include_pack=include_pack,
+ )
+ if not rows:
+ break
+
+ prepared_rows: list[tuple[int, str, str, Optional[Any]]] = []
+ skipped = 0
+ for row in rows:
+ try:
+ source_cursor = int(self._row_value(row, "source_cursor"))
+ tid = int(self._row_value(row, "tid"))
+ username = str(self._row_value(row, "user_name", "") or "").strip()
+ content = _decode_sns_text_blob(self._row_value(row, "content"))
+ if (
+ not username
+ or not _looks_like_xml_text(content)
+ or "7" in content
+ ):
+ skipped += 1
+ continue
+ pack = self._row_value(row, "pack_info_buf") if include_pack else None
+ prepared_rows.append((tid, username, content, pack))
+ max_tid_unsigned = max(max_tid_unsigned, tid & 0xFFFFFFFFFFFFFFFF)
+ except Exception:
+ skipped += 1
+ continue
+
+ result = _upsert_sns_timeline_rows_to_decrypted_db(
+ job.account_dir,
+ prepared_rows,
+ source="sns.full-sync",
+ )
+ if not bool(result.get("success")):
+ self._finish_error(
+ job,
+ code="snapshot_write_failed",
+ message="朋友圈本地快照写入失败，可稍后重试",
+ exc=None,
+ started_monotonic=started_monotonic,
+ )
+ return
+
+ after_cursor = max(
+ int(self._row_value(row, "source_cursor")) for row in rows
+ )
+ snapshot = _build_sns_snapshot_status(job.account_dir)
+ with self._mu:
+ job.progress.source_rows_scanned += len(rows)
+ job.progress.batches_completed += 1
+ job.progress.prepared += int(result.get("prepared") or 0)
+ job.progress.changed += int(result.get("changed") or 0)
+ job.progress.unchanged += int(result.get("unchanged") or 0)
+ job.progress.skipped += skipped
+ job.snapshot_version = str(snapshot.get("version") or "")
+
+ logger.info(
+ "[sns.full-sync] status=running sync_id=%s phase=scanning batches=%s scanned=%s total=%s prepared=%s changed=%s unchanged=%s skipped=%s elapsed_ms=%s",
+ job.sync_id,
+ job.progress.batches_completed,
+ job.progress.source_rows_scanned,
+ job.progress.source_rows_total,
+ job.progress.prepared,
+ job.progress.changed,
+ job.progress.unchanged,
+ job.progress.skipped,
+ int((time.monotonic() - started_monotonic) * 1000),
+ )
+ self._publish(job, "full_sync_progress")
+
+ if len(rows) < _BATCH_SIZE:
+ break
+
+ if job.cancel_event.is_set():
+ self._finish_cancelled(job, started_monotonic)
+ return
+
+ with self._mu:
+ job.progress.phase = "finalizing"
+ self._publish(job, "full_sync_progress")
+
+ if max_tid_unsigned > 0:
+ state = _read_sns_realtime_sync_state(job.account_dir)
+ state["maxId"] = str(max_tid_unsigned)
+ state["updatedAt"] = int(time.time() * 1000)
+ if not _write_sns_realtime_sync_state(job.account_dir, state):
+ self._finish_error(
+ job,
+ code="sync_state_write_failed",
+ message="朋友圈同步状态写入失败，可安全重试",
+ exc=None,
+ started_monotonic=started_monotonic,
+ )
+ return
+
+ snapshot = _build_sns_snapshot_status(job.account_dir)
+ with self._mu:
+ job.status = "done"
+ job.finished_at = int(time.time() * 1000)
+ job.snapshot_version = str(snapshot.get("version") or "")
+ logger.info(
+ "[sns.full-sync] status=done sync_id=%s phase=finalizing batches=%s scanned=%s total=%s prepared=%s changed=%s unchanged=%s skipped=%s elapsed_ms=%s",
+ job.sync_id,
+ job.progress.batches_completed,
+ job.progress.source_rows_scanned,
+ job.progress.source_rows_total,
+ job.progress.prepared,
+ job.progress.changed,
+ job.progress.unchanged,
+ job.progress.skipped,
+ int((time.monotonic() - started_monotonic) * 1000),
+ )
+ self._publish(job, "full_sync_done")
+ except Exception as exc:
+ self._finish_error(
+ job,
+ code="full_sync_failed",
+ message="朋友圈全量同步失败，请稍后重试",
+ exc=exc,
+ started_monotonic=started_monotonic,
+ )
+ finally:
+ if slot_acquired:
+ try:
+ self._global_slot.release()
+ except Exception:
+ pass
+
+
+SNS_FULL_SYNC = SnsFullSyncManager()
diff --git a/src/wechat_decrypt_tool/sns_realtime_autosync.py b/src/wechat_decrypt_tool/sns_realtime_autosync.py
index 9648f953..3cf56e5a 100644
--- a/src/wechat_decrypt_tool/sns_realtime_autosync.py
+++ b/src/wechat_decrypt_tool/sns_realtime_autosync.py
@@ -176,8 +176,9 @@ def _bootstrap_accounts(self) -> None:
"""启动时只枚举一次账号；后续账号由 SSE 连接动态注册。"""
try:
accounts = list(_list_decrypted_accounts() or [])
- except Exception:
+ except Exception as exc:
logger.exception("[sns-autosync] 初始账号枚举失败")
+ logger.error("[sns.incremental-sync] status=error phase=account-scan error_type=%s", type(exc).__name__)
return
for account in accounts:
if self._stop.is_set():
@@ -314,9 +315,10 @@ def _watch_directory(self, watch_key: str) -> None:
if not self._stop.is_set():
logger.error("[sns-autosync] 系统文件监听意外结束")
self._mark_watcher_failed(watch_key, "sns_file_watch_unavailable")
- except Exception:
+ except Exception as exc:
if not self._stop.is_set():
logger.exception("[sns-autosync] 系统文件监听失败")
+ logger.error("[sns.incremental-sync] status=error phase=file-watch error_type=%s", type(exc).__name__)
self._mark_watcher_failed(watch_key, "sns_file_watch_unavailable")
def _mark_watcher_failed(self, watch_key: str, code: str) -> None:
@@ -365,13 +367,17 @@ def _schedule_sync(self, account: str, *, reason: str) -> None:
try:
worker.start()
- except Exception:
+ except Exception as exc:
with self._mu:
state = self._states.get(account)
if state is not None and state.worker is worker:
state.sync_running = False
state.worker = None
logger.exception("[sns-autosync] 启动同步线程失败 account=%s", account)
+ logger.error(
+ "[sns.incremental-sync] status=error phase=worker-start error_type=%s",
+ type(exc).__name__,
+ )
self._publish_error(
account,
source_revision=revision,
@@ -386,6 +392,12 @@ def _sync_account_runner(self, account: str, reason: str, revision: int) -> None
if reason == "startup":
self._refresh_native_moments_once(account)
while not self._stop.is_set():
+ sync_id = uuid.uuid4().hex
+ started = time.monotonic()
+ logger.info(
+ "[sns.incremental-sync] status=running request_id=%s phase=scanning",
+ sync_id,
+ )
result, superseded = self._sync_with_bounded_retries(account, revision)
# WCDB 读取可能刷新共享内存文件；短暂忽略纯 -shm 事件，防止读取自身形成事件环。
with self._mu:
@@ -397,6 +409,15 @@ def _sync_account_runner(self, account: str, reason: str, revision: int) -> None
)
if not superseded and not self._stop.is_set():
self._publish_sync_result(account, reason, revision, result)
+ status = str((result or {}).get("status") or "error").strip().lower()
+ logger.info(
+ "[sns.incremental-sync] status=%s request_id=%s phase=finalizing scanned=%s changed=%s elapsed_ms=%s",
+ status,
+ sync_id,
+ int((result or {}).get("scanned") or 0),
+ int((result or {}).get("changed") or (result or {}).get("upserted") or 0),
+ int((time.monotonic() - started) * 1000),
+ )
with self._mu:
state = self._states.get(account)
@@ -463,12 +484,17 @@ def _refresh_native_moments_once(self, account: str) -> None:
)
client.refresh_wechat_moments(context.name, context.account_dir)
logger.info("[sns-autosync] native refresh 调用完成 account=%s", account)
+ logger.info("[sns.incremental-sync] status=done phase=native-refresh")
except Exception as exc:
logger.warning(
"[sns-autosync] native refresh 调用失败 account=%s error=%s；保留手动刷新",
account,
exc,
)
+ logger.warning(
+ "[sns.incremental-sync] status=error phase=native-refresh error_type=%s",
+ type(exc).__name__,
+ )
def _sync_with_bounded_retries(self, account: str, revision: int) -> tuple[dict[str, Any], bool]:
last_result: dict[str, Any] = {"status": "error", "error": "sns_sync_failed"}
@@ -480,8 +506,13 @@ def _sync_with_bounded_retries(self, account: str, revision: int) -> tuple[dict[
if self._stop.is_set():
return {"status": "skipped", "reason": "service_stopping"}, False
last_result = dict(self._sync_account(account) or {})
- except Exception:
+ except Exception as exc:
logger.exception("[sns-autosync] 同步失败 account=%s", account)
+ logger.error(
+ "[sns.incremental-sync] status=error phase=scanning error_type=%s attempt=%s",
+ type(exc).__name__,
+ attempt + 1,
+ )
last_result = {"status": "error", "error": "sns_sync_failed"}
if not self._should_retry(last_result) or attempt >= len(self._retry_delays):
@@ -534,8 +565,12 @@ def _sync_account(self, account: str) -> dict[str, Any]:
)
except HTTPException as exc:
return {"status": "error", "error": str(exc.detail or "sns_sync_failed")}
- except Exception:
+ except Exception as exc:
logger.exception("[sns-autosync] 增量同步调用失败 account=%s", account)
+ logger.error(
+ "[sns.incremental-sync] status=error phase=scanning error_type=%s",
+ type(exc).__name__,
+ )
return {"status": "error", "error": "sns_sync_failed"}
def subscribe(
@@ -661,5 +696,9 @@ def _publish_event(self, account: str, event: dict[str, Any]) -> None:
except Exception:
pass
+ def publish_external_event(self, account: str, event: dict[str, Any]) -> None:
+ """向当前账号的 SSE 订阅者投递外部同步事件。"""
+ self._publish_event(account, event)
+
SNS_REALTIME_AUTOSYNC = SnsRealtimeAutoSyncService()
diff --git a/tests/test_sns_full_sync.py b/tests/test_sns_full_sync.py
new file mode 100644
index 00000000..11bd3caa
--- /dev/null
+++ b/tests/test_sns_full_sync.py
@@ -0,0 +1,364 @@
+import json
+import sqlite3
+import sys
+import threading
+import time
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest import mock
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+
+from wechat_decrypt_tool import sns_full_sync
+from wechat_decrypt_tool.routers import sns as sns_router
+
+
+class _FakeConnection:
+ def __init__(self, db_storage_dir: Path):
+ self.handle = 1
+ self.db_storage_dir = Path(db_storage_dir)
+ self.lock = threading.RLock()
+
+
+def _sqlite_query(_connection, source_path: Path, sql: str):
+ conn = sqlite3.connect(str(source_path))
+ conn.row_factory = sqlite3.Row
+ try:
+ return [dict(row) for row in conn.execute(sql).fetchall()]
+ finally:
+ conn.close()
+
+
+def _wait_job(manager, account_dir: Path, *, timeout: float = 10.0):
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ job = manager.get(account_dir)
+ if job and job.get("status") not in {"queued", "running"}:
+ return job
+ time.sleep(0.01)
+ raise AssertionError(f"朋友圈全量同步任务未在 {timeout} 秒内结束: {manager.get(account_dir)}")
+
+
+def _create_source_db(root: Path, rows, *, with_pack: bool = True, without_rowid: bool = False):
+ source_dir = root / "sns"
+ source_dir.mkdir(parents=True, exist_ok=True)
+ source_path = source_dir / "sns.db"
+ conn = sqlite3.connect(str(source_path))
+ try:
+ pack_sql = ", pack_info_buf BLOB" if with_pack else ""
+ suffix = " WITHOUT ROWID" if without_rowid else ""
+ conn.execute(
+ f"CREATE TABLE SnsTimeLine(tid INTEGER PRIMARY KEY, user_name TEXT, content TEXT{pack_sql}){suffix}"
+ )
+ if with_pack:
+ conn.executemany(
+ "INSERT INTO SnsTimeLine(tid, user_name, content, pack_info_buf) VALUES (?, ?, ?, ?)",
+ rows,
+ )
+ else:
+ conn.executemany(
+ "INSERT INTO SnsTimeLine(tid, user_name, content) VALUES (?, ?, ?)",
+ rows,
+ )
+ conn.commit()
+ finally:
+ conn.close()
+ return source_path
+
+
+class TestSnsFullSync(unittest.TestCase):
+ def _run_with_source(self, manager, account_dir: Path, source_root: Path, *, events=None):
+ connection = _FakeConnection(source_root)
+ event_list = events if events is not None else []
+ with (
+ mock.patch.object(sns_full_sync.WCDB_REALTIME, "ensure_connected", return_value=connection),
+ mock.patch.object(manager, "_query", side_effect=_sqlite_query),
+ mock.patch.object(
+ sns_full_sync.SNS_REALTIME_AUTOSYNC,
+ "publish_external_event",
+ side_effect=lambda _account, event: event_list.append(event),
+ ),
+ ):
+ started, reused = manager.start(account_dir)
+ self.assertFalse(reused)
+ self.assertTrue(started.get("syncId"))
+ return _wait_job(manager, account_dir), event_list
+
+ def test_full_sync_reads_more_than_2000_rows_and_second_run_is_unchanged(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_dir = root / "decrypted" / "account-a"
+ account_dir.mkdir(parents=True)
+ source_root = root / "source-a"
+ rows = []
+ for tid in range(1, 2206):
+ username = "friend-main" if tid <= 1600 else f"friend-{tid % 7}"
+ rows.append((tid, username, f"1{tid}", None))
+ for tid in range(-5, 0):
+ rows.append((tid, "friend-negative", f"1{tid}", None))
+ _create_source_db(source_root, rows, with_pack=True)
+
+ manager = sns_full_sync.SnsFullSyncManager()
+ events = []
+ first, events = self._run_with_source(manager, account_dir, source_root, events=events)
+
+ self.assertEqual(first["status"], "done")
+ self.assertEqual(first["progress"]["sourceRowsTotal"], len(rows))
+ self.assertEqual(first["progress"]["sourceRowsScanned"], len(rows))
+ self.assertEqual(first["progress"]["prepared"], len(rows))
+ self.assertEqual(first["progress"]["changed"], len(rows))
+ self.assertEqual(first["progress"]["percent"], 100)
+ self.assertGreater(first["progress"]["batchesCompleted"], 10)
+
+ conn = sqlite3.connect(str(account_dir / "sns.db"))
+ try:
+ count = conn.execute("SELECT COUNT(*) FROM SnsTimeLine").fetchone()[0]
+ main_count = conn.execute(
+ "SELECT COUNT(*) FROM SnsTimeLine WHERE user_name = ?",
+ ("friend-main",),
+ ).fetchone()[0]
+ finally:
+ conn.close()
+ self.assertEqual(count, len(rows))
+ self.assertGreater(main_count, 200)
+
+ progress_events = [event["job"] for event in events if event.get("type") == "full_sync_progress"]
+ changed_counts = [job["progress"]["changed"] for job in progress_events]
+ percents = [job["progress"]["percent"] for job in progress_events]
+ self.assertEqual(changed_counts, sorted(changed_counts))
+ self.assertEqual(percents, sorted(percents))
+ self.assertLessEqual(max(percents), 99)
+ self.assertEqual(events[-1].get("type"), "full_sync_done")
+ self.assertEqual(events[-1]["job"]["progress"]["percent"], 100)
+ self.assertTrue(events[-1]["job"].get("snapshotVersion"))
+
+ second, _ = self._run_with_source(manager, account_dir, source_root)
+ self.assertEqual(second["status"], "done")
+ self.assertEqual(second["progress"]["changed"], 0)
+ self.assertEqual(second["progress"]["unchanged"], len(rows))
+
+ def test_full_sync_ignores_highwater_and_supports_signed_tid_old_schema(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_dir = root / "decrypted" / "account-b"
+ account_dir.mkdir(parents=True)
+ (account_dir / "_sns_realtime_sync_state.json").write_text(
+ json.dumps({"maxId": "999999"}),
+ encoding="utf-8",
+ )
+ source_root = root / "source-b"
+ rows = [
+ (-2, "friend-a", "1"),
+ (1, "friend-a", "7"),
+ (2, "friend-b", "damaged-but-nonempty"),
+ (3, "friend-b", "1"),
+ ]
+ _create_source_db(source_root, rows, with_pack=False, without_rowid=True)
+
+ manager = sns_full_sync.SnsFullSyncManager()
+ result, _ = self._run_with_source(manager, account_dir, source_root)
+
+ self.assertEqual(result["status"], "done")
+ self.assertEqual(result["progress"]["sourceRowsScanned"], 4)
+ self.assertEqual(result["progress"]["prepared"], 2)
+ self.assertEqual(result["progress"]["skipped"], 2)
+ conn = sqlite3.connect(str(account_dir / "sns.db"))
+ try:
+ tids = {row[0] for row in conn.execute("SELECT tid FROM SnsTimeLine")}
+ columns = {row[1] for row in conn.execute("PRAGMA table_info(SnsTimeLine)")}
+ finally:
+ conn.close()
+ self.assertEqual(tids, {-2, 3})
+ self.assertNotIn("pack_info_buf", columns)
+ state = json.loads((account_dir / "_sns_realtime_sync_state.json").read_text(encoding="utf-8"))
+ self.assertEqual(state["maxId"], str((-2) & 0xFFFFFFFFFFFFFFFF))
+
+ def test_full_sync_backfills_rows_below_existing_highwater_without_regressing_it(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_dir = root / "decrypted" / "account-low-history"
+ account_dir.mkdir(parents=True)
+ state_path = account_dir / "_sns_realtime_sync_state.json"
+ state_path.write_text(json.dumps({"maxId": "999999"}), encoding="utf-8")
+ source_root = root / "source-low-history"
+ rows = [
+ (tid, "friend-history", "1", None)
+ for tid in range(1, 351)
+ ]
+ _create_source_db(source_root, rows)
+
+ manager = sns_full_sync.SnsFullSyncManager()
+ result, _ = self._run_with_source(manager, account_dir, source_root)
+
+ self.assertEqual(result["status"], "done")
+ self.assertEqual(result["progress"]["changed"], 350)
+ conn = sqlite3.connect(str(account_dir / "sns.db"))
+ try:
+ self.assertEqual(conn.execute("SELECT COUNT(*) FROM SnsTimeLine").fetchone()[0], 350)
+ finally:
+ conn.close()
+ state = json.loads(state_path.read_text(encoding="utf-8"))
+ self.assertEqual(state["maxId"], "999999")
+
+ def test_duplicate_reuses_job_other_account_queues_and_cancel_keeps_batches(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_a = root / "decrypted" / "account-a"
+ account_b = root / "decrypted" / "account-b"
+ account_a.mkdir(parents=True)
+ account_b.mkdir(parents=True)
+ source_a = root / "source-a"
+ source_b = root / "source-b"
+ _create_source_db(
+ source_a,
+ [(tid, "friend-a", "1", None) for tid in range(1, 451)],
+ )
+ _create_source_db(
+ source_b,
+ [(tid, "friend-b", "1", None) for tid in range(1, 11)],
+ )
+ connections = {
+ str(account_a.resolve()): _FakeConnection(source_a),
+ str(account_b.resolve()): _FakeConnection(source_b),
+ }
+ first_batch_entered = threading.Event()
+ release_first_batch = threading.Event()
+ real_upsert = sns_router._upsert_sns_timeline_rows_to_decrypted_db
+ events = []
+
+ def slow_upsert(account_dir, rows, *, source):
+ result = real_upsert(account_dir, rows, source=source)
+ if Path(account_dir).resolve() == account_a.resolve() and not first_batch_entered.is_set():
+ first_batch_entered.set()
+ release_first_batch.wait(timeout=3)
+ return result
+
+ manager = sns_full_sync.SnsFullSyncManager()
+ with (
+ mock.patch.object(
+ sns_full_sync.WCDB_REALTIME,
+ "ensure_connected",
+ side_effect=lambda account_dir, timeout=15.0: connections[str(Path(account_dir).resolve())],
+ ),
+ mock.patch.object(manager, "_query", side_effect=_sqlite_query),
+ mock.patch.object(sns_router, "_upsert_sns_timeline_rows_to_decrypted_db", side_effect=slow_upsert),
+ mock.patch.object(
+ sns_full_sync.SNS_REALTIME_AUTOSYNC,
+ "publish_external_event",
+ side_effect=lambda account, event: events.append((account, event)),
+ ),
+ ):
+ first, reused = manager.start(account_a)
+ self.assertFalse(reused)
+ self.assertTrue(first_batch_entered.wait(timeout=3))
+
+ duplicate, reused = manager.start(account_a)
+ self.assertTrue(reused)
+ self.assertEqual(duplicate["syncId"], first["syncId"])
+
+ queued, reused = manager.start(account_b)
+ self.assertFalse(reused)
+ self.assertEqual(queued["status"], "queued")
+
+ current, accepted = manager.cancel(account_a, "stale-sync-id")
+ self.assertFalse(accepted)
+ self.assertEqual(current["syncId"], first["syncId"])
+ self.assertFalse(current["cancelRequested"])
+
+ cancelling, accepted = manager.cancel(account_a, first["syncId"])
+ self.assertTrue(accepted)
+ self.assertTrue(cancelling["cancelRequested"])
+ release_first_batch.set()
+
+ cancelled = _wait_job(manager, account_a)
+ completed = _wait_job(manager, account_b)
+
+ self.assertEqual(cancelled["status"], "cancelled")
+ self.assertEqual(cancelled["progress"]["batchesCompleted"], 1)
+ self.assertFalse((account_a / "_sns_realtime_sync_state.json").exists())
+ conn = sqlite3.connect(str(account_a / "sns.db"))
+ try:
+ self.assertEqual(conn.execute("SELECT COUNT(*) FROM SnsTimeLine").fetchone()[0], 200)
+ finally:
+ conn.close()
+ self.assertEqual(completed["status"], "done")
+ self.assertEqual(completed["progress"]["changed"], 10)
+ event_types = [event.get("type") for _account, event in events]
+ self.assertIn("full_sync_cancelled", event_types)
+ self.assertIn("full_sync_done", event_types)
+
+ def test_failures_use_stable_public_errors(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_dir = root / "account"
+ account_dir.mkdir()
+ manager = sns_full_sync.SnsFullSyncManager()
+ sentinel = "C:/private/path/account-secret"
+ events = []
+ with (
+ mock.patch.object(
+ sns_full_sync.WCDB_REALTIME,
+ "ensure_connected",
+ side_effect=RuntimeError(sentinel),
+ ),
+ mock.patch.object(
+ sns_full_sync.SNS_REALTIME_AUTOSYNC,
+ "publish_external_event",
+ side_effect=lambda _account, event: events.append(event),
+ ),
+ mock.patch.object(sns_full_sync.logger, "error") as log_error,
+ ):
+ manager.start(account_dir)
+ result = _wait_job(manager, account_dir)
+
+ self.assertEqual(result["status"], "error")
+ self.assertEqual(result["error"]["code"], "realtime_not_available")
+ self.assertEqual(events[-1].get("type"), "full_sync_error")
+ rendered = "\n".join(" ".join(map(str, call.args)) for call in log_error.call_args_list)
+ self.assertNotIn(sentinel, rendered)
+
+ def test_batch_write_failure_does_not_advance_highwater(self):
+ with TemporaryDirectory() as td:
+ root = Path(td)
+ account_dir = root / "account"
+ account_dir.mkdir()
+ source_root = root / "source"
+ _create_source_db(
+ source_root,
+ [(1, "friend", "1", None)],
+ )
+ manager = sns_full_sync.SnsFullSyncManager()
+ connection = _FakeConnection(source_root)
+ with (
+ mock.patch.object(sns_full_sync.WCDB_REALTIME, "ensure_connected", return_value=connection),
+ mock.patch.object(manager, "_query", side_effect=_sqlite_query),
+ mock.patch.object(
+ sns_router,
+ "_upsert_sns_timeline_rows_to_decrypted_db",
+ return_value={"success": False, "prepared": 1, "changed": 0, "unchanged": 0},
+ ),
+ mock.patch.object(sns_full_sync.SNS_REALTIME_AUTOSYNC, "publish_external_event"),
+ ):
+ manager.start(account_dir)
+ result = _wait_job(manager, account_dir)
+
+ self.assertEqual(result["status"], "error")
+ self.assertEqual(result["error"]["code"], "snapshot_write_failed")
+ self.assertFalse((account_dir / "_sns_realtime_sync_state.json").exists())
+
+ def test_routes_expose_start_status_and_exact_cancel(self):
+ methods_by_path = {
+ (getattr(route, "path", ""), tuple(sorted(getattr(route, "methods", ()) or ())))
+ for route in sns_router.router.routes
+ }
+ self.assertIn(("/api/sns/realtime/full_sync", ("POST",)), methods_by_path)
+ self.assertIn(("/api/sns/realtime/full_sync/status", ("GET",)), methods_by_path)
+ self.assertIn(("/api/sns/realtime/full_sync", ("DELETE",)), methods_by_path)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_sns_page_decrypted_source.py b/tests/test_sns_page_decrypted_source.py
index 0e4e5c51..3e37dd74 100644
--- a/tests/test_sns_page_decrypted_source.py
+++ b/tests/test_sns_page_decrypted_source.py
@@ -177,6 +177,9 @@ def test_sns_page_loads_local_snapshot_before_event_connection(self):
self.assertIn("query.set('usernames', params.usernames.join(','))", api)
self.assertIn("syncSnsRealtimeLatest,", api)
self.assertIn("getSnsSnapshotStatus,", api)
+ self.assertIn("startSnsFullSync,", api)
+ self.assertIn("getSnsFullSyncStatus,", api)
+ self.assertIn("cancelSnsFullSync,", api)
self.assertRegex(
page,
re.compile(
@@ -201,19 +204,19 @@ def test_sns_page_loads_local_snapshot_before_event_connection(self):
refresh = page.split("const refreshSnsData = async () => {", 1)[1].split(
"\n\nlet postsRequestGeneration", 1
)[0]
- self.assertIn("await syncLatestSnsWithTimeout(account, {", refresh)
- self.assertIn("maxScan: SNS_MANUAL_REFRESH_SCAN_LIMIT", refresh)
- self.assertIn("scanOffset: reconcileWindow.scanOffset", refresh)
- self.assertIn("waitForCurrent: true", refresh)
- self.assertIn("await activeReconcile", refresh)
- self.assertIn("mergeVisiblePostsWindow(reconcileWindow)", refresh)
- self.assertIn("loadSnsUsers({ preserveExisting: true })", refresh)
- self.assertIn("if (shouldMergeTimeline)", refresh)
+ self.assertIn("await api.startSnsFullSync({ account })", refresh)
+ self.assertNotIn("syncLatestSnsWithTimeout", refresh)
+ self.assertNotIn("selectedSnsUser", refresh)
+ self.assertNotIn("scanOffset", refresh)
+ self.assertNotIn("usernames", refresh)
self.assertNotIn("loadPosts({ reset: true })", refresh)
self.assertNotIn("posts.value = []", refresh)
self.assertIn('@click="refreshSnsData"', page)
- self.assertIn("refreshQueued = true", page)
- self.assertIn("syncStatus === 'ok' || syncStatus === 'noop'", page)
+ self.assertIn("const cancelSnsFullSync = async () =>", page)
+ self.assertIn("await api.cancelSnsFullSync({ account, sync_id: syncId })", page)
+ self.assertIn("await restoreSnsFullSyncStatus(String(v || ''))", page)
+ self.assertIn("source.addEventListener('full_sync_progress', onSnsFullSyncEvent)", page)
+ self.assertIn("queueSnsFullSyncMerge(job, { final })", page)
self.assertIn("实时同步失败，当前显示本地快照", page)
self.assertIn("const describeSnsSyncFailure = (failure) =>", page)
self.assertIn("实时同步响应超时，后台任务仍可能完成", page)