417 lines
17 KiB
Python
417 lines
17 KiB
Python
"""Joplin Data API client for importing notes into Joplin desktop."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# HTTP timeout for regular API calls (seconds). Notes can be large Markdown
|
|
# files so we allow more time than a typical JSON API call.
|
|
# Override with JOPLIN_REQUEST_TIMEOUT env var if you have very large conversations.
|
|
_REQUEST_TIMEOUT: int = int(os.getenv("JOPLIN_REQUEST_TIMEOUT", "30"))
|
|
|
|
|
|
class JoplinError(Exception):
|
|
"""Raised when the Joplin API returns an error or is unreachable."""
|
|
|
|
|
|
class JoplinClient:
|
|
"""HTTP client for the Joplin local REST API (Web Clipper service).
|
|
|
|
Requires Joplin desktop to be running with the Web Clipper service enabled.
|
|
Get your API token from: Joplin → Tools → Options → Web Clipper.
|
|
|
|
Args:
|
|
base_url: Joplin API base URL (default: http://localhost:41184).
|
|
token: API authorization token from Joplin Web Clipper settings.
|
|
"""
|
|
|
|
def __init__(self, base_url: str, token: str) -> None:
|
|
self._base_url = base_url.rstrip("/")
|
|
self._token = token
|
|
# In-memory cache: (parent_id | None, title) → folder ID
|
|
self._notebook_cache: dict[tuple[str | None, str], str] = {}
|
|
self._notebooks_loaded = False
|
|
logger.debug("[joplin] Client initialised with base_url=%s", self._base_url)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Connectivity
|
|
# ------------------------------------------------------------------
|
|
|
|
def ping(self) -> bool:
|
|
"""Return True if the Joplin API is reachable and responding.
|
|
|
|
Note: /ping does not require authentication. A successful ping only
|
|
confirms Joplin is running — not that the token is valid. Call
|
|
``validate_token()`` to confirm authentication separately.
|
|
|
|
Raises:
|
|
JoplinError: If the API returns an unexpected non-connection error.
|
|
"""
|
|
url = f"{self._base_url}/ping"
|
|
logger.debug("[joplin] GET %s", url)
|
|
try:
|
|
resp = requests.get(url, timeout=5)
|
|
resp.raise_for_status()
|
|
ok = "JoplinClipperServer" in resp.text
|
|
logger.debug("[joplin] ping → %s (body: %r)", "OK" if ok else "unexpected response", resp.text[:80])
|
|
return ok
|
|
except requests.exceptions.ConnectionError:
|
|
logger.debug("[joplin] ping → connection refused at %s", url)
|
|
return False
|
|
except requests.exceptions.Timeout:
|
|
logger.debug("[joplin] ping → timed out after 5s at %s", url)
|
|
return False
|
|
except requests.exceptions.RequestException as e:
|
|
raise JoplinError(f"Joplin ping failed: {e}") from e
|
|
|
|
def validate_token(self) -> None:
|
|
"""Verify the API token is accepted by Joplin.
|
|
|
|
Does a minimal authenticated call (GET /folders?limit=1) and raises
|
|
``JoplinError`` if authentication fails.
|
|
|
|
Raises:
|
|
JoplinError: If the token is rejected (401) or Joplin is unreachable.
|
|
"""
|
|
logger.debug("[joplin] Validating API token…")
|
|
self._get("/folders", params={"limit": 1, "fields": "id"})
|
|
logger.debug("[joplin] Token validated OK")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Notebooks (folders)
|
|
# ------------------------------------------------------------------
|
|
|
|
def list_notebooks(self) -> list[dict]:
|
|
"""Return all Joplin notebooks (folders), handling pagination.
|
|
|
|
Returns:
|
|
List of folder dicts with at least ``id``, ``title``, and ``parent_id`` keys.
|
|
"""
|
|
results: list[dict] = []
|
|
page = 1
|
|
while True:
|
|
logger.debug("[joplin] GET /folders page=%d", page)
|
|
resp = self._get("/folders", params={"page": page, "fields": "id,title,parent_id"})
|
|
items = resp.get("items", [])
|
|
results.extend(items)
|
|
logger.debug("[joplin] /folders page=%d → %d items, has_more=%s", page, len(items), resp.get("has_more"))
|
|
if not resp.get("has_more"):
|
|
break
|
|
page += 1
|
|
return results
|
|
|
|
def get_or_create_notebook(self, title: str, parent_id: str | None = None) -> str:
|
|
"""Return the Joplin folder ID for ``title`` under ``parent_id``, creating if needed.
|
|
|
|
Args:
|
|
title: Notebook display name.
|
|
parent_id: ID of the parent folder, or None for a root notebook.
|
|
|
|
Returns:
|
|
Joplin folder ID string.
|
|
"""
|
|
if not self._notebooks_loaded:
|
|
self._load_notebook_cache()
|
|
|
|
key = (parent_id, title)
|
|
if key in self._notebook_cache:
|
|
folder_id = self._notebook_cache[key]
|
|
logger.debug("[joplin] Notebook cache hit: %r (parent=%s) → %s", title, parent_id, folder_id)
|
|
return folder_id
|
|
|
|
# Not found — create it
|
|
logger.info("[joplin] Creating notebook: %r (parent=%s)", title, parent_id)
|
|
data: dict = {"title": title}
|
|
if parent_id:
|
|
data["parent_id"] = parent_id
|
|
resp = self._post("/folders", data)
|
|
folder_id = resp["id"]
|
|
self._notebook_cache[key] = folder_id
|
|
logger.debug("[joplin] Notebook created: %r → %s", title, folder_id)
|
|
return folder_id
|
|
|
|
def get_or_create_notebook_path(self, path: list[str]) -> str:
|
|
"""Ensure a nested notebook path exists and return the leaf folder ID.
|
|
|
|
Creates intermediate notebooks as needed.
|
|
|
|
Args:
|
|
path: Ordered list of notebook names, e.g. ["AI-ChatGPT", "No Project"].
|
|
|
|
Returns:
|
|
Joplin folder ID of the deepest (leaf) notebook.
|
|
"""
|
|
parent_id: str | None = None
|
|
for name in path:
|
|
parent_id = self.get_or_create_notebook(name, parent_id)
|
|
assert parent_id is not None
|
|
return parent_id
|
|
|
|
# ------------------------------------------------------------------
|
|
# Notes
|
|
# ------------------------------------------------------------------
|
|
|
|
def create_note(self, title: str, body: str, parent_id: str) -> str:
|
|
"""Create a new note in the specified notebook.
|
|
|
|
Args:
|
|
title: Note title.
|
|
body: Note body (Markdown).
|
|
parent_id: Notebook (folder) ID.
|
|
|
|
Returns:
|
|
ID of the created note.
|
|
"""
|
|
logger.debug(
|
|
"[joplin] Creating note: %r in notebook %s (%d chars)",
|
|
title, parent_id, len(body),
|
|
)
|
|
resp = self._post("/notes", {"title": title, "body": body, "parent_id": parent_id})
|
|
note_id = resp["id"]
|
|
logger.info("[joplin] Note created: %r → %s", title, note_id)
|
|
return note_id
|
|
|
|
def update_note(self, note_id: str, title: str, body: str) -> None:
|
|
"""Update the title and body of an existing note.
|
|
|
|
Args:
|
|
note_id: Joplin note ID.
|
|
title: New note title.
|
|
body: New note body (Markdown).
|
|
"""
|
|
logger.debug(
|
|
"[joplin] Updating note %s: %r (%d chars)",
|
|
note_id, title, len(body),
|
|
)
|
|
self._put(f"/notes/{note_id}", {"title": title, "body": body})
|
|
logger.info("[joplin] Note updated: %r (%s)", title, note_id)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Resources (embedded media)
|
|
# ------------------------------------------------------------------
|
|
|
|
def create_resource(self, file_path: "Path", title: str | None = None) -> str:
|
|
"""Upload a local file as a Joplin resource and return its ID.
|
|
|
|
Joplin's ``POST /resources`` is multipart: a ``data`` file part plus a
|
|
``props`` JSON part. The returned ID is referenced from note bodies as
|
|
``:/<id>``.
|
|
"""
|
|
url = f"{self._base_url}/resources"
|
|
props = json.dumps({"title": title or file_path.name})
|
|
logger.debug("[joplin] POST /resources (%s)", file_path.name)
|
|
try:
|
|
with file_path.open("rb") as fh:
|
|
resp = requests.post(
|
|
url,
|
|
params={"token": self._token},
|
|
files={
|
|
"data": (file_path.name, fh),
|
|
"props": (None, props),
|
|
},
|
|
timeout=_REQUEST_TIMEOUT,
|
|
)
|
|
resp.raise_for_status()
|
|
resource_id = resp.json()["id"]
|
|
logger.info("[joplin] Resource created: %s → %s", file_path.name, resource_id)
|
|
return resource_id
|
|
except requests.exceptions.ConnectionError as e:
|
|
raise JoplinError(
|
|
"Cannot connect to Joplin. Is Joplin desktop running with Web Clipper enabled?"
|
|
) from e
|
|
except requests.exceptions.Timeout as e:
|
|
raise JoplinError(_timeout_message("POST", "/resources")) from e
|
|
except requests.exceptions.HTTPError as e:
|
|
raise JoplinError(_http_error_message("POST", "/resources", e)) from e
|
|
except (requests.exceptions.RequestException, KeyError, OSError) as e:
|
|
raise JoplinError(f"Joplin resource upload failed for {file_path.name}: {e}") from e
|
|
|
|
# ------------------------------------------------------------------
|
|
# HTTP helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _get(self, path: str, params: dict | None = None) -> dict[str, Any]:
|
|
url = f"{self._base_url}{path}"
|
|
query = {"token": self._token, **(params or {})}
|
|
logger.debug("[joplin] GET %s params=%s", path, {k: v for k, v in (params or {}).items()})
|
|
try:
|
|
resp = requests.get(url, params=query, timeout=_REQUEST_TIMEOUT)
|
|
logger.debug("[joplin] GET %s → HTTP %d", path, resp.status_code)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.ConnectionError as e:
|
|
raise JoplinError(
|
|
"Cannot connect to Joplin. Is Joplin desktop running with Web Clipper enabled?"
|
|
) from e
|
|
except requests.exceptions.Timeout as e:
|
|
raise JoplinError(_timeout_message("GET", path)) from e
|
|
except requests.exceptions.HTTPError as e:
|
|
raise JoplinError(_http_error_message("GET", path, e)) from e
|
|
except requests.exceptions.RequestException as e:
|
|
raise JoplinError(f"Joplin GET {path} failed: {e}") from e
|
|
|
|
def _post(self, path: str, data: dict) -> dict[str, Any]:
|
|
url = f"{self._base_url}{path}"
|
|
logger.debug("[joplin] POST %s", path)
|
|
try:
|
|
resp = requests.post(url, params={"token": self._token}, json=data, timeout=_REQUEST_TIMEOUT)
|
|
logger.debug("[joplin] POST %s → HTTP %d", path, resp.status_code)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.ConnectionError as e:
|
|
raise JoplinError(
|
|
"Cannot connect to Joplin. Is Joplin desktop running with Web Clipper enabled?"
|
|
) from e
|
|
except requests.exceptions.Timeout as e:
|
|
raise JoplinError(_timeout_message("POST", path)) from e
|
|
except requests.exceptions.HTTPError as e:
|
|
raise JoplinError(_http_error_message("POST", path, e)) from e
|
|
except requests.exceptions.RequestException as e:
|
|
raise JoplinError(f"Joplin POST {path} failed: {e}") from e
|
|
|
|
def _put(self, path: str, data: dict) -> dict[str, Any]:
|
|
url = f"{self._base_url}{path}"
|
|
logger.debug("[joplin] PUT %s", path)
|
|
try:
|
|
resp = requests.put(url, params={"token": self._token}, json=data, timeout=_REQUEST_TIMEOUT)
|
|
logger.debug("[joplin] PUT %s → HTTP %d", path, resp.status_code)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except requests.exceptions.ConnectionError as e:
|
|
raise JoplinError(
|
|
"Cannot connect to Joplin. Is Joplin desktop running with Web Clipper enabled?"
|
|
) from e
|
|
except requests.exceptions.Timeout as e:
|
|
raise JoplinError(_timeout_message("PUT", path)) from e
|
|
except requests.exceptions.HTTPError as e:
|
|
raise JoplinError(_http_error_message("PUT", path, e)) from e
|
|
except requests.exceptions.RequestException as e:
|
|
raise JoplinError(f"Joplin PUT {path} failed: {e}") from e
|
|
|
|
def _load_notebook_cache(self) -> None:
|
|
logger.debug("[joplin] Loading notebook list from Joplin…")
|
|
notebooks = self.list_notebooks()
|
|
self._notebook_cache = {
|
|
(nb.get("parent_id") or None, nb["title"]): nb["id"]
|
|
for nb in notebooks
|
|
}
|
|
self._notebooks_loaded = True
|
|
logger.debug("[joplin] Notebook cache loaded: %d notebooks", len(self._notebook_cache))
|
|
for (parent_id, title), folder_id in self._notebook_cache.items():
|
|
logger.debug("[joplin] (%s) %r → %s", parent_id or "root", title, folder_id)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Error message helper
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _timeout_message(method: str, path: str) -> str:
|
|
"""Build a clear timeout error message with actionable suggestions."""
|
|
return (
|
|
f"Joplin {method} {path} timed out after {_REQUEST_TIMEOUT}s. "
|
|
"Possible causes:\n"
|
|
" • The note body is very large and Joplin is slow to process it.\n"
|
|
" • Joplin is busy (syncing, indexing, or loading a large library).\n"
|
|
" • Joplin has frozen — try restarting it.\n"
|
|
f"If this happens repeatedly, increase JOPLIN_REQUEST_TIMEOUT in your .env "
|
|
f"(currently {_REQUEST_TIMEOUT}s)."
|
|
)
|
|
|
|
|
|
def _http_error_message(method: str, path: str, e: requests.exceptions.HTTPError) -> str:
|
|
"""Build a human-friendly error message from an HTTP error, with auth hint on 401."""
|
|
resp = e.response
|
|
status = resp.status_code if resp is not None else "?"
|
|
if status == 401:
|
|
return (
|
|
f"Joplin rejected the API token (HTTP 401 on {method} {path}). "
|
|
"Check that JOPLIN_API_TOKEN is correct: "
|
|
"Joplin → Tools → Options → Web Clipper → Authorization token."
|
|
)
|
|
if status == 404:
|
|
return f"Joplin resource not found (HTTP 404 on {method} {path}). The note may have been deleted in Joplin."
|
|
body_snippet = ""
|
|
if resp is not None:
|
|
try:
|
|
body_snippet = f" — {resp.text[:120]}"
|
|
except Exception:
|
|
pass
|
|
return f"Joplin {method} {path} failed: HTTP {status}{body_snippet}"
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Embedded-media rewriting
|
|
# ------------------------------------------------------------------
|
|
|
|
# Markdown image/link targets pointing at the local media/ sibling directory,
|
|
# e.g.  or [name](media/clip.wav).
|
|
_MEDIA_LINK_RE = re.compile(r"(!?\[[^\]]*\])\((media/[^)\s]+)\)")
|
|
|
|
|
|
def upload_media_and_rewrite(
|
|
body: str,
|
|
note_dir: Path,
|
|
client: "JoplinClient",
|
|
resource_map: dict[str, str],
|
|
) -> tuple[str, dict[str, str]]:
|
|
"""Rewrite local ``media/…`` links to Joplin ``:/resourceId`` references.
|
|
|
|
Uploads each referenced file as a Joplin resource (once), rewriting both
|
|
image embeds and file links. ``resource_map`` (relative-path → resource_id)
|
|
is the idempotency record from the cache: known paths are reused, new ones
|
|
are added and returned so the caller can persist them. Missing files are
|
|
left as-is so the link still points at the on-disk copy.
|
|
"""
|
|
new_map = dict(resource_map)
|
|
|
|
def replace(match: re.Match) -> str:
|
|
label, rel_path = match.group(1), match.group(2)
|
|
resource_id = new_map.get(rel_path)
|
|
if resource_id is None:
|
|
file_path = (note_dir / rel_path).resolve()
|
|
if not file_path.is_file():
|
|
logger.warning("[joplin] Media file missing, leaving link: %s", rel_path)
|
|
return match.group(0)
|
|
resource_id = client.create_resource(file_path)
|
|
new_map[rel_path] = resource_id
|
|
return f"{label}(:/{resource_id})"
|
|
|
|
return _MEDIA_LINK_RE.sub(replace, body), new_map
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Notebook naming helper
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
_PROVIDER_DISPLAY = {
|
|
"chatgpt": "AI-ChatGPT",
|
|
"claude": "AI-Claude",
|
|
# Decision 2026-06-12: Claude Code coding projects nest under the same
|
|
# AI-Claude parent, alongside Claude web projects.
|
|
"claude-code": "AI-Claude",
|
|
}
|
|
|
|
|
|
def notebook_path(provider: str, project: str | None) -> tuple[str, str]:
|
|
"""Return (parent_notebook, child_notebook) for the given provider and project.
|
|
|
|
The parent is the top-level provider notebook; the child is the project name.
|
|
|
|
Examples:
|
|
notebook_path("chatgpt", None) → ("AI-ChatGPT", "No Project")
|
|
notebook_path("chatgpt", "no-project") → ("AI-ChatGPT", "No Project")
|
|
notebook_path("claude", "budget-tracker") → ("AI-Claude", "Budget Tracker")
|
|
"""
|
|
parent = _PROVIDER_DISPLAY.get(provider, f"AI-{provider.capitalize()}")
|
|
child = (project or "no-project").replace("-", " ").title()
|
|
return parent, child
|