From f1eae121a697f93f0a9a3a59417c5d92f239433e Mon Sep 17 00:00:00 2001 From: CodeGhost21 <164498022+CodeGhost21@users.noreply.github.com> Date: Sat, 23 May 2026 01:23:47 +0530 Subject: [PATCH] channels: wechat message scraping into context and memory (follow-up to #1991) (#1990) (#2264) Co-authored-by: Cursor --- app/src-tauri/src/lib.rs | 2 + app/src-tauri/src/webview_accounts/mod.rs | 31 +- .../src/wechat_scanner/dom_snapshot.rs | 351 +++++++++++++++++ app/src-tauri/src/wechat_scanner/mod.rs | 240 ++++++++++++ src/openhuman/webview_accounts/mod.rs | 12 +- src/openhuman/webview_accounts/ops.rs | 17 + .../webview_accounts/wechat_ingest.rs | 358 ++++++++++++++++++ .../webview_accounts/wechat_ingest_test.rs | 54 +++ 8 files changed, 1060 insertions(+), 5 deletions(-) create mode 100644 app/src-tauri/src/wechat_scanner/dom_snapshot.rs create mode 100644 app/src-tauri/src/wechat_scanner/mod.rs create mode 100644 src/openhuman/webview_accounts/wechat_ingest.rs create mode 100644 src/openhuman/webview_accounts/wechat_ingest_test.rs diff --git a/app/src-tauri/src/lib.rs b/app/src-tauri/src/lib.rs index ffad8923d..f3c95f5b0 100644 --- a/app/src-tauri/src/lib.rs +++ b/app/src-tauri/src/lib.rs @@ -32,6 +32,7 @@ mod slack_scanner; mod telegram_scanner; mod webview_accounts; mod webview_apis; +mod wechat_scanner; mod whatsapp_scanner; mod window_state; @@ -2493,6 +2494,7 @@ pub fn run() { let builder = builder.manage(slack_scanner::ScannerRegistry::new()); let builder = builder.manage(discord_scanner::ScannerRegistry::new()); let builder = builder.manage(telegram_scanner::ScannerRegistry::new()); + let builder = builder.manage(wechat_scanner::ScannerRegistry::new()); let builder = builder.manage(screen_capture::ScreenShareState::new()); let builder = builder.manage(meet_call::MeetCallState::new()); let builder = builder.manage(meet_audio::MeetAudioState::new()); diff --git a/app/src-tauri/src/webview_accounts/mod.rs b/app/src-tauri/src/webview_accounts/mod.rs index aada3efc1..15419daa5 100644 --- a/app/src-tauri/src/webview_accounts/mod.rs +++ b/app/src-tauri/src/webview_accounts/mod.rs @@ -62,9 +62,9 @@ fn provider_url(provider: &str) -> Option<&'static str> { } /// Returns the injected recipe.js for providers that still rely on the -/// JS-bridge ingest path. Migrated providers (whatsapp, telegram, slack, -/// discord, browserscan) return `None` — their scraping runs natively via -/// CDP in the per-provider scanner modules. +/// JS-bridge ingest path. Migrated providers (whatsapp, wechat, telegram, +/// slack, discord, browserscan) return `None` — their scraping runs natively +/// via CDP in the per-provider scanner modules. fn provider_recipe_js(provider: &str) -> Option<&'static str> { match provider { "linkedin" => Some(LINKEDIN_RECIPE_JS), @@ -967,6 +967,11 @@ fn teardown_all_account_scanners(app: &AppHandle) { { total += registry.inner().forget_all(); } + if let Some(registry) = + app.try_state::>() + { + total += registry.inner().forget_all(); + } if total > 0 { log::info!( "[webview-accounts] aborted {} provider scanner task(s) for shutdown", @@ -999,6 +1004,11 @@ fn teardown_account_scanners(app: &AppHandle, account_id: &str) { { registry.inner().forget(account_id); } + if let Some(registry) = + app.try_state::>() + { + registry.inner().forget(account_id); + } } #[derive(Debug, Clone)] @@ -1761,7 +1771,7 @@ fn data_directory_for(app: &AppHandle, account_id: &str) -> Resul /// /// Empty for the 6 zero-injection providers (whatsapp, wechat, telegram, /// slack, discord, browserscan) — they load with ZERO injected JS. Some have -/// native/CDP scraper paths; WeChat is shell-only for now. The per-account +/// native/CDP scraper paths (`wechat_scanner`, etc.). The per-account /// CDP session opener (`cdp::session`) still injects the notification-permission /// shim via `Page.addScriptToEvaluateOnNewDocument` before the real provider /// URL loads. The 2 deferred providers (linkedin, google-meet) still get the @@ -2542,6 +2552,19 @@ pub async fn webview_account_open( } else { log::warn!("[webview-accounts] discord ScannerRegistry not in app state"); } + } else if args.provider == "wechat" { + if let Some(registry) = app + .try_state::>() + .map(|s| s.inner().clone()) + { + registry.ensure_scanner( + app.clone(), + args.account_id.clone(), + scanner_url_prefix.clone(), + ); + } else { + log::warn!("[webview-accounts] wechat ScannerRegistry not in app state"); + } } // Browser Notification interception, native CEF path. The renderer diff --git a/app/src-tauri/src/wechat_scanner/dom_snapshot.rs b/app/src-tauri/src/wechat_scanner/dom_snapshot.rs new file mode 100644 index 000000000..b750d9a1b --- /dev/null +++ b/app/src-tauri/src/wechat_scanner/dom_snapshot.rs @@ -0,0 +1,351 @@ +//! WeChat Web DOM scrape via `DOMSnapshot.captureSnapshot` (pure CDP). + +use serde_json::Value; + +use crate::cdp::{CdpConn, Snapshot}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ChatRow { + pub name: String, + pub preview: Option, + pub unread: u32, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MessageRow { + pub chat_id: String, + pub chat_name: String, + pub sender: Option, + pub body: String, + pub ts: Option, +} + +pub struct DomScan { + pub chat_rows: Vec, + pub messages: Vec, + pub active_chat_name: Option, + pub unread: u32, + pub hash: u64, +} + +pub async fn scan(cdp: &mut CdpConn, session: &str) -> Result { + let snap = Snapshot::capture(cdp, session).await?; + let mut chat_rows = Vec::new(); + let mut unread: u32 = 0; + for idx in snap.find_all(is_chat_list_row) { + let name = find_row_title(&snap, idx).unwrap_or_default(); + let preview = find_row_preview(&snap, idx); + let badge = find_row_unread(&snap, idx); + if name.is_empty() && preview.as_deref().map(str::is_empty).unwrap_or(true) { + continue; + } + unread = unread.saturating_add(badge); + chat_rows.push(ChatRow { + name, + preview, + unread: badge, + }); + } + let active_chat_name = find_active_chat_title(&snap); + let chat_id_base = active_chat_name + .as_deref() + .filter(|s| !s.is_empty()) + .unwrap_or("active"); + let mut messages = Vec::new(); + for idx in snap.find_all(is_message_bubble) { + let body = snap.text_content(idx); + if body.len() < 2 { + continue; + } + messages.push(MessageRow { + chat_id: chat_id_base.to_string(), + chat_name: active_chat_name + .clone() + .unwrap_or_else(|| chat_id_base.to_string()), + sender: find_message_sender(&snap, idx), + body, + ts: None, + }); + } + let hash = hash_scan(&chat_rows, &messages, unread); + Ok(DomScan { + chat_rows, + messages, + active_chat_name, + unread, + hash, + }) +} + +pub fn scan_to_core_payload( + account_id: &str, + scan: &DomScan, +) -> openhuman_core::openhuman::webview_accounts::WechatScanPayload { + use openhuman_core::openhuman::webview_accounts::{ + WechatChatRow, WechatMessageRow, WechatScanPayload, + }; + WechatScanPayload { + account_id: account_id.to_string(), + chat_rows: scan + .chat_rows + .iter() + .map(|r| WechatChatRow { + name: r.name.clone(), + preview: r.preview.clone(), + unread: r.unread, + }) + .collect(), + messages: scan + .messages + .iter() + .map(|m| WechatMessageRow { + chat_id: m.chat_id.clone(), + chat_name: m.chat_name.clone(), + sender: m.sender.clone(), + body: m.body.clone(), + ts: m.ts, + }) + .collect(), + unread: scan.unread, + snapshot_key: format!("{:x}", scan.hash), + source: "cdp-dom".to_string(), + } +} + +#[allow(dead_code)] +pub fn ingest_payload_for_scan(scan: &DomScan) -> Value { + openhuman_core::openhuman::webview_accounts::list_ingest_payload(&scan_to_core_payload( + "test-account", + scan, + )) +} + +fn is_chat_list_row(snap: &Snapshot, idx: usize) -> bool { + if !snap.is_element(idx) { + return false; + } + let tag = snap.tag(idx); + (tag.eq_ignore_ascii_case("LI") || tag.eq_ignore_ascii_case("DIV")) + && (class_matches_any( + snap, + idx, + &[ + "session", + "chat-item", + "chat_item", + "conversation-item", + "recent", + "nav-item", + ], + ) || snap.attr(idx, "data-chat-id").is_some()) +} + +fn is_message_bubble(snap: &Snapshot, idx: usize) -> bool { + snap.is_element(idx) + && (class_matches_any( + snap, + idx, + &[ + "message", + "msg", + "bubble", + "chat-message", + "message-item", + "msg-item", + ], + ) || snap.attr(idx, "data-message-id").is_some()) +} + +fn class_matches_any(snap: &Snapshot, idx: usize, needles: &[&str]) -> bool { + snap.classes(idx).any(|c| { + let lower = c.to_ascii_lowercase(); + needles.iter().any(|n| lower.contains(n)) + }) +} + +fn find_row_title(snap: &Snapshot, root: usize) -> Option { + find_text_by_class_hints( + snap, + root, + &[ + "nickname", + "nick-name", + "title", + "name", + "user-name", + "session-name", + ], + ) +} + +fn find_row_preview(snap: &Snapshot, root: usize) -> Option { + find_text_by_class_hints( + snap, + root, + &["preview", "last-msg", "msg-preview", "desc", "subtitle"], + ) +} + +fn find_row_unread(snap: &Snapshot, root: usize) -> u32 { + find_text_by_class_hints(snap, root, &["badge", "unread", "count", "num"]) + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(0) +} + +fn find_active_chat_title(snap: &Snapshot) -> Option { + snap.find_all(|s, i| { + s.is_element(i) + && class_matches_any( + s, + i, + &["chat-title", "conversation-title", "header-title", "title"], + ) + }) + .into_iter() + .find_map(|idx| { + let t = snap.text_content(idx); + if t.is_empty() { + None + } else { + Some(t) + } + }) +} + +fn find_message_sender(snap: &Snapshot, bubble: usize) -> Option { + parent_of(snap, bubble) + .and_then(|parent| find_text_by_class_hints(snap, parent, &["sender", "nickname", "name"])) +} + +fn find_text_by_class_hints(snap: &Snapshot, root: usize, hints: &[&str]) -> Option { + let node = snap.find_descendant(root, |s, i| { + s.is_element(i) && class_matches_any(s, i, hints) + })?; + let t = snap.text_content(node); + if t.is_empty() { + None + } else { + Some(t) + } +} + +fn parent_of(snap: &Snapshot, idx: usize) -> Option { + (0..snap.len()).find(|&i| snap.children(i).contains(&idx)) +} + +fn hash_scan(chat_rows: &[ChatRow], messages: &[MessageRow], unread: u32) -> u64 { + let mut h: u64 = 0xcbf29ce484222325; + fn mix(h: &mut u64, b: u8) { + *h ^= b as u64; + *h = h.wrapping_mul(0x100000001b3); + } + for b in (chat_rows.len() as u32).to_le_bytes() { + mix(&mut h, b); + } + for b in (messages.len() as u32).to_le_bytes() { + mix(&mut h, b); + } + for b in unread.to_le_bytes() { + mix(&mut h, b); + } + for r in chat_rows { + for b in r.name.as_bytes() { + mix(&mut h, *b); + } + mix(&mut h, 0x7c); + if let Some(p) = &r.preview { + for b in p.as_bytes() { + mix(&mut h, *b); + } + } + mix(&mut h, 0x7c); + for b in r.unread.to_le_bytes() { + mix(&mut h, b); + } + } + for m in messages { + for b in m.chat_id.as_bytes() { + mix(&mut h, *b); + } + mix(&mut h, 0x7c); + for b in m.chat_name.as_bytes() { + mix(&mut h, *b); + } + mix(&mut h, 0x7c); + if let Some(sender) = &m.sender { + for b in sender.as_bytes() { + mix(&mut h, *b); + } + } + mix(&mut h, 0x7c); + for b in m.body.as_bytes() { + mix(&mut h, *b); + } + mix(&mut h, 0x7c); + if let Some(ts) = m.ts { + for b in ts.to_le_bytes() { + mix(&mut h, b); + } + } + mix(&mut h, 0x7c); + } + h +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hash_changes_when_message_body_changes() { + let row = ChatRow { + name: "A".into(), + preview: None, + unread: 0, + }; + let first = MessageRow { + chat_id: "c".into(), + chat_name: "A".into(), + sender: Some("alice".into()), + body: "hello".into(), + ts: Some(1), + }; + let second = MessageRow { + body: "world".into(), + ..first.clone() + }; + assert_ne!( + hash_scan(&[row.clone()], &[first], 0), + hash_scan(&[row], &[second], 0) + ); + } + + #[test] + fn hash_changes_when_unread_moves_between_chats() { + let a1 = ChatRow { + name: "A".into(), + preview: None, + unread: 2, + }; + let b1 = ChatRow { + name: "B".into(), + preview: None, + unread: 0, + }; + let a2 = ChatRow { + name: "A".into(), + preview: None, + unread: 0, + }; + let b2 = ChatRow { + name: "B".into(), + preview: None, + unread: 2, + }; + assert_ne!( + hash_scan(&[a1, b1], &[], 2), + hash_scan(&[a2, b2], &[], 2), + "per-chat unread distribution must affect the hash" + ); + } +} diff --git a/app/src-tauri/src/wechat_scanner/mod.rs b/app/src-tauri/src/wechat_scanner/mod.rs new file mode 100644 index 000000000..e6ef04304 --- /dev/null +++ b/app/src-tauri/src/wechat_scanner/mod.rs @@ -0,0 +1,240 @@ +//! WeChat Web scanner over CDP — chat list + active conversation DOM scrape. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use openhuman_core::openhuman::webview_accounts::{ + list_ingest_envelope, memory_doc_ingest_list_snapshot, memory_doc_ingest_peer_transcript, + validate_scan, WechatMessageRow, WechatScanPayload, +}; +use parking_lot::Mutex; +use serde_json::{json, Value}; +use tauri::{AppHandle, Emitter, Runtime}; +use tokio::task::AbortHandle; +use tokio::time::sleep; + +mod dom_snapshot; + +const SCAN_INTERVAL: Duration = Duration::from_secs(3); +const STARTUP_DELAY: Duration = Duration::from_secs(8); + +pub fn wechat_scanner_disabled() -> bool { + matches!( + std::env::var("OPENHUMAN_DISABLE_WECHAT_SCANNER") + .ok() + .as_deref() + .map(str::trim), + Some("1") | Some("true") | Some("yes") + ) +} + +pub fn spawn_scanner( + app: AppHandle, + account_id: String, + url_prefix: String, +) -> AbortHandle { + tokio::spawn(async move { + let fragment = crate::cdp::target_url_fragment(&account_id); + log::info!( + "[wechat] scanner up account={} url_prefix={} fragment={}", + account_id, + url_prefix, + fragment + ); + sleep(STARTUP_DELAY).await; + let mut last_hash: Option = None; + loop { + match scan_once(&url_prefix, &fragment).await { + Ok(scan) => { + if Some(scan.hash) == last_hash { + sleep(SCAN_INTERVAL).await; + continue; + } + last_hash = Some(scan.hash); + let payload = dom_snapshot::scan_to_core_payload(&account_id, &scan); + if validate_scan(&payload).is_err() { + sleep(SCAN_INTERVAL).await; + continue; + } + log::info!( + "[wechat][{}] dom scan chats={} msgs={} unread={}", + account_id, + scan.chat_rows.len(), + scan.messages.len(), + scan.unread + ); + emit_and_persist(&app, &account_id, &payload); + } + Err(e) => log::debug!("[wechat][{}] dom scan failed: {}", account_id, e), + } + sleep(SCAN_INTERVAL).await; + } + }) + .abort_handle() +} + +async fn scan_once(url_prefix: &str, url_fragment: &str) -> Result { + let prefix = url_prefix.to_string(); + let fragment = url_fragment.to_string(); + let (mut cdp, session) = crate::cdp::connect_and_attach_matching(move |t| { + t.url.starts_with(&prefix) && t.url.ends_with(&fragment) + }) + .await?; + let scan = dom_snapshot::scan(&mut cdp, &session).await; + crate::cdp::detach_session(&mut cdp, &session).await; + scan +} + +fn emit_and_persist(app: &AppHandle, account_id: &str, payload: &WechatScanPayload) { + if let Err(e) = app.emit( + "webview:event", + &list_ingest_envelope(account_id, payload, chrono_now_millis()), + ) { + log::warn!("[wechat][{}] ingest emit failed: {}", account_id, e); + } + if !payload.chat_rows.is_empty() { + let acct = account_id.to_string(); + let list = payload.clone(); + tokio::spawn(async move { + if let Err(e) = post_memory_doc(&acct, memory_doc_ingest_list_snapshot(&list)).await { + log::warn!("[wechat][{}] list memory failed: {}", acct, e); + } + }); + } + let mut groups: HashMap)> = HashMap::new(); + for m in &payload.messages { + if m.body.trim().is_empty() { + continue; + } + let e = groups.entry(m.chat_id.clone()).or_default(); + if e.0.is_empty() { + e.0 = m.chat_name.clone(); + } + e.1.push(m.clone()); + } + for (chat_id, (chat_name, rows)) in groups { + let acct = account_id.to_string(); + tokio::spawn(async move { + match memory_doc_ingest_peer_transcript(&acct, &chat_id, &chat_name, &rows) { + Ok(params) => { + if let Err(e) = post_memory_doc(&acct, Ok(params)).await { + log::warn!( + "[wechat][{}] peer memory upsert failed chat_id={}: {}", + acct, + chat_id, + e + ); + } + } + Err(e) => log::warn!( + "[wechat][{}] peer transcript build failed chat_id={}: {}", + acct, + chat_id, + e + ), + } + }); + } +} + +async fn post_memory_doc( + account_id: &str, + params: Result, String>, +) -> Result<(), String> { + let params = params?; + let body = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "openhuman.memory_doc_ingest", + "params": Value::Object(params), + }); + let url = crate::core_rpc::core_rpc_url_value(); + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(15)) + .build() + .map_err(|e| format!("http client: {e}"))?; + let resp = crate::core_rpc::apply_auth(client.post(&url)) + .map_err(|e| format!("prepare {url}: {e}"))? + .json(&body) + .send() + .await + .map_err(|e| format!("POST {url}: {e}"))?; + if !resp.status().is_success() { + return Err(format!( + "{}: {}", + resp.status(), + resp.text().await.unwrap_or_default() + )); + } + let v: Value = resp.json().await.map_err(|e| format!("decode: {e}"))?; + if v.get("error").is_some() { + return Err(format!("rpc error: {}", v["error"])); + } + log::info!("[wechat][{}] memory upsert ok", account_id); + Ok(()) +} + +fn chrono_now_millis() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +#[derive(Default)] +pub struct ScannerRegistry { + started: Mutex>, +} + +impl ScannerRegistry { + pub fn new() -> Arc { + Arc::new(Self::default()) + } + + pub fn ensure_scanner( + &self, + app: AppHandle, + account_id: String, + url_prefix: String, + ) { + if wechat_scanner_disabled() { + return; + } + let mut g = self.started.lock(); + if g.contains_key(&account_id) { + return; + } + let scanner_account_id = account_id.clone(); + g.insert( + account_id, + spawn_scanner(app, scanner_account_id, url_prefix), + ); + } + + pub fn forget(&self, account_id: &str) { + if let Some(h) = self.started.lock().remove(account_id) { + h.abort(); + } + } + + pub fn forget_all(&self) -> usize { + let entries: Vec<_> = self.started.lock().drain().collect(); + for (_, h) in &entries { + h.abort(); + } + entries.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn disabled_env_var_is_honored() { + std::env::set_var("OPENHUMAN_DISABLE_WECHAT_SCANNER", "1"); + assert!(wechat_scanner_disabled()); + std::env::remove_var("OPENHUMAN_DISABLE_WECHAT_SCANNER"); + } +} diff --git a/src/openhuman/webview_accounts/mod.rs b/src/openhuman/webview_accounts/mod.rs index c4624803b..ba36331a4 100644 --- a/src/openhuman/webview_accounts/mod.rs +++ b/src/openhuman/webview_accounts/mod.rs @@ -2,7 +2,7 @@ //! //! The Tauri shell hosts CEF-backed webviews for third-party accounts //! (Gmail, WhatsApp, Telegram, Slack, Discord, LinkedIn, Zoom, Google -//! Messages). Their HTTP cookies live in a single shared Chromium +//! Messages, WeChat). Their HTTP cookies live in a single shared Chromium //! cookie store at `{CEF_USER_DATA_DIR}/Default/Cookies` — a SQLite //! database. The core runs as a child sidecar and has no direct handle //! to CEF, so the Tauri shell exports `OPENHUMAN_CEF_COOKIES_DB` @@ -22,5 +22,15 @@ //! user has an active session for that provider. mod ops; +pub mod wechat_ingest; + +#[cfg(test)] +#[path = "wechat_ingest_test.rs"] +mod tests; pub use ops::detect_webview_logins; +pub use wechat_ingest::{ + list_ingest_envelope, list_ingest_payload, memory_doc_ingest_list_snapshot, + memory_doc_ingest_peer_transcript, validate_scan, WechatChatRow, WechatMessageRow, + WechatScanPayload, +}; diff --git a/src/openhuman/webview_accounts/ops.rs b/src/openhuman/webview_accounts/ops.rs index fa5171e08..6ed77ec8a 100644 --- a/src/openhuman/webview_accounts/ops.rs +++ b/src/openhuman/webview_accounts/ops.rs @@ -43,6 +43,11 @@ pub(crate) const PROVIDERS: &[Provider] = &[ host_suffix: "web.whatsapp.com", session_cookie_names: &["wa_ul", "wa_build"], }, + Provider { + key: "wechat", + host_suffix: "web.wechat.com", + session_cookie_names: &["wxuin", "webwx_data_ticket", "webwx_auth_ticket"], + }, Provider { key: "telegram", host_suffix: "web.telegram.org", @@ -317,6 +322,18 @@ mod tests { std::env::remove_var(COOKIES_DB_ENV); } + #[test] + fn detects_wechat_via_wxuin_cookie() { + let _lock = lock_env(); + let tmp = TempDir::new().unwrap(); + let db = tmp.path().join("Cookies"); + make_cookies_db(&db, &[("web.wechat.com", "wxuin")]); + std::env::set_var(COOKIES_DB_ENV, &db); + let v = detect_webview_logins(); + assert_eq!(v["wechat"], Value::Bool(true)); + std::env::remove_var(COOKIES_DB_ENV); + } + #[test] fn detects_slack_and_linkedin() { let _lock = lock_env(); diff --git a/src/openhuman/webview_accounts/wechat_ingest.rs b/src/openhuman/webview_accounts/wechat_ingest.rs new file mode 100644 index 000000000..253c8573c --- /dev/null +++ b/src/openhuman/webview_accounts/wechat_ingest.rs @@ -0,0 +1,358 @@ +//! WeChat Web ingest contract — normalized payloads for context + memory. + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Map, Value}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct WechatChatRow { + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub preview: Option, + #[serde(default)] + pub unread: u32, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct WechatMessageRow { + pub chat_id: String, + pub chat_name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sender: Option, + pub body: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ts: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct WechatScanPayload { + pub account_id: String, + #[serde(default)] + pub chat_rows: Vec, + #[serde(default)] + pub messages: Vec, + #[serde(default)] + pub unread: u32, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub snapshot_key: String, + #[serde(default = "default_source")] + pub source: String, +} + +fn default_source() -> String { + "cdp-dom".to_string() +} + +pub fn validate_scan(payload: &WechatScanPayload) -> Result<(), String> { + if payload.account_id.trim().is_empty() { + return Err("account_id is required".into()); + } + if payload.chat_rows.is_empty() && payload.messages.is_empty() { + return Err("scan has no chat rows or messages".into()); + } + Ok(()) +} + +pub fn list_ingest_envelope( + account_id: &str, + payload: &WechatScanPayload, + ts_millis: i64, +) -> Value { + json!({ + "account_id": account_id, + "provider": "wechat", + "kind": "ingest", + "payload": list_ingest_payload(payload), + "ts": ts_millis, + }) +} + +pub fn list_ingest_payload(payload: &WechatScanPayload) -> Value { + let messages: Vec = payload + .chat_rows + .iter() + .enumerate() + .map(|(idx, row)| { + let id = if row.name.is_empty() { + format!("wechat:row:{idx}") + } else { + format!("wechat:{idx}:{}", row.name) + }; + json!({ + "id": id, + "from": if row.name.is_empty() { Value::Null } else { json!(row.name) }, + "body": row.preview.clone().map(Value::String).unwrap_or(Value::Null), + "unread": row.unread, + }) + }) + .collect(); + json!({ + "messages": messages, + "unread": payload.unread, + "snapshotKey": payload.snapshot_key, + }) +} + +pub fn memory_doc_ingest_list_snapshot( + payload: &WechatScanPayload, +) -> Result, String> { + validate_scan(payload)?; + if payload.chat_rows.is_empty() { + return Err("no chat rows for list snapshot".into()); + } + let namespace = format!("wechat-web:{}", payload.account_id); + let key = if payload.snapshot_key.is_empty() { + format!("list:{}", chrono_day_key()) + } else { + format!("list:{}", payload.snapshot_key) + }; + Ok(memory_doc_params( + namespace, + key, + format!( + "WeChat · chat list · {}", + short_account(&payload.account_id) + ), + format_list_transcript(payload), + json!({ + "provider": "wechat", + "account_id": payload.account_id, + "kind": "chat-list", + "chat_count": payload.chat_rows.len(), + "unread": payload.unread, + }), + vec!["wechat", "chat-list"], + )) +} + +pub fn memory_doc_ingest_peer_transcript( + account_id: &str, + chat_id: &str, + chat_name: &str, + rows: &[WechatMessageRow], +) -> Result, String> { + if account_id.trim().is_empty() { + return Err("account_id is required".into()); + } + if chat_id.trim().is_empty() { + return Err("chat_id is required".into()); + } + if rows.is_empty() { + return Err("no messages for peer transcript".into()); + } + let mut sorted: Vec<&WechatMessageRow> = rows.iter().collect(); + sorted.sort_by_key(|m| m.ts.unwrap_or(0)); + let first_day = ts_to_ymd(sorted.first().and_then(|m| m.ts).unwrap_or(0)); + let last_day = ts_to_ymd(sorted.last().and_then(|m| m.ts).unwrap_or(0)); + let transcript: String = sorted + .iter() + .map(|m| { + let stamp = m.ts.map(format_message_stamp).unwrap_or_else(|| "?".into()); + let who = m.sender.as_deref().filter(|s| !s.is_empty()).unwrap_or("?"); + format!("[{stamp}] {who}: {}", m.body.replace(['\r', '\n'], " ")) + }) + .collect::>() + .join("\n"); + let peer_label = if chat_name.trim().is_empty() { + chat_id + } else { + chat_name + }; + let header = format!( + "# WeChat — {peer_label}\nchat_id: {chat_id}\naccount_id: {account_id}\nmessages: {}\nrange: {first_day} → {last_day}\n\n", + sorted.len() + ); + let key = if peer_key_looks_clean(chat_name) { + format!("{chat_id}:{chat_name}") + } else { + chat_id.to_string() + }; + Ok(memory_doc_params( + format!("wechat-web:{account_id}"), + key, + format!("WeChat · {peer_label}"), + format!("{header}{transcript}"), + json!({ + "provider": "wechat", + "account_id": account_id, + "chat_id": chat_id, + "chat_name": chat_name, + "message_count": sorted.len(), + }), + vec!["wechat", "peer-transcript"], + )) +} + +fn memory_doc_params( + namespace: String, + key: String, + title: String, + content: String, + metadata: Value, + tags: Vec<&str>, +) -> Map { + let mut params = Map::new(); + params.insert("namespace".into(), json!(namespace)); + params.insert("key".into(), json!(key)); + params.insert("title".into(), json!(title)); + params.insert("content".into(), json!(content)); + params.insert("source_type".into(), json!("wechat-web")); + params.insert("priority".into(), json!("medium")); + params.insert("tags".into(), json!(tags)); + params.insert("metadata".into(), metadata); + params.insert("category".into(), json!("core")); + params +} + +fn format_list_transcript(payload: &WechatScanPayload) -> String { + let mut lines = vec![ + "# WeChat — chat list".to_string(), + format!("account_id: {}", payload.account_id), + format!("chats: {}", payload.chat_rows.len()), + format!("unread: {}", payload.unread), + String::new(), + ]; + for row in &payload.chat_rows { + let preview = row.preview.as_deref().unwrap_or(""); + let badge = if row.unread > 0 { + format!(" [{} unread]", row.unread) + } else { + String::new() + }; + lines.push(format!("- {}{}: {}", row.name, badge, preview)); + } + lines.join("\n") +} + +fn short_account(account_id: &str) -> String { + if account_id.chars().count() <= 8 { + account_id.to_string() + } else { + account_id.chars().take(8).collect() + } +} + +fn peer_key_looks_clean(name: &str) -> bool { + !name.is_empty() + && name + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.') +} + +fn format_message_stamp(ts: i64) -> String { + let day = ts_to_ymd(ts); + let secs_of_day = (ts.rem_euclid(86_400)) as u32; + format!( + "{} {:02}:{:02}Z", + day, + secs_of_day / 3600, + (secs_of_day / 60) % 60 + ) +} + +fn ts_to_ymd(secs: i64) -> String { + if secs <= 0 { + return String::new(); + } + let days = secs.div_euclid(86_400); + let z = days + 719_468; + let era = if z >= 0 { z } else { z - 146_096 } / 146_097; + let doe = (z - era * 146_097) as u64; + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe as i64 + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = (doy - (153 * mp + 2) / 5 + 1) as u32; + let m = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32; + let y_real = (if m <= 2 { y + 1 } else { y }) as i32; + format!("{:04}-{:02}-{:02}", y_real, m, d) +} + +fn chrono_day_key() -> String { + ts_to_ymd( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn validate_rejects_empty_account() { + let mut p = WechatScanPayload { + account_id: "acct".into(), + chat_rows: vec![WechatChatRow { + name: "A".into(), + preview: None, + unread: 0, + }], + messages: vec![], + unread: 0, + snapshot_key: String::new(), + source: "cdp-dom".into(), + }; + p.account_id = " ".into(); + assert!(validate_scan(&p).is_err()); + } + + #[test] + fn validate_rejects_empty_scan() { + assert!(validate_scan(&WechatScanPayload { + account_id: "acct".into(), + chat_rows: vec![], + messages: vec![], + unread: 0, + snapshot_key: String::new(), + source: "cdp-dom".into(), + }) + .is_err()); + } + + #[test] + fn list_ingest_payload_has_messages() { + let v = list_ingest_payload(&WechatScanPayload { + account_id: "a".into(), + chat_rows: vec![WechatChatRow { + name: "Bob".into(), + preview: Some("hi".into()), + unread: 1, + }], + messages: vec![], + unread: 1, + snapshot_key: "k".into(), + source: "cdp-dom".into(), + }); + assert_eq!(v["messages"].as_array().map(|a| a.len()), Some(1)); + } + + #[test] + fn peer_transcript_rejects_empty_messages() { + assert!(memory_doc_ingest_peer_transcript("acct", "c1", "Alice", &[]).is_err()); + } + + #[test] + fn peer_transcript_key_includes_chat_id_for_clean_names() { + let rows = vec![WechatMessageRow { + chat_id: "chat-1".into(), + chat_name: "Alice".into(), + sender: None, + body: "hello".into(), + ts: Some(1), + }]; + + let first = memory_doc_ingest_peer_transcript("acct", "chat-1", "Alice", &rows).unwrap(); + let second = memory_doc_ingest_peer_transcript("acct", "chat-2", "Alice", &rows).unwrap(); + + assert_eq!(first["key"].as_str(), Some("chat-1:Alice")); + assert_eq!(second["key"].as_str(), Some("chat-2:Alice")); + } + + #[test] + fn short_account_truncates_on_char_boundary() { + assert_eq!(short_account("acct-123"), "acct-123"); + assert_eq!(short_account("ééééééééé"), "éééééééé"); + } +} diff --git a/src/openhuman/webview_accounts/wechat_ingest_test.rs b/src/openhuman/webview_accounts/wechat_ingest_test.rs new file mode 100644 index 000000000..6651019cc --- /dev/null +++ b/src/openhuman/webview_accounts/wechat_ingest_test.rs @@ -0,0 +1,54 @@ +use super::wechat_ingest::{ + list_ingest_envelope, memory_doc_ingest_peer_transcript, validate_scan, WechatChatRow, + WechatMessageRow, WechatScanPayload, +}; + +#[test] +fn envelope_includes_provider_and_kind() { + let payload = WechatScanPayload { + account_id: "acct-x".into(), + chat_rows: vec![WechatChatRow { + name: "Bob".into(), + preview: Some("ping".into()), + unread: 1, + }], + messages: vec![], + unread: 1, + snapshot_key: "deadbeef".into(), + source: "cdp-dom".into(), + }; + let env = list_ingest_envelope("acct-x", &payload, 1_234); + assert_eq!(env["provider"].as_str(), Some("wechat")); + assert_eq!(env["kind"].as_str(), Some("ingest")); +} + +#[test] +fn validate_accepts_messages_only_scan() { + let payload = WechatScanPayload { + account_id: "acct".into(), + chat_rows: vec![], + messages: vec![WechatMessageRow { + chat_id: "c1".into(), + chat_name: "Alice".into(), + sender: None, + body: "hello".into(), + ts: None, + }], + unread: 0, + snapshot_key: String::new(), + source: "cdp-dom".into(), + }; + assert!(validate_scan(&payload).is_ok()); +} + +#[test] +fn peer_transcript_rejects_blank_chat_id() { + let rows = vec![WechatMessageRow { + chat_id: " ".into(), + chat_name: "x".into(), + sender: None, + body: "y".into(), + ts: None, + }]; + assert!(memory_doc_ingest_peer_transcript("acct", " ", "x", &rows).is_err()); +}