229 lines
7.7 KiB
Python
Executable file
229 lines
7.7 KiB
Python
Executable file
#!/usr/bin/python3
|
|
|
|
import sys
|
|
import logging
|
|
import argparse
|
|
import os.path
|
|
from typing import Iterable
|
|
from pathlib import Path
|
|
from datetime import timedelta
|
|
import json
|
|
|
|
import yaml
|
|
from icalendar import Calendar, Event
|
|
|
|
|
|
def get_parser():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("files", nargs="+", type=str)
|
|
p.add_argument("--hm-json", default='./hackmeeting.json', type=Path)
|
|
p.add_argument(
|
|
"--out-talks-dir",
|
|
help="Output directory for markdown files",
|
|
default="talks/",
|
|
type=Path,
|
|
)
|
|
p.add_argument("--trust-location", type=str, nargs="*", default=[])
|
|
p.add_argument(
|
|
"--slot-size",
|
|
type=int,
|
|
metavar="MINUTES",
|
|
default=15,
|
|
help="Round times to the nearest */MINUTES",
|
|
)
|
|
p.add_argument("--night-threshold", metavar="HOUR", default=5, type=int)
|
|
p.add_argument("--mode", choices=["pelican"], default="pelican")
|
|
return p
|
|
|
|
|
|
def round_down(num, divisor):
|
|
"""
|
|
>>> round_down(1000, 10)
|
|
1000
|
|
>>> round_down(1001, 10)
|
|
1000
|
|
>>> round_down(1009, 10)
|
|
1000
|
|
"""
|
|
return num - (num % divisor)
|
|
|
|
|
|
def round_down_time(hhmm: str, divisor: int):
|
|
hh = hhmm[:-2]
|
|
mm = hhmm[-2:]
|
|
mm = round_down(int(mm, base=10), divisor)
|
|
return int('%s%02d' % (hh,mm) , base=10)
|
|
|
|
|
|
class Converter:
|
|
"""
|
|
This class takes care of everything converter-related.
|
|
|
|
Objects are used to enable multiple output formats to be added pretty easily by subclassing
|
|
"""
|
|
|
|
def __init__(self, args):
|
|
self.args = args
|
|
self.rooms = []
|
|
self.talks = {}
|
|
self.talk_room = {} # map talk uid to room name
|
|
self.talk_location = {} # same, but see --trust-location
|
|
self.changed_files = []
|
|
|
|
def _fname_to_room(self, fpath: str) -> str:
|
|
base = os.path.splitext(os.path.basename(fpath))[0]
|
|
if base == 'ALL':
|
|
return '*'
|
|
return base
|
|
|
|
def get_vevents_from_calendar(self, cal: Calendar) -> Iterable[Event]:
|
|
for subc in cal.subcomponents:
|
|
if type(subc) is Event:
|
|
yield subc
|
|
|
|
def load_input(self):
|
|
with self.args.hm_json.open() as buf:
|
|
self.hackmeeting_metadata = json.load(buf)
|
|
for fpath in self.args.files:
|
|
room = self._fname_to_room(fpath)
|
|
with open(fpath) as buf:
|
|
file_content = buf.read()
|
|
cal = Calendar.from_ical(file_content, multiple=True)
|
|
for subcal in cal:
|
|
for ev in self.get_vevents_from_calendar(subcal):
|
|
if ev.decoded('DTSTART').year != self.hackmeeting_metadata['year']:
|
|
continue
|
|
uid = ev.decoded("uid").decode("ascii")
|
|
self.talks[uid] = ev
|
|
self.talk_room[uid] = room
|
|
self.talk_location[uid] = room
|
|
if fpath in self.args.trust_location:
|
|
try:
|
|
self.talk_location[uid] = ev.decoded("location").decode(
|
|
"utf8"
|
|
)
|
|
except:
|
|
pass
|
|
|
|
def run(self):
|
|
self.rooms = [self._fname_to_room(fpath) for fpath in self.args.files]
|
|
self.load_input()
|
|
self.output()
|
|
for fpath in self.changed_files:
|
|
print(fpath)
|
|
|
|
|
|
class PelicanConverter(Converter):
|
|
"""
|
|
add relevant output features to the base converter
|
|
"""
|
|
|
|
def load_input(self):
|
|
super().load_input()
|
|
talks_meta = self.args.out_talks_dir / 'meta.yaml'
|
|
with talks_meta.open() as buf:
|
|
self.talks_metadata = yaml.safe_load(buf)
|
|
|
|
def output_markdown(self):
|
|
for uid in sorted(self.talks):
|
|
talk = self.talks[uid]
|
|
fname = 'meta.yaml'
|
|
talkdir = self.args.out_talks_dir / uid
|
|
talkdir.mkdir(exist_ok=True)
|
|
fpath = talkdir / fname
|
|
self.changed_files.append(fpath)
|
|
day = (talk.decoded('DTSTART').date() - self.talks_metadata['startdate']).days
|
|
after_midnight = talk.decoded('DTSTART').hour < self.args.night_threshold
|
|
if after_midnight:
|
|
day -= 1
|
|
frontmatter = dict(
|
|
key=uid,
|
|
title=talk.decoded("SUMMARY").decode("utf8"),
|
|
format="conference",
|
|
start=talk.decoded("DTSTART"),
|
|
time=talk.decoded("DTSTART").strftime('%H:%M'),
|
|
day=day,
|
|
end=talk.decoded("DTEND"),
|
|
room=self.talk_location[uid],
|
|
duration=int(
|
|
(talk.decoded("DTEND") - talk.decoded("DTSTART")).total_seconds()
|
|
// 60
|
|
),
|
|
tags=[],
|
|
)
|
|
if "CATEGORIES" in talk:
|
|
try:
|
|
vobject = talk.get("CATEGORIES")
|
|
if hasattr(vobject, "cats"):
|
|
vobject = vobject.cats
|
|
frontmatter["tags"] = [str(t) for t in vobject]
|
|
else:
|
|
frontmatter["tags"] = [str(vobject)]
|
|
except Exception as exc:
|
|
logging.warning("Error parsing categories: %s", str(exc))
|
|
if "base" in frontmatter["tags"]:
|
|
frontmatter["level"] = "beginner"
|
|
if "DESCRIPTION" in talk:
|
|
frontmatter['text'] = talk.decoded("DESCRIPTION").decode("utf8")
|
|
else:
|
|
frontmatter['text'] = ''
|
|
with open(str(fpath), "w") as buf:
|
|
yaml.safe_dump(frontmatter, buf)
|
|
# body
|
|
|
|
def output_schedule(self):
|
|
days = {}
|
|
for uid in sorted(self.talks):
|
|
talk = self.talks[uid]
|
|
# TODO: talk just after midnight should belong to the preceding day
|
|
dt = talk.decoded("dtstart")
|
|
after_midnight = dt.time().hour < self.args.night_threshold
|
|
if after_midnight:
|
|
dt = dt - timedelta(days=1)
|
|
day = dt.strftime("%Y-%m-%d")
|
|
|
|
hour = talk.decoded("dtstart").time().hour
|
|
minute = talk.decoded("dtstart").time().minute
|
|
if after_midnight:
|
|
hour += 24
|
|
|
|
start = "%02d:%02d" % (hour, minute)
|
|
if day not in days:
|
|
days[day] = dict(day=day, start=start, rooms={})
|
|
if days[day]["start"] > start:
|
|
days[day]["start"] = start
|
|
|
|
room = self.talk_room[uid]
|
|
days[day]["rooms"].setdefault(room, dict(room=room, slots=[]))
|
|
|
|
talkstart = round_down_time('%02d%02d' % (hour, minute), self.args.slot_size)
|
|
duration = talk.decoded("dtend") - talk.decoded("dtstart")
|
|
duration_minutes = int(duration.total_seconds() // 60)
|
|
duration_minutes = round_down(duration_minutes, self.args.slot_size)
|
|
slot = "%04d-%dmin" % (talkstart, duration_minutes)
|
|
days[day]["rooms"][room]["slots"].append(dict(slot=slot, talk=uid))
|
|
|
|
# convert from our intermediate format to the correct one
|
|
for d in sorted(days):
|
|
# vanity: let's sort
|
|
for room in sorted(days[d]["rooms"]):
|
|
days[d]["rooms"][room]["slots"].sort(key=lambda x: x["slot"])
|
|
# convert dict to list
|
|
days[d]["rooms"] = [days[d]["rooms"][k] for k in sorted(days[d]["rooms"])]
|
|
out = {"schedule": [days[k] for k in sorted(days)]}
|
|
|
|
# XXX: dump, finally
|
|
|
|
def output(self):
|
|
self.output_markdown()
|
|
|
|
|
|
def main():
|
|
converter_register = {"pelican": PelicanConverter}
|
|
args = get_parser().parse_args()
|
|
c = converter_register[args.mode](args)
|
|
c.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|