channels: wechat message scraping into context and memory (follow-up to #1991) (#1990) (#2264)

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
CodeGhost21
2026-05-22 12:53:47 -07:00
committed by GitHub
co-authored by Cursor
parent e854a07ece
commit f1eae121a6
8 changed files with 1060 additions and 5 deletions
+2
View File
@@ -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());
+27 -4
View File
@@ -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<R: Runtime>(app: &AppHandle<R>) {
{
total += registry.inner().forget_all();
}
if let Some(registry) =
app.try_state::<std::sync::Arc<crate::wechat_scanner::ScannerRegistry>>()
{
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<R: Runtime>(app: &AppHandle<R>, account_id: &str) {
{
registry.inner().forget(account_id);
}
if let Some(registry) =
app.try_state::<std::sync::Arc<crate::wechat_scanner::ScannerRegistry>>()
{
registry.inner().forget(account_id);
}
}
#[derive(Debug, Clone)]
@@ -1761,7 +1771,7 @@ fn data_directory_for<R: Runtime>(app: &AppHandle<R>, 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<R: Runtime>(
} else {
log::warn!("[webview-accounts] discord ScannerRegistry not in app state");
}
} else if args.provider == "wechat" {
if let Some(registry) = app
.try_state::<std::sync::Arc<crate::wechat_scanner::ScannerRegistry>>()
.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
@@ -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<String>,
pub unread: u32,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MessageRow {
pub chat_id: String,
pub chat_name: String,
pub sender: Option<String>,
pub body: String,
pub ts: Option<i64>,
}
pub struct DomScan {
pub chat_rows: Vec<ChatRow>,
pub messages: Vec<MessageRow>,
pub active_chat_name: Option<String>,
pub unread: u32,
pub hash: u64,
}
pub async fn scan(cdp: &mut CdpConn, session: &str) -> Result<DomScan, String> {
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<String> {
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<String> {
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<String> {
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<String> {
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<String> {
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<usize> {
(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"
);
}
}
+240
View File
@@ -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<R: Runtime>(
app: AppHandle<R>,
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<u64> = 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<dom_snapshot::DomScan, String> {
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<R: Runtime>(app: &AppHandle<R>, 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<String, (String, Vec<WechatMessageRow>)> = 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<serde_json::Map<String, Value>, 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<HashMap<String, AbortHandle>>,
}
impl ScannerRegistry {
pub fn new() -> Arc<Self> {
Arc::new(Self::default())
}
pub fn ensure_scanner<R: Runtime>(
&self,
app: AppHandle<R>,
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");
}
}
+11 -1
View File
@@ -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,
};
+17
View File
@@ -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();
@@ -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<String>,
#[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<String>,
pub body: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ts: Option<i64>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WechatScanPayload {
pub account_id: String,
#[serde(default)]
pub chat_rows: Vec<WechatChatRow>,
#[serde(default)]
pub messages: Vec<WechatMessageRow>,
#[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<Value> = 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<Map<String, Value>, 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<Map<String, Value>, 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::<Vec<_>>()
.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<String, Value> {
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("ééééééééé"), "éééééééé");
}
}
@@ -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());
}