docs_examples.py

"""Shared engine for the documentation example generators.

Used by `update_export_examples.py` and `update_import_examples.py` to run the
Data Contract CLI and inject the resulting command + output snippets into the
docs pages between AUTOGENERATED markers.
"""

import re
import shutil
import subprocess
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent
DOCS = REPO / "docs" / "docs"
EXAMPLES_URL = "https://github.com/datacontract/datacontract-cli/blob/main/examples"

# JSX comments (not HTML comments) — MDX 3 rejects `<!-- -->`.
START = "{/* AUTOGENERATED EXAMPLE: do not edit by hand; regenerate with the update script. */}"
END = "{/* END AUTOGENERATED EXAMPLE */}"

# Per-language line-comment used as a truncation marker for excerpts.
COMMENT = {
    "yaml": "# …",
    "sql": "-- …",
    "python": "# …",
    "go": "// …",
    "text": "…",
    "turtle": "# …",
    "markdown": "<!-- … -->",
    "protobuf": "// …",
    "mermaid": "%% …",
    "json": "// …",
}


def datacontract_bin() -> str:
    """Locate the datacontract CLI, preferring the running interpreter's env."""
    local = Path(sys.executable).parent / "datacontract"
    return str(local) if local.exists() else (shutil.which("datacontract") or "datacontract")


def run_cli(args: list[str]) -> str:
    """Run `datacontract <args>` and return stdout, raising on failure."""
    result = subprocess.run([datacontract_bin(), *args], capture_output=True, text=True)
    if result.returncode != 0:
        raise RuntimeError(f"`datacontract {' '.join(args)}` failed:\n{result.stdout}{result.stderr}")
    return result.stdout.strip("\n")


def fence(lang: str, body: str, max_lines: int | None) -> str:
    """Render a fenced code block, truncating to max_lines with a comment marker."""
    lines = body.split("\n")
    if max_lines and len(lines) > max_lines:
        lines = lines[:max_lines] + [COMMENT.get(lang, "…")]
    return f"```{lang}\n" + "\n".join(lines) + "\n```"


def upsert(page: Path, block: str, fallback: re.Pattern) -> None:
    """Replace the AUTOGENERATED block in `page`.

    If the markers exist, replace between them. Otherwise replace the first
    region matching `fallback` (the hand-written example) in place, which
    installs the markers on the first run.
    """
    text = page.read_text()
    marked = f"{START}\n\n{block}\n\n{END}"
    if START in text and END in text:
        new = re.sub(re.escape(START) + r".*?" + re.escape(END), lambda _: marked, text, count=1, flags=re.DOTALL)
    else:
        match = fallback.search(text)
        if not match:
            raise RuntimeError(f"{page.relative_to(REPO)}: no markers and no example block to replace")
        new = text[: match.start()] + marked + text[match.end() :]
    page.write_text(new)