scripts/release/release_common.py

from __future__ import annotations

import csv
import fnmatch
import hashlib
import io
import json
import os
import re
import shutil
import tempfile
import zipfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence

TOOLKIT_VERSION = "1.0.1"
EXPECTED_ACCOUNT = "AiLLMBS"
DEFAULT_REPO_ID = "AiLLMBS/neurocell-lm-ad"

TEXT_SUFFIXES = {
    ".md", ".txt", ".py", ".ps1", ".json", ".csv", ".tsv", ".toml",
    ".yaml", ".yml", ".cff", ".ini", ".cfg", ".gitattributes", ".gitignore",
}

PROTECTED_PATTERNS = [
    "**/*.h5ad", "**/*.h5", "**/*.loom", "**/*.zarr/**", "**/*.parquet",
    "**/*.joblib", "**/*.pkl", "**/*.pickle", "**/*.safetensors", "**/*.bin",
    "**/*.pt", "**/*.pth", "**/*.ckpt", "**/*.npz", "**/*.npy",
    "**/chunks/**", "**/.cache/**", "**/hf_cache/**", "**/__pycache__/**",
    "**/*.token", "**/*.pem", "**/*.key", "**/.env", "**/.env.*",
    "**/*lockbox*", "**/*sealed*", "**/*predictions*", "**/*donor*manifest*",
]

SECRET_PATTERNS: dict[str, re.Pattern[str]] = {
    "hf_token": re.compile(r"\bhf_[A-Za-z0-9]{20,}\b"),
    "aws_access_key": re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
    "aws_secret_assignment": re.compile(
        r"(?i)(aws_secret_access_key|secret_access_key)\s*[:=]\s*['\"]?[^\s'\"]{12,}"
    ),
    "private_key": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----"),
    "generic_bearer": re.compile(r"(?i)authorization\s*:\s*bearer\s+[A-Za-z0-9._~+/=-]{12,}"),
    "generic_token_assignment": re.compile(
        r"(?i)\b(token|api[_-]?key|access[_-]?key|client[_-]?secret)\b\s*[:=]\s*['\"]([A-Za-z0-9._~+/=-]{18,})['\"]"
    ),
}

ABSOLUTE_PATH_PATTERNS: dict[str, re.Pattern[str]] = {
    "windows_drive_path": re.compile(r"(?<![A-Za-z0-9])(?:[A-Za-z]:\\[^\r\n\"'<>|]+)"),
    "windows_user_path": re.compile(r"(?i)C:\\Users\\[^\\\s]+"),
    "unix_home_path": re.compile(r"(?<![A-Za-z0-9])/(?:home|Users)/[^/\s]+(?:/[^\s\"']*)?"),
}

EMAIL_PATTERN = re.compile(r"(?i)\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b")
DONOR_PATTERN = re.compile(r"\bH(?:20|21)\.33\.\d{3}\b")
CELL_BARCODE_PATTERN = re.compile(r"\b[ACGT]{16}-\d+(?:_[A-Za-z0-9]+)?\b")

ALLOWED_PUBLIC_EMAILS = {
    "terms@alleninstitute.org",
    "communications@alleninstitute.org",
    "security@huggingface.co",
}


class ReleaseError(RuntimeError):
    pass


@dataclass(frozen=True)
class InventoryRow:
    source_path: str
    release_path: str
    classification: str
    include: bool
    reason: str
    contains_donor_level_data: bool
    contains_absolute_paths: bool
    license_or_redistribution_risk: str
    cleanup_action: str
    source_sha256: str = ""
    release_sha256: str = ""
    source_size_bytes: int | None = None
    release_size_bytes: int | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "source_path": self.source_path,
            "release_path": self.release_path,
            "classification": self.classification,
            "include": self.include,
            "reason": self.reason,
            "contains_donor_level_data": self.contains_donor_level_data,
            "contains_absolute_paths": self.contains_absolute_paths,
            "license_or_redistribution_risk": self.license_or_redistribution_risk,
            "cleanup_action": self.cleanup_action,
            "source_sha256": self.source_sha256,
            "release_sha256": self.release_sha256,
            "source_size_bytes": self.source_size_bytes,
            "release_size_bytes": self.release_size_bytes,
        }


def now_utc() -> str:
    return datetime.now(timezone.utc).isoformat()


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sha256_file(path: Path, chunk_size: int = 8 * 1024 * 1024) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(chunk_size), b""):
            digest.update(block)
    return digest.hexdigest()


def canonical_json_bytes(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    ).encode("utf-8")


def canonical_json_sha256(value: Any) -> str:
    return sha256_bytes(canonical_json_bytes(value))


def read_json(path: Path) -> Any:
    with path.open("r", encoding="utf-8-sig") as handle:
        return json.load(handle)


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(value, indent=2, ensure_ascii=False, sort_keys=True) + "\n",
        encoding="utf-8",
        newline="\n",
    )


def read_csv(path: Path) -> list[dict[str, str]]:
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))


def write_csv(path: Path, rows: Sequence[Mapping[str, Any]], fieldnames: Sequence[str] | None = None) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    if fieldnames is None:
        keys: list[str] = []
        seen: set[str] = set()
        for row in rows:
            for key in row:
                if key not in seen:
                    seen.add(key)
                    keys.append(str(key))
        fieldnames = keys
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(fieldnames), extrasaction="ignore")
        writer.writeheader()
        for row in rows:
            writer.writerow({key: _csv_value(row.get(key)) for key in fieldnames})


def _csv_value(value: Any) -> Any:
    if value is None:
        return ""
    if isinstance(value, bool):
        return str(value).lower()
    if isinstance(value, (dict, list, tuple)):
        return json.dumps(value, sort_keys=True, ensure_ascii=False)
    return value


def normalize_relpath(path: str | Path) -> str:
    value = str(path).replace("\\", "/")
    while value.startswith("./"):
        value = value[2:]
    value = value.lstrip("/")
    while "//" in value:
        value = value.replace("//", "/")
    return value


def is_text_path(path: Path) -> bool:
    if path.name in {"README", "LICENSE", "NOTICE", ".gitignore", ".gitattributes"}:
        return True
    return path.suffix.lower() in TEXT_SUFFIXES


def read_text_safely(path: Path, max_bytes: int = 8 * 1024 * 1024) -> str | None:
    if path.stat().st_size > max_bytes or not is_text_path(path):
        return None
    data = path.read_bytes()
    if b"\x00" in data:
        return None
    for encoding in ("utf-8-sig", "utf-8", "cp1252"):
        try:
            return data.decode(encoding)
        except UnicodeDecodeError:
            continue
    return None


def sanitize_public_text(text: str) -> tuple[str, list[dict[str, str]]]:
    actions: list[dict[str, str]] = []
    replacements = [
        (re.compile(r"(?i)D:\\LLM_Hugging_Face\\neurocell-lm-ad(?:\\[^\r\n\"']*)?"), "${NEUROCELL_PROJECT_ROOT}"),
        (re.compile(r"(?i)D:/LLM_Hugging_Face/neurocell-lm-ad(?:/[^\r\n\"']*)?"), "${NEUROCELL_PROJECT_ROOT}"),
        (re.compile(r"(?i)C:\\Users\\[^\\\s]+"), "<LOCAL_USER_HOME>"),
        (re.compile(r"(?<![A-Za-z0-9])/(?:home|Users)/[^/\s]+"), "<LOCAL_USER_HOME>"),
    ]
    sanitized = text
    for pattern, replacement in replacements:
        sanitized, count = pattern.subn(replacement, sanitized)
        if count:
            actions.append({"pattern": pattern.pattern, "replacement": replacement, "count": str(count)})
    return sanitized, actions


def scan_text(text: str, *, allow_donor_ids: bool = False, allow_emails: Iterable[str] = ()) -> list[dict[str, Any]]:
    findings: list[dict[str, Any]] = []
    allowed_emails = set(ALLOWED_PUBLIC_EMAILS) | {email.lower() for email in allow_emails}
    for name, pattern in SECRET_PATTERNS.items():
        for match in pattern.finditer(text):
            findings.append({"severity": "ERROR", "kind": name, "offset": match.start(), "match": _redact(match.group(0))})
    for name, pattern in ABSOLUTE_PATH_PATTERNS.items():
        for match in pattern.finditer(text):
            findings.append({"severity": "ERROR", "kind": name, "offset": match.start(), "match": _redact_path(match.group(0))})
    for match in EMAIL_PATTERN.finditer(text):
        email = match.group(0).lower()
        if email not in allowed_emails:
            findings.append({"severity": "ERROR", "kind": "email_address", "offset": match.start(), "match": _redact(email)})
    if not allow_donor_ids:
        for match in DONOR_PATTERN.finditer(text):
            findings.append({"severity": "ERROR", "kind": "donor_identifier", "offset": match.start(), "match": match.group(0)})
        for match in CELL_BARCODE_PATTERN.finditer(text):
            findings.append({"severity": "ERROR", "kind": "cell_identifier", "offset": match.start(), "match": _redact(match.group(0))})
    return findings


def _redact(value: str) -> str:
    if len(value) <= 8:
        return "<redacted>"
    return value[:4] + "..." + value[-4:]


def _redact_path(value: str) -> str:
    if len(value) <= 24:
        return "<absolute-path>"
    return value[:3] + "..." + Path(value.replace("\\", "/")).name


def matches_any(path: str | Path, patterns: Sequence[str]) -> bool:
    rel = normalize_relpath(path)
    candidates = {rel, rel.lower()}
    for pattern in patterns:
        pat = normalize_relpath(pattern)
        variants = [pat]
        if pat.startswith("**/"):
            variants.append(pat[3:])
        for candidate in candidates:
            for variant in variants:
                if fnmatch.fnmatchcase(candidate, variant) or fnmatch.fnmatchcase(candidate, variant.lower()):
                    return True
    return False


def verify_file_hash(path: Path, expected: str, label: str | None = None) -> str:
    if not path.exists():
        raise ReleaseError(f"Missing required file: {path}")
    observed = sha256_file(path)
    if observed.lower() != expected.lower():
        raise ReleaseError(
            f"SHA-256 mismatch for {label or path.name}: expected {expected.lower()}, observed {observed.lower()}."
        )
    return observed


def index_zip_by_basename(path: Path) -> dict[str, list[zipfile.ZipInfo]]:
    result: dict[str, list[zipfile.ZipInfo]] = {}
    with zipfile.ZipFile(path) as archive:
        for info in archive.infolist():
            if info.is_dir():
                continue
            result.setdefault(Path(info.filename).name, []).append(info)
    return result


def read_zip_member_by_basename(path: Path, basename: str) -> bytes:
    with zipfile.ZipFile(path) as archive:
        matches = [info for info in archive.infolist() if not info.is_dir() and Path(info.filename).name == basename]
        if len(matches) != 1:
            raise ReleaseError(f"Expected exactly one {basename!r} in {path.name}; found {len(matches)}.")
        return archive.read(matches[0])


def verify_zip_against_manifest(zip_path: Path, manifest_rows: Sequence[Mapping[str, str]]) -> dict[str, Any]:
    by_name = index_zip_by_basename(zip_path)
    verified: list[dict[str, Any]] = []
    for row in manifest_rows:
        source = row.get("Path") or row.get("path") or ""
        expected = row.get("Hash") or row.get("hash") or ""
        basename = Path(source.replace("\\", "/")).name
        matches = by_name.get(basename, [])
        if len(matches) != 1:
            raise ReleaseError(
                f"Final archive must contain exactly one {basename!r}; found {len(matches)} in {zip_path.name}."
            )
        with zipfile.ZipFile(zip_path) as archive:
            data = archive.read(matches[0])
        observed = sha256_bytes(data)
        if observed.lower() != expected.lower():
            raise ReleaseError(
                f"Archive member hash mismatch for {basename}: expected {expected.lower()}, observed {observed.lower()}."
            )
        verified.append({"file": basename, "sha256": observed, "size_bytes": len(data)})
    return {"archive": zip_path.name, "verified_members": verified, "member_count": len(verified)}


def extract_zip_member_to(path: Path, member_basename: str, destination: Path) -> Path:
    data = read_zip_member_by_basename(path, member_basename)
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_bytes(data)
    return destination


def locate_project_file(project_root: Path, relative_path: str, script_archive: Path | None = None) -> tuple[str, bytes]:
    direct = project_root / Path(relative_path)
    if direct.exists() and direct.is_file():
        return str(direct), direct.read_bytes()
    if script_archive and script_archive.exists():
        basename = Path(relative_path).name
        data = read_zip_member_by_basename(script_archive, basename)
        return f"{script_archive}::{basename}", data
    raise ReleaseError(f"Could not locate approved source file: {relative_path}")


def ensure_separate_release_dir(project_root: Path, release_dir: Path) -> None:
    project = project_root.resolve()
    release = release_dir.resolve()
    if project == release:
        raise ReleaseError("Release directory cannot equal the frozen project directory.")
    try:
        release.relative_to(project)
    except ValueError:
        return
    raise ReleaseError("Release directory must not be inside the frozen project directory.")


def parse_snapshot_inventory(path: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    pattern = re.compile(r"^\[FILE\]\s+(.+?)\s+\|\s+Size:\s+(\d+)\s+bytes\s+\|")
    with path.open("r", encoding="utf-8-sig", errors="replace") as handle:
        for line in handle:
            match = pattern.match(line.rstrip("\r\n"))
            if not match:
                continue
            rows.append({"path": normalize_relpath(match.group(1)), "size_bytes": int(match.group(2))})
    if not rows:
        raise ReleaseError(f"No file inventory entries found in snapshot: {path}")
    return rows


def classify_project_path(path: str, size_bytes: int, approved_source_names: set[str]) -> dict[str, Any]:
    rel = normalize_relpath(path)
    lower = rel.lower()
    name = Path(rel).name
    donor_level = any(token in lower for token in [
        "donor", "prediction", "cell_label", "cell_metadata", "sample_manifest", "index_manifest",
        "score_manifest", "common_cell", "metadata.parquet", "eligibility.csv", "per_donor",
    ])
    absolute = False
    risk = "low"
    cleanup = "none"
    classification = "LOCAL_ONLY"
    include = False
    reason = "Local research artifact not required in the public release."
    release_path = ""

    if lower.startswith(("data_external/", "data_interim/")):
        classification = "PROHIBITED_OR_UNCERTAIN"
        risk = "high: upstream SEA-AD/Allen or source data"
        reason = "Raw or upstream-derived source data are not redistributed."
    elif lower.startswith("data_processed/"):
        classification = "LOCAL_ONLY"
        risk = "high: donor/cell-level derived data"
        reason = "Processed cell-, donor-, feature-, or prediction-level artifacts remain local."
    elif lower.startswith("outputs/") and ("chunks/" in lower or lower.endswith((".npz", ".npy"))):
        classification = "PROHIBITED_OR_UNCERTAIN"
        risk = "high: embedding vectors and cell-level metadata"
        reason = "C2S embedding chunks and vector outputs are excluded."
    elif lower.startswith("models/") or lower.endswith((".joblib", ".pkl", ".pickle", ".safetensors", ".bin", ".pt", ".pth")):
        classification = "PROHIBITED_OR_UNCERTAIN"
        risk = "medium/high: serialized executable model and derivative-rights uncertainty"
        reason = "Serialized model artifacts are excluded; recreate locally from the frozen code and source data."
    elif "lockbox" in lower or "sealed" in lower:
        classification = "PROHIBITED_OR_UNCERTAIN"
        risk = "critical: lockbox-related"
        reason = "Lockbox-related files and identifiers must never be uploaded."
    elif lower.startswith(("__pycache__/", ".cache/", "hf_cache/", "logs/")) or lower.endswith((".pyc", ".pyo", ".tmp", ".part")):
        classification = "LOCAL_ONLY"
        risk = "low"
        reason = "Cache, log, compiled, or temporary file."
    elif name in approved_source_names:
        classification = "PUBLIC_AS_IS"
        include = True
        reason = "Approved outcome-safe reproduction source file."
        release_path = f"scripts/research_pipeline/{name}"
    elif re.match(r"^(?:13_inspect_mtg_raw_matrix(?:_v2)?|16_build_ranked_gene_records|17_audit_c2s_tokenizer|27_extract_c2s_frozen_embeddings|45_build_mec_ranked_gene_records|51_fit_ad_severity_training_baselines)\.py$", name):
        classification = "ARCHIVE_ONLY"
        reason = "Superseded or failed implementation preserved only in the frozen research archive."
    elif lower.startswith("reports/"):
        if donor_level or lower.endswith(".parquet"):
            classification = "LOCAL_ONLY"
            risk = "high: donor/cell-level or local path content"
            reason = "Donor-level report or prediction-related audit remains local."
        else:
            classification = "PUBLIC_AFTER_CLEANUP"
            risk = "medium: absolute paths and potentially detailed identifiers"
            cleanup = "regenerate aggregate-only public summary"
            reason = "Only an aggregate, path-sanitized derivative is published."
    elif lower.startswith("configs/"):
        classification = "LOCAL_ONLY"
        risk = "medium: split/lockbox identifiers"
        reason = "Original configuration may expose donor assignments or lockbox policy details; publish a sanitized protocol instead."
    elif lower.endswith((".zip", ".7z", ".tar", ".gz")):
        classification = "ARCHIVE_ONLY"
        risk = "unknown bundled content"
        reason = "Research archive is retained locally and is not uploaded wholesale."
    elif lower.endswith((".py", ".ps1", ".md", ".txt")):
        classification = "ARCHIVE_ONLY"
        risk = "low/unknown"
        reason = "Not part of the approved public source subset."
    elif size_bytes > 50 * 1024 * 1024:
        classification = "LOCAL_ONLY"
        risk = "large generated artifact"
        reason = "Unexpectedly large file is excluded by default."

    return {
        "source_path": rel,
        "release_path": release_path,
        "classification": classification,
        "include": include,
        "reason": reason,
        "contains_donor_level_data": donor_level,
        "contains_absolute_paths": absolute,
        "license_or_redistribution_risk": risk,
        "cleanup_action": cleanup,
        "source_sha256": "",
        "release_sha256": "",
        "source_size_bytes": size_bytes,
        "release_size_bytes": None,
    }


def build_release_manifest(release_dir: Path, *, exclude: Sequence[str] = ()) -> dict[str, Any]:
    files: list[dict[str, Any]] = []
    for path in sorted(release_dir.rglob("*")):
        if not path.is_file():
            continue
        rel = normalize_relpath(path.relative_to(release_dir))
        if matches_any(rel, list(exclude)):
            continue
        files.append({
            "path": rel,
            "size_bytes": path.stat().st_size,
            "sha256": sha256_file(path),
        })
    return {
        "schema_version": "1.0",
        "generated_at_utc": now_utc(),
        "file_count": len(files),
        "total_size_bytes": sum(item["size_bytes"] for item in files),
        "files": files,
    }


def manifest_content_hash(manifest: Mapping[str, Any]) -> str:
    normalized = {
        "schema_version": manifest.get("schema_version"),
        "files": sorted(
            [
                {"path": item["path"], "size_bytes": int(item["size_bytes"]), "sha256": item["sha256"]}
                for item in manifest.get("files", [])
            ],
            key=lambda item: item["path"],
        ),
    }
    return canonical_json_sha256(normalized)


def verify_release_manifest(release_dir: Path, manifest_path: Path, *, allow_extra: Iterable[str] = ()) -> dict[str, Any]:
    manifest = read_json(manifest_path)
    expected = {item["path"]: item for item in manifest.get("files", [])}
    errors: list[str] = []
    verified: list[dict[str, Any]] = []
    for rel, item in expected.items():
        path = release_dir / Path(rel)
        if not path.exists():
            errors.append(f"missing:{rel}")
            continue
        observed_size = path.stat().st_size
        observed_hash = sha256_file(path)
        if observed_size != int(item["size_bytes"]):
            errors.append(f"size:{rel}")
        if observed_hash.lower() != str(item["sha256"]).lower():
            errors.append(f"sha256:{rel}")
        verified.append({"path": rel, "sha256": observed_hash, "size_bytes": observed_size})
    allowed = {normalize_relpath(item) for item in allow_extra}
    observed_paths = {
        normalize_relpath(path.relative_to(release_dir))
        for path in release_dir.rglob("*") if path.is_file()
    }
    extras = sorted(observed_paths - set(expected) - allowed)
    if extras:
        errors.extend(f"extra:{item}" for item in extras)
    return {
        "status": "PASS" if not errors else "FAIL",
        "errors": errors,
        "verified_files": len(verified),
        "manifest_file_count": len(expected),
        "content_sha256": manifest_content_hash(manifest),
    }


def scan_release_tree(
    release_dir: Path,
    *,
    deny_patterns: Sequence[str],
    max_file_size_bytes: int,
    allow_large_paths: Sequence[str] = (),
    allow_donor_paths: Sequence[str] = (),
    content_scan_exclude_paths: Sequence[str] = (),
) -> dict[str, Any]:
    findings: list[dict[str, Any]] = []
    scanned_files = 0
    for path in sorted(release_dir.rglob("*")):
        if not path.is_file():
            continue
        scanned_files += 1
        rel = normalize_relpath(path.relative_to(release_dir))
        if matches_any(rel, deny_patterns):
            findings.append({"severity": "ERROR", "kind": "denylisted_path", "path": rel})
        if path.stat().st_size > max_file_size_bytes and not matches_any(rel, allow_large_paths):
            findings.append({
                "severity": "ERROR", "kind": "unexpected_large_file", "path": rel,
                "size_bytes": path.stat().st_size, "limit_bytes": max_file_size_bytes,
            })
        if matches_any(rel, content_scan_exclude_paths):
            continue
        text = read_text_safely(path)
        if text is None:
            continue
        allow_donors = matches_any(rel, allow_donor_paths)
        for finding in scan_text(text, allow_donor_ids=allow_donors):
            findings.append({"path": rel, **finding})
    return {
        "status": "PASS" if not [f for f in findings if f["severity"] == "ERROR"] else "FAIL",
        "scanned_files": scanned_files,
        "findings": findings,
        "error_count": len([f for f in findings if f["severity"] == "ERROR"]),
    }


def copy_bytes(destination: Path, data: bytes) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_bytes(data)


def copy_text_sanitized(destination: Path, text: str) -> list[dict[str, str]]:
    sanitized, actions = sanitize_public_text(text)
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(sanitized.replace("\r\n", "\n"), encoding="utf-8", newline="\n")
    return actions


def format_float(value: Any, digits: int = 4, missing: str = "—") -> str:
    if value is None or value == "":
        return missing
    return f"{float(value):.{digits}f}"


def markdown_table(headers: Sequence[str], rows: Sequence[Sequence[Any]]) -> str:
    lines = [
        "| " + " | ".join(headers) + " |",
        "|" + "|".join(["---"] * len(headers)) + "|",
    ]
    for row in rows:
        lines.append("| " + " | ".join(str(value) for value in row) + " |")
    return "\n".join(lines)


def assert_no_nan_inf(value: Any, path: str = "root") -> None:
    import math
    if isinstance(value, dict):
        for key, item in value.items():
            assert_no_nan_inf(item, f"{path}.{key}")
    elif isinstance(value, list):
        for index, item in enumerate(value):
            assert_no_nan_inf(item, f"{path}[{index}]")
    elif isinstance(value, float) and (math.isnan(value) or math.isinf(value)):
        raise ReleaseError(f"Non-finite value at {path}")


def safe_rmtree(path: Path, *, protected_root: Path | None = None) -> None:
    resolved = path.resolve()
    if protected_root is not None:
        protected = protected_root.resolve()
        if resolved == protected:
            raise ReleaseError("Refusing to delete the protected project root.")
        try:
            resolved.relative_to(protected)
        except ValueError:
            pass
        else:
            raise ReleaseError("Refusing to delete a directory inside the protected project root.")
    if path.exists():
        shutil.rmtree(path)


def temporary_directory(parent: Path, prefix: str) -> Path:
    parent.mkdir(parents=True, exist_ok=True)
    return Path(tempfile.mkdtemp(prefix=prefix, dir=parent))