From 4908a705816366379037b45689c72700a017dd3c Mon Sep 17 00:00:00 2001 From: xiaosheng <73678111+xiaoshengbao@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:10:38 +0800 Subject: [PATCH 1/2] fix(sns): add full-history sync and privacy-safe logs --- frontend/composables/useApi.js | 25 + frontend/pages/sns.vue | 359 ++++++++--- .../tests/sns-page-initialization.test.mjs | 22 +- src/wechat_decrypt_tool/api.py | 7 +- src/wechat_decrypt_tool/logging_config.py | 4 + src/wechat_decrypt_tool/request_logging.py | 29 +- src/wechat_decrypt_tool/routers/sns.py | 167 ++++-- src/wechat_decrypt_tool/sns_export_service.py | 18 +- src/wechat_decrypt_tool/sns_full_sync.py | 557 ++++++++++++++++++ src/wechat_decrypt_tool/sns_media.py | 32 +- .../sns_realtime_autosync.py | 66 ++- tests/test_request_log_redaction.py | 34 ++ tests/test_sns_full_sync.py | 364 ++++++++++++ tests/test_sns_log_privacy.py | 174 ++++++ tests/test_sns_media.py | 6 +- tests/test_sns_media_route_weflow_default.py | 24 +- tests/test_sns_page_decrypted_source.py | 23 +- 17 files changed, 1685 insertions(+), 226 deletions(-) create mode 100644 src/wechat_decrypt_tool/sns_full_sync.py create mode 100644 tests/test_sns_full_sync.py create mode 100644 tests/test_sns_log_privacy.py 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 18bddbc9..3cb4c43f 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('') @@ -2013,12 +2050,6 @@ const saveSnsExportToSelectedFolder = async (options = {}) => { } const response = await fetch(getSnsExportDownloadUrl(exportId)) if (!response.ok) { - await reportServerErrorFromResponse(response, { - method: 'GET', - requestUrl: getSnsExportDownloadUrl(exportId), - message: `\u4e0b\u8f7d\u5bfc\u51fa\u6587\u4ef6\u5931\u8d25\uff08${response.status}\uff09`, - source: 'sns.exportDownload' - }) throw new Error(`\u4e0b\u8f7d\u5bfc\u51fa\u6587\u4ef6\u5931\u8d25\uff08${response.status}\uff09`) } exportSaveBytesTotal.value = asNumber(response.headers.get('Content-Length')) @@ -2261,14 +2292,8 @@ const loadSelfInfo = async () => { const unchanged = Object.keys(resp).every((key) => resp[key] === selfInfo.value?.[key]) if (!unchanged) selfInfo.value = resp } - } catch (e) { - await reportServerErrorFromError(e, { - method: 'GET', - requestUrl, - source: 'sns.loadSelfInfo', - apiBase, - }) - console.error('获取个人信息失败', e) + } catch { + console.error('[sns.self-info] status=error phase=load') } } @@ -2302,7 +2327,7 @@ const loadSnsUsers = async ({ preserveExisting = false } = {}) => { for (const item of nextByUsername.values()) merged.push(item) snsUsers.value = merged } catch (e) { - console.error('加载朋友圈联系人失败', e) + console.error('[sns.users] status=error phase=load') // 后台刷新失败时保留已显示的联系人，避免侧边栏闪空。 } } @@ -2578,7 +2603,7 @@ const onCopyPostTextClick = async () => { const ok = await copyTextToClipboard(text) if (!ok) showErrorAlert('复制失败：无法写入剪贴板') } catch (e) { - console.error('复制失败:', e) + console.error('[sns.copy] status=error phase=clipboard-write') showErrorAlert('复制失败') } finally { closeContextMenu() @@ -2596,7 +2621,7 @@ const onCopyPostJsonClick = async () => { const ok = await copyTextToClipboard(json) if (!ok) showErrorAlert('复制失败：无法写入剪贴板') } catch (e) { - console.error('复制失败:', e) + console.error('[sns.copy] status=error phase=clipboard-write') showErrorAlert('复制失败') } finally { closeContextMenu() @@ -3369,11 +3394,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 @@ -3384,6 +3410,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 @@ -3438,7 +3468,7 @@ const beginSnsRealtimeSync = ( const syncLatestSnsWithTimeout = async ( account, { - maxScan = SNS_MANUAL_REFRESH_SCAN_LIMIT, + maxScan = SNS_INCREMENTAL_DEFAULT_SCAN_LIMIT, scanOffset = null, usernames = [], waitForCurrent = false @@ -3515,80 +3545,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 {} - } - 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) - ) + 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 }) } - 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) => { @@ -3887,13 +3895,152 @@ const mergeVisiblePostsWindow = async (windowRange = getSnsVisibleReconcileWindo scheduleSnsVisibleWindowUpdate() return true } catch (e) { - console.warn('合并朋友圈浮动窗口失败', e) + console.warn('[sns.timeline] status=error phase=viewport-merge') return false } } 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() @@ -4051,7 +4198,7 @@ const queueSnsRealtimeReconcile = (eventPayload) => { if (account === String(selectedAccount.value || '').trim()) { syncWarning.value = describeSnsSyncFailure(e) } - console.warn('朋友圈事件对账失败，继续使用本地快照', e) + console.warn('[sns.incremental-sync] status=error phase=viewport-reconcile') } } return changed @@ -4074,6 +4221,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 @@ -4121,6 +4269,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() @@ -4139,6 +4305,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() @@ -4155,8 +4325,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) { @@ -4181,6 +4356,7 @@ watch( resetSnsMediaErrors() if (previewCtx.value) closeImagePreview() await loadLocalSnsData() + await restoreSnsFullSyncStatus(String(v || '')) // 首屏就绪后建立事件连接；后端启动同步或重连差异由 ready 事件补齐。 connectSnsEventStream() } @@ -4257,6 +4433,7 @@ const runPassiveSnsRefresh = async () => { if (!String(selectedAccount.value || '').trim()) return // 窗口重新可见时只核对一次本地版本，然后恢复 SSE。 await reconcileSnsSnapshotOnce() + await restoreSnsFullSyncStatus(String(selectedAccount.value || '')) connectSnsEventStream() } @@ -4299,7 +4476,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..a475f279 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('\n\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..dceb4a48 100644 --- a/src/wechat_decrypt_tool/api.py +++ b/src/wechat_decrypt_tool/api.py @@ -281,8 +281,11 @@ async def _startup_background_jobs() -> None: logger.exception("Failed to start realtime autosync service") try: SNS_REALTIME_AUTOSYNC.start() - except Exception: - logger.exception("Failed to start SNS realtime autosync service") + except Exception as exc: + 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/logging_config.py b/src/wechat_decrypt_tool/logging_config.py index 9142d7d0..bd90a73d 100644 --- a/src/wechat_decrypt_tool/logging_config.py +++ b/src/wechat_decrypt_tool/logging_config.py @@ -274,6 +274,10 @@ def setup_logging(self, log_level: str = "INFO"): if console_handler is not None: fastapi_logger.addHandler(console_handler) fastapi_logger.setLevel(level) + + # httpx 的 INFO 请求日志会展开完整远程 URL；朋友圈媒体 URL 可能含 token 和用户标识。 + logging.getLogger("httpx").setLevel(logging.WARNING) + logging.getLogger("httpcore").setLevel(logging.WARNING) # 记录初始化信息 logger = logging.getLogger(__name__) diff --git a/src/wechat_decrypt_tool/request_logging.py b/src/wechat_decrypt_tool/request_logging.py index 70fe16c6..348e5fef 100644 --- a/src/wechat_decrypt_tool/request_logging.py +++ b/src/wechat_decrypt_tool/request_logging.py @@ -94,10 +94,14 @@ def redact_sensitive_log_data(value: Any, *, _depth: int = 0) -> Any: def redact_sensitive_query_text(value: Any) -> str: """Redact sensitive query values without decoding or rewriting unrelated values.""" text = str(value or "") - if "?" not in text: + prefix, separator, query = text.partition("?") + normalized_path = prefix.lower().rstrip("/") + if normalized_path == "/sns" or "/sns/" in f"{normalized_path}/": + # 朋友圈查询参数可能包含账号、联系人、动态/媒体 ID、URL、路径、时间和尺寸； + # 路径段也可能包含导出或文件 ID，因此统一压缩为固定命名空间。 + return "/api/sns" if "/api/sns" in normalized_path else "/sns" + if not separator: return text - - prefix, query = text.split("?", 1) redacted_parts: list[str] = [] for part in query.split("&"): raw_key, separator, raw_value = part.partition("=") @@ -232,20 +236,31 @@ def _extract_response_detail_from_body(response: Response, body: bytes) -> str: async def log_server_errors_middleware(logger, request: Request, call_next): method = str(request.method or "").upper() or "GET" path = str(request.url.path or "").strip() or "/" + normalized_path = path.lower().rstrip("/") + is_sns_request = normalized_path == "/sns" or "/sns/" in f"{normalized_path}/" + safe_path = "/api/sns" if is_sns_request else path try: response = await call_next(request) except Exception as exc: - logger.exception("[server-exception] method=%s path=%s error=%s", method, path, exc) + if is_sns_request: + logger.error( + "[server-exception] method=%s path=%s code=sns_request_failed error_type=%s", + method, + safe_path, + type(exc).__name__, + ) + else: + logger.exception("[server-exception] method=%s path=%s error=%s", method, path, exc) raise status = int(getattr(response, "status_code", 0) or 0) if status >= 500: response, body = await _buffer_response_body(response) detail = _extract_response_detail_from_body(response, body) or _extract_response_detail(response) - if detail: - logger.error("[server-5xx] status=%s method=%s path=%s detail=%s", status, method, path, detail) + if detail and not is_sns_request: + logger.error("[server-5xx] status=%s method=%s path=%s detail=%s", status, method, safe_path, detail) else: - logger.error("[server-5xx] status=%s method=%s path=%s", status, method, path) + logger.error("[server-5xx] status=%s method=%s path=%s", status, method, safe_path) return response diff --git a/src/wechat_decrypt_tool/routers/sns.py b/src/wechat_decrypt_tool/routers/sns.py index b7af8e09..4aaa0600 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, @@ -527,11 +529,9 @@ def _pack_blob(value: Any) -> Optional[bytes]: "unchanged": unchanged, } except Exception as e: - raw_error_text = f"{type(e).__name__}: {e}" - error_text = raw_error_text.encode("ascii", errors="backslashreplace").decode("ascii") + error_text = type(e).__name__ logger.warning( - "[sns] decrypted sns.db upsert failed source=%s prepared=%s err=%s", - source, + "[sns.incremental-sync] status=error phase=writing prepared=%s error_type=%s", len(rows), error_text, ) @@ -1571,7 +1571,7 @@ def _get_sns_covers( # 利用 exec_query 强行查 rows = _wcdb_exec_query(conn.handle, kind="media", path=str(sns_db_path), sql=cover_sql) or [] except Exception as e: - logger.warning("[sns] WCDB cover fetch failed: %s", e) + logger.warning("[sns.cover] status=error phase=source-read error_type=%s", type(e).__name__) # 2) Fallback to local decrypted snapshot sns.db. if not rows: @@ -1585,7 +1585,7 @@ def _get_sns_covers( conn_sq.close() rows = [{"tid": r["tid"], "content": r["content"]} for r in (rows_sq or [])] except Exception as e: - logger.warning("[sns] SQLite cover fetch failed: %s", e) + logger.warning("[sns.cover] status=error phase=snapshot-read error_type=%s", type(e).__name__) out: list[dict[str, Any]] = [] seen: set[str] = set() @@ -1643,7 +1643,7 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): if requested_source == "auto" and account_prefers_decrypted_snapshot(account_dir): requested_source = "decrypted" - logger.info(f"[self_info] 开始获取账号信息, 预设 wxid: {wxid}") + logger.info("[sns.self-info] status=running phase=lookup") nickname = wxid result_source = "wxid_dir" @@ -1658,10 +1658,10 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): if names_map and names_map.get(wxid): nickname = names_map[wxid] result_source = "wcdb_realtime" - logger.info(f"[self_info] 从 WCDB 实时连接获取成功: {nickname}") + logger.info("[sns.self-info] status=done phase=wcdb") return {"wxid": wxid, "nickname": nickname, "source": result_source} except Exception as e: - logger.debug(f"[self_info] WCDB 路径跳过或失败: {e}") + logger.debug("[sns.self-info] status=skipped phase=wcdb error_type=%s", type(e).__name__) contact_db_path = account_dir / "contact.db" if contact_db_path.exists(): @@ -1673,7 +1673,7 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): cursor = conn.execute("PRAGMA table_info(contact)") cols = {row["name"].lower() for row in cursor.fetchall()} - logger.debug(f"[self_info] contact 表现有字段: {cols}") + logger.debug("[sns.self-info] status=running phase=snapshot-schema") target_nick_col = "nick_name" if "nick_name" in cols else ("nickname" if "nickname" in cols else None) @@ -1697,18 +1697,18 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): nickname = raw_alias result_source = "contact_db_alias" - logger.info(f"[self_info] 从数据库提取成功: {nickname} (src: {result_source})") + logger.info("[sns.self-info] status=done phase=snapshot") else: - logger.warning("[self_info] contact 表中找不到任何昵称相关字段") + logger.warning("[sns.self-info] status=skipped phase=snapshot-schema code=nickname_column_missing") except sqlite3.OperationalError as e: - logger.error(f"[self_info] 数据库繁忙或锁定: {e}") + logger.error("[sns.self-info] status=error phase=snapshot-read error_type=%s", type(e).__name__) except Exception as e: - logger.exception(f"[self_info] 查询异常: {e}") + logger.error("[sns.self-info] status=error phase=snapshot-read error_type=%s", type(e).__name__) finally: if conn: conn.close() else: - logger.warning(f"[self_info] 找不到 contact.db: {contact_db_path}") + logger.warning("[sns.self-info] status=skipped phase=snapshot-read code=contact_db_missing") return { "wxid": wxid, @@ -1796,6 +1796,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 +1831,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 +1883,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 +1910,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 +1927,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 +2164,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(( @@ -2117,8 +2184,8 @@ def _sync_response( )) if not snapshot_complete: logger.warning( - "[sns-sync] snapshot write incomplete account=%s scanned=%s prepared=%s changed=%s unchanged=%s missing_required=%s", - account_dir.name, + "[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, @@ -2138,10 +2205,9 @@ def _sync_response( if backlog_truncated: logger.warning( - "[sns-sync] backlog exceeds scan cap account=%s scanned=%s last_max_id=%s", - account_dir.name, + "[sns.incremental-sync] status=skipped request_id=%s phase=scanning code=scan_cap_reached scanned=%s", + sync_request_id, len(rows), - last_max_id_u, ) return _sync_response({ "status": "skipped", @@ -2160,7 +2226,10 @@ def _sync_response( st2["maxId"] = str(committed_max_id_u) 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", @@ -2258,8 +2327,8 @@ def _list_from_decrypted_sqlite() -> dict[str, Any]: try: rows2 = conn2.execute(sql, params_with_page).fetchall() except sqlite3.OperationalError as e: - logger.warning("[sns] query failed: %s", e) - raise HTTPException(status_code=500, detail=f"sns.db query failed: {e}") + logger.warning("[sns.timeline] status=error phase=snapshot-read error_type=%s", type(e).__name__) + raise HTTPException(status_code=500, detail="sns.db query failed") finally: conn2.close() @@ -2925,10 +2994,10 @@ def _clean_name(v: Any) -> str: return wcdb_resp except WCDBRealtimeError as e: - logger.info("[sns] wcdb realtime unavailable: %s", e) + logger.info("[sns.timeline] status=skipped phase=source-read error_type=%s", type(e).__name__) fallback_reason = str(e) except Exception as e: - logger.warning("[sns] wcdb realtime failed: %s", e) + logger.warning("[sns.timeline] status=error phase=source-read error_type=%s", type(e).__name__) fallback_reason = str(e) fallback = _list_from_decrypted_sqlite() @@ -2981,7 +3050,7 @@ def worker() -> None: finally: conn.close() except Exception as exc: - logger.info("[sns] background index creation deferred: %s", exc) + logger.info("[sns.timeline] status=skipped phase=index error_type=%s", type(exc).__name__) with _SNS_INDEX_SCHEDULE_LOCK: _SNS_INDEX_SCHEDULED.discard(key) @@ -3522,35 +3591,25 @@ async def get_sns_media( media_type_i = int(media_type or 2) md5_norm = _normalize_hex32(md5) request_id = f"sns-media-{time.time_ns()}-{threading.get_ident()}" - _trace_id, trace = create_perf_trace( + _trace_id, raw_trace = create_perf_trace( logger, "sns.media", requestId=request_id, - account=str(account_dir.name), - accountDir=str(account_dir), - wxidDir=str(wxid_dir or ""), - postId=str(post_id or ""), - mediaId=str(media_id or ""), - postType=post_type_i, - mediaType=media_type_i, - createTime=int(create_time or 0), - width=int(width or 0), - height=int(height or 0), - totalSize=int(total_size or 0), - idx=max(0, int(idx or 0)), - md5=md5_norm, - variant=variant_norm, - preferRemoteOriginal=prefer_remote_original, - useCacheRequested=str(use_cache), - useCacheEffective=use_cache_flag, - tokenPresent=bool(str(token or "")), - tokenLength=len(str(token or "")), - tokenHash=_sns_media_value_hash(token), - keyPresent=bool(str(key or "")), - keyLength=len(str(key or "")), - keyHash=_sns_media_value_hash(key), - **_sns_media_url_trace_fields(url), ) + + def trace(phase: str, **fields: Any) -> None: + """严格筛选媒体诊断字段，避免后续调用误把用户数据写入日志。""" + safe_fields: dict[str, Any] = {} + for field_name in ("result", "statusCode", "elapsedMs"): + if field_name in fields: + safe_fields[field_name] = fields[field_name] + error_type = str(fields.get("errorType") or "") + if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,79}", error_type): + safe_fields["errorType"] = error_type + safe_phase = str(phase or "unknown") + if not re.fullmatch(r"[a-z0-9:_-]{1,80}", safe_phase, flags=re.I): + safe_phase = "unknown" + raw_trace(safe_phase, **safe_fields) trace("request:start") # 点击预览需要高清原图：本地 sns 缓存有时只命中缩略图，所以 full/original 请求先按 @@ -3799,7 +3858,7 @@ async def proxy_article_thumb(url: str): ) except Exception as e: - logger.warning(f"[sns] 提取公众号封面失败 url={u[:50]}... : {e}") + logger.warning("[sns.article-thumb] status=error phase=fetch error_type=%s", type(e).__name__) raise HTTPException(status_code=404, detail="无法获取文章封面") diff --git a/src/wechat_decrypt_tool/sns_export_service.py b/src/wechat_decrypt_tool/sns_export_service.py index 16a21fc0..996f0cc3 100644 --- a/src/wechat_decrypt_tool/sns_export_service.py +++ b/src/wechat_decrypt_tool/sns_export_service.py @@ -247,7 +247,10 @@ async def worker(http_client: httpx.AsyncClient) -> None: except Exception as exc: result.failed += 1 result.missing.append(task_id) - logger.info("sns media prefetch failed: kind=%s url=%s error=%s", task.kind, task.url, exc) + logger.info( + "[sns.export] status=error phase=media-prefetch error_type=%s", + type(exc).__name__, + ) finally: completed += 1 if on_progress is not None: @@ -1631,7 +1634,12 @@ def _run_job_safe(self, job: ExportJob, account_dir: Path) -> None: pass self.commit_staged_files(job.export_id) except Exception as e: - logger.exception("sns export job failed: %s: %s", job.export_id, e) + logger.error( + "[sns.export] status=error export_id=%s phase=%s error_type=%s", + job.export_id, + str(job.progress.phase or "unknown"), + type(e).__name__, + ) with self._lock: job.status = "error" job.error = str(e) @@ -1696,7 +1704,11 @@ def set_phase(name: str) -> None: if sync_status not in {"ok", "noop"}: job.warning = "实时同步未完成，已继续导出本地历史快照。" except Exception as exc: # 导出必须可在无 native broker 的环境中降级运行 - logger.warning("sns realtime sync before export failed: export=%s error=%s", job.export_id, exc) + logger.warning( + "[sns.export] status=error export_id=%s phase=syncing error_type=%s", + job.export_id, + type(exc).__name__, + ) job.warning = "实时同步失败，已继续导出本地历史快照。" job.freshness = { "status": "warning", 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_media.py b/src/wechat_decrypt_tool/sns_media.py index 0de619f3..7605c8c6 100644 --- a/src/wechat_decrypt_tool/sns_media.py +++ b/src/wechat_decrypt_tool/sns_media.py @@ -91,33 +91,20 @@ def _sns_remote_diagnostic_log( error: Optional[BaseException] = None, **fields: object, ) -> None: - raw_url = str(url or "").strip() - try: - host = str(urlparse(raw_url).hostname or "").strip().lower() - except Exception: - host = "" - - stable_url = normalize_sns_cache_url(raw_url) + # 朋友圈日志只保留诊断阶段和受控结果，不记录 URL、Host、密钥摘要或响应尺寸。 payload: dict[str, object] = { "diagnosticId": str(diagnostic_id or ""), "event": str(event or ""), - "urlHost": host, - "urlIdentity": ( - hashlib.sha256(stable_url.encode("utf-8", errors="ignore")).hexdigest()[:16] - if stable_url - else "" - ), - **fields, } + for field_name in ("stage", "result", "statusCode", "elapsedMs", "attempt"): + if field_name in fields: + payload[field_name] = fields[field_name] if error is not None: - error_text = str(error).strip() or repr(error) - for sensitive in (raw_url, str(key or ""), str(token or "")): - if sensitive: - error_text = error_text.replace(sensitive, "") - error_text = re.sub(r"https?://[^\s\"']+", "", error_text, flags=re.I) - payload["errorType"] = type(error).__name__ - payload["errorText"] = error_text[:500] + error_type = type(error).__name__ + payload["errorType"] = ( + error_type if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,79}", error_type) else "Exception" + ) response = getattr(error, "response", None) status_code = getattr(response, "status_code", None) @@ -125,7 +112,7 @@ def _sns_remote_diagnostic_log( try: payload["statusCode"] = int(status_code) except Exception: - payload["statusCode"] = str(status_code) + payload["statusCode"] = 0 logger.info( "[sns_media] %s", @@ -1082,4 +1069,3 @@ async def try_fetch_and_decrypt_sns_image_remote( x_enc=str(x_enc or "").strip(), cache_path=cache_path, ) - diff --git a/src/wechat_decrypt_tool/sns_realtime_autosync.py b/src/wechat_decrypt_tool/sns_realtime_autosync.py index 9648f953..f232f7cd 100644 --- a/src/wechat_decrypt_tool/sns_realtime_autosync.py +++ b/src/wechat_decrypt_tool/sns_realtime_autosync.py @@ -176,8 +176,8 @@ def _bootstrap_accounts(self) -> None: """启动时只枚举一次账号；后续账号由 SSE 连接动态注册。""" try: accounts = list(_list_decrypted_accounts() or []) - except Exception: - logger.exception("[sns-autosync] 初始账号枚举失败") + except Exception as exc: + 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 +314,9 @@ 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: @@ -358,20 +358,23 @@ def _schedule_sync(self, account: str, *, reason: str) -> None: worker = threading.Thread( target=self._sync_account_runner, args=(account, reason, revision), - name=f"sns-event-sync-{account}", + name="sns-incremental-sync", daemon=True, ) state.worker = worker 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 +389,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 +406,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) @@ -462,12 +480,11 @@ def _refresh_native_moments_once(self, account: str) -> None: client, NativeCoreFeature.WECHAT_MOMENTS_REFRESH ) 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, + "[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]: @@ -480,8 +497,12 @@ 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: - logger.exception("[sns-autosync] 同步失败 account=%s", account) + except Exception as exc: + 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 +555,11 @@ 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: - logger.exception("[sns-autosync] 增量同步调用失败 account=%s", account) + except Exception as exc: + logger.error( + "[sns.incremental-sync] status=error phase=scanning error_type=%s", + type(exc).__name__, + ) return {"status": "error", "error": "sns_sync_failed"} def subscribe( @@ -599,14 +623,6 @@ def _publish_sync_result(self, account: str, reason: str, revision: int, result: "timestamp": int(time.time() * 1000), } self._publish_event(account, event) - logger.info( - "[sns-autosync] 事件同步完成 account=%s reason=%s revision=%s status=%s changed=%s", - account, - reason, - revision, - status, - event["changed"], - ) return raw_code = str((result or {}).get("error") or (result or {}).get("reason") or "sns_sync_failed").strip() @@ -661,5 +677,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_request_log_redaction.py b/tests/test_request_log_redaction.py index ae19880b..c486b7eb 100644 --- a/tests/test_request_log_redaction.py +++ b/tests/test_request_log_redaction.py @@ -118,6 +118,40 @@ def test_uvicorn_access_filter_redacts_query_secrets(self): self.assertNotIn("AES_SECRET", rendered) self.assertIn("db_storage_path=C%3A%5Cdb", rendered) + def test_uvicorn_access_filter_removes_all_sns_query_values(self): + from wechat_decrypt_tool.request_logging import ( + SensitiveQueryLogFilter, + redact_sensitive_query_text, + ) + + record = logging.LogRecord( + "uvicorn.access", + logging.INFO, + __file__, + 1, + '%s - "%s %s HTTP/%s" %d', + ( + "127.0.0.1:1234", + "GET", + "/api/sns/media?account=ACCOUNT_SENTINEL&post_id=POST_SENTINEL&url=https%3A%2F%2Fprivate.example", + "1.1", + 200, + ), + None, + ) + + self.assertTrue(SensitiveQueryLogFilter().filter(record)) + rendered = record.getMessage() + self.assertIn("/api/sns", rendered) + self.assertNotIn("/media", rendered) + self.assertNotIn("ACCOUNT_SENTINEL", rendered) + self.assertNotIn("POST_SENTINEL", rendered) + self.assertNotIn("private.example", rendered) + self.assertEqual( + redact_sensitive_query_text("/api/sns/exports/EXPORT_SENTINEL/files/private-file.jpg"), + "/api/sns", + ) + if __name__ == "__main__": unittest.main() 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_log_privacy.py b/tests/test_sns_log_privacy.py new file mode 100644 index 00000000..89914819 --- /dev/null +++ b/tests/test_sns_log_privacy.py @@ -0,0 +1,174 @@ +import asyncio +import hashlib +import sys +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 fastapi import HTTPException + +from wechat_decrypt_tool import sns_export_service, sns_media +from wechat_decrypt_tool import request_logging +from wechat_decrypt_tool.routers import sns as sns_router + + +def _render_calls(calls) -> str: + rendered = [] + for item in calls: + args = item.args + if not args: + continue + try: + rendered.append(str(args[0]) % tuple(args[1:])) + except Exception: + rendered.append(" ".join(str(value) for value in args)) + return "\n".join(rendered) + + +class TestSnsLogPrivacy(unittest.TestCase): + def test_sns_server_exception_log_omits_exception_text_and_traceback(self): + sentinel = "C:/private/account-sentinel/source.db" + + class _Url: + path = "/api/sns/media" + + class _Request: + method = "GET" + url = _Url() + + async def fail(_request): + raise RuntimeError(sentinel) + + fake_logger = mock.Mock() + with self.assertRaises(RuntimeError): + asyncio.run(request_logging.log_server_errors_middleware(fake_logger, _Request(), fail)) + + logs = _render_calls(fake_logger.error.call_args_list) + self.assertIn("code=sns_request_failed", logs) + self.assertIn("error_type=RuntimeError", logs) + self.assertNotIn(sentinel, logs) + fake_logger.exception.assert_not_called() + + def test_remote_media_log_omits_all_user_and_derived_sentinels(self): + sentinels = { + "account": "privacy-account-sentinel", + "nickname": "隐私昵称哨兵", + "content": "朋友圈正文哨兵", + "url": "https://private.example.test/secret/path?token=url-token-sentinel", + "path": "C:/Users/private-user/朋友圈/private-file.jpg", + "post": "post-id-sentinel", + "media": "media-id-sentinel", + "key": "key-sentinel-value", + "token": "token-sentinel-value", + } + derived = { + hashlib.sha256(value.encode("utf-8")).hexdigest() + for value in sentinels.values() + } + + with mock.patch.object(sns_media.logger, "info") as log_info: + sns_media._sns_remote_diagnostic_log( + "remote:download-error", + url=sentinels["url"], + diagnostic_id="diag-safe-1", + key=sentinels["key"], + token=sentinels["token"], + error=RuntimeError(f"failed at {sentinels['path']} {sentinels['content']}"), + account=sentinels["account"], + nickname=sentinels["nickname"], + postId=sentinels["post"], + mediaId=sentinels["media"], + candidatePath=sentinels["path"], + responseSha256=next(iter(derived)), + width=123, + height=456, + createTime=1234567890, + statusCode=503, + ) + + logs = _render_calls(log_info.call_args_list) + self.assertIn("diag-safe-1", logs) + self.assertIn('"statusCode": 503', logs) + self.assertIn('"errorType": "RuntimeError"', logs) + for value in (*sentinels.values(), *derived): + self.assertNotIn(value, logs) + for forbidden_field in ( + "urlHost", + "urlIdentity", + "errorText", + "candidatePath", + "responseSha256", + "postId", + "mediaId", + "width", + "height", + "createTime", + ): + self.assertNotIn(forbidden_field, logs) + + def test_incremental_unavailable_log_does_not_include_account_or_path(self): + with TemporaryDirectory() as td: + account_dir = Path(td) / "private-account-sentinel" + account_dir.mkdir() + with ( + mock.patch.object(sns_router, "_resolve_account_dir", return_value=account_dir), + mock.patch.object(sns_router.WCDB_REALTIME, "get_status", return_value={}), + mock.patch.object(sns_router.logger, "info") as log_info, + mock.patch.object(sns_router.logger, "error") as log_error, + ): + with self.assertRaises(HTTPException): + sns_router.sync_sns_realtime_timeline_latest(account=account_dir.name) + + logs = _render_calls(log_info.call_args_list + log_error.call_args_list) + self.assertIn("[sns.incremental-sync]", logs) + self.assertIn("realtime_not_available", logs) + self.assertNotIn(account_dir.name, logs) + self.assertNotIn(str(account_dir), logs) + + def test_export_prefetch_failure_log_omits_url_and_exception_text(self): + task = sns_export_service.SnsRemoteMediaTask( + kind="image", + url="https://private.example.test/media/url-sentinel", + key="private-key-sentinel", + token="private-token-sentinel", + ) + + async def fail_fetch(**_kwargs): + raise RuntimeError("C:/private/path-sentinel") + + async def run(account_dir: Path): + return await sns_export_service._prefetch_sns_remote_media( + account_dir=account_dir, + tasks=[task], + use_cache=False, + concurrency=1, + ) + + with TemporaryDirectory() as td: + account_dir = Path(td) / "account-sentinel" + account_dir.mkdir() + with ( + mock.patch.object( + sns_export_service, + "_try_fetch_and_decrypt_sns_image_remote", + side_effect=fail_fetch, + ), + mock.patch.object(sns_export_service.logger, "info") as log_info, + ): + result = asyncio.run(run(account_dir)) + + self.assertEqual(result.failed, 1) + logs = _render_calls(log_info.call_args_list) + self.assertIn("error_type=RuntimeError", logs) + for sentinel in (task.url, task.key, task.token, "C:/private/path-sentinel", account_dir.name): + self.assertNotIn(sentinel, logs) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sns_media.py b/tests/test_sns_media.py index 305b3a7c..716968e3 100644 --- a/tests/test_sns_media.py +++ b/tests/test_sns_media.py @@ -265,8 +265,10 @@ async def run(account_dir: Path): self.assertIn('"diagnosticId": "diag-http-400"', logs) self.assertIn('"errorType": "HTTPStatusError"', logs) self.assertIn('"statusCode": 400', logs) - self.assertIn('"urlHost": "mmsns.qpic.cn"', logs) - self.assertIn('"urlIdentity":', logs) + self.assertNotIn("mmsns.qpic.cn", logs) + self.assertNotIn('"urlHost":', logs) + self.assertNotIn('"urlIdentity":', logs) + self.assertNotIn('"errorText":', logs) self.assertNotIn(raw_url_token, logs) self.assertNotIn(request_token, logs) self.assertNotIn(request_key, logs) diff --git a/tests/test_sns_media_route_weflow_default.py b/tests/test_sns_media_route_weflow_default.py index d05f262e..1a5084af 100644 --- a/tests/test_sns_media_route_weflow_default.py +++ b/tests/test_sns_media_route_weflow_default.py @@ -201,16 +201,17 @@ def test_route_logs_redacted_identity_and_local_match_details(self): self.assertIn("sns.media request:start", logs) self.assertIn("sns.media local-key-post:probe", logs) self.assertIn("sns.media local-heuristic:probe", logs) - self.assertIn('"matchedBy": "local-heuristic"', logs) - self.assertIn('"candidatePath":', logs) - self.assertIn("matched-local.jpg", logs) - self.assertIn('"postId": "post-42"', logs) - self.assertIn('"mediaId": "media-7"', logs) - self.assertIn('"urlHost": "mmsns.qpic.cn"', logs) - self.assertIn('"urlIdentity":', logs) - self.assertIn('"tokenHash":', logs) - self.assertIn('"keyHash":', logs) - self.assertIn('"responseSha256":', logs) + self.assertIn('"result": "local-cache"', logs) + self.assertNotIn('"matchedBy":', logs) + self.assertNotIn('"candidatePath":', logs) + self.assertNotIn("matched-local.jpg", logs) + self.assertNotIn("post-42", logs) + self.assertNotIn("media-7", logs) + self.assertNotIn("mmsns.qpic.cn", logs) + self.assertNotIn('"urlIdentity":', logs) + self.assertNotIn('"tokenHash":', logs) + self.assertNotIn('"keyHash":', logs) + self.assertNotIn('"responseSha256":', logs) self.assertNotIn("super-secret-token", logs) self.assertNotIn("super-secret-key", logs) self.assertNotIn("url-secret-token", logs) @@ -338,7 +339,8 @@ def test_route_logs_exact_key_and_md5_match_strategies(self): logs = self._render_log_calls(log_info) self.assertEqual(resp.status_code, 200) - self.assertIn(f'"matchedBy": "{expected_match}"', logs) + self.assertIn(f"sns.media {expected_match}:probe", logs) + self.assertNotIn('"matchedBy":', logs) heuristic.assert_not_called() remote.assert_not_called() 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) From 55eaa1be2717fd23649e63978bd640047ed414be Mon Sep 17 00:00:00 2001 From: xiaosheng <73678111+xiaoshengbao@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:25:34 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E4=BF=9D?= =?UTF-8?q?=E7=95=99=E5=8E=9F=E6=9C=89=E6=9C=8B=E5=8F=8B=E5=9C=88=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E5=B9=B6=E6=96=B0=E5=A2=9E=E5=90=8C=E6=AD=A5=E8=AF=8A?= =?UTF-8?q?=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/wechat_decrypt_tool/api.py | 1 + src/wechat_decrypt_tool/logging_config.py | 4 - src/wechat_decrypt_tool/request_logging.py | 29 +-- src/wechat_decrypt_tool/routers/sns.py | 101 ++++++---- src/wechat_decrypt_tool/sns_export_service.py | 18 +- src/wechat_decrypt_tool/sns_media.py | 31 +++- .../sns_realtime_autosync.py | 21 ++- tests/test_request_log_redaction.py | 34 ---- tests/test_sns_log_privacy.py | 174 ------------------ tests/test_sns_media.py | 6 +- tests/test_sns_media_route_weflow_default.py | 24 ++- 11 files changed, 133 insertions(+), 310 deletions(-) delete mode 100644 tests/test_sns_log_privacy.py diff --git a/src/wechat_decrypt_tool/api.py b/src/wechat_decrypt_tool/api.py index dceb4a48..e4372a08 100644 --- a/src/wechat_decrypt_tool/api.py +++ b/src/wechat_decrypt_tool/api.py @@ -282,6 +282,7 @@ async def _startup_background_jobs() -> None: try: SNS_REALTIME_AUTOSYNC.start() 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__, diff --git a/src/wechat_decrypt_tool/logging_config.py b/src/wechat_decrypt_tool/logging_config.py index bd90a73d..9142d7d0 100644 --- a/src/wechat_decrypt_tool/logging_config.py +++ b/src/wechat_decrypt_tool/logging_config.py @@ -274,10 +274,6 @@ def setup_logging(self, log_level: str = "INFO"): if console_handler is not None: fastapi_logger.addHandler(console_handler) fastapi_logger.setLevel(level) - - # httpx 的 INFO 请求日志会展开完整远程 URL；朋友圈媒体 URL 可能含 token 和用户标识。 - logging.getLogger("httpx").setLevel(logging.WARNING) - logging.getLogger("httpcore").setLevel(logging.WARNING) # 记录初始化信息 logger = logging.getLogger(__name__) diff --git a/src/wechat_decrypt_tool/request_logging.py b/src/wechat_decrypt_tool/request_logging.py index 348e5fef..70fe16c6 100644 --- a/src/wechat_decrypt_tool/request_logging.py +++ b/src/wechat_decrypt_tool/request_logging.py @@ -94,14 +94,10 @@ def redact_sensitive_log_data(value: Any, *, _depth: int = 0) -> Any: def redact_sensitive_query_text(value: Any) -> str: """Redact sensitive query values without decoding or rewriting unrelated values.""" text = str(value or "") - prefix, separator, query = text.partition("?") - normalized_path = prefix.lower().rstrip("/") - if normalized_path == "/sns" or "/sns/" in f"{normalized_path}/": - # 朋友圈查询参数可能包含账号、联系人、动态/媒体 ID、URL、路径、时间和尺寸； - # 路径段也可能包含导出或文件 ID，因此统一压缩为固定命名空间。 - return "/api/sns" if "/api/sns" in normalized_path else "/sns" - if not separator: + if "?" not in text: return text + + prefix, query = text.split("?", 1) redacted_parts: list[str] = [] for part in query.split("&"): raw_key, separator, raw_value = part.partition("=") @@ -236,31 +232,20 @@ def _extract_response_detail_from_body(response: Response, body: bytes) -> str: async def log_server_errors_middleware(logger, request: Request, call_next): method = str(request.method or "").upper() or "GET" path = str(request.url.path or "").strip() or "/" - normalized_path = path.lower().rstrip("/") - is_sns_request = normalized_path == "/sns" or "/sns/" in f"{normalized_path}/" - safe_path = "/api/sns" if is_sns_request else path try: response = await call_next(request) except Exception as exc: - if is_sns_request: - logger.error( - "[server-exception] method=%s path=%s code=sns_request_failed error_type=%s", - method, - safe_path, - type(exc).__name__, - ) - else: - logger.exception("[server-exception] method=%s path=%s error=%s", method, path, exc) + logger.exception("[server-exception] method=%s path=%s error=%s", method, path, exc) raise status = int(getattr(response, "status_code", 0) or 0) if status >= 500: response, body = await _buffer_response_body(response) detail = _extract_response_detail_from_body(response, body) or _extract_response_detail(response) - if detail and not is_sns_request: - logger.error("[server-5xx] status=%s method=%s path=%s detail=%s", status, method, safe_path, detail) + if detail: + logger.error("[server-5xx] status=%s method=%s path=%s detail=%s", status, method, path, detail) else: - logger.error("[server-5xx] status=%s method=%s path=%s", status, method, safe_path) + logger.error("[server-5xx] status=%s method=%s path=%s", status, method, path) return response diff --git a/src/wechat_decrypt_tool/routers/sns.py b/src/wechat_decrypt_tool/routers/sns.py index 4aaa0600..9330fcf1 100644 --- a/src/wechat_decrypt_tool/routers/sns.py +++ b/src/wechat_decrypt_tool/routers/sns.py @@ -529,12 +529,19 @@ def _pack_blob(value: Any) -> Optional[bytes]: "unchanged": unchanged, } except Exception as e: - error_text = type(e).__name__ + raw_error_text = f"{type(e).__name__}: {e}" + error_text = raw_error_text.encode("ascii", errors="backslashreplace").decode("ascii") logger.warning( - "[sns.incremental-sync] status=error phase=writing prepared=%s error_type=%s", + "[sns] decrypted sns.db upsert failed source=%s prepared=%s err=%s", + source, 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: @@ -1571,7 +1578,7 @@ def _get_sns_covers( # 利用 exec_query 强行查 rows = _wcdb_exec_query(conn.handle, kind="media", path=str(sns_db_path), sql=cover_sql) or [] except Exception as e: - logger.warning("[sns.cover] status=error phase=source-read error_type=%s", type(e).__name__) + logger.warning("[sns] WCDB cover fetch failed: %s", e) # 2) Fallback to local decrypted snapshot sns.db. if not rows: @@ -1585,7 +1592,7 @@ def _get_sns_covers( conn_sq.close() rows = [{"tid": r["tid"], "content": r["content"]} for r in (rows_sq or [])] except Exception as e: - logger.warning("[sns.cover] status=error phase=snapshot-read error_type=%s", type(e).__name__) + logger.warning("[sns] SQLite cover fetch failed: %s", e) out: list[dict[str, Any]] = [] seen: set[str] = set() @@ -1643,7 +1650,7 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): if requested_source == "auto" and account_prefers_decrypted_snapshot(account_dir): requested_source = "decrypted" - logger.info("[sns.self-info] status=running phase=lookup") + logger.info(f"[self_info] 开始获取账号信息, 预设 wxid: {wxid}") nickname = wxid result_source = "wxid_dir" @@ -1658,10 +1665,10 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): if names_map and names_map.get(wxid): nickname = names_map[wxid] result_source = "wcdb_realtime" - logger.info("[sns.self-info] status=done phase=wcdb") + logger.info(f"[self_info] 从 WCDB 实时连接获取成功: {nickname}") return {"wxid": wxid, "nickname": nickname, "source": result_source} except Exception as e: - logger.debug("[sns.self-info] status=skipped phase=wcdb error_type=%s", type(e).__name__) + logger.debug(f"[self_info] WCDB 路径跳过或失败: {e}") contact_db_path = account_dir / "contact.db" if contact_db_path.exists(): @@ -1673,7 +1680,7 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): cursor = conn.execute("PRAGMA table_info(contact)") cols = {row["name"].lower() for row in cursor.fetchall()} - logger.debug("[sns.self-info] status=running phase=snapshot-schema") + logger.debug(f"[self_info] contact 表现有字段: {cols}") target_nick_col = "nick_name" if "nick_name" in cols else ("nickname" if "nickname" in cols else None) @@ -1697,18 +1704,18 @@ def api_sns_self_info(account: Optional[str] = None, source: str = "auto"): nickname = raw_alias result_source = "contact_db_alias" - logger.info("[sns.self-info] status=done phase=snapshot") + logger.info(f"[self_info] 从数据库提取成功: {nickname} (src: {result_source})") else: - logger.warning("[sns.self-info] status=skipped phase=snapshot-schema code=nickname_column_missing") + logger.warning("[self_info] contact 表中找不到任何昵称相关字段") except sqlite3.OperationalError as e: - logger.error("[sns.self-info] status=error phase=snapshot-read error_type=%s", type(e).__name__) + logger.error(f"[self_info] 数据库繁忙或锁定: {e}") except Exception as e: - logger.error("[sns.self-info] status=error phase=snapshot-read error_type=%s", type(e).__name__) + logger.exception(f"[self_info] 查询异常: {e}") finally: if conn: conn.close() else: - logger.warning("[sns.self-info] status=skipped phase=snapshot-read code=contact_db_missing") + logger.warning(f"[self_info] 找不到 contact.db: {contact_db_path}") return { "wxid": wxid, @@ -2183,6 +2190,15 @@ def _sync_response( not missing_required_tids, )) if not snapshot_complete: + logger.warning( + "[sns-sync] snapshot write incomplete account=%s scanned=%s prepared=%s changed=%s unchanged=%s missing_required=%s", + account_dir.name, + len(rows), + prepared_count, + changed_count, + 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, @@ -2204,6 +2220,12 @@ def _sync_response( }, prepared=prepared_count, changed=changed_count, unchanged=unchanged_count) if backlog_truncated: + logger.warning( + "[sns-sync] backlog exceeds scan cap account=%s scanned=%s last_max_id=%s", + account_dir.name, + 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, @@ -2226,6 +2248,7 @@ def _sync_response( st2["maxId"] = str(committed_max_id_u) 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, @@ -2327,8 +2350,8 @@ def _list_from_decrypted_sqlite() -> dict[str, Any]: try: rows2 = conn2.execute(sql, params_with_page).fetchall() except sqlite3.OperationalError as e: - logger.warning("[sns.timeline] status=error phase=snapshot-read error_type=%s", type(e).__name__) - raise HTTPException(status_code=500, detail="sns.db query failed") + logger.warning("[sns] query failed: %s", e) + raise HTTPException(status_code=500, detail=f"sns.db query failed: {e}") finally: conn2.close() @@ -2994,10 +3017,10 @@ def _clean_name(v: Any) -> str: return wcdb_resp except WCDBRealtimeError as e: - logger.info("[sns.timeline] status=skipped phase=source-read error_type=%s", type(e).__name__) + logger.info("[sns] wcdb realtime unavailable: %s", e) fallback_reason = str(e) except Exception as e: - logger.warning("[sns.timeline] status=error phase=source-read error_type=%s", type(e).__name__) + logger.warning("[sns] wcdb realtime failed: %s", e) fallback_reason = str(e) fallback = _list_from_decrypted_sqlite() @@ -3050,7 +3073,7 @@ def worker() -> None: finally: conn.close() except Exception as exc: - logger.info("[sns.timeline] status=skipped phase=index error_type=%s", type(exc).__name__) + logger.info("[sns] background index creation deferred: %s", exc) with _SNS_INDEX_SCHEDULE_LOCK: _SNS_INDEX_SCHEDULED.discard(key) @@ -3591,25 +3614,35 @@ async def get_sns_media( media_type_i = int(media_type or 2) md5_norm = _normalize_hex32(md5) request_id = f"sns-media-{time.time_ns()}-{threading.get_ident()}" - _trace_id, raw_trace = create_perf_trace( + _trace_id, trace = create_perf_trace( logger, "sns.media", requestId=request_id, + account=str(account_dir.name), + accountDir=str(account_dir), + wxidDir=str(wxid_dir or ""), + postId=str(post_id or ""), + mediaId=str(media_id or ""), + postType=post_type_i, + mediaType=media_type_i, + createTime=int(create_time or 0), + width=int(width or 0), + height=int(height or 0), + totalSize=int(total_size or 0), + idx=max(0, int(idx or 0)), + md5=md5_norm, + variant=variant_norm, + preferRemoteOriginal=prefer_remote_original, + useCacheRequested=str(use_cache), + useCacheEffective=use_cache_flag, + tokenPresent=bool(str(token or "")), + tokenLength=len(str(token or "")), + tokenHash=_sns_media_value_hash(token), + keyPresent=bool(str(key or "")), + keyLength=len(str(key or "")), + keyHash=_sns_media_value_hash(key), + **_sns_media_url_trace_fields(url), ) - - def trace(phase: str, **fields: Any) -> None: - """严格筛选媒体诊断字段，避免后续调用误把用户数据写入日志。""" - safe_fields: dict[str, Any] = {} - for field_name in ("result", "statusCode", "elapsedMs"): - if field_name in fields: - safe_fields[field_name] = fields[field_name] - error_type = str(fields.get("errorType") or "") - if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,79}", error_type): - safe_fields["errorType"] = error_type - safe_phase = str(phase or "unknown") - if not re.fullmatch(r"[a-z0-9:_-]{1,80}", safe_phase, flags=re.I): - safe_phase = "unknown" - raw_trace(safe_phase, **safe_fields) trace("request:start") # 点击预览需要高清原图：本地 sns 缓存有时只命中缩略图，所以 full/original 请求先按 @@ -3858,7 +3891,7 @@ async def proxy_article_thumb(url: str): ) except Exception as e: - logger.warning("[sns.article-thumb] status=error phase=fetch error_type=%s", type(e).__name__) + logger.warning(f"[sns] 提取公众号封面失败 url={u[:50]}... : {e}") raise HTTPException(status_code=404, detail="无法获取文章封面") diff --git a/src/wechat_decrypt_tool/sns_export_service.py b/src/wechat_decrypt_tool/sns_export_service.py index 996f0cc3..16a21fc0 100644 --- a/src/wechat_decrypt_tool/sns_export_service.py +++ b/src/wechat_decrypt_tool/sns_export_service.py @@ -247,10 +247,7 @@ async def worker(http_client: httpx.AsyncClient) -> None: except Exception as exc: result.failed += 1 result.missing.append(task_id) - logger.info( - "[sns.export] status=error phase=media-prefetch error_type=%s", - type(exc).__name__, - ) + logger.info("sns media prefetch failed: kind=%s url=%s error=%s", task.kind, task.url, exc) finally: completed += 1 if on_progress is not None: @@ -1634,12 +1631,7 @@ def _run_job_safe(self, job: ExportJob, account_dir: Path) -> None: pass self.commit_staged_files(job.export_id) except Exception as e: - logger.error( - "[sns.export] status=error export_id=%s phase=%s error_type=%s", - job.export_id, - str(job.progress.phase or "unknown"), - type(e).__name__, - ) + logger.exception("sns export job failed: %s: %s", job.export_id, e) with self._lock: job.status = "error" job.error = str(e) @@ -1704,11 +1696,7 @@ def set_phase(name: str) -> None: if sync_status not in {"ok", "noop"}: job.warning = "实时同步未完成，已继续导出本地历史快照。" except Exception as exc: # 导出必须可在无 native broker 的环境中降级运行 - logger.warning( - "[sns.export] status=error export_id=%s phase=syncing error_type=%s", - job.export_id, - type(exc).__name__, - ) + logger.warning("sns realtime sync before export failed: export=%s error=%s", job.export_id, exc) job.warning = "实时同步失败，已继续导出本地历史快照。" job.freshness = { "status": "warning", diff --git a/src/wechat_decrypt_tool/sns_media.py b/src/wechat_decrypt_tool/sns_media.py index 7605c8c6..5faba142 100644 --- a/src/wechat_decrypt_tool/sns_media.py +++ b/src/wechat_decrypt_tool/sns_media.py @@ -91,20 +91,33 @@ def _sns_remote_diagnostic_log( error: Optional[BaseException] = None, **fields: object, ) -> None: - # 朋友圈日志只保留诊断阶段和受控结果，不记录 URL、Host、密钥摘要或响应尺寸。 + raw_url = str(url or "").strip() + try: + host = str(urlparse(raw_url).hostname or "").strip().lower() + except Exception: + host = "" + + stable_url = normalize_sns_cache_url(raw_url) payload: dict[str, object] = { "diagnosticId": str(diagnostic_id or ""), "event": str(event or ""), + "urlHost": host, + "urlIdentity": ( + hashlib.sha256(stable_url.encode("utf-8", errors="ignore")).hexdigest()[:16] + if stable_url + else "" + ), + **fields, } - for field_name in ("stage", "result", "statusCode", "elapsedMs", "attempt"): - if field_name in fields: - payload[field_name] = fields[field_name] if error is not None: - error_type = type(error).__name__ - payload["errorType"] = ( - error_type if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,79}", error_type) else "Exception" - ) + error_text = str(error).strip() or repr(error) + for sensitive in (raw_url, str(key or ""), str(token or "")): + if sensitive: + error_text = error_text.replace(sensitive, "") + error_text = re.sub(r"https?://[^\s\"']+", "", error_text, flags=re.I) + payload["errorType"] = type(error).__name__ + payload["errorText"] = error_text[:500] response = getattr(error, "response", None) status_code = getattr(response, "status_code", None) @@ -112,7 +125,7 @@ def _sns_remote_diagnostic_log( try: payload["statusCode"] = int(status_code) except Exception: - payload["statusCode"] = 0 + payload["statusCode"] = str(status_code) logger.info( "[sns_media] %s", diff --git a/src/wechat_decrypt_tool/sns_realtime_autosync.py b/src/wechat_decrypt_tool/sns_realtime_autosync.py index f232f7cd..3cf56e5a 100644 --- a/src/wechat_decrypt_tool/sns_realtime_autosync.py +++ b/src/wechat_decrypt_tool/sns_realtime_autosync.py @@ -177,6 +177,7 @@ def _bootstrap_accounts(self) -> None: try: accounts = list(_list_decrypted_accounts() or []) 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: @@ -316,6 +317,7 @@ def _watch_directory(self, watch_key: str) -> None: self._mark_watcher_failed(watch_key, "sns_file_watch_unavailable") 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") @@ -358,7 +360,7 @@ def _schedule_sync(self, account: str, *, reason: str) -> None: worker = threading.Thread( target=self._sync_account_runner, args=(account, reason, revision), - name="sns-incremental-sync", + name=f"sns-event-sync-{account}", daemon=True, ) state.worker = worker @@ -371,6 +373,7 @@ def _schedule_sync(self, account: str, *, reason: str) -> None: 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__, @@ -480,8 +483,14 @@ def _refresh_native_moments_once(self, account: str) -> None: client, NativeCoreFeature.WECHAT_MOMENTS_REFRESH ) 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__, @@ -498,6 +507,7 @@ def _sync_with_bounded_retries(self, account: str, revision: int) -> tuple[dict[ return {"status": "skipped", "reason": "service_stopping"}, False last_result = dict(self._sync_account(account) or {}) 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__, @@ -556,6 +566,7 @@ 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 as exc: + logger.exception("[sns-autosync] 增量同步调用失败 account=%s", account) logger.error( "[sns.incremental-sync] status=error phase=scanning error_type=%s", type(exc).__name__, @@ -623,6 +634,14 @@ def _publish_sync_result(self, account: str, reason: str, revision: int, result: "timestamp": int(time.time() * 1000), } self._publish_event(account, event) + logger.info( + "[sns-autosync] 事件同步完成 account=%s reason=%s revision=%s status=%s changed=%s", + account, + reason, + revision, + status, + event["changed"], + ) return raw_code = str((result or {}).get("error") or (result or {}).get("reason") or "sns_sync_failed").strip() diff --git a/tests/test_request_log_redaction.py b/tests/test_request_log_redaction.py index c486b7eb..ae19880b 100644 --- a/tests/test_request_log_redaction.py +++ b/tests/test_request_log_redaction.py @@ -118,40 +118,6 @@ def test_uvicorn_access_filter_redacts_query_secrets(self): self.assertNotIn("AES_SECRET", rendered) self.assertIn("db_storage_path=C%3A%5Cdb", rendered) - def test_uvicorn_access_filter_removes_all_sns_query_values(self): - from wechat_decrypt_tool.request_logging import ( - SensitiveQueryLogFilter, - redact_sensitive_query_text, - ) - - record = logging.LogRecord( - "uvicorn.access", - logging.INFO, - __file__, - 1, - '%s - "%s %s HTTP/%s" %d', - ( - "127.0.0.1:1234", - "GET", - "/api/sns/media?account=ACCOUNT_SENTINEL&post_id=POST_SENTINEL&url=https%3A%2F%2Fprivate.example", - "1.1", - 200, - ), - None, - ) - - self.assertTrue(SensitiveQueryLogFilter().filter(record)) - rendered = record.getMessage() - self.assertIn("/api/sns", rendered) - self.assertNotIn("/media", rendered) - self.assertNotIn("ACCOUNT_SENTINEL", rendered) - self.assertNotIn("POST_SENTINEL", rendered) - self.assertNotIn("private.example", rendered) - self.assertEqual( - redact_sensitive_query_text("/api/sns/exports/EXPORT_SENTINEL/files/private-file.jpg"), - "/api/sns", - ) - if __name__ == "__main__": unittest.main() diff --git a/tests/test_sns_log_privacy.py b/tests/test_sns_log_privacy.py deleted file mode 100644 index 89914819..00000000 --- a/tests/test_sns_log_privacy.py +++ /dev/null @@ -1,174 +0,0 @@ -import asyncio -import hashlib -import sys -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 fastapi import HTTPException - -from wechat_decrypt_tool import sns_export_service, sns_media -from wechat_decrypt_tool import request_logging -from wechat_decrypt_tool.routers import sns as sns_router - - -def _render_calls(calls) -> str: - rendered = [] - for item in calls: - args = item.args - if not args: - continue - try: - rendered.append(str(args[0]) % tuple(args[1:])) - except Exception: - rendered.append(" ".join(str(value) for value in args)) - return "\n".join(rendered) - - -class TestSnsLogPrivacy(unittest.TestCase): - def test_sns_server_exception_log_omits_exception_text_and_traceback(self): - sentinel = "C:/private/account-sentinel/source.db" - - class _Url: - path = "/api/sns/media" - - class _Request: - method = "GET" - url = _Url() - - async def fail(_request): - raise RuntimeError(sentinel) - - fake_logger = mock.Mock() - with self.assertRaises(RuntimeError): - asyncio.run(request_logging.log_server_errors_middleware(fake_logger, _Request(), fail)) - - logs = _render_calls(fake_logger.error.call_args_list) - self.assertIn("code=sns_request_failed", logs) - self.assertIn("error_type=RuntimeError", logs) - self.assertNotIn(sentinel, logs) - fake_logger.exception.assert_not_called() - - def test_remote_media_log_omits_all_user_and_derived_sentinels(self): - sentinels = { - "account": "privacy-account-sentinel", - "nickname": "隐私昵称哨兵", - "content": "朋友圈正文哨兵", - "url": "https://private.example.test/secret/path?token=url-token-sentinel", - "path": "C:/Users/private-user/朋友圈/private-file.jpg", - "post": "post-id-sentinel", - "media": "media-id-sentinel", - "key": "key-sentinel-value", - "token": "token-sentinel-value", - } - derived = { - hashlib.sha256(value.encode("utf-8")).hexdigest() - for value in sentinels.values() - } - - with mock.patch.object(sns_media.logger, "info") as log_info: - sns_media._sns_remote_diagnostic_log( - "remote:download-error", - url=sentinels["url"], - diagnostic_id="diag-safe-1", - key=sentinels["key"], - token=sentinels["token"], - error=RuntimeError(f"failed at {sentinels['path']} {sentinels['content']}"), - account=sentinels["account"], - nickname=sentinels["nickname"], - postId=sentinels["post"], - mediaId=sentinels["media"], - candidatePath=sentinels["path"], - responseSha256=next(iter(derived)), - width=123, - height=456, - createTime=1234567890, - statusCode=503, - ) - - logs = _render_calls(log_info.call_args_list) - self.assertIn("diag-safe-1", logs) - self.assertIn('"statusCode": 503', logs) - self.assertIn('"errorType": "RuntimeError"', logs) - for value in (*sentinels.values(), *derived): - self.assertNotIn(value, logs) - for forbidden_field in ( - "urlHost", - "urlIdentity", - "errorText", - "candidatePath", - "responseSha256", - "postId", - "mediaId", - "width", - "height", - "createTime", - ): - self.assertNotIn(forbidden_field, logs) - - def test_incremental_unavailable_log_does_not_include_account_or_path(self): - with TemporaryDirectory() as td: - account_dir = Path(td) / "private-account-sentinel" - account_dir.mkdir() - with ( - mock.patch.object(sns_router, "_resolve_account_dir", return_value=account_dir), - mock.patch.object(sns_router.WCDB_REALTIME, "get_status", return_value={}), - mock.patch.object(sns_router.logger, "info") as log_info, - mock.patch.object(sns_router.logger, "error") as log_error, - ): - with self.assertRaises(HTTPException): - sns_router.sync_sns_realtime_timeline_latest(account=account_dir.name) - - logs = _render_calls(log_info.call_args_list + log_error.call_args_list) - self.assertIn("[sns.incremental-sync]", logs) - self.assertIn("realtime_not_available", logs) - self.assertNotIn(account_dir.name, logs) - self.assertNotIn(str(account_dir), logs) - - def test_export_prefetch_failure_log_omits_url_and_exception_text(self): - task = sns_export_service.SnsRemoteMediaTask( - kind="image", - url="https://private.example.test/media/url-sentinel", - key="private-key-sentinel", - token="private-token-sentinel", - ) - - async def fail_fetch(**_kwargs): - raise RuntimeError("C:/private/path-sentinel") - - async def run(account_dir: Path): - return await sns_export_service._prefetch_sns_remote_media( - account_dir=account_dir, - tasks=[task], - use_cache=False, - concurrency=1, - ) - - with TemporaryDirectory() as td: - account_dir = Path(td) / "account-sentinel" - account_dir.mkdir() - with ( - mock.patch.object( - sns_export_service, - "_try_fetch_and_decrypt_sns_image_remote", - side_effect=fail_fetch, - ), - mock.patch.object(sns_export_service.logger, "info") as log_info, - ): - result = asyncio.run(run(account_dir)) - - self.assertEqual(result.failed, 1) - logs = _render_calls(log_info.call_args_list) - self.assertIn("error_type=RuntimeError", logs) - for sentinel in (task.url, task.key, task.token, "C:/private/path-sentinel", account_dir.name): - self.assertNotIn(sentinel, logs) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_sns_media.py b/tests/test_sns_media.py index 716968e3..305b3a7c 100644 --- a/tests/test_sns_media.py +++ b/tests/test_sns_media.py @@ -265,10 +265,8 @@ async def run(account_dir: Path): self.assertIn('"diagnosticId": "diag-http-400"', logs) self.assertIn('"errorType": "HTTPStatusError"', logs) self.assertIn('"statusCode": 400', logs) - self.assertNotIn("mmsns.qpic.cn", logs) - self.assertNotIn('"urlHost":', logs) - self.assertNotIn('"urlIdentity":', logs) - self.assertNotIn('"errorText":', logs) + self.assertIn('"urlHost": "mmsns.qpic.cn"', logs) + self.assertIn('"urlIdentity":', logs) self.assertNotIn(raw_url_token, logs) self.assertNotIn(request_token, logs) self.assertNotIn(request_key, logs) diff --git a/tests/test_sns_media_route_weflow_default.py b/tests/test_sns_media_route_weflow_default.py index 1a5084af..d05f262e 100644 --- a/tests/test_sns_media_route_weflow_default.py +++ b/tests/test_sns_media_route_weflow_default.py @@ -201,17 +201,16 @@ def test_route_logs_redacted_identity_and_local_match_details(self): self.assertIn("sns.media request:start", logs) self.assertIn("sns.media local-key-post:probe", logs) self.assertIn("sns.media local-heuristic:probe", logs) - self.assertIn('"result": "local-cache"', logs) - self.assertNotIn('"matchedBy":', logs) - self.assertNotIn('"candidatePath":', logs) - self.assertNotIn("matched-local.jpg", logs) - self.assertNotIn("post-42", logs) - self.assertNotIn("media-7", logs) - self.assertNotIn("mmsns.qpic.cn", logs) - self.assertNotIn('"urlIdentity":', logs) - self.assertNotIn('"tokenHash":', logs) - self.assertNotIn('"keyHash":', logs) - self.assertNotIn('"responseSha256":', logs) + self.assertIn('"matchedBy": "local-heuristic"', logs) + self.assertIn('"candidatePath":', logs) + self.assertIn("matched-local.jpg", logs) + self.assertIn('"postId": "post-42"', logs) + self.assertIn('"mediaId": "media-7"', logs) + self.assertIn('"urlHost": "mmsns.qpic.cn"', logs) + self.assertIn('"urlIdentity":', logs) + self.assertIn('"tokenHash":', logs) + self.assertIn('"keyHash":', logs) + self.assertIn('"responseSha256":', logs) self.assertNotIn("super-secret-token", logs) self.assertNotIn("super-secret-key", logs) self.assertNotIn("url-secret-token", logs) @@ -339,8 +338,7 @@ def test_route_logs_exact_key_and_md5_match_strategies(self): logs = self._render_log_calls(log_info) self.assertEqual(resp.status_code, 200) - self.assertIn(f"sns.media {expected_match}:probe", logs) - self.assertNotIn('"matchedBy":', logs) + self.assertIn(f'"matchedBy": "{expected_match}"', logs) heuristic.assert_not_called() remote.assert_not_called()