Skills › Scientific Agent Skills › pptx-posters › _common.py
_common.py
397 lines · skills/pptx-posters/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared, dependency-free safety helpers for the poster command-line tools."""
4from __future__ import annotations
6import hashlib
7import json
8import math
9import os
10import re
11import stat
12import tempfile
13from datetime import datetime
14from pathlib import Path
15from typing import Any, Iterable
17MAX_JSON_BYTES = 2 * 1024 * 1024
18MAX_INPUT_BYTES = 512 * 1024 * 1024
19MAX_ASSET_BYTES = 128 * 1024 * 1024
20MAX_REPORT_BYTES = 8 * 1024 * 1024
21PPTX_MIN_INCHES = 1.0
22PPTX_MAX_INCHES = 56.0
23EMU_PER_INCH = 914_400
25_REMOTE_OR_SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:")
26_HEX_COLOR = re.compile(r"^#[0-9A-Fa-f]{6}$")
27_PLACEHOLDER_PATTERNS = (
28 re.compile(
29 r"\b(?:TODO|TBD|TBC|FIXME|LOREM\s+IPSUM|"
30 r"REPLACE(?:[_ -]?ME)?(?:[_ -][A-Z0-9]+)*|"
31 r"PLACEHOLDER|INSERT[_ -]HERE|YOUR[_ -](?:TITLE|NAME|TEXT|URL))\b",
32 re.IGNORECASE,
33 ),
34 re.compile(
35 r"(?:<|\[)\s*(?:title|author|affiliation|content|citation|source|"
36 r"url|doi|path|value)\s*(?:>|\])",
37 re.IGNORECASE,
38 ),
39)
42class CliError(ValueError):
43 """An expected validation or command-line error."""
46def checked_input_file(
47 value: str | os.PathLike[str],
48 *,
49 max_bytes: int = MAX_INPUT_BYTES,
50 suffixes: Iterable[str] | None = None,
51) -> Path:
52 """Return a bounded regular input file while rejecting a final symlink."""
53 path = Path(value)
54 if path.is_symlink():
55 raise CliError(f"input must not be a symlink: {path}")
56 try:
57 info = path.stat()
58 except OSError as exc:
59 raise CliError(f"cannot access input file {path}: {exc}") from exc
60 if not stat.S_ISREG(info.st_mode):
61 raise CliError(f"input is not a regular file: {path}")
62 if info.st_size > max_bytes:
63 raise CliError(
64 f"input is {info.st_size} bytes; limit is {max_bytes} bytes"
65 )
66 resolved = path.resolve()
67 if suffixes is not None:
68 allowed = {suffix.lower() for suffix in suffixes}
69 if resolved.suffix.lower() not in allowed:
70 raise CliError(
71 f"input suffix must be one of {', '.join(sorted(allowed))}: {path}"
72 )
73 return resolved
76def checked_output_file(
77 value: str | os.PathLike[str],
78 *,
79 suffix: str | None = None,
80) -> Path:
81 """Validate a new output path; existing destinations are never overwritten."""
82 path = Path(value)
83 if path.name in {"", ".", ".."}:
84 raise CliError("output must name a file")
85 if path.is_symlink():
86 raise CliError(f"output must not be a symlink: {path}")
87 parent = path.parent
88 if not parent.exists() or not parent.is_dir():
89 raise CliError(f"output parent directory does not exist: {parent}")
90 if parent.is_symlink():
91 raise CliError(f"output parent must not be a symlink: {parent}")
92 destination = parent.resolve() / path.name
93 if destination.exists():
94 raise CliError(f"refusing to overwrite existing output: {destination}")
95 if suffix is not None and destination.suffix.lower() != suffix.lower():
96 raise CliError(f"output must use the {suffix} suffix: {destination}")
97 return destination
100def private_temp_file(destination: Path, *, suffix: str = ".tmp") -> tuple[int, Path]:
101 """Create a private same-directory temporary file for an output."""
102 descriptor, raw_path = tempfile.mkstemp(
103 prefix=f".{destination.name}.",
104 suffix=suffix,
105 dir=destination.parent,
106 )
107 temporary = Path(raw_path)
108 os.chmod(temporary, 0o600)
109 return descriptor, temporary
112def commit_temp_file(temporary: Path, destination: Path) -> None:
113 """Publish a temporary file without replacing an existing destination."""
114 if destination.exists():
115 raise CliError(f"refusing to overwrite existing output: {destination}")
116 try:
117 os.link(temporary, destination)
118 except FileExistsError as exc:
119 raise CliError(
120 f"refusing to overwrite existing output: {destination}"
121 ) from exc
122 except OSError as exc:
123 raise CliError(f"cannot publish output {destination}: {exc}") from exc
124 os.chmod(destination, 0o600)
127def atomic_write_bytes(path: Path, payload: bytes) -> None:
128 """Write bytes through a private temporary file without replacement."""
129 destination = checked_output_file(path)
130 descriptor, temporary = private_temp_file(destination)
131 try:
132 with os.fdopen(descriptor, "wb") as handle:
133 handle.write(payload)
134 handle.flush()
135 os.fsync(handle.fileno())
136 commit_temp_file(temporary, destination)
137 finally:
138 try:
139 temporary.unlink()
140 except FileNotFoundError:
141 pass
144def emit_json(
145 document: dict[str, Any],
146 *,
147 output: str | os.PathLike[str] | None = None,
148) -> None:
149 """Print deterministic JSON or safely write it to a new file."""
150 payload = (
151 json.dumps(document, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
152 ).encode("utf-8")
153 if len(payload) > MAX_REPORT_BYTES:
154 raise CliError(
155 f"report is {len(payload)} bytes; limit is {MAX_REPORT_BYTES} bytes"
156 )
157 if output is None:
158 print(payload.decode("utf-8"), end="")
159 return
160 atomic_write_bytes(Path(output), payload)
163def _object_without_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
164 result: dict[str, Any] = {}
165 for key, value in pairs:
166 if key in result:
167 raise CliError(f"duplicate JSON key: {key!r}")
168 result[key] = value
169 return result
172def _reject_nonfinite_constant(value: str) -> None:
173 raise CliError(f"non-finite JSON number is not allowed: {value}")
176def load_json_file(
177 value: str | os.PathLike[str],
178 *,
179 max_bytes: int = MAX_JSON_BYTES,
180) -> tuple[Path, Any]:
181 """Read strict UTF-8 JSON with duplicate and non-finite values rejected."""
182 path = checked_input_file(value, max_bytes=max_bytes, suffixes={".json"})
183 try:
184 raw = path.read_bytes()
185 text = raw.decode("utf-8")
186 except UnicodeDecodeError as exc:
187 raise CliError(f"JSON must be UTF-8: {path}") from exc
188 except OSError as exc:
189 raise CliError(f"cannot read JSON {path}: {exc}") from exc
190 try:
191 document = json.loads(
192 text,
193 object_pairs_hook=_object_without_duplicates,
194 parse_constant=_reject_nonfinite_constant,
195 )
196 except json.JSONDecodeError as exc:
197 raise CliError(
198 f"invalid JSON in {path} at line {exc.lineno}, column {exc.colno}: "
199 f"{exc.msg}"
200 ) from exc
201 return path, document
204def reject_unknown_keys(
205 value: Any,
206 *,
207 context: str,
208 allowed: set[str],
209 required: set[str],
210) -> dict[str, Any]:
211 """Return an object after enforcing exact keys."""
212 if not isinstance(value, dict):
213 raise CliError(f"{context} must be an object")
214 unknown = sorted(set(value) - allowed)
215 missing = sorted(required - set(value))
216 if unknown:
217 raise CliError(f"{context} has unknown key(s): {', '.join(unknown)}")
218 if missing:
219 raise CliError(f"{context} is missing key(s): {', '.join(missing)}")
220 return value
223def require_string(
224 value: Any,
225 *,
226 context: str,
227 minimum: int = 1,
228 maximum: int = 10_000,
229) -> str:
230 """Validate a bounded string without trimming or changing its content."""
231 if not isinstance(value, str):
232 raise CliError(f"{context} must be a string")
233 if not minimum <= len(value) <= maximum:
234 raise CliError(
235 f"{context} length must be between {minimum} and {maximum} characters"
236 )
237 if any(ord(character) < 32 and character not in "\t\n\r" for character in value):
238 raise CliError(f"{context} contains a forbidden control character")
239 if any(0xD800 <= ord(character) <= 0xDFFF for character in value):
240 raise CliError(f"{context} contains an unpaired Unicode surrogate")
241 return value
244def require_bool(value: Any, *, context: str) -> bool:
245 if not isinstance(value, bool):
246 raise CliError(f"{context} must be true or false")
247 return value
250def finite_number(
251 value: Any,
252 *,
253 context: str,
254 minimum: float | None = None,
255 maximum: float | None = None,
256) -> float:
257 """Validate a finite JSON number, excluding booleans."""
258 if isinstance(value, bool) or not isinstance(value, (int, float)):
259 raise CliError(f"{context} must be a number")
260 number = float(value)
261 if not math.isfinite(number):
262 raise CliError(f"{context} must be finite")
263 if minimum is not None and number < minimum:
264 raise CliError(f"{context} must be at least {minimum}")
265 if maximum is not None and number > maximum:
266 raise CliError(f"{context} must be at most {maximum}")
267 return number
270def positive_int(value: Any, *, context: str, maximum: int) -> int:
271 if isinstance(value, bool) or not isinstance(value, int):
272 raise CliError(f"{context} must be an integer")
273 if value < 1 or value > maximum:
274 raise CliError(f"{context} must be between 1 and {maximum}")
275 return value
278def canonical_json_hash(value: Any) -> str:
279 """Return SHA-256 for canonical UTF-8 JSON."""
280 payload = json.dumps(
281 value,
282 ensure_ascii=False,
283 allow_nan=False,
284 sort_keys=True,
285 separators=(",", ":"),
286 ).encode("utf-8")
287 return hashlib.sha256(payload).hexdigest()
290def sha256_file(path: Path, *, max_bytes: int = MAX_ASSET_BYTES) -> str:
291 """Hash a bounded regular file without loading it all into memory."""
292 checked = checked_input_file(path, max_bytes=max_bytes)
293 digest = hashlib.sha256()
294 with checked.open("rb") as handle:
295 while True:
296 block = handle.read(1024 * 1024)
297 if not block:
298 break
299 digest.update(block)
300 return digest.hexdigest()
303def resolve_local_asset(
304 manifest_path: Path,
305 relative_value: Any,
306 *,
307 suffixes: set[str],
308) -> Path:
309 """Resolve a manifest-relative local file while preventing path escape."""
310 raw = require_string(relative_value, context="asset.path", maximum=512)
311 if _REMOTE_OR_SCHEME.match(raw) or raw.startswith(("/", "\\")):
312 raise CliError(f"asset.path must be a relative local path: {raw!r}")
313 if "\\" in raw:
314 raise CliError("asset.path must use forward slashes")
315 relative = Path(raw)
316 if any(part in {"", ".", ".."} for part in relative.parts):
317 raise CliError(f"asset.path contains an unsafe segment: {raw!r}")
318 root = manifest_path.parent.resolve()
319 candidate = root / relative
320 resolved = candidate.resolve()
321 try:
322 resolved.relative_to(root)
323 except ValueError as exc:
324 raise CliError(f"asset.path escapes the manifest directory: {raw!r}") from exc
325 return checked_input_file(
326 candidate,
327 max_bytes=MAX_ASSET_BYTES,
328 suffixes=suffixes,
329 )
332def has_placeholder(value: Any) -> tuple[str, str] | None:
333 """Return the first JSON path and placeholder string, if present."""
335 def walk(item: Any, path: str) -> tuple[str, str] | None:
336 if isinstance(item, str):
337 for pattern in _PLACEHOLDER_PATTERNS:
338 if pattern.search(item):
339 return path, item
340 return None
341 if isinstance(item, list):
342 for index, child in enumerate(item):
343 found = walk(child, f"{path}[{index}]")
344 if found:
345 return found
346 return None
347 if isinstance(item, dict):
348 for key, child in item.items():
349 found = walk(child, f"{path}.{key}")
350 if found:
351 return found
352 return None
354 return walk(value, "$")
357def parse_aware_datetime(value: Any, *, context: str) -> str:
358 """Validate an ISO 8601 timestamp that includes a UTC offset."""
359 text = require_string(value, context=context, maximum=64)
360 normalized = text[:-1] + "+00:00" if text.endswith("Z") else text
361 try:
362 parsed = datetime.fromisoformat(normalized)
363 except ValueError as exc:
364 raise CliError(f"{context} must be an ISO 8601 timestamp") from exc
365 if parsed.tzinfo is None or parsed.utcoffset() is None:
366 raise CliError(f"{context} must include a UTC offset")
367 return text
370def parse_hex_color(value: Any, *, context: str) -> str:
371 """Validate and normalize an opaque six-digit sRGB color."""
372 text = require_string(value, context=context, maximum=7)
373 if not _HEX_COLOR.fullmatch(text):
374 raise CliError(f"{context} must be a six-digit hex color such as #1A2B3C")
375 return text.upper()
378def _linear_channel(channel: int) -> float:
379 encoded = channel / 255.0
380 if encoded <= 0.04045:
381 return encoded / 12.92
382 return ((encoded + 0.055) / 1.055) ** 2.4
385def relative_luminance(color: str) -> float:
386 channels = tuple(int(color[index : index + 2], 16) for index in (1, 3, 5))
387 red, green, blue = (_linear_channel(channel) for channel in channels)
388 return 0.2126 * red + 0.7152 * green + 0.0722 * blue
391def contrast_ratio(first: str, second: str) -> float:
392 """Return the WCAG contrast ratio of two normalized sRGB colors."""
393 first_luminance = relative_luminance(first)
394 second_luminance = relative_luminance(second)
395 lighter = max(first_luminance, second_luminance)
396 darker = min(first_luminance, second_luminance)
397 return (lighter + 0.05) / (darker + 0.05)