feat: add plug-and-play devRant API client examples with Python and JavaScript
Add complete set of example scripts for the devRant REST protocol under examples/devrant/, including reusable client libraries, a rant poster, a live feed watcher with keyword-based auto-upvote, and an end-to-end smoke test that exercises every endpoint. Both Python (stdlib-only) and JavaScript (Node 18+ global fetch) implementations are provided.
This commit is contained in:
@@ -0,0 +1,100 @@
|
||||
# DevPlace XML-RPC examples
|
||||
|
||||
Runnable Python examples for the DevPlace XML-RPC bridge (`/xmlrpc`). They use only the
|
||||
standard library (`xmlrpc.client`); no dependencies to install. Full reference:
|
||||
`/docs/xmlrpc.html`.
|
||||
|
||||
## Files
|
||||
|
||||
| File | What it does |
|
||||
|------|--------------|
|
||||
| `client.py` | Reusable `DevPlace` wrapper: credential injection, all three auth styles, dotted-name calls, introspection helpers. Imported by the others. |
|
||||
| `introspect.py` | Prints every available method with its help text. Needs no credentials. |
|
||||
| `recipes.py` | Post, reply, vote, react, follow, message, and paginate the feed. |
|
||||
| `mention_bot.py` | A bot that polls notifications and replies once to every `@mention`. |
|
||||
| `chatbot.py` | An interactive chat bot over direct messages: polls the inbox, answers commands (`help`, `feed`, `echo`, `whoami`), and replies. |
|
||||
| `ai_chatbot.py` | An AI-powered DM bot: message I/O over XML-RPC, replies generated by the DevPlace AI gateway, with per-user history. |
|
||||
| `dm_client.py` | A two-way terminal chat client: send and receive direct messages with one user live. |
|
||||
| `feed_watch.py` | Live feed ticker: prints new posts as they appear, optionally auto-upvotes posts matching keywords. |
|
||||
| `digest.py` | Builds a digest of the latest posts, top members, and developer news; prints it or posts it. |
|
||||
| `gist_backup.py` | Downloads all your gists to local files named by language. |
|
||||
| `deploy_project.py` | Pushes a local directory into a new DevPlace project's virtual filesystem. |
|
||||
| `rss_to_posts.py` | Polls an RSS/Atom feed and creates a post for each new entry. |
|
||||
| `export_data.py` | Exports your profile, posts, gists, projects, bookmarks, and notifications to one JSON file. |
|
||||
|
||||
## Configuration
|
||||
|
||||
The scripts read their settings from environment variables:
|
||||
|
||||
| Variable | Default | Purpose |
|
||||
|----------|---------|---------|
|
||||
| `DEVPLACE_URL` | `https://devplace.net` | Full base URL of the DevPlace instance (scheme + host, no path) |
|
||||
| `DEVPLACE_API_KEY` | `YOUR_API_KEY` | Your API key (from your profile page) |
|
||||
| `DEVPLACE_POLL_SECONDS` | varies | Poll interval for the bots/watchers |
|
||||
| `DEVPLACE_REPLY` | (a default) | Mention bot reply text |
|
||||
|
||||
`DEVPLACE_URL` defaults to `https://devplace.net`, so you only need to set `DEVPLACE_API_KEY`
|
||||
to run the authenticated examples against the public instance.
|
||||
|
||||
## Run
|
||||
|
||||
```bash
|
||||
# discover the API (no credentials needed)
|
||||
python introspect.py
|
||||
|
||||
# run the automation recipes
|
||||
DEVPLACE_API_KEY=xxxx python recipes.py
|
||||
|
||||
# run the mention bot (loops forever)
|
||||
DEVPLACE_API_KEY=xxxx python mention_bot.py
|
||||
|
||||
# run the direct-message chat bot (loops forever, answers commands)
|
||||
DEVPLACE_API_KEY=xxxx python chatbot.py
|
||||
|
||||
# open a two-way terminal chat with one user
|
||||
DEVPLACE_API_KEY=xxxx python dm_client.py alice
|
||||
|
||||
# AI-powered DM bot (replies via the DevPlace AI gateway)
|
||||
DEVPLACE_API_KEY=xxxx python ai_chatbot.py
|
||||
|
||||
# watch the feed live and auto-upvote posts mentioning "rust" or "python"
|
||||
DEVPLACE_API_KEY=xxxx DEVPLACE_UPVOTE_KEYWORDS=rust,python python feed_watch.py
|
||||
|
||||
# build a digest and post it to the feed
|
||||
DEVPLACE_API_KEY=xxxx DEVPLACE_DIGEST_POST=1 python digest.py
|
||||
|
||||
# back up every gist to ./gists_backup/
|
||||
DEVPLACE_API_KEY=xxxx python gist_backup.py
|
||||
|
||||
# deploy a local folder as a new project
|
||||
DEVPLACE_API_KEY=xxxx python deploy_project.py ./my-app "My App"
|
||||
|
||||
# mirror an RSS feed into posts
|
||||
DEVPLACE_API_KEY=xxxx DEVPLACE_RSS_URL=https://example.com/feed.xml python rss_to_posts.py
|
||||
|
||||
# export all your data to JSON
|
||||
DEVPLACE_API_KEY=xxxx python export_data.py alice
|
||||
```
|
||||
|
||||
To run against a local dev server, point `DEVPLACE_URL` at it:
|
||||
|
||||
```bash
|
||||
DEVPLACE_URL=http://localhost:10500 DEVPLACE_API_KEY=xxxx python recipes.py
|
||||
```
|
||||
|
||||
## Authentication
|
||||
|
||||
`client.py` supports all three methods documented at `/docs/xmlrpc.html`:
|
||||
|
||||
```python
|
||||
from client import DevPlace
|
||||
|
||||
# 1. API key (sent inside each call's parameter struct)
|
||||
dp = DevPlace("https://devplace.net", api_key="YOUR_API_KEY")
|
||||
|
||||
# 2. Username and password (HTTP Basic via a credentialed URL)
|
||||
dp = DevPlace("https://devplace.net", username="alice", password="secret123")
|
||||
```
|
||||
|
||||
To authenticate with the API key as an HTTP header instead, see the `ApiKeyTransport` example
|
||||
in `/docs/xmlrpc.html`.
|
||||
@@ -0,0 +1,81 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import urllib.request
|
||||
import xmlrpc.client
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
POLL_SECONDS = int(os.environ.get("DEVPLACE_POLL_SECONDS", "10"))
|
||||
MODEL = os.environ.get("DEVPLACE_AI_MODEL", "molodetz")
|
||||
SYSTEM_PROMPT = os.environ.get(
|
||||
"DEVPLACE_AI_SYSTEM",
|
||||
"You are a helpful assistant on the DevPlace developer network. Keep replies short.",
|
||||
)
|
||||
HISTORY_TURNS = 8
|
||||
FALLBACK = "Sorry, I cannot reach my brain right now. Try again in a moment."
|
||||
|
||||
|
||||
def ask_ai(prompt_history: list[dict]) -> str:
|
||||
body = json.dumps({"model": MODEL, "messages": prompt_history}).encode("utf-8")
|
||||
request = urllib.request.Request(
|
||||
f"{URL}/openai/v1/chat/completions",
|
||||
data=body,
|
||||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}"},
|
||||
)
|
||||
with urllib.request.urlopen(request, timeout=60) as response:
|
||||
payload = json.loads(response.read())
|
||||
return payload["choices"][0]["message"]["content"].strip()
|
||||
|
||||
|
||||
def run() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
histories: dict[str, list[dict]] = {}
|
||||
answered: set[str] = set()
|
||||
print(f"AI chat bot started; polling {URL} every {POLL_SECONDS}s.")
|
||||
while True:
|
||||
try:
|
||||
for conversation in dp.call("messages.inbox").get("conversations", []):
|
||||
if not conversation.get("unread"):
|
||||
continue
|
||||
other = conversation.get("other_user") or {}
|
||||
if not other.get("uid"):
|
||||
continue
|
||||
thread = dp.call("messages.inbox", with_uid=other["uid"])
|
||||
incoming = [m for m in thread.get("messages", []) if not m.get("is_mine")]
|
||||
if not incoming:
|
||||
continue
|
||||
last = incoming[-1]
|
||||
if last["message"]["uid"] in answered:
|
||||
continue
|
||||
answered.add(last["message"]["uid"])
|
||||
|
||||
history = histories.setdefault(
|
||||
other["uid"], [{"role": "system", "content": SYSTEM_PROMPT}]
|
||||
)
|
||||
history.append({"role": "user", "content": last["message"]["content"]})
|
||||
history[:] = history[:1] + history[-HISTORY_TURNS * 2 :]
|
||||
try:
|
||||
reply = ask_ai(history)
|
||||
except Exception as error:
|
||||
print("AI error:", error)
|
||||
reply = FALLBACK
|
||||
else:
|
||||
history.append({"role": "assistant", "content": reply})
|
||||
dp.call("messages.send", content=reply, receiver_uid=other["uid"])
|
||||
print(f"Answered @{other.get('username')}")
|
||||
except xmlrpc.client.Fault as fault:
|
||||
print("Poll error:", fault.faultCode, fault.faultString)
|
||||
except OSError as error:
|
||||
print("Network error, retrying:", error)
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
@@ -0,0 +1,72 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
import xmlrpc.client
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
POLL_SECONDS = int(os.environ.get("DEVPLACE_POLL_SECONDS", "10"))
|
||||
|
||||
|
||||
def respond(dp: DevPlace, text: str, sender: dict) -> str:
|
||||
text = (text or "").strip()
|
||||
command = text.lower()
|
||||
name = sender.get("username", "there")
|
||||
if command in ("hi", "hello", "hey", "help", "/help", "/start"):
|
||||
return (
|
||||
f"Hi @{name}! I am a DevPlace chat bot reachable over XML-RPC. "
|
||||
"Try: 'feed' for the latest posts, 'echo your message', or 'whoami'."
|
||||
)
|
||||
if command == "whoami":
|
||||
return f"You are @{name} (uid {sender.get('uid')})."
|
||||
if command == "feed":
|
||||
page = dp.call("feed.list")
|
||||
lines = [
|
||||
f"- {item['post'].get('title') or item['post']['uid']} by @{item['author']['username']}"
|
||||
for item in page.get("posts", [])[:3]
|
||||
]
|
||||
return "Latest posts:\n" + "\n".join(lines) if lines else "The feed is empty."
|
||||
if command.startswith("echo "):
|
||||
return text[5:]
|
||||
return "I did not understand that. Send 'help' for the list of commands."
|
||||
|
||||
|
||||
def run() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
answered: set[str] = set()
|
||||
print(f"Chat bot started; polling {URL} every {POLL_SECONDS}s.")
|
||||
while True:
|
||||
try:
|
||||
inbox = dp.call("messages.inbox")
|
||||
for conversation in inbox.get("conversations", []):
|
||||
if not conversation.get("unread"):
|
||||
continue
|
||||
other = conversation.get("other_user") or {}
|
||||
if not other.get("uid"):
|
||||
continue
|
||||
thread = dp.call("messages.inbox", with_uid=other["uid"])
|
||||
incoming = [m for m in thread.get("messages", []) if not m.get("is_mine")]
|
||||
if not incoming:
|
||||
continue
|
||||
last = incoming[-1]
|
||||
message_uid = last["message"]["uid"]
|
||||
if message_uid in answered:
|
||||
continue
|
||||
answered.add(message_uid)
|
||||
reply = respond(dp, last["message"]["content"], other)
|
||||
dp.call("messages.send", content=reply, receiver_uid=other["uid"])
|
||||
print(f"Replied to @{other.get('username')}: {last['message']['content']!r}")
|
||||
except xmlrpc.client.Fault as fault:
|
||||
print("Poll error:", fault.faultCode, fault.faultString)
|
||||
except OSError as error:
|
||||
print("Network error, retrying:", error)
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
@@ -0,0 +1,38 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import xmlrpc.client
|
||||
from typing import Any
|
||||
from urllib.parse import quote, urlsplit, urlunsplit
|
||||
|
||||
|
||||
class DevPlace:
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
api_key: str | None = None,
|
||||
username: str | None = None,
|
||||
password: str | None = None,
|
||||
) -> None:
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_key = api_key
|
||||
endpoint = self.base_url + "/xmlrpc"
|
||||
if username and password:
|
||||
parts = urlsplit(self.base_url)
|
||||
credentials = f"{quote(username, safe='')}:{quote(password, safe='')}"
|
||||
endpoint = urlunsplit(
|
||||
(parts.scheme, f"{credentials}@{parts.netloc}", parts.path + "/xmlrpc", "", "")
|
||||
)
|
||||
self.proxy = xmlrpc.client.ServerProxy(endpoint, allow_none=True)
|
||||
|
||||
def call(self, method: str, **params: Any) -> Any:
|
||||
if self.api_key and "api_key" not in params:
|
||||
params["api_key"] = self.api_key
|
||||
return getattr(self.proxy, method)(params)
|
||||
|
||||
def methods(self) -> list[str]:
|
||||
return sorted(self.proxy.system.listMethods())
|
||||
|
||||
def help(self, method: str) -> str:
|
||||
return self.proxy.system.methodHelp(method)
|
||||
@@ -0,0 +1,60 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import xmlrpc.client
|
||||
from pathlib import Path
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
SOURCE = Path(os.environ.get("DEVPLACE_SOURCE") or (sys.argv[1] if len(sys.argv) > 1 else "."))
|
||||
TITLE = os.environ.get("DEVPLACE_PROJECT_TITLE") or (sys.argv[2] if len(sys.argv) > 2 else SOURCE.resolve().name)
|
||||
|
||||
SKIP_DIRS = {".git", "__pycache__", "node_modules", ".venv", "venv"}
|
||||
MAX_BYTES = 512 * 1024
|
||||
|
||||
|
||||
def text_files(root: Path):
|
||||
for path in sorted(root.rglob("*")):
|
||||
if not path.is_file():
|
||||
continue
|
||||
if any(part in SKIP_DIRS for part in path.relative_to(root).parts):
|
||||
continue
|
||||
if path.stat().st_size > MAX_BYTES:
|
||||
print("skip (too big):", path)
|
||||
continue
|
||||
try:
|
||||
content = path.read_text(encoding="utf-8")
|
||||
except UnicodeDecodeError:
|
||||
print("skip (binary):", path)
|
||||
continue
|
||||
yield path.relative_to(root).as_posix(), content
|
||||
|
||||
|
||||
def main() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
created = dp.call(
|
||||
"projects.create",
|
||||
title=TITLE,
|
||||
description=f"Deployed from {SOURCE} via the XML-RPC bridge.",
|
||||
project_type="software",
|
||||
)
|
||||
slug = created["data"]["slug"]
|
||||
print("created project:", slug)
|
||||
count = 0
|
||||
for rel_path, content in text_files(SOURCE):
|
||||
try:
|
||||
dp.call("project.files.write", project_slug=slug, path=rel_path, content=content)
|
||||
count += 1
|
||||
print("wrote", rel_path)
|
||||
except xmlrpc.client.Fault as fault:
|
||||
print("failed", rel_path, "-", fault.faultCode, fault.faultString)
|
||||
print(f"deployed {count} files to {URL}/projects/{slug}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,48 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
POST_IT = os.environ.get("DEVPLACE_DIGEST_POST") == "1"
|
||||
|
||||
|
||||
def build_digest(dp: DevPlace) -> str:
|
||||
lines = ["# DevPlace digest", ""]
|
||||
|
||||
posts = dp.call("feed.list").get("posts", [])[:5]
|
||||
lines.append("## Latest posts")
|
||||
for item in posts:
|
||||
post = item["post"]
|
||||
lines.append(f"- {post.get('title') or post['uid']} by @{item['author']['username']}")
|
||||
lines.append("")
|
||||
|
||||
leaders = dp.call("leaderboard").get("entries", [])[:5]
|
||||
lines.append("## Top members")
|
||||
for entry in leaders:
|
||||
lines.append(f"- #{entry.get('rank')} @{entry.get('username')} ({entry.get('stars')} stars)")
|
||||
lines.append("")
|
||||
|
||||
articles = dp.call("news.list").get("articles", [])[:5]
|
||||
lines.append("## Developer news")
|
||||
for article in articles:
|
||||
lines.append(f"- {article.get('title')} ({article.get('source_name')})")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
digest = build_digest(dp)
|
||||
print(digest)
|
||||
if POST_IT:
|
||||
result = dp.call("posts.create", content=digest, title="Daily digest", topic="devlog")
|
||||
print("\nposted:", result.get("redirect"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,67 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
POLL_SECONDS = int(os.environ.get("DEVPLACE_POLL_SECONDS", "3"))
|
||||
PEER = os.environ.get("DEVPLACE_PEER") or (sys.argv[1] if len(sys.argv) > 1 else "")
|
||||
|
||||
|
||||
def resolve_uid(dp: DevPlace, username: str) -> str:
|
||||
detail = dp.call("profile.detail", username=username)
|
||||
return detail["profile_user"]["uid"]
|
||||
|
||||
|
||||
def receive_loop(dp: DevPlace, peer_uid: str, stop: threading.Event) -> None:
|
||||
seen: set[str] = set()
|
||||
primed = False
|
||||
while not stop.is_set():
|
||||
try:
|
||||
thread = dp.call("messages.inbox", with_uid=peer_uid)
|
||||
for item in thread.get("messages", []):
|
||||
message_uid = item["message"]["uid"]
|
||||
if message_uid in seen:
|
||||
continue
|
||||
seen.add(message_uid)
|
||||
if primed and not item.get("is_mine"):
|
||||
sender = item.get("sender") or {}
|
||||
sys.stdout.write(
|
||||
f"\n{sender.get('username', 'them')}> {item['message']['content']}\nyou> "
|
||||
)
|
||||
sys.stdout.flush()
|
||||
primed = True
|
||||
except Exception as error:
|
||||
sys.stderr.write(f"\n[receive error: {error}]\n")
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
def run() -> None:
|
||||
if not PEER:
|
||||
print("Usage: DEVPLACE_PEER=<username> python dm_client.py (or pass it as an argument)")
|
||||
return
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
peer_uid = resolve_uid(dp, PEER)
|
||||
stop = threading.Event()
|
||||
worker = threading.Thread(target=receive_loop, args=(dp, peer_uid, stop), daemon=True)
|
||||
worker.start()
|
||||
print(f"Chatting with @{PEER}. Type a message and press enter; Ctrl-C to quit.")
|
||||
try:
|
||||
while True:
|
||||
text = input("you> ")
|
||||
if text.strip():
|
||||
dp.call("messages.send", content=text, receiver_uid=peer_uid)
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
stop.set()
|
||||
print("\nbye")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
@@ -0,0 +1,39 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
USERNAME = os.environ.get("DEVPLACE_USERNAME") or (sys.argv[1] if len(sys.argv) > 1 else "")
|
||||
OUT_FILE = Path(os.environ.get("DEVPLACE_EXPORT_FILE", "devplace_export.json"))
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if not USERNAME:
|
||||
print("Usage: DEVPLACE_USERNAME=<you> python export_data.py")
|
||||
return
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
export = {
|
||||
"profile": dp.call("profile.detail", username=USERNAME),
|
||||
"posts": dp.call("profile.detail", username=USERNAME, tab="posts").get("posts", []),
|
||||
"gists": dp.call("profile.detail", username=USERNAME, tab="gists").get("gists", []),
|
||||
"projects": dp.call("profile.detail", username=USERNAME, tab="projects").get("projects", []),
|
||||
"bookmarks": dp.call("bookmarks.saved").get("items", []),
|
||||
"notifications": dp.call("notifications.list").get("notification_groups", []),
|
||||
}
|
||||
OUT_FILE.write_text(json.dumps(export, indent=2), encoding="utf-8")
|
||||
print(
|
||||
f"exported profile, {len(export['posts'])} posts, {len(export['gists'])} gists, "
|
||||
f"{len(export['projects'])} projects to {OUT_FILE}"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,57 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
import xmlrpc.client
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
POLL_SECONDS = int(os.environ.get("DEVPLACE_POLL_SECONDS", "20"))
|
||||
UPVOTE_KEYWORDS = [k.strip().lower() for k in os.environ.get("DEVPLACE_UPVOTE_KEYWORDS", "").split(",") if k.strip()]
|
||||
|
||||
|
||||
def matches(post: dict) -> bool:
|
||||
if not UPVOTE_KEYWORDS:
|
||||
return False
|
||||
haystack = f"{post.get('title') or ''} {post.get('content') or ''}".lower()
|
||||
return any(keyword in haystack for keyword in UPVOTE_KEYWORDS)
|
||||
|
||||
|
||||
def run() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
seen: set[str] = set()
|
||||
primed = False
|
||||
print(f"Watching the feed on {URL}; Ctrl-C to quit.")
|
||||
if UPVOTE_KEYWORDS:
|
||||
print("Auto-upvoting posts matching:", ", ".join(UPVOTE_KEYWORDS))
|
||||
while True:
|
||||
try:
|
||||
for item in dp.call("feed.list").get("posts", []):
|
||||
post = item["post"]
|
||||
if post["uid"] in seen:
|
||||
continue
|
||||
seen.add(post["uid"])
|
||||
if not primed:
|
||||
continue
|
||||
author = item["author"]["username"]
|
||||
print(f"[{item.get('time_ago')}] @{author}: {post.get('title') or post['uid']}")
|
||||
if API_KEY != "YOUR_API_KEY" and matches(post):
|
||||
try:
|
||||
dp.call("votes.cast", target_type="post", target_uid=post["uid"], value="1")
|
||||
print(" upvoted (keyword match)")
|
||||
except xmlrpc.client.Fault as fault:
|
||||
print(" upvote failed:", fault.faultCode)
|
||||
primed = True
|
||||
except xmlrpc.client.Fault as fault:
|
||||
print("error:", fault.faultCode, fault.faultString)
|
||||
except OSError as error:
|
||||
print("network error:", error)
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
@@ -0,0 +1,58 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
OUT_DIR = Path(os.environ.get("DEVPLACE_GIST_DIR", "gists_backup"))
|
||||
|
||||
EXTENSIONS = {
|
||||
"python": "py",
|
||||
"javascript": "js",
|
||||
"typescript": "ts",
|
||||
"html": "html",
|
||||
"css": "css",
|
||||
"c": "c",
|
||||
"cpp": "cpp",
|
||||
"java": "java",
|
||||
"go": "go",
|
||||
"rust": "rs",
|
||||
"sql": "sql",
|
||||
"bash": "sh",
|
||||
"json": "json",
|
||||
"markdown": "md",
|
||||
"plaintext": "txt",
|
||||
}
|
||||
|
||||
|
||||
def all_gists(dp: DevPlace):
|
||||
cursor = None
|
||||
while True:
|
||||
page = dp.call("gists.list", before=cursor) if cursor else dp.call("gists.list")
|
||||
for item in page.get("gists", []):
|
||||
yield item["gist"]
|
||||
cursor = page.get("next_cursor")
|
||||
if not cursor:
|
||||
return
|
||||
|
||||
|
||||
def main() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
saved = 0
|
||||
for gist in all_gists(dp):
|
||||
extension = EXTENSIONS.get(gist.get("language") or "plaintext", "txt")
|
||||
path = OUT_DIR / f"{gist['slug']}.{extension}"
|
||||
path.write_text(gist.get("source_code") or "", encoding="utf-8")
|
||||
saved += 1
|
||||
print("saved", path)
|
||||
print(f"backed up {saved} gists into {OUT_DIR}/")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,23 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
dp = DevPlace(URL)
|
||||
names = dp.methods()
|
||||
print(f"{len(names)} methods available on {URL}/xmlrpc\n")
|
||||
for name in names:
|
||||
print("###", name)
|
||||
print(dp.help(name))
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,68 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import xmlrpc.client
|
||||
from typing import Iterator
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
POLL_SECONDS = int(os.environ.get("DEVPLACE_POLL_SECONDS", "30"))
|
||||
REPLY = os.environ.get(
|
||||
"DEVPLACE_REPLY",
|
||||
"Thanks for the mention! This is an automated reply via the XML-RPC bridge.",
|
||||
)
|
||||
|
||||
POST_SLUG = re.compile(r"/posts/([^/#?]+)")
|
||||
|
||||
|
||||
def unread_mentions(dp: DevPlace) -> Iterator[dict]:
|
||||
result = dp.call("notifications.list")
|
||||
for group in result.get("notification_groups", []):
|
||||
for entry in group.get("entries", []):
|
||||
note = entry["notification"]
|
||||
if note.get("type") == "mention" and not note.get("read"):
|
||||
yield note
|
||||
|
||||
|
||||
def reply_to_mention(dp: DevPlace, note: dict) -> bool:
|
||||
match = POST_SLUG.search(note.get("target_url") or "")
|
||||
if not match:
|
||||
return False
|
||||
detail = dp.call("posts.detail", post_slug=match.group(1))
|
||||
dp.call(
|
||||
"comments.create",
|
||||
content=REPLY,
|
||||
target_uid=detail["post"]["uid"],
|
||||
target_type="post",
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def run() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
print(f"Mention bot started; polling {URL} every {POLL_SECONDS}s.")
|
||||
while True:
|
||||
try:
|
||||
for note in unread_mentions(dp):
|
||||
try:
|
||||
if reply_to_mention(dp, note):
|
||||
print("Replied to:", note.get("message"))
|
||||
except xmlrpc.client.Fault as fault:
|
||||
print("Could not reply:", fault.faultCode, fault.faultString)
|
||||
finally:
|
||||
dp.call("notifications.mark.read", notification_uid=note["uid"])
|
||||
except xmlrpc.client.Fault as fault:
|
||||
print("Poll error:", fault.faultCode, fault.faultString)
|
||||
except OSError as error:
|
||||
print("Network error, retrying:", error)
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
@@ -0,0 +1,64 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Iterator
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
|
||||
|
||||
def all_feed_posts(dp: DevPlace) -> Iterator[dict]:
|
||||
cursor = None
|
||||
while True:
|
||||
page = dp.call("feed.list", before=cursor) if cursor else dp.call("feed.list")
|
||||
for item in page["posts"]:
|
||||
yield item
|
||||
cursor = page.get("next_cursor")
|
||||
if not cursor:
|
||||
return
|
||||
|
||||
|
||||
def create_post(dp: DevPlace, content: str, title: str = "") -> str:
|
||||
result = dp.call("posts.create", content=content, title=title)
|
||||
return result["data"]["slug"]
|
||||
|
||||
|
||||
def reply_to_post(dp: DevPlace, post_slug: str, content: str) -> None:
|
||||
detail = dp.call("posts.detail", post_slug=post_slug)
|
||||
dp.call("comments.create", content=content, target_uid=detail["post"]["uid"])
|
||||
|
||||
|
||||
def upvote_post(dp: DevPlace, post_uid: str) -> None:
|
||||
dp.call("votes.cast", target_type="post", target_uid=post_uid, value="1")
|
||||
|
||||
|
||||
def react_to_post(dp: DevPlace, post_uid: str, emoji: str = "rocket") -> None:
|
||||
dp.call("reactions.toggle", target_type="post", target_uid=post_uid, emoji=emoji)
|
||||
|
||||
|
||||
def follow(dp: DevPlace, username: str) -> None:
|
||||
dp.call("follow.user", username=username)
|
||||
|
||||
|
||||
def send_message(dp: DevPlace, receiver_uid: str, content: str) -> None:
|
||||
dp.call("messages.send", content=content, receiver_uid=receiver_uid)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
slug = create_post(dp, "Posted from the XML-RPC recipes example.", title="Hello")
|
||||
print("created post:", slug)
|
||||
reply_to_post(dp, slug, "And here is an automated reply.")
|
||||
print("listing the first few feed posts:")
|
||||
for index, item in enumerate(all_feed_posts(dp)):
|
||||
print("-", item["author"]["username"], ":", item["post"].get("title") or item["post"]["uid"])
|
||||
if index >= 4:
|
||||
break
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,70 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
import urllib.request
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
|
||||
from client import DevPlace
|
||||
|
||||
URL = os.environ.get("DEVPLACE_URL", "https://devplace.net")
|
||||
API_KEY = os.environ.get("DEVPLACE_API_KEY", "YOUR_API_KEY")
|
||||
FEED_URL = os.environ.get("DEVPLACE_RSS_URL", "https://news.ycombinator.com/rss")
|
||||
TOPIC = os.environ.get("DEVPLACE_RSS_TOPIC", "devlog")
|
||||
POLL_SECONDS = int(os.environ.get("DEVPLACE_POLL_SECONDS", "900"))
|
||||
SEEN_FILE = Path(os.environ.get("DEVPLACE_RSS_SEEN", "rss_seen.txt"))
|
||||
|
||||
ATOM = "{http://www.w3.org/2005/Atom}"
|
||||
|
||||
|
||||
def fetch_items():
|
||||
with urllib.request.urlopen(FEED_URL, timeout=20) as response:
|
||||
root = ET.fromstring(response.read())
|
||||
items = []
|
||||
for node in root.iter():
|
||||
tag = node.tag.split("}")[-1]
|
||||
if tag in ("item", "entry"):
|
||||
title = node.findtext("title") or node.findtext(f"{ATOM}title") or ""
|
||||
link = node.findtext("link") or ""
|
||||
if not link:
|
||||
link_node = node.find(f"{ATOM}link")
|
||||
link = link_node.get("href") if link_node is not None else ""
|
||||
if title and link:
|
||||
items.append((title.strip(), link.strip()))
|
||||
return items
|
||||
|
||||
|
||||
def load_seen() -> set[str]:
|
||||
if SEEN_FILE.exists():
|
||||
return set(SEEN_FILE.read_text(encoding="utf-8").splitlines())
|
||||
return set()
|
||||
|
||||
|
||||
def remember(link: str) -> None:
|
||||
with SEEN_FILE.open("a", encoding="utf-8") as handle:
|
||||
handle.write(link + "\n")
|
||||
|
||||
|
||||
def run() -> None:
|
||||
dp = DevPlace(URL, api_key=API_KEY)
|
||||
print(f"RSS poster started for {FEED_URL}; polling every {POLL_SECONDS}s.")
|
||||
while True:
|
||||
seen = load_seen()
|
||||
try:
|
||||
for title, link in fetch_items():
|
||||
if link in seen:
|
||||
continue
|
||||
content = f"{title}\n\n{link}"
|
||||
dp.call("posts.create", content=content, title=title[:120], topic=TOPIC)
|
||||
remember(link)
|
||||
print("posted:", title)
|
||||
except Exception as error:
|
||||
print("error:", error)
|
||||
time.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
Reference in New Issue
Block a user