~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).
"""Browser code for the Launchpad root page."""

__metaclass__ = type
__all__ = [
    'LaunchpadRootIndexView',
    'LaunchpadSearchView',
    ]


import re
import time

import feedparser
from lazr.batchnavigator.z3batching import batch
from zope.app.form.interfaces import ConversionError
from zope.component import getUtility
from zope.interface import Interface
from zope.schema import TextLine
from zope.schema.interfaces import TooLong
from zope.schema.vocabulary import getVocabularyRegistry

from canonical.config import config
from canonical.launchpad import _
from lp.services.statistics.interfaces.statistic import (
    ILaunchpadStatisticSet,
    )
from canonical.launchpad.webapp import LaunchpadView
from canonical.launchpad.webapp.authorization import check_permission
from canonical.launchpad.webapp.batching import BatchNavigator
from canonical.launchpad.webapp.publisher import canonical_url
from canonical.launchpad.webapp.vhosts import allvhosts
from canonical.lazr.timeout import urlfetch
from lp.answers.interfaces.questioncollection import IQuestionSet
from lp.app.browser.launchpadform import (
    action,
    LaunchpadFormView,
    safe_action,
    )
from lp.app.errors import NotFoundError
from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.app.validators.name import sanitize_name
from lp.blueprints.interfaces.specification import ISpecificationSet
from lp.bugs.interfaces.bug import IBugSet
from lp.code.interfaces.branchcollection import IAllBranches
from lp.registry.browser.announcement import HasAnnouncementsView
from lp.registry.interfaces.person import IPersonSet
from lp.registry.interfaces.pillar import IPillarNameSet
from lp.registry.interfaces.product import IProductSet
from lp.services.googlesearch.interfaces import (
    GoogleResponseError,
    ISearchService,
    )
from lp.services.propertycache import cachedproperty


shipit_faq_url = 'http://www.ubuntu.com/getubuntu/shipit-faq'


class LaunchpadRootIndexView(HasAnnouncementsView, LaunchpadView):
    """An view for the default view of the LaunchpadRoot."""

    page_title = 'Launchpad'
    featured_projects = []
    featured_projects_top = None

    # Used by the footer to display the lp-arcana section.
    is_root_page = True

    @staticmethod
    def _get_day_of_year():
        """Calculate the number of the current day.

        This method gets overridden in tests to make the selection of the
        top featured project deterministic.
        """
        return time.gmtime()[7]

    def initialize(self):
        """Set up featured projects list and the top featured project."""
        super(LaunchpadRootIndexView, self).initialize()
        # The maximum number of projects to be displayed as defined by the
        # number of items plus one top featured project.
        self.featured_projects = list(
            getUtility(IPillarNameSet).featured_projects)
        self._setFeaturedProjectsTop()

    def _setFeaturedProjectsTop(self):
        """Set the top featured project and remove it from the list."""
        project_count = len(self.featured_projects)
        if project_count > 0:
            top_project = self._get_day_of_year() % project_count
            self.featured_projects_top = self.featured_projects.pop(
                top_project)

    @cachedproperty
    def apphomes(self):
        return {
            'answers': canonical_url(self.context, rootsite='answers'),
            'blueprints': canonical_url(self.context, rootsite='blueprints'),
            'bugs': canonical_url(self.context, rootsite='bugs'),
            'code': canonical_url(self.context, rootsite='code'),
            'translations': canonical_url(self.context,
                                          rootsite='translations'),
            'ubuntu': canonical_url(
                getUtility(ILaunchpadCelebrities).ubuntu),
            }

    @property
    def branch_count(self):
        """The total branch count of public branches in all of Launchpad."""
        return getUtility(IAllBranches).visibleByUser(None).count()

    @property
    def bug_count(self):
        """The total bug count in all of Launchpad."""
        return getUtility(ILaunchpadStatisticSet).value('bug_count')

    @property
    def project_count(self):
        """The total project count in all of Launchpad."""
        return getUtility(IProductSet).count_all()

    @property
    def translation_count(self):
        """The total count of translatable strings in all of Launchpad """
        return getUtility(ILaunchpadStatisticSet).value('pomsgid_count')

    @property
    def blueprint_count(self):
        """The total blueprint count in all of Launchpad."""
        return getUtility(ISpecificationSet).specification_count

    @property
    def answer_count(self):
        """The total blueprint count in all of Launchpad."""
        return getUtility(ILaunchpadStatisticSet).value('question_count')

    def getRecentBlogPosts(self):
        """Return the parsed feed of the most recent blog posts.

        It returns a list of dict with keys title, description, link and date.

        The date is formatted and the description which may contain HTML is
        sanitized.

        The number of blog posts to display is controlled through
        launchpad.homepage_recent_posts_count. The posts are fetched
        from the feed specified in launchpad.homepage_recent_posts_feed.

        Since the feed is parsed everytime, the template should cache this
        through memcached.

        FeedParser takes care of sanitizing the HTML contained in the feed.
        """
        # Use urlfetch which supports timeout
        try:
            data = urlfetch(config.launchpad.homepage_recent_posts_feed)
        except IOError:
            return []
        feed = feedparser.parse(data)
        posts = []
        max_count = config.launchpad.homepage_recent_posts_count
        # FeedParser takes care of HTML sanitisation.
        for entry in feed.entries[:max_count]:
            posts.append({
                'title': entry.title,
                'description': entry.description,
                'link': entry.link,
                'date': time.strftime('%d %b %Y', entry.updated_parsed),
                })
        return posts


class LaunchpadSearchFormView(LaunchpadView):
    """A view to display the global search form in any page."""
    id_suffix = '-secondary'
    text = None
    focusedElementScript = None
    form_wide_errors = None
    errors = None
    error_count = None
    error = None
    error_class = None

    @property
    def rooturl(self):
        """Return the site's root url."""
        return allvhosts.configs['mainsite'].rooturl


class LaunchpadPrimarySearchFormView(LaunchpadSearchFormView):
    """A view to display the global search form in the page."""
    id_suffix = ''

    @property
    def text(self):
        """The search text submitted to the context view."""
        return self.context.text

    @property
    def focusedElementScript(self):
        """The context view's focusedElementScript."""
        return self.context.focusedElementScript

    @property
    def form_wide_errors(self):
        """The context view's form_wide_errors."""
        return self.context.form_wide_errors

    @property
    def errors(self):
        """The context view's errors."""
        return self.context.errors

    @property
    def error_count(self):
        """The context view's error_count."""
        return self.context.error_count

    @property
    def error(self):
        """The context view's text field error."""
        return self.context.getFieldError('text')

    @property
    def error_class(self):
        """Return the 'error' if there is an error, or None."""
        if self.error:
            return 'error'
        return None


class ILaunchpadSearch(Interface):
    """The Schema for performing searches across all Launchpad."""

    text = TextLine(
        title=_('Search text'), required=False, max_length=250)


class LaunchpadSearchView(LaunchpadFormView):
    """A view to search for Launchpad pages and objects."""
    schema = ILaunchpadSearch
    field_names = ['text']

    shipit_keywords = set([
        'ubuntu', 'kubuntu', 'edubuntu',
        'ship', 'shipit', 'send', 'get', 'mail', 'free',
        'cd', 'cds', 'dvd', 'dvds', 'disc'])
    shipit_anti_keywords = set([
        'burn', 'burning', 'enable', 'error', 'errors', 'image', 'iso',
        'read', 'rip', 'write'])

    def __init__(self, context, request):
        """Initialize the view.

        Set the state of the search_params and matches.
        """
        super(LaunchpadSearchView, self).__init__(context, request)
        self.has_page_service = True
        self._bug = None
        self._question = None
        self._person_or_team = None
        self._pillar = None
        self._pages = None
        self.search_params = self._getDefaultSearchParams()
        # The Search Action should always run.
        self.request.form['field.actions.search'] = 'Search'

    def _getDefaultSearchParams(self):
        """Return a dict of the search param set to their default state."""
        return {
            'text': None,
            'start': 0,
            }

    def _updateSearchParams(self):
        """Sanitize the search_params and add the BatchNavigator params."""
        if self.search_params['text'] is not None:
            text = self.search_params['text'].strip()
            if text == '':
                self.search_params['text'] = None
            else:
                self.search_params['text'] = text
        request_start = self.request.get('start', self.search_params['start'])
        try:
            start = int(request_start)
        except (ValueError, TypeError):
            return
        self.search_params['start'] = start

    @property
    def text(self):
        """Return the text or None."""
        return self.search_params['text']

    @property
    def start(self):
        """Return the start index of the batch."""
        return self.search_params['start']

    @property
    def page_title(self):
        """Page title."""
        return self.page_heading

    @property
    def page_heading(self):
        """Heading to display above the search results."""
        if self.text is None:
            return 'Search Launchpad'
        else:
            return 'Pages matching "%s" in Launchpad' % self.text

    @property
    def batch_heading(self):
        """Heading to display in the batch navigation."""
        if self.has_exact_matches:
            return ('other page matching "%s"' % self.text,
                    'other pages matching "%s"' % self.text)
        else:
            return ('page matching "%s"' % self.text,
                    'pages matching "%s"' % self.text)

    @property
    def focusedElementScript(self):
        """Focus the first widget when there are no matches."""
        if self.has_matches:
            return None
        return super(LaunchpadSearchView, self).focusedElementScript()

    @property
    def bug(self):
        """Return the bug that matched the terms, or None."""
        return self._bug

    @property
    def question(self):
        """Return the question that matched the terms, or None."""
        return self._question

    @property
    def pillar(self):
        """Return the project that matched the terms, or None."""
        return self._pillar

    @property
    def person_or_team(self):
        """Return the person or team that matched the terms, or None."""
        return self._person_or_team

    @property
    def pages(self):
        """Return the pages that matched the terms, or None."""
        return self._pages

    @property
    def has_shipit(self):
        """Return True is the search text contains shipit keywords."""
        if self.text is None:
            return False
        terms = set(self.text.lower().split())
        anti_matches = self.shipit_anti_keywords.intersection(terms)
        if len(anti_matches) >= 1:
            return False
        matches = self.shipit_keywords.intersection(terms)
        return len(matches) >= 2

    @property
    def has_exact_matches(self):
        """Return True if something exactly matched the search terms."""
        kinds = (self.bug, self.question, self.pillar,
                 self.person_or_team, self.has_shipit)
        return self.containsMatchingKind(kinds)

    @property
    def shipit_faq_url(self):
        """The shipit FAQ URL."""
        return shipit_faq_url

    @property
    def has_matches(self):
        """Return True if something matched the search terms, or False."""
        kinds = (self.bug, self.question, self.pillar,
                 self.person_or_team, self.has_shipit, self.pages)
        return self.containsMatchingKind(kinds)

    @property
    def url(self):
        """Return the requested URL."""
        if 'QUERY_STRING' in self.request:
            query_string = self.request['QUERY_STRING']
        else:
            query_string = ''
        return self.request.getURL() + '?' + query_string

    def containsMatchingKind(self, kinds):
        """Return True if one of the items in kinds is not None, or False."""
        for kind in kinds:
            if kind is not None and kind is not False:
                return True
        return False

    def validate(self, data):
        """See `LaunchpadFormView`"""
        errors = list(self.errors)
        for error in errors:
            if isinstance(error, ConversionError):
                self.setFieldError(
                    'text', 'Can not convert your search term.')
            elif isinstance(error, unicode):
                continue
            elif (error.field_name == 'text'
                and isinstance(error.errors, TooLong)):
                self.setFieldError(
                    'text', 'The search text cannot exceed 250 characters.')

    @safe_action
    @action(u'Search', name='search')
    def search_action(self, action, data):
        """The Action executed when the user uses the search button.

        Saves the user submitted search parameters in an instance
        attribute.
        """
        self.search_params.update(**data)
        self._updateSearchParams()
        if self.text is None:
            return

        if self.start == 0:
            numeric_token = self._getNumericToken(self.text)
            if numeric_token is not None:
                try:
                    bug = getUtility(IBugSet).get(numeric_token)
                    if check_permission("launchpad.View", bug):
                        self._bug = bug
                except NotFoundError:
                    # Let self._bug remain None.
                    pass
                self._question = getUtility(IQuestionSet).get(numeric_token)

            name_token = self._getNameToken(self.text)
            if name_token is not None:
                self._person_or_team = self._getPersonOrTeam(name_token)
                self._pillar = self._getDistributionOrProductOrProjectGroup(
                    name_token)

        self._pages = self.searchPages(self.text, start=self.start)

    def _getNumericToken(self, text):
        """Return the first group of numbers in the search text, or None."""
        numeric_pattern = re.compile(r'(\d+)')
        match = numeric_pattern.search(text)
        if match is None:
            return None
        return match.group(1)

    def _getNameToken(self, text):
        """Return the search text as a Launchpad name.

        Launchpad names may contain ^[a-z0-9][a-z0-9\+\.\-]+$.
        See `valid_name_pattern`.
        """
        hypen_pattern = re.compile(r'[ _]')
        name = hypen_pattern.sub('-', text.strip().lower())
        return sanitize_name(name)

    def _getPersonOrTeam(self, name):
        """Return the matching active person or team."""
        person_or_team = getUtility(IPersonSet).getByName(name)
        if (person_or_team is not None
            and person_or_team.is_valid_person_or_team
            and check_permission('launchpad.View', person_or_team)):
            return person_or_team
        return None

    def _getDistributionOrProductOrProjectGroup(self, name):
        """Return the matching distribution, product or project, or None."""
        vocabulary_registry = getVocabularyRegistry()
        vocab = vocabulary_registry.get(
            None, 'DistributionOrProductOrProjectGroup')
        try:
            return vocab.getTermByToken(name).value
        except LookupError:
            return None

    def searchPages(self, query_terms, start=0):
        """Return the up to 20 pages that match the query_terms, or None.

        :param query_terms: The unescaped terms to query Google.
        :param start: The index of the page that starts the set of pages.
        :return: A GooglBatchNavigator or None.
        """
        if query_terms in [None, '']:
            return None
        google_search = getUtility(ISearchService)
        try:
            page_matches = google_search.search(
                terms=query_terms, start=start)
        except GoogleResponseError:
            # There was a connectivity or Google service issue that means
            # there is no data available at this moment.
            self.has_page_service = False
            return None
        if len(page_matches) == 0:
            return None
        navigator = GoogleBatchNavigator(
            page_matches, self.request, start=start)
        navigator.setHeadings(*self.batch_heading)
        return navigator


class WindowedList:
    """A list that contains a subset of items (a window) of a virtual list."""

    def __init__(self, window, start, total):
        """Create a WindowedList from a smaller list.

        :param window: The list with real items.
        :param start: An int, the list's starting index in the virtual list.
        :param total: An int, the total number of items in the virtual list.
        """
        self._window = window
        self._start = start
        self._total = total
        self._end = start + len(window)

    def __len__(self):
        """Return the length of the virtual list."""
        return self._total

    def __getitem__(self, key):
        """Return the key item or None if key belongs to the virtual list."""
        # When the key is a slice, return a list of items.
        if isinstance(key, (tuple, slice)):
            if isinstance(key, (slice)):
                indices = key.indices(len(self))
            else:
                indices = key
            return [self[index] for index in range(*indices)]
        # If the index belongs to the window return a real item.
        if key >= self._start and key < self._end:
            window_index = key - self._start
            return self._window[window_index]
        # Otherwise the index belongs to the virtual list.
        return None

    def __iter__(self):
        """Yield each item, or None if the index is virtual."""
        for index in range(0, self._total):
            yield self[index]


class WindowedListBatch(batch._Batch):
    """A batch class that does not include None objects when iterating."""

    def __iter__(self):
        """Iterate over objects that are not None."""
        for item in super(WindowedListBatch, self).__iter__():
            if item is not None:
                # Never yield None
                yield item

    def endNumber(self):
        """Return the end index of the batch, not including None objects."""
        # This class should know about the private _window attribute.
        # pylint: disable-msg=W0212
        return self.start + len(self.list._window)


class GoogleBatchNavigator(BatchNavigator):
    """A batch navigator with a fixed size of 20 items per batch."""

    _batch_factory = WindowedListBatch
    # Searches generally don't show the 'Last' link when there is a
    # good chance of getting over 100,000 results.
    show_last_link = False

    singular_heading = 'page'
    plural_heading = 'pages'

    def __init__(self, results, request, start=0, size=20, callback=None,
                 transient_parameters=None, force_start=False,
                 range_factory=None):
        """See `BatchNavigator`.

        :param results: A `PageMatches` object that contains the matching
            pages to iterate over.
        :param request: An `IBrowserRequest` that contains the form
            parameters.
        :param start: an int that represents the start of the current batch.
        :param size: The batch size is fixed to 20, The param is not used.
        :param callback: Not used.
        """
        results = WindowedList(results, start, results.total)
        super(GoogleBatchNavigator, self).__init__(results, request,
            start=start, size=size, callback=callback,
            transient_parameters=transient_parameters,
            force_start=force_start, range_factory=range_factory)

    def determineSize(self, size, batch_params_source):
        # Force the default and users requested sizes to 20.
        self.default_size = 20
        return 20