nextcloud_notes.py
Rama main · réplica de lectura del repositorio original.
"""piSynapse Nextcloud Notes
REST API integration for Nextcloud Notes (v1).
"""
import asyncio
import json
import logging
import threading
from typing import Any
from utils import retry, sanitize_external_text
logger = logging.getLogger("piSynapse")
# Singleton pattern — reuse connection
_notes_client: "NextcloudNotesClient | None" = None
# Notes run in worker threads (asyncio.to_thread); the lock makes lazy
# singleton creation single-flight across concurrent requests.
_notes_lock = threading.Lock()
class NotFoundError(Exception):
"""Raised when Nextcloud returns 404 (resource does not exist)."""
class NextcloudNotesClient:
"""Nextcloud Notes REST API client."""
def __init__(self, creds: dict | None = None):
from config import NEXTCLOUD_PASSWORD, NEXTCLOUD_TIMEOUT, NEXTCLOUD_URL, NEXTCLOUD_USER
creds = creds or {}
self._base = (creds.get("url") or NEXTCLOUD_URL).rstrip("/")
self._user = creds.get("user") or NEXTCLOUD_USER
self._password = creds.get("password") or NEXTCLOUD_PASSWORD
self._timeout = NEXTCLOUD_TIMEOUT
self._list_cache: list[dict] | None = None
self._list_cache_ts: float = 0
def _request(self, method: str, path: str, data: dict | None = None) -> Any:
"""Make an authenticated request to Nextcloud Notes API.
Raises NotFoundError on 404 (so callers can tell "does not exist"
from a network/server failure), and the raw urllib error otherwise.
"""
import base64
import urllib.error
import urllib.request
url = f"{self._base}/index.php/apps/notes/api/v1/{path.lstrip('/')}"
headers = {
"Authorization": f"Basic {base64.b64encode(f'{self._user}:{self._password}'.encode()).decode()}",
"Content-Type": "application/json",
"Accept": "application/json",
}
body = json.dumps(data).encode("utf-8") if data else None
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=self._timeout) as resp:
raw = resp.read().decode("utf-8")
return json.loads(raw) if raw.strip() else None
except urllib.error.HTTPError as e:
if e.code == 404:
raise NotFoundError(f"Not found: {path}") from e
raise Exception(f"Nextcloud API error {e.code}: {e.reason}")
@retry(attempts=2, delay=1.0)
def list_notes(self) -> list[dict]:
"""Fetch all notes from Nextcloud (paginated, cached 30s)."""
import time
now = time.time()
if self._list_cache and now - self._list_cache_ts < 30:
return self._list_cache
notes: list[dict] = []
page = 1
per_page = 100
while page <= 200:
result = self._request("GET", f"notes?page={page}&itemsPerPage={per_page}")
if isinstance(result, dict):
data = result.get("data", [])
batch = data if isinstance(data, list) else (data.get("notes", []) if isinstance(data, dict) else [])
else:
batch = result or []
if not batch:
break
notes.extend(batch)
if len(batch) < per_page:
break
seen_ids = {n.get("id") for n in notes[:-len(batch)]}
new_ids = {n.get("id") for n in batch}
if new_ids.issubset(seen_ids):
break # server ignores pagination params — stop to avoid infinite loop
page += 1
self._list_cache = notes
self._list_cache_ts = now
return self._list_cache
@retry(attempts=2, delay=1.0)
def get_note(self, note_id: int) -> dict | None:
"""Fetch a single note by ID."""
return self._request("GET", f"notes/{note_id}")
@retry(attempts=2, delay=1.0)
def create_note(self, title: str, content: str = "", category: str = "", tags: list[str] | None = None) -> dict:
"""Create a new note."""
payload = {"title": title, "content": content}
if category:
payload["category"] = category
if tags:
payload["tags"] = tags
return self._request("POST", "notes", payload)
@retry(attempts=2, delay=1.0)
def update_note(self, note_id: int, title: str | None = None, content: str | None = None,
category: str | None = None, tags: list[str] | None = None) -> dict | None:
"""Update an existing note. Only fields passed are updated.
The GET (merge base) and the PUT share a single retry scope — fetching
via the decorated ``get_note`` would nest two retry layers (up to 4
attempts and doubled latency), so the merge is read here directly.
"""
try:
current = self._request("GET", f"notes/{note_id}")
except NotFoundError:
return None
if not current or current.get("etag") is None:
return None
payload: dict[str, Any] = {"etag": current.get("etag", "")}
if title is not None:
payload["title"] = title
if content is not None:
payload["content"] = content
if category is not None:
payload["category"] = category
if tags is not None:
payload["tags"] = tags
return self._request("PUT", f"notes/{note_id}", payload)
@retry(attempts=2, delay=1.0)
def delete_note(self, note_id: int) -> bool:
"""Delete a note by ID. Returns False only when the note does not exist.
Network/server failures raise instead of being reported as "not found".
"""
try:
self._request("DELETE", f"notes/{note_id}")
except NotFoundError:
return False
return True
@retry(attempts=2, delay=1.0)
def search_notes_server(self, query: str, limit: int = 10) -> list[dict]:
"""Search notes using Nextcloud server-side search API (if supported).
Falls back to client-side filtering if the API doesn't support search param.
Returns up to `limit` matches.
"""
# Try server-side search first (Nextcloud Notes API v1.0.2+ supports ?search=)
try:
encoded_query = query.replace(" ", "+")
result = self._request("GET", f"notes?search={encoded_query}&page=1&itemsPerPage={limit}")
if isinstance(result, dict):
data = result.get("data", [])
notes = data if isinstance(data, list) else (data.get("notes", []) if isinstance(data, dict) else [])
else:
notes = result or []
if notes:
return notes[:limit]
except Exception:
pass # Fall back to client-side
# Fallback: client-side filtering
all_notes = self.list_notes()
q = query.lower()
matches = [
n for n in all_notes
if q in (n.get("title", "") + " " + n.get("content", "")).lower()
]
return matches[:limit]
def _get_client() -> NextcloudNotesClient | None:
"""Return singleton client, create if needed (or ephemeral per-user one).
The shared .env account serves admins only (NC_SHARED_OK); everyone else
needs personal credentials, else None (callers report "not configured").
"""
global _notes_client
from config import NEXTCLOUD_PASSWORD, NEXTCLOUD_URL
from utils import NC_CREDS, NC_SHARED_OK
creds = NC_CREDS.get()
if creds:
return NextcloudNotesClient(creds=creds)
if not NEXTCLOUD_URL or not NEXTCLOUD_PASSWORD:
return None
if not NC_SHARED_OK.get():
return None
if _notes_client is None:
with _notes_lock:
if _notes_client is None:
_notes_client = NextcloudNotesClient()
return _notes_client
def _invalidate_list_cache() -> None:
"""Drop the 30s listing cache after a successful write (create/update/delete),
so the next listing reflects the change immediately.
"""
if _notes_client is not None:
_notes_client._list_cache = None
_notes_client._list_cache_ts = 0.0
# -- Async wrappers for FastAPI/tool dispatcher compatibility --
async def list_notes() -> tuple[str, list[dict]]:
"""List all notes.
Returns ``(display_text, items)`` where ``items`` are the raw note dicts
in the exact order they are numbered in ``display_text`` (the model only
sees list positions; real IDs never appear in tool output).
"""
client = _get_client()
if not client:
return "ERROR: Nextcloud credentials missing.", []
try:
notes = await asyncio.to_thread(client.list_notes)
if not notes:
return "No notes found.", []
lines = [" Notes:\n"]
for i, n in enumerate(notes, 1):
title = n.get("title", "Untitled")
category = n.get("category", "")
tags = n.get("tags", [])
starred = n.get("starred", False)
content = n.get("content", "")
star = " * " if starred else " "
lines.append(f"{star}{i}. {title}")
meta = []
if category:
meta.append(f"Category: {category}")
if tags:
meta.append(f"Tags: {', '.join(tags)}")
if meta:
lines.append(f" {' | '.join(meta)}")
if content:
lines.append(f" Preview: {sanitize_external_text(content)[:80].replace(chr(10), ' ')}")
lines.append("")
return "\n".join(lines), notes
except Exception as e:
logger.error(f"Nextcloud Notes Error: {e}")
return "ERROR: Failed to list notes.", []
async def get_note(note_id: int) -> str:
"""Get a single note."""
client = _get_client()
if not client:
return "ERROR: Nextcloud credentials missing."
try:
note = await asyncio.to_thread(client.get_note, note_id)
if not note:
return "ERROR: Note not found."
title = note.get("title", "Untitled")
content = note.get("content", "")
category = note.get("category", "")
tags = note.get("tags", [])
modified = note.get("modified", "")
starred = note.get("starred", False)
lines = []
star = " * " if starred else " "
lines.append(f"{star}{title}")
meta = []
if category:
meta.append(f"Category: {category}")
if tags:
meta.append(f"Tags: {', '.join(tags)}")
if modified:
meta.append(f"Modified: {modified}")
if meta:
lines.append(f" {' | '.join(meta)}")
lines.append("")
lines.append(content)
return "\n".join(lines)
except NotFoundError:
return "Note not found."
except Exception as e:
logger.error(f"Nextcloud Notes Error: {e}")
return "ERROR: Failed to get note."
async def create_note(title: str, content: str = "", category: str = "") -> tuple[str, int | None]:
"""Create a new note.
Returns ``(result_text, note_id)`` — the id is forwarded by the dispatcher
so the verification hook can re-read the note and confirm it persisted.
"""
client = _get_client()
if not client:
return "ERROR: Nextcloud credentials missing.", None
try:
resp = await asyncio.to_thread(client.create_note, title, content, category)
_invalidate_list_cache()
note_id = resp.get("id") if isinstance(resp, dict) else None
return f"OK Note '{title}' created.", note_id
except Exception as e:
logger.error(f"Nextcloud Notes Error: {e}")
return "ERROR: Failed to create note.", None
async def update_note(note_id: int, title: str | None = None, content: str | None = None,
category: str | None = None, tags: list[str] | None = None) -> str:
"""Update an existing note. Only fields passed are forwarded to Nextcloud."""
client = _get_client()
if not client:
return "ERROR: Nextcloud credentials missing."
try:
result = await asyncio.to_thread(client.update_note, note_id, title, content, category, tags)
if not result:
return "NOOP: Note not found."
_invalidate_list_cache()
return "OK Note updated."
except NotFoundError:
return "NOOP: Note not found."
except Exception as e:
logger.error(f"Nextcloud Notes Error: {e}")
return "ERROR: Failed to update note."
async def delete_note(note_id: int) -> str:
"""Delete a note."""
client = _get_client()
if not client:
return "ERROR: Nextcloud credentials missing."
try:
ok = await asyncio.to_thread(client.delete_note, note_id)
if ok:
_invalidate_list_cache()
return "OK Note deleted." if ok else "NOOP: Note not found."
except NotFoundError:
return "NOOP: Note not found."
except Exception as e:
logger.error(f"Nextcloud Notes Error: {e}")
return "ERROR: Failed to delete note."
async def search_notes(query: str, limit: int = 10) -> tuple[str, list[dict]]:
"""Search notes by title or content.
Returns ``(display_text, items)`` — same contract as ``list_notes``.
Uses server-side search API when available (Nextcloud Notes v1.0.2+).
"""
client = _get_client()
if not client:
return "ERROR: Nextcloud credentials missing.", []
try:
matches = await asyncio.to_thread(client.search_notes_server, query, limit)
if not matches:
return f"'{query}' not found in notes.", []
lines = [f" Search Results for '{query}':\n"]
for i, n in enumerate(matches, 1):
title = n.get("title", "Untitled")
preview = n.get("content", "")[:100].replace("\n", " ")
lines.append(f" {i}. {title}")
if preview:
lines.append(f" Preview: {preview}...")
lines.append("")
return "\n".join(lines), matches
except Exception as e:
logger.error(f"Nextcloud Notes Error: {e}")
return "ERROR: Failed to search notes.", []