pagesift_engine/
change.rs

1//! 变化检测（Firecrawl 的 `changeTracking` 格式）：和存储里同一 URL、同一 `tag` 上一次抓到的
2//! 正文 Markdown 比较，给出 `new`/`same`/`changed`/`removed`，按需附上 git diff 风格的差异。
3//!
4//! 每次抓取都把这次的正文追加成一份快照（内容按哈希去重，没变的页面不多占空间）。
5//! 快照的第一个字节标明种类：正文，或者页面已经不存在（404/410）。
6
7use std::time::SystemTime;
8
9use pagesift_api::types::{ChangeTracking, Format};
10use pagesift_api::{ApiError, rfc3339};
11use pagesift_net::Url;
12use pagesift_store::{ContentHash, Store};
13use serde_json::{Map, Value, json};
14
15const PAGE: u8 = b'M';
16const GONE: u8 = b'R';
17/// git diff 默认的上下文行数。
18const CONTEXT_LINES: usize = 3;
19
20/// 解析好的 `changeTracking` 格式参数。
21#[derive(Debug, Clone, Default, PartialEq)]
22pub struct Tracking {
23    pub git_diff: bool,
24    pub json: bool,
25    pub schema: Option<Value>,
26    pub prompt: Option<String>,
27    pub tag: Option<String>,
28}
29
30impl Tracking {
31    pub fn parse(f: &Format) -> Result<Self, ApiError> {
32        let mut t = Tracking {
33            schema: f.options.get("schema").cloned(),
34            prompt: f
35                .options
36                .get("prompt")
37                .and_then(Value::as_str)
38                .map(String::from),
39            tag: f
40                .options
41                .get("tag")
42                .and_then(Value::as_str)
43                .map(String::from),
44            ..Default::default()
45        };
46        match f.options.get("modes") {
47            None | Some(Value::Null) => {}
48            Some(Value::Array(modes)) => {
49                for m in modes {
50                    match m.as_str() {
51                        Some("git-diff") => t.git_diff = true,
52                        Some("json") => t.json = true,
53                        _ => {
54                            return Err(ApiError::invalid(format!(
55                                "changeTracking mode must be \"git-diff\" or \"json\", got {m}"
56                            )));
57                        }
58                    }
59                }
60            }
61            Some(other) => {
62                return Err(ApiError::invalid(format!(
63                    "changeTracking modes must be an array, got {other}"
64                )));
65            }
66        }
67        Ok(t)
68    }
69
70    /// 快照在存储里的键：同一 URL 的不同 `tag` 各自一条历史。
71    fn key(&self, url: &str) -> String {
72        let url = Url::parse(url)
73            .map(|u| crate::crawl::normalize(&u, false))
74            .unwrap_or_else(|_| url.to_string());
75        format!(
76            "changeTracking\u{1f}{}\u{1f}{url}",
77            self.tag.as_deref().unwrap_or("")
78        )
79    }
80}
81
82/// 上一份快照的内容。
83#[derive(Debug, Clone, PartialEq)]
84pub enum Previous {
85    Page(String),
86    Gone,
87    /// 快照还在历史里，但内容已经被存储淘汰：只能按哈希判断变没变，给不出差异。
88    Evicted,
89}
90
91/// 比较的结果：给 Firecrawl 的字段，加上计算 json 模式需要的上一版正文。
92pub struct Outcome {
93    pub tracking: ChangeTracking,
94    pub previous: Previous,
95    pub warnings: Vec<String>,
96}
97
98/// 这一次抓到的内容。
99#[derive(Debug, Clone, Copy, PartialEq)]
100pub enum Current<'a> {
101    Page(&'a str),
102    /// 404/410。
103    Gone,
104    /// 条件请求得到 304：内容和上一次一样，不另存快照。
105    NotModified,
106}
107
108fn encode(current: Current<'_>) -> Vec<u8> {
109    match current {
110        Current::Page(md) => {
111            let mut v = Vec::with_capacity(md.len() + 1);
112            v.push(PAGE);
113            v.extend_from_slice(md.as_bytes());
114            v
115        }
116        Current::Gone | Current::NotModified => vec![GONE],
117    }
118}
119
120fn decode(bytes: &[u8]) -> Previous {
121    match bytes.split_first() {
122        Some((&PAGE, md)) => Previous::Page(String::from_utf8_lossy(md).into_owned()),
123        _ => Previous::Gone,
124    }
125}
126
127/// 上一份快照：(时间, 内容哈希, 内容)。
128type Snapshot = (SystemTime, ContentHash, Previous);
129
130/// 取出上一份快照，再存入这一次的（`current` 为 `None` 时不存）。
131fn load_and_store(
132    store: &Store,
133    key: &str,
134    current: Option<&[u8]>,
135    now: SystemTime,
136) -> Result<(Option<Snapshot>, Vec<String>), pagesift_store::Error> {
137    let snaps = store.snapshots();
138    let previous = match snaps.latest(key)? {
139        Some(meta) => {
140            let content = match snaps.get(&meta.hash)? {
141                Some(bytes) => decode(&bytes),
142                None => Previous::Evicted,
143            };
144            Some((meta.at, meta.hash, content))
145        }
146        None => None,
147    };
148    let mut warnings = Vec::new();
149    let Some(current) = current else {
150        return Ok((previous, warnings));
151    };
152    match snaps.put_at(key, current, now) {
153        Ok(_) => {}
154        Err(pagesift_store::Error::ValueTooLarge { .. }) => {
155            warnings.push("change_tracking_snapshot_too_large".into());
156        }
157        Err(e) => return Err(e),
158    }
159    Ok((previous, warnings))
160}
161
162/// 和上一次比较并记录这一次。
163pub async fn track(
164    store: &Store,
165    tracking: &Tracking,
166    url: &str,
167    current: Current<'_>,
168) -> Result<Outcome, ApiError> {
169    let key = tracking.key(url);
170    let bytes = encode(current);
171    let hash = ContentHash::of(&bytes);
172    let now = SystemTime::now();
173    let stored = (current != Current::NotModified).then_some(bytes);
174    let (previous, mut warnings) = store
175        .blocking(move |s| load_and_store(s, &key, stored.as_deref(), now))
176        .await
177        .map_err(|e| {
178            ApiError::new(
179                pagesift_api::ErrorCode::Internal,
180                format!("change tracking store: {e}"),
181            )
182        })?;
183    let Some((at, prev_hash, prev)) = previous else {
184        return Ok(Outcome {
185            tracking: result(None, "new", None, None),
186            previous: Previous::Gone,
187            warnings,
188        });
189    };
190    let at = Some(rfc3339(at));
191    let cur = match current {
192        Current::Page(md) => md,
193        Current::Gone => {
194            return Ok(Outcome {
195                tracking: result(at, "removed", None, None),
196                previous: prev,
197                warnings,
198            });
199        }
200        Current::NotModified => "",
201    };
202    if prev_hash == hash || current == Current::NotModified {
203        return Ok(Outcome {
204            tracking: result(at, "same", None, None),
205            previous: prev,
206            warnings,
207        });
208    }
209    let old = match &prev {
210        Previous::Page(md) => Some(md.as_str()),
211        Previous::Gone => Some(""),
212        Previous::Evicted => None,
213    };
214    let (diff, ext) = match old {
215        Some(old) => {
216            let report = pagesift_diff::diff(old, cur);
217            let ext = json!({
218                "substantive": report.substantive,
219                "stats": report.stats,
220            });
221            (tracking.git_diff.then(|| git_diff(old, cur)), Some(ext))
222        }
223        None => {
224            if tracking.git_diff {
225                warnings.push("change_tracking_previous_evicted".into());
226            }
227            (None, None)
228        }
229    };
230    Ok(Outcome {
231        tracking: result(at, "changed", diff, ext),
232        previous: prev,
233        warnings,
234    })
235}
236
237fn result(
238    previous_scrape_at: Option<String>,
239    status: &str,
240    diff: Option<Value>,
241    pagesift: Option<Value>,
242) -> ChangeTracking {
243    ChangeTracking {
244        previous_scrape_at,
245        change_status: status.into(),
246        visibility: "visible".into(),
247        diff,
248        json: None,
249        pagesift,
250    }
251}
252
253/// `{"text": ..., "json": ...}`：文本是统一格式的 hunk，JSON 是 parse-diff（Firecrawl 用的库）的结构。
254pub fn git_diff(old: &str, new: &str) -> Value {
255    let d = pagesift_diff::unified_diff(old, new, CONTEXT_LINES);
256    let chunks: Vec<Value> = d
257        .hunks
258        .iter()
259        .map(|h| {
260            let changes: Vec<Value> = h
261                .lines
262                .iter()
263                .map(|l| {
264                    let content = l.prefixed();
265                    match l.kind {
266                        pagesift_diff::LineKind::Normal => json!({
267                            "type": "normal", "normal": true,
268                            "ln1": l.old_line, "ln2": l.new_line, "content": content,
269                        }),
270                        pagesift_diff::LineKind::Add => json!({
271                            "type": "add", "add": true, "ln": l.new_line, "content": content,
272                        }),
273                        pagesift_diff::LineKind::Del => json!({
274                            "type": "del", "del": true, "ln": l.old_line, "content": content,
275                        }),
276                    }
277                })
278                .collect();
279            json!({
280                "content": h.header(),
281                "changes": changes,
282                "oldStart": h.old_start,
283                "oldLines": h.old_lines,
284                "newStart": h.new_start,
285                "newLines": h.new_lines,
286            })
287        })
288        .collect();
289    json!({
290        "text": d.text(),
291        "json": {
292            "files": [{
293                "from": null,
294                "to": null,
295                "chunks": chunks,
296                "additions": d.additions,
297                "deletions": d.deletions,
298            }],
299        },
300    })
301}
302
303/// json 模式：两版提取结果里值不同的字段，写成 `{"字段": {"previous": 旧值, "current": 新值}}`。
304pub fn json_changes(previous: &Value, current: &Value) -> Value {
305    let empty = Map::new();
306    let (p, c) = match (previous, current) {
307        (Value::Object(p), Value::Object(c)) => (p, c),
308        (Value::Null, Value::Object(c)) => (&empty, c),
309        _ if previous == current => return Value::Object(Map::new()),
310        _ => return json!({ "previous": previous, "current": current }),
311    };
312    let mut out = Map::new();
313    for k in p.keys().chain(c.keys().filter(|k| !p.contains_key(*k))) {
314        let (pv, cv) = (
315            p.get(k).unwrap_or(&Value::Null),
316            c.get(k).unwrap_or(&Value::Null),
317        );
318        if pv != cv {
319            out.insert(k.clone(), json!({ "previous": pv, "current": cv }));
320        }
321    }
322    Value::Object(out)
323}
324
325#[cfg(test)]
326mod tests {
327    use super::*;
328
329    fn fmt(v: Value) -> Format {
330        serde_json::from_value(v).unwrap()
331    }
332
333    #[test]
334    fn parses_modes_and_tag() {
335        let t = Tracking::parse(&fmt(json!({"type": "changeTracking"}))).unwrap();
336        assert_eq!(t, Tracking::default());
337        let t = Tracking::parse(&fmt(json!({
338            "type": "changeTracking", "modes": ["git-diff", "json"], "tag": "daily",
339            "schema": {"type": "object"}, "prompt": "price",
340        })))
341        .unwrap();
342        assert!(t.git_diff && t.json);
343        assert_eq!(t.tag.as_deref(), Some("daily"));
344        assert_eq!(t.prompt.as_deref(), Some("price"));
345        assert!(
346            Tracking::parse(&fmt(json!({"type": "changeTracking", "modes": ["word"]}))).is_err()
347        );
348        assert!(Tracking::parse(&fmt(json!({"type": "changeTracking", "modes": "json"}))).is_err());
349        assert_ne!(
350            t.key("https://example.com/a#x"),
351            Tracking::default().key("https://example.com/a"),
352            "tags keep separate histories"
353        );
354        assert_eq!(
355            Tracking::default().key("https://EXAMPLE.com/a#x"),
356            Tracking::default().key("https://example.com/a")
357        );
358    }
359
360    #[test]
361    fn git_diff_follows_parse_diff_shape() {
362        let v = git_diff("# T\n\nold content\n", "# T\n\nnew content\n");
363        assert_eq!(
364            v["text"],
365            "@@ -1,3 +1,3 @@\n # T\n \n-old content\n+new content"
366        );
367        let file = &v["json"]["files"][0];
368        assert_eq!(file["from"], Value::Null);
369        assert_eq!(
370            (file["additions"].as_u64(), file["deletions"].as_u64()),
371            (Some(1), Some(1))
372        );
373        let chunk = &file["chunks"][0];
374        assert_eq!(chunk["content"], "@@ -1,3 +1,3 @@");
375        assert_eq!(
376            chunk["changes"][0],
377            json!({"type": "normal", "normal": true, "ln1": 1, "ln2": 1, "content": " # T"})
378        );
379        assert_eq!(
380            chunk["changes"][2],
381            json!({"type": "del", "del": true, "ln": 3, "content": "-old content"})
382        );
383        assert_eq!(
384            chunk["changes"][3],
385            json!({"type": "add", "add": true, "ln": 3, "content": "+new content"})
386        );
387    }
388
389    #[test]
390    fn json_changes_lists_changed_fields_only() {
391        let p = json!({"title": "A", "price": 10, "gone": 1});
392        let c = json!({"title": "A", "price": 12, "new": true});
393        assert_eq!(
394            json_changes(&p, &c),
395            json!({
396                "price": {"previous": 10, "current": 12},
397                "gone": {"previous": 1, "current": null},
398                "new": {"previous": null, "current": true},
399            })
400        );
401        assert_eq!(json_changes(&json!([1]), &json!([1])), json!({}));
402        assert_eq!(
403            json_changes(&json!([1]), &json!([2])),
404            json!({"previous": [1], "current": [2]})
405        );
406    }
407
408    #[tokio::test]
409    async fn statuses_follow_the_snapshot_history() {
410        let dir = tempfile::tempdir().unwrap();
411        let store = Store::open_dir(dir.path()).unwrap();
412        let t = Tracking {
413            git_diff: true,
414            ..Default::default()
415        };
416        let url = "https://example.com/p";
417        let first = track(&store, &t, url, Current::Page("# P\n\nv1\n"))
418            .await
419            .unwrap();
420        assert_eq!(first.tracking.change_status, "new");
421        assert_eq!(first.tracking.previous_scrape_at, None);
422        let same = track(&store, &t, url, Current::Page("# P\n\nv1\n"))
423            .await
424            .unwrap();
425        assert_eq!(same.tracking.change_status, "same");
426        assert!(same.tracking.previous_scrape_at.is_some());
427        assert!(same.tracking.diff.is_none());
428        let changed = track(&store, &t, url, Current::Page("# P\n\nv2\n"))
429            .await
430            .unwrap();
431        assert_eq!(changed.tracking.change_status, "changed");
432        assert_eq!(changed.previous, Previous::Page("# P\n\nv1\n".into()));
433        assert_eq!(
434            changed.tracking.diff.as_ref().unwrap()["text"],
435            "@@ -1,3 +1,3 @@\n # P\n \n-v1\n+v2"
436        );
437        assert_eq!(
438            changed.tracking.pagesift.as_ref().unwrap()["substantive"],
439            true
440        );
441        // 304：和上一次一样，也不另存快照
442        let unchanged = track(&store, &t, url, Current::NotModified).await.unwrap();
443        assert_eq!(unchanged.tracking.change_status, "same");
444        assert_eq!(unchanged.previous, Previous::Page("# P\n\nv2\n".into()));
445        assert_eq!(store.snapshots().list(&t.key(url), ..).unwrap().len(), 3);
446        let removed = track(&store, &t, url, Current::Gone).await.unwrap();
447        assert_eq!(removed.tracking.change_status, "removed");
448        let back = track(&store, &t, url, Current::Page("# P\n\nv3\n"))
449            .await
450            .unwrap();
451        assert_eq!(back.tracking.change_status, "changed");
452        assert_eq!(back.previous, Previous::Gone);
453        // 另一个 tag 有自己的历史
454        let tagged = Tracking {
455            tag: Some("x".into()),
456            ..Default::default()
457        };
458        let other = track(&store, &tagged, url, Current::Page("# P\n\nv3\n"))
459            .await
460            .unwrap();
461        assert_eq!(other.tracking.change_status, "new");
462    }
463
464    #[tokio::test]
465    async fn noise_only_changes_are_marked_not_substantive() {
466        let dir = tempfile::tempdir().unwrap();
467        let store = Store::open_dir(dir.path()).unwrap();
468        let t = Tracking::default();
469        let url = "https://example.com/n";
470        track(
471            &store,
472            &t,
473            url,
474            Current::Page("Updated 2026-01-01.\n\nBody text.\n"),
475        )
476        .await
477        .unwrap();
478        let o = track(
479            &store,
480            &t,
481            url,
482            Current::Page("Updated 2026-02-03.\n\nBody text.\n"),
483        )
484        .await
485        .unwrap();
486        assert_eq!(o.tracking.change_status, "changed");
487        assert!(o.tracking.diff.is_none(), "git-diff mode not requested");
488        assert_eq!(o.tracking.pagesift.unwrap()["substantive"], false);
489    }
490}
pagesift_engine/change.rs

pagesift_engine/
change.rs
