Skills › Scientific Agent Skills › scientific-writing › _common.py
_common.py
251 lines · skills/scientific-writing/scripts/_common.py at 5f2bc03
1"""Shared, dependency-free safety helpers for scientific-writing CLIs."""
3from __future__ import annotations
5import csv
6import io
7import json
8import sys
9from collections.abc import Iterable
10from dataclasses import asdict, dataclass
11from datetime import date
12from pathlib import Path
13from typing import Any
15MAX_FILE_BYTES = 5_000_000
16MAX_RECORDS = 10_000
17MAX_JSON_NODES = 100_000
18MAX_JSON_DEPTH = 50
19MAX_CSV_FIELD_BYTES = 100_000
22class InputError(ValueError):
23 """Raised when a local input fails a bounded safety check."""
26@dataclass(frozen=True)
27class Issue:
28 severity: str
29 code: str
30 location: str | None = None
31 item_id: str | None = None
33 def to_dict(self) -> dict[str, str]:
34 return {key: value for key, value in asdict(self).items() if value is not None}
37def _checked_file(path_value: str | Path, suffixes: Iterable[str]) -> Path:
38 path = Path(path_value)
39 allowed = {suffix.lower() for suffix in suffixes}
40 if path.is_symlink():
41 raise InputError("symbolic-link inputs are not accepted")
42 if not path.is_file():
43 raise InputError("input must be an existing regular file")
44 if path.suffix.lower() not in allowed:
45 raise InputError(
46 f"input extension must be one of: {', '.join(sorted(allowed))}"
47 )
48 if path.stat().st_size > MAX_FILE_BYTES:
49 raise InputError(f"input exceeds {MAX_FILE_BYTES} bytes")
50 return path
53def read_text(path_value: str | Path, suffixes: Iterable[str]) -> str:
54 path = _checked_file(path_value, suffixes)
55 try:
56 return path.read_text(encoding="utf-8")
57 except UnicodeDecodeError as exc:
58 raise InputError("input must be UTF-8 text") from exc
61def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
62 result: dict[str, Any] = {}
63 for key, value in pairs:
64 if key in result:
65 raise InputError("JSON contains a duplicate object key")
66 result[key] = value
67 return result
70def _reject_nonfinite_json(_value: str) -> None:
71 raise InputError("JSON non-finite numbers are not accepted")
74def _check_json_bounds(value: Any, depth: int = 0) -> int:
75 if depth > MAX_JSON_DEPTH:
76 raise InputError(f"JSON nesting exceeds {MAX_JSON_DEPTH} levels")
77 count = 1
78 if isinstance(value, dict):
79 for child in value.values():
80 count += _check_json_bounds(child, depth + 1)
81 elif isinstance(value, list):
82 if len(value) > MAX_RECORDS:
83 raise InputError(f"JSON array exceeds {MAX_RECORDS} records")
84 for child in value:
85 count += _check_json_bounds(child, depth + 1)
86 if count > MAX_JSON_NODES:
87 raise InputError(f"JSON exceeds {MAX_JSON_NODES} nodes")
88 return count
91def read_json(path_value: str | Path) -> Any:
92 text = read_text(path_value, {".json"})
93 try:
94 value = json.loads(
95 text,
96 object_pairs_hook=_reject_duplicate_keys,
97 parse_constant=_reject_nonfinite_json,
98 )
99 except (json.JSONDecodeError, RecursionError) as exc:
100 raise InputError("input is not valid bounded JSON") from exc
101 _check_json_bounds(value)
102 return value
105def read_csv(path_value: str | Path) -> tuple[list[str], list[dict[str, str]]]:
106 text = read_text(path_value, {".csv"})
107 csv.field_size_limit(MAX_CSV_FIELD_BYTES)
108 try:
109 reader = csv.DictReader(io.StringIO(text, newline=""))
110 fields = list(reader.fieldnames or [])
111 if not fields or any(not field for field in fields):
112 raise InputError("CSV requires a non-empty header row")
113 if len(set(fields)) != len(fields):
114 raise InputError("CSV contains duplicate header names")
115 rows: list[dict[str, str]] = []
116 for index, row in enumerate(reader, start=1):
117 if index > MAX_RECORDS:
118 raise InputError(f"CSV exceeds {MAX_RECORDS} data rows")
119 if None in row:
120 raise InputError("CSV row has more fields than the header")
121 rows.append({key: value or "" for key, value in row.items()})
122 except csv.Error as exc:
123 raise InputError("input is not valid bounded CSV") from exc
124 return fields, rows
127def issue(
128 severity: str,
129 code: str,
130 *,
131 location: str | None = None,
132 item_id: str | None = None,
133) -> Issue:
134 if severity not in {"error", "warning", "info"}:
135 raise ValueError("unsupported issue severity")
136 return Issue(severity=severity, code=code, location=location, item_id=item_id)
139def emit_report(
140 tool: str,
141 issues: Iterable[Issue],
142 *,
143 summary: dict[str, Any] | None = None,
144) -> int:
145 ordered = sorted(
146 issues,
147 key=lambda item: (
148 {"error": 0, "warning": 1, "info": 2}[item.severity],
149 item.code,
150 item.location or "",
151 item.item_id or "",
152 ),
153 )
154 error_count = sum(item.severity == "error" for item in ordered)
155 warning_count = sum(item.severity == "warning" for item in ordered)
156 payload = {
157 "tool": tool,
158 "status": "fail" if error_count else "pass",
159 "summary": {
160 "errors": error_count,
161 "warnings": warning_count,
162 **(summary or {}),
163 },
164 "issues": [item.to_dict() for item in ordered],
165 }
166 print(json.dumps(payload, indent=2, sort_keys=True))
167 return 1 if error_count else 0
170def emit_input_error(tool: str, exc: Exception) -> int:
171 return emit_report(
172 tool,
173 [issue("error", "INVALID_INPUT", location=type(exc).__name__)],
174 )
177def require_object(value: Any, label: str = "root") -> dict[str, Any]:
178 if not isinstance(value, dict):
179 raise InputError(f"{label} must be a JSON object")
180 return value
183def require_list(value: Any, label: str) -> list[Any]:
184 if not isinstance(value, list):
185 raise InputError(f"{label} must be a JSON array")
186 if len(value) > MAX_RECORDS:
187 raise InputError(f"{label} exceeds {MAX_RECORDS} records")
188 return value
191def is_nonempty_string(value: Any) -> bool:
192 return isinstance(value, str) and bool(value.strip())
195def is_iso_date(value: Any) -> bool:
196 """Accept an actual calendar date in the exact YYYY-MM-DD form."""
197 if not isinstance(value, str):
198 return False
199 try:
200 return date.fromisoformat(value).isoformat() == value
201 except ValueError:
202 return False
205def is_placeholder(value: Any) -> bool:
206 if not isinstance(value, str):
207 return False
208 lowered = value.strip().lower()
209 return (
210 not lowered
211 or "[[todo" in lowered
212 or lowered in {"todo", "tbd", "tk", "replace_me", "unknown"}
213 )
216def write_new_text(
217 path_value: str | Path, content: str, suffixes: Iterable[str]
218) -> Path:
219 path = Path(path_value)
220 allowed = {suffix.lower() for suffix in suffixes}
221 if path.suffix.lower() not in allowed:
222 raise InputError(
223 f"output extension must be one of: {', '.join(sorted(allowed))}"
224 )
225 if path.exists() or path.is_symlink():
226 raise InputError("output already exists; refusing to overwrite")
227 parent = path.parent
228 if not parent.is_dir() or parent.is_symlink():
229 raise InputError("output parent must be an existing regular directory")
230 encoded = content.encode("utf-8")
231 if len(encoded) > MAX_FILE_BYTES:
232 raise InputError(f"output exceeds {MAX_FILE_BYTES} bytes")
233 with path.open("x", encoding="utf-8", newline="\n") as handle:
234 handle.write(content)
235 return path
238def main_guard(tool: str, callback: Any) -> int:
239 try:
240 return int(callback())
241 except (InputError, OSError, ValueError) as exc:
242 return emit_input_error(tool, exc)
245def run(tool: str, callback: Any) -> None:
246 raise SystemExit(main_guard(tool, callback))
249if __name__ == "__main__":
250 print("This module is imported by the scientific-writing command-line tools.")
251 sys.exit(0)