393 linhas
13 KiB
Python
393 linhas
13 KiB
Python
'''
|
|
Manage talks scheduling in a semantic way
|
|
'''
|
|
|
|
|
|
from __future__ import print_function
|
|
import os
|
|
import io
|
|
from functools import wraps
|
|
import logging
|
|
import re
|
|
import datetime
|
|
import shutil
|
|
import time
|
|
from copy import copy
|
|
import locale
|
|
from contextlib import contextmanager
|
|
from babel.dates import format_date, format_datetime, format_time
|
|
|
|
from markdown import markdown
|
|
from docutils import nodes
|
|
from docutils.parsers.rst import directives, Directive
|
|
|
|
from pelican import signals, generators
|
|
import jinja2
|
|
|
|
pelican = None # This will be set during register()
|
|
|
|
|
|
|
|
|
|
def memoize(function):
|
|
'''decorators to cache'''
|
|
memo = {}
|
|
|
|
@wraps(function)
|
|
def wrapper(*args):
|
|
if args in memo:
|
|
return memo[args]
|
|
else:
|
|
rv = function(*args)
|
|
memo[args] = rv
|
|
return rv
|
|
return wrapper
|
|
|
|
|
|
@contextmanager
|
|
def setlocale(name):
|
|
saved = locale.setlocale(locale.LC_ALL)
|
|
try:
|
|
yield locale.setlocale(locale.LC_ALL, name)
|
|
finally:
|
|
locale.setlocale(locale.LC_ALL, saved)
|
|
|
|
|
|
@memoize
|
|
def get_talk_names():
|
|
return [name for name in os.listdir(pelican.settings['TALKS_PATH'])
|
|
if not name.startswith('_') and
|
|
get_talk_data(name) is not None
|
|
]
|
|
|
|
|
|
def all_talks():
|
|
return [get_talk_data(tn) for tn in get_talk_names()]
|
|
|
|
|
|
def unique_attr(iterable, attr):
|
|
return {x[attr] for x in iterable
|
|
if attr in x}
|
|
|
|
|
|
@memoize
|
|
def get_global_data():
|
|
fname = os.path.join(pelican.settings['TALKS_PATH'], 'meta.yaml')
|
|
if not os.path.isfile(fname):
|
|
return None
|
|
with io.open(fname, encoding='utf8') as buf:
|
|
try:
|
|
data = yaml.load(buf)
|
|
except Exception:
|
|
logging.exception("Syntax error reading %s; skipping", fname)
|
|
return None
|
|
if data is None:
|
|
return None
|
|
if 'startdate' not in data:
|
|
logging.error("Missing startdate in global data")
|
|
data['startdate'] = datetime.datetime.now()
|
|
return data
|
|
|
|
|
|
@memoize
|
|
def get_talk_data(talkname):
|
|
fname = os.path.join(pelican.settings['TALKS_PATH'], talkname, 'meta.yaml')
|
|
if not os.path.isfile(fname):
|
|
return None
|
|
with io.open(fname, encoding='utf8') as buf:
|
|
try:
|
|
data = yaml.load(buf)
|
|
except:
|
|
logging.exception("Syntax error reading %s; skipping", fname)
|
|
return None
|
|
if data is None:
|
|
return None
|
|
try:
|
|
gridstep = pelican.settings['TALKS_GRID_STEP']
|
|
if 'title' not in data:
|
|
logging.warn("Talk <{}> has no `title` field".format(talkname))
|
|
data['title'] = talkname
|
|
if 'text' not in data:
|
|
logging.warn("Talk <{}> has no `text` field".format(talkname))
|
|
data['text'] = ''
|
|
if 'duration' not in data:
|
|
logging.info("Talk <{}> has no `duration` field (50min used)"
|
|
.format(talkname))
|
|
data['duration'] = 50
|
|
data['duration'] = int(data['duration'])
|
|
if data['duration'] < gridstep:
|
|
logging.info("Talk <{}> lasts only {} minutes; changing to {}"
|
|
.format(talkname, data['duration'], gridstep))
|
|
data['duration'] = gridstep
|
|
if 'links' not in data or not data['links']:
|
|
data['links'] = []
|
|
if 'contacts' not in data or not data['contacts']:
|
|
data['contacts'] = []
|
|
if 'needs' not in data or not data['needs']:
|
|
data['needs'] = []
|
|
if 'room' not in data:
|
|
logging.warn("Talk <{}> has no `room` field".format(talkname))
|
|
if 'time' not in data or 'day' not in data:
|
|
logging.warn("Talk <{}> has no `time` or `day`".format(talkname))
|
|
if 'time' in data:
|
|
del data['time']
|
|
if 'day' in data:
|
|
del data['day']
|
|
if 'day' in data:
|
|
data['day'] = get_global_data()['startdate'] + \
|
|
datetime.timedelta(days=data['day'])
|
|
if 'time' in data and 'day' in data:
|
|
timeparts = re.findall(r'\d+', str(data['time']))
|
|
if 4 > len(timeparts) > 0:
|
|
timeparts = [int(p) for p in timeparts]
|
|
data['time'] = datetime.datetime.combine(
|
|
data['day'], datetime.time(*timeparts))
|
|
else:
|
|
logging.error("Talk <%s> has malformed `time`", talkname)
|
|
data['id'] = talkname
|
|
resdir = os.path.join(pelican.settings['TALKS_PATH'], talkname,
|
|
pelican.settings['TALKS_ATTACHMENT_PATH'])
|
|
if os.path.isdir(resdir) and os.listdir(resdir):
|
|
data['resources'] = resdir
|
|
return data
|
|
except:
|
|
logging.exception("Error on talk %s", talkname)
|
|
raise
|
|
|
|
|
|
@memoize
|
|
def jinja_env():
|
|
env = jinja2.Environment(
|
|
loader=jinja2.FileSystemLoader(os.path.join(pelican.settings['TALKS_PATH'], '_templates')),
|
|
autoescape=True,
|
|
)
|
|
env.filters['markdown'] = lambda text: jinja2.Markup(markdown(text))
|
|
env.filters['dateformat'] = format_date
|
|
env.filters['datetimeformat'] = format_datetime
|
|
env.filters['timeformat'] = format_time
|
|
return env
|
|
|
|
|
|
class TalkListDirective(Directive):
|
|
final_argument_whitespace = True
|
|
has_content = True
|
|
option_spec = {
|
|
'lang': directives.unchanged
|
|
}
|
|
|
|
def run(self):
|
|
lang = self.options.get('lang', 'C')
|
|
tmpl = jinja_env().get_template('talk.html')
|
|
|
|
def _sort_date(name):
|
|
'''
|
|
This function is a helper to sort talks by start date
|
|
|
|
When no date is available, put at the beginning
|
|
'''
|
|
d = get_talk_data(name)
|
|
room = d.get('room', '')
|
|
time = d.get('time', datetime.datetime(1, 1, 1))
|
|
title = d.get('title', '')
|
|
return (time, room, title)
|
|
|
|
return [
|
|
nodes.raw('', tmpl.render(lang=lang, **get_talk_data(n)),
|
|
format='html')
|
|
for n in sorted(get_talk_names(),
|
|
key=_sort_date)
|
|
]
|
|
|
|
|
|
class TalkDirective(Directive):
|
|
required_arguments = 1
|
|
final_argument_whitespace = True
|
|
has_content = True
|
|
option_spec = {
|
|
'lang': directives.unchanged
|
|
}
|
|
|
|
def run(self):
|
|
lang = self.options.get('lang', 'C')
|
|
tmpl = jinja_env().get_template('talk.html')
|
|
data = get_talk_data(self.arguments[0])
|
|
if data is None:
|
|
return []
|
|
return [
|
|
nodes.raw('', tmpl.render(lang=lang, **data),
|
|
format='html')
|
|
]
|
|
|
|
|
|
class TalkGridDirective(Directive):
|
|
'''A complete grid'''
|
|
final_argument_whitespace = True
|
|
has_content = True
|
|
option_spec = {
|
|
'lang': directives.unchanged
|
|
}
|
|
|
|
def run(self):
|
|
lang = self.options.get('lang', 'C')
|
|
tmpl = jinja_env().get_template('grid.html')
|
|
output = []
|
|
days = unique_attr(all_talks(), 'day')
|
|
gridstep = pelican.settings['TALKS_GRID_STEP']
|
|
for day in sorted(days):
|
|
talks = {talk['id'] for talk in all_talks()
|
|
if talk.get('day', None) == day
|
|
and 'time' in talk
|
|
and 'room' in talk}
|
|
if not talks:
|
|
continue
|
|
talks = [get_talk_data(t) for t in talks]
|
|
rooms = set()
|
|
for t in talks:
|
|
if type(t['room']) is list:
|
|
for r in t['room']:
|
|
rooms.add(r)
|
|
else:
|
|
rooms.add(t['room'])
|
|
rooms = list(sorted(rooms))
|
|
# room=* is not a real room.
|
|
# Remove it unless that day only has special rooms
|
|
if '*' in rooms and len(rooms) > 1:
|
|
del rooms[rooms.index('*')]
|
|
mintime = min({talk['time'].hour * 60 +
|
|
talk['time'].minute
|
|
for talk in talks}) // gridstep * gridstep
|
|
maxtime = max({talk['time'].hour * 60 +
|
|
talk['time'].minute +
|
|
talk['duration']
|
|
for talk in talks})
|
|
times = {}
|
|
|
|
for t in range(mintime, maxtime, gridstep):
|
|
times[t] = [None] * len(rooms)
|
|
for talk in sorted(talks, key=lambda x: x['time']):
|
|
talktime = talk['time'].hour * 60 + talk['time'].minute
|
|
position = talktime // gridstep * gridstep # round
|
|
assert position in times
|
|
if talk['room'] == '*':
|
|
roomnums = range(len(rooms))
|
|
elif type(talk['room']) is list:
|
|
roomnums = [rooms.index(r) for r in talk['room']]
|
|
else:
|
|
roomnums = [rooms.index(talk['room'])]
|
|
for roomnum in roomnums:
|
|
if times[position][roomnum] is not None:
|
|
logging.error("Talk %s and %s overlap! (room %s)",
|
|
times[position][roomnum]['id'],
|
|
talk['id'],
|
|
rooms[roomnum]
|
|
)
|
|
continue
|
|
times[position][roomnum] = copy(talk)
|
|
times[position][roomnum]['skip'] = False
|
|
for i in range(1, talk['duration'] // gridstep):
|
|
times[position + i*gridstep][roomnum] = copy(talk)
|
|
times[position + i*gridstep][roomnum]['skip'] = True
|
|
|
|
render = tmpl.render(times=times,
|
|
rooms=rooms,
|
|
mintime=mintime, maxtime=maxtime,
|
|
timestep=gridstep,
|
|
lang=lang,
|
|
)
|
|
output.append(nodes.raw(
|
|
'', u'<h4>%s</h4>' % format_date(day, format='full',
|
|
locale=lang),
|
|
format='html'))
|
|
output.append(nodes.raw('', render, format='html'))
|
|
return output
|
|
|
|
|
|
def talks_to_ics():
|
|
content = u'BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:pelican\n'
|
|
for t in all_talks():
|
|
try:
|
|
content += talk_to_ics(t)
|
|
except:
|
|
logging.exception("Error producing calendar for talk %s", t['id'])
|
|
content += 'END:VCALENDAR\n'
|
|
return content
|
|
|
|
|
|
def talk_to_ics(talk):
|
|
if 'time' not in talk or 'duration' not in talk:
|
|
return ''
|
|
start = talk['time']
|
|
end = start + datetime.timedelta(minutes=talk['duration'])
|
|
content = 'BEGIN:VEVENT\n'
|
|
content += "UID:%s@%d.hackmeeting.org\n" % (talk['id'], talk['day'].year)
|
|
content += "SUMMARY:%s\n" % talk['title']
|
|
content += "DTSTAMP:%s\n" % time.strftime('%Y%m%dT%H%M%SZ',
|
|
time.gmtime(float(
|
|
start.strftime('%s'))))
|
|
content += "DTSTART:%s\n" % time.strftime('%Y%m%dT%H%M%SZ',
|
|
time.gmtime(float(
|
|
start.strftime('%s'))))
|
|
content += "DTEND:%s\n" % time.strftime('%Y%m%dT%H%M%SZ',
|
|
time.gmtime(float(
|
|
end.strftime('%s'))))
|
|
|
|
content += "LOCATION:%s\n" % (talk['room'] if 'room' in talk else 'todo')
|
|
content += 'END:VEVENT\n'
|
|
return content
|
|
|
|
|
|
class TalksGenerator(generators.Generator):
|
|
def __init__(self, *args, **kwargs):
|
|
self.talks = []
|
|
super(TalksGenerator, self).__init__(*args, **kwargs)
|
|
|
|
def generate_context(self):
|
|
self.talks = {n: get_talk_data(n) for n in get_talk_names()}
|
|
self._update_context(('talks',))
|
|
|
|
def generate_output(self, writer=None):
|
|
for talkname in self.talks:
|
|
if 'resources' in self.talks[talkname]:
|
|
outdir = os.path.join(self.output_path,
|
|
pelican.settings['TALKS_PATH'], talkname,
|
|
pelican.settings['TALKS_ATTACHMENT_PATH'])
|
|
if os.path.isdir(outdir):
|
|
shutil.rmtree(outdir)
|
|
shutil.copytree(self.talks[talkname]['resources'], outdir)
|
|
with io.open(os.path.join(self.output_path, pelican.settings.get('TALKS_ICS')),
|
|
'w',
|
|
encoding='utf8') as buf:
|
|
buf.write(talks_to_ics())
|
|
|
|
|
|
def add_talks_option_defaults(pelican):
|
|
pelican.settings.setdefault('TALKS_PATH', 'talks')
|
|
pelican.settings.setdefault('TALKS_ATTACHMENT_PATH', 'res')
|
|
pelican.settings.setdefault('TALKS_ICS', 'schedule.ics')
|
|
pelican.settings.setdefault('TALKS_GRID_STEP', 30)
|
|
|
|
|
|
def get_generators(gen):
|
|
return TalksGenerator
|
|
|
|
|
|
def pelican_init(pelicanobj):
|
|
global pelican
|
|
pelican = pelicanobj
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
print('ERROR: yaml not found. Talks plugins will be disabled')
|
|
|
|
def register():
|
|
pass
|
|
else:
|
|
|
|
def register():
|
|
signals.initialized.connect(pelican_init)
|
|
signals.get_generators.connect(get_generators)
|
|
signals.initialized.connect(add_talks_option_defaults)
|
|
directives.register_directive('talklist', TalkListDirective)
|
|
directives.register_directive('talk', TalkDirective)
|
|
directives.register_directive('talkgrid', TalkGridDirective)
|