#!/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()