Produits Entr'ouvert » w.c.s. » Au quotidien

import time

import datetime

import urllib2

import vobject

from quixote import get_request

from qommon.publisher import get_publisher_class

from qommon.storage import StorableObject

from qommon.cron import CronJob

from qommon import misc

class Event(StorableObject):

    _names = 'events'

    title = None

    description = None

    url = None

    date_start = None

    date_end = None

    def in_month(self, year, month):

        if not self.date_end: # a single date

            return tuple(self.date_start[:2]) == (year, month)

        else:

            # an interval

            if tuple(self.date_start[:2]) > (year, month): # start later

                return False

            if tuple(self.date_end[:2]) < (year, month): # ended before

                return False

        return True

    def format_date(self):

        d = {

            'year_start': self.date_start[0],

            'month_start': misc.get_month_name(self.date_start[1]),

            'day_start': self.date_start[2]

            }

        if self.date_end and self.date_start[:3] != self.date_end[:3]:

            d.update({

                'year_end': self.date_end[0],

                'month_end': misc.get_month_name(self.date_end[1]),

                'day_end': self.date_end[2]

                })

            d2 = datetime.date(*self.date_start[:3]) + datetime.timedelta()

            if tuple(self.date_end[:2]) == (d2.year, d2.month): # two consecutive days

                if self.date_start[1] == self.date_end[1]:

                    return _('On %(month_start)s %(day_start)s and %(day_end)s') % d

                else:

                    return _('On %(month_start)s %(day_start)s and %(month_end)s %(day_end)s') % d

            else:

                if self.date_start[0] == self.date_end[0]: # same year

                    if self.date_start[1] == self.date_end[1]: # same month

                        return _('From %(month_start)s %(day_start)s to %(day_end)s') % d

                    else:

                        return _('From %(month_start)s %(day_start)s '

                                 'to %(month_end)s %(day_end)s') % d

                else:

                    return _('From %(month_start)s %(day_start)s %(year_start)s '

                             'to %(month_end)s %(day_end)s %(year_end)s') % d

        else:

            return _('On %(month_start)s %(day_start)s') % d

    def as_vevent(self):

        vevent = vobject.newFromBehavior('vevent')

        vevent.add('uid').value = '%04d%02d%02d-%s@%s' % (self.date_start[:3] + (self.id,

            get_request().get_server().lower().split(':')[0].rstrip('.')))

        vevent.add('summary').value = unicode(self.title, 'iso-8859-15')

        vevent.add('dtstart').value = datetime.date(*self.date_start[:3])

        vevent.dtstart.value_param = 'DATE'

        if self.date_end:

            vevent.add('dtend').value = datetime.date(*self.date_end[:3])

            vevent.dtend.value_param = 'DATE'

        if self.description:

            vevent.add('description').value = unicode(self.description.strip(), 'iso-8859-15')

        if self.url:

            vevent.add('url').value = unicode(self.url, 'iso-8859-15')

        vevent.add('class').value = 'PUBLIC'

        return vevent

    def as_vcalendar(cls):

        cal = vobject.iCalendar()

        cal.add('prodid').value = '-//Entr\'ouvert//NON SGML Au Quotidien'

        for x in cls.select():

            cal.add(x.as_vevent())

        return cal.serialize()

    as_vcalendar = classmethod(as_vcalendar)

class RemoteCalendar(StorableObject):

    _names = 'remote_calendars'

    label = None

    url = None

    content = None

    events = None

    error = None  # (time, string, params)

    def download_and_parse(self):

        old_content = self.content

        try:

            self.content = urllib2.urlopen(self.url).read()

        except urllib2.HTTPError, e:

            self.error = (time.localtime(), N_('HTTP Error %s on download'), (e.code,))

            self.store()

            return

        except urllib2.URLError, e:

            self.error = (time.localtime(), N_('Error on download'), ())

            self.store()

            return

        if self.error:

            self.error = None

            self.store()

        if self.content == old_content:

            return

        self.events = []

        try:

            parsed_cal = vobject.readOne(self.content)

        except vobject.base.ParseError:

            self.error = (time.localtime(), N_('Failed to parse file'), ())

            self.store()

            return

        for vevent in parsed_cal.vevent_list:

            ev = Event()

            ev.title = vevent.summary.value.encode('iso-8859-15')

            try:

                ev.url = vevent.url.value.encode('iso-8859-15')

            except AttributeError:

                pass

            ev.date_start = vevent.dtstart.value.timetuple()

            if vevent.dtend:

                try:

                    ev.date_end = vevent.dtend.value.timetuple()

                except AttributeError:

                    pass

            try:

                ev.description = vevent.description.value.encode('iso-8859-15')

            except AttributeError:

                pass

            self.events.append(ev)

        self.store()

    def get_error_message(self):

        if not self.error:

            return None

        return '(%s) %s' % (misc.localstrftime(self.error[0]),

                _(self.error[1]) % self.error[2])

def update_remote_calendars(publisher):

    for source in RemoteCalendar.select():

        source.download_and_parse()

get_publisher_class().register_cronjob(CronJob(update_remote_calendars, minutes = [0]))