import json
from collections import OrderedDict
from urllib.parse import urlparse
import requests
from requests_file import FileAdapter
from .base import BaseScheduleSupport, filter_additional_data, schedule_time_to_timedelta
s = requests.Session()
s.mount('file://', FileAdapter())
class ScheduleJSONSupport(BaseScheduleSupport):
identifier = 'schedule-json'
readonly = True
# fmt: off
configuration_fields = {
# 'key': (type, default value, mandatory, translation text)
'feedback': ('bool', True, False, 'Enable/Disable feedback url generation'),
'kind': ('string', 'assembly', False, 'kind of events, either `assembly` or `official` or `sos` or `lightning`'),
'headers': ('dict', {}, False, 'HTTP headers to send with the request e.g. Authorization'),
'auth': ('string', None, False, 'HTTP Authentification header e.g. `Token 123456`'),
'legacy_id': ('string', None, False, 'Use `drop` to regenerate legacy ids from the event guid, to avoid collisions with other sources'),
'legacy_id_offset': ('int', None, False, 'Offset to add to the legacy integer `id` to avoid collisions with other sources'),
}
# fmt: on
def ready(self):
r = s.head(self.remote_url)
return r.ok
def fetch(self):
"""
This method is the workhorse of the schedule support module:
its job is to query upstream for the current set of data.
It shall return a dictionary with keys 'rooms' and 'events',
each containing a dictionary with entries mapping a source id
to a dictionary which can be understood by Room.from_dict()
and Event.from_dict() respectively.
The hub will update events it already knows by the source id:
all events need to have an unique but stable identifier, i.e.
if the name of the event changes the identifier must not change!
"""
headers = {}
if self.conf_value('headers'):
headers = self.conf_value('headers')
if self.conf_value('auth'):
headers['Authorization'] = self.conf_value('auth')
schedule = ScheduleJSON.from_url(self.remote_url, headers={**headers, 'Accept-Language': 'en'})
schedule_de = ScheduleJSON.from_url(self.remote_url, headers={**headers, 'Accept-Language': 'de'}, event_map=True)
instance = urlparse(schedule.get('base_url', self.remote_url))
host = f'//{instance.netloc}'
kind = self.conf_value('kind')
other_fields_to_drop = []
if self.conf_value('legacy_id') == 'drop':
other_fields_to_drop.append('id')
def ensure_full_url(uri):
if not uri:
return None
if not uri.startswith('http') and not uri.startswith('//'):
return f'{host}{uri}'
return uri
return {
'version': schedule.version(),
'rooms': {r['name']: r for r in schedule.rooms()},
'events': {
e.get('id'): {
'guid': e.get('guid'),
'slug': e.get('slug').split(f"{e.get('id')}-")[1][0:150].strip('-') or e.get('slug')[0:150].strip('-'),
'name': e.get('title'),
'language': e.get('language'),
'abstract': e.get('abstract') or '',
'description': e.get('description') or '',
'description_en': e.get('description') or '',
'description_de': schedule_de.event(e.get('guid')).get('description') or '',
'track': e.get('track'),
'room': e.get('room'),
'schedule_start': e.get('date'),
'schedule_duration': str(schedule_time_to_timedelta(e.get('duration'))),
'is_public': True,
'kind': kind,
'speakers': e.get('persons', []),
'banner_image_url': ensure_full_url(e.get('logo')),
'additional_data': filter_additional_data(e, self.computed_data(e), other_fields_to_drop),
}
for e in schedule.events()
},
}
def computed_data(self, event: dict):
data = {}
data['origin_url'] = event['url']
# add feedback_url if feedback is enabled via configuraiton_fields in ScheduleSource config
if self.conf_value('feedback'):
data['feedback_url'] = f"{event['url']}feedback/"
if self.conf_value('legacy_id_offset'):
data['id'] = event['id'] + self.conf_value('legacy_id_offset')
return data
class ScheduleJSON:
"""
Schedule from JSON document
"""
_schedule = None
_events = None
def __init__(self, json, event_map=False):
self._schedule = json
if event_map:
self._events = {e.get('guid'): e for e in self.events()}
@classmethod
def from_url(cls, url, client=None, headers=None, event_map=False):
r = (client if client else s).get(url, headers=headers)
if r.ok is False:
raise Exception(f'Request failed, HTTP {r.status_code}.')
# maintain order from input file
schedule = json.JSONDecoder(object_pairs_hook=OrderedDict).decode(r.text)
# Close the raw file handle if it's still open
if hasattr(r, 'raw') and r.raw.closed is False:
r.raw.close()
return ScheduleJSON(json=schedule['schedule'], event_map=event_map)
def __getitem__(self, key):
return self._schedule[key]
def get(self, key, default=None):
return self._schedule.get(key, default)
def schedule(self):
return self._schedule
def version(self):
return self._schedule.get('version')
def days(self):
return self._schedule.get('conference').get('days')
def rooms(self):
# try to access the room dict from schedule.json gen 2021
rooms = self._schedule.get('conference', {}).get('rooms', [])
if rooms:
return list(rooms)
# looks like we have an older schudule.json (gen 2020), without a dedicated room list
# so we have use a fallback and iterate all days adding the rooms to a set, creating uniqueness
rooms = set()
for day in self.days():
for roomname in day.get('rooms'):
rooms.add(roomname)
return [{'name': name} for name in rooms]
def events(self):
for day in self.days():
for room in day.get('rooms'):
yield from day.get('rooms')[room]
def event(self, guid):
if guid in self._events:
return self._events[guid]
return None
def __str__(self):
return json.dumps(self._schedule, indent=2)