nextcloud_tasks.py
Rama main · réplica de lectura del repositorio original.
"""piSynapse Nextcloud Tasks
CalDAV integration for Nextcloud Tasks (VTODO).
"""
import asyncio
import logging
import threading
import uuid
from datetime import date, datetime
from icalendar import Todo
from utils import nc_cache_key, retry, sanitize_external_text
logger = logging.getLogger("piSynapse")
# Singleton — reuse client (shared admin account only; per-user clients are
# built fresh per call in _get_dav_client)
_client = None
# Task calendar per Nextcloud identity (utils.nc_cache_key). The credential
# gate in _get_task_calendar runs BEFORE this cache is consulted — a cached
# hit must never bypass NC_CREDS/NC_SHARED_OK (reopens the 2026-09-27
# invisibility hole otherwise).
_task_calendars: dict[str, object] = {}
# 30s TTL cache of fetched todos: (include_completed, nc_cache_key, todos).
# The flag must match the request (a pending-only fetch must never serve
# show_completed=True and vice versa) and the key must match the caller's
# Nextcloud identity (2026-10-08 analysis — cross-user todo leak).
_todos_cache: tuple[bool, str, list] | None = None
_todos_cache_ts: float = 0
# Tasks run in worker threads (asyncio.to_thread); the lock makes the lazy
# singleton creation single-flight across concurrent requests.
_client_lock = threading.Lock()
def _build_dav_client(url: str, username: str, password: str, timeout: int):
import caldav
caldav_url = f"{url.rstrip('/')}/remote.php/dav/"
return caldav.DAVClient(url=caldav_url, username=username, password=password, timeout=timeout)
def _get_dav_client():
"""Return cached CalDAV client (ephemeral per-user one, else admin-shared)."""
global _client
from config import NEXTCLOUD_PASSWORD, NEXTCLOUD_TIMEOUT, NEXTCLOUD_URL, NEXTCLOUD_USER
from utils import NC_CREDS, NC_SHARED_OK
creds = NC_CREDS.get()
if creds and creds.get("url") and creds.get("password"):
return _build_dav_client(creds["url"], creds.get("user", ""), creds["password"], NEXTCLOUD_TIMEOUT)
if not NEXTCLOUD_URL or not NEXTCLOUD_PASSWORD:
return None
if not NC_SHARED_OK.get():
return None
if _client is None:
with _client_lock:
if _client is None:
_client = _build_dav_client(NEXTCLOUD_URL, NEXTCLOUD_USER, NEXTCLOUD_PASSWORD, NEXTCLOUD_TIMEOUT)
return _client
def _get_task_calendar():
"""Find the first calendar that supports VTODO (tasks).
Order matters: the credential gate (_get_dav_client) runs FIRST and the
cache is consulted only after it — returning a cached calendar without
re-checking NC_CREDS/NC_SHARED_OK would serve it to callers who lost (or
never had) access. Entries are keyed per identity (utils.nc_cache_key).
"""
client = _get_dav_client()
if not client:
return None
key = nc_cache_key()
cached = _task_calendars.get(key)
if cached is not None:
return cached
try:
principal = client.principal()
calendars = principal.calendars()
for cal in calendars:
try:
components = cal.get_supported_components()
if "VTODO" in components:
_task_calendars[key] = cal
return cal
except Exception:
continue
if calendars:
_task_calendars[key] = calendars[0]
return calendars[0]
return None
except Exception as e:
logger.error(f"Failed to find task calendar: {e}")
return None
def _todo_to_dict(todo) -> dict:
"""Extract a clean dict from a caldav Todo object."""
try:
component = getattr(todo, "icalendar_component", None) or todo.icalendar_instance
# caldav may return the VCALENDAR wrapper — drill into the VTODO
# child, otherwise every field reads as missing ('Untitled').
if getattr(component, "name", "") == "VCALENDAR":
vtodos = [sc for sc in component.subcomponents if sc.name == "VTODO"]
if vtodos:
component = vtodos[0]
summary = str(component.get("SUMMARY", "Untitled"))
uid = str(component.get("UID", ""))
status = str(component.get("STATUS", "NEEDS-ACTION"))
priority = int(component.get("PRIORITY", 0) or 0)
percent = int(component.get("PERCENT-COMPLETE", 0) or 0)
description = str(component.get("DESCRIPTION", "") or "")
categories = component.get("CATEGORIES", "")
if isinstance(categories, list):
categories = ", ".join(categories)
else:
categories = str(categories) if categories else ""
due = component.get("DUE")
due_str = ""
if due:
due_val = due.dt if hasattr(due, "dt") else due
if isinstance(due_val, datetime):
due_str = due_val.strftime("%Y-%m-%d %H:%M")
elif isinstance(due_val, date):
due_str = due_val.strftime("%Y-%m-%d")
return {
"uid": uid,
"summary": summary,
"status": status,
"priority": priority,
"percent": percent,
"description": description[:500],
"categories": categories,
"due": due_str,
"completed": status == "COMPLETED",
}
except Exception as e:
logger.error(f"Error parsing todo: {e}")
return {"uid": "?", "summary": "Error parsing task", "status": "UNKNOWN"}
# -- Sync helpers (called via asyncio.to_thread) --
def _calendar_or_config_error() -> str:
"""Explain why the task calendar is unavailable.
``_get_task_calendar`` returns None for two very different reasons:
Nextcloud isn't configured at all (credentials missing) or it is
configured but no VTODO-capable calendar exists. Reporting the latter
for the former is misleading — the user would create a calendar they
can't use.
"""
if _get_dav_client() is None:
return "ERROR: Nextcloud not configured. Set NEXTCLOUD_URL and NEXTCLOUD_PASSWORD in settings."
return "ERROR: No task calendar found. Create a task list in Nextcloud Tasks app first."
@retry(attempts=2, delay=1.0)
def _create_task_sync(summary: str, due: str, priority: int, notes: str) -> tuple[str, str]:
global _todos_cache, _todos_cache_ts
cal = _get_task_calendar()
if not cal:
return _calendar_or_config_error(), ""
todo = Todo()
uid = f"{uuid.uuid4()}@pisynapse"
todo.add("uid", uid)
todo.add("summary", summary)
todo.add("dtstart", datetime.now())
todo.add("dtstamp", datetime.now())
todo.add("status", "NEEDS-ACTION")
todo.add("percent-complete", 0)
if priority:
todo.add("priority", min(max(priority, 1), 9))
if notes:
todo.add("description", notes)
due_warning = ""
if due:
parsed = False
try:
due_dt = datetime.fromisoformat(due)
todo.add("due", due_dt)
parsed = True
except ValueError:
try:
due_dt = datetime.strptime(due, "%Y-%m-%d")
todo.add("due", due_dt)
parsed = True
except ValueError:
pass
if not parsed:
due_warning = (
f" Warning: could not parse due date '{due}' — the task was created WITHOUT a due date."
)
cal.save_todo(todo)
# Drop the listing cache so the new task shows up immediately.
_todos_cache = None
_todos_cache_ts = 0
return f"OK Task '{summary}' created.{due_warning}", uid
def _fetch_todos_cached(include_completed: bool) -> list:
"""Fetch todos via the 30s cache (flag- and identity-aware, see module docstring).
A failed fetch raises (NOT swallowed): retry fires via @retry and the
async wrapper reports a real error instead of "No tasks found."
"""
global _todos_cache, _todos_cache_ts
import time
now = time.time()
key = nc_cache_key()
if (_todos_cache is None or _todos_cache[0] != include_completed
or _todos_cache[1] != key or now - _todos_cache_ts > 30):
_todos_cache = (
include_completed,
key,
_get_task_calendar().todos(include_completed=include_completed),
)
_todos_cache_ts = now
return _todos_cache[2]
@retry(attempts=2, delay=1.0)
def _list_tasks_sync(show_completed: bool) -> tuple[str, list[dict]]:
"""Build the numbered task listing.
Returns ``(display_text, items)`` where ``items`` are the task dicts in
the exact order they are numbered in ``display_text`` (the model only
sees list positions; real UIDs never appear in tool output).
"""
global _todos_cache, _todos_cache_ts
cal = _get_task_calendar()
if not cal:
return _calendar_or_config_error(), []
todos = _fetch_todos_cached(show_completed)
if not todos:
return "No tasks found.", []
tasks = []
for t in todos:
d = _todo_to_dict(t)
if not show_completed and d.get("completed"):
continue
tasks.append(d)
if not tasks:
return "No pending tasks." if not show_completed else "No tasks found.", []
lines = [" Tasks:\n"]
for i, t in enumerate(tasks, 1):
status_icon = "x" if t["completed"] else "o"
lines.append(f" {i}. [{status_icon}] {t['summary']}")
meta = []
if t["due"]:
meta.append(f"Due: {t['due']}")
if t["priority"]:
meta.append(f"P: {t['priority']}")
if t["categories"]:
meta.append(f"Tags: {t['categories']}")
if meta:
lines.append(f" {' | '.join(meta)}")
desc = t.get("description", "")
if desc:
lines.append(f" Notes: {sanitize_external_text(desc)[:120]}")
lines.append("")
return "\n".join(lines), tasks
@retry(attempts=2, delay=1.0)
def _complete_task_sync(uid_prefix: str) -> str:
global _todos_cache, _todos_cache_ts
cal = _get_task_calendar()
if not cal:
return _calendar_or_config_error()
# A failed fetch raises (NOT swallowed): retry fires via @retry and the
# async wrapper reports a real error instead of "not found".
todos = cal.todos()
for t in todos:
d = _todo_to_dict(t)
if d["uid"].startswith(uid_prefix) and not d["completed"]:
t.complete()
_todos_cache = None
_todos_cache_ts = 0
return f"OK '{d['summary']}' marked as done."
return f"NOOP: Task with UID '{uid_prefix}' not found or already completed."
@retry(attempts=2, delay=1.0)
def _delete_task_sync(uid_prefix: str) -> str:
global _todos_cache, _todos_cache_ts
cal = _get_task_calendar()
if not cal:
return _calendar_or_config_error()
# A failed fetch raises (NOT swallowed): retry fires via @retry and the
# async wrapper reports a real error instead of "not found".
todos = cal.todos()
for t in todos:
d = _todo_to_dict(t)
if d["uid"].startswith(uid_prefix):
t.delete()
_todos_cache = None
_todos_cache_ts = 0
return f"OK '{d['summary']}' deleted."
return f"NOOP: Task with UID '{uid_prefix}' not found."
@retry(attempts=2, delay=1.0)
def _search_tasks_sync(query: str, limit: int = 10) -> tuple[str, list[dict]]:
"""Search tasks by summary/description — same contract as ``_list_tasks_sync``."""
global _todos_cache, _todos_cache_ts
cal = _get_task_calendar()
if not cal:
return _calendar_or_config_error(), []
todos = _fetch_todos_cached(False)
q = query.lower()
matches = []
for t in todos:
d = _todo_to_dict(t)
if q in d.get("summary", "").lower() or q in d.get("description", "").lower():
matches.append(d)
if not matches:
return f"'{query}' not found in tasks.", []
lines = [f" Search Results for '{query}':\n"]
for i, t in enumerate(matches[:limit], 1):
status_icon = "x" if t["completed"] else "o"
lines.append(f" {i}. [{status_icon}] {t['summary']}")
meta = []
if t["due"]:
meta.append(f"Due: {t['due']}")
if meta:
lines.append(f" {' | '.join(meta)}")
lines.append("")
return "\n".join(lines), matches[:limit]
# -- Async wrappers (for FastAPI / tool dispatcher) --
async def create_task(summary: str, due: str = "", priority: int = 0, notes: str = "") -> tuple[str, str]:
try:
return await asyncio.to_thread(_create_task_sync, summary, due, priority, notes)
except Exception as e:
logger.error(f"Tasks Error: {e}")
return "ERROR: Failed to create task.", ""
async def list_tasks(show_completed: bool = False) -> tuple[str, list[dict]]:
try:
# Data-fetch budget (3x the connection timeout): a hung CalDAV
# server must fail the turn, not the whole request.
return await asyncio.wait_for(
asyncio.to_thread(_list_tasks_sync, show_completed), timeout=30
)
except Exception as e:
logger.error(f"Tasks Error: {e}")
return "ERROR: Failed to list tasks.", []
async def complete_task(uid_prefix: str) -> str:
try:
return await asyncio.to_thread(_complete_task_sync, uid_prefix)
except Exception as e:
logger.error(f"Tasks Error: {e}")
return "ERROR: Failed to complete task."
async def delete_task(uid_prefix: str) -> str:
try:
return await asyncio.to_thread(_delete_task_sync, uid_prefix)
except Exception as e:
logger.error(f"Tasks Error: {e}")
return "ERROR: Failed to delete task."
async def search_tasks(query: str, limit: int = 10) -> tuple[str, list[dict]]:
try:
return await asyncio.to_thread(_search_tasks_sync, query, limit)
except Exception as e:
logger.error(f"Tasks Error: {e}")
return "ERROR: Failed to search tasks.", []