135 lines
3.2 KiB
Python
135 lines
3.2 KiB
Python
import logging
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from functools import total_ordering
|
|
from typing import Iterator
|
|
|
|
from caldav import Calendar
|
|
from caldav.lib.error import ReportError
|
|
from pydantic import BaseModel, validator
|
|
from vobject.icalendar import VEvent
|
|
|
|
from . import caldav_principal, get_ttl_hash, run_in_executor, timed_alru_cache
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _string_strip(in_str: str) -> str:
|
|
return in_str.strip()
|
|
|
|
|
|
@total_ordering
|
|
class CalEvent(BaseModel):
|
|
summary: str = ""
|
|
description: str = ""
|
|
dtstart: datetime = datetime.utcnow()
|
|
dtend: datetime = datetime.utcnow()
|
|
|
|
class Config:
|
|
frozen = True
|
|
|
|
def __lt__(self, other: "CalEvent") -> bool:
|
|
return self.dtstart < other.dtstart
|
|
|
|
def __eq__(self, other: "CalEvent") -> bool:
|
|
return self.dict() == other.dict()
|
|
|
|
_validate_summary = validator(
|
|
"summary",
|
|
allow_reuse=True,
|
|
)(_string_strip)
|
|
|
|
_validate_description = validator(
|
|
"description",
|
|
allow_reuse=True,
|
|
)(_string_strip)
|
|
|
|
@classmethod
|
|
def from_vevent(cls, event: VEvent) -> "CalEvent":
|
|
data = {}
|
|
|
|
for key in cls().dict().keys():
|
|
try:
|
|
data[key] = event.contents[key][0].value
|
|
|
|
except KeyError:
|
|
pass
|
|
|
|
return cls.parse_obj(data)
|
|
|
|
|
|
@timed_alru_cache(maxsize=20)
|
|
async def _get_calendar(
|
|
calendar_name: str,
|
|
) -> Calendar:
|
|
|
|
@run_in_executor
|
|
def _inner() -> Calendar:
|
|
return caldav_principal().calendar(calendar_name)
|
|
|
|
return await _inner()
|
|
|
|
|
|
@timed_alru_cache(maxsize=20)
|
|
async def _get_calendar_events(
|
|
calendar_name: str,
|
|
) -> list[CalEvent]:
|
|
|
|
@run_in_executor
|
|
def _inner() -> Iterator[VEvent]:
|
|
_logger.info(f"updating {calendar_name}")
|
|
print(f"updating {calendar_name}")
|
|
|
|
calendar = caldav_principal().calendar(calendar_name)
|
|
|
|
date_start = datetime.utcnow().date()
|
|
time_min = datetime.min.time()
|
|
dt_start = datetime.combine(date_start, time_min)
|
|
dt_end = dt_start + timedelta(days=365)
|
|
|
|
try:
|
|
search_result = calendar.date_search(
|
|
start=dt_start,
|
|
end=dt_end,
|
|
expand=True,
|
|
verify_expand=True,
|
|
)
|
|
|
|
except ReportError:
|
|
_logger.warn("CalDAV server does not support expanded search")
|
|
|
|
search_result = calendar.date_search(
|
|
start=dt_start,
|
|
end=dt_end,
|
|
expand=False,
|
|
)
|
|
|
|
return (
|
|
vevent
|
|
for event in search_result
|
|
for vevent in event.vobject_instance.contents["vevent"]
|
|
)
|
|
|
|
return sorted([
|
|
CalEvent.from_vevent(vevent)
|
|
for vevent in await _inner()
|
|
])
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DavCalendar:
|
|
calendar_name: str
|
|
|
|
@property
|
|
async def calendar(self) -> Calendar:
|
|
return await _get_calendar(
|
|
ttl_hash=get_ttl_hash(20),
|
|
calendar_name=self.calendar_name,
|
|
)
|
|
|
|
@property
|
|
async def events(self) -> list[CalEvent]:
|
|
return await _get_calendar_events(
|
|
ttl_hash=get_ttl_hash(20),
|
|
calendar_name=self.calendar_name,
|
|
)
|