Skills › Scientific Agent Skills › market-research-reports › _common.py
_common.py
314 lines · skills/market-research-reports/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared, dependency-free validation helpers for local market-research CLIs."""
4from __future__ import annotations
6import csv
7import json
8import math
9import os
10import re
11import tempfile
12from datetime import date
13from pathlib import Path
14from typing import Any, Iterable
16MAX_FILE_BYTES = 5 * 1024 * 1024
17MAX_ROWS = 10_000
18MAX_CELL_CHARS = 20_000
19MAX_IDENTIFIER_CHARS = 96
21IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._:-]{0,95}$")
22CURRENCY_RE = re.compile(r"^[A-Z]{3}$")
25class ValidationError(ValueError):
26 """A deterministic, user-correctable input validation error."""
29def safe_input_path(raw_path: str | Path, suffixes: Iterable[str]) -> Path:
30 """Resolve a bounded regular local file and reject symlink inputs."""
31 path = Path(raw_path).expanduser()
32 if path.is_symlink():
33 raise ValidationError(f"symlink inputs are not allowed: {path}")
34 try:
35 resolved = path.resolve(strict=True)
36 except FileNotFoundError as exc:
37 raise ValidationError(f"input file does not exist: {path}") from exc
38 if not resolved.is_file():
39 raise ValidationError(f"input path is not a regular file: {resolved}")
40 allowed = {suffix.lower() for suffix in suffixes}
41 if resolved.suffix.lower() not in allowed:
42 choices = ", ".join(sorted(allowed))
43 raise ValidationError(f"expected one of [{choices}], got: {resolved.suffix}")
44 size = resolved.stat().st_size
45 if size > MAX_FILE_BYTES:
46 raise ValidationError(
47 f"input exceeds {MAX_FILE_BYTES} bytes: {resolved} ({size} bytes)"
48 )
49 return resolved
52def safe_output_path(
53 raw_path: str | Path, suffix: str, *, force: bool = False
54) -> Path:
55 """Resolve an output in an existing directory without implicit overwrites."""
56 path = Path(raw_path).expanduser()
57 if path.suffix.lower() != suffix.lower():
58 raise ValidationError(f"output must use {suffix}: {path}")
59 try:
60 parent = path.parent.resolve(strict=True)
61 except FileNotFoundError as exc:
62 raise ValidationError(f"output parent does not exist: {path.parent}") from exc
63 if not parent.is_dir():
64 raise ValidationError(f"output parent is not a directory: {parent}")
65 resolved = parent / path.name
66 if resolved.is_symlink():
67 raise ValidationError(f"symlink outputs are not allowed: {resolved}")
68 if resolved.exists():
69 if not resolved.is_file():
70 raise ValidationError(f"output is not a regular file: {resolved}")
71 if not force:
72 raise ValidationError(
73 f"output already exists; pass --force to replace it: {resolved}"
74 )
75 return resolved
78def read_json(raw_path: str | Path) -> Any:
79 """Read one bounded UTF-8 JSON document."""
80 path = safe_input_path(raw_path, {".json"})
81 try:
82 with path.open("r", encoding="utf-8") as handle:
83 return json.load(handle)
84 except UnicodeDecodeError as exc:
85 raise ValidationError(f"JSON must be UTF-8: {path}") from exc
86 except json.JSONDecodeError as exc:
87 raise ValidationError(
88 f"invalid JSON at line {exc.lineno}, column {exc.colno}: {exc.msg}"
89 ) from exc
92def read_csv_records(
93 raw_path: str | Path,
94 *,
95 required_fields: Iterable[str],
96 max_rows: int = MAX_ROWS,
97) -> list[dict[str, str]]:
98 """Read strict UTF-8 CSV records with unique headers and bounded cells."""
99 path = safe_input_path(raw_path, {".csv"})
100 required = tuple(required_fields)
101 try:
102 with path.open("r", encoding="utf-8-sig", newline="") as handle:
103 reader = csv.DictReader(handle)
104 headers = reader.fieldnames
105 if not headers:
106 raise ValidationError(f"CSV has no header: {path}")
107 normalized = [header.strip() for header in headers]
108 if any(not header for header in normalized):
109 raise ValidationError("CSV headers must not be blank")
110 if len(normalized) != len(set(normalized)):
111 raise ValidationError("CSV headers must be unique")
112 missing = sorted(set(required) - set(normalized))
113 if missing:
114 raise ValidationError(
115 f"CSV is missing required columns: {', '.join(missing)}"
116 )
117 reader.fieldnames = normalized
118 records: list[dict[str, str]] = []
119 for line_number, row in enumerate(reader, start=2):
120 if line_number - 1 > max_rows:
121 raise ValidationError(f"CSV exceeds {max_rows} data rows")
122 if None in row:
123 raise ValidationError(
124 f"row {line_number} has more cells than the header"
125 )
126 cleaned: dict[str, str] = {}
127 for key, value in row.items():
128 cell = "" if value is None else value.strip()
129 if "\x00" in cell:
130 raise ValidationError(
131 f"row {line_number}, column {key} contains a NUL byte"
132 )
133 if len(cell) > MAX_CELL_CHARS:
134 raise ValidationError(
135 f"row {line_number}, column {key} exceeds "
136 f"{MAX_CELL_CHARS} characters"
137 )
138 cleaned[key] = cell
139 if any(cleaned.values()):
140 records.append(cleaned)
141 except UnicodeDecodeError as exc:
142 raise ValidationError(f"CSV must be UTF-8: {path}") from exc
143 except csv.Error as exc:
144 raise ValidationError(f"invalid CSV: {exc}") from exc
145 if not records:
146 raise ValidationError("CSV must contain at least one data row")
147 return records
150def require_object(value: Any, context: str) -> dict[str, Any]:
151 if not isinstance(value, dict):
152 raise ValidationError(f"{context} must be a JSON object")
153 return value
156def require_list(
157 value: Any,
158 context: str,
159 *,
160 minimum: int = 0,
161 maximum: int = MAX_ROWS,
162) -> list[Any]:
163 if not isinstance(value, list):
164 raise ValidationError(f"{context} must be a JSON array")
165 if not minimum <= len(value) <= maximum:
166 raise ValidationError(
167 f"{context} must contain between {minimum} and {maximum} items"
168 )
169 return value
172def require_text(
173 value: Any,
174 context: str,
175 *,
176 allow_empty: bool = False,
177 maximum: int = MAX_CELL_CHARS,
178) -> str:
179 if not isinstance(value, str):
180 raise ValidationError(f"{context} must be a string")
181 text = value.strip()
182 if not allow_empty and not text:
183 raise ValidationError(f"{context} must not be empty")
184 if len(text) > maximum:
185 raise ValidationError(f"{context} exceeds {maximum} characters")
186 if "\x00" in text:
187 raise ValidationError(f"{context} contains a NUL byte")
188 return text
191def require_identifier(value: Any, context: str) -> str:
192 identifier = require_text(
193 value, context, maximum=MAX_IDENTIFIER_CHARS, allow_empty=False
194 )
195 if not IDENTIFIER_RE.fullmatch(identifier):
196 raise ValidationError(
197 f"{context} must match {IDENTIFIER_RE.pattern}: {identifier!r}"
198 )
199 return identifier
202def require_unique_identifiers(values: Iterable[str], context: str) -> None:
203 seen: set[str] = set()
204 duplicates: set[str] = set()
205 for value in values:
206 if value in seen:
207 duplicates.add(value)
208 seen.add(value)
209 if duplicates:
210 raise ValidationError(
211 f"{context} contains duplicate IDs: {', '.join(sorted(duplicates))}"
212 )
215def parse_iso_date(value: Any, context: str) -> str:
216 text = require_text(value, context, maximum=10)
217 if not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", text):
218 raise ValidationError(f"{context} must be YYYY-MM-DD: {text!r}")
219 try:
220 date.fromisoformat(text)
221 except ValueError as exc:
222 raise ValidationError(f"{context} must be YYYY-MM-DD: {text!r}") from exc
223 return text
226def parse_year(value: Any, context: str) -> int:
227 if type(value) is int:
228 parsed = value
229 elif isinstance(value, str) and re.fullmatch(r"[0-9]{4}", value.strip()):
230 parsed = int(value.strip())
231 else:
232 raise ValidationError(f"{context} must be an integer year")
233 if not 1800 <= parsed <= 2200:
234 raise ValidationError(f"{context} must be between 1800 and 2200")
235 return parsed
238def parse_number(
239 value: Any,
240 context: str,
241 *,
242 minimum: float,
243 maximum: float,
244) -> float:
245 if isinstance(value, bool):
246 raise ValidationError(f"{context} must be numeric")
247 try:
248 number = float(value)
249 except (TypeError, ValueError) as exc:
250 raise ValidationError(f"{context} must be numeric") from exc
251 if not math.isfinite(number):
252 raise ValidationError(f"{context} must be finite")
253 if not minimum <= number <= maximum:
254 raise ValidationError(
255 f"{context} must be between {minimum} and {maximum}: {number}"
256 )
257 return number
260def parse_fraction(value: Any, context: str) -> float:
261 return parse_number(value, context, minimum=0.0, maximum=1.0)
264def parse_currency(value: Any, context: str, *, allow_empty: bool = False) -> str:
265 text = require_text(value, context, allow_empty=allow_empty, maximum=3).upper()
266 if not text and allow_empty:
267 return ""
268 if not CURRENCY_RE.fullmatch(text):
269 raise ValidationError(f"{context} must be a three-letter currency code")
270 return text
273def split_ids(value: Any, context: str, *, allow_empty: bool = False) -> list[str]:
274 text = require_text(value, context, allow_empty=allow_empty)
275 if not text:
276 return []
277 identifiers = [part.strip() for part in text.split(";")]
278 if any(not part for part in identifiers):
279 raise ValidationError(f"{context} contains an empty semicolon-delimited ID")
280 parsed = [require_identifier(part, context) for part in identifiers]
281 require_unique_identifiers(parsed, context)
282 return parsed
285def write_json_report(
286 data: Any, output: str | Path | None, *, force: bool = False
287) -> None:
288 """Print JSON or atomically write it to an explicitly selected local file."""
289 serialized = json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
290 if output is None:
291 print(serialized, end="")
292 return
293 destination = safe_output_path(output, ".json", force=force)
294 temporary_name: str | None = None
295 try:
296 with tempfile.NamedTemporaryFile(
297 "w",
298 encoding="utf-8",
299 dir=destination.parent,
300 prefix=f".{destination.name}.",
301 suffix=".tmp",
302 delete=False,
303 ) as handle:
304 handle.write(serialized)
305 temporary_name = handle.name
306 os.replace(temporary_name, destination)
307 finally:
308 if temporary_name and Path(temporary_name).exists():
309 Path(temporary_name).unlink()
312def error_exit(exc: ValidationError) -> int:
313 print(f"ERROR: {exc}", file=os.sys.stderr)
314 return 2