Announce Noisebridge Events on Mary Poppins

announce_events.py

#!/usr/bin/env python3
# coding: utf-8

from datetime import datetime, timedelta
import logging
import random
import requests
import urllib.parse

logging.basicConfig(level=logging.INFO)

weekdays = ["Monday", "Tuesday", "Wednesday", "Thurday", "Friday", "Saturday", "Sunday"]

def english_and(seq):
    if len(seq) < 3:
        return " and ".join(seq)
    return ", ".join(seq[:-1]) + ", and " + seq[-1]

def reservoir_sample(seq, k=1):
    sample = []
    for i, item in enumerate(seq, start=1):
        if len(sample) < k:
            sample.append(item)
            continue

        if random.random() < k / i:
            sample[random.randrange(k)] = item
    return sample

class Event:
    """A Meetup event object containing pertinent fields and helpers."""

    def __init__(self, name, start, duration):
        self.name = name
        self.start = start
        self.duration = duration

    @property
    def end(self):
        return self.start + self.duration

    @property
    def has_ended(self):
        return datetime.now() > self.end

    @property
    def has_started(self):
        return datetime.now() > self.start

    def __str__(self):
        return '"{self.name}" at {self.start}'.format(self=self)

    def __repr__(self):
        return 'Event(name={self.name!r}, start={self.start!r}, duration={self.duration!r})'.format(self=self)

    @classmethod    def from_json(cls, obj):
        """Create Event object from Meetup API response object."""
        return cls(
            name = obj['name'],
            start = datetime.fromtimestamp(obj['time'] // 1000),
            duration = timedelta(milliseconds=obj['duration']),
        )

class Client:
    def __init__(self, url="http://api.meetup.com", group="noisebridge"):
        self.base_url = url
        self.group_url = self.base_url  + "/" + group
        self.events_url = self.group_url + "/events"

    def fetch_upcoming_events(self, lookahead=timedelta(days=7)):
        resp = requests.get(self.events_url, {'no_later_than': (datetime.now() + lookahead).isoformat()})
        resp.raise_for_status
        for obj in resp.json():
            event = Event.from_json(obj)            if not event.has_started:
                yield event

def events_announcement(events):
    desc = "Some events happending this week at Noisebridge. "

    # Group events by their start date.
    events_by_date = {}
    for event in events:
        day = event.start.date()
        if day in events_by_date:
            events_by_date[day].append(event)
        else:
            events_by_date[day] = [event]

    for day in sorted(events_by_date.keys()):
        desc += "On %s we have " % weekdays[day.weekday()]
        desc += english_and(["%s at %s" % (e.name, e.start.time().strftime("%H:%M")) for e in events_by_date[day]])
        desc += ". "

    return desc

def mary_speak(text):
    escaped = text.replace("&", "%26")
    escaped = escaped.replace("#", "%23")
    escaped = escaped.lower()
    logging.info('Sending "%s" to Mary Poppins' % escaped)
    requests.get("http://pegasus.noise:5000", {"text": escaped}).raise_for_status()

if __name__ == "__main__":
    events = Client().fetch_upcoming_events()
    announcement = events_announcement(reservoir_sample(events, k=2))
    print("Text length is %d." % len(announcement))
    mary_speak(announcement)