373 lines
No EOL
13 KiB
Python
373 lines
No EOL
13 KiB
Python
"""
|
|
title: Readeck Integration
|
|
author: Assistant
|
|
author_url: https://github.com/assistant
|
|
funding_url: https://github.com/assistant
|
|
version: 0.1.0
|
|
license: MIT
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
from typing import Dict, List, Optional, Any
|
|
from pydantic import BaseModel, Field
|
|
import logging
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Tools:
|
|
class Valves(BaseModel):
|
|
readeck_url: str = Field(
|
|
default="http://localhost:8000",
|
|
description="Readeck instance URL (e.g., http://localhost:8000 or https://readeck.example.com)"
|
|
)
|
|
api_token: str = Field(
|
|
default="",
|
|
description="Readeck API token for authentication"
|
|
)
|
|
|
|
def __init__(self):
|
|
self.valves = self.Valves()
|
|
|
|
def _get_headers(self) -> Dict[str, str]:
|
|
"""Get headers for Readeck API requests"""
|
|
return {
|
|
"Authorization": f"Bearer {self.valves.api_token}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
}
|
|
|
|
def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict[str, Any]:
|
|
"""Make a request to the Readeck API"""
|
|
url = f"{self.valves.readeck_url.rstrip('/')}/api{endpoint}"
|
|
|
|
try:
|
|
response = requests.get(
|
|
url,
|
|
headers=self._get_headers(),
|
|
params=params or {},
|
|
timeout=30
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Error making request to {url}: {e}")
|
|
return {"error": str(e), "status_code": getattr(e.response, 'status_code', None)}
|
|
|
|
async def get_bookmarks(
|
|
self,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
collection_id: Optional[str] = None,
|
|
search: Optional[str] = None,
|
|
labels: Optional[str] = None,
|
|
__user__: Optional[dict] = None
|
|
) -> str:
|
|
"""
|
|
Get bookmarks from Readeck
|
|
|
|
Args:
|
|
limit: Number of bookmarks to return (default: 20, max: 100)
|
|
offset: Number of bookmarks to skip (default: 0)
|
|
collection_id: Filter by collection ID
|
|
search: Search query for bookmark titles/content
|
|
labels: Comma-separated list of labels to filter by
|
|
"""
|
|
|
|
if not self.valves.readeck_url or not self.valves.api_token:
|
|
return "Error: Readeck URL and API token must be configured in the tool settings."
|
|
|
|
# Build query parameters
|
|
params = {
|
|
"limit": min(limit, 100), # Cap at 100 to prevent excessive requests
|
|
"offset": offset
|
|
}
|
|
|
|
if collection_id:
|
|
params["collection_id"] = collection_id
|
|
if search:
|
|
params["q"] = search
|
|
if labels:
|
|
params["labels"] = labels
|
|
|
|
# Make request to Readeck API
|
|
result = self._make_request("/bookmarks", params)
|
|
|
|
if "error" in result:
|
|
return f"Error fetching bookmarks: {result['error']}"
|
|
|
|
# Format the response
|
|
bookmarks = result.get("bookmarks", [])
|
|
total = result.get("total", len(bookmarks))
|
|
|
|
if not bookmarks:
|
|
return "No bookmarks found matching your criteria."
|
|
|
|
# Format bookmarks for display
|
|
formatted_bookmarks = []
|
|
for bookmark in bookmarks:
|
|
formatted_bookmark = {
|
|
"id": bookmark.get("id"),
|
|
"title": bookmark.get("title", "Untitled"),
|
|
"url": bookmark.get("url"),
|
|
"excerpt": bookmark.get("excerpt", "")[:200] + "..." if bookmark.get("excerpt", "") else "",
|
|
"labels": bookmark.get("labels", []),
|
|
"collection": bookmark.get("collection", {}).get("name", ""),
|
|
"created_at": bookmark.get("created_at"),
|
|
"updated_at": bookmark.get("updated_at")
|
|
}
|
|
formatted_bookmarks.append(formatted_bookmark)
|
|
|
|
return {
|
|
"bookmarks": formatted_bookmarks,
|
|
"total": total,
|
|
"limit": params["limit"],
|
|
"offset": params["offset"]
|
|
}
|
|
|
|
async def get_collections(
|
|
self,
|
|
__user__: Optional[dict] = None
|
|
) -> str:
|
|
"""
|
|
Get collections from Readeck
|
|
"""
|
|
|
|
if not self.valves.readeck_url or not self.valves.api_token:
|
|
return "Error: Readeck URL and API token must be configured in the tool settings."
|
|
|
|
# Make request to Readeck API
|
|
result = self._make_request("/collections")
|
|
|
|
if "error" in result:
|
|
return f"Error fetching collections: {result['error']}"
|
|
|
|
collections = result.get("collections", [])
|
|
|
|
if not collections:
|
|
return "No collections found."
|
|
|
|
# Format collections for display
|
|
formatted_collections = []
|
|
for collection in collections:
|
|
formatted_collection = {
|
|
"id": collection.get("id"),
|
|
"name": collection.get("name"),
|
|
"description": collection.get("description", ""),
|
|
"bookmark_count": collection.get("bookmark_count", 0),
|
|
"created_at": collection.get("created_at"),
|
|
"updated_at": collection.get("updated_at")
|
|
}
|
|
formatted_collections.append(formatted_collection)
|
|
|
|
return {
|
|
"collections": formatted_collections,
|
|
"total": len(formatted_collections)
|
|
}
|
|
|
|
async def get_bookmark_by_id(
|
|
self,
|
|
bookmark_id: str,
|
|
__user__: Optional[dict] = None
|
|
) -> str:
|
|
"""
|
|
Get a specific bookmark by its ID
|
|
|
|
Args:
|
|
bookmark_id: The ID of the bookmark to retrieve
|
|
"""
|
|
|
|
if not self.valves.readeck_url or not self.valves.api_token:
|
|
return "Error: Readeck URL and API token must be configured in the tool settings."
|
|
|
|
if not bookmark_id:
|
|
return "Error: bookmark_id is required."
|
|
|
|
# Make request to Readeck API
|
|
result = self._make_request(f"/bookmarks/{bookmark_id}")
|
|
|
|
if "error" in result:
|
|
return f"Error fetching bookmark: {result['error']}"
|
|
|
|
bookmark = result.get("bookmark")
|
|
if not bookmark:
|
|
return f"Bookmark with ID {bookmark_id} not found."
|
|
|
|
# Format bookmark for display
|
|
formatted_bookmark = {
|
|
"id": bookmark.get("id"),
|
|
"title": bookmark.get("title", "Untitled"),
|
|
"url": bookmark.get("url"),
|
|
"content": bookmark.get("content", ""),
|
|
"excerpt": bookmark.get("excerpt", ""),
|
|
"labels": bookmark.get("labels", []),
|
|
"collection": bookmark.get("collection", {}),
|
|
"created_at": bookmark.get("created_at"),
|
|
"updated_at": bookmark.get("updated_at"),
|
|
"reading_time": bookmark.get("reading_time"),
|
|
"word_count": bookmark.get("word_count")
|
|
}
|
|
|
|
return formatted_bookmark
|
|
|
|
async def search_bookmarks(
|
|
self,
|
|
query: str,
|
|
limit: int = 10,
|
|
__user__: Optional[dict] = None
|
|
) -> str:
|
|
"""
|
|
Search bookmarks by title, content, or URL
|
|
|
|
Args:
|
|
query: Search query
|
|
limit: Number of results to return (default: 10)
|
|
"""
|
|
|
|
if not query:
|
|
return "Error: search query is required."
|
|
|
|
return await self.get_bookmarks(
|
|
limit=limit,
|
|
search=query,
|
|
__user__=__user__
|
|
)
|
|
|
|
async def get_labels(
|
|
self,
|
|
__user__: Optional[dict] = None
|
|
) -> str:
|
|
"""
|
|
Get all labels from Readeck
|
|
"""
|
|
|
|
if not self.valves.readeck_url or not self.valves.api_token:
|
|
return "Error: Readeck URL and API token must be configured in the tool settings."
|
|
|
|
# Make request to Readeck API
|
|
result = self._make_request("/bookmarks/labels")
|
|
|
|
if "error" in result:
|
|
return f"Error fetching labels: {result['error']}"
|
|
|
|
labels = result.get("labels", [])
|
|
|
|
if not labels:
|
|
return "No labels found."
|
|
|
|
# Format labels for display
|
|
formatted_labels = []
|
|
for label in labels:
|
|
if isinstance(label, str):
|
|
# Simple string label
|
|
formatted_labels.append({
|
|
"name": label,
|
|
"bookmark_count": None
|
|
})
|
|
elif isinstance(label, dict):
|
|
# Label object with additional info
|
|
formatted_labels.append({
|
|
"name": label.get("name"),
|
|
"bookmark_count": label.get("bookmark_count"),
|
|
"created_at": label.get("created_at"),
|
|
"updated_at": label.get("updated_at")
|
|
})
|
|
|
|
return {
|
|
"labels": formatted_labels,
|
|
"total": len(formatted_labels)
|
|
}
|
|
|
|
async def get_bookmarks_by_label(
|
|
self,
|
|
label_name: str,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
__user__: Optional[dict] = None
|
|
) -> str:
|
|
"""
|
|
Get bookmarks filtered by a specific label
|
|
|
|
Args:
|
|
label_name: Name of the label to filter by
|
|
limit: Number of bookmarks to return (default: 20, max: 100)
|
|
offset: Number of bookmarks to skip (default: 0)
|
|
"""
|
|
|
|
if not self.valves.readeck_url or not self.valves.api_token:
|
|
return "Error: Readeck URL and API token must be configured in the tool settings."
|
|
|
|
if not label_name:
|
|
return "Error: label_name is required."
|
|
|
|
# First, get the label metadata to retrieve the href_bookmarks URI
|
|
label_result = self._make_request(f"/bookmarks/labels/{label_name}")
|
|
|
|
if "error" in label_result:
|
|
return f"Error fetching label '{label_name}': {label_result['error']}"
|
|
|
|
# Extract the href_bookmarks URI
|
|
href_bookmarks = label_result.get("href_bookmarks")
|
|
if not href_bookmarks:
|
|
return f"No href_bookmarks found for label '{label_name}'. Label may not exist or have no bookmarks."
|
|
|
|
# Build query parameters for the bookmarks request
|
|
params = {
|
|
"limit": min(limit, 100), # Cap at 100 to prevent excessive requests
|
|
"offset": offset
|
|
}
|
|
|
|
# Extract the endpoint path from href_bookmarks (remove the base URL if present)
|
|
if href_bookmarks.startswith(self.valves.readeck_url):
|
|
endpoint_path = href_bookmarks[len(self.valves.readeck_url):]
|
|
if endpoint_path.startswith("/api"):
|
|
endpoint_path = endpoint_path[4:] # Remove /api prefix
|
|
else:
|
|
# Assume it's a relative path
|
|
endpoint_path = href_bookmarks
|
|
if endpoint_path.startswith("/api"):
|
|
endpoint_path = endpoint_path[4:] # Remove /api prefix
|
|
|
|
# Make request to get the actual bookmarks
|
|
bookmarks_result = self._make_request(endpoint_path, params)
|
|
|
|
if "error" in bookmarks_result:
|
|
return f"Error fetching bookmarks for label '{label_name}': {bookmarks_result['error']}"
|
|
|
|
# Format the response
|
|
bookmarks = bookmarks_result.get("bookmarks", [])
|
|
total = bookmarks_result.get("total", len(bookmarks))
|
|
|
|
if not bookmarks:
|
|
return f"No bookmarks found with label '{label_name}'."
|
|
|
|
# Format bookmarks for display
|
|
formatted_bookmarks = []
|
|
for bookmark in bookmarks:
|
|
formatted_bookmark = {
|
|
"id": bookmark.get("id"),
|
|
"title": bookmark.get("title", "Untitled"),
|
|
"url": bookmark.get("url"),
|
|
"excerpt": bookmark.get("excerpt", "")[:200] + "..." if bookmark.get("excerpt", "") else "",
|
|
"labels": bookmark.get("labels", []),
|
|
"collection": bookmark.get("collection", {}).get("name", ""),
|
|
"created_at": bookmark.get("created_at"),
|
|
"updated_at": bookmark.get("updated_at")
|
|
}
|
|
formatted_bookmarks.append(formatted_bookmark)
|
|
|
|
return {
|
|
"label": label_name,
|
|
"label_info": {
|
|
"name": label_result.get("name"),
|
|
"bookmark_count": label_result.get("bookmark_count"),
|
|
"href_bookmarks": href_bookmarks
|
|
},
|
|
"bookmarks": formatted_bookmarks,
|
|
"total": total,
|
|
"limit": params["limit"],
|
|
"offset": params["offset"]
|
|
} |