- fetch <content-id|perma-url> [--save]: löst Perma-Links selbst auf - space-list + resolve_container: container_id akzeptiert Slug/URL/ID - SKILL.md entsprechend ergänzt Gleicht das Plugin an die kanonische humhub.py (dokumentierer) an. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
421 lines
16 KiB
Python
Executable File
421 lines
16 KiB
Python
Executable File
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = [
|
|
# "requests",
|
|
# "python-dotenv",
|
|
# ]
|
|
# ///
|
|
"""CLI for interacting with HumHub via REST API (wiki pages and posts)."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import requests
|
|
from dotenv import load_dotenv, find_dotenv
|
|
|
|
|
|
BASE_URL = "https://humhub.inmedias.it/api/v1"
|
|
|
|
|
|
def get_session() -> tuple[requests.Session, str]:
|
|
load_dotenv(Path.home() / ".config/dokumentierer/.env")
|
|
load_dotenv() # CWD/Projektverzeichnis, überschreibt nichts
|
|
token = os.environ.get("HUMHUB_API_TOKEN", "")
|
|
if not token:
|
|
sys.exit("Error: HUMHUB_API_TOKEN is not set in .env")
|
|
session = requests.Session()
|
|
session.headers.update({"Authorization": f"Bearer {token}"})
|
|
return session, token
|
|
|
|
|
|
def api_get(session: requests.Session, path: str) -> dict:
|
|
r = session.get(f"{BASE_URL}{path}")
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def api_put(session: requests.Session, path: str, payload: dict) -> dict:
|
|
r = session.put(f"{BASE_URL}{path}", json=payload)
|
|
if not r.ok:
|
|
sys.exit(f"Error {r.status_code}: {r.text}")
|
|
return r.json()
|
|
|
|
|
|
def api_post(session: requests.Session, path: str, payload: dict) -> dict:
|
|
r = session.post(f"{BASE_URL}{path}", json=payload)
|
|
if not r.ok:
|
|
sys.exit(f"Error {r.status_code}: {r.text}")
|
|
return r.json()
|
|
|
|
|
|
def resolve_container(session: requests.Session, value: str) -> int:
|
|
"""Resolve a container ID, space slug, or space URL to a contentcontainer_id."""
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
pass
|
|
# Strip full URL down to slug
|
|
slug = value.rstrip("/").split("/s/")[-1].rstrip("/") if "/s/" in value else value.strip("/")
|
|
page = 1
|
|
while True:
|
|
data = api_get(session, f"/space?limit=50&page={page}")
|
|
results = data.get("results", [])
|
|
if not results:
|
|
break
|
|
for space in results:
|
|
space_slug = space.get("url", "").rstrip("/").split("/s/")[-1].rstrip("/")
|
|
if space_slug == slug or space.get("name", "").lower() == slug.lower():
|
|
return space["contentcontainer_id"]
|
|
if len(results) < 50:
|
|
break
|
|
page += 1
|
|
sys.exit(f"Error: Space '{value}' nicht gefunden. Verfügbare Spaces: uv run humhub.py space-list")
|
|
|
|
|
|
def parse_topics(topics_str: str | None) -> list[dict]:
|
|
if not topics_str:
|
|
return []
|
|
return [{"name": t.strip()} for t in topics_str.split(",") if t.strip()]
|
|
|
|
|
|
def read_file(path: str) -> str:
|
|
if path == "-":
|
|
return sys.stdin.read()
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
return fh.read()
|
|
except OSError as exc:
|
|
sys.exit(f"Error reading file: {exc}")
|
|
|
|
|
|
# ── Wiki commands ────────────────────────────────────────────────────────────
|
|
|
|
def cmd_wiki_list(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
data = api_get(session, "/wiki?per-page=200")
|
|
query = args.filter.lower() if args.filter else None
|
|
for page in data["results"]:
|
|
if query and query not in page["title"].lower():
|
|
continue
|
|
print(f"{page['id']:>4} {page['title']}")
|
|
|
|
|
|
def cmd_wiki_get(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
data = api_get(session, f"/wiki/page/{args.id}")
|
|
title = data.get("title", "")
|
|
content = data.get("latest_revision", {}).get("content", "")
|
|
topics = [t["name"] for t in data.get("content", {}).get("topics", [])]
|
|
url = data.get("content", {}).get("metadata", {}).get("url", "")
|
|
print(f"Title: {title}")
|
|
print(f"Topics: {', '.join(topics) or '(keine)'}")
|
|
print(f"URL: https://humhub.inmedias.it{url}")
|
|
print(f"Content ({len(content)} Zeichen):")
|
|
print(content)
|
|
|
|
|
|
def cmd_wiki_update(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
content = read_file(args.file) if args.file else None
|
|
|
|
current = api_get(session, f"/wiki/page/{args.id}")
|
|
title = args.title or current.get("title", "")
|
|
|
|
if content is None:
|
|
content = current.get("latest_revision", {}).get("content", "")
|
|
|
|
wiki_page: dict = {"title": title}
|
|
if args.parent_id is not None:
|
|
wiki_page["parent_page_id"] = args.parent_id
|
|
|
|
payload: dict = {
|
|
"WikiPage": wiki_page,
|
|
"WikiPageRevision": {"content": content},
|
|
}
|
|
topics = parse_topics(args.topics)
|
|
if topics:
|
|
payload["content"] = {"topics": topics}
|
|
|
|
data = api_put(session, f"/wiki/page/{args.id}", payload)
|
|
saved_topics = [t["name"] for t in data.get("content", {}).get("topics", [])]
|
|
parent_id = data.get("parent_page_id")
|
|
print(f"Aktualisiert: '{data.get('title')}' (Revision {data.get('latest_revision', {}).get('id')}){f', parent={parent_id}' if parent_id else ''}")
|
|
if saved_topics:
|
|
print(f"Topics: {', '.join(saved_topics)}")
|
|
|
|
|
|
def cmd_wiki_create(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
container_id = resolve_container(session, args.container_id)
|
|
content = read_file(args.file) if args.file else ""
|
|
|
|
wiki_page: dict = {"title": args.title}
|
|
if args.parent_id is not None:
|
|
wiki_page["parent_page_id"] = args.parent_id
|
|
|
|
payload: dict = {
|
|
"WikiPage": wiki_page,
|
|
"WikiPageRevision": {"content": content},
|
|
}
|
|
topics = parse_topics(args.topics)
|
|
if topics:
|
|
payload["content"] = {"topics": topics}
|
|
|
|
data = api_post(session, f"/wiki/container/{container_id}", payload)
|
|
page_id = data.get("id")
|
|
url = data.get("content", {}).get("metadata", {}).get("url", "")
|
|
print(f"Erstellt: '{data.get('title')}' (ID {page_id})")
|
|
if url:
|
|
print(f"URL: https://humhub.inmedias.it{url}")
|
|
|
|
|
|
# ── Post commands ────────────────────────────────────────────────────────────
|
|
|
|
def cmd_post_list(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
container_id = resolve_container(session, args.container_id)
|
|
data = api_get(session, f"/post/container/{container_id}?limit={args.limit}")
|
|
results = data.get("results", [])
|
|
if not results:
|
|
print("(keine Posts)")
|
|
return
|
|
for post in results:
|
|
pid = post.get("id")
|
|
msg = post.get("message", "").replace("\n", " ")[:80]
|
|
topics = [t["name"] for t in post.get("content", {}).get("topics", [])]
|
|
topic_str = f" [{', '.join(topics)}]" if topics else ""
|
|
print(f"ID {pid}{topic_str}: {msg}")
|
|
|
|
|
|
def cmd_post_create(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
container_id = resolve_container(session, args.container_id)
|
|
message = read_file(args.file)
|
|
topics = parse_topics(args.topics)
|
|
|
|
payload: dict = {"data": {"message": message}}
|
|
if topics:
|
|
payload["data"]["content"] = {"topics": topics}
|
|
|
|
data = api_post(session, f"/post/container/{container_id}", payload)
|
|
pid = data.get("id")
|
|
saved_topics = [t["name"] for t in data.get("content", {}).get("topics", [])]
|
|
print(f"Post erstellt (ID {pid})")
|
|
if saved_topics:
|
|
print(f"Topics: {', '.join(saved_topics)}")
|
|
|
|
|
|
def cmd_post_update(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
message = read_file(args.file)
|
|
topics = parse_topics(args.topics)
|
|
|
|
payload: dict = {"data": {"message": message}}
|
|
if topics:
|
|
payload["data"]["content"] = {"topics": topics}
|
|
|
|
data = api_put(session, f"/post/{args.id}", payload)
|
|
pid = data.get("id")
|
|
saved_topics = [t["name"] for t in data.get("content", {}).get("topics", [])]
|
|
print(f"Post aktualisiert (ID {pid})")
|
|
if saved_topics:
|
|
print(f"Topics: {', '.join(saved_topics)}")
|
|
|
|
|
|
def cmd_space_list(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
page = 1
|
|
found = False
|
|
while True:
|
|
data = api_get(session, f"/space?limit=50&page={page}")
|
|
results = data.get("results", [])
|
|
if not results:
|
|
break
|
|
for space in results:
|
|
slug = space.get("url", "").rstrip("/").split("/s/")[-1].rstrip("/")
|
|
ccid = space.get("contentcontainer_id", "?")
|
|
name = space.get("name", "")
|
|
print(f"ccid={ccid:>4} slug={slug:<30} {name}")
|
|
found = True
|
|
if len(results) < 50:
|
|
break
|
|
page += 1
|
|
if not found:
|
|
print("(keine Spaces gefunden)")
|
|
|
|
|
|
# ── Fetch (perma-/content-id-agnostisch) ─────────────────────────────────────
|
|
|
|
def parse_content_ref(ref: str) -> int:
|
|
"""Akzeptiert eine blanke Content-ID oder einen Perma-Link
|
|
(…/content/perma?id=NNNN) und liefert die Content-ID."""
|
|
ref = ref.strip()
|
|
if ref.isdigit():
|
|
return int(ref)
|
|
qs = parse_qs(urlparse(ref).query)
|
|
if "id" in qs and qs["id"][0].isdigit():
|
|
return int(qs["id"][0])
|
|
sys.exit(f"Error: keine Content-ID erkennbar in '{ref}'")
|
|
|
|
|
|
def slugify(text: str) -> str:
|
|
text = text.lower()
|
|
for a, b in (("ä", "ae"), ("ö", "oe"), ("ü", "ue"), ("ß", "ss")):
|
|
text = text.replace(a, b)
|
|
text = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
|
|
return text[:60] or "inhalt"
|
|
|
|
|
|
def resolve_content(session: requests.Session, content_id: int) -> dict:
|
|
"""Löst eine Content-ID auf den zugrundeliegenden Datensatz auf
|
|
(WikiPage oder Post) und liefert Titel, Inhalt und Metadaten."""
|
|
envelope = api_get(session, f"/content/{content_id}")
|
|
meta = envelope.get("metadata", {})
|
|
model = meta.get("object_model", "")
|
|
object_id = meta.get("object_id")
|
|
page_url = meta.get("url", "")
|
|
topics = [t["name"] for t in envelope.get("topics", [])]
|
|
updated_at = meta.get("updated_at", "")
|
|
|
|
if model.endswith("WikiPage"):
|
|
data = api_get(session, f"/wiki/page/{object_id}")
|
|
title = data.get("title", "") or f"Wiki-Seite {object_id}"
|
|
content = data.get("latest_revision", {}).get("content", "")
|
|
kind = f"WikiPage (page {object_id})"
|
|
seg = page_url.rstrip("/").split("/")[-1]
|
|
slug = seg if seg and not seg.isdigit() else slugify(title)
|
|
elif model.endswith("Post"):
|
|
data = api_get(session, f"/post/{object_id}")
|
|
content = data.get("data", {}).get("message") or data.get("message", "")
|
|
first = content.splitlines()[0].lstrip("# ").strip() if content else ""
|
|
title = first or f"Post {object_id}"
|
|
kind = f"Post (id {object_id})"
|
|
slug = slugify(title)
|
|
else:
|
|
sys.exit(f"Error: Objekttyp nicht unterstützt: {model or '(unbekannt)'}")
|
|
|
|
return {
|
|
"content_id": content_id,
|
|
"title": title,
|
|
"content": content,
|
|
"kind": kind,
|
|
"topics": topics,
|
|
"updated_at": updated_at,
|
|
"page_url": f"https://humhub.inmedias.it{page_url}" if page_url else "",
|
|
"slug": slug,
|
|
}
|
|
|
|
|
|
def render_resource(info: dict) -> str:
|
|
"""Inhalt mit Herkunfts-Kopf als Markdown-Kommentar."""
|
|
perma = f"https://humhub.inmedias.it/content/perma?id={info['content_id']}"
|
|
header = (
|
|
"<!--\n"
|
|
f"Quelle: {perma}\n"
|
|
f"Seite: {info['page_url']}\n"
|
|
f"Titel: {info['title']}\n"
|
|
f"Typ: {info['kind']}\n"
|
|
f"Topics: {', '.join(info['topics']) or '(keine)'}\n"
|
|
f"Stand: {info['updated_at']}\n"
|
|
f"Abgerufen: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n"
|
|
"-->\n\n"
|
|
)
|
|
return header + info["content"].rstrip() + "\n"
|
|
|
|
|
|
def cmd_fetch(args: argparse.Namespace) -> None:
|
|
session, _ = get_session()
|
|
content_id = parse_content_ref(args.ref)
|
|
info = resolve_content(session, content_id)
|
|
body = render_resource(info)
|
|
|
|
if args.save:
|
|
outdir = Path(args.out)
|
|
outdir.mkdir(parents=True, exist_ok=True)
|
|
path = outdir / f"humhub-{content_id}-{info['slug']}.md"
|
|
path.write_text(body, encoding="utf-8")
|
|
print(f"Gespeichert: {path}")
|
|
print(f" {info['kind']} — '{info['title']}' ({len(info['content'])} Zeichen)")
|
|
else:
|
|
sys.stdout.write(body)
|
|
|
|
|
|
# ── Main ─────────────────────────────────────────────────────────────────────
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="HumHub REST API CLI")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# fetch
|
|
p = sub.add_parser("fetch", help="Content-ID/Perma-Link auflösen und Inhalt holen (WikiPage oder Post)")
|
|
p.add_argument("ref", help="Content-ID oder Perma-Link (…/content/perma?id=NNNN)")
|
|
p.add_argument("--save", action="store_true", help="In Datei ablegen statt auf stdout")
|
|
p.add_argument("--out", default="resources", help="Zielverzeichnis für --save (Standard: resources/)")
|
|
p.set_defaults(func=cmd_fetch)
|
|
|
|
# wiki-list
|
|
p = sub.add_parser("wiki-list", help="Alle Wiki-Seiten auflisten")
|
|
p.add_argument("filter", nargs="?", help="Filter nach Titel (case-insensitiv)")
|
|
p.set_defaults(func=cmd_wiki_list)
|
|
|
|
# wiki-get
|
|
p = sub.add_parser("wiki-get", help="Wiki-Seite anzeigen")
|
|
p.add_argument("id", type=int, help="Wiki-Seiten-ID")
|
|
p.set_defaults(func=cmd_wiki_get)
|
|
|
|
# wiki-update
|
|
p = sub.add_parser("wiki-update", help="Wiki-Seite aktualisieren")
|
|
p.add_argument("id", type=int, help="Wiki-Seiten-ID")
|
|
p.add_argument("file", nargs="?", default=None, help="Markdown-Datei oder '-' für stdin (optional)")
|
|
p.add_argument("--title", default=None, help="Neuer Titel (sonst unverändert)")
|
|
p.add_argument("--topics", default=None, help="Komma-getrennte Topic-Namen (ersetzt bestehende)")
|
|
p.add_argument("--parent-id", type=int, default=None, dest="parent_id", help="ID der Oberseite")
|
|
p.set_defaults(func=cmd_wiki_update)
|
|
|
|
# space-list
|
|
p = sub.add_parser("space-list", help="Spaces auflisten mit contentcontainer_id")
|
|
p.set_defaults(func=cmd_space_list)
|
|
|
|
# wiki-create
|
|
p = sub.add_parser("wiki-create", help="Neue Wiki-Seite anlegen")
|
|
p.add_argument("container_id", help="Space-Slug, URL oder numerische Container-ID")
|
|
p.add_argument("title", help="Seitentitel")
|
|
p.add_argument("file", nargs="?", default=None, help="Markdown-Datei oder '-' für stdin (optional)")
|
|
p.add_argument("--topics", default=None, help="Komma-getrennte Topic-Namen")
|
|
p.add_argument("--parent-id", type=int, default=None, dest="parent_id", help="ID der Oberseite")
|
|
p.set_defaults(func=cmd_wiki_create)
|
|
|
|
# post-list
|
|
p = sub.add_parser("post-list", help="Posts eines Containers auflisten")
|
|
p.add_argument("container_id", help="Space-Slug, URL oder numerische Container-ID")
|
|
p.add_argument("--limit", type=int, default=10, help="Max. Anzahl Posts (Standard: 10)")
|
|
p.set_defaults(func=cmd_post_list)
|
|
|
|
# post-create
|
|
p = sub.add_parser("post-create", help="Neuen Post erstellen")
|
|
p.add_argument("container_id", help="Space-Slug, URL oder numerische Container-ID")
|
|
p.add_argument("file", help="Markdown-Datei oder '-' für stdin")
|
|
p.add_argument("--topics", default=None, help="Komma-getrennte Topic-Namen")
|
|
p.set_defaults(func=cmd_post_create)
|
|
|
|
# post-update
|
|
p = sub.add_parser("post-update", help="Post aktualisieren")
|
|
p.add_argument("id", type=int, help="Post-ID")
|
|
p.add_argument("file", help="Markdown-Datei oder '-' für stdin")
|
|
p.add_argument("--topics", default=None, help="Komma-getrennte Topic-Namen (ersetzt bestehende)")
|
|
p.set_defaults(func=cmd_post_update)
|
|
|
|
args = parser.parse_args()
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|