Browse Source

Added feature to filter events from a calendar via phylter

master
Johann Schmitz 5 years ago
parent
commit
87a41c52fc
Signed by: ercpe GPG Key ID: A084064277C501ED
  1. 11
      doc/maxd.cfg
  2. 3
      requirements.txt
  3. 3
      src/maxd/config.py
  4. 26
      src/maxd/worker.py
  5. 19
      tests/test_worker.py

11
doc/maxd.cfg

@ -42,13 +42,14 @@
# Example of a calendar definition.
# [cal1]
# # url can either be an url (http:// or https://) or a path to a local file. Please note, that max does not
# # support CalDAV (yet).
# url can either be an url (http:// or https://) or a path to a local file. Please note, that max does not support CalDAV (yet).
# url =
# # Optional authentication (ignored for local files)
# # username =
# # password =
# Optional authentication (ignored for local files)
# username =
# password =
# optional phylter query (https://code.not-your-server.de/phylter.git) query to filter events for this calendar
# filter =
# Static week program
# each day takes comma-separated list of periods (hh:mm - hh:mm)

3
requirements.txt

@ -4,4 +4,5 @@ pytz
python-dateutil
pymax>=0.2
CacheControl
requests
requests
phylter

3
src/maxd/config.py

@ -85,11 +85,12 @@ def time_range(s):
yield int(m.group(1)), int(m.group(2)), int(m.group(3)), int(m.group(4))
class CalendarConfig(collections.namedtuple('CalendarConfig', ('name', 'url', 'username', 'password'))):
class CalendarConfig(collections.namedtuple('CalendarConfig', ('name', 'url', 'username', 'password', 'filter'))):
def __new__(cls, **kwargs):
kwargs.setdefault('username', None)
kwargs.setdefault('password', None)
kwargs.setdefault('filter', None)
return super(CalendarConfig, cls).__new__(cls, **kwargs)
@property

26
src/maxd/worker.py

@ -7,8 +7,7 @@ from dateutil import rrule
import pytz
import dateutil.tz
from maxd.fetcher import HTTPCalendarEventFetcher
from maxd.fetcher import LocalCalendarEventFetcher
from phylter.parser import Parser
from pymax.cube import Discovery, Cube
from pymax.objects import ProgramSchedule
@ -201,8 +200,19 @@ class Worker(object):
logger.info("Applying range filter to fetched events from %s" % calendar_config.name)
events = self.apply_range_filter(events, start, end)
if calendar_config.filter is not None:
logger.info("Applying user filter \"%s\" to %s events" % (calendar_config.filter, len(events)))
events = self.apply_user_filter(calendar_config.filter, events)
logger.debug("Event list contains now %s events from calendar %s" % (len(events), calendar_config.name))
else:
logger.debug("Filter query not set in calendar config")
return events
def apply_user_filter(self, query_string, events):
q = Parser().parse(query_string)
return q.apply(events)
def apply_range_filter(self, events, start, end):
start = (start.astimezone(pytz.UTC) if start.tzinfo else start).replace(hour=0, minute=0, second=0)
end = (end.astimezone(pytz.UTC) if end.tzinfo else end).replace(hour=23, minute=59, second=59)
@ -228,28 +238,28 @@ class Worker(object):
rule = rrule.rrulestr(cal_event.get('RRULE').to_ical().decode('utf-8'), dtstart=event_start_utc.replace(tzinfo=None))
if rule._until:
# The until field in the RRULE may contain a timezone (even if it's UTC).
# The until identifier in the RRULE may contain a timezone (even if it's UTC).
# Make sure its UTC and remove the tzinfo
rule._until = rule._until.astimezone(pytz.UTC).replace(tzinfo=None)
for dt in rule.between(start.replace(tzinfo=None), end.replace(tzinfo=None), inc=True):
if all_day:
s, e = _to_all_day(dt.date())
yield Event(name=None, start=s, end=e)
yield Event(name=str(cal_event['SUMMARY']), start=s, end=e)
else:
dt = dt.replace(tzinfo=pytz.UTC)
if 'duration' in cal_event:
duration = cal_event['duration'].dt # it's already a timedelta
else:
duration = cal_event['DTEND'].dt - cal_event['DTSTART'].dt
yield Event(name=None, start=dt, end=dt + duration)
yield Event(name=str(cal_event['SUMMARY']), start=dt, end=dt + duration)
else:
if all_day:
s, e = _to_all_day(cal_event['DTSTART'].dt)
yield Event(name=None, start=s, end=e)
yield Event(name=str(cal_event['SUMMARY']), start=s, end=e)
else:
yield Event(name=None, start=cal_event['DTSTART'].dt.astimezone(pytz.UTC), end=cal_event['DTEND'].dt.astimezone(pytz.UTC))
except Exception as ex:
yield Event(name=str(cal_event['SUMMARY']), start=cal_event['DTSTART'].dt.astimezone(pytz.UTC), end=cal_event['DTEND'].dt.astimezone(pytz.UTC))
except:
logger.exception("Failed to apply range filter to event %s" % cal_event)
for event in _build_all_events():

19
tests/test_worker.py

@ -14,7 +14,7 @@ try:
except ImportError:
from io import StringIO
from maxd.config import Configuration
from maxd.worker import Worker, Schedule, _to_utc_datetime
from maxd.worker import Worker, Schedule, _to_utc_datetime, Event
if sys.version_info.major == 2 or (sys.version_info.major == 3 and sys.version_info.minor <= 2):
from mock import Mock, patch
@ -44,12 +44,27 @@ class TestWorker(object):
datetime.datetime(2015, 12, 28, tzinfo=pytz.UTC),
datetime.datetime(2016, 1, 1, tzinfo=pytz.UTC))
filtered = list(filtered)
print(filtered)
# weekly event: once (2015-12-29)
# daily event: 4 (2015-12-28 till 2015-12-31)
assert len(filtered) == 5
def test_apply_user_filter(self):
w = Worker(Configuration('/dev/null'))
with open('tests/fixtures/calendars/repeating.ics', 'r') as f:
events = [o for o in icalendar.Calendar.from_ical(f.read()).walk() if o.name == 'VEVENT']
filtered = w.apply_range_filter(events,
datetime.datetime(2015, 12, 28, tzinfo=pytz.UTC),
datetime.datetime(2016, 1, 1, tzinfo=pytz.UTC))
filtered = list(filtered)
filtered = w.apply_user_filter("name == 'Ending repeating event'", filtered)
filtered = list(filtered)
assert len(filtered) == 4
def test_create_schedule(self):
w = Worker(Configuration('/dev/null'))