374 lines
12 KiB
Python
374 lines
12 KiB
Python
"""Khal CLI client for calendar operations.
|
|
|
|
This module provides a client that uses the khal CLI tool to interact with
|
|
calendar data stored in vdir format.
|
|
"""
|
|
|
|
import subprocess
|
|
import logging
|
|
from datetime import datetime, date, timedelta
|
|
from typing import Optional, List
|
|
|
|
from src.calendar.backend import CalendarBackend, Event
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class KhalClient(CalendarBackend):
|
|
"""Calendar backend using khal CLI."""
|
|
|
|
def __init__(self, config_path: Optional[str] = None):
|
|
"""Initialize the Khal client.
|
|
|
|
Args:
|
|
config_path: Optional path to khal config file
|
|
"""
|
|
self.config_path = config_path
|
|
|
|
def _run_khal(
|
|
self, args: List[str], capture_output: bool = True
|
|
) -> subprocess.CompletedProcess:
|
|
"""Run a khal command.
|
|
|
|
Args:
|
|
args: Command arguments (after 'khal')
|
|
capture_output: Whether to capture stdout/stderr
|
|
|
|
Returns:
|
|
CompletedProcess result
|
|
"""
|
|
cmd = ["khal"] + args
|
|
if self.config_path:
|
|
cmd.extend(["-c", self.config_path])
|
|
|
|
logger.debug(f"Running khal command: {' '.join(cmd)}")
|
|
|
|
return subprocess.run(
|
|
cmd,
|
|
capture_output=capture_output,
|
|
text=True,
|
|
)
|
|
|
|
def _parse_event_line(
|
|
self, line: str, day_header_date: Optional[date] = None
|
|
) -> Optional[Event]:
|
|
"""Parse a single event line from khal list output.
|
|
|
|
Expected format: title|start-time|end-time|start|end|location|uid|description|organizer|url|categories|status|recurring
|
|
|
|
Args:
|
|
line: The line to parse
|
|
day_header_date: Current day being parsed (from day headers)
|
|
|
|
Returns:
|
|
Event if successfully parsed, None otherwise
|
|
"""
|
|
# Skip empty lines and day headers
|
|
if not line or "|" not in line:
|
|
return None
|
|
|
|
parts = line.split("|")
|
|
if len(parts) < 5:
|
|
return None
|
|
|
|
try:
|
|
title = parts[0].strip()
|
|
start_str = parts[3].strip() # Full datetime
|
|
end_str = parts[4].strip() # Full datetime
|
|
location = parts[5].strip() if len(parts) > 5 else ""
|
|
uid = parts[6].strip() if len(parts) > 6 else ""
|
|
description = parts[7].strip() if len(parts) > 7 else ""
|
|
organizer = parts[8].strip() if len(parts) > 8 else ""
|
|
url = parts[9].strip() if len(parts) > 9 else ""
|
|
categories = parts[10].strip() if len(parts) > 10 else ""
|
|
status = parts[11].strip() if len(parts) > 11 else ""
|
|
recurring_symbol = parts[12].strip() if len(parts) > 12 else ""
|
|
|
|
# Parse datetimes (format: YYYY-MM-DD HH:MM)
|
|
start = datetime.strptime(start_str, "%Y-%m-%d %H:%M")
|
|
end = datetime.strptime(end_str, "%Y-%m-%d %H:%M")
|
|
|
|
# Check for all-day events (typically start at 00:00 and end at 00:00 next day)
|
|
all_day = (
|
|
start.hour == 0
|
|
and start.minute == 0
|
|
and end.hour == 0
|
|
and end.minute == 0
|
|
and (end.date() - start.date()).days >= 1
|
|
)
|
|
|
|
# Check if event is recurring (repeat symbol is typically a loop arrow)
|
|
recurring = bool(recurring_symbol)
|
|
|
|
return Event(
|
|
uid=uid or f"{title}_{start_str}",
|
|
title=title,
|
|
start=start,
|
|
end=end,
|
|
location=location,
|
|
description=description,
|
|
organizer=organizer,
|
|
url=url,
|
|
categories=categories,
|
|
status=status,
|
|
all_day=all_day,
|
|
recurring=recurring,
|
|
)
|
|
except (ValueError, IndexError) as e:
|
|
logger.warning(f"Failed to parse event line '{line}': {e}")
|
|
return None
|
|
|
|
def get_events(
|
|
self,
|
|
start_date: date,
|
|
end_date: date,
|
|
calendar: Optional[str] = None,
|
|
) -> List[Event]:
|
|
"""Get events in a date range.
|
|
|
|
Args:
|
|
start_date: Start of range (inclusive)
|
|
end_date: End of range (inclusive)
|
|
calendar: Optional calendar name to filter by
|
|
|
|
Returns:
|
|
List of events in the range, sorted by start time
|
|
"""
|
|
# Format dates for khal
|
|
start_str = start_date.strftime("%Y-%m-%d")
|
|
# Add one day to end_date to make it inclusive
|
|
end_dt = end_date + timedelta(days=1)
|
|
end_str = end_dt.strftime("%Y-%m-%d")
|
|
|
|
# Build command
|
|
# Format: title|start-time|end-time|start|end|location|uid|description|organizer|url|categories|status|recurring
|
|
format_str = "{title}|{start-time}|{end-time}|{start}|{end}|{location}|{uid}|{description}|{organizer}|{url}|{categories}|{status}|{repeat-symbol}"
|
|
args = ["list", "-f", format_str, start_str, end_str]
|
|
|
|
if calendar:
|
|
args.extend(["-a", calendar])
|
|
|
|
result = self._run_khal(args)
|
|
|
|
if result.returncode != 0:
|
|
logger.error(f"khal list failed: {result.stderr}")
|
|
return []
|
|
|
|
events = []
|
|
current_day: Optional[date] = None
|
|
|
|
for line in result.stdout.strip().split("\n"):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
# Check for day headers (e.g., "Today, 2025-12-18" or "Monday, 2025-12-22")
|
|
if ", " in line and "|" not in line:
|
|
try:
|
|
# Extract date from header
|
|
date_part = line.split(", ")[-1]
|
|
current_day = datetime.strptime(date_part, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
pass
|
|
continue
|
|
|
|
event = self._parse_event_line(line, current_day)
|
|
if event:
|
|
events.append(event)
|
|
|
|
# Sort by start time
|
|
events.sort(key=lambda e: e.start)
|
|
|
|
return events
|
|
|
|
def get_event(self, uid: str) -> Optional[Event]:
|
|
"""Get a single event by UID.
|
|
|
|
Args:
|
|
uid: Event unique identifier
|
|
|
|
Returns:
|
|
Event if found, None otherwise
|
|
"""
|
|
# khal doesn't have a direct "get by uid" command
|
|
# We search for it instead
|
|
result = self._run_khal(["search", uid])
|
|
|
|
if result.returncode != 0 or not result.stdout.strip():
|
|
return None
|
|
|
|
# Parse the first result
|
|
# Search output format is different, so we need to handle it
|
|
lines = result.stdout.strip().split("\n")
|
|
if lines:
|
|
# For now, return None - would need more parsing
|
|
# This is a limitation of khal's CLI
|
|
return None
|
|
|
|
return None
|
|
|
|
def get_calendars(self) -> List[str]:
|
|
"""Get list of available calendar names.
|
|
|
|
Returns:
|
|
List of calendar names
|
|
"""
|
|
result = self._run_khal(["printcalendars"])
|
|
|
|
if result.returncode != 0:
|
|
logger.error(f"khal printcalendars failed: {result.stderr}")
|
|
return []
|
|
|
|
calendars = []
|
|
for line in result.stdout.strip().split("\n"):
|
|
line = line.strip()
|
|
if line:
|
|
calendars.append(line)
|
|
|
|
return calendars
|
|
|
|
def create_event(
|
|
self,
|
|
title: str,
|
|
start: datetime,
|
|
end: datetime,
|
|
calendar: Optional[str] = None,
|
|
location: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
all_day: bool = False,
|
|
) -> Event:
|
|
"""Create a new event.
|
|
|
|
Args:
|
|
title: Event title
|
|
start: Start datetime
|
|
end: End datetime
|
|
calendar: Calendar to add event to
|
|
location: Event location
|
|
description: Event description
|
|
all_day: Whether this is an all-day event
|
|
|
|
Returns:
|
|
The created event
|
|
"""
|
|
# Build khal new command
|
|
# Format: khal new [-a calendar] start end title [:: description] [-l location]
|
|
if all_day:
|
|
start_str = start.strftime("%Y-%m-%d")
|
|
end_str = end.strftime("%Y-%m-%d")
|
|
else:
|
|
start_str = start.strftime("%Y-%m-%d %H:%M")
|
|
end_str = end.strftime("%H:%M") # End time only if same day
|
|
if end.date() != start.date():
|
|
end_str = end.strftime("%Y-%m-%d %H:%M")
|
|
|
|
args = ["new"]
|
|
if calendar:
|
|
args.extend(["-a", calendar])
|
|
if location:
|
|
args.extend(["-l", location])
|
|
|
|
args.extend([start_str, end_str, title])
|
|
|
|
if description:
|
|
args.extend(["::", description])
|
|
|
|
result = self._run_khal(args)
|
|
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Failed to create event: {result.stderr}")
|
|
|
|
# Return a constructed event (khal doesn't return the created event)
|
|
return Event(
|
|
uid=f"new_{title}_{start.isoformat()}",
|
|
title=title,
|
|
start=start,
|
|
end=end,
|
|
location=location or "",
|
|
description=description or "",
|
|
calendar=calendar or "",
|
|
all_day=all_day,
|
|
)
|
|
|
|
def delete_event(self, uid: str) -> bool:
|
|
"""Delete an event.
|
|
|
|
Args:
|
|
uid: Event unique identifier
|
|
|
|
Returns:
|
|
True if deleted successfully
|
|
"""
|
|
# khal edit with --delete flag
|
|
# This is tricky because khal edit is interactive
|
|
# We might need to use khal's Python API directly for this
|
|
logger.warning("delete_event not fully implemented for khal CLI")
|
|
return False
|
|
|
|
def update_event(
|
|
self,
|
|
uid: str,
|
|
title: Optional[str] = None,
|
|
start: Optional[datetime] = None,
|
|
end: Optional[datetime] = None,
|
|
location: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
) -> Optional[Event]:
|
|
"""Update an existing event.
|
|
|
|
Args:
|
|
uid: Event unique identifier
|
|
title: New title (if provided)
|
|
start: New start time (if provided)
|
|
end: New end time (if provided)
|
|
location: New location (if provided)
|
|
description: New description (if provided)
|
|
|
|
Returns:
|
|
Updated event if successful, None otherwise
|
|
"""
|
|
# khal edit is interactive, so this is limited via CLI
|
|
logger.warning("update_event not fully implemented for khal CLI")
|
|
return None
|
|
|
|
def search_events(self, query: str) -> List[Event]:
|
|
"""Search for events matching a query string.
|
|
|
|
Args:
|
|
query: Search string to match against event titles and descriptions
|
|
|
|
Returns:
|
|
List of matching events
|
|
"""
|
|
if not query:
|
|
return []
|
|
|
|
# Use khal search with custom format
|
|
format_str = "{title}|{start-time}|{end-time}|{start}|{end}|{location}|{uid}|{description}|{organizer}|{url}|{categories}|{status}|{repeat-symbol}"
|
|
args = ["search", "-f", format_str, query]
|
|
|
|
result = self._run_khal(args)
|
|
|
|
if result.returncode != 0:
|
|
logger.error(f"khal search failed: {result.stderr}")
|
|
return []
|
|
|
|
events = []
|
|
for line in result.stdout.strip().split("\n"):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
# Skip day headers
|
|
if ", " in line and "|" not in line:
|
|
continue
|
|
|
|
event = self._parse_event_line(line)
|
|
if event:
|
|
events.append(event)
|
|
|
|
# Sort by start time
|
|
events.sort(key=lambda e: e.start)
|
|
|
|
return events
|