ics2mdwn/ics2yaml.py

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