depot/py/icalfilter/app.py

import asyncio
import concurrent.futures
import dataclasses
import datetime
import json
import os
from typing import Any, Dict, List, Optional, Set, Union

import aiohttp
import attrs
import icalendar
import icalevents.icalparser
from dateutil.tz import UTC
from quart import Quart, Response, render_template


async def parse_ical(calendar_text: str) -> icalendar.Calendar:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        process_pool, icalendar.Calendar.from_ical, calendar_text
    )


async def serialize_ical(calendar: icalendar.Calendar) -> str:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(process_pool, calendar.to_ical)


def maybe_fromisoformat(val: Optional[str]) -> Optional[datetime.datetime]:
    if not val:
        return None
    if isinstance(val, datetime.datetime):
        return val
    dt = datetime.datetime.fromisoformat(val)
    if not dt.tzinfo:
        dt = dt.replace(tzinfo=UTC)
    return dt


@attrs.frozen
class Calendar:
    name: str
    source_url: str
    keep_hashtags: Set[str] = attrs.field(converter=frozenset, factory=frozenset)
    drop_hashtags: Set[str] = attrs.field(converter=frozenset, factory=frozenset)
    minimize: bool = True
    skip_minimize_hashtags: Set[str] = attrs.field(
        converter=frozenset, factory=frozenset
    )
    skip_before: Optional[datetime.datetime] = attrs.field(
        converter=maybe_fromisoformat, default=None
    )
    skip_if_declined: Set[str] = attrs.field(converter=frozenset, factory=frozenset)
    retitle_to: Optional[str] = None

    async def fetch_events(self, session: aiohttp.ClientSession) -> icalendar.Calendar:
        async with session.get(self.source_url) as response:
            response.raise_for_status()
            text = await response.text()
        return await parse_ical(text)


def make_calendars(
    in_calendars: Union[List[Dict[str, Any]], List[Calendar]]
) -> List[Calendar]:
    if len(in_calendars) == 0:
        return []
    elif isinstance(in_calendars[0], Calendar):
        return in_calendars
    return [Calendar(**c) for c in in_calendars]


@attrs.frozen
class Config:
    calendars: List[Calendar] = attrs.field(converter=make_calendars)

    async def fetch_calendars(
        self, session: aiohttp.ClientSession
    ) -> List[icalendar.Calendar]:
        return await asyncio.gather(*[c.fetch_events(session) for c in self.calendars])


def load_config(fn: str) -> Config:
    with open(fn, "rt") as f:
        return Config(**json.load(f))


app = Quart(__name__)
config = load_config(os.environ.get("ICALFILTER_CONFIG", "config/config.json"))
process_pool = concurrent.futures.ProcessPoolExecutor(max_workers=4)


def contains_any_hashtag(text: str, hashtags: Set[str]) -> bool:
    return any(hashtag in text for hashtag in hashtags)


def _all_occurrences_before_expensive(
    event: icalendar.Event,
    cutoff: datetime.datetime,
    seen_timezones: Dict[str, icalendar.Timezone],
) -> bool:
    # Recurring events are... more complicated.
    rrule_or_rruleset = icalevents.icalparser.parse_rrule(event)
    try:
        return not rrule_or_rruleset.after(cutoff)
    except TypeError:
        return not rrule_or_rruleset.after(cutoff.replace(tzinfo=None))


async def all_occurrences_before(
    event: icalendar.Event,
    cutoff: datetime.datetime,
    seen_timezones: Dict[str, icalendar.Timezone],
) -> bool:
    parsed_event = icalevents.icalparser.create_event(event, cutoff.tzinfo)
    if not parsed_event.recurring:
        if type(parsed_event.end) is datetime.date:
            return parsed_event.end < cutoff.date()
        return parsed_event.end < cutoff

    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        process_pool, _all_occurrences_before_expensive, event, cutoff, seen_timezones
    )


async def keep_event(
    source_cal: Calendar,
    event: icalendar.Event,
    seen_timezones: Dict[str, icalendar.Timezone],
) -> bool:
    summary = event.get('summary', '')
    if source_cal.keep_hashtags:
        if not contains_any_hashtag(summary, source_cal.keep_hashtags):
            return False
    if source_cal.drop_hashtags:
        if contains_any_hashtag(summary, source_cal.drop_hashtags):
            return False
    if source_cal.skip_before:
        if "dtstart" not in event:
            print("no dtstart?", event)
            return False
        if await all_occurrences_before(event, source_cal.skip_before, seen_timezones):
            return False
    if source_cal.skip_if_declined:
        attendees = event.get("attendee", [])
        if not isinstance(attendees, list):
            # Sometimes it's just a plain vCalAddress
            attendees = [attendees]
        for attendee in attendees:
            if str(attendee) not in source_cal.skip_if_declined:
                continue
            if attendee.params.get("PARTSTAT", None) == "DECLINED":
                return False
    return True


def maybe_strip_event(source_cal: Calendar, event: icalendar.Event) -> icalendar.Event:
    if not source_cal.minimize:
        return event
    if source_cal.retitle_to:
        event["summary"] = source_cal.retitle_to
    out_event = icalendar.Event()
    for prop in [
        "uid",
        "dtstamp",
        "summary",
        "dtstart",
        "dtend",
        "duration",
        "recurrence-id",
        "sequence",
        "rrule",
        "rdate",
        "exdate",
    ]:
        if prop in event:
            out_event[prop] = event[prop]
    return out_event


class Cache:
    def __init__(self, timeout, cb):
        self._timeout = timeout
        self._cb = cb
        self._cond = asyncio.Condition()
        self._value = None
        self._value_ttl = None
        self._acquiring_value = False

    def _check_value(self):
        now = datetime.datetime.utcnow()
        if self._value_ttl and now >= self._value_ttl:
            self._value = None
            self._value_ttl = None
        if not self._value_ttl:
            return (None, False)
        return (self._value, True)

    async def get_value(self):
        await self._cond.acquire()
        while True:
            if self._acquiring_value:
                await self._cond.wait()
            value, ok = self._check_value()
            if ok:
                self._cond.release()
                return value

            self._acquiring_value = True
            self._cond.release()
            try:
                value = await self._cb()
            except:
                await self._cond.acquire()
                self._acquiring_value = False
                self._cond.notify_all()
                self._cond.release()
                raise
            await self._cond.acquire()
            self._value = value
            self._value_ttl = datetime.datetime.utcnow() + self._timeout
            self._acquiring_value = False
            self._cond.notify_all()
            self._cond.release()
            return value


async def render_ical():
    async with aiohttp.ClientSession() as session:
        icals = await config.fetch_calendars(session)

    cal = icalendar.Calendar()
    cal.add("prodid", "-//icalfilter//lukegb.com//")
    cal.add("version", "2.0")

    seen_timezones = {}
    for source_cal, source_ical in zip(config.calendars, icals):
        for event in source_ical.subcomponents:
            if isinstance(event, icalendar.Timezone):
                if event["tzid"] in seen_timezones:
                    continue
                seen_timezones[event["tzid"]] = event
                cal.add_component(event)

    for source_cal, source_ical in zip(config.calendars, icals):
        for event in source_ical.subcomponents:
            if not isinstance(event, icalendar.Event):
                continue
            if not await keep_event(source_cal, event, seen_timezones):
                continue
            cal.add_component(maybe_strip_event(source_cal, event))

    return await serialize_ical(cal)


ical_cache = Cache(datetime.timedelta(hours=3), render_ical)


@app.get("/ical.ics")
async def render_ical_view():
    cal_text = await ical_cache.get_value()
    return Response(cal_text, mimetype="text/calendar")


@app.get("/")
async def index():
    return await render_template("index.html")


if __name__ == "__main__":
    import sys

    print('This is intended to be run with "quart run".', file=sys.stderr)
    sys.exit(1)