326 lines
12 KiB
Python
326 lines
12 KiB
Python
import os
|
|
import caldav
|
|
import json
|
|
import traceback
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
from app.db.session import SessionLocal
|
|
from app.models import Event, Task
|
|
|
|
# Berlin timezone for display
|
|
BERLIN_TZ = ZoneInfo("Europe/Berlin")
|
|
UTC_TZ = ZoneInfo("UTC")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _get_caldav_principal():
|
|
url = os.getenv("NEXTCLOUD_URL")
|
|
username = os.getenv("NEXTCLOUD_USERNAME")
|
|
password = os.getenv("NEXTCLOUD_PASSWORD")
|
|
|
|
if not all([url, username, password]):
|
|
raise ValueError("Nextcloud credentials missing. Set NEXTCLOUD_URL, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD.")
|
|
|
|
caldav_url = url if url.endswith("/remote.php/dav/") else f"{url.rstrip('/')}/remote.php/dav/"
|
|
client = caldav.DAVClient(url=caldav_url, username=username, password=password)
|
|
return client.principal()
|
|
|
|
|
|
def _to_berlin(dt):
|
|
"""Convert a datetime to Berlin timezone. Handles naive and aware datetimes."""
|
|
if dt is None:
|
|
return None
|
|
if not isinstance(dt, datetime):
|
|
# date-only: treat as midnight Berlin time
|
|
return datetime.combine(dt, datetime.min.time()).replace(tzinfo=BERLIN_TZ)
|
|
if dt.tzinfo is None:
|
|
# Naive datetime — assume UTC
|
|
dt = dt.replace(tzinfo=UTC_TZ)
|
|
return dt.astimezone(BERLIN_TZ)
|
|
|
|
|
|
def _is_own_calendar(cal_url_str: str) -> bool:
|
|
"""Check if a calendar belongs to the authenticated user (not shared by others)."""
|
|
return "_shared_by_" not in cal_url_str
|
|
|
|
|
|
def _get_nextcloud_base_url() -> str:
|
|
"""Get the base Nextcloud URL for building links."""
|
|
url = os.getenv("NEXTCLOUD_URL", "")
|
|
return url.rstrip("/")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Calendar events
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def sync_calendar_events():
|
|
try:
|
|
principal = _get_caldav_principal()
|
|
calendars = principal.calendars()
|
|
except Exception as e:
|
|
print(f"Failed to connect to Nextcloud CalDAV: {e}")
|
|
return False
|
|
|
|
db = SessionLocal()
|
|
|
|
for calendar in calendars:
|
|
cal_url_str = str(calendar.url)
|
|
if not _is_own_calendar(cal_url_str):
|
|
print(f"Skipping shared calendar: {calendar.get_display_name()}")
|
|
continue
|
|
|
|
try:
|
|
events = calendar.events()
|
|
for event in events:
|
|
ical = event.icalendar_component
|
|
if ical.name == "VEVENT":
|
|
title = str(ical.get("SUMMARY", "Untitled Event"))
|
|
dtstart = ical.get("DTSTART")
|
|
dtend = ical.get("DTEND")
|
|
|
|
if not dtstart or not dtend:
|
|
continue
|
|
|
|
start_time = _to_berlin(dtstart.dt)
|
|
end_time = _to_berlin(dtend.dt)
|
|
|
|
existing = db.query(Event).filter(Event.title == title, Event.source == "nextcloud").first()
|
|
if not existing:
|
|
db.add(Event(title=title, start_time=start_time, end_time=end_time, source="nextcloud"))
|
|
except Exception as e:
|
|
print(f"Error reading calendar {calendar}: {e}")
|
|
|
|
db.commit()
|
|
db.close()
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToDo sync (two-way, only own tasks)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _parse_vtodo_datetime(val):
|
|
"""Parse a VTODO datetime and convert to Berlin timezone."""
|
|
if val is None:
|
|
return None
|
|
return _to_berlin(val.dt)
|
|
|
|
|
|
def _has_local_changes(task, remote_title: str, remote_description: str, remote_status: str) -> bool:
|
|
"""Check if the local task differs from what was last synced (i.e., user edited locally)."""
|
|
# If never synced, treat as no local changes (first sync)
|
|
if task.last_synced_at is None:
|
|
return False
|
|
# If the task was modified after last sync, local has changes
|
|
# We compare local fields against remote to detect divergence
|
|
local_title = task.title or ""
|
|
local_desc = task.description or ""
|
|
local_status = task.status or "open"
|
|
# If local differs from remote, and we have a last_synced_at, local was changed
|
|
if local_title != remote_title:
|
|
return True
|
|
if local_desc != (remote_description or ""):
|
|
return True
|
|
if local_status != remote_status:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _has_remote_changes(task, remote_title: str, remote_description: str, remote_status: str) -> bool:
|
|
"""Check if remote data differs from what's currently stored locally.
|
|
|
|
This detects whether the remote side changed since our last sync.
|
|
If the remote values differ from what we have locally, remote changed.
|
|
"""
|
|
local_title = task.title or ""
|
|
local_desc = task.description or ""
|
|
local_status = task.status or "open"
|
|
if local_title != remote_title:
|
|
return True
|
|
if local_desc != (remote_description or ""):
|
|
return True
|
|
if local_status != remote_status:
|
|
return True
|
|
return False
|
|
|
|
|
|
def sync_todo_tasks():
|
|
"""Import VTODOs from own Nextcloud todo lists into local Task table.
|
|
|
|
Only imports from calendars owned by the authenticated user (skips shared).
|
|
Detects conflicts when both local and remote changed since last sync.
|
|
"""
|
|
try:
|
|
principal = _get_caldav_principal()
|
|
calendars = principal.calendars()
|
|
except Exception as e:
|
|
print(f"Failed to connect to Nextcloud CalDAV: {e}")
|
|
return False
|
|
|
|
db = SessionLocal()
|
|
nc_username = os.getenv("NEXTCLOUD_USERNAME", "")
|
|
base_url = _get_nextcloud_base_url()
|
|
|
|
for calendar in calendars:
|
|
cal_url_str = str(calendar.url)
|
|
cal_name = calendar.get_display_name() or "Tasks"
|
|
|
|
# Skip shared calendars (these belong to coworkers)
|
|
if not _is_own_calendar(cal_url_str):
|
|
print(f"Skipping shared todo list: {cal_name}")
|
|
continue
|
|
|
|
try:
|
|
todos = calendar.todos(include_completed=True)
|
|
except Exception:
|
|
continue # calendar may not support VTODOs
|
|
|
|
for todo in todos:
|
|
try:
|
|
ical = todo.icalendar_component
|
|
if ical.name != "VTODO":
|
|
continue
|
|
|
|
uid = str(ical.get("UID", ""))
|
|
if not uid:
|
|
continue
|
|
|
|
title = str(ical.get("SUMMARY", "Untitled Task"))
|
|
description = str(ical.get("DESCRIPTION", "") or "")
|
|
due_date = _parse_vtodo_datetime(ical.get("DUE"))
|
|
vtodo_status = str(ical.get("STATUS", "NEEDS-ACTION")).upper()
|
|
status = "completed" if vtodo_status == "COMPLETED" else "open"
|
|
|
|
# Build origin label with calendar name
|
|
origin_label = f"nextcloud_todo:{cal_name}"
|
|
|
|
# Build link to Nextcloud tasks app
|
|
source_url = f"{base_url}/apps/tasks/#/calendars/{cal_name}" if base_url else None
|
|
|
|
existing = db.query(Task).filter(Task.source_id == uid).first()
|
|
if existing:
|
|
# Don't touch dismissed or completed tasks (user already handled them)
|
|
if existing.status in ("dismissed", "completed"):
|
|
continue
|
|
|
|
# --- Conflict detection ---
|
|
# Check if local was modified since last sync
|
|
local_changed = _has_local_changes(existing, title, description, status)
|
|
remote_changed = _has_remote_changes(existing, title, description, status)
|
|
|
|
if local_changed and remote_changed:
|
|
# Both sides changed → conflict
|
|
existing.has_conflict = True
|
|
existing.conflict_data = json.dumps({
|
|
"title": title,
|
|
"status": status,
|
|
"description": description,
|
|
})
|
|
continue
|
|
|
|
if not remote_changed:
|
|
# Only local changed (or nothing changed) → push local to remote if needed
|
|
if local_changed and existing.source_id:
|
|
push_updates = {}
|
|
if existing.title != title:
|
|
push_updates["title"] = existing.title
|
|
if existing.description != (description or ""):
|
|
push_updates["description"] = existing.description
|
|
if existing.status != status:
|
|
push_updates["status"] = existing.status
|
|
if push_updates:
|
|
update_remote_todo(existing.source_id, push_updates)
|
|
existing.last_synced_at = datetime.now(tz=UTC_TZ)
|
|
continue
|
|
|
|
# Only remote changed → update local normally
|
|
# Don't revert in_progress or review tasks back to open
|
|
if existing.status in ("in_progress", "review") and status == "open":
|
|
existing.title = title
|
|
existing.description = description
|
|
existing.due_date = due_date
|
|
existing.origin = origin_label
|
|
existing.source_url = source_url
|
|
existing.last_synced_at = datetime.now(tz=UTC_TZ)
|
|
continue
|
|
|
|
existing.title = title
|
|
existing.description = description
|
|
existing.due_date = due_date
|
|
existing.status = status
|
|
existing.origin = origin_label
|
|
existing.source_url = source_url
|
|
existing.last_synced_at = datetime.now(tz=UTC_TZ)
|
|
else:
|
|
max_pos = db.query(Task).count()
|
|
db.add(Task(
|
|
title=title,
|
|
description=description,
|
|
priority=4,
|
|
position=max_pos,
|
|
status=status,
|
|
origin=origin_label,
|
|
source_id=uid,
|
|
source_url=source_url,
|
|
due_date=due_date,
|
|
last_synced_at=datetime.now(tz=UTC_TZ),
|
|
))
|
|
except Exception as e:
|
|
print(f"Error processing VTODO: {e}")
|
|
|
|
db.commit()
|
|
db.close()
|
|
return True
|
|
|
|
|
|
def update_remote_todo(source_id: str, updates: dict):
|
|
"""Push local task changes back to the Nextcloud server.
|
|
|
|
updates: dict with any of: title, description, status
|
|
"""
|
|
try:
|
|
principal = _get_caldav_principal()
|
|
calendars = principal.calendars()
|
|
except Exception as e:
|
|
print(f"Failed to connect to Nextcloud CalDAV for push: {e}")
|
|
return False
|
|
|
|
for calendar in calendars:
|
|
cal_url_str = str(calendar.url)
|
|
# Only push to own calendars
|
|
if not _is_own_calendar(cal_url_str):
|
|
continue
|
|
|
|
try:
|
|
todos = calendar.todos(include_completed=True)
|
|
except Exception:
|
|
continue
|
|
|
|
for todo in todos:
|
|
try:
|
|
ical = todo.icalendar_component
|
|
if ical.name != "VTODO":
|
|
continue
|
|
if str(ical.get("UID", "")) != source_id:
|
|
continue
|
|
|
|
# Apply updates
|
|
if "title" in updates:
|
|
ical["SUMMARY"] = updates["title"]
|
|
if "description" in updates:
|
|
ical["DESCRIPTION"] = updates["description"]
|
|
if "status" in updates:
|
|
ical["STATUS"] = "COMPLETED" if updates["status"] == "completed" else "NEEDS-ACTION"
|
|
|
|
todo.save()
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error updating remote VTODO {source_id}: {e}")
|
|
|
|
print(f"Remote VTODO with UID {source_id} not found.")
|
|
return False
|