|
@@ -1,5 +1,5 @@
|
|
|
import urllib.request
|
|
import urllib.request
|
|
|
-from datetime import date, datetime, timedelta
|
|
|
|
|
|
|
+from datetime import date, datetime
|
|
|
|
|
|
|
|
from room_cal._utils import CalendarEntry, EventSource, get_config
|
|
from room_cal._utils import CalendarEntry, EventSource, get_config
|
|
|
from room_cal.store import EventRecord
|
|
from room_cal.store import EventRecord
|
|
@@ -8,8 +8,6 @@ _source = EventSource('ical_feeds')
|
|
|
|
|
|
|
|
|
|
|
|
|
_TIMEOUT = 15
|
|
_TIMEOUT = 15
|
|
|
-_DEFAULT_PAST_DAYS = 365
|
|
|
|
|
-_DEFAULT_FUTURE_DAYS = 730
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _webcal_to_https(url: str) -> str:
|
|
def _webcal_to_https(url: str) -> str:
|
|
@@ -58,35 +56,8 @@ def _parse_ical(raw_bytes: bytes, calendar_name: str) -> list[EventRecord]:
|
|
|
return events
|
|
return events
|
|
|
|
|
|
|
|
|
|
|
|
|
-async def _fetch_caldav(entry: CalendarEntry) -> list[EventRecord]:
|
|
|
|
|
- from room_cal.store import get_calendar
|
|
|
|
|
-
|
|
|
|
|
- calendar_name = entry.caldav
|
|
|
|
|
- today = datetime.today().date()
|
|
|
|
|
- start_date = today - timedelta(days=entry.past_days if entry.past_days is not None else _DEFAULT_PAST_DAYS)
|
|
|
|
|
- end_date = today + timedelta(days=entry.future_days if entry.future_days is not None else _DEFAULT_FUTURE_DAYS)
|
|
|
|
|
-
|
|
|
|
|
- cal = get_calendar(calendar_name)
|
|
|
|
|
- raw_events = await cal.fetch_date_range(start_date, end_date)
|
|
|
|
|
- return [
|
|
|
|
|
- EventRecord(
|
|
|
|
|
- uid=e['uid'],
|
|
|
|
|
- start=e['start'],
|
|
|
|
|
- end=e['end'],
|
|
|
|
|
- all_day=e['all_day'],
|
|
|
|
|
- title=e['title'],
|
|
|
|
|
- description=e['description'],
|
|
|
|
|
- x_data=e.get('x_data'),
|
|
|
|
|
- _meta={'calendar': entry.name},
|
|
|
|
|
- )
|
|
|
|
|
- for e in raw_events
|
|
|
|
|
- ]
|
|
|
|
|
-
|
|
|
|
|
-
|
|
|
|
|
async def fetch_calendar(entry: CalendarEntry) -> list[EventRecord]:
|
|
async def fetch_calendar(entry: CalendarEntry) -> list[EventRecord]:
|
|
|
- """Fetch one CalendarEntry (with either url or caldav) and return its events."""
|
|
|
|
|
- if entry.caldav is not None:
|
|
|
|
|
- return await _fetch_caldav(entry)
|
|
|
|
|
|
|
+ """Fetch one CalendarEntry by URL and return its events."""
|
|
|
url = _webcal_to_https(entry.url)
|
|
url = _webcal_to_https(entry.url)
|
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'room_cal/1.0'})
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'room_cal/1.0'})
|
|
|
with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
|
|
with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
|