ics2mdwn/ics2mdwn.py

#!/usr/bin/python3

import argparse
import os.path
from typing import Iterable
from pathlib import Path
import yaml
from datetime import timedelta

from icalendar import Calendar, Event


def get_parser():
    p = argparse.ArgumentParser()
    p.add_argument("files", nargs="+", type=str)
    p.add_argument(
        "--out-md-dir",
        help="Output directory for markdown files",
        default="content/talks/",
        type=Path,
    )
    p.add_argument(
        "--out-schedule",
        help="Output schedule.yml file",
        default="data/schedule.yml",
        type=Path,
    )
    p.add_argument('--trust-location', type=str, nargs='*', default=[])
    p.add_argument("--night-threshold", metavar="HOUR", default=5, type=int)
    p.add_argument("--mode", choices=["hugo"], default="hugo")
    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)


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:
        return os.path.splitext(os.path.basename(fpath))[0]

    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):
        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):
                    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 HugoConverter(Converter):
    """
    add relevant output features to the base converter
    """

    def output_markdown(self):
        for uid in sorted(self.talks):
            talk = self.talks[uid]
            fname = "%s.md" % uid
            fpath = self.args.out_md_dir / fname
            self.changed_files.append(fpath)
            frontmatter = dict(
                    key=uid,
                    title=talk.decoded("SUMMARY").decode("utf8"),
                    format="conference",
                    start=talk.decoded("DTSTART"),
                    end=talk.decoded("DTEND"),
                    location=self.talk_location[uid],
                    duration=int(
                        (
                            talk.decoded("DTEND") - talk.decoded("DTSTART")
                        ).total_seconds()
                        // 60
                    ),
                )
            if "CATEGORIES" in talk:
                frontmatter["tags"] = [str(t) for t in talk.get("CATEGORIES").cats]
            else:
                frontmatter["tags"] = []
            if 'base' in frontmatter['tags']:
                frontmatter['level'] = 'Principiante'
            with open(str(fpath), "w") as buf:
                buf.write("---\n")
                yaml.safe_dump(frontmatter, buf)
                buf.write("---\n\n")
                # body
                if "DESCRIPTION" in talk:
                    buf.write(talk.decoded("DESCRIPTION").decode("utf8"))

    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(hour * 100 + minute, 10)
            duration = talk.decoded("dtend") - talk.decoded("dtstart")
            duration_minutes = int(duration.total_seconds() // 60)
            duration_minutes = round_down(duration_minutes, 30)
            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)]}

        # dump, finally
        with open(str(self.args.out_schedule), "w") as buf:
            yaml.safe_dump(out, buf)
        self.changed_files.append(self.args.out_schedule)

    def output(self):
        self.output_markdown()
        self.output_schedule()


def main():
    converter_register = {"hugo": HugoConverter}
    args = get_parser().parse_args()
    c = converter_register[args.mode](args)
    c.run()


if __name__ == "__main__":
    main()