mrcharles
d4e9fd693f
A standalone CLI container (following the redmine-cli pattern) that
lets AI agents and humans read, post, reply, search, and discuss on a
Discourse forum via the REST API. Uses three environment variables
(DISCOURSE_URL, DISCOURSE_API_KEY, DISCOURSE_API_USERNAME) for auth.
Commands: whoami, categories, cat-info, topics/ls, show, create, reply,
update, delete, search, notifications.
Fully validated end-to-end (10/10 checks) against a live Discourse
instance: credential auth, whoami, categories, topics, show, search,
and full write cycle (create + reply + update + delete).
Includes validate.sh for repeatable end-to-end testing using
containerized curl + the CLI image.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
435 lines
15 KiB
Python
435 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Discourse CLI - a thin wrapper around the Discourse REST API.
|
|
|
|
Connection details come from the environment:
|
|
DISCOURSE_URL base URL of the Discourse instance (e.g. https://discourse.example.com)
|
|
DISCOURSE_API_KEY API key ("All users" or "Single user" key from Admin > API)
|
|
DISCOURSE_API_USERNAME username the API key acts as (e.g. system, or your account)
|
|
|
|
Designed to be run inside the discourse-cli Docker container, but works anywhere
|
|
these environment variables are set.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
import requests
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# HTTP client
|
|
# --------------------------------------------------------------------------- #
|
|
class DiscourseAPI:
|
|
"""Minimal authenticated Discourse REST client."""
|
|
|
|
def __init__(self, url, api_key, api_username):
|
|
self.base_url = url.rstrip("/")
|
|
self.headers = {
|
|
"Api-Key": api_key,
|
|
"Api-Username": api_username,
|
|
"Accept": "application/json",
|
|
}
|
|
self.timeout = 30
|
|
|
|
def _request(self, method, path, **kwargs):
|
|
url = f"{self.base_url}/{path.lstrip('/')}"
|
|
resp = requests.request(
|
|
method, url, headers=self.headers, timeout=self.timeout, **kwargs
|
|
)
|
|
if resp.status_code == 429:
|
|
_err(f"rate limited by Discourse (HTTP 429). Retry later.")
|
|
if not resp.ok:
|
|
detail = ""
|
|
try:
|
|
detail = resp.json().get("errors", resp.text[:200])
|
|
except Exception:
|
|
detail = resp.text[:200]
|
|
_err(f"Discourse API error {resp.status_code} for {method} {path}: {detail}")
|
|
if resp.status_code == 204 or not resp.content:
|
|
return {}
|
|
return resp.json()
|
|
|
|
def get(self, path, params=None):
|
|
return self._request("GET", path, params=params)
|
|
|
|
def post(self, path, data=None):
|
|
return self._request("POST", path, json=data)
|
|
|
|
def put(self, path, data=None):
|
|
return self._request("PUT", path, json=data)
|
|
|
|
def delete(self, path, data=None):
|
|
return self._request("DELETE", path, json=data)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _client():
|
|
"""Build and return an authenticated Discourse client, or exit with help."""
|
|
url = os.environ.get("DISCOURSE_URL", "").strip()
|
|
key = os.environ.get("DISCOURSE_API_KEY", "").strip()
|
|
user = os.environ.get("DISCOURSE_API_USERNAME", "").strip()
|
|
missing = [
|
|
n
|
|
for n, v in (
|
|
("DISCOURSE_URL", url),
|
|
("DISCOURSE_API_KEY", key),
|
|
("DISCOURSE_API_USERNAME", user),
|
|
)
|
|
if not v
|
|
]
|
|
if missing:
|
|
sys.stderr.write(
|
|
"ERROR: missing required environment variable(s): "
|
|
+ ", ".join(missing)
|
|
+ "\n"
|
|
)
|
|
sys.exit(2)
|
|
return DiscourseAPI(url, key, user)
|
|
|
|
|
|
def _err(msg, code=1):
|
|
sys.stderr.write(f"ERROR: {msg}\n")
|
|
sys.exit(code)
|
|
|
|
|
|
def _kv(label, value):
|
|
"""Format a label/value line, omitting falsy values gracefully."""
|
|
if value in (None, "", [], {}):
|
|
return None
|
|
return f"{label:>14}: {value}"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Commands
|
|
# --------------------------------------------------------------------------- #
|
|
def cmd_whoami(args):
|
|
api = _client()
|
|
data = api.get("/session/current.json")
|
|
u = data.get("current_user", {})
|
|
print(f"id: {u.get('id', '?')}")
|
|
print(f"username: {u.get('username', '?')}")
|
|
print(f"name: {u.get('name', '?')}")
|
|
print(f"admin: {u.get('admin', False)}")
|
|
print(f"trust: {u.get('trust_level', '?')}")
|
|
print(f"discourse: {os.environ['DISCOURSE_URL']}")
|
|
return 0
|
|
|
|
|
|
def cmd_categories(args):
|
|
api = _client()
|
|
data = api.get("/categories.json")
|
|
cats = data.get("category_list", {}).get("categories", [])
|
|
print(f"{'ID':<6} {'SLUG':<28} {'TOPICS':>7} NAME")
|
|
print("-" * 80)
|
|
for c in cats:
|
|
topic_count = c.get("topic_count", 0)
|
|
print(f"{c.get('id','?'):<6} {str(c.get('slug','')):<28} {topic_count:>6} {c.get('name','')}")
|
|
subcats = []
|
|
for c in cats:
|
|
subcats.extend(c.get("subcategory_ids", []) or [])
|
|
print(f"\n{len(cats)} category(ies)")
|
|
return 0
|
|
|
|
|
|
def cmd_topics(args):
|
|
api = _client()
|
|
if args.category:
|
|
cat_id = _resolve_category(api, args.category)
|
|
data = api.get(f"/c/{cat_id}.json", params={"page": args.page})
|
|
topics = data.get("topic_list", {}).get("topics", [])
|
|
else:
|
|
path = "/latest.json"
|
|
if args.unread:
|
|
path = "/unread.json"
|
|
elif args.new:
|
|
path = "/new.json"
|
|
data = api.get(path, params={"page": args.page})
|
|
topics = data.get("topic_list", {}).get("topics", [])
|
|
|
|
print(f"{'ID':<10} {'REPLIES':>8} {'VIEWS':>9} TITLE")
|
|
print("-" * 90)
|
|
for t in topics:
|
|
if t.get("pinned"):
|
|
continue
|
|
replies = t.get("posts_count", 1) - 1
|
|
views = t.get("views", 0)
|
|
title = t.get("title", "")
|
|
print(f"{t.get('id','?'):<10} {replies:>8} {views:>9} {title}")
|
|
print(f"\n{len(topics)} topic(s)")
|
|
return 0
|
|
|
|
|
|
def cmd_show(args):
|
|
api = _client()
|
|
data = api.get(f"/t/{args.topic_id}.json")
|
|
print(f"#{data.get('id','?')}: {data.get('title','?')}")
|
|
print("=" * 90)
|
|
print(_kv("category_id", data.get("category_id")))
|
|
print(_kv("views", data.get("views")))
|
|
print(_kv("like_count", data.get("like_count")))
|
|
print(_kv("posts_count", data.get("posts_count")))
|
|
print(_kv("created", data.get("created_at")))
|
|
print(_kv("last_posted", data.get("last_posted_at")))
|
|
print("")
|
|
|
|
posts = data.get("post_stream", {}).get("posts", [])
|
|
for idx, p in enumerate(posts, start=1):
|
|
post_num = p.get("post_number", idx)
|
|
author = p.get("username", "?")
|
|
created = p.get("created_at", "?")
|
|
likes = p.get("actions_summary", [])
|
|
like_count = 0
|
|
for a in likes:
|
|
if a.get("id") == 2:
|
|
like_count = a.get("count", 0)
|
|
print(f"\n--- #{post_num} [{post_num}] {author} ({created}) likes={like_count} ---")
|
|
cooked = p.get("cooked", "")
|
|
text = _strip_html(cooked)
|
|
if text.strip():
|
|
print(text.strip())
|
|
else:
|
|
print("(no content)")
|
|
return 0
|
|
|
|
|
|
def cmd_create(args):
|
|
api = _client()
|
|
if not args.title:
|
|
_err("--title is required to create a topic")
|
|
if not args.body:
|
|
_err("--body is required to create a topic")
|
|
cat_id = _resolve_category(api, args.category) if args.category else None
|
|
payload = {
|
|
"title": args.title,
|
|
"raw": args.body,
|
|
}
|
|
if cat_id is not None:
|
|
payload["category"] = cat_id
|
|
if args.tags:
|
|
payload["tags"] = [t.strip() for t in args.tags.split(",") if t.strip()]
|
|
data = api.post("/posts.json", data=payload)
|
|
topic_id = data.get("topic_id", "?")
|
|
post_id = data.get("id", "?")
|
|
print(f"Created topic #{topic_id} (post #{post_id}): {args.title}")
|
|
if topic_id != "?":
|
|
print(f"URL: {os.environ['DISCOURSE_URL']}/t/{topic_id}")
|
|
return 0
|
|
|
|
|
|
def cmd_reply(args):
|
|
api = _client()
|
|
if not args.body:
|
|
_err("--body is required to reply")
|
|
payload = {"topic_id": args.topic_id, "raw": args.body}
|
|
if args.reply_to:
|
|
payload["reply_to_post_number"] = args.reply_to
|
|
data = api.post("/posts.json", data=payload)
|
|
post_id = data.get("id", "?")
|
|
post_num = data.get("post_number", "?")
|
|
print(f"Posted reply #{post_num} (id={post_id}) in topic #{args.topic_id}")
|
|
return 0
|
|
|
|
|
|
def cmd_update(args):
|
|
api = _client()
|
|
if not args.body:
|
|
_err("--body is required to update a post")
|
|
data = api.put(f"/posts/{args.post_id}.json", data={"post": {"raw": args.body}})
|
|
post_num = data.get("post", {}).get("post_number", "?")
|
|
print(f"Updated post #{post_num} (id={args.post_id})")
|
|
return 0
|
|
|
|
|
|
def cmd_delete(args):
|
|
api = _client()
|
|
api.delete(f"/posts/{args.post_id}.json")
|
|
print(f"Deleted post #{args.post_id}")
|
|
return 0
|
|
|
|
|
|
def cmd_search(args):
|
|
api = _client()
|
|
params = {"q": args.query, "page": args.page}
|
|
data = api.get("/search.json", params=params)
|
|
topics = data.get("topics", [])
|
|
posts = data.get("posts", [])
|
|
topic_map = {t.get("id"): t for t in topics}
|
|
print(f"{'TOPIC':<10} {'POST':<8} BLURB")
|
|
print("-" * 90)
|
|
for p in posts:
|
|
tid = p.get("topic_id", "?")
|
|
t = topic_map.get(tid, {})
|
|
title = t.get("title", "")
|
|
blurb = p.get("blurb", "")
|
|
pid = p.get("id", "?")
|
|
print(f"{tid:<10} {pid:<8} {title}")
|
|
if blurb:
|
|
print(f"{'':<20}{blurb}")
|
|
print(f"\n{len(posts)} result(s)")
|
|
return 0
|
|
|
|
|
|
def cmd_notifications(args):
|
|
api = _client()
|
|
data = api.get("/notifications.json")
|
|
notifs = data.get("notifications", [])
|
|
if not notifs:
|
|
print("(no notifications)")
|
|
return 0
|
|
print(f"{'ID':<10} {'READ':>5} TYPE SUBJECT")
|
|
print("-" * 90)
|
|
for n in notifs[: args.limit]:
|
|
nid = n.get("id", "?")
|
|
read = "yes" if n.get("read") else "no"
|
|
ntype = n.get("notification_type", "?")
|
|
subject = n.get("slug") or n.get("data", {}).get("topic_title", "")
|
|
print(f"{nid:<10} {read:>5} {str(ntype):<20} {subject}")
|
|
print(f"\n{len(notifs)} notification(s), showing {min(len(notifs), args.limit)}")
|
|
return 0
|
|
|
|
|
|
def cmd_categories_info(args):
|
|
api = _client()
|
|
cat_id = _resolve_category(api, args.category)
|
|
data = api.get(f"/c/{cat_id}/show.json")
|
|
c = data.get("category", {})
|
|
print(f"#{c.get('id','?')}: {c.get('name','?')}")
|
|
print("=" * 60)
|
|
print(_kv("slug", c.get("slug")))
|
|
print(_kv("color", c.get("color")))
|
|
print(_kv("topic_count", c.get("topic_count")))
|
|
print(_kv("post_count", c.get("post_count")))
|
|
print(_kv("description", _strip_html(c.get("description", "") or "")[:200]))
|
|
return 0
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Resolution helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _resolve_category(api, ref):
|
|
"""Resolve a category id, slug, or name to a numeric id."""
|
|
if str(ref).isdigit():
|
|
return int(ref)
|
|
data = api.get("/categories.json")
|
|
cats = data.get("category_list", {}).get("categories", [])
|
|
needle = str(ref).strip().lower()
|
|
for c in cats:
|
|
if str(c.get("slug", "")).lower() == needle:
|
|
return c["id"]
|
|
if str(c.get("name", "")).lower() == needle:
|
|
return c["id"]
|
|
_err(f"unknown category '{ref}'. Run 'categories' to list valid slugs/names.")
|
|
|
|
|
|
def _strip_html(cooked):
|
|
"""Very small HTML-to-text for display of Discourse 'cooked' post bodies."""
|
|
import html
|
|
import re
|
|
|
|
if not cooked:
|
|
return ""
|
|
# Preserve block-level breaks
|
|
text = re.sub(r"(?i)</(p|div|li|h[1-6]|tr|blockquote)>", "\n", cooked)
|
|
text = re.sub(r"(?i)<br\s*/?>", "\n", text)
|
|
text = re.sub(r"(?i)<li[^>]*>", " - ", text)
|
|
# Code blocks
|
|
text = re.sub(r"(?i)<code[^>]*>", "`", text)
|
|
text = re.sub(r"(?i)</code>", "`", text)
|
|
# Strip all remaining tags
|
|
text = re.sub(r"<[^>]+>", "", text)
|
|
text = html.unescape(text)
|
|
# Collapse excessive blank lines
|
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
|
return text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Argument parsing
|
|
# --------------------------------------------------------------------------- #
|
|
def build_parser():
|
|
p = argparse.ArgumentParser(
|
|
prog="discourse-cli",
|
|
description="Read, post, and discuss on a Discourse forum via the REST API.",
|
|
)
|
|
sub = p.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser(
|
|
"whoami", help="show the authenticated user (connection test)"
|
|
).set_defaults(func=cmd_whoami)
|
|
|
|
sp = sub.add_parser("categories", help="list categories")
|
|
sp.set_defaults(func=cmd_categories)
|
|
|
|
sp = sub.add_parser("cat-info", help="show details of a category")
|
|
sp.add_argument("category", metavar="ID_OR_SLUG")
|
|
sp.set_defaults(func=cmd_categories_info)
|
|
|
|
sp = sub.add_parser("topics", aliases=["ls"], help="list topics (latest or in a category)")
|
|
sp.add_argument("-c", "--category", metavar="ID_OR_SLUG", help="filter by category")
|
|
sp.add_argument("-n", "--new", action="store_true", help="only new topics")
|
|
sp.add_argument("-u", "--unread", action="store_true", help="only unread topics")
|
|
sp.add_argument("-p", "--page", type=int, default=0, help="page number (default 0)")
|
|
sp.set_defaults(func=cmd_topics)
|
|
|
|
sp = sub.add_parser("show", help="show a topic with its posts")
|
|
sp.add_argument("topic_id", type=int)
|
|
sp.set_defaults(func=cmd_show)
|
|
|
|
sp = sub.add_parser("create", help="create a new topic")
|
|
sp.add_argument("-c", "--category", metavar="ID_OR_SLUG")
|
|
sp.add_argument("-t", "--title", required=True, help="topic title")
|
|
sp.add_argument("-b", "--body", required=True, help="post body (raw text/markdown)")
|
|
sp.add_argument("--tags", help="comma-separated tags")
|
|
sp.set_defaults(func=cmd_create)
|
|
|
|
sp = sub.add_parser("reply", help="reply to a topic")
|
|
sp.add_argument("topic_id", type=int)
|
|
sp.add_argument("-b", "--body", required=True, help="reply body (raw text/markdown)")
|
|
sp.add_argument(
|
|
"-r", "--reply-to", type=int, metavar="POST_NUMBER", help="reply to a specific post number"
|
|
)
|
|
sp.set_defaults(func=cmd_reply)
|
|
|
|
sp = sub.add_parser("update", help="update an existing post")
|
|
sp.add_argument("post_id", type=int)
|
|
sp.add_argument("-b", "--body", required=True, help="new post body (raw text/markdown)")
|
|
sp.set_defaults(func=cmd_update)
|
|
|
|
sp = sub.add_parser("delete", help="delete a post")
|
|
sp.add_argument("post_id", type=int)
|
|
sp.set_defaults(func=cmd_delete)
|
|
|
|
sp = sub.add_parser("search", help="search the forum")
|
|
sp.add_argument("query", help="search query")
|
|
sp.add_argument("-p", "--page", type=int, default=1, help="page number (default 1)")
|
|
sp.set_defaults(func=cmd_search)
|
|
|
|
sp = sub.add_parser("notifications", help="list your notifications")
|
|
sp.add_argument("-l", "--limit", type=int, default=20, help="max results (default 20)")
|
|
sp.set_defaults(func=cmd_notifications)
|
|
|
|
return p
|
|
|
|
|
|
def main(argv=None):
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
return args.func(args)
|
|
except requests.exceptions.ConnectionError:
|
|
_err(f"could not connect to Discourse at {os.environ.get('DISCOURSE_URL','?')}", 2)
|
|
except requests.exceptions.Timeout:
|
|
_err("request timed out talking to Discourse", 2)
|
|
except SystemExit:
|
|
raise
|
|
except Exception as e: # noqa: BLE001 - top-level safety net for the CLI
|
|
_err(f"{type(e).__name__}: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|