Skills › Scientific Agent Skills › autoskill › fetch_window.py
fetch_window.py
63 lines · skills/autoskill/scripts/fetch_window.py at 5f2bc03
1import warnings
3_MAX_PAGES = 10_000 # bounded exit: hard ceiling so the loop cannot spin forever
6def fetch_window(client, start_time, end_time, page_size=50, token=None):
7 """Fetch local timeline rows; memories/parsed records are not activity events."""
8 if isinstance(page_size, bool) or not isinstance(page_size, int) or page_size < 1:
9 raise ValueError("page_size must be a positive integer")
10 events = []
11 skipped_types = set()
12 offset = 0
13 headers = {"Authorization": f"Bearer {token}"} if token else {}
14 for _page in range(_MAX_PAGES):
15 response = client.get("/search", params={
16 "start_time": start_time,
17 "end_time": end_time,
18 "limit": page_size,
19 "offset": offset,
20 "content_type": "all",
21 "include_frames": "false",
22 "include_cloud": "false",
23 # Upstream filter_pii can use a remote enclave. Redact locally instead.
24 "filter_pii": "false",
25 }, headers=headers)
26 response.raise_for_status()
27 payload = response.json()
28 data = payload.get("data")
29 pagination = payload.get("pagination", {})
30 total = pagination.get("total")
31 if (not isinstance(data, list) or isinstance(total, bool)
32 or not isinstance(total, int) or total < 0):
33 raise ValueError("Screenpipe /search must return data[] and pagination.total")
34 if pagination.get("offset", offset) != offset:
35 raise ValueError("Screenpipe /search returned an unexpected pagination offset")
37 for item in data:
38 content = item.get("content", {})
39 kind = item.get("type", "").lower()
40 if kind not in {"ocr", "ui", "accessibility", "input", "audio"}:
41 skipped_types.add(kind or "unknown")
42 continue
43 if not content.get("timestamp"):
44 raise ValueError(f"Screenpipe {kind} row has no timestamp")
45 events.append({
46 "ts": content.get("timestamp"),
47 "app": content.get("app_name") or "",
48 "window_title": content.get("window_name") or content.get("window_title") or "",
49 "text": content.get("text") or content.get("transcription") or content.get("text_content") or "",
50 "content_type": kind,
51 })
53 offset += len(data)
54 if offset >= total:
55 break
56 if not data:
57 raise RuntimeError("Screenpipe search ended before pagination.total; rerun a fixed time window")
58 else:
59 raise RuntimeError("Screenpipe search exceeded the page limit; request a smaller time window")
60 if skipped_types:
61 warnings.warn("Skipped non-timeline Screenpipe types: " + ", ".join(sorted(skipped_types)),
62 stacklevel=2)
63 return events