214 lines
7.7 KiB
Python
214 lines
7.7 KiB
Python
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = ["click", "requests"]
|
|
# ///
|
|
"""@@NAME@@ — @@DESCRIPTION@@
|
|
|
|
The whole skill lives in this one file. `uv` reads the PEP-723 header above and
|
|
builds/caches a venv on first run, so `click`/`requests` need no install step and
|
|
nothing is added to the system Python.
|
|
|
|
Usage:
|
|
|
|
@@NAME@@ run <target> [--step wait:<sel>] [--step click:<sel>] [--limit N]
|
|
@@NAME@@ send <title> [--body <text>] [--retries 3]
|
|
|
|
Global flags come before the subcommand:
|
|
|
|
@@NAME@@ --json run <target> machine-readable output on stdout
|
|
@@NAME@@ -v run <target> progress chatter on stderr
|
|
|
|
Conventions this skeleton follows (keep them — the rest of the homelab's Python
|
|
skills do):
|
|
|
|
* stdout is the RESULT, stderr is the CHATTER. `--json` makes stdout a single
|
|
JSON object so a calling agent can parse it without scraping prose.
|
|
* exit 0 = success, 1 = handled failure (a message on stderr, never a
|
|
traceback), 2 = usage error (click's own).
|
|
* private defaults (tokens, hosts) come from the gitignored `.env.claude`,
|
|
never from hardcoded literals — see load_env_claude() below.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import click
|
|
import requests
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
# repo root = <root>/.claude/skills/@@NAME@@/scripts -> up 4
|
|
REPO_ROOT = SCRIPT_DIR.parents[3]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# shared helpers — every Python skill needs these three; copy them as-is
|
|
# --------------------------------------------------------------------------- #
|
|
def load_env_claude() -> None:
|
|
"""Load private defaults from the gitignored .env.claude (walk up to repo root).
|
|
|
|
Mirrors the bash `set -a; . .env.claude`: simple KEY=value lines, values
|
|
exported into the environment (overriding), comments/blank lines skipped.
|
|
"""
|
|
d = SCRIPT_DIR
|
|
while d != d.parent:
|
|
f = d / ".env.claude"
|
|
if f.is_file():
|
|
for line in f.read_text().splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if line.startswith("export "):
|
|
line = line[len("export "):]
|
|
if "=" not in line:
|
|
continue
|
|
key, _, val = line.partition("=")
|
|
key, val = key.strip(), val.strip()
|
|
if len(val) >= 2 and val[0] == val[-1] and val[0] in "\"'":
|
|
val = val[1:-1]
|
|
os.environ[key] = val
|
|
return
|
|
d = d.parent
|
|
|
|
|
|
load_env_claude()
|
|
|
|
|
|
def err(msg: str) -> None:
|
|
"""Chatter — always stderr, so it never pollutes a piped result."""
|
|
click.echo(msg, err=True)
|
|
|
|
|
|
def die(msg: str, code: int = 1) -> None:
|
|
"""Handled failure: a one-line reason, no traceback."""
|
|
err(f"error: {msg}")
|
|
sys.exit(code)
|
|
|
|
|
|
def emit(ctx: click.Context, payload: dict, human: str) -> None:
|
|
"""Print the command's result: JSON when --json, otherwise the human line."""
|
|
if ctx.obj["json"]:
|
|
click.echo(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
else:
|
|
click.echo(human)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# param types
|
|
# --------------------------------------------------------------------------- #
|
|
STEP_KINDS = ("wait", "click", "type", "shot")
|
|
|
|
|
|
class Step(click.ParamType):
|
|
"""A `kind:value` step, e.g. `wait:#app` or `click:button.submit`.
|
|
|
|
ONE repeatable option carrying a typed value is the idiom for an *ordered*
|
|
list of heterogeneous actions. Do not split it into `--wait/--click/--shot`:
|
|
click collects each option's occurrences into its own tuple, so the relative
|
|
order between different options is lost — exactly the thing a step list needs.
|
|
"""
|
|
|
|
name = "step"
|
|
|
|
def convert(self, value, param, ctx):
|
|
kind, sep, arg = value.partition(":")
|
|
if not sep or kind not in STEP_KINDS:
|
|
self.fail(
|
|
f"{value!r} is not a step; expected <kind>:<value> "
|
|
f"with kind in {'|'.join(STEP_KINDS)}",
|
|
param,
|
|
ctx,
|
|
)
|
|
return (kind, arg)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# HTTP — retries + a fallback endpoint, the shape most skills end up needing
|
|
# --------------------------------------------------------------------------- #
|
|
def post_json(urls: list[str], payload: dict, retries: int = 3, timeout: float = 10.0) -> dict:
|
|
"""POST `payload` to the first URL that answers, retrying each in turn.
|
|
|
|
Building the body as a dict and letting requests serialise it is the whole
|
|
point of leaving bash behind: no shell quoting, no jq, and no E2BIG when the
|
|
payload carries a base64 blob.
|
|
"""
|
|
last = "no endpoint configured"
|
|
for url in urls:
|
|
for attempt in range(1, retries + 1):
|
|
try:
|
|
r = requests.post(url, json=payload, timeout=timeout)
|
|
r.raise_for_status()
|
|
return r.json() if r.content else {}
|
|
except requests.RequestException as exc:
|
|
last = f"{url}: {exc}"
|
|
if attempt < retries:
|
|
err(f" retry {attempt}/{retries - 1} — {exc}")
|
|
die(f"all endpoints failed ({last})")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# CLI
|
|
# --------------------------------------------------------------------------- #
|
|
@click.group(context_settings={"help_option_names": ["-h", "--help"]})
|
|
@click.option("--json", "as_json", is_flag=True, help="Emit the result as JSON on stdout.")
|
|
@click.option("-v", "--verbose", is_flag=True, help="Log progress to stderr.")
|
|
@click.version_option("0.1.0", prog_name="@@NAME@@")
|
|
@click.pass_context
|
|
def cli(ctx: click.Context, as_json: bool, verbose: bool) -> None:
|
|
"""@@DESCRIPTION@@"""
|
|
ctx.obj = {"json": as_json, "verbose": verbose}
|
|
|
|
|
|
@cli.command()
|
|
@click.argument("target")
|
|
@click.option(
|
|
"--step",
|
|
"steps",
|
|
type=Step(),
|
|
multiple=True,
|
|
metavar="KIND:VALUE",
|
|
help=f"Repeatable, order-preserving step ({'|'.join(STEP_KINDS)}). May be given many times.",
|
|
)
|
|
@click.option("--limit", type=click.IntRange(1, 100), default=10, show_default=True,
|
|
help="Maximum number of results.")
|
|
@click.pass_context
|
|
def run(ctx: click.Context, target: str, steps: tuple, limit: int) -> None:
|
|
"""FILL IN — do the skill's main job against TARGET."""
|
|
if ctx.obj["verbose"]:
|
|
err(f"target={target} limit={limit} steps={len(steps)}")
|
|
|
|
results = []
|
|
for kind, arg in steps:
|
|
if ctx.obj["verbose"]:
|
|
err(f" {kind}: {arg}")
|
|
results.append({"kind": kind, "value": arg}) # FILL IN — actually do the step
|
|
|
|
emit(
|
|
ctx,
|
|
{"target": target, "limit": limit, "steps": results},
|
|
f"{target}: ran {len(results)} step(s)",
|
|
)
|
|
|
|
|
|
@cli.command()
|
|
@click.argument("title")
|
|
@click.option("--body", default="", help="Message body.")
|
|
@click.option("--retries", type=click.IntRange(1, 10), default=3, show_default=True,
|
|
help="Attempts per endpoint.")
|
|
@click.pass_context
|
|
def send(ctx: click.Context, title: str, body: str, retries: int) -> None:
|
|
"""FILL IN — example of posting a JSON payload with retries + fallback."""
|
|
# Endpoints from the environment (.env.claude / .env), never hardcoded.
|
|
urls = [u for u in (os.environ.get("@@ENVPREFIX@@_URL"), "http://127.0.0.1:8096/api/health") if u]
|
|
if not urls:
|
|
die("set @@ENVPREFIX@@_URL in .env.claude")
|
|
|
|
reply = post_json(urls, {"title": title, "body": body}, retries=retries)
|
|
emit(ctx, {"sent": True, "title": title, "reply": reply}, f"sent: {title}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|