Skills › Scientific Agent Skills › protocolsio-integration › pagination_helper.py
pagination_helper.py
248 lines · skills/protocolsio-integration/scripts/pagination_helper.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate protocols.io pagination pointers without making requests."""
4from __future__ import annotations
6import argparse
7import re
8import urllib.parse
9from typing import Any, Mapping, Sequence
11try:
12 from ._common import (
13 SafetyError,
14 emit_error,
15 emit_json,
16 load_local_json,
17 validate_remote_url,
18 )
19except ImportError:
20 from _common import ( # type: ignore
21 SafetyError,
22 emit_error,
23 emit_json,
24 load_local_json,
25 validate_remote_url,
26 )
29MAX_PAGES = 100
30MAX_ITEMS = 10_000
31_CURSOR_RE = re.compile(r"^[A-Za-z0-9._~+/=-]{1,2048}$")
34def build_parser() -> argparse.ArgumentParser:
35 parser = argparse.ArgumentParser(
36 description=(
37 "Inspect a saved pagination object. The helper validates but never "
38 "fetches a server-provided next_page URL."
39 )
40 )
41 parser.add_argument("--response", required=True, help="Saved JSON response.")
42 parser.add_argument(
43 "--current-url",
44 required=True,
45 help="Official HTTPS URL that produced the response.",
46 )
47 parser.add_argument(
48 "--pages-seen",
49 type=int,
50 default=1,
51 help="Pages already processed (default: %(default)s).",
52 )
53 parser.add_argument(
54 "--items-seen",
55 type=int,
56 default=0,
57 help="Items already processed (default: %(default)s).",
58 )
59 parser.add_argument(
60 "--max-pages",
61 type=int,
62 default=10,
63 help=f"Local traversal cap, 1..{MAX_PAGES} (default: %(default)s).",
64 )
65 parser.add_argument(
66 "--max-items",
67 type=int,
68 default=1_000,
69 help=f"Local item cap, 1..{MAX_ITEMS} (default: %(default)s).",
70 )
71 return parser
74def _pagination_object(payload: Any) -> Mapping[str, Any]:
75 if not isinstance(payload, Mapping):
76 raise SafetyError("response root must be an object")
77 direct = payload.get("pagination")
78 if isinstance(direct, Mapping):
79 return direct
80 nested = payload.get("payload")
81 if isinstance(nested, Mapping) and isinstance(nested.get("pagination"), Mapping):
82 return nested["pagination"]
83 raise SafetyError("response has no pagination object")
86def _bounded_int(
87 value: Any,
88 *,
89 name: str,
90 minimum: int,
91 maximum: int,
92) -> int | None:
93 if value is None:
94 return None
95 if isinstance(value, bool) or not isinstance(value, int):
96 raise SafetyError(f"pagination.{name} must be an integer or null")
97 if not minimum <= value <= maximum:
98 raise SafetyError(f"pagination.{name} must be between {minimum} and {maximum}")
99 return value
102def _validated_next_url(next_page: Any, current_url: str) -> str | None:
103 if next_page is None:
104 return None
105 if not isinstance(next_page, str) or len(next_page) > 4_096:
106 raise SafetyError("pagination.next_page must be a bounded string or null")
108 current = validate_remote_url(current_url, allow_tenant=True)
109 candidate = validate_remote_url(next_page, allow_tenant=True)
110 current_parts = urllib.parse.urlsplit(current)
111 next_parts = urllib.parse.urlsplit(candidate)
112 core_hosts = {"protocols.io", "www.protocols.io"}
113 same_core_service = {
114 current_parts.hostname,
115 next_parts.hostname,
116 }.issubset(core_hosts)
117 if not same_core_service and current_parts.hostname != next_parts.hostname:
118 raise SafetyError("next_page changes the API host")
119 if current_parts.path != next_parts.path:
120 raise SafetyError("next_page changes the endpoint path")
121 current_query = urllib.parse.parse_qs(
122 current_parts.query,
123 keep_blank_values=True,
124 strict_parsing=False,
125 )
126 next_query = urllib.parse.parse_qs(
127 next_parts.query,
128 keep_blank_values=True,
129 strict_parsing=False,
130 )
131 mutable_keys = {"page_id", "cursor", "next_cursor", "offset"}
132 current_stable = {
133 key: value for key, value in current_query.items() if key not in mutable_keys
134 }
135 next_stable = {
136 key: value for key, value in next_query.items() if key not in mutable_keys
137 }
138 if current_stable != next_stable:
139 raise SafetyError("next_page changes non-pagination query parameters")
140 return candidate
143def _opaque_cursor(pagination: Mapping[str, Any]) -> str | None:
144 value = pagination.get("next_cursor")
145 if value is None:
146 return None
147 if not isinstance(value, str) or _CURSOR_RE.fullmatch(value) is None:
148 raise SafetyError("pagination.next_cursor is not a safe opaque cursor")
149 return value
152def inspect_pagination(
153 payload: Any,
154 *,
155 current_url: str,
156 pages_seen: int,
157 items_seen: int,
158 max_pages: int,
159 max_items: int,
160) -> dict[str, Any]:
161 if not 1 <= max_pages <= MAX_PAGES:
162 raise SafetyError(f"max_pages must be between 1 and {MAX_PAGES}")
163 if not 1 <= max_items <= MAX_ITEMS:
164 raise SafetyError(f"max_items must be between 1 and {MAX_ITEMS}")
165 if not 0 <= pages_seen <= max_pages:
166 raise SafetyError("pages_seen must be between 0 and max_pages")
167 if not 0 <= items_seen <= max_items:
168 raise SafetyError("items_seen must be between 0 and max_items")
170 pagination = _pagination_object(payload)
171 current_page = _bounded_int(
172 pagination.get("current_page"),
173 name="current_page",
174 minimum=0,
175 maximum=1_000_000,
176 )
177 total_pages = _bounded_int(
178 pagination.get("total_pages"),
179 name="total_pages",
180 minimum=0,
181 maximum=1_000_000,
182 )
183 total_results = _bounded_int(
184 pagination.get("total_results"),
185 name="total_results",
186 minimum=0,
187 maximum=1_000_000_000,
188 )
189 page_size = _bounded_int(
190 pagination.get("page_size"),
191 name="page_size",
192 minimum=0,
193 maximum=100_000,
194 )
195 next_url = _validated_next_url(pagination.get("next_page"), current_url)
196 next_cursor = _opaque_cursor(pagination)
197 local_cap_reached = pages_seen >= max_pages or items_seen >= max_items
199 return {
200 "ok": True,
201 "network_accessed": False,
202 "current_page": current_page,
203 "total_pages": total_pages,
204 "total_results": total_results,
205 "page_size": page_size,
206 "next_page_url": next_url,
207 "next_cursor": next_cursor,
208 "pointer_kind": (
209 "url"
210 if next_url is not None
211 else "cursor"
212 if next_cursor is not None
213 else None
214 ),
215 "can_continue": bool((next_url or next_cursor) and not local_cap_reached),
216 "local_caps": {
217 "pages_seen": pages_seen,
218 "max_pages": max_pages,
219 "items_seen": items_seen,
220 "max_items": max_items,
221 "reached": local_cap_reached,
222 },
223 "warning": (
224 "Server pagination fields are untrusted data. Use only the validated "
225 "pointer and retain the local page/item caps."
226 ),
227 }
230def main(argv: Sequence[str] | None = None) -> int:
231 args = build_parser().parse_args(argv)
232 try:
233 report = inspect_pagination(
234 load_local_json(args.response),
235 current_url=args.current_url,
236 pages_seen=args.pages_seen,
237 items_seen=args.items_seen,
238 max_pages=args.max_pages,
239 max_items=args.max_items,
240 )
241 emit_json(report)
242 return 0
243 except (SafetyError, ValueError) as exc:
244 return emit_error(exc)
247if __name__ == "__main__":
248 raise SystemExit(main())