| """A small client for a self-hosted Penpot's RPC API (JSON over /api/rpc/command/<name>). |
| |
| Signs in as a test profile it registers on first use (credentials live in a local file, never in the repo), so it |
| is only for a Penpot you run yourself: local Docker or a sandbox. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import http.cookiejar |
| import json |
| import mimetypes |
| import os |
| import secrets |
| import urllib.error |
| import urllib.request |
| import uuid |
| from pathlib import Path |
|
|
|
|
| class PenpotError(RuntimeError): |
| pass |
|
|
|
|
| def sniff_mime(data: bytes) -> str | None: |
| if data[:8] == b"\x89PNG\r\n\x1a\n": |
| return "image/png" |
| if data[:3] == b"\xff\xd8\xff": |
| return "image/jpeg" |
| if data[:4] == b"RIFF" and data[8:12] == b"WEBP": |
| return "image/webp" |
| if data.lstrip()[:5] in (b"<?xml", b"<svg "): |
| return "image/svg+xml" |
| return None |
|
|
|
|
| class PenpotClient: |
| def __init__(self, base_url: str, cred_path: str | Path, headers: dict[str, str] | None = None): |
| self.base = base_url.rstrip("/") |
| self.headers = dict(headers or {}) |
| self.cred_path = Path(cred_path) |
| self.jar = http.cookiejar.CookieJar() |
| self._op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar)) |
| self.profile: dict = {} |
|
|
| |
|
|
| def rpc(self, cmd: str, **params): |
| req = urllib.request.Request(f"{self.base}/api/rpc/command/{cmd}", data=json.dumps(params).encode(), |
| method="POST", headers={"Content-Type": "application/json", |
| "Accept": "application/json", **self.headers}) |
| return self._send(req, cmd) |
|
|
| def _send(self, req, cmd, wait_s: float = 180): |
| """One RPC; a 502/503/504 or refused connection (Penpot still booting or restarting) is retried.""" |
| import time |
|
|
| t0 = time.time() |
| while True: |
| try: |
| with self._op.open(req, timeout=120) as r: |
| body = r.read() |
| return json.loads(body) if body else None |
| except urllib.error.HTTPError as e: |
| if e.code in (502, 503, 504) and time.time() - t0 < wait_s: |
| time.sleep(3) |
| continue |
| raise PenpotError(f"{cmd}: HTTP {e.code}: {e.read()[:600].decode(errors='replace')}") from None |
| except (urllib.error.URLError, ConnectionError) as e: |
| if time.time() - t0 < wait_s: |
| time.sleep(3) |
| continue |
| raise PenpotError(f"{cmd}: {e}") from None |
|
|
| def upload_media(self, file_id: str, name: str, data: bytes, mime: str | None = None) -> dict: |
| """Add an image to a file's local library (the Assets panel); returns the media object. Penpot checks the |
| bytes against the extension, so the type is sniffed from the content and `name` gets the matching one.""" |
| mime = mime or sniff_mime(data) or mimetypes.guess_type(name)[0] or "image/png" |
| name = f"{Path(name).stem}{mimetypes.guess_extension(mime) or ''}".replace(".jpe", ".jpg") |
| boundary = secrets.token_hex(12) |
| parts = [] |
| for k, v in (("file-id", file_id), ("is-local", "true"), ("name", name)): |
| parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()) |
| parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="content"; filename="{name}"\r\n' |
| f"Content-Type: {mime}\r\n\r\n".encode() + data + b"\r\n") |
| parts.append(f"--{boundary}--\r\n".encode()) |
| req = urllib.request.Request(f"{self.base}/api/rpc/command/upload-file-media-object", data=b"".join(parts), |
| method="POST", headers={"Content-Type": f"multipart/form-data; boundary={boundary}", |
| "Accept": "application/json", **self.headers}) |
| return self._send(req, "upload-file-media-object") |
|
|
| |
|
|
| def login(self) -> dict: |
| """Sign in as the test profile, registering it the first time.""" |
| if not self.cred_path.exists(): |
| self.cred_path.parent.mkdir(parents=True, exist_ok=True) |
| self.cred_path.write_text(json.dumps({"email": f"agent-{secrets.token_hex(4)}@design-canvas.test", |
| |
| "password": secrets.token_urlsafe(18) + "aZ7!", |
| "fullname": "design-canvas agent"})) |
| os.chmod(self.cred_path, 0o600) |
| cred = json.loads(self.cred_path.read_text()) |
| try: |
| self.rpc("login-with-password", email=cred["email"], password=cred["password"]) |
| except PenpotError: |
| tok = self.rpc("prepare-register-profile", fullname=cred["fullname"], email=cred["email"], |
| password=cred["password"]) |
| self.rpc("register-profile", token=tok["token"], fullname=cred["fullname"], |
| acceptTermsAndPrivacy=True, acceptNewsletterSubscription=False) |
| self.rpc("login-with-password", email=cred["email"], password=cred["password"]) |
| self.profile = self.rpc("get-profile") |
| return self.profile |
|
|
| def cookies(self) -> list[dict]: |
| """The session cookies, in Playwright's add_cookies format.""" |
| return [{"name": c.name, "value": c.value, "url": self.base, "httpOnly": True, "sameSite": "Lax"} |
| for c in self.jar] |
|
|
| |
|
|
| def create_file(self, name: str, project_id: str | None = None) -> dict: |
| return self.rpc("create-file", name=name, projectId=project_id or self.profile["defaultProjectId"]) |
|
|
| def get_file(self, file_id: str) -> dict: |
| return self.rpc("get-file", id=file_id) |
|
|
| def update_file(self, file_id: str, changes: list[dict], revn: int, vern: int = 0) -> dict: |
| return self.rpc("update-file", id=file_id, sessionId=str(uuid.uuid4()), revn=revn, vern=vern, |
| changes=changes) |
|
|
