diff --git a/.github/release-notes.md b/.github/release-notes.md index c4b3f47..8b0cc58 100644 --- a/.github/release-notes.md +++ b/.github/release-notes.md @@ -1,7 +1,6 @@ -## MPider v2.0.0 +## MPider v2.0.1 -- 新增本地微信推送监测。 -- 优化文章下载与限流配置。 +- 推送监测与待处理链接分开控制，已保存链接可独立处理。 - 其他体验改进和问题修复。 提供 Windows x64 安装版与便携版。 diff --git a/CHANGELOG.md b/CHANGELOG.md index 8706b35..6ce8918 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ 本页记录面向使用者的功能与已知限制。开发完成、实机验证和正式发布分别说明，发行状态以 [GitHub Releases](https://github.com/areyoubugcoder/mpider/releases) 为准。 +## v2.0.1 + +- 推送监测与待处理链接分开控制，已保存链接可独立处理。 +- 其他体验改进和问题修复。 + ## v2.0.0 - 新增本地微信推送监测。 diff --git a/Cargo.lock b/Cargo.lock index 64f0205..a90e559 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -102,7 +102,7 @@ dependencies = [ [[package]] name = "article-md" -version = "2.0.0" +version = "2.0.1" dependencies = [ "htmd", "regex", @@ -2933,7 +2933,7 @@ dependencies = [ [[package]] name = "mpider-core" -version = "2.0.0" +version = "2.0.1" dependencies = [ "ammonia", "anyhow", @@ -2965,7 +2965,7 @@ dependencies = [ [[package]] name = "mpider-tauri" -version = "2.0.0" +version = "2.0.1" dependencies = [ "mpider-core", "rfd", diff --git a/Cargo.toml b/Cargo.toml index 2c678b6..e0a709d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ members = ["crates/article-md", "crates/core", "src-tauri"] [workspace.package] edition = "2021" -version = "2.0.0" +version = "2.0.1" license = "MIT" repository = "https://github.com/areyoubugcoder/mpider" diff --git a/README.md b/README.md index c472d3d..af6f8b9 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ [项目官网](https://areyoubugcoder.github.io/mpider/) · [下载客户端](https://github.com/areyoubugcoder/mpider/releases/latest) · [全部版本](https://github.com/areyoubugcoder/mpider/releases) · [使用说明](docs/windows-standalone.md) · [常见问题](docs/faq.md) · [反馈问题](https://github.com/areyoubugcoder/mpider/issues/new/choose) · [演示视频](https://xhslink.cn/o/8c23uHExcT6) · [进群交流](https://qm.qq.com/q/s4JavFxA3e) -> **v2.0.0 · Windows x64 正式开源版**。当前仅验证微信 **4.1.15.56 x64** 的模块与函数指纹；微信升级后需要重新适配。macOS、Linux 不支持本次原生采集。 +> **v2.0.1 · Windows x64 正式开源版**。当前仅验证微信 **4.1.15.56 x64** 的模块与函数指纹；微信升级后需要重新适配。macOS、Linux 不支持本次原生采集。 ## 什么是账号池？有什么风险？ @@ -75,7 +75,7 @@ flowchart TD | 功能 | 使用方式 | | --- | --- | -| 推送监测 | 后台发现本机微信中的文章链接，通过网页校验后自动去重入库，详见[使用说明](docs/push-monitor.md) | +| 推送监测 | 后台发现本机微信中的文章链接，独立处理已保存链接并去重入库，详见[使用说明](docs/push-monitor.md) | | 主页识别 | 开启后在微信手动打开公众号主页，登记公众号与会话；识别本身不入库文章 | | 同步最新文章 | 每个公众号单独获取首页，去重入库，不改变历史检查点 | | 采集历史文章 | 按追加篇数或截止日期创建任务，支持暂停和检查点续采 | @@ -97,9 +97,9 @@ flowchart TD | 下载文件 | 用途 | | --- | --- | -| `MPider_2.0.0_x64-setup.exe` | 推荐，NSIS 安装版，支持应用内更新 | -| `MPider_2.0.0_x64_zh-CN.msi` | 中文 MSI 安装版 | -| `MPider-v2.0.0-windows-x64.zip` | 便携版，完整解压后运行 `MPider.exe`，保留旁边的 `runtime/` | +| `MPider_2.0.1_x64-setup.exe` | 推荐，NSIS 安装版，支持应用内更新 | +| `MPider_2.0.1_x64_zh-CN.msi` | 中文 MSI 安装版 | +| `MPider-v2.0.1-windows-x64.zip` | 便携版，完整解压后运行 `MPider.exe`，保留旁边的 `runtime/` | | `SHA256SUMS` | 核对下载文件的 SHA-256 | 三个客户端包均携带 Python **3.12.10**、Frida **17.23.3**、cryptography **50.0.2** 及所需许可，无需另外安装 Python、Node.js、pnpm 或 Rust。微信和 WebView2 需要自行安装。不要只复制便携包内的 EXE。 @@ -107,7 +107,7 @@ flowchart TD 安装包附带应用更新签名，用于 MPider 验证更新文件；它不等同于 Windows 开发者代码签名，系统仍可能提示发布者未知。可先核对附件摘要： ```powershell -Get-FileHash .\MPider_2.0.0_x64-setup.exe -Algorithm SHA256 +Get-FileHash .\MPider_2.0.1_x64-setup.exe -Algorithm SHA256 ``` ## 开始采集 diff --git a/crates/core/examples/push_monitor.rs b/crates/core/examples/push_monitor.rs index b7f5eee..2bf522e 100644 --- a/crates/core/examples/push_monitor.rs +++ b/crates/core/examples/push_monitor.rs @@ -38,6 +38,8 @@ async fn main() -> Result<()> { store.wb_set_policy(policy)?; let id = store.push_start(Config::default())?; let runner = tokio::spawn(push_monitor::run(store.clone(), id)); + let queue_id = store.push_queue_start()?; + let queue = tokio::spawn(push_monitor::run_queue(store.clone(), queue_id)); let until = tokio::time::Instant::now() + std::time::Duration::from_secs(seconds); loop { tokio::time::sleep(std::time::Duration::from_secs(2)).await; @@ -52,7 +54,9 @@ async fn main() -> Result<()> { } } store.push_end(id, "stopped", None)?; + store.push_queue_end(queue_id, "paused", None)?; tokio::time::timeout(std::time::Duration::from_secs(30), runner).await??; + tokio::time::timeout(std::time::Duration::from_secs(30), queue).await??; Ok(()) } } diff --git a/crates/core/src/store.rs b/crates/core/src/store.rs index 226dafa..5387a3e 100644 --- a/crates/core/src/store.rs +++ b/crates/core/src/store.rs @@ -378,6 +378,7 @@ impl Store { conn.execute_batch(native::SCHEMA)?; conn.execute_batch(workbench::SCHEMA)?; conn.execute_batch(push_monitor::SCHEMA)?; + push_monitor::migrate(&conn)?; // 系统日志迁移必须确认成功，禁止因磁盘或权限失败而静默丢失日志字段。 let has_context = conn .prepare("PRAGMA table_info(app_log)")? diff --git a/crates/core/src/store/push_monitor.rs b/crates/core/src/store/push_monitor.rs index 8164606..a9275ec 100644 --- a/crates/core/src/store/push_monitor.rs +++ b/crates/core/src/store/push_monitor.rs @@ -1,7 +1,7 @@ //! 推送候选、事件与文章提交同事务；运行世代挡住停止/重启后的晚到结果。 use super::*; use crate::workbench::push_monitor::{ - Candidate, Config, EventCursor, EventPage, Metadata, MonitorEvent, Run, Scan, Status, + Candidate, Config, EventCursor, EventPage, Metadata, MonitorEvent, QueueRun, Run, Scan, Status, MAX_CANDIDATES, }; use anyhow::{ensure, Context}; @@ -15,13 +15,30 @@ CREATE TABLE IF NOT EXISTS push_candidates( first_seen_at REAL NOT NULL,state TEXT NOT NULL DEFAULT 'pending',attempts INTEGER NOT NULL DEFAULT 0, retry_at REAL NOT NULL DEFAULT 0,finished_at REAL,biz TEXT NOT NULL DEFAULT '', nickname TEXT NOT NULL DEFAULT '',title TEXT NOT NULL DEFAULT '',published_at INTEGER, - article_id INTEGER,reason TEXT + article_id INTEGER,reason TEXT,config TEXT ); CREATE INDEX IF NOT EXISTS push_pending ON push_candidates(state,retry_at,id); CREATE INDEX IF NOT EXISTS push_discovered_time ON push_candidates(first_seen_at DESC,id DESC); CREATE INDEX IF NOT EXISTS push_event_time ON push_candidates(COALESCE(finished_at,first_seen_at) DESC,id DESC); CREATE INDEX IF NOT EXISTS push_request_budget ON wb_requests(json_extract(document,'$.purpose'),json_extract(document,'$.started_at')); "#; +pub(super) fn migrate(conn: &Connection) -> Result<()> { + let has_config = conn + .prepare("PRAGMA table_info(push_candidates)")? + .query_map([], |r| r.get::<_, String>(1))? + .collect::>>()? + .iter() + .any(|name| name == "config"); + if !has_config { + conn.execute("ALTER TABLE push_candidates ADD COLUMN config TEXT", [])?; + } + // 旧候选采用升级前保存的范围；之后重新开启扫描不再改变它们的筛选条件。 + conn.execute( + "UPDATE push_candidates SET config=COALESCE((SELECT json_extract(document,'$.config') FROM push_monitor WHERE id=1),'{}') WHERE config IS NULL", + [], + )?; + Ok(()) +} fn run_from(conn: &Connection) -> Result { let text: Option = conn .query_row("SELECT document FROM push_monitor WHERE id=1", [], |r| { @@ -53,6 +70,10 @@ fn active(tx: &Transaction<'_>, id: i64) -> Result { let r = run_from(tx)?; Ok(r.active && r.id == id) } +fn queue_active(tx: &Transaction<'_>, id: i64) -> Result { + let r = run_from(tx)?; + Ok(r.queue.active && r.queue.id == id) +} fn terminal( tx: &Transaction<'_>, candidate: i64, @@ -74,6 +95,10 @@ impl Store { let r = self.push_run()?; Ok(r.active && r.id == id) } + pub fn push_queue_is_active(&self, id: i64) -> Result { + let r = self.push_run()?; + Ok(r.queue.active && r.queue.id == id) + } /// 此底层入口也供离线测试使用；桌面 IPC 在进入前检查 Windows 平台。 pub fn push_start(&self, config: Config) -> Result { config.validate()?; @@ -87,20 +112,16 @@ impl Store { phase: "starting".into(), config, started_at: now(), + queue: old.queue, ..Default::default() }; - // 上次停止时正在处理的候选重新排队；终态和历史请求预算保留。 - tx.execute( - "UPDATE push_candidates SET state='pending' WHERE state='fetching'", - [], - )?; tx.execute("DELETE FROM push_candidates WHERE state NOT IN ('pending','fetching') AND finished_at Result<()> { - let id = self.push_run()?.id; - self.push_end( - id, - "stopped", - Some("应用已重启；待处理链接已保留，请手动开启推送监测"), - ) + let run = self.push_run()?; + self.push_end(run.id, "stopped", Some("应用已重启；新链接扫描已停止"))?; + self.push_queue_end( + run.queue.id, + "paused", + Some("应用已重启；待处理链接已保留，可独立开始处理"), + )?; + // 兼容旧版只有扫描开关时遗留的正在处理状态。 + self.conn().execute( + "UPDATE push_candidates SET state='pending' WHERE state='fetching'", + [], + )?; + Ok(()) + } + pub fn push_queue_start(&self) -> Result { + let mut conn = self.conn(); + let tx = conn.transaction()?; + let mut run = run_from(&tx)?; + ensure!(!run.queue.active, "待处理链接已经在处理中"); + run.queue = QueueRun { + id: run.queue.id.checked_add(1).context("处理运行编号已耗尽")?, + active: true, + phase: "running".into(), + ..Default::default() + }; + tx.execute( + "UPDATE push_candidates SET state='pending' WHERE state='fetching'", + [], + )?; + save(&tx, &run)?; + system_logs::write( + &tx, + "push_monitor", + "info", + "待处理链接已开始处理", + json!({"queue_run_id":run.queue.id}), + )?; + tx.commit()?; + Ok(run.queue.id) + } + pub fn push_queue_end(&self, id: i64, phase: &str, reason: Option<&str>) -> Result<()> { + ensure!(["paused", "failed"].contains(&phase), "处理状态无效"); + let mut conn = self.conn(); + let tx = conn.transaction()?; + let mut run = run_from(&tx)?; + if !run.queue.active || run.queue.id != id { + return Ok(()); + } + run.queue.active = false; + run.queue.phase = phase.into(); + run.queue.reason = reason.map(str::to_owned); + run.queue.next_fetch_at = 0.; + tx.execute( + "UPDATE push_candidates SET state='pending' WHERE state='fetching'", + [], + )?; + save(&tx, &run)?; + system_logs::write( + &tx, + "push_monitor", + if phase == "failed" { "error" } else { "info" }, + "待处理链接已暂停，扫描状态不变", + json!({"queue_run_id":id,"state":phase,"reason":reason}), + )?; + tx.commit()?; + Ok(()) } pub fn push_scanned(&self, id: i64, scan: Scan, ts: f64) -> Result<()> { let mut conn = self.conn(); @@ -199,8 +279,8 @@ impl Store { } } else if remaining > 0 { tx.execute( - "INSERT INTO push_candidates(key,url,first_seen_at,biz) VALUES(?,?,?,?)", - params![link.key, link.url, ts, link.biz.as_deref().unwrap_or("")], + "INSERT INTO push_candidates(key,url,first_seen_at,biz,config) VALUES(?,?,?,?,?)", + params![link.key, link.url, ts, link.biz.as_deref().unwrap_or(""), serde_json::to_string(&run.config)?], )?; remaining -= 1; } @@ -230,7 +310,7 @@ impl Store { pub fn push_next(&self, id: i64, ts: f64) -> Result> { let conn = self.conn(); let run = run_from(&conn)?; - if !run.active || run.id != id || ts < run.next_fetch_at { + if !run.queue.active || run.queue.id != id || ts < run.queue.next_fetch_at { return Ok(None); } Ok(conn.query_row("SELECT id,url,attempts,first_seen_at,key FROM push_candidates WHERE state='pending' AND retry_at<=? ORDER BY COALESCE(reason IN ('redirect_pending','link_updated'),0) DESC,(attempts=0) DESC,first_seen_at DESC,id DESC LIMIT 1",[ts],|r|Ok(Candidate{id:r.get(0)?,url:r.get(1)?,attempts:r.get(2)?,first_seen_at:r.get(3)?,key:r.get(4)?})).optional()?) @@ -238,7 +318,7 @@ impl Store { pub fn push_claim(&self, id: i64, candidate: i64) -> Result { let mut conn = self.conn(); let tx = conn.transaction()?; - if !active(&tx, id)? { + if !queue_active(&tx, id)? { return Ok(false); } let limit = request_limit(&tx)?; @@ -247,7 +327,8 @@ impl Store { tx.execute("UPDATE push_candidates SET state='retry_limit',reason='retry_limit',finished_at=? WHERE id=? AND state='pending' AND attempts>=?",params![now(),candidate,limit])?; } let mut run = run_from(&tx)?; - run.next_fetch_at = 0.; + run.queue.next_fetch_at = 0.; + run.queue.reason = None; save(&tx, &run)?; tx.commit()?; Ok(changed == 1) @@ -255,12 +336,12 @@ impl Store { pub fn push_deferred(&self, id: i64, until: f64, reason: Option<&str>) -> Result<()> { let mut conn = self.conn(); let tx = conn.transaction()?; - if !active(&tx, id)? { + if !queue_active(&tx, id)? { return Ok(()); } let mut run = run_from(&tx)?; - run.next_fetch_at = until; - run.reason = reason.map(str::to_owned); + run.queue.next_fetch_at = until; + run.queue.reason = reason.map(str::to_owned); save(&tx, &run)?; tx.commit()?; Ok(()) @@ -275,7 +356,7 @@ impl Store { ) -> Result<()> { let mut conn = self.conn(); let tx = conn.transaction()?; - if !active(&tx, id)? { + if !queue_active(&tx, id)? { return Ok(()); } let attempts: u32 = tx.query_row( @@ -303,7 +384,7 @@ impl Store { crate::workbench::push_monitor::normalize(url).context("跳转目标不是有效的微信文章")?; let mut conn = self.conn(); let tx = conn.transaction()?; - if !active(&tx, id)? { + if !queue_active(&tx, id)? { return Ok(()); } let attempts: u32 = tx.query_row( @@ -328,15 +409,15 @@ impl Store { pub fn push_import(&self, id: i64, candidate: i64, meta: &Metadata, ts: f64) -> Result<()> { let mut conn = self.conn(); let tx = conn.transaction()?; - if !active(&tx, id)? { + if !queue_active(&tx, id)? { return Ok(()); } - let run = run_from(&tx)?; - let (first, key): (f64, String) = tx.query_row( - "SELECT first_seen_at,key FROM push_candidates WHERE id=? AND state='fetching'", + let (first, key, config): (f64, String, String) = tx.query_row( + "SELECT first_seen_at,key,COALESCE(config,'{}') FROM push_candidates WHERE id=? AND state='fetching'", [candidate], - |r| Ok((r.get(0)?, r.get(1)?)), + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), )?; + let config: Config = serde_json::from_str(&config)?; ensure!( !key.starts_with("article:") || key == meta.key(), "跳转后的文章身份不一致" @@ -351,11 +432,11 @@ impl Store { candidate ], )?; - let ignored = if !run.config.bizzes.is_empty() && !run.config.bizzes.contains(&meta.biz) { + let ignored = if !config.bizzes.is_empty() && !config.bizzes.contains(&meta.biz) { Some("不在监测范围") } else if meta.published_at as f64 > ts + 1800. { Some("发布时间无效") - } else if (meta.published_at as f64) < first - run.config.recent_hours as f64 * 3600. { + } else if (meta.published_at as f64) < first - config.recent_hours as f64 * 3600. { Some("超出文章时间窗") } else { None @@ -411,7 +492,7 @@ impl Store { "push_monitor", "info", "监测发现的文章已保存", - json!({"monitor_run_id":id,"article_id":article_id,"biz":meta.biz,"new_articles":i64::from(existing.is_none())}), + json!({"queue_run_id":id,"article_id":article_id,"biz":meta.biz,"new_articles":i64::from(existing.is_none())}), )?; tx.commit()?; Ok(()) diff --git a/crates/core/src/store/workbench.rs b/crates/core/src/store/workbench.rs index beab920..02f9700 100644 --- a/crates/core/src/store/workbench.rs +++ b/crates/core/src/store/workbench.rs @@ -321,7 +321,7 @@ impl Store { value.cooldown_until = old.unwrap_or(0.); let text = serde_json::to_string(&value)?; tx.execute("INSERT INTO wb_policy VALUES(1,?) ON CONFLICT(id) DO UPDATE SET document=excluded.document",[&text])?; - tx.execute("UPDATE push_monitor SET document=json_set(document,'$.next_fetch_at',0,'$.reason',NULL) WHERE id=1 AND json_extract(document,'$.active')=1 AND COALESCE(json_extract(document,'$.next_fetch_at'),0)>0",[])?; + tx.execute("UPDATE push_monitor SET document=json_set(document,'$.queue.next_fetch_at',0,'$.queue.reason',NULL) WHERE id=1 AND json_extract(document,'$.queue.active')=1 AND COALESCE(json_extract(document,'$.queue.next_fetch_at'),0)>0",[])?; tx.execute( "INSERT INTO wb_policy_history VALUES(?,?,?)", params![value.version, now(), text], @@ -1568,7 +1568,10 @@ impl Store { r.get(0) })?; let run: crate::workbench::push_monitor::Run = serde_json::from_str(&text)?; - ensure!(run.active && run.id == owner, "推送监测已停止"); + ensure!( + run.queue.active && run.queue.id == owner, + "待处理链接已暂停" + ); } let article = ["body", crate::workbench::push_monitor::CATEGORY].contains(&category); // 包含旧版 body/push_metadata 和新版 webpage，历史请求不用重写。 diff --git a/crates/core/src/workbench/push_monitor.rs b/crates/core/src/workbench/push_monitor.rs index b6f59d5..fc0d48d 100644 --- a/crates/core/src/workbench/push_monitor.rs +++ b/crates/core/src/workbench/push_monitor.rs @@ -70,12 +70,22 @@ pub struct Run { pub started_at: f64, pub last_scan_at: f64, pub next_scan_at: f64, - pub next_fetch_at: f64, + pub queue: QueueRun, pub processes: usize, pub read_bytes: u64, pub rounds: u64, pub reason: Option, } +/// 解析队列有独立运行世代；扫描开关不能取消请求或接受旧处理器的晚到结果。 +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +#[serde(default)] +pub struct QueueRun { + pub id: i64, + pub active: bool, + pub phase: String, + pub next_fetch_at: f64, + pub reason: Option, +} #[derive(Clone, Debug, Serialize)] pub struct MonitorEvent { pub id: i64, @@ -471,23 +481,12 @@ pub async fn run(store: Arc, id: i64) { } pub async fn run_with_scanner(store: Arc, id: i64, scanner: S) -> Result<()> { - run_with_scanner_at(store, id, scanner, None).await -} - -// 测试仅覆盖 HTTP 目的地；生产始终使用已验证的文章地址。 -async fn run_with_scanner_at( - store: Arc, - id: i64, - scanner: S, - endpoint: Option<&str>, -) -> Result<()> { static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); let _guard = LOCK.lock().await; - let client = client()?; let cancel = Arc::new(AtomicBool::new(false)); let mut scanner = Some(scanner); let mut task: Option)>> = None; - // 扫描在后台进行；已有队列不必等待一轮几十秒的内存扫描结束。 + // 扫描在后台进行，停止时可及时发出取消信号。 let result = async { while store.push_is_active(id)? { if task.as_ref().is_some_and(|task| task.is_finished()) { @@ -503,13 +502,6 @@ async fn run_with_scanner_at( (current, result) })); } - if let Some(candidate) = store.push_next(id, now())? { - if let Some(endpoint) = endpoint { - process_at(&store, &client, id, candidate, endpoint).await?; - } else { - process(&store, &client, id, candidate).await?; - } - } tokio::time::sleep(Duration::from_millis(100)).await; } Ok(()) @@ -523,6 +515,35 @@ async fn run_with_scanner_at( result } +/// 显式处理已保存链接，不创建扫描器，也不要求微信进程存在。 +pub async fn run_queue(store: Arc, id: i64) { + if let Err(error) = run_queue_at(store.clone(), id, None).await { + let _ = store.push_queue_end( + id, + "failed", + Some(&crate::store::system_logs::safe_text(&error.to_string())), + ); + } +} + +// 测试可注入本机 HTTP fixture；生产使用已验证候选的文章地址。 +async fn run_queue_at(store: Arc, id: i64, endpoint: Option<&str>) -> Result<()> { + static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + let _guard = LOCK.lock().await; + let client = client()?; + while store.push_queue_is_active(id)? { + if let Some(candidate) = store.push_next(id, now())? { + if let Some(endpoint) = endpoint { + process_at(&store, &client, id, candidate, endpoint).await?; + } else { + process(&store, &client, id, candidate).await?; + } + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + Ok(()) +} + async fn process( store: &Store, client: &reqwest::Client, @@ -564,7 +585,7 @@ async fn process_at( tokio::select! { result = &mut response => break result, _ = tokio::time::sleep(Duration::from_millis(200)) => { - if !store.push_is_active(id)? { + if !store.push_queue_is_active(id)? { completion.finish("cancelled", serde_json::json!({}), now())?; return Ok(()); } diff --git a/crates/core/src/workbench/push_monitor/tests.rs b/crates/core/src/workbench/push_monitor/tests.rs index 04e88fa..95e3826 100644 --- a/crates/core/src/workbench/push_monitor/tests.rs +++ b/crates/core/src/workbench/push_monitor/tests.rs @@ -20,9 +20,16 @@ var ct = "{}"; now() as i64 ) } +// 既有解析回归显式开启两个独立运行；首次运行编号恰好一致。 +fn start_both(store: &Store, config: Config) -> i64 { + let scan_id = store.push_start(config).unwrap(); + let queue_id = store.push_queue_start().unwrap(); + assert_eq!(scan_id, queue_id); + queue_id +} fn setup(config: Config) -> (Store, i64, i64) { let store = Store::open_in_memory().unwrap(); - let id = store.push_start(config).unwrap(); + let id = start_both(&store, config); store .push_scanned( id, @@ -75,7 +82,7 @@ fn long_checksum_survives_discovery_and_requeues_failed_candidate_without_resett assert_eq!(discover(bytes), vec![signed.clone()]); } let store = Store::open_in_memory().unwrap(); - let id = store.push_start(Config::default()).unwrap(); + let id = start_both(&store, Config::default()); let mut unsigned = link("FAKE_A"); unsigned.url = unsigned.url.split("&chksm=").next().unwrap().to_owned(); store @@ -147,7 +154,7 @@ async fn queued_webpage_is_fetched_while_memory_scan_is_still_running() { } } let store = Store::open_in_memory().unwrap(); - let id = store.push_start(Config::default()).unwrap(); + let id = start_both(&store, Config::default()); store .push_scanned( id, @@ -161,10 +168,13 @@ async fn queued_webpage_is_fetched_while_memory_scan_is_still_running() { let store = Arc::new(store); let entered = Arc::new(AtomicBool::new(false)); let (endpoint, server) = fixture(200, html("FAKE_A")).await; + let scan_task = tokio::spawn(run_with_scanner( + store.clone(), + id, + WaitingScanner(entered.clone()), + )); let clone = store.clone(); - let scanner = WaitingScanner(entered.clone()); - let task = - tokio::spawn(async move { run_with_scanner_at(clone, id, scanner, Some(&endpoint)).await }); + let task = tokio::spawn(async move { run_queue_at(clone, id, Some(&endpoint)).await }); tokio::time::timeout(Duration::from_secs(3), server) .await .unwrap() @@ -178,6 +188,13 @@ async fn queued_webpage_is_fetched_while_memory_scan_is_still_running() { .unwrap(); assert_eq!(store.push_run().unwrap().rounds, 1); store.push_end(id, "stopped", None).unwrap(); + assert!(store.push_queue_is_active(id).unwrap()); + store.push_queue_end(id, "paused", None).unwrap(); + tokio::time::timeout(Duration::from_secs(3), scan_task) + .await + .unwrap() + .unwrap() + .unwrap(); tokio::time::timeout(Duration::from_secs(3), task) .await .unwrap() @@ -233,7 +250,7 @@ fn stop_restart_preserves_queue_and_rejects_late_results() { let meta = metadata(&link("FAKE_A"), &html("FAKE_A")).unwrap(); store.push_import(id, candidate, &meta, now()).unwrap(); assert!(store.list_accounts().unwrap().is_empty()); - let next = store.push_start(Config::default()).unwrap(); + let next = store.push_queue_start().unwrap(); assert_ne!(next, id); assert_eq!(store.push_next(next, now()).unwrap().unwrap().id, candidate); store.push_import(id, candidate, &meta, now()).unwrap(); @@ -251,7 +268,7 @@ fn event_article_and_account_commit_atomically_and_never_touch_sessions() { (now() * 1000000.) as u64 )); let store = Store::open(&path).unwrap(); - let id = store.push_start(Config::default()).unwrap(); + let id = start_both(&store, Config::default()); store .push_scanned( id, @@ -383,8 +400,8 @@ fn article_daily_budget_survives_runs_and_is_shared_with_downloads() { .unwrap() .request_id .is_some()); - store.push_end(id, "stopped", None).unwrap(); - let next = store.push_start(Config::default()).unwrap(); + store.push_queue_end(id, "paused", None).unwrap(); + let next = store.push_queue_start().unwrap(); for category in [CATEGORY, "body"] { let purpose = if category == CATEGORY { PURPOSE @@ -455,7 +472,7 @@ fn article_interval_is_shared_and_live_policy_change_releases_waiting_monitor() policy.body_gap_ms = 5000; policy.body_daily_budget = 3; store.wb_set_policy(policy).unwrap(); - assert_eq!(store.push_run().unwrap().next_fetch_at, 0.); + assert_eq!(store.push_run().unwrap().queue.next_fetch_at, 0.); assert!(store .wb_request_permit(CATEGORY, PURPOSE, id, "FAKE_A", ts + 38.) .unwrap() @@ -556,7 +573,7 @@ async fn redirect_waits_for_daily_article_budget_after_reopen_without_hidden_htt .unwrap(); server.await.unwrap(); store.push_restart().unwrap(); - let next_run = store.push_start(Config::default()).unwrap(); + let next_run = store.push_queue_start().unwrap(); let candidate = store.push_next(next_run, now()).unwrap().unwrap(); process_at( &store, @@ -569,7 +586,7 @@ async fn redirect_waits_for_daily_article_budget_after_reopen_without_hidden_htt .unwrap(); assert_eq!(store.push_status().unwrap().requests_24h, 1); assert_eq!(store.push_status().unwrap().pending, 1); - assert!(store.push_run().unwrap().next_fetch_at > now() + 86300.); + assert!(store.push_run().unwrap().queue.next_fetch_at > now() + 86300.); } #[tokio::test] @@ -745,7 +762,7 @@ async fn stop_during_http_does_not_commit_late_reply() { process_at(&clone, &client().unwrap(), id, candidate, &endpoint).await }); let (socket, _) = listener.accept().await.unwrap(); - store.push_end(id, "stopped", None).unwrap(); + store.push_queue_end(id, "paused", None).unwrap(); tokio::time::timeout(Duration::from_secs(3), task) .await .unwrap() @@ -757,7 +774,10 @@ async fn stop_during_http_does_not_commit_late_reply() { let requests = store.wb_snapshot().unwrap().requests; assert_eq!(requests[0].outcome, "cancelled"); assert_eq!(requests[0].owner_kind, "push_monitor"); - assert_eq!(json!(store.push_status().unwrap())["run"]["active"], false); + assert_eq!( + json!(store.push_status().unwrap())["run"]["queue"]["active"], + false + ); } #[test] @@ -770,7 +790,7 @@ fn persisted_queue_and_request_budget_survive_database_reopen() { { let store = Store::open(&path).unwrap(); article_budget(&store, 1); - let id = store.push_start(Config::default()).unwrap(); + let id = start_both(&store, Config::default()); store .push_scanned( id, @@ -791,7 +811,8 @@ fn persisted_queue_and_request_budget_survive_database_reopen() { let store = Store::open(&path).unwrap(); store.push_restart().unwrap(); assert!(!store.push_status().unwrap().run.active); - let id = store.push_start(Config::default()).unwrap(); + assert!(!store.push_status().unwrap().run.queue.active); + let id = store.push_queue_start().unwrap(); assert_eq!(store.push_next(id, now()).unwrap().unwrap().attempts, 1); assert!(store .wb_request_permit(CATEGORY, PURPOSE, id, "FAKE_A", now()) @@ -864,7 +885,7 @@ async fn scanner_stop_is_cooperative_and_partial_results_are_discarded() { #[test] fn recent_discoveries_are_hour_bounded_and_keyset_pages_do_not_skip_or_repeat() { let store = Store::open_in_memory().unwrap(); - let id = store.push_start(Config::default()).unwrap(); + let id = start_both(&store, Config::default()); let ts = now(); store .push_scanned( @@ -1013,3 +1034,247 @@ fn saving_policy_preserves_stopped_monitor_failure_reason() { Some("扫描器读取失败") ); } + +#[test] +fn scanning_and_processing_have_independent_generations_and_controls() { + let store = Store::open_in_memory().unwrap(); + let scan = store.push_start(Config::default()).unwrap(); + store + .push_scanned( + scan, + Scan { + links: vec![link("FAKE_A")], + ..Default::default() + }, + now(), + ) + .unwrap(); + assert!(!store.push_run().unwrap().queue.active); + assert!(store.push_next(scan, now()).unwrap().is_none()); + assert!(store + .wb_request_permit(CATEGORY, PURPOSE, scan, "FAKE_A", now()) + .is_err()); + store.push_end(scan, "stopped", None).unwrap(); + let scan = store.push_start(Config::default()).unwrap(); + let queue = store.push_queue_start().unwrap(); + assert_ne!(scan, queue); + assert!(store.push_queue_start().is_err()); + let candidate = store.push_next(queue, now()).unwrap().unwrap(); + assert!(store.push_claim(queue, candidate.id).unwrap()); + store.push_queue_end(queue, "paused", None).unwrap(); + assert!(store.push_is_active(scan).unwrap()); + store + .push_scanned( + scan, + Scan { + links: vec![link("FAKE_B")], + ..Default::default() + }, + now(), + ) + .unwrap(); + assert_eq!(store.push_status().unwrap().pending, 2); + let next = store.push_queue_start().unwrap(); + store + .push_import( + queue, + candidate.id, + &metadata(&link("FAKE_A"), &html("FAKE_A")).unwrap(), + now(), + ) + .unwrap(); + store + .push_queue_end(queue, "failed", Some("旧处理器晚到错误")) + .unwrap(); + assert!(store.push_queue_is_active(next).unwrap()); + assert_eq!(store.push_status().unwrap().imported, 0); + store.push_end(scan, "stopped", None).unwrap(); + assert!(store.push_queue_is_active(next).unwrap()); + assert!(store.push_claim(next, candidate.id).unwrap()); + store + .push_import( + next, + candidate.id, + &metadata(&link("FAKE_A"), &html("FAKE_A")).unwrap(), + now(), + ) + .unwrap(); + assert_eq!(store.push_status().unwrap().imported, 1); +} + +#[tokio::test] +async fn queue_worker_processes_saved_links_with_scanning_stopped() { + let store = Arc::new(Store::open_in_memory().unwrap()); + let scan = store.push_start(Config::default()).unwrap(); + store + .push_scanned( + scan, + Scan { + links: vec![link("FAKE_A")], + ..Default::default() + }, + now(), + ) + .unwrap(); + store.push_end(scan, "stopped", None).unwrap(); + let queue = store.push_queue_start().unwrap(); + let (endpoint, server) = fixture(200, html("FAKE_A")).await; + let clone = store.clone(); + let worker = tokio::spawn(async move { run_queue_at(clone, queue, Some(&endpoint)).await }); + tokio::time::timeout(Duration::from_secs(5), async { + server.await.unwrap(); + while store.push_status().unwrap().imported != 1 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + let status = store.push_status().unwrap(); + assert!(!status.run.active); + assert!(status.run.queue.active); + assert_eq!(status.run.rounds, 1); + assert_eq!(status.pending, 0); + assert_eq!(status.requests_24h, 1); + store.push_queue_end(queue, "paused", None).unwrap(); + tokio::time::timeout(Duration::from_secs(3), worker) + .await + .unwrap() + .unwrap() + .unwrap(); +} + +#[tokio::test] +async fn stopping_scan_keeps_inflight_queue_request_and_its_result() { + let (store, id, _) = setup(Config::default()); + let store = Arc::new(store); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}/s", listener.local_addr().unwrap()); + let candidate = store.push_next(id, now()).unwrap().unwrap(); + let clone = store.clone(); + let worker = tokio::spawn(async move { + process_at(&clone, &client().unwrap(), id, candidate, &endpoint).await + }); + let (mut socket, _) = tokio::time::timeout(Duration::from_secs(3), listener.accept()) + .await + .unwrap() + .unwrap(); + let mut request = [0u8; 4096]; + let received = socket.read(&mut request).await.unwrap(); + assert!( + received > 0, + "queue request must arrive before stopping scan" + ); + store.push_end(id, "stopped", None).unwrap(); + let body = html("FAKE_A"); + socket + .write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ) + .as_bytes(), + ) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(3), worker) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(store.push_status().unwrap().imported, 1); + assert!(store.push_run().unwrap().queue.active); + assert!(!store.push_run().unwrap().active); +} + +#[test] +fn changing_scan_scope_does_not_reclassify_saved_links_or_clear_queue_wait() { + let config = Config { + bizzes: vec!["FAKE_A".into()], + recent_hours: 168, + ..Default::default() + }; + let (store, queue, candidate) = setup(config); + store + .push_deferred(queue, now() + 3600., Some("测试额度等待")) + .unwrap(); + store + .push_end(store.push_run().unwrap().id, "stopped", None) + .unwrap(); + store + .push_start(Config { + bizzes: vec!["FAKE_B".into()], + recent_hours: 1, + ..Default::default() + }) + .unwrap(); + let run = store.push_run().unwrap(); + assert!(run.queue.active); + assert!(run.queue.next_fetch_at > now()); + assert_eq!(run.queue.reason.as_deref(), Some("测试额度等待")); + assert!(store.push_claim(queue, candidate).unwrap()); + let mut meta = metadata(&link("FAKE_A"), &html("FAKE_A")).unwrap(); + meta.published_at -= 2 * 86400; + store.push_import(queue, candidate, &meta, now()).unwrap(); + assert_eq!(store.push_status().unwrap().imported, 1); +} + +#[test] +fn legacy_queue_migration_preserves_candidates_attempts_and_discovery_scope() { + let path = std::env::temp_dir().join(format!( + "mpider-push-migrate-{}-{}.db", + std::process::id(), + (now() * 1000000.) as u64 + )); + { + let store = Store::open(&path).unwrap(); + let id = store + .push_start(Config { + bizzes: vec!["FAKE_A".into()], + ..Default::default() + }) + .unwrap(); + store + .push_scanned( + id, + Scan { + links: vec![link("FAKE_A")], + ..Default::default() + }, + now(), + ) + .unwrap(); + let conn = rusqlite::Connection::open(&path).unwrap(); + conn.execute_batch("ALTER TABLE push_candidates DROP COLUMN config; UPDATE push_candidates SET state='fetching',attempts=1; UPDATE push_monitor SET document=json_remove(document,'$.queue');").unwrap(); + } + { + let store = Store::open(&path).unwrap(); + store.push_restart().unwrap(); + assert!(!store.push_run().unwrap().active); + assert!(!store.push_run().unwrap().queue.active); + assert_eq!( + store.push_events(None, 10).unwrap().items[0].outcome, + "pending" + ); + store + .push_start(Config { + bizzes: vec!["FAKE_B".into()], + ..Default::default() + }) + .unwrap(); + let queue = store.push_queue_start().unwrap(); + let candidate = store.push_next(queue, now()).unwrap().unwrap(); + assert_eq!(candidate.attempts, 1); + store.push_claim(queue, candidate.id).unwrap(); + store + .push_import( + queue, + candidate.id, + &metadata(&link("FAKE_A"), &html("FAKE_A")).unwrap(), + now(), + ) + .unwrap(); + assert_eq!(store.push_status().unwrap().imported, 1); + } + let _ = std::fs::remove_file(path); +} diff --git a/docs/app-updates.md b/docs/app-updates.md index d34eacf..4a46f13 100644 --- a/docs/app-updates.md +++ b/docs/app-updates.md @@ -12,7 +12,7 @@ ## 版本与接口 -- 当前版本取 Tauri 项目配置（当前为 `2.0.0`），`package.json`、根 `Cargo.toml` 的工作区版本、`src-tauri/tauri.conf.json` 必须一致。发布标签为 `v<当前项目版本>`，不根据历史 GitHub Release 自动推算或递增。 +- 当前版本取 Tauri 项目配置（当前为 `2.0.1`），`package.json`、根 `Cargo.toml` 的工作区版本、`src-tauri/tauri.conf.json` 必须一致。发布标签为 `v<当前项目版本>`，不根据历史 GitHub Release 自动推算或递增。 - 更新地址固定为 `https://github.com/areyoubugcoder/mpider/releases/latest/download/latest.json`，公开仓库无需在客户端放 GitHub Token。仅更高的稳定语义版本有效，同版、旧版、预发布版均不安装。草稿未公开时不会被客户端下载。 - 静态清单包含 `version`、`pub_date`、`notes` 和 `platforms`；目前仅含 `windows-x86_64`。平台项含 `url` 和 `.sig` 文件的完整内容 `signature`。 - Tauri updater 负责 HTTPS 请求、平台选择与签名验证，版本清单和安装包通过 GitHub Release 附件分发，不增加业务 HTTP 服务或数据库表。 @@ -44,7 +44,7 @@ pnpm tauri signer generate -w ~/.tauri/mpider-updater.key ## GitHub 发布 -1. 核对发布提交与标签；本次项目版本为 `2.0.0`，标签应为 `v2.0.0`。保留历史 Release 与标签，不覆盖仍在服务用户的版本。 +1. 核对发布提交与标签；本次项目版本为 `2.0.1`，标签应为 `v2.0.1`。保留历史 Release 与标签，不覆盖仍在服务用户的版本。 2. 按项目主线/开源发布流程准备发布提交。需要升级版本时，同时修改三个版本文件及锁文件；本功能不会自行修改项目版本。 3. 打标签前在 main 手动运行 Release（`gh workflow run release.yml --ref main`）预演完整构建。预演只留存 artifact，不创建 Release，并写入后续标签可读取的 Rust 缓存。通过后将匹配当前项目版本的 `v*` 标签推送到 GitHub。工作流先核对版本与签名配置，再为 Windows x64 构建；缺少签名配置时标签发布立即失败。 4. Windows 构建成功后整理附件并生成 `latest.json`，创建 Release **预发布草稿**；保留完整便携 ZIP、自检与摘要，同时提供携带相同 Python 运行环境和完整许可证的 NSIS/MSI。Windows 更新默认使用 NSIS `-setup.exe`；MSI 仍提供手动安装。工作流不构建或上传 macOS 产物。 diff --git a/docs/faq.md b/docs/faq.md index 8a10967..fbf0457 100644 --- a/docs/faq.md +++ b/docs/faq.md @@ -36,7 +36,7 @@ ## 推送监测与主页识别有什么区别？ -“推送监测”持续发现微信内存中的文章链接，通过网页校验后自动加入文章库；“主页识别”在你打开公众号主页时登记公众号与采集会话，不自动采集文章。两者独立开启，可同时使用。推送监测的发现时间不等于推送送达时间，也可能发现手动打开的旧文章。详见[推送监测](push-monitor.md)。 +“推送监测”持续发现微信内存中的文章链接并追加到待处理队列；独立开启“开始处理”后，通过网页校验的文章自动加入文章库。停止监测后仍可处理已保存链接；“主页识别”在你打开公众号主页时登记公众号与采集会话，不自动采集文章。两者独立开启，可同时使用。推送监测的发现时间不等于推送送达时间，也可能发现手动打开的旧文章。详见[推送监测](push-monitor.md)。 ## 提示会话过期，是否表示历史游标也失效了？ diff --git a/docs/push-monitor.md b/docs/push-monitor.md index 8a58668..adec1d2 100644 --- a/docs/push-monitor.md +++ b/docs/push-monitor.md @@ -1,12 +1,18 @@ # 推送监测 -在 Windows 上保持微信登录，从左侧进入“推送监测”，选择范围与频率后开启。程序只读扫描微信进程中存在的文章链接，持久化候选，再获取文章网页核对公众号、标题和发布时间。符合条件的文章自动加入现有文章库；下载文章和导出仍使用文章库的入口。 +从左侧进入“推送监测”，分别控制新链接扫描和已保存链接处理： + +- “开启推送监测”：在 Windows 上只读扫描已登录微信的进程，持续发现并保存新链接。仅开启监测不会请求文章网页。 +- “开始处理”：独立处理已经保存的链接，获取文章网页核对公众号、标题和发布时间。无需开启监测或保持微信运行；符合条件的文章自动加入文章库。队列清空后保持等待，后续追加的链接也会自动处理。 + +两个功能可以同时开启，也可以单独使用。下载文章和导出仍使用文章库的入口。 ## 与主页识别的区别 | 操作 | 用途 | 是否自动新增文章 | | --- | --- | --- | -| 推送监测 | 后台发现本机微信内存中的文章链接 | 是，通过网页身份和时间窗校验后入库 | +| 推送监测 | 后台发现本机微信内存中的文章链接 | 否，保存到待处理队列 | +| 处理待处理链接 | 解析已保存链接并校验身份和时间窗 | 是 | | 主页识别 | 开启后手动打开公众号主页，登记公众号和采集会话 | 否 | | 同步最新文章 | 对已登记公众号显式获取首页列表 | 是 | @@ -15,7 +21,7 @@ ## 时间、范围和请求 - 默认每轮扫描结束后等待 10 秒；扫描本身和网页处理也需要时间，不承诺固定送达延迟。 -- 默认处理在首次发现前 24 小时内发布的文章，可设置 1～168 小时。首次开启也会处理内存中已有的近期链接，记录可能包含用户打开的旧文章。 +- 默认处理在首次发现前 24 小时内发布的文章，可设置 1～168 小时。首次开启监测也会发现内存中已有的近期链接，记录可能包含用户打开的旧文章。 - “发现时间”表示软件观察到链接的时间，不是微信推送送达时间。 - 可以监测全部公众号，或按已登记公众号的稳定标识选择范围；支持按名称或 ID 模糊搜索，搜索及加载更多时保留已有选择。短链需先解析网页才能知道所属公众号，可能占用一次请求额度后被过滤。 - 文章请求的间隔、重试和近 24 小时预算统一在“限流配置 → 文章下载”调整。推送页不再提供独立网页间隔或每小时上限。新配置默认文章下载间隔 1 秒、近 24 小时预算 86,400 次；已保存间隔保持原值，旧配置缺少文章预算时补入初始值。滚动额度跨停止、重启保留，跳转、失败和重试同样计数。 @@ -24,13 +30,13 @@ “最近发现”只展示首次发现时间在最近一小时内的记录，首次只加载 20 条，点击一次“加载更多”才再读取下一批 20 条，重新进入页面从首批开始；不再固定截断为 100 条。已过期记录不会因后来解析完成而重新出现在列表中，后台候选与文章不受这一展示窗口影响。 -停止监测会取消正在等待的网页响应，不接受晚到结果。应用重启后监测保持关闭，需要手动重新开启；未完成候选继续排队，不会被吸收为已完成基线。删除公众号或文章后，删除前已发现的迟到结果不会将其重建。 +停止监测只停止扫描新链接，已开启的队列处理继续运行；“暂停处理”只取消正在等待的网页响应、保留未完成链接，不影响扫描。重新开始处理会继续已有队列，暂停前的晚到结果不会写入。应用重启后两个功能均保持关闭，可按需分别开启；未完成链接、请求次数和预算均保留。每条链接使用发现时保存的公众号范围与时间窗，之后修改监测设置不影响已有链接。删除公众号或文章后，删除前已发现的迟到结果不会将其重建。 ## 支持范围 当前实现面向 64 位 Windows 微信的 `Weixin.exe`、`WeChat.exe`、`WeChatAppEx.exe`。主进程每轮扫描，内嵌浏览器按热点及每 30 轮全量兜底扫描。支持 ASCII 与 UTF-16LE 链接，长链保留 `sn` 及 16～256 位十六进制 `chksm`，不会将合法的较长签名丢弃；同一文章补齐签名后，失败候选在剩余次数内重新排队。读取已提交的可写、非代码映像内存区。不注入、不操作微信窗口、不加载第三方监控 EXE。 -只有进入可扫描内存且保留足够久的链接才能被发现。微信退出、URL 释放、内存布局变化、权限不一致或网页不可访问都可能影响结果；不能用它保证完整补齐离线期间的推送或代替历史翻页。macOS 可以查看记录，但不能开启此监测。 +只有进入可扫描内存且保留足够久的链接才能被发现。微信退出、URL 释放、内存布局变化、权限不一致或网页不可访问都可能影响结果；不能用它保证完整补齐离线期间的推送或代替历史翻页。macOS 可以查看记录并处理已保存链接，但不能扫描微信进程。 ## 本地诊断 @@ -45,8 +51,19 @@ cargo run --locked -p mpider-core --example push_monitor -- run .\push-test.db 1 ## 本地接口约定 -Tauri 提供 `workbench_push_status`、`workbench_push_events({before, limit})`、`workbench_push_start({config})` 和 `workbench_push_stop`。监测配置只包含 `scan_seconds`（5～300）、`recent_hours`（1～168）和 `bizzes`（空集合表示全部）；旧版独立请求间隔及每小时上限字段被忽略。文章预算使用限流策略中的 `body_daily_budget`（正整数），请求工作位数量使用 `body_workers`，每个工作位在网页响应结束后按 `body_gap_ms` 等待；首个请求立即发送，与文章下载共享工作位和预算。正在运行时不能重复开启；先停止再调整设置。状态返回平台支持性、运行状态、队列和入库计数、共享的近24小时文章请求数及预算；状态接口不读取发现列表，发现记录仅在分页接口按批次读取，不返回内存片段或候选原始 URL。 +Tauri 提供以下本地命令： + +| 命令 | 行为 | +| --- | --- | +| `workbench_push_status` | 返回 `run`（扫描状态）、`run.queue`（独立处理状态）及队列、入库、请求预算计数 | +| `workbench_push_events({before, limit})` | 分页读取最近发现记录 | +| `workbench_push_start({config})` / `workbench_push_stop` | 仅开启／停止 Windows 新链接扫描 | +| `workbench_push_queue_start` / `workbench_push_queue_stop` | 独立开启／暂停已保存链接处理，不要求 Windows 或微信进程；开启返回处理运行编号 | + +`run.queue` 包含 `id`、`active`、`phase`、`next_fetch_at`、`reason`；请求等待时间和原因属于处理队列，不再覆盖扫描状态。两种运行分别防止重复开启、拒绝各自停止或重启前的晚到结果。已有数据库升级时保留候选、重试次数和升级前保存的筛选范围。 + +监测配置只包含 `scan_seconds`（5～300）、`recent_hours`（1～168）和 `bizzes`（空集合表示全部）；旧版独立请求间隔及每小时上限字段被忽略。文章预算使用限流策略中的 `body_daily_budget`（正整数），请求工作位数量使用 `body_workers`，每个工作位在网页响应结束后按 `body_gap_ms` 等待；首个请求立即发送，与文章下载共享工作位和预算。监测运行时先停止扫描再调整监测设置；期间可以继续处理已有链接。状态返回平台支持性、运行状态、队列和入库计数、共享的近24小时文章请求数及预算；状态接口不读取发现列表，发现记录仅在分页接口按批次读取，不返回内存片段或候选原始 URL。 发现记录分页按 `(first_seen_at, id)` 降序游标继续，`before` 传上一页的 `next_cursor`；末页返回空游标。`limit` 为 1～100，服务端每次读取均限制在最近一小时。状态与分页均不返回候选原始 URL。 -离线回归使用虚构页面和本机 HTTP 服务，覆盖身份不匹配、跨号去重、原子入库、停止期间取消、重启恢复、删除屏障、共享文章间隔、24小时预算、最近一小时游标分页及验证页短暂重试。它们验证程序行为；实际微信版本上的发现能力还需要单独实机验收。 +离线回归使用虚构页面和本机 HTTP 服务，覆盖扫描与处理独立开关、关闭扫描后的独立 HTTP 处理、旧队列迁移、身份不匹配、跨号去重、原子入库、停止期间取消、重启恢复、删除屏障、共享文章间隔、24小时预算、最近一小时游标分页及验证页短暂重试。它们验证程序行为；实际微信版本上的发现能力还需要单独实机验收。 diff --git a/docs/windows-standalone.md b/docs/windows-standalone.md index cf42c3a..46fc82e 100644 --- a/docs/windows-standalone.md +++ b/docs/windows-standalone.md @@ -2,16 +2,16 @@ 当前发行目标为 Windows 10/11 x64 原生工作台。首次登记公众号和更新会话需要已登录、解锁且可交互的微信桌面；保存有效会话后的历史翻页、正文和图片任务仍需电脑与应用运行，但不依赖持续点击微信窗口。不提供无人登录的 Windows 服务或自动唤醒。 -v2.0.0 包含独立监测、按公众号任务、Markdown/HTML 双格式导出和 GitHub 签名更新。下载时请核对 Release 的源码版本、文件摘要与验证范围；离线 CI 不代替真实采集与安装升级验证。 +v2.0.1 包含独立监测、按公众号任务、Markdown/HTML 双格式导出和 GitHub 签名更新。下载时请核对 Release 的源码版本、文件摘要与验证范围；离线 CI 不代替真实采集与安装升级验证。 ## 使用发行包 -推荐安装 `MPider_2.0.0_x64-setup.exe`，也可选择 MSI；安装版携带与便携包相同的 Python 环境和许可，并支持应用内签名更新。以下步骤介绍便携 ZIP 的使用。 +推荐安装 `MPider_2.0.1_x64-setup.exe`，也可选择 MSI；安装版携带与便携包相同的 Python 环境和许可，并支持应用内签名更新。以下步骤介绍便携 ZIP 的使用。 1. 从 [GitHub Releases](https://github.com/areyoubugcoder/mpider/releases) 下载 `MPider-v版本-windows-x64.zip` 和 `SHA256SUMS`，核对来源及摘要： ```powershell - Get-FileHash .\MPider-v2.0.0-windows-x64.zip -Algorithm SHA256 + Get-FileHash .\MPider-v2.0.1-windows-x64.zip -Algorithm SHA256 ``` 2. **完整解压**到可写目录，保留 `MPider.exe` 旁的 `runtime`。不要只复制 EXE。发行包附带 Python 3.12.10、Frida 17.23.3、cryptography 50.0.2 及其依赖，无需安装 Node.js、pnpm、Rust 或系统 Python。 @@ -42,7 +42,7 @@ powershell -NoProfile -ExecutionPolicy Bypass -File tools\build-windows.ps1 -Rel `Bypass` 仅作用于当前 PowerShell 进程，不修改机器策略。脚本遵循已有 `CARGO_TARGET_DIR`，运行 `pnpm tauri build --no-bundle`，把前端与已内嵌的 native worker、JS 和 profiles 编译到 EXE，再附带隔离 Python/Frida 环境，产物位于： -- `dist-windows/MPider-v2.0.0-windows-x64.zip`：完整便携包。 +- `dist-windows/MPider-v2.0.1-windows-x64.zip`：完整便携包。 - `dist-windows/SHA256SUMS`：ZIP 的 SHA-256。 - `dist-windows/verification.json`：离线验收记录。 @@ -57,7 +57,7 @@ powershell -NoProfile -ExecutionPolicy Bypass -File tools\build-windows.ps1 -Rel `release-manifest.json` 列出包内文件摘要、源码 revision 与依赖版本。独立验证命令为： ```powershell -python tools/verify-windows-release.py --directory dist-windows\MPider-v2.0.0-windows-x64 +python tools/verify-windows-release.py --directory dist-windows\MPider-v2.0.1-windows-x64 ``` 该检查只核对文件完整性、x64 PE、实际 Python/依赖版本以及虚构材料 AES-GCM 和当前用户 DPAPI 往返，不启动微信、不读取真实会话、不发送采集请求。**离线检查成功不代表真实采集或长期稳定性通过。**页面交互验收需在交互桌面另行进行，真实数据不得进入公开包。 diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 7cc6d85..ea4fa22 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -20,7 +20,7 @@ import { CursorLab } from "@/views/CursorLab"; import { SystemLogs } from "@/views/SystemLogs"; import { Faq } from "@/views/Faq"; import { PushMonitor } from "@/views/PushMonitor"; -import { pushStatus, pushPhase } from "@/lib/push-monitor"; +import { pushStatus, pushPhase, pushQueuePhase } from "@/lib/push-monitor"; import type { View } from "@/views/types"; export function App() { @@ -60,7 +60,7 @@ export function App() { {!!snapshot.data?.active_queues?.history.task_count && {setTaskKind("history");setView("jobs");}} />} - + ; } diff --git a/frontend/src/lib/push-monitor.ts b/frontend/src/lib/push-monitor.ts index 1f5b48f..b7908df 100644 --- a/frontend/src/lib/push-monitor.ts +++ b/frontend/src/lib/push-monitor.ts @@ -1,11 +1,12 @@ -/** 推送监测与主页识别独立开关；URL 候选与内存内容不经过前端。 */ +/** 新链接扫描、已保存链接处理与主页识别分别控制；原始 URL 不经过前端。 */ import { invoke } from "@tauri-apps/api/core"; export interface PushConfig { scan_seconds: number; recent_hours: number; bizzes: string[] } export interface PushRun { id: number; active: boolean; phase: string; config: PushConfig; started_at: number; - last_scan_at: number; next_scan_at: number; next_fetch_at: number; processes: number; + last_scan_at: number; next_scan_at: number; queue: PushQueueRun; processes: number; read_bytes: number; rounds: number; reason: string | null; } +export interface PushQueueRun { id: number; active: boolean; phase: string; next_fetch_at: number; reason: string | null } export interface PushEvent { id: number; first_seen_at: number; finished_at: number | null; outcome: string; nickname: string; title: string; biz: string; article_id: number | null; @@ -18,6 +19,8 @@ export const pushEvents = (before: PushEventCursor | null) => invoke invoke("workbench_push_status"); export const startPushMonitor = (config: PushConfig) => invoke("workbench_push_start", { config }); export const stopPushMonitor = () => invoke("workbench_push_stop"); +export const startPushQueue = () => invoke("workbench_push_queue_start"); +export const stopPushQueue = () => invoke("workbench_push_queue_stop"); export const defaultPushConfig: PushConfig = { scan_seconds: 10, recent_hours: 24, bizzes: [] }; export const pushOutcome = (value: string) => ({ retry_limit: "已达到文章重试上限", pending: "等待处理", fetching: "正在解析", imported: "新增入库", existing: "文章已存在", ignored: "已过滤", @@ -30,6 +33,12 @@ export function pushPhase(run?: PushRun) { if (!run.active) return run.phase === "failed" ? "运行异常" : "已停止"; if (run.phase === "starting") return "正在启动"; if (run.phase === "waiting_wechat") return "等待微信启动"; - if (run.next_fetch_at > Date.now()/1000) return "监测中 · 文章解析等待中"; return "监测中"; } +export function pushQueuePhase(queue?: PushQueueRun, pending = 0, now = Date.now() / 1000) { + if (!queue) return "读取中"; + if (!queue.active) return queue.phase === "failed" ? "处理异常" : "已暂停"; + if (pending === 0) return "等待待处理链接"; + if (queue.next_fetch_at > now) return "等待请求额度或间隔"; + return "正在处理"; +} diff --git a/frontend/src/views/PushMonitor.tsx b/frontend/src/views/PushMonitor.tsx index 413de0b..9e40d86 100644 --- a/frontend/src/views/PushMonitor.tsx +++ b/frontend/src/views/PushMonitor.tsx @@ -9,7 +9,7 @@ import { Card, CardHeader, CardTitle, CardDescription, CardContent } from "@/com import { Table, TableBody, TableRow, TableHead, TableHeader, TableCell } from "@/components/ui/table"; import { logUiError, sourcePage } from "@/lib/workbench"; import { recentPushEvents } from "@/lib/push-feed"; -import { defaultPushConfig, pushStatus, pushEvents, startPushMonitor, stopPushMonitor, pushPhase, pushOutcome, type PushConfig, type PushEventCursor } from "@/lib/push-monitor"; +import { defaultPushConfig, pushStatus, pushEvents, startPushMonitor, stopPushMonitor, startPushQueue, stopPushQueue, pushPhase, pushQueuePhase, pushOutcome, type PushConfig, type PushEventCursor } from "@/lib/push-monitor"; const date = (value?: number | null) => value ? new Date(value * 1000).toLocaleString() : "—"; export function PushMonitor({ onArticles, onRate }: { onArticles: (biz: string) => void; onRate: () => void }) { @@ -36,8 +36,10 @@ export function PushMonitor({ onArticles, onRate }: { onArticles: (biz: string) const [allAccounts, setAllAccounts] = useState(true); const [dirty, setDirty] = useState(false); const [busy, setBusy] = useState(false); + const [queueBusy, setQueueBusy] = useState(false); const data = status.data; const active = !!data?.run.active; + const processing = !!data?.run.queue.active; useEffect(() => { if (data && !dirty) { setConfig(data.run.config); setAllAccounts(!data.run.config.bizzes.length); } }, [data, dirty]); const change = (value: Partial) => { setDirty(true); setConfig(previous => ({ ...previous, ...value })); }; async function act() { @@ -49,21 +51,34 @@ export function PushMonitor({ onArticles, onRate }: { onArticles: (biz: string) } catch (e) { toast.error(String(e)); void logUiError("action_failed", String(e)).catch(() => {}); } finally { setBusy(false); } } + async function processQueue() { + setQueueBusy(true); + try { + if (processing) await stopPushQueue(); + else await startPushQueue(); + await qc.invalidateQueries({ queryKey: ["push-monitor"] }); + } catch (e) { toast.error(String(e)); void logUiError("action_failed", String(e)).catch(() => {}); } + finally { setQueueBusy(false); } + } const number = (key: "scan_seconds" | "recent_hours", text: string, min: number, max: number) =>
change({ [key]: Number(e.target.value) })} />
; const events = recentPushEvents(feed.data?.pages ?? [], Date.now() / 1000); const options = accounts.data?.pages.flatMap(page => page.items) ?? []; return
-

推送监测

持续发现本机微信中的文章链接，解析后自动加入文章库。

+

推送监测

扫描新链接与处理已保存链接分别控制，解析后的文章自动加入文章库。

{status.error &&

无法读取监测状态：{String(status.error)}

} - {data && !data.supported &&

请在已登录微信的 Windows 电脑开启推送监测。此处可以查看已有记录。

} -
与「主页识别」独立运行，无需为每次扫描打开公众号主页。仅能发现微信内存中存在的链接。
 + {data && !data.supported &&

扫描新链接需在已登录微信的 Windows 电脑开启；此处仍可独立处理已保存的链接和查看记录。

} +
持续扫描微信内存并追加新链接；停止监测不影响下方已开启的链接处理。与「主页识别」独立运行。

{pushPhase(data?.run)}

{data?.run.reason &&

{data.run.reason}

} - {!!data?.run.next_fetch_at && data.run.next_fetch_at > Date.now() / 1000 &&

文章解析不早于 {date(data.run.next_fetch_at)} 继续；内存扫描和待处理队列保留。

} -
{[["新增入库", data?.imported ?? 0], ["待处理链接", data?.pending ?? 0], ["近24小时文章请求", `${data?.requests_24h ?? 0} / ${data?.daily_budget ?? "—"}`], ["微信进程", data?.run.processes ?? 0]].map(([name, value]) =>

{name}

{value}

)}
-

上次扫描：{date(data?.run.last_scan_at)} · {data?.run.rounds ?? 0} 轮 · 最近一轮读取 {((data?.run.read_bytes ?? 0) / 1048576).toFixed(0)} MB。监测记录保留 30 天；文章保存在文章库。

+

上次扫描：{date(data?.run.last_scan_at)} · {data?.run.rounds ?? 0} 轮 · 最近一轮读取 {((data?.run.read_bytes ?? 0) / 1048576).toFixed(0)} MB · 微信进程 {data?.run.processes ?? 0}。监测记录保留 30 天；文章保存在文章库。

+ +
待处理链接独立解析已保存的链接，无需开启监测或保持微信运行。处理开启后，新追加的链接也会自动处理；暂停处理不影响扫描。
 +

{pushQueuePhase(data?.run.queue, data?.pending)}

+ {data?.run.queue.reason &&

{data.run.queue.reason}

} + {processing && !!data?.pending && data.run.queue.next_fetch_at > Date.now() / 1000 &&

文章解析不早于 {date(data.run.queue.next_fetch_at)} 继续。

} +
{[["待处理链接", data?.pending ?? 0], ["新增入库", data?.imported ?? 0], ["近24小时文章请求", `${data?.requests_24h ?? 0} / ${data?.daily_budget ?? "—"}`]].map(([name, value]) =>

{name}

{value}

)}
 - 监测范围与频率开启时保存设置。首次开启也会处理内存中的近期文章，停止和重启保留待处理链接。 + 监测范围与频率开启监测时保存设置，仅影响之后发现的链接；已有链接保留发现时的范围与时间窗。停止和重启均保留待处理链接。
{number("scan_seconds", "每轮扫描后的间隔（秒）", 5, 300)}{number("recent_hours", "发现时回溯文章时间窗（小时）", 1, 168)}
公众号范围 diff --git a/package.json b/package.json index bf3cb6e..66933b1 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "mpider-desktop", "private": true, - "version": "2.0.0", + "version": "2.0.1", "type": "module", "packageManager": "pnpm@10.29.2", "description": "mpider 桌面外壳的前端 + Tauri CLI（前端源码在 frontend/，Tauri 壳在 src-tauri/）", @@ -14,7 +14,7 @@ "preview": "vite preview", "tauri": "tauri", "test": "pnpm test:frontend && pnpm test:native", - "test:frontend": "node --test tools/test_cursor_lab.ts tools/test_home_sync.ts tools/test_pagination.ts tools/test_relative_time.ts tools/test_advertisements.ts tools/test_action_visibility.ts tools/test_export_progress.ts tools/test_release_updater.mjs tools/test_push_feed.ts", + "test:frontend": "node --test tools/test_cursor_lab.ts tools/test_home_sync.ts tools/test_pagination.ts tools/test_relative_time.ts tools/test_advertisements.ts tools/test_action_visibility.ts tools/test_export_progress.ts tools/test_release_updater.mjs tools/test_push_feed.ts tools/test_push_monitor.ts", "test:native": "node --test tools/test_native_bridge_filter.mjs tools/test_session_agent.mjs" }, "dependencies": { diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 75809d6..21fa0f7 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2608,6 +2608,8 @@ pub fn run() { workbench::workbench_push_events, workbench::workbench_push_start, workbench::workbench_push_stop, + workbench::workbench_push_queue_start, + workbench::workbench_push_queue_stop, workbench::workbench_article_body_states, workbench::workbench_source_page, workbench::workbench_task_page, diff --git a/src-tauri/src/workbench.rs b/src-tauri/src/workbench.rs index 9f24715..bf1f1a4 100644 --- a/src-tauri/src/workbench.rs +++ b/src-tauri/src/workbench.rs @@ -53,6 +53,24 @@ pub fn workbench_push_stop(state: State<'_, AppState>) -> Result<(), String> { .map_err(|e| e.to_string()) } #[tauri::command] +pub fn workbench_push_queue_start(state: State<'_, AppState>) -> Result { + root(&state)?; + let id = state.store.push_queue_start().map_err(|e| e.to_string())?; + let store = state.store.clone(); + tauri::async_runtime::spawn(async move { + core::push_monitor::run_queue(store, id).await; + }); + Ok(id) +} +#[tauri::command] +pub fn workbench_push_queue_stop(state: State<'_, AppState>) -> Result<(), String> { + let id = state.store.push_run().map_err(|e| e.to_string())?.queue.id; + state + .store + .push_queue_end(id, "paused", None) + .map_err(|e| e.to_string()) +} +#[tauri::command] pub fn workbench_snapshot(state: State<'_, AppState>) -> Result { state.store.wb_snapshot().map_err(|e| e.to_string()) } diff --git a/src-tauri/tauri.conf.json b/src-tauri/tauri.conf.json index bf773f7..fda5893 100644 --- a/src-tauri/tauri.conf.json +++ b/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "MPider", - "version": "2.0.0", + "version": "2.0.1", "identifier": "io.github.areyoubugcoder.mpider", "build": { "frontendDist": "../frontend/dist", diff --git a/tools/test_push_monitor.ts b/tools/test_push_monitor.ts new file mode 100644 index 0000000..d151450 --- /dev/null +++ b/tools/test_push_monitor.ts @@ -0,0 +1,17 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { pushPhase, pushQueuePhase, type PushRun, type PushQueueRun } from "../frontend/src/lib/push-monitor.ts"; + +test("停止监测时仍显示独立的队列处理状态", () => { + const queue: PushQueueRun = { id: 1, active: true, phase: "running", next_fetch_at: 0, reason: null }; + assert.equal(pushPhase({ active: false, phase: "stopped", queue } as PushRun), "已停止"); + assert.equal(pushQueuePhase(queue, 10, 1000), "正在处理"); + assert.equal(pushQueuePhase(queue, 0, 1000), "等待待处理链接"); +}); +test("解析限流和暂停不改变监测状态，队列原因独立显示", () => { + const queue: PushQueueRun = { id: 2, active: true, phase: "running", next_fetch_at: 2000, reason: "预算已用完" }; + assert.equal(pushPhase({ active: true, phase: "running", queue } as PushRun), "监测中"); + assert.equal(pushQueuePhase(queue, 10, 1000), "等待请求额度或间隔"); + assert.equal(pushQueuePhase({ ...queue, active: false, phase: "paused" }, 10, 1000), "已暂停"); + assert.equal(pushQueuePhase({ ...queue, active: false, phase: "failed" }, 10, 1000), "处理异常"); +});