Skills › Scientific Agent Skills › scientific-visualization › _common.py
_common.py
142 lines · skills/scientific-visualization/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared, network-free safety helpers for visualization command-line tools."""
4from __future__ import annotations
6import json
7import os
8import stat
9import tempfile
10from pathlib import Path
11from typing import Any
13MAX_INPUT_BYTES = 200 * 1024 * 1024
14MAX_REPORT_BYTES = 4 * 1024 * 1024
17class CliError(ValueError):
18 """An expected command-line validation error."""
21def checked_input_file(
22 value: str | os.PathLike[str],
23 *,
24 max_bytes: int = MAX_INPUT_BYTES,
25) -> Path:
26 """Return a bounded regular input file, rejecting symlinks."""
27 path = Path(value).expanduser()
28 if path.is_symlink():
29 raise CliError(f"input must not be a symlink: {path}")
30 try:
31 info = path.stat()
32 except OSError as exc:
33 raise CliError(f"cannot access input file {path}: {exc}") from exc
34 if not stat.S_ISREG(info.st_mode):
35 raise CliError(f"input is not a regular file: {path}")
36 if info.st_size > max_bytes:
37 raise CliError(
38 f"input is {info.st_size} bytes; limit is {max_bytes} bytes"
39 )
40 return path.resolve()
43def checked_output_file(
44 value: str | os.PathLike[str],
45 *,
46 force: bool = False,
47 mkdir: bool = False,
48) -> Path:
49 """Validate an explicit output path without following a destination symlink."""
50 path = Path(value).expanduser()
51 if path.name in {"", ".", ".."}:
52 raise CliError("output must name a file")
53 if path.is_symlink():
54 raise CliError(f"output must not be a symlink: {path}")
56 parent = path.parent
57 if mkdir:
58 parent.mkdir(parents=True, exist_ok=True)
59 if not parent.exists() or not parent.is_dir():
60 raise CliError(f"output parent directory does not exist: {parent}")
61 if parent.is_symlink():
62 raise CliError(f"output parent must not be a symlink: {parent}")
64 if path.exists():
65 if not path.is_file():
66 raise CliError(f"output exists and is not a regular file: {path}")
67 if not force:
68 raise CliError(f"refusing to overwrite existing output: {path}")
69 return parent.resolve() / path.name
72def atomic_write_bytes(path: Path, payload: bytes, *, force: bool = False) -> None:
73 """Write bytes through a same-directory temporary file."""
74 destination = checked_output_file(path, force=force)
75 descriptor, temporary_name = tempfile.mkstemp(
76 prefix=f".{destination.name}.",
77 suffix=".tmp",
78 dir=destination.parent,
79 )
80 temporary = Path(temporary_name)
81 try:
82 with os.fdopen(descriptor, "wb") as handle:
83 handle.write(payload)
84 handle.flush()
85 os.fsync(handle.fileno())
86 os.chmod(temporary, 0o600)
87 if force:
88 os.replace(temporary, destination)
89 else:
90 # A second existence check followed by replace still races another
91 # writer. Linking a same-filesystem temporary refuses collisions.
92 try:
93 os.link(temporary, destination)
94 except FileExistsError as exc:
95 raise CliError(f"refusing to overwrite existing output: {destination}") from exc
96 finally:
97 try:
98 temporary.unlink()
99 except FileNotFoundError:
100 pass
103def emit_json(
104 document: dict[str, Any],
105 *,
106 output: str | os.PathLike[str] | None = None,
107 force: bool = False,
108) -> None:
109 """Print deterministic JSON or write it safely to an explicit file."""
110 payload = (
111 json.dumps(document, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
112 ).encode("utf-8")
113 if len(payload) > MAX_REPORT_BYTES:
114 raise CliError(
115 f"report is {len(payload)} bytes; limit is {MAX_REPORT_BYTES} bytes"
116 )
117 if output is None:
118 print(payload.decode("utf-8"), end="")
119 return
120 atomic_write_bytes(Path(output), payload, force=force)
123def positive_int(value: str) -> int:
124 """Argparse converter for strictly positive integers."""
125 try:
126 parsed = int(value)
127 except ValueError as exc:
128 raise CliError(f"expected an integer, got {value!r}") from exc
129 if parsed <= 0:
130 raise CliError("value must be greater than zero")
131 return parsed
134def positive_float(value: str) -> float:
135 """Argparse converter for finite, strictly positive floats."""
136 try:
137 parsed = float(value)
138 except ValueError as exc:
139 raise CliError(f"expected a number, got {value!r}") from exc
140 if not (parsed > 0 and parsed < float("inf")):
141 raise CliError("value must be finite and greater than zero")
142 return parsed