pycalendar/aggregator.py

   1 # Copyright
   2
   3 class Aggregator (list):
   4     r"""An iCalendar feed aggregator
   5
   6     Figure out where the example feeds are located, relative to the
   7     directory from which you run this doctest (i.e., the project's
   8     root directory).
   9
  10     >>> import os
  11     >>> root_dir = os.curdir
  12     >>> data_dir = os.path.abspath(os.path.join(root_dir, 'test', 'data'))
  13     >>> base_url = 'file://{}'.format(data_dir.replace(os.sep, '/'))
  14
  15     >>> from .feed import Feed
  16
  17     You can set processing hooks to analyze and manipulate feeds as
  18     they come in.
  19
  20     >>> processors = [lambda feed: print("I'm processing {!r}".format(feed))]
  21
  22     >>> a = Aggregator(
  23     ...     prodid='-//pycalendar//NONSGML testing//EN',
  24     ...     feeds=[
  25     ...         Feed(url='{}/{}'.format(base_url, name))
  26     ...         for name in ['geohash.ics',]],
  27     ...     processors=processors,
  28     ...     )
  29     >>> a  # doctest: +ELLIPSIS
  30     [<Feed url:file://.../test/data/geohash.ics>]
  31     >>> a.fetch()  # doctest: +ELLIPSIS
  32     I'm processing <Feed url:file://.../test/data/geohash.ics>
  33
  34     Generate aggregate calendars with the ``.write`` method.
  35
  36     >>> import io
  37     >>> stream = io.StringIO()
  38     >>> a.write(stream=stream)
  39     >>> value = stream.getvalue()
  40     >>> value  # doctest: +ELLIPSIS
  41     'BEGIN:VCALENDAR\r\nVERSION:2.0\r\n...END:VCALENDAR\r\n'
  42     >>> print(value.replace('\r\n', '\n'))
  43     BEGIN:VCALENDAR
  44     VERSION:2.0
  45     PRODID:-//pycalendar//NONSGML testing//EN
  46     BEGIN:VEVENT
  47     UID:2013-06-30@geohash.invalid
  48     DTSTAMP:2013-06-30T00:00:00Z
  49     DTSTART;VALUE=DATE:20130630
  50     DTEND;VALUE=DATE:20130701
  51     SUMMARY:XKCD geohashing\, Boston graticule
  52     URL:http://xkcd.com/426/
  53     LOCATION:Snow Hill\, Dover\, Massachusetts
  54     GEO:42.226663,-71.28676
  55     END:VEVENT
  56     END:VCALENDAR
  57     <BLANKLINE>
  58     """
  59     def __init__(self, prodid, version='2.0', feeds=None, processors=None):
  60         super(Aggregator, self).__init__()
  61         self.prodid = prodid
  62         self.version = version
  63         if feeds:
  64             self.extend(feeds)
  65         if not processors:
  66             processors = []
  67         self.processors = processors
  68
  69     def fetch(self):
  70         for feed in self:
  71             feed.fetch()
  72             for processor in self.processors:
  73                 processor(feed)
  74
  75     def write(self, stream):
  76         stream.write('BEGIN:VCALENDAR\r\n')
  77         stream.write('VERSION:{}\r\n'.format(self.version))
  78         stream.write('PRODID:{}\r\n'.format(self.prodid))
  79         for feed in self:
  80             for entry in feed:
  81                 entry.write(stream=stream)
  82         stream.write('END:VCALENDAR\r\n')