core/_shim.py

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"))