SJNmusic/cli/sjnmusic

#!/usr/bin/env python3
"""
sjnmusic - tiny CLI for sjnmusicd HTTP daemon.
Talks to the webserver at http://127.0.0.1:14567 by default. Override with
--host or --port, or set SJNMUSIC_HOST / SJNMUSIC_PORT environment variables.
"""

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 14567


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="sjnmusic",
        description="Talk to the sjnmusicd HTTP daemon.",
    )
    p.add_argument("--host", default=os.environ.get("SJNMUSIC_HOST", DEFAULT_HOST))
    p.add_argument(
        "--port",
        type=int,
        default=int(os.environ.get("SJNMUSIC_PORT", str(DEFAULT_PORT))),
    )
    p.add_argument(
        "--token",
        default=os.environ.get("SJNMUSIC_TOKEN"),
        help=(
            "bearer token for the daemon (Authorization: Bearer <token>). "
            "Required when talking to a non-loopback daemon. Can also be set "
            "via the SJNMUSIC_TOKEN environment variable. On loopback the "
            "CLI auto-discovers the token from GET /auth/token."
        ),
    )
    p.add_argument("--json", action="store_true", help="print raw JSON responses")

    sub = p.add_subparsers(dest="cmd", required=True)

    def add_search_fields(sp):
        sp.add_argument("query", nargs="*", help="search text")
        sp.add_argument("-q", "--query", dest="query_flag", help="search text")
        return sp

    add_search_fields(sub.add_parser("play", help="search & start playing immediately"))
    add_search_fields(sub.add_parser("add", help="search & append to queue"))
    add_search_fields(sub.add_parser("del", help="search & delete from disk+db"))
    init_p = add_search_fields(
        sub.add_parser(
            "init",
            help="download via yt-dlp and register (async); picks a yt-dlp candidate first if interactive",
        )
    )
    init_p.add_argument(
        "--pick",
        action="store_true",
        help="force interactive candidate picker even when stdin is not a TTY",
    )
    init_p.add_argument(
        "--auto",
        action="store_true",
        help="skip the candidate picker and download yt-dlp's top result",
    )
    init_p.add_argument(
        "--limit",
        type=int,
        default=None,
        help="number of yt-dlp candidates to fetch from /search/yt (default: daemon config)",
    )
    add_search_fields(sub.add_parser("search", help="search the song database (best match)"))
    add_search_fields(sub.add_parser("search-all", help="search every match above threshold"))

    sub.add_parser("queue", help="show the current queue")
    sub.add_parser("now", help="alias for /now-playing")
    sub.add_parser("status", help="one-line summary of current playback")
    sub.add_parser("skip", help="skip the next song")
    sub.add_parser("pause", help="pause playback")
    sub.add_parser("resume", help="resume playback")
    sub.add_parser("shuffle", help="shuffle the queue in place")
    sub.add_parser("clear", help="empty the queue")
    sub.add_parser("songs", help="list all known songs")
    sub.add_parser("playlists", help="list all playlists")
    sub.add_parser("downloads", help="list recent yt-dlp download jobs")
    sub.add_parser("history", help="show recently played songs")
    sub.add_parser("stats", help="aggregate stats from history")
    sub.add_parser("help", help="list HTTP endpoints")

    vol = sub.add_parser("volume", help="get/set volume (0-100 or floating 0.0-2.0)")
    vol.add_argument("value", nargs="?", type=float, help="0-100 (interpreted as percent)")

    seek = sub.add_parser("seek", help="jump to position within the current track")
    seek.add_argument("secs", type=float, help="position in seconds")

    repeat = sub.add_parser("repeat", help="get/set repeat mode (off/one/all)")
    repeat.add_argument("mode", nargs="?", help="off, one, all")

    pl_new = sub.add_parser("pl-new", help="create a playlist")
    pl_new.add_argument("name")

    pl_show = sub.add_parser("pl-show", help="show playlist contents")
    pl_show.add_argument("name")

    pl_add = sub.add_parser("pl-add", help="add a song to a playlist")
    pl_add.add_argument("name")
    pl_add.add_argument("query", nargs="*")
    pl_add.add_argument("-q", "--query", dest="query_flag")

    pl_play = sub.add_parser("pl-play", help="load a playlist into the queue")
    pl_play.add_argument("name")

    pl_rm = sub.add_parser("pl-rm", help="delete a playlist")
    pl_rm.add_argument("name")

    pl_rm_song = sub.add_parser("pl-rm-song", help="remove a song from a playlist")
    pl_rm_song.add_argument("name")
    pl_rm_song.add_argument("song_id", type=int)

    pl_rename = sub.add_parser("pl-rename", help="rename a playlist")
    pl_rename.add_argument("old")
    pl_rename.add_argument("new")

    pl_dup = sub.add_parser("pl-dup", help="duplicate a playlist under a new name")
    pl_dup.add_argument("src")
    pl_dup.add_argument("dest")

    pl_reorder = sub.add_parser("pl-reorder", help="reorder songs in a playlist")
    pl_reorder.add_argument("name")
    pl_reorder.add_argument("from_pos", type=int)
    pl_reorder.add_argument("to_pos", type=int)

    init_batch = sub.add_parser("init-batch", help="batch-download several songs at once")
    init_batch.add_argument("queries", nargs="+")

    return p


def join_query(args) -> str:
    parts = []
    if getattr(args, "query_flag", None):
        parts.append(args.query_flag)
    if getattr(args, "query", None):
        parts.extend(args.query)
    return " ".join(parts).strip()


def call(host, port, method, route, body=None, query=None, timeout=15, headers=None):
    qs = ""
    if query:
        qs = "?" + urllib.parse.urlencode(query)
    url = f"http://{host}:{port}{route}{qs}"

    data = None
    hdrs = dict(headers or {})
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        hdrs.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(url=url, data=data, headers=hdrs, method=method)

    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            payload = resp.read().decode("utf-8")
            return resp.status, payload
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", errors="replace")
    except urllib.error.URLError as e:
        print(f"error: cannot reach daemon at {url}: {e.reason}", file=sys.stderr)
        sys.exit(2)


# Loopback addresses that skip the bearer-token check on the daemon
# side. Kept in sync with the daemon's `auth::is_local_request` so the
# CLI auto-fetch fallback only ever fires when the daemon would
# actually let us through unauthenticated.
LOCAL_HOSTS = frozenset({"127.0.0.1", "::1", "localhost"})


def is_local(host):
    return host in LOCAL_HOSTS


def fetch_token(host, port, timeout=5):
    """
    Best-effort localhost-only token probe.

    GET /auth/token is itself loopback-only on the daemon side, so
    reaching it implies we're a process running on the same host. We
    do NOT retry on failure — a 403 here means a remote daemon is
    pretending to be local and refusing (we should propagate the
    error), a 404 means the daemon has auth disabled (don't auto-fetch
    any token), and any other non-200 is opaque. We surface the
    decoded JSON `token` field on success and `None` on every other
    outcome.
    """
    try:
        status, payload = call(
            host, port, "GET", "/auth/token", timeout=timeout
        )
    except SystemExit:
        # `call` exits the process on URLError; treat that as
        # "couldn't reach daemon, no token". We re-raise because the
        # caller is already going to error out on the original
        # request anyway, but only after a transparent re-raise so
        # surfacing happens at the right stack frame.
        return None
    if status != 200:
        return None
    obj = try_parse(payload)
    if not isinstance(obj, dict):
        return None
    tok = obj.get("token")
    if isinstance(tok, str) and tok:
        return tok
    return None


def request(args, host, port, method, route, body=None, query=None, timeout=15):
    """
    Auth-aware wrapper around `call`. Order of operations:

    1. If the operator supplied `--token` (or `SJNMUSIC_TOKEN`), attach
       `Authorization: Bearer <token>` and send straight to the
       daemon. We never auto-fetch on top of an explicit token — that
       would silently swap a user-supplied secret with the daemon's
       current one (which is rarely what an operator wants).
    2. Otherwise: if the peer is loopback, send without a token and
       then, if the daemon rejected with 401, do a one-shot
       `GET /auth/token` and retry. This keeps today's CLI workflows
       unchanged: a user with auto-fetch works even though the
       daemon now requires tokens for non-loopback callers. We
       decline to silently fetch on a non-loopback host because that
       would defeat the whole point of forcing operators to give an
       explicit token to remote daemons.
    3. If the peer is non-loopback and no token was supplied, the
       call proceeds without a header so the operator gets a
       daemon-issued 401 instead of a CLI-level error — easier to
       debug in the daemon logs.
    """
    explicit_token = getattr(args, "token", None) or ""
    hdrs = {}
    if explicit_token:
        hdrs["Authorization"] = f"Bearer {explicit_token}"

    status, payload = call(
        host, port, method, route, body=body, query=query,
        timeout=timeout, headers=hdrs,
    )

    # Auto-fetch only on FIRST observation of 401 on a loopback peer
    # when no explicit token was supplied AND this is not already the
    # token endpoint itself (which would obviously never require a
    # token). The 401 must have come from a request method that the
    # server would normally respond 2xx to without auth; if /auth/token
    # is what we just hit, we'd loop forever.
    if (
        status == 401
        and not explicit_token
        and is_local(host)
        and route != "/auth/token"
    ):
        fetched = fetch_token(host, port)
        if fetched:
            hdrs["Authorization"] = f"Bearer {fetched}"
            status, payload = call(
                host, port, method, route, body=body, query=query,
                timeout=timeout, headers=hdrs,
            )

    return status, payload


def fmt_secs(s):
    if s is None:
        return "--:--"
    try:
        s = float(s)
    except (TypeError, ValueError):
        return "--:--"
    if s < 0:
        return "--:--"
    m, sec = divmod(int(s), 60)
    return f"{m:d}:{sec:02d}"


def try_parse(payload):
    try:
        return json.loads(payload)
    except json.JSONDecodeError:
        return None


def print_candidate(idx, c):
    """Pretty-print one ranked candidate row for the picker."""
    score = c.get("score")
    score_str = f"score={score:+d}" if score is not None else ""
    flags = " ".join(c.get("flags") or [])
    flag_str = f" [{flags}]" if flags else ""
    print(
        f"[{idx}] {c.get('title', '?')}  --  "
        f"{c.get('uploader') or '(unknown)':24}  "
        f"({fmt_secs(c.get('duration_secs')):>5})  "
        f"{score_str}{flag_str}"
    )


def interactive_pick(candidates):
    """Show the picker; return the user's choice (or candidates[0])."""
    for i, c in enumerate(candidates, 1):
        print_candidate(i, c)
    try:
        raw = input("Pick a song [1]: ").strip()
    except EOFError:
        print(
            "error: no input available; re-run with --auto or supply a piped number",
            file=sys.stderr,
        )
        sys.exit(2)
    if raw == "":
        return candidates[0]
    try:
        idx = int(raw) - 1
    except ValueError:
        print(f"error: not a number: {raw!r}", file=sys.stderr)
        sys.exit(2)
    if idx < 0 or idx >= len(candidates):
        print(
            f"error: choice out of range (1-{len(candidates)})",
            file=sys.stderr,
        )
        sys.exit(2)
    return candidates[idx]


def legacy_pick(results, force_auto):
    """Legacy /search/yt picker (no scoring fields)."""
    if force_auto:
        return results[0]
    return interactive_pick(results)


def do_init(args, host, port, q, force_pick, force_auto, limit):
    """
    Drive the `init` subcommand through the new ranked pipeline.

    Default behaviour: `GET /pick?q=…` and let the daemon either
    auto-select (when the top score beats the runner-up by margin) or
    surface ranked candidates for the picker. Falls back to the legacy
    `/search/yt` picker if the daemon doesn't speak `/pick` (older
    builds), so the CLI stays forward/backward compatible.

    Returns the chosen candidate dict (with at least a 'url' key) so
    `main()` can dispatch the actual `/init` POST uniformly.
    """
    pick_query = {"q": q}
    if limit is not None and limit > 0:
        pick_query["limit"] = str(limit)
    # 3 parallel yt-dlp fetches can easily take 30+ seconds each
    # on a slow network; the daemon drops its state mutex before
    # invoking them, but the network round-trips themselves are
    # unchanged. Generous timeout to avoid ghost-failing.
    pick_status, pick_payload = request(
        args, host, port, "GET", "/pick", query=pick_query, timeout=180
    )
    pick_obj = try_parse(pick_payload)
    if pick_status != 200 or pick_obj is None or "kind" not in pick_obj:
        # Fall back to legacy `/search/yt` if the daemon is too old
        # to know about `/pick`. Keeps the CLI working across daemon
        # upgrades in either direction.
        legacy_status, legacy_payload = request(
            args, host, port, "GET", "/search/yt",
            query=pick_query, timeout=120,
        )
        legacy_obj = try_parse(legacy_payload) or {}
        legacy_results = legacy_obj.get("results") or []
        if legacy_status != 200:
            return (legacy_status, legacy_payload)
        if not legacy_results:
            print(
                f"error: no yt-dlp results for {q!r}",
                file=sys.stderr,
            )
            sys.exit(1)
        return (
            200,
            json.dumps({"chosen": legacy_pick(legacy_results, force_auto)}),
        )

    kind = pick_obj["kind"]
    if kind == "empty":
        print(
            f"error: no yt-dlp results for {q!r}: "
            f"{pick_obj.get('message', '')}",
            file=sys.stderr,
        )
        sys.exit(1)
    if kind == "auto":
        # Daemon is confident — surface the choice so the user sees
        # what was picked, then download.
        score = pick_obj.get("score", 0)
        print(
            f"[auto-pick score={score}] "
            f"{pick_obj.get('title', '?')} -- "
            f"{pick_obj.get('uploader', '(unknown)')}  "
            f"({fmt_secs(pick_obj.get('duration_secs'))})"
        )
        if force_pick:
            # `--pick` overrides the auto decision: ask the user.
            ranked_status, ranked_payload = request(
                args, host, port, "GET", "/search/yt/ranked",
                query=pick_query, timeout=180,
            )
            ranked_obj = try_parse(ranked_payload) or {}
            ranked_results = ranked_obj.get("results") or []
            if ranked_status != 200 or not ranked_results:
                print_response_args(args, ranked_status, ranked_payload)
                sys.exit(1)
            return (
                200,
                json.dumps({"chosen": interactive_pick(ranked_results)}),
            )
        return (
            200,
            json.dumps({
                "chosen": {
                    "url": pick_obj.get("url"),
                    "title": pick_obj.get("title"),
                    "uploader": pick_obj.get("uploader"),
                    "score": score,
                    "duration_secs": pick_obj.get("duration_secs"),
                    "flags": pick_obj.get("flags", []),
                },
            }),
        )
    if kind == "needs_choice":
        candidates = pick_obj.get("candidates") or []
        if not candidates:
            print(
                f"error: no yt-dlp results for {q!r}",
                file=sys.stderr,
            )
            sys.exit(1)
        if force_auto:
            chosen = candidates[0]
            print(
                f"[auto-pick score={chosen.get('score', 0)}] "
                f"{chosen.get('title', '?')} -- "
                f"{chosen.get('uploader', '(unknown)')}  "
                f"({fmt_secs(chosen.get('duration_secs'))})"
            )
        else:
            chosen = interactive_pick(candidates)
        return (200, json.dumps({"chosen": chosen}))
    print(f"error: unknown /pick kind {kind!r}", file=sys.stderr)
    sys.exit(1)


def print_response_args(args, status, payload):
    """Helper for subroutes that already pre-formatted a payload."""
    print_response(args, status, payload)


def print_response(args, status, payload):
    if args.json:
        print(f"HTTP {status}\n{payload}")
        return
    obj = try_parse(payload)
    if obj is None:
        print(f"HTTP {status}: {payload}")
        return
    if isinstance(obj, dict) and "error" in obj:
        print(f"error: {obj['error']}", file=sys.stderr)
        sys.exit(1)
    # Specialised formatters used by status / history / stats / downloads.
    if (
        isinstance(obj, dict)
        and "current" in obj
        and "queue_len" in obj
        and "elapsed_secs" in obj
    ):
        cur = obj.get("current") or {}
        name = cur.get("name") or "(idle)"
        elapsed = fmt_secs(obj.get("elapsed_secs"))
        total = fmt_secs(obj.get("duration_secs"))
        print(
            f"[{cur.get('id', '-')}] {name}  {elapsed}/{total}  "
            f"q={obj.get('queue_len', 0)}  "
            f"vol={obj.get('volume', 0):.2f}  "
            f"paused={obj.get('paused', False)}  "
            f"repeat={obj.get('repeat', 'off')}"
        )
        return
    if isinstance(obj, dict) and "history" in obj:
        for h in obj["history"]:
            ts = h.get("played_at", 0)
            when = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
            print(
                f"{when}  [{h.get('song_id', '-')}] {h.get('song_name') or '(missing)'}  "
                f"{fmt_secs(h.get('duration_secs_played', 0))}"
            )
        return
    if isinstance(obj, dict) and "top_songs" in obj:
        print(f"total plays: {obj.get('total_plays', 0)}")
        print(f"total time:  {fmt_secs(obj.get('total_secs', 0))}")
        for t in obj.get("top_songs", []):
            print(
                f"  - [{t.get('song_id', '-')}] {t.get('name') or '(missing)'}  "
                f"plays={t.get('plays', 0)}  time={fmt_secs(t.get('total_secs', 0))}"
            )
        return
    if isinstance(obj, dict) and "downloads" in obj:
        for d in obj["downloads"]:
            extras = []
            if d.get("song_id") is not None:
                extras.append(f"song={d['song_id']}")
            if d.get("error"):
                extras.append(f"err={d['error']}")
            extra_str = ("  " + "  ".join(extras)) if extras else ""
            print(f"[{d.get('id', '-')}] {d.get('status', '?')}  {d.get('name', '?')}{extra_str}")
        return
    if isinstance(obj, dict) and "matches" in obj:
        for s in obj["matches"]:
            print(f"[{s['id']}] {s['name']}")
        return
    if isinstance(obj, dict) and "queue" in obj and "current" in obj:
        repeat = obj.get("repeat", "off")
        print(f"queue (repeat={repeat}):")
        for s in obj.get("queue", []):
            print(f"  - [{s['id']}] {s['name']}")
        cur = obj.get("current")
        if cur:
            print(f"now playing: [{cur['id']}] {cur['name']}")
        else:
            print("now playing: (idle)")
        return
    if isinstance(obj, dict) and "songs" in obj:
        for s in obj["songs"]:
            print(f"[{s['id']}] {s['name']}")
        return
    if isinstance(obj, list):
        if obj and isinstance(obj[0], dict) and obj[0].get("songs") is not None:
            for pl in obj:
                print(f"{pl['name']}  ({len(pl.get('songs', []))} songs)")
                for s in pl.get("songs", []):
                    print(f"   - [{s['id']}] {s['name']}")
        else:
            for s in obj:
                print(s)
        return
    if isinstance(obj, dict) and "name" in obj and "songs" in obj:
        pl = obj
        print(f"playlist: {pl['name']}")
        for s in pl.get("songs", []):
            print(f"  - [{s['id']}] {s['name']}")
        return
    if isinstance(obj, dict) and "endpoints" in obj:
        for ep in obj["endpoints"]:
            print(ep)
        return
    if isinstance(obj, dict) and "added" in obj:
        print(f"added [{obj['added']['id']}] {obj['added']['name']} -> {obj['playlist']['name']}")
        return
    print(json.dumps(obj, indent=2))


def main(argv=None):
    args = build_parser().parse_args(argv)
    host, port = args.host, args.port
    q = join_query(args)

    def with_q(body):
        body = dict(body or {})
        if q:
            body.setdefault("query", q)
        return body

    def pl_route(name):
        return f"/playlists/{urllib.parse.quote(name)}"

    if args.cmd == "play":
        status, payload = request(args, host, port, "POST", "/play", body=with_q(None))
    elif args.cmd == "add":
        status, payload = request(args, host, port, "POST", "/add", body=with_q(None))
    elif args.cmd == "del":
        if not q:
            print("error: 'del' requires a search query", file=sys.stderr)
            sys.exit(2)
        status, payload = request(args, host, port, "POST", "/del", body=with_q(None))
    elif args.cmd == "init":
        if not q:
            print("error: 'init' requires a song name", file=sys.stderr)
            sys.exit(2)
        force_pick = getattr(args, "pick", False)
        force_auto = getattr(args, "auto", False)
        limit = getattr(args, "limit", None)

        # Drive the ranked pipeline (do_init returns the chosen
        # candidate as a JSON envelope so we don't need separate
        # branches below for `/init`). Routing through `request`
        # means auto-fetch-from-/auth/token also works for the
        # search (/pick) and the eventual /init POST.
        _, envelope_payload = do_init(
            args, host, port, q, force_pick, force_auto, limit,
        )
        chosen_obj = try_parse(envelope_payload) or {}
        chosen = chosen_obj.get("chosen") or {}
        status, payload = request(
            args, host, port, "POST", "/init",
            body={"name": q, "url": chosen.get("url") or ""},
        )
    elif args.cmd == "init-batch":
        status, payload = request(
            args, host, port, "POST", "/init/batch", body={"songs": args.queries}
        )
    elif args.cmd == "search":
        if not q:
            print("error: 'search' requires a query", file=sys.stderr)
            sys.exit(2)
        status, payload = request(args, host, port, "GET", "/search", query={"q": q})
    elif args.cmd == "search-all":
        if not q:
            print("error: 'search-all' requires a query", file=sys.stderr)
            sys.exit(2)
        status, payload = request(args, host, port, "GET", "/search/all", query={"q": q})
    elif args.cmd == "queue":
        status, payload = request(args, host, port, "GET", "/queue")
    elif args.cmd in ("now", "status"):
        status, payload = request(args, host, port, "GET", "/now-playing")
    elif args.cmd == "skip":
        status, payload = request(args, host, port, "POST", "/skip")
    elif args.cmd == "pause":
        status, payload = request(args, host, port, "POST", "/pause")
    elif args.cmd == "resume":
        status, payload = request(args, host, port, "POST", "/resume")
    elif args.cmd == "seek":
        status, payload = request(args, host, port, "POST", "/seek", body={"secs": args.secs})
    elif args.cmd == "shuffle":
        status, payload = request(args, host, port, "POST", "/queue/shuffle")
    elif args.cmd == "clear":
        status, payload = request(args, host, port, "POST", "/queue/clear")
    elif args.cmd == "songs":
        status, payload = request(args, host, port, "GET", "/songs")
    elif args.cmd == "downloads":
        status, payload = request(args, host, port, "GET", "/downloads")
    elif args.cmd == "history":
        status, payload = request(args, host, port, "GET", "/history")
    elif args.cmd == "stats":
        status, payload = request(args, host, port, "GET", "/stats")
    elif args.cmd == "volume":
        if args.value is None:
            status, payload = request(args, host, port, "GET", "/volume")
        else:
            # Treat 0..100 as a percentage of rodio's default 1.0.
            v = args.value
            if v > 2.0:
                v = v / 100.0
            status, payload = request(args, host, port, "POST", "/volume", body={"vol": v})
    elif args.cmd == "repeat":
        if args.mode is None:
            status, payload = request(args, host, port, "GET", "/repeat")
        else:
            status, payload = request(args, host, port, "POST", "/repeat", body={"mode": args.mode})
    elif args.cmd == "playlists":
        status, payload = request(args, host, port, "GET", "/playlists")
    elif args.cmd == "help":
        status, payload = request(args, host, port, "GET", "/help")
    elif args.cmd == "pl-new":
        status, payload = request(args, host, port, "POST", "/playlists", body={"name": args.name})
    elif args.cmd == "pl-show":
        status, payload = request(args, host, port, "GET", pl_route(args.name))
    elif args.cmd == "pl-add":
        if not q:
            print("error: 'pl-add' requires a song query", file=sys.stderr)
            sys.exit(2)
        status, payload = request(
            args, host, port, "POST", f"{pl_route(args.name)}/add", body={"query": q}
        )
    elif args.cmd == "pl-play":
        status, payload = request(args, host, port, "POST", f"{pl_route(args.name)}/play")
    elif args.cmd == "pl-rm":
        status, payload = request(args, host, port, "DELETE", pl_route(args.name))
    elif args.cmd == "pl-rm-song":
        status, payload = request(
            args, host, port, "DELETE", f"{pl_route(args.name)}/songs/{args.song_id}"
        )
    elif args.cmd == "pl-rename":
        status, payload = request(
            args, host, port, "PATCH", pl_route(args.old), body={"name": args.new}
        )
    elif args.cmd == "pl-dup":
        status, payload = request(
            args, host, port, "POST", f"{pl_route(args.src)}/duplicate", body={"name": args.dest}
        )
    elif args.cmd == "pl-reorder":
        status, payload = request(
            args, host, port, "POST", f"{pl_route(args.name)}/reorder",
            body={"from": args.from_pos, "to": args.to_pos},
        )
    else:
        status, payload = request(args, host, port, "GET", "/help")

    print_response(args, status, payload)


if __name__ == "__main__":
    main()