feat: group private apps into andreknie-privat
This commit is contained in:
@@ -0,0 +1,325 @@
|
||||
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
|
||||
Reference in New Issue
Block a user