from __future__ import annotations import argparse import hashlib import json import random import re import unicodedata from pathlib import Path from typing import Any, Iterable, Iterator SLOT_RE = re.compile(r"⟦SLOT:([A-Z_]+):(\d+)⟧") PLACEHOLDER_RE = re.compile( r"^(?:\$\{[^}]+\}|<[^>]+>|YOUR[_-]?[A-Z_]+|changeme|example|placeholder|replace(?:-me)?|)$", re.I, ) REFERENCE_RE = re.compile( r"^(?:[A-Za-z_]\w*|\$\{[^}]+\}|os\.environ\[[^]]+\]|[A-Za-z_$][\w$]*(?:\.[A-Za-z_$][\w$]*)+)$" ) LIVE_PREFIX_RE = re.compile(r"^(?:gh[opsru]_|xox[baprs]-|sk-|AKIA|AIza|glpat-|eyJ)", re.I) INTERPOLATION_RE = re.compile(r"^(?:[#$%]?\{[\w.\[\]'\"|-]{1,64}\}|%\([\w.-]{1,64}\)[a-z])$") CALL_RE = re.compile(r"^[A-Za-z_$][\w$.]*\([^()]{0,80}\)?$") INDEX_RE = re.compile(r"^[A-Za-z_$][\w$.]*\[[\w\"'.-]{0,64}\]?$") MARKER_VALUE_RE = re.compile(r"^\[(?:PII_)?REDACTED(?::[A-Z_]+)?\]$") def is_inert_value(value: str, quoted: bool = False) -> bool: if not value or not any(character.isalnum() for character in value): return True if PLACEHOLDER_RE.match(value) or MARKER_VALUE_RE.match(value) or SLOT_RE.search(value) or INTERPOLATION_RE.match(value): return True if LIVE_PREFIX_RE.match(value): return False if quoted: return False return bool(REFERENCE_RE.match(value) or CALL_RE.match(value) or INDEX_RE.match(value) or value.startswith(("//", "http://", "https://", "{", ":"))) def nfc(value: str) -> str: return unicodedata.normalize("NFC", value) def stable_id(*parts: object, length: int = 20) -> str: return hashlib.sha256("\0".join(map(str, parts)).encode()).hexdigest()[:length] def seeded_rng(seed: int, *namespace: object) -> random.Random: return random.Random(int(stable_id(seed, *namespace), 16)) def read_jsonl(path: Path) -> Iterator[dict[str, Any]]: if not path.exists(): return with path.open(encoding="utf-8") as stream: for line_number, line in enumerate(stream, 1): if line.strip(): try: yield json.loads(line) except json.JSONDecodeError as exc: raise ValueError(f"invalid JSONL in {path}:{line_number}") from exc def write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> int: path.parent.mkdir(parents=True, exist_ok=True) count = 0 with path.open("w", encoding="utf-8", newline="\n") as stream: for row in rows: stream.write(json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) stream.write("\n") count += 1 return count def write_report(staging: Path, name: str, **counts: object) -> None: report = staging / "reports" / f"{name}.json" report.parent.mkdir(parents=True, exist_ok=True) report.write_text(json.dumps(counts, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") def add_common_args(parser: argparse.ArgumentParser) -> None: parser.add_argument("--seed", type=int, default=42) parser.add_argument("--staging", type=Path, default=Path("staging"))