~launchpad-pqm/launchpad/devel

# Copyright 2009 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

"""Base classes for feeds.

Supported feeds include Atom, Javascript, and HTML-snippets.
Future support may include feeds such as sparklines.
"""

__metaclass__ = type

__all__ = [
    'FeedBase',
    'FeedEntry',
    'FeedPerson',
    'FeedTypedData',
    'MINUTES',
    ]

import operator
import os
import time
from urlparse import urljoin
from xml.sax.saxutils import escape as xml_escape

from BeautifulSoup import BeautifulSoup
from z3c.ptcompat import ViewPageTemplateFile
from zope.component import getUtility
from zope.datetime import rfc1123_date
from zope.interface import implements

from lp.services.config import config
from lp.services.feeds.interfaces.feed import (
    IFeed,
    IFeedEntry,
    IFeedPerson,
    IFeedTypedData,
    UnsupportedFeedFormat,
    )
from lp.services.propertycache import cachedproperty
from lp.services.utils import utc_now
from lp.services.webapp import (
    canonical_url,
    LaunchpadView,
    urlappend,
    urlparse,
    )
# XXX: bac 2007-09-20 bug=153795: modules in canonical.lazr should not import
# from canonical.launchpad, but we're doing it here as an expediency to get a
# working prototype.
from lp.services.webapp.interfaces import ILaunchpadRoot
from lp.services.webapp.vhosts import allvhosts


SUPPORTED_FEEDS = ('.atom', '.html')
MINUTES = 60 # Seconds in a minute.


class FeedBase(LaunchpadView):
    """See `IFeed`.

    Base class for feeds.
    """

    implements(IFeed)

    # convert to seconds
    max_age = config.launchpad.max_feed_cache_minutes * MINUTES
    quantity = 25
    items = None
    rootsite = 'mainsite'
    template_files = {'atom': 'templates/feed-atom.pt',
                      'html': 'templates/feed-html.pt'}

    def __init__(self, context, request):
        super(FeedBase, self).__init__(context, request)
        self.format = self.feed_format
        self.root_url = canonical_url(getUtility(ILaunchpadRoot),
                                      rootsite=self.rootsite)

    @property
    def title(self):
        """See `IFeed`."""
        raise NotImplementedError

    @property
    def link_self(self):
        """See `IFeed`."""

        # The self link is the URL for this particular feed.  For example:
        # http://feeds.launchpad.net/ubuntu/announcments.atom
        path = "%s.%s" % (self.feedname, self.format)
        return urlappend(canonical_url(self.context, rootsite="feeds"),
                         path)

    @property
    def site_url(self):
        """See `IFeed`."""
        return allvhosts.configs['mainsite'].rooturl[:-1]

    @property
    def link_alternate(self):
        """See `IFeed`."""
        return canonical_url(self.context, rootsite=self.rootsite)

    @property
    def feed_id(self):
        """See `IFeed`.

        Override this method if the context used does not create a
        meaningful id.
        """
        # Get the creation date, if available.  Otherwise use a fixed date, as
        # allowed by the RFC.
        if getattr(self.context, 'datecreated', None) is not None:
            datecreated = self.context.datecreated.date().isoformat()
        elif getattr(self.context, 'date_created', None) is not None:
            datecreated = self.context.date_created.date().isoformat()
        else:
            datecreated = "2008"
        url_path = urlparse(self.link_alternate)[2]
        if self.rootsite != 'mainsite':
            id_ = 'tag:launchpad.net,%s:/%s%s' % (
                datecreated,
                self.rootsite,
                url_path)
        else:
            id_ = 'tag:launchpad.net,%s:%s' % (
                datecreated,
                url_path)
        return id_

    def getItems(self):
        """See `IFeed`."""
        if self.items is None:
            self.items = self._getItemsWorker()
        return self.items

    def _getItemsWorker(self):
        """Create the list of items.

        Called by getItems which may cache the results.  The caching is
        necessary since `getItems` is called multiple times in the course of
        constructing a single feed and pulling together the list of items is
        potentially expensive.
        """
        raise NotImplementedError

    @property
    def feed_format(self):
        """See `IFeed`."""
        # If the full URL is http://feeds.launchpad.dev/announcements.atom/foo
        # getURL() will return http://feeds.launchpad.dev/announcements.atom
        # when traversing the feed, which will allow os.path.splitext()
        # to split off ".atom" correctly.
        path = self.request.getURL()
        extension = os.path.splitext(path)[1]
        if extension in SUPPORTED_FEEDS:
            return extension[1:]
        else:
            raise UnsupportedFeedFormat('%s is not supported' % path)

    @property
    def logo(self):
        """See `IFeed`."""
        raise NotImplementedError

    @property
    def icon(self):
        """See `IFeed`."""
        return "%s/@@/launchpad" % self.site_url

    @cachedproperty
    def date_updated(self):
        """See `IFeed`."""
        sorted_items = sorted(self.getItems(),
                              key=operator.attrgetter('last_modified'),
                              reverse=True)
        if len(sorted_items) == 0:
            # datetime.isoformat() doesn't place the necessary "+00:00"
            # for the feedvalidator's check of the iso8601 date format
            # unless a timezone is specified with tzinfo.
            return utc_now()
        last_modified = sorted_items[0].last_modified
        if last_modified is None:
            raise AssertionError, 'All feed entries require a date updated.'
        return last_modified

    def render(self):
        """See `IFeed`."""
        expires = rfc1123_date(time.time() + self.max_age)
        if self.date_updated is not None:
            last_modified = rfc1123_date(
                time.mktime(self.date_updated.timetuple()))
        else:
            last_modified = rfc1123_date(time.time())
        response = self.request.response
        response.setHeader('Expires', expires)
        response.setHeader('Cache-Control', 'max-age=%d' % self.max_age)
        response.setHeader('X-Cache-Control', 'max-age=%d' % self.max_age)
        response.setHeader('Last-Modified', last_modified)

        if self.format == 'atom':
            return self.renderAtom()
        elif self.format == 'html':
            return self.renderHTML()
        else:
            raise UnsupportedFeedFormat("Format %s is not supported" %
                                        self.format)

    def renderAtom(self):
        """See `IFeed`."""
        self.request.response.setHeader('content-type',
                                        'application/atom+xml;charset=utf-8')
        template_file = ViewPageTemplateFile(self.template_files['atom'])
        result = template_file(self)
        # XXX EdwinGrubbs 2008-01-10 bug=181903
        # Zope3 requires the content-type to start with "text/" if
        # the result is a unicode object.
        return result.encode('utf-8')

    def renderHTML(self):
        """See `IFeed`."""
        return ViewPageTemplateFile(self.template_files['html'])(self)


class FeedEntry:
    """See `IFeedEntry`.

    An individual entry for a feed.
    """

    implements(IFeedEntry)

    def __init__(self,
                 title,
                 link_alternate,
                 date_created,
                 date_updated,
                 date_published=None,
                 authors=None,
                 contributors=None,
                 content=None,
                 id_=None,
                 generator=None,
                 logo=None,
                 icon=None):
        self.title = title
        self.link_alternate = link_alternate
        self.content = content
        self.date_created = date_created
        self.date_updated = date_updated
        self.date_published = date_published
        if date_updated is None:
            raise AssertionError, 'date_updated is required by RFC 4287'
        if authors is None:
            authors = []
        self.authors = authors
        self.contributors = contributors
        if id_ is None:
            self.id = self.construct_id()
        else:
            self.id = id_

    @property
    def last_modified(self):
        if self.date_published is not None:
            return max(self.date_published, self.date_updated)
        return self.date_updated

    def construct_id(self):
        url_path = urlparse(self.link_alternate)[2]
        return 'tag:launchpad.net,%s:%s' % (
            self.date_created.date().isoformat(),
            url_path)


class FeedTypedData:
    """Data for a feed that includes its type."""

    implements(IFeedTypedData)

    content_types = ['text', 'html', 'xhtml']

    def __init__(self, content, content_type='text', root_url=None):
        self._content = content
        if content_type not in self.content_types:
            raise UnsupportedFeedFormat("%s: is not valid" % content_type)
        self.content_type = content_type
        self.root_url = root_url

    @property
    def content(self):
        if (self.content_type in ('html', 'xhtml') and
            self.root_url is not None):
            # Unqualified hrefs must be qualified using the original subdomain
            # or they will try be served from http://feeds.launchpad.net,
            # which will not work.
            soup = BeautifulSoup(self._content)
            a_tags = soup.findAll('a')
            for a_tag in a_tags:
                if a_tag['href'].startswith('/'):
                    a_tag['href'] = urljoin(self.root_url, a_tag['href'])
            altered_content = unicode(soup)
        else:
            altered_content = self._content

        if self.content_type in ('text', 'html'):
            altered_content = xml_escape(altered_content)
        elif self.content_type == 'xhtml':
            soup = BeautifulSoup(
                altered_content,
                convertEntities=BeautifulSoup.HTML_ENTITIES)
            altered_content = unicode(soup)
        return altered_content


class FeedPerson:
    """See `IFeedPerson`.

    If this class is consistently used we will not accidentally leak email
    addresses.
    """

    implements(IFeedPerson)

    def __init__(self, person, rootsite):
        self.name = person.displayname
        # We don't want to disclose email addresses in public feeds.
        self.email = None
        self.uri = canonical_url(person, rootsite=rootsite)