File size: 3,104 Bytes
0936c1f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | 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"))
|