jpmschweitzerandClaude Opus 4.5
9446d6bf9a
Replace username/password login flow with simpler API token auth: - Use WIKI_GRAPHQL_API environment variable for JWT token - Remove login() method and session management - Add list_pages() method for fetching all pages - Keep legacy auth fields in config for backwards compatibility 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
633 lines
18 KiB
Python
633 lines
18 KiB
Python
"""
|
|
Wiki.js GraphQL client for Library Desk.
|
|
|
|
Provides async Wiki.js operations via GraphQL API:
|
|
- Page CRUD (create, read, update, delete)
|
|
- Search and listing
|
|
- Tag management
|
|
- Multi-tenancy via path namespaces
|
|
"""
|
|
|
|
import httpx
|
|
from typing import Optional, List, Dict, Any
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WikiJSClient:
|
|
"""
|
|
Wiki.js GraphQL API client.
|
|
|
|
Documentation: https://docs.requarks.io/dev/api
|
|
Authentication: API token (JWT) generated from Wiki.js admin panel
|
|
"""
|
|
|
|
def __init__(self, base_url: str, api_token: str):
|
|
"""
|
|
Initialize Wiki.js client.
|
|
|
|
Args:
|
|
base_url: Wiki.js base URL (e.g., "http://wiki:3000")
|
|
api_token: Wiki.js API token (JWT from admin panel)
|
|
"""
|
|
self.base_url = base_url.rstrip("/")
|
|
self.graphql_url = f"{self.base_url}/graphql"
|
|
self.api_token = api_token
|
|
self.client = httpx.AsyncClient(timeout=30.0)
|
|
logger.info(f"Initialized Wiki.js client: {base_url} (using API token)")
|
|
|
|
async def close(self):
|
|
"""Close HTTP client"""
|
|
await self.client.aclose()
|
|
|
|
def _ensure_authenticated(self):
|
|
"""Verify API token is configured."""
|
|
if not self.api_token:
|
|
raise Exception("Wiki.js API token not configured")
|
|
|
|
async def _execute_query(
|
|
self,
|
|
query: str,
|
|
variables: Optional[Dict] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Execute GraphQL query.
|
|
|
|
Args:
|
|
query: GraphQL query string
|
|
variables: Query variables
|
|
|
|
Returns:
|
|
Query result data
|
|
|
|
Raises:
|
|
Exception: If query fails or returns errors
|
|
"""
|
|
# Ensure API token is configured
|
|
self._ensure_authenticated()
|
|
|
|
payload = {
|
|
"query": query,
|
|
"variables": variables or {}
|
|
}
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {self.api_token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
try:
|
|
response = await self.client.post(
|
|
self.graphql_url,
|
|
headers=headers,
|
|
json=payload
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
|
|
if "errors" in result:
|
|
logger.error(f"GraphQL errors: {result['errors']}")
|
|
raise Exception(f"GraphQL errors: {result['errors']}")
|
|
|
|
return result.get("data", {})
|
|
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"GraphQL query failed: {e}", exc_info=True)
|
|
raise
|
|
|
|
async def list_pages(
|
|
self,
|
|
path_prefix: str = "",
|
|
tags: Optional[List[str]] = None,
|
|
limit: int = 50
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
List pages with optional filtering.
|
|
|
|
Multi-tenancy: Use path_prefix to filter by user namespace.
|
|
|
|
Args:
|
|
path_prefix: Filter by path prefix (e.g., "/users/jpmschweitzer")
|
|
tags: Filter by tags (e.g., ["projects"])
|
|
limit: Maximum results
|
|
|
|
Returns:
|
|
List of page objects
|
|
"""
|
|
query = """
|
|
query ListPages($limit: Int, $orderBy: PageOrderBy) {
|
|
pages {
|
|
list(limit: $limit, orderBy: $orderBy) {
|
|
id
|
|
path
|
|
title
|
|
description
|
|
tags
|
|
createdAt
|
|
updatedAt
|
|
isPublished
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
variables = {
|
|
"limit": limit,
|
|
"orderBy": "TITLE"
|
|
}
|
|
|
|
data = await self._execute_query(query, variables)
|
|
pages = data.get("pages", {}).get("list", [])
|
|
|
|
# Ensure tags is always a list
|
|
for page in pages:
|
|
if "tags" not in page or page["tags"] is None:
|
|
page["tags"] = []
|
|
|
|
# Filter by path prefix (client-side if API doesn't support)
|
|
if path_prefix:
|
|
# Normalize paths to have leading slash for consistent comparison
|
|
normalized_prefix = "/" + path_prefix.lstrip("/")
|
|
pages = [
|
|
p for p in pages
|
|
if ("/" + p["path"].lstrip("/")).startswith(normalized_prefix)
|
|
]
|
|
|
|
# Filter by tags (dossiers)
|
|
if tags:
|
|
pages = [
|
|
p for p in pages
|
|
if any(tag in (p.get("tags") or []) for tag in tags)
|
|
]
|
|
|
|
return pages
|
|
|
|
async def list_all_pages(
|
|
self,
|
|
path_prefix: str = "",
|
|
tags: Optional[List[str]] = None,
|
|
batch_size: int = 100
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
List ALL pages with pagination support.
|
|
|
|
Fetches pages in batches until all are retrieved.
|
|
|
|
Args:
|
|
path_prefix: Filter by path prefix (e.g., "users/jpmschweitzer")
|
|
tags: Filter by tags
|
|
batch_size: Number of pages per batch (max 100)
|
|
|
|
Returns:
|
|
Complete list of page objects
|
|
"""
|
|
all_pages = []
|
|
offset = 0
|
|
|
|
while True:
|
|
# Wiki.js list doesn't support offset, but limit is enough
|
|
# since we filter client-side by path_prefix
|
|
# Just fetch a large batch
|
|
pages = await self.list_pages(
|
|
path_prefix=path_prefix,
|
|
tags=tags,
|
|
limit=1000 # Fetch up to 1000 at once
|
|
)
|
|
|
|
if not pages:
|
|
break
|
|
|
|
all_pages = pages
|
|
break # Wiki.js list doesn't paginate, so one call is enough
|
|
|
|
logger.info(f"list_all_pages: found {len(all_pages)} pages (prefix: {path_prefix or 'all'})")
|
|
return all_pages
|
|
|
|
async def get_taxonomy_structure(
|
|
self,
|
|
user_namespace: str,
|
|
limit: int = 500
|
|
) -> Dict[str, List[str]]:
|
|
"""
|
|
Get the existing taxonomy structure for a user namespace.
|
|
|
|
Returns a dict mapping top-level categories to their subcategories.
|
|
This helps the LLM choose existing paths rather than creating new ones.
|
|
|
|
Args:
|
|
user_namespace: User namespace (e.g., "users/jpmschweitzer")
|
|
limit: Maximum pages to fetch
|
|
|
|
Returns:
|
|
Dict like {"reference": ["political-entities", "tech"], "places": ["the-netherlands"]}
|
|
"""
|
|
from collections import defaultdict
|
|
|
|
pages = await self.list_pages(path_prefix=user_namespace, limit=limit)
|
|
|
|
# Extract structure: category -> set of subcategories
|
|
structure: Dict[str, set] = defaultdict(set)
|
|
prefix = user_namespace.strip("/") + "/"
|
|
|
|
for page in pages:
|
|
path = page.get("path", "")
|
|
if not path.startswith(prefix):
|
|
continue
|
|
|
|
# Get relative path within user namespace
|
|
rel_path = path[len(prefix):]
|
|
parts = rel_path.split("/")
|
|
|
|
if len(parts) >= 1:
|
|
category = parts[0]
|
|
if len(parts) >= 2:
|
|
# Has subcategory (e.g., reference/political-entities/nato)
|
|
structure[category].add(parts[1])
|
|
else:
|
|
# Just ensure category exists in structure
|
|
structure[category] # Access to ensure key exists
|
|
|
|
# Convert sets to sorted lists
|
|
return {cat: sorted(list(subs)) for cat, subs in sorted(structure.items())}
|
|
|
|
async def get_page(self, page_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get single page by ID.
|
|
|
|
Args:
|
|
page_id: Page ID
|
|
|
|
Returns:
|
|
Page object or None if not found
|
|
"""
|
|
query = """
|
|
query GetPage($id: Int!) {
|
|
pages {
|
|
single(id: $id) {
|
|
id
|
|
path
|
|
title
|
|
description
|
|
content
|
|
tags {
|
|
tag
|
|
}
|
|
createdAt
|
|
updatedAt
|
|
isPublished
|
|
editor
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
try:
|
|
# Ensure page_id is int (GraphQL requires Int, not String)
|
|
data = await self._execute_query(query, {"id": int(page_id)})
|
|
page = data.get("pages", {}).get("single")
|
|
|
|
# Extract tag strings from tag objects
|
|
if page and "tags" in page and page["tags"]:
|
|
page["tags"] = [t["tag"] for t in page["tags"]]
|
|
elif page:
|
|
page["tags"] = []
|
|
|
|
return page
|
|
except Exception as e:
|
|
logger.error(f"Failed to get page {page_id}: {e}")
|
|
return None
|
|
|
|
async def create_page(
|
|
self,
|
|
path: str,
|
|
title: str,
|
|
content: str,
|
|
description: str = "",
|
|
tags: Optional[List[str]] = None,
|
|
is_published: bool = True,
|
|
is_private: bool = False,
|
|
editor: str = "markdown"
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create new page.
|
|
|
|
Multi-tenancy: Ensure path starts with user namespace.
|
|
|
|
Args:
|
|
path: Page path (e.g., "/users/jpmschweitzer/projects/library-desk")
|
|
title: Page title
|
|
content: Page content (markdown)
|
|
description: Short description
|
|
tags: List of tags (for dossier organization)
|
|
is_published: Whether page is published
|
|
is_private: Whether page is private
|
|
editor: Editor type (markdown, wysiwyg, etc.)
|
|
|
|
Returns:
|
|
Created page object
|
|
|
|
Raises:
|
|
Exception: If creation fails
|
|
"""
|
|
mutation = """
|
|
mutation CreatePage(
|
|
$content: String!,
|
|
$description: String!,
|
|
$editor: String!,
|
|
$isPublished: Boolean!,
|
|
$isPrivate: Boolean!,
|
|
$locale: String!,
|
|
$path: String!,
|
|
$tags: [String]!,
|
|
$title: String!
|
|
) {
|
|
pages {
|
|
create(
|
|
content: $content,
|
|
description: $description,
|
|
editor: $editor,
|
|
isPublished: $isPublished,
|
|
isPrivate: $isPrivate,
|
|
locale: $locale,
|
|
path: $path,
|
|
tags: $tags,
|
|
title: $title
|
|
) {
|
|
responseResult {
|
|
succeeded
|
|
errorCode
|
|
message
|
|
}
|
|
page {
|
|
id
|
|
path
|
|
title
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
variables = {
|
|
"path": path,
|
|
"title": title,
|
|
"content": content,
|
|
"description": description,
|
|
"tags": tags or [],
|
|
"isPublished": is_published,
|
|
"isPrivate": is_private,
|
|
"editor": editor,
|
|
"locale": "en"
|
|
}
|
|
|
|
data = await self._execute_query(mutation, variables)
|
|
result = data.get("pages", {}).get("create", {})
|
|
|
|
if not result.get("responseResult", {}).get("succeeded"):
|
|
error = result.get("responseResult", {})
|
|
raise Exception(f"Failed to create page: {error}")
|
|
|
|
logger.info(f"Created page: {path}")
|
|
return result.get("page")
|
|
|
|
async def update_page(
|
|
self,
|
|
page_id: int,
|
|
content: Optional[str] = None,
|
|
title: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
tags: Optional[List[str]] = None,
|
|
is_published: Optional[bool] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Update existing page.
|
|
|
|
Args:
|
|
page_id: Page ID
|
|
content: New content (optional)
|
|
title: New title (optional)
|
|
description: New description (optional)
|
|
tags: New tags (optional)
|
|
is_published: Published status (optional)
|
|
|
|
Returns:
|
|
Updated page object
|
|
|
|
Raises:
|
|
Exception: If update fails
|
|
"""
|
|
mutation = """
|
|
mutation UpdatePage(
|
|
$id: Int!,
|
|
$content: String,
|
|
$title: String,
|
|
$description: String,
|
|
$tags: [String],
|
|
$isPublished: Boolean
|
|
) {
|
|
pages {
|
|
update(
|
|
id: $id,
|
|
content: $content,
|
|
title: $title,
|
|
description: $description,
|
|
tags: $tags,
|
|
isPublished: $isPublished
|
|
) {
|
|
responseResult {
|
|
succeeded
|
|
errorCode
|
|
message
|
|
}
|
|
page {
|
|
id
|
|
updatedAt
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
variables = {"id": page_id}
|
|
if content is not None:
|
|
variables["content"] = content
|
|
if title is not None:
|
|
variables["title"] = title
|
|
if description is not None:
|
|
variables["description"] = description
|
|
if tags is not None:
|
|
variables["tags"] = tags
|
|
if is_published is not None:
|
|
variables["isPublished"] = is_published
|
|
|
|
data = await self._execute_query(mutation, variables)
|
|
result = data.get("pages", {}).get("update", {})
|
|
|
|
if not result.get("responseResult", {}).get("succeeded"):
|
|
error = result.get("responseResult", {})
|
|
raise Exception(f"Failed to update page: {error}")
|
|
|
|
logger.info(f"Updated page: {page_id}")
|
|
return result.get("page")
|
|
|
|
async def delete_page(self, page_id: int):
|
|
"""
|
|
Delete page.
|
|
|
|
Args:
|
|
page_id: Page ID
|
|
|
|
Raises:
|
|
Exception: If deletion fails
|
|
"""
|
|
mutation = """
|
|
mutation DeletePage($id: Int!) {
|
|
pages {
|
|
delete(id: $id) {
|
|
responseResult {
|
|
succeeded
|
|
errorCode
|
|
message
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
data = await self._execute_query(mutation, {"id": page_id})
|
|
result = data.get("pages", {}).get("delete", {})
|
|
|
|
if not result.get("responseResult", {}).get("succeeded"):
|
|
error = result.get("responseResult", {})
|
|
raise Exception(f"Failed to delete page: {error}")
|
|
|
|
logger.info(f"Deleted page: {page_id}")
|
|
|
|
async def search_pages(
|
|
self,
|
|
query: str,
|
|
path_prefix: Optional[str] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Search pages by content.
|
|
|
|
Args:
|
|
query: Search query
|
|
path_prefix: Optional path prefix filter
|
|
|
|
Returns:
|
|
List of matching pages
|
|
"""
|
|
gql_query = """
|
|
query SearchPages($query: String!) {
|
|
pages {
|
|
search(query: $query) {
|
|
results {
|
|
id
|
|
path
|
|
title
|
|
description
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
data = await self._execute_query(gql_query, {"query": query})
|
|
results = data.get("pages", {}).get("search", {}).get("results", [])
|
|
|
|
# Filter by path prefix if provided
|
|
if path_prefix:
|
|
results = [r for r in results if r["path"].startswith(path_prefix)]
|
|
|
|
return results
|
|
|
|
async def get_page_tree(self, path: str = "/") -> List[Dict[str, Any]]:
|
|
"""
|
|
Get page tree structure.
|
|
|
|
Args:
|
|
path: Root path
|
|
|
|
Returns:
|
|
Tree structure of pages
|
|
"""
|
|
query = """
|
|
query GetPageTree($parent: Int, $mode: String!) {
|
|
pages {
|
|
tree(parent: $parent, mode: $mode) {
|
|
id
|
|
path
|
|
title
|
|
isFolder
|
|
pageId
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
try:
|
|
data = await self._execute_query(
|
|
query,
|
|
{"parent": 0, "mode": "all"}
|
|
)
|
|
return data.get("pages", {}).get("tree", [])
|
|
except Exception as e:
|
|
logger.error(f"Failed to get page tree: {e}")
|
|
return []
|
|
|
|
async def move_page(
|
|
self,
|
|
page_id: int,
|
|
new_path: str,
|
|
locale: str = "en"
|
|
) -> bool:
|
|
"""
|
|
Move/rename page.
|
|
|
|
Args:
|
|
page_id: Page ID
|
|
new_path: New page path
|
|
locale: Page locale
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
mutation = """
|
|
mutation MovePage($id: Int!, $destinationPath: String!, $destinationLocale: String!) {
|
|
pages {
|
|
move(id: $id, destinationPath: $destinationPath, destinationLocale: $destinationLocale) {
|
|
responseResult {
|
|
succeeded
|
|
errorCode
|
|
message
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
try:
|
|
data = await self._execute_query(
|
|
mutation,
|
|
{
|
|
"id": page_id,
|
|
"destinationPath": new_path,
|
|
"destinationLocale": locale
|
|
}
|
|
)
|
|
result = data.get("pages", {}).get("move", {})
|
|
success = result.get("responseResult", {}).get("succeeded", False)
|
|
|
|
if success:
|
|
logger.info(f"Moved page {page_id} to {new_path}")
|
|
return success
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to move page: {e}")
|
|
return False
|