Commit 02f97591 authored by HgO's avatar HgO

refactor agenda, rename WikiCalendar to Wiki, add ICalendar, CalDav and NextCloud calendar classes

parent c6d881b1
import json
import logging
import pytz
import os
import requests
import sys
import urllib.parse
from datetime import datetime
from datetime import date, datetime, timedelta
try:
import icalendar
import caldav
except ImportError:
pass
from mastodon import MastodonError
import config
......@@ -22,73 +27,76 @@ def load_class(string):
return cls
def localize(dt, timezone = None):
if timezone is None:
timezone = datetime.now().astimezone().tzinfo
if isinstance(dt, int):
dt = datetime.fromtimestamp(dt)
return dt.replace(tzinfo=None) - timezone.utcoffset(dt)
if type(dt) is date:
return datetime.combine(dt, datetime.min.time())
return dt.replace(tzinfo=None) + timezone.utcoffset(dt)
class Calendar:
def __init__(self, url):
self.url = url
def query(self):
return self.url
def get(self):
raise NotImplementedError
def events(self):
events = map(self.create_event, self.get())
return sorted(event for event in events
if event.start_date and event.start_date > datetime.now())
@staticmethod
def create_event(item):
raise NotImplementedError
class WikiCalendar(Calendar):
SEMANTIC_QUERY = (
'[[Has event type::+]] [[Has date::>{}]]' \
.format(datetime.now().isoformat()),
'?Has title=title',
'?Has date=start_date',
'?Has end date=end_date',
'?Has page=page',
'?Has location=location',
'?In topic=topics',
'?Organized by=workgroups',
'?Has event type=type',
'sort=Has date',
'order=asc',
'limit=10'
)
API_QUERY = '/api.php?action=ask&format=json' \
+ '&api_version=3&formatversion=latest&utf8=1&query='
class Wiki(Calendar):
def __init__(self, url):
super().__init__(url)
self.SEMANTIC_QUERY = (
'[[Has event type::+]] [[Has date::>{}]]' \
.format(datetime.now().isoformat()),
'?Has title=title',
'?Has date=start_date',
'?Has end date=end_date',
'?Has page=page',
'?Has location=location',
'?In topic=topics',
'?Organized by=workgroups',
'?Has event type=type',
'sort=Has date',
'order=asc',
'limit=10'
)
self.API_QUERY = '/api.php?action=ask&format=json' \
+ '&api_version=3&formatversion=latest&utf8=1&query='
def query(self):
q = self.url + self.API_QUERY \
return self.url + self.API_QUERY \
+ urllib.parse.quote('|'.join(self.SEMANTIC_QUERY))
def get(self):
response = requests.get(self.query())
response = requests.get(q)
if response.status_code != 200:
return
return json.loads(response.text)
def events(self):
response = self.query()
if response is None:
return
pages = response['query']['results']
for page in pages:
event = self._create_event(page)
response_json = json.loads(response.text)
return iter(response_json['query']['results'])
if not event.title:
logging.warn("{event} - Event has no title".format(event=event.url))
continue
if not event.start_date:
logging.warn("{event} - Event has no date".format(
event=event.url))
continue
if cache.is_same(event):
continue
yield event
@staticmethod
def _create_event(page):
def create_event(page):
page = next(iter(page.values()))
details = page['printouts']
......@@ -109,14 +117,12 @@ class WikiCalendar(Calendar):
title = None
try:
start_timestamp = int(details['start_date'][0]['timestamp'])
start_date = WikiCalendar._datefromtimestamp(start_timestamp)
start_date = localize(int(details['start_date'][0]['timestamp']))
except IndexError:
start_date = None
try:
end_timestamp = int(details['end_date'][0]['timestamp'])
end_date = WikiCalendar._datefromtimestamp(end_timestamp)
end_date = localize(int(details['end_date'][0]['timestamp']))
except IndexError:
end_date = None
......@@ -132,12 +138,105 @@ class WikiCalendar(Calendar):
return model.data.Event(id, url, title, start_date, end_date, location,
icon, topics=topics, workgroups=workgroups)
class ICalendar(Calendar):
def query(self):
return self.url
def get(self):
response = requests.get(self.query())
if response.status_code != 200:
return
yield from self._vevents(response.content)
@staticmethod
def _vevents(content):
ical = icalendar.Calendar.from_ical(content)
for component in ical.walk():
if component.name == 'VEVENT':
yield component
@staticmethod
def _datefromtimestamp(timestamp, timezone = pytz.timezone('Europe/Brussels')):
date = datetime.fromtimestamp(timestamp)
return date - timezone.utcoffset(date)
def create_event(vevent):
id = vevent['uid']
url = None
title = vevent.get('summary')
start_date = localize(vevent['dtstart'].dt)
end_date = vevent.get('dtend')
if not end_date:
end_date = start_date
else:
end_date = localize(end_date.dt)
location = vevent.get('location')
return model.data.Event(id, url, title, start_date, end_date, location)
class CalDav(ICalendar):
def __init__(self, url):
super().__init__(url)
calendar_config = config.parser['calendar']
self.client = caldav.DAVClient(url,
username=calendar_config['user'],
password=calendar_config['password'])
def get(self):
start_date = datetime.now()
calendar = self.client.principal().calendar(cal_id=self.url)
content = calendar.date_search(start_date)
for event_dav in content:
yield from self._vevents(event_dav.data)
class NextCloud(ICalendar):
def __init__(self, url, range = None):
super().__init__(url)
if range is None:
range = timedelta(31)
self._range = range
def query(self):
start_date = datetime.now()
end_date = start_date + self._range
query = 'export&start={start}&end={end}&componentType=VEVENT&expand=1'.format(
start=int(start_date.timestamp()),
end=int(end_date.timestamp()))
scheme, netloc, path, _, _ = urllib.parse.urlsplit(self.url)
if '/dav/' not in self.url:
path = os.path.dirname(path)
path = path.replace('index.php', 'remote.php')
path = path.replace('/apps/calendar/p/', '/dav/public-calendars/')
if '/dav/public-calendars/' not in path:
return
return '{scheme}://{netloc}/{path}?{query}'.format(
scheme=scheme,
netloc=netloc,
path=path,
query=query)
def get(self):
query = self.query()
if not query:
return CalDav(self.url).get()
return ICalendar(query).get()
db = model.connect()
cache = model.table.Cache(db)
......@@ -145,6 +244,13 @@ cache = model.table.Cache(db)
calendar_cls = load_class(config.parser['calendar']['class'])
calendar = calendar_cls(config.parser['calendar']['url'])
for event in calendar.events():
if not event.title:
logging.warn("{event} - Event has no title".format(event=event.url))
continue
if cache.is_same(event):
continue
print(str(event))
try:
......
......@@ -12,4 +12,4 @@ path = parbot.log
[calendar]
url = https://wiki.pirateparty.be
class = WikiCalendar
class = Wiki
......@@ -205,8 +205,9 @@ class Event:
details.append(self.location)
details.append('')
details.append(self.url)
details.append('')
if self.url:
details.append(self.url)
details.append('')
if len(self.workgroups):
details.append(i18n.t('event.organized_by'))
......@@ -238,7 +239,7 @@ class Event:
if minutes == 0:
minutes = ''
else:
minutes = '{0>2}'.format(minutes)
minutes = '{:0>2}'.format(minutes)
if hours > 0:
date_str += ' ' + i18n.t('event.duration').format(
......@@ -247,6 +248,10 @@ class Event:
return date_str
def __hash__(self):
return hash((self.id, self.url, self.title,
self.start_date, self.end_date, self.location, self.icon))
def __eq__(self, other):
other_vars = vars(other)
for var, value in vars(self).items():
......@@ -257,3 +262,9 @@ class Event:
def __ne__(self, other):
return not (self == other)
def __lt__(self, other):
return self.start_date < other.start_date
def __gt__(self, other):
return not (self < other)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment