bvandeusen
2fd9a2300a
Drift-audit Group 5 #7/#8 + Group 7 (CalDAV write-path): - Point events no longer fabricate a 60-min DTEND: caldav.create_event emits DTSTART-only when there's no end and no duration, so the next pull doesn't read it back as duration_minutes=60 and silently lengthen the event. - Recurrence edits now propagate: caldav.update_event gains a recurrence param (sentinel = leave unchanged; value/empty = set/clear RRULE), and _push_update passes the local event's rule so a changed/cleared RRULE isn't overwritten by the stale remote rule on the next pull. - Event deletions propagate to CalDAV: trash.delete captures an event's caldav_uid before soft-deleting and fires _push_delete, so a UI/MCP delete removes the remote copy instead of leaving it to linger. (delete_event the service primitive is kept — still tested/usable — rather than removed.) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
771 lines
26 KiB
Python
771 lines
26 KiB
Python
"""CalDAV calendar integration service."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import date as date_type, datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import caldav
|
|
import icalendar
|
|
|
|
from fabledassistant.services.settings import get_all_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CALDAV_SETTING_KEYS = ["caldav_url", "caldav_username", "caldav_password", "caldav_calendar_name", "caldav_timezone"]
|
|
|
|
# Sentinel: distinguishes "leave the RRULE untouched" from "clear it" (None/"")
|
|
# in update_event, since None is a meaningful value for recurrence.
|
|
_RECURRENCE_UNSET = object()
|
|
|
|
|
|
async def get_caldav_config(user_id: int) -> dict[str, str]:
|
|
"""Return the user's CalDAV config from their settings."""
|
|
all_settings = await get_all_settings(user_id)
|
|
return {k: all_settings.get(k, "") for k in CALDAV_SETTING_KEYS}
|
|
|
|
|
|
async def is_caldav_configured(user_id: int) -> bool:
|
|
"""Check if the user has configured an external CalDAV server."""
|
|
config = await get_caldav_config(user_id)
|
|
return bool(config.get("caldav_url") and config.get("caldav_username") and config.get("caldav_password"))
|
|
|
|
|
|
def _get_calendar(client: caldav.DAVClient, calendar_name: str) -> caldav.Calendar:
|
|
"""Get a named calendar or the first available one (synchronous)."""
|
|
principal = client.principal()
|
|
calendars = principal.calendars()
|
|
if not calendars:
|
|
raise ValueError("No calendars found on the CalDAV server.")
|
|
if calendar_name:
|
|
for cal in calendars:
|
|
if cal.name == calendar_name:
|
|
return cal
|
|
names = [c.name for c in calendars]
|
|
raise ValueError(f"Calendar '{calendar_name}' not found. Available: {', '.join(names)}")
|
|
return calendars[0]
|
|
|
|
|
|
def _get_all_calendars(client: caldav.DAVClient) -> list[caldav.Calendar]:
|
|
"""Get all calendars for the user (synchronous)."""
|
|
principal = client.principal()
|
|
calendars = principal.calendars()
|
|
if not calendars:
|
|
raise ValueError("No calendars found on the CalDAV server.")
|
|
return calendars
|
|
|
|
|
|
def _make_client(config: dict[str, str]) -> caldav.DAVClient:
|
|
"""Create a CalDAV client from config dict."""
|
|
return caldav.DAVClient(
|
|
url=config["caldav_url"],
|
|
username=config.get("caldav_username") or None,
|
|
password=config.get("caldav_password") or None,
|
|
)
|
|
|
|
|
|
def _parse_vevent(component) -> dict | None:
|
|
"""Extract event data from a VEVENT component."""
|
|
if component.name != "VEVENT":
|
|
return None
|
|
title = str(component.get("SUMMARY", ""))
|
|
dtstart = component.get("DTSTART")
|
|
dtend = component.get("DTEND")
|
|
location = str(component.get("LOCATION", ""))
|
|
description = str(component.get("DESCRIPTION", ""))
|
|
uid = str(component.get("UID", ""))
|
|
start_str = dtstart.dt.isoformat() if dtstart else ""
|
|
end_str = dtend.dt.isoformat() if dtend else ""
|
|
|
|
result = {
|
|
"uid": uid,
|
|
"title": title,
|
|
"start": start_str,
|
|
"end": end_str,
|
|
"location": location,
|
|
"description": description,
|
|
}
|
|
|
|
# Extract recurrence rule
|
|
rrule = component.get("RRULE")
|
|
if rrule:
|
|
result["recurrence"] = rrule.to_ical().decode("utf-8")
|
|
|
|
# Extract alarms
|
|
alarms = []
|
|
for sub in component.subcomponents:
|
|
if sub.name == "VALARM":
|
|
trigger = sub.get("TRIGGER")
|
|
if trigger and trigger.dt:
|
|
minutes = abs(int(trigger.dt.total_seconds() // 60))
|
|
alarms.append({"minutes_before": minutes})
|
|
if alarms:
|
|
result["alarms"] = alarms
|
|
|
|
# Extract attendees
|
|
attendees = component.get("ATTENDEE")
|
|
if attendees:
|
|
if not isinstance(attendees, list):
|
|
attendees = [attendees]
|
|
result["attendees"] = [str(a).replace("mailto:", "") for a in attendees]
|
|
|
|
return result
|
|
|
|
|
|
def _parse_vtodo(component) -> dict | None:
|
|
"""Extract todo data from a VTODO component."""
|
|
if component.name != "VTODO":
|
|
return None
|
|
uid = str(component.get("UID", ""))
|
|
summary = str(component.get("SUMMARY", ""))
|
|
description = str(component.get("DESCRIPTION", ""))
|
|
status = str(component.get("STATUS", ""))
|
|
due = component.get("DUE")
|
|
due_str = due.dt.isoformat() if due else ""
|
|
priority = component.get("PRIORITY")
|
|
priority_val = int(priority) if priority else None
|
|
|
|
return {
|
|
"uid": uid,
|
|
"summary": summary,
|
|
"description": description,
|
|
"due": due_str,
|
|
"status": status,
|
|
"priority": priority_val,
|
|
}
|
|
|
|
|
|
def _apply_timezone(dt: datetime, timezone: str | None) -> datetime:
|
|
"""Apply a timezone to a naive datetime. Returns dt unchanged if already aware."""
|
|
if dt.tzinfo is not None:
|
|
return dt
|
|
if timezone:
|
|
return dt.replace(tzinfo=ZoneInfo(timezone))
|
|
return dt
|
|
|
|
|
|
def _build_valarm(minutes_before: int) -> icalendar.Alarm:
|
|
"""Create a DISPLAY alarm component triggered N minutes before the event."""
|
|
alarm = icalendar.Alarm()
|
|
alarm.add("action", "DISPLAY")
|
|
alarm.add("description", "Reminder")
|
|
alarm.add("trigger", timedelta(minutes=-minutes_before))
|
|
return alarm
|
|
|
|
|
|
def _add_attendees(event: icalendar.Event, attendees: list[str]) -> None:
|
|
"""Add mailto: attendees to an iCalendar event."""
|
|
for email in attendees:
|
|
attendee = icalendar.vCalAddress(f"mailto:{email}")
|
|
event.add("attendee", attendee)
|
|
|
|
|
|
def _check_config(config: dict[str, str]) -> None:
|
|
"""Raise if CalDAV is not configured."""
|
|
if not config.get("caldav_url"):
|
|
raise ValueError("CalDAV is not configured. Go to Settings → Calendar to enter your server URL.")
|
|
|
|
|
|
async def create_event(
|
|
user_id: int,
|
|
title: str,
|
|
start: str,
|
|
end: str | None = None,
|
|
duration: int | None = None,
|
|
description: str | None = None,
|
|
location: str | None = None,
|
|
all_day: bool = False,
|
|
recurrence: str | None = None,
|
|
timezone: str | None = None,
|
|
reminder_minutes: int | None = None,
|
|
attendees: list[str] | None = None,
|
|
calendar_name: str | None = None,
|
|
uid: str | None = None,
|
|
) -> dict:
|
|
"""Create a calendar event.
|
|
|
|
start/end are ISO date (YYYY-MM-DD) or datetime strings.
|
|
If all_day is True, DTSTART/DTEND use DATE values.
|
|
recurrence is an iCalendar RRULE string (e.g. "FREQ=YEARLY").
|
|
"""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
|
|
tz = timezone or config.get("caldav_timezone") or None
|
|
|
|
cal = icalendar.Calendar()
|
|
cal.add("prodid", "-//FabledAssistant//EN")
|
|
cal.add("version", "2.0")
|
|
|
|
event = icalendar.Event()
|
|
if uid:
|
|
# Remove auto-generated UID if the library added one, then inject ours
|
|
if "UID" in event:
|
|
del event["UID"]
|
|
event.add("uid", uid)
|
|
event.add("summary", title)
|
|
|
|
if all_day:
|
|
# All-day events use DATE values (no time component)
|
|
d_start = datetime.fromisoformat(start).date() if "T" in start else date_type.fromisoformat(start)
|
|
if end:
|
|
d_end = datetime.fromisoformat(end).date() if "T" in end else date_type.fromisoformat(end)
|
|
else:
|
|
d_end = d_start + timedelta(days=1)
|
|
event.add("dtstart", d_start)
|
|
event.add("dtend", d_end)
|
|
result_start = d_start.isoformat()
|
|
result_end = d_end.isoformat()
|
|
else:
|
|
dt_start = _apply_timezone(datetime.fromisoformat(start), tz)
|
|
event.add("dtstart", dt_start)
|
|
result_start = dt_start.isoformat()
|
|
if end:
|
|
dt_end = _apply_timezone(datetime.fromisoformat(end), tz)
|
|
elif duration:
|
|
dt_end = dt_start + timedelta(minutes=duration)
|
|
else:
|
|
dt_end = None
|
|
if dt_end is not None:
|
|
event.add("dtend", dt_end)
|
|
result_end = dt_end.isoformat()
|
|
else:
|
|
# Point event (no end, no duration): emit DTSTART only. Fabricating
|
|
# a 60-min DTEND here would round-trip back on the next pull as
|
|
# duration_minutes=60, silently lengthening a point event.
|
|
result_end = None
|
|
|
|
if description:
|
|
event.add("description", description)
|
|
if location:
|
|
event.add("location", location)
|
|
if recurrence:
|
|
# Parse RRULE string like "FREQ=YEARLY" into a vRecur dict
|
|
rrule_parts = {}
|
|
for part in recurrence.split(";"):
|
|
if "=" in part:
|
|
key, value = part.split("=", 1)
|
|
rrule_parts[key.strip().lower()] = value.strip()
|
|
event.add("rrule", rrule_parts)
|
|
if reminder_minutes is not None:
|
|
event.add_component(_build_valarm(reminder_minutes))
|
|
if attendees:
|
|
_add_attendees(event, attendees)
|
|
|
|
cal.add_component(event)
|
|
|
|
ical_str = cal.to_ical().decode("utf-8")
|
|
|
|
def _save():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
calendar = _get_calendar(client, cal_name)
|
|
calendar.save_event(ical_str)
|
|
|
|
await asyncio.to_thread(_save)
|
|
|
|
result = {
|
|
"title": title,
|
|
"start": result_start,
|
|
"end": result_end,
|
|
"all_day": all_day,
|
|
}
|
|
if recurrence:
|
|
result["recurrence"] = recurrence
|
|
return result
|
|
|
|
|
|
async def list_events(user_id: int, date_from: str, date_to: str) -> list[dict]:
|
|
"""List calendar events in a date range. Dates are ISO datetime strings.
|
|
|
|
Searches all calendars unless caldav_calendar_name is configured.
|
|
"""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
|
|
dt_from = datetime.fromisoformat(date_from)
|
|
dt_to = datetime.fromisoformat(date_to)
|
|
|
|
def _search():
|
|
client = _make_client(config)
|
|
cal_name = config.get("caldav_calendar_name", "")
|
|
if cal_name:
|
|
calendars = [_get_calendar(client, cal_name)]
|
|
else:
|
|
calendars = _get_all_calendars(client)
|
|
all_results = []
|
|
for calendar in calendars:
|
|
try:
|
|
all_results.extend(calendar.date_search(dt_from, dt_to))
|
|
except Exception:
|
|
logger.warning("Failed to search calendar '%s'", getattr(calendar, 'name', '?'))
|
|
return all_results
|
|
|
|
results = await asyncio.to_thread(_search)
|
|
|
|
events = []
|
|
for result in results:
|
|
cal = icalendar.Calendar.from_ical(result.data)
|
|
for component in cal.walk():
|
|
parsed = _parse_vevent(component)
|
|
if parsed:
|
|
events.append(parsed)
|
|
return events
|
|
|
|
|
|
async def search_events(user_id: int, query: str, days_ahead: int = 90) -> list[dict]:
|
|
"""Search events by keyword in the next N days."""
|
|
now = datetime.now()
|
|
date_from = now.isoformat()
|
|
date_to = (now + timedelta(days=days_ahead)).isoformat()
|
|
|
|
all_events = await list_events(user_id, date_from, date_to)
|
|
q = query.lower()
|
|
return [
|
|
e for e in all_events
|
|
if q in e["title"].lower() or q in e.get("location", "").lower() or q in e.get("description", "").lower()
|
|
]
|
|
|
|
|
|
async def update_event(
|
|
user_id: int,
|
|
query: str,
|
|
title: str | None = None,
|
|
start: str | None = None,
|
|
end: str | None = None,
|
|
description: str | None = None,
|
|
location: str | None = None,
|
|
timezone: str | None = None,
|
|
calendar_name: str | None = None,
|
|
recurrence: str | None | object = _RECURRENCE_UNSET,
|
|
) -> dict:
|
|
"""Update a calendar event matching the query.
|
|
|
|
``recurrence``: leave at the sentinel to keep the existing RRULE; pass an
|
|
RRULE string to set it, or None/"" to remove it. The push path passes the
|
|
local event's recurrence so RRULE edits propagate to the server.
|
|
"""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
tz = timezone or config.get("caldav_timezone") or None
|
|
|
|
def _do_update():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
now = datetime.now()
|
|
if cal_name:
|
|
calendars = [_get_calendar(client, cal_name)]
|
|
else:
|
|
calendars = _get_all_calendars(client)
|
|
results = []
|
|
for cal in calendars:
|
|
try:
|
|
results.extend(cal.date_search(now - timedelta(days=30), now + timedelta(days=365)))
|
|
except Exception:
|
|
logger.warning("Failed to search calendar '%s'", getattr(cal, 'name', '?'))
|
|
|
|
q = query.lower()
|
|
matches = []
|
|
for r in results:
|
|
cal_obj = icalendar.Calendar.from_ical(r.data)
|
|
for component in cal_obj.walk():
|
|
if component.name == "VEVENT":
|
|
event_title = str(component.get("SUMMARY", ""))
|
|
if q in event_title.lower():
|
|
matches.append((r, component))
|
|
|
|
if not matches:
|
|
raise ValueError(f"No event found matching '{query}'.")
|
|
if len(matches) > 3:
|
|
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
|
|
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
|
|
|
|
event_obj, component = matches[0]
|
|
if title:
|
|
component["SUMMARY"] = title
|
|
if start:
|
|
dt_start = _apply_timezone(datetime.fromisoformat(start), tz)
|
|
del component["DTSTART"]
|
|
component.add("dtstart", dt_start)
|
|
if end:
|
|
dt_end = _apply_timezone(datetime.fromisoformat(end), tz)
|
|
if "DTEND" in component:
|
|
del component["DTEND"]
|
|
component.add("dtend", dt_end)
|
|
if description is not None:
|
|
if "DESCRIPTION" in component:
|
|
del component["DESCRIPTION"]
|
|
component.add("description", description)
|
|
if location is not None:
|
|
if "LOCATION" in component:
|
|
del component["LOCATION"]
|
|
component.add("location", location)
|
|
if recurrence is not _RECURRENCE_UNSET:
|
|
# Authoritatively sync the RRULE to the local event: drop the old
|
|
# rule, then re-add if a non-empty rule was provided (else clear it).
|
|
if "RRULE" in component:
|
|
del component["RRULE"]
|
|
if recurrence:
|
|
rrule_parts = {}
|
|
for part in str(recurrence).split(";"):
|
|
if "=" in part:
|
|
key, value = part.split("=", 1)
|
|
rrule_parts[key.strip().lower()] = value.strip()
|
|
component.add("rrule", rrule_parts)
|
|
|
|
# Rebuild ical data and save
|
|
cal_data = icalendar.Calendar()
|
|
cal_data.add("prodid", "-//FabledAssistant//EN")
|
|
cal_data.add("version", "2.0")
|
|
cal_data.add_component(component)
|
|
event_obj.data = cal_data.to_ical().decode("utf-8")
|
|
event_obj.save()
|
|
|
|
return _parse_vevent(component)
|
|
|
|
return await asyncio.to_thread(_do_update)
|
|
|
|
|
|
async def delete_event(
|
|
user_id: int,
|
|
query: str,
|
|
calendar_name: str | None = None,
|
|
) -> dict:
|
|
"""Delete a calendar event matching the query."""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
|
|
def _do_delete():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
now = datetime.now()
|
|
if cal_name:
|
|
calendars = [_get_calendar(client, cal_name)]
|
|
else:
|
|
calendars = _get_all_calendars(client)
|
|
results = []
|
|
for cal in calendars:
|
|
try:
|
|
results.extend(cal.date_search(now - timedelta(days=30), now + timedelta(days=365)))
|
|
except Exception:
|
|
logger.warning("Failed to search calendar '%s'", getattr(cal, 'name', '?'))
|
|
|
|
q = query.lower()
|
|
matches = []
|
|
for r in results:
|
|
cal_obj = icalendar.Calendar.from_ical(r.data)
|
|
for component in cal_obj.walk():
|
|
if component.name == "VEVENT":
|
|
event_title = str(component.get("SUMMARY", ""))
|
|
if q in event_title.lower():
|
|
matches.append((r, component))
|
|
|
|
if not matches:
|
|
raise ValueError(f"No event found matching '{query}'.")
|
|
if len(matches) > 3:
|
|
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
|
|
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
|
|
|
|
event_obj, component = matches[0]
|
|
parsed = _parse_vevent(component)
|
|
event_obj.delete()
|
|
return parsed
|
|
|
|
return await asyncio.to_thread(_do_delete)
|
|
|
|
|
|
async def list_calendars(user_id: int) -> list[dict]:
|
|
"""List all calendars for the user."""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
|
|
def _list():
|
|
client = _make_client(config)
|
|
principal = client.principal()
|
|
calendars = principal.calendars()
|
|
return [{"name": c.name, "url": str(c.url)} for c in calendars]
|
|
|
|
return await asyncio.to_thread(_list)
|
|
|
|
|
|
async def create_todo(
|
|
user_id: int,
|
|
summary: str,
|
|
due: str | None = None,
|
|
description: str | None = None,
|
|
priority: int | None = None,
|
|
reminder_minutes: int | None = None,
|
|
timezone: str | None = None,
|
|
calendar_name: str | None = None,
|
|
) -> dict:
|
|
"""Create a CalDAV todo (VTODO)."""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
tz = timezone or config.get("caldav_timezone") or None
|
|
|
|
def _create():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
calendar = _get_calendar(client, cal_name)
|
|
|
|
kwargs = {"summary": summary}
|
|
if due:
|
|
dt_due = datetime.fromisoformat(due)
|
|
dt_due = _apply_timezone(dt_due, tz)
|
|
kwargs["due"] = dt_due
|
|
|
|
todo = calendar.save_todo(**kwargs)
|
|
|
|
# Modify component for extra fields
|
|
cal_obj = icalendar.Calendar.from_ical(todo.data)
|
|
modified = False
|
|
for component in cal_obj.walk():
|
|
if component.name == "VTODO":
|
|
if description:
|
|
component.add("description", description)
|
|
modified = True
|
|
if priority is not None:
|
|
component.add("priority", priority)
|
|
modified = True
|
|
if reminder_minutes is not None:
|
|
component.add_component(_build_valarm(reminder_minutes))
|
|
modified = True
|
|
if modified:
|
|
todo.data = cal_obj.to_ical().decode("utf-8")
|
|
todo.save()
|
|
return _parse_vtodo(component)
|
|
|
|
return {"summary": summary}
|
|
|
|
return await asyncio.to_thread(_create)
|
|
|
|
|
|
async def list_todos(
|
|
user_id: int,
|
|
include_completed: bool = False,
|
|
calendar_name: str | None = None,
|
|
) -> list[dict]:
|
|
"""List CalDAV todos."""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
|
|
def _list():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
calendar = _get_calendar(client, cal_name)
|
|
todos = calendar.todos(include_completed=include_completed)
|
|
results = []
|
|
for t in todos:
|
|
cal_obj = icalendar.Calendar.from_ical(t.data)
|
|
for component in cal_obj.walk():
|
|
parsed = _parse_vtodo(component)
|
|
if parsed:
|
|
results.append(parsed)
|
|
return results
|
|
|
|
return await asyncio.to_thread(_list)
|
|
|
|
|
|
async def search_todos(
|
|
user_id: int,
|
|
query: str,
|
|
include_completed: bool = False,
|
|
calendar_name: str | None = None,
|
|
) -> list[dict]:
|
|
"""Search CalDAV todos by keyword in summary or description."""
|
|
todos = await list_todos(user_id, include_completed=include_completed, calendar_name=calendar_name)
|
|
q = query.lower()
|
|
return [
|
|
t for t in todos
|
|
if q in t.get("summary", "").lower() or q in (t.get("description") or "").lower()
|
|
]
|
|
|
|
|
|
async def complete_todo(
|
|
user_id: int,
|
|
query: str,
|
|
calendar_name: str | None = None,
|
|
) -> dict:
|
|
"""Complete a CalDAV todo matching the query."""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
|
|
def _complete():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
calendar = _get_calendar(client, cal_name)
|
|
todos = calendar.todos(include_completed=False)
|
|
|
|
q = query.lower()
|
|
matches = []
|
|
for t in todos:
|
|
cal_obj = icalendar.Calendar.from_ical(t.data)
|
|
for component in cal_obj.walk():
|
|
if component.name == "VTODO":
|
|
s = str(component.get("SUMMARY", ""))
|
|
if q in s.lower():
|
|
matches.append((t, component))
|
|
|
|
if not matches:
|
|
raise ValueError(f"No todo found matching '{query}'.")
|
|
if len(matches) > 3:
|
|
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
|
|
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
|
|
|
|
todo_obj, component = matches[0]
|
|
todo_obj.complete()
|
|
|
|
# Re-parse after completing
|
|
cal_obj = icalendar.Calendar.from_ical(todo_obj.data)
|
|
for comp in cal_obj.walk():
|
|
parsed = _parse_vtodo(comp)
|
|
if parsed:
|
|
return parsed
|
|
return {"summary": str(component.get("SUMMARY", "")), "status": "COMPLETED"}
|
|
|
|
return await asyncio.to_thread(_complete)
|
|
|
|
|
|
async def update_todo(
|
|
user_id: int,
|
|
query: str,
|
|
summary: str | None = None,
|
|
due: str | None = None,
|
|
description: str | None = None,
|
|
priority: int | None = None,
|
|
timezone: str | None = None,
|
|
calendar_name: str | None = None,
|
|
) -> dict:
|
|
"""Update a CalDAV todo matching the query."""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
tz = timezone or config.get("caldav_timezone") or None
|
|
|
|
def _do_update():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
calendar = _get_calendar(client, cal_name)
|
|
todos = calendar.todos(include_completed=True)
|
|
|
|
q = query.lower()
|
|
matches = []
|
|
for t in todos:
|
|
cal_obj = icalendar.Calendar.from_ical(t.data)
|
|
for component in cal_obj.walk():
|
|
if component.name == "VTODO":
|
|
s = str(component.get("SUMMARY", ""))
|
|
if q in s.lower():
|
|
matches.append((t, component))
|
|
|
|
if not matches:
|
|
raise ValueError(f"No todo found matching '{query}'.")
|
|
if len(matches) > 3:
|
|
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
|
|
raise ValueError(
|
|
f"Too many matches ({len(matches)}) for '{query}'. "
|
|
f"Be more specific. Found: {', '.join(titles[:10])}"
|
|
)
|
|
|
|
todo_obj, component = matches[0]
|
|
|
|
if summary:
|
|
component["SUMMARY"] = summary
|
|
if description is not None:
|
|
if "DESCRIPTION" in component:
|
|
del component["DESCRIPTION"]
|
|
component.add("description", description)
|
|
if priority is not None:
|
|
if "PRIORITY" in component:
|
|
del component["PRIORITY"]
|
|
component.add("priority", priority)
|
|
if due:
|
|
if "DUE" in component:
|
|
del component["DUE"]
|
|
try:
|
|
dt = datetime.fromisoformat(due)
|
|
dt = _apply_timezone(dt, tz)
|
|
component.add("due", dt)
|
|
except ValueError:
|
|
component.add("due", date_type.fromisoformat(due))
|
|
|
|
# Rebuild ical data and save
|
|
cal_data = icalendar.Calendar()
|
|
cal_data.add("prodid", "-//FabledAssistant//EN")
|
|
cal_data.add("version", "2.0")
|
|
cal_data.add_component(component)
|
|
todo_obj.data = cal_data.to_ical().decode("utf-8")
|
|
todo_obj.save()
|
|
|
|
return _parse_vtodo(component)
|
|
|
|
return await asyncio.to_thread(_do_update)
|
|
|
|
|
|
async def delete_todo(
|
|
user_id: int,
|
|
query: str,
|
|
calendar_name: str | None = None,
|
|
) -> dict:
|
|
"""Delete a CalDAV todo matching the query."""
|
|
config = await get_caldav_config(user_id)
|
|
_check_config(config)
|
|
|
|
def _delete():
|
|
client = _make_client(config)
|
|
cal_name = calendar_name or config.get("caldav_calendar_name", "")
|
|
calendar = _get_calendar(client, cal_name)
|
|
todos = calendar.todos(include_completed=True)
|
|
|
|
q = query.lower()
|
|
matches = []
|
|
for t in todos:
|
|
cal_obj = icalendar.Calendar.from_ical(t.data)
|
|
for component in cal_obj.walk():
|
|
if component.name == "VTODO":
|
|
s = str(component.get("SUMMARY", ""))
|
|
if q in s.lower():
|
|
matches.append((t, component))
|
|
|
|
if not matches:
|
|
raise ValueError(f"No todo found matching '{query}'.")
|
|
if len(matches) > 3:
|
|
titles = [str(m[1].get("SUMMARY", "")) for m in matches]
|
|
raise ValueError(f"Too many matches ({len(matches)}) for '{query}'. Be more specific. Found: {', '.join(titles[:10])}")
|
|
|
|
todo_obj, component = matches[0]
|
|
parsed = _parse_vtodo(component)
|
|
todo_obj.delete()
|
|
return parsed
|
|
|
|
return await asyncio.to_thread(_delete)
|
|
|
|
|
|
async def test_connection(user_id: int) -> dict:
|
|
"""Test the CalDAV connection and return status."""
|
|
config = await get_caldav_config(user_id)
|
|
if not config.get("caldav_url"):
|
|
return {"success": False, "error": "CalDAV is not configured."}
|
|
|
|
def _test():
|
|
client = _make_client(config)
|
|
principal = client.principal()
|
|
calendars = principal.calendars()
|
|
return [c.name for c in calendars]
|
|
|
|
try:
|
|
calendar_names = await asyncio.to_thread(_test)
|
|
return {
|
|
"success": True,
|
|
"calendars": calendar_names,
|
|
"message": f"Connected successfully. Found {len(calendar_names)} calendar(s).",
|
|
}
|
|
except Exception as e:
|
|
error_msg = str(e)
|
|
if "401" in error_msg or "403" in error_msg or "Unauthorized" in error_msg:
|
|
error_msg = "Authentication failed. Check your username and password."
|
|
elif "404" in error_msg or "Not Found" in error_msg:
|
|
error_msg = "CalDAV endpoint not found. Check your URL."
|
|
elif "Connection" in error_msg or "resolve" in error_msg:
|
|
error_msg = f"Connection failed: {error_msg}"
|
|
return {"success": False, "error": error_msg}
|