canned-prompts/reference/canned_prompts.py

#!/usr/bin/env python3
"""Tiny reference CLI for Canned Prompt Format v0.1.

This implementation intentionally favors readability over features. It uses a
filesystem-backed local catalog and registry and never calls a model.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Any, Iterable

import yaml

FORMAT = "canned-prompt/v0.1"
PLACEHOLDER_RE = re.compile(r"{{\s*([A-Za-z_][A-Za-z0-9_.-]*)\s*}}")
SEMVER_RE = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:[-+].*)?$")
REQUIRED_FIELDS = ("format", "id", "name", "version", "summary", "template")


class CannedPromptError(Exception):
    pass


def home_dir() -> Path:
    return Path(os.environ.get("CANNED_PROMPTS_HOME", Path.home() / ".canned-prompts"))


def default_catalog() -> Path:
    return home_dir() / "catalog"


def default_registry() -> Path:
    return home_dir() / "registry"


def read_manifest(package_dir: Path) -> dict[str, Any]:
    manifest_path = package_dir / "prompt.yaml"
    if not manifest_path.is_file():
        raise CannedPromptError(f"missing manifest: {manifest_path}")
    try:
        data = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
    except yaml.YAMLError as exc:
        raise CannedPromptError(f"invalid YAML in {manifest_path}: {exc}") from exc
    if not isinstance(data, dict):
        raise CannedPromptError("prompt.yaml must contain a mapping")
    return data


def safe_relative_file(package_dir: Path, relative: str, field: str) -> Path:
    if not isinstance(relative, str) or not relative.strip():
        raise CannedPromptError(f"{field} must be a non-empty relative path")
    candidate = (package_dir / relative).resolve()
    root = package_dir.resolve()
    try:
        candidate.relative_to(root)
    except ValueError as exc:
        raise CannedPromptError(f"{field} escapes the package: {relative}") from exc
    if not candidate.is_file():
        raise CannedPromptError(f"{field} does not reference a file: {relative}")
    return candidate


def declared_names(manifest: dict[str, Any]) -> set[str]:
    names: set[str] = set()
    inputs = manifest.get("inputs") or []
    if not isinstance(inputs, list):
        raise CannedPromptError("inputs must be a list")
    for item in inputs:
        if not isinstance(item, dict) or not isinstance(item.get("name"), str):
            raise CannedPromptError("each input must be a mapping with a string name")
        name = item["name"]
        if name in names:
            raise CannedPromptError(f"duplicate input/parameter name: {name}")
        names.add(name)

    parameters = manifest.get("parameters") or {}
    if not isinstance(parameters, dict):
        raise CannedPromptError("parameters must be a mapping")
    for name, spec in parameters.items():
        if not isinstance(name, str) or not isinstance(spec, dict):
            raise CannedPromptError("parameters must map names to mappings")
        if name in names:
            raise CannedPromptError(f"duplicate input/parameter name: {name}")
        names.add(name)
    return names


def validate_package(package_dir: Path) -> dict[str, Any]:
    package_dir = package_dir.resolve()
    if not package_dir.is_dir():
        raise CannedPromptError(f"package directory not found: {package_dir}")

    manifest = read_manifest(package_dir)
    missing = [field for field in REQUIRED_FIELDS if field not in manifest]
    if missing:
        raise CannedPromptError("missing required fields: " + ", ".join(missing))

    if manifest["format"] != FORMAT:
        raise CannedPromptError(f"unsupported format: {manifest['format']!r}")

    package_id = manifest["id"]
    if not isinstance(package_id, str) or not package_id.strip():
        raise CannedPromptError("id must be a non-empty string")
    if any(part in ("", ".", "..") for part in package_id.split("/")):
        raise CannedPromptError("id contains an invalid path segment")
    if re.search(r"[^A-Za-z0-9._/-]", package_id):
        raise CannedPromptError("id contains unsupported characters")

    version = manifest["version"]
    if not isinstance(version, str) or not SEMVER_RE.match(version):
        raise CannedPromptError("version must be semantic-version-like, e.g. 1.2.0")

    template_path = safe_relative_file(package_dir, manifest["template"], "template")

    for field in ("examples", "evals"):
        refs = manifest.get(field) or []
        if not isinstance(refs, list):
            raise CannedPromptError(f"{field} must be a list")
        for relative in refs:
            safe_relative_file(package_dir, relative, field)

    names = declared_names(manifest)
    template = template_path.read_text(encoding="utf-8")
    placeholders = set(PLACEHOLDER_RE.findall(template))
    undeclared = sorted(placeholders - names)
    if undeclared:
        raise CannedPromptError(
            "template contains undeclared placeholders: " + ", ".join(undeclared)
        )

    return manifest


def id_path(store: Path, package_id: str) -> Path:
    parts = package_id.split("/")
    if any(part in ("", ".", "..") for part in parts):
        raise CannedPromptError("unsafe package id")
    return store.joinpath(*parts)


def package_path(store: Path, package_id: str, version: str) -> Path:
    return id_path(store, package_id) / version


def parse_semver(value: str) -> tuple[int, int, int, str]:
    match = SEMVER_RE.match(value)
    if not match:
        return (-1, -1, -1, value)
    return (int(match.group(1)), int(match.group(2)), int(match.group(3)), value)


def versions_for(store: Path, package_id: str) -> list[str]:
    base = id_path(store, package_id)
    if not base.is_dir():
        return []
    versions = [p.name for p in base.iterdir() if p.is_dir() and (p / "prompt.yaml").is_file()]
    return sorted(versions, key=parse_semver, reverse=True)


def resolve_installed(store: Path, package_id: str, version: str | None) -> Path:
    if version:
        path = package_path(store, package_id, version)
        if not path.is_dir():
            raise CannedPromptError(f"package not found: {package_id}@{version}")
        return path
    versions = versions_for(store, package_id)
    if not versions:
        raise CannedPromptError(f"package not found: {package_id}")
    return package_path(store, package_id, versions[0])


def copy_immutable(src: Path, dst: Path, what: str) -> None:
    if dst.exists():
        raise CannedPromptError(f"{what} already exists: {dst}")
    dst.parent.mkdir(parents=True, exist_ok=True)
    shutil.copytree(src, dst)


def iter_packages(store: Path) -> Iterable[tuple[Path, dict[str, Any]]]:
    if not store.exists():
        return
    for manifest_path in store.rglob("prompt.yaml"):
        package_dir = manifest_path.parent
        try:
            manifest = validate_package(package_dir)
        except CannedPromptError:
            continue
        yield package_dir, manifest


def cmd_add(args: argparse.Namespace) -> None:
    src = Path(args.path)
    manifest = validate_package(src)
    catalog = Path(args.catalog).expanduser()
    dst = package_path(catalog, manifest["id"], manifest["version"])
    copy_immutable(src.resolve(), dst, "catalog package")
    print(f"added {manifest['id']}@{manifest['version']} -> {dst}")


def cmd_publish(args: argparse.Namespace) -> None:
    src = Path(args.path)
    manifest = validate_package(src)
    registry = Path(args.registry).expanduser()
    dst = package_path(registry, manifest["id"], manifest["version"])
    copy_immutable(src.resolve(), dst, "published package")
    print(f"published {manifest['id']}@{manifest['version']} -> {dst}")


def cmd_install(args: argparse.Namespace) -> None:
    registry = Path(args.registry).expanduser()
    catalog = Path(args.catalog).expanduser()
    src = resolve_installed(registry, args.id, args.version)
    manifest = validate_package(src)
    dst = package_path(catalog, manifest["id"], manifest["version"])
    copy_immutable(src, dst, "catalog package")
    print(f"installed {manifest['id']}@{manifest['version']} -> {dst}")


def cmd_search(args: argparse.Namespace) -> None:
    catalog = Path(args.catalog).expanduser()
    query = args.query.lower()
    matches: list[dict[str, str]] = []
    for _, manifest in iter_packages(catalog):
        haystack = " ".join(
            [
                str(manifest.get("id", "")),
                str(manifest.get("name", "")),
                str(manifest.get("summary", "")),
                " ".join(str(tag) for tag in (manifest.get("tags") or [])),
            ]
        ).lower()
        if query in haystack:
            matches.append(
                {
                    "id": manifest["id"],
                    "version": manifest["version"],
                    "name": manifest["name"],
                    "summary": manifest["summary"],
                }
            )
    matches.sort(key=lambda m: (m["id"], parse_semver(m["version"])), reverse=False)
    if args.json:
        print(json.dumps(matches, indent=2, ensure_ascii=False))
        return
    if not matches:
        print("no matches")
        return
    for item in matches:
        print(f"{item['id']}@{item['version']}  {item['name']}")
        print(f"  {item['summary']}")


def cmd_show(args: argparse.Namespace) -> None:
    catalog = Path(args.catalog).expanduser()
    package_dir = resolve_installed(catalog, args.id, args.version)
    manifest = validate_package(package_dir)
    if args.json:
        print(json.dumps(manifest, indent=2, ensure_ascii=False))
    else:
        print(yaml.safe_dump(manifest, sort_keys=False, allow_unicode=True).rstrip())


def coerce_value(raw: str, spec: dict[str, Any]) -> Any:
    kind = spec.get("type", "string")
    if kind == "boolean":
        lowered = raw.lower()
        if lowered in {"true", "1", "yes", "on"}:
            return True
        if lowered in {"false", "0", "no", "off"}:
            return False
        raise CannedPromptError(f"cannot parse boolean value: {raw}")
    if kind == "integer":
        try:
            return int(raw)
        except ValueError as exc:
            raise CannedPromptError(f"cannot parse integer value: {raw}") from exc
    if kind == "number":
        try:
            return float(raw)
        except ValueError as exc:
            raise CannedPromptError(f"cannot parse numeric value: {raw}") from exc
    if kind == "enum":
        values = spec.get("values") or []
        if raw not in values:
            raise CannedPromptError(f"invalid enum value {raw!r}; expected one of {values}")
    return raw


def supplied_values(pairs: list[str]) -> dict[str, str]:
    values: dict[str, str] = {}
    for pair in pairs:
        if "=" not in pair:
            raise CannedPromptError(f"--set expects name=value, got: {pair}")
        name, value = pair.split("=", 1)
        if not name:
            raise CannedPromptError("--set name cannot be empty")
        values[name] = value
    return values


def resolve_values(manifest: dict[str, Any], raw_values: dict[str, str]) -> dict[str, Any]:
    resolved: dict[str, Any] = {}
    known: set[str] = set()

    inputs = manifest.get("inputs") or []
    for item in inputs:
        name = item["name"]
        known.add(name)
        if name in raw_values:
            resolved[name] = raw_values[name]
        elif item.get("required", False):
            raise CannedPromptError(f"missing required input: {name}")

    parameters = manifest.get("parameters") or {}
    for name, spec in parameters.items():
        known.add(name)
        if name in raw_values:
            resolved[name] = coerce_value(raw_values[name], spec)
        elif "default" in spec:
            resolved[name] = spec["default"]

    unknown = sorted(set(raw_values) - known)
    if unknown:
        raise CannedPromptError("unknown values: " + ", ".join(unknown))

    return resolved


def render_template(template: str, values: dict[str, Any]) -> str:
    def replace(match: re.Match[str]) -> str:
        name = match.group(1)
        if name not in values:
            raise CannedPromptError(f"unresolved placeholder: {name}")
        value = values[name]
        if isinstance(value, bool):
            return "true" if value else "false"
        if isinstance(value, (dict, list)):
            return json.dumps(value, ensure_ascii=False)
        return str(value)

    return PLACEHOLDER_RE.sub(replace, template)


def cmd_render(args: argparse.Namespace) -> None:
    catalog = Path(args.catalog).expanduser()
    package_dir = resolve_installed(catalog, args.id, args.version)
    manifest = validate_package(package_dir)
    template_path = safe_relative_file(package_dir, manifest["template"], "template")
    raw = supplied_values(args.set_values)
    values = resolve_values(manifest, raw)
    rendered = render_template(template_path.read_text(encoding="utf-8"), values)
    print(rendered, end="" if rendered.endswith("\n") else "\n")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="canned-prompts", description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    add = sub.add_parser("add", help="add a package directory to the local catalog")
    add.add_argument("path")
    add.add_argument("--catalog", default=str(default_catalog()))
    add.set_defaults(func=cmd_add)

    search = sub.add_parser("search", help="search the local catalog")
    search.add_argument("query")
    search.add_argument("--catalog", default=str(default_catalog()))
    search.add_argument("--json", action="store_true")
    search.set_defaults(func=cmd_search)

    show = sub.add_parser("show", help="show an installed package manifest")
    show.add_argument("id")
    show.add_argument("--version")
    show.add_argument("--catalog", default=str(default_catalog()))
    show.add_argument("--json", action="store_true")
    show.set_defaults(func=cmd_show)

    render = sub.add_parser("render", help="render an installed prompt template")
    render.add_argument("id")
    render.add_argument("--version")
    render.add_argument("--catalog", default=str(default_catalog()))
    render.add_argument("--set", dest="set_values", action="append", default=[], metavar="NAME=VALUE")
    render.set_defaults(func=cmd_render)

    install = sub.add_parser("install", help="install a package from a filesystem registry")
    install.add_argument("id")
    install.add_argument("--version")
    install.add_argument("--catalog", default=str(default_catalog()))
    install.add_argument("--registry", default=str(default_registry()))
    install.set_defaults(func=cmd_install)

    publish = sub.add_parser("publish", help="publish a package to a filesystem registry")
    publish.add_argument("path")
    publish.add_argument("--registry", default=str(default_registry()))
    publish.set_defaults(func=cmd_publish)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        args.func(args)
        return 0
    except CannedPromptError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())