software-center 13.10-0ubuntu4 / usr / share / software-center / softwarecenter / db / database.py

# Copyright (C) 2009 Canonical
#
# Authors:
#  Michael Vogt
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; version 3.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

import locale
import logging
import os
import re
import string
import threading
import xapian
from softwarecenter.db.application import Application
from softwarecenter.db.pkginfo import get_pkg_info
import softwarecenter.paths

from gi.repository import GObject, Gio, GLib

from softwarecenter.utils import ExecutionTime
from softwarecenter.enums import (
    AVAILABLE_FOR_PURCHASE_MAGIC_CHANNEL_NAME,
    PkgStates,
    XapianValues
)

from softwarecenter.paths import XAPIAN_BASE_PATH_SOFTWARE_CENTER_AGENT
from gettext import gettext as _

LOG = logging.getLogger(__name__)


def get_reinstall_previous_purchases_query():
    """Return a query to get applications purchased

    :return: a xapian query to get all the apps that are purchased
    """
    # this query will give us all documents that have a purchase date != ""
    query = xapian.Query(xapian.Query.OP_VALUE_GE,
                         XapianValues.PURCHASED_DATE,
                         "1")
    return query


def parse_axi_values_file(filename="/var/lib/apt-xapian-index/values"):
    """ parse the apt-xapian-index "values" file and provide the
    information in the self._axi_values dict
    """
    axi_values = {}
    if not os.path.exists(filename):
        return axi_values
    for raw_line in open(filename):
        line = string.split(raw_line, "#", 1)[0]
        if line.strip() == "":
            continue
        (key, value) = line.split()
        axi_values[key] = int(value)
    return axi_values


class SearchQuery(list):
    """ a list wrapper for a search query. it can take a search string
        or a list of search strings

        It provides __eq__ to easily compare two search query lists
    """
    def __init__(self, query_string_or_list):
        if query_string_or_list is None:
            pass
        # turn single queries into a single item list
        elif isinstance(query_string_or_list, xapian.Query):
            self.append(query_string_or_list)
        else:
            self.extend(query_string_or_list)

    def __eq__(self, other):
        # turn single queries into a single item list
        if isinstance(other, xapian.Query):
            other = [other]
        q1 = [str(q) for q in self]
        q2 = [str(q) for q in other]
        return q1 == q2

    def __ne__(self, other):
        return not self.__eq__(other)

    def __repr__(self):
        return "[%s]" % ",".join([str(q) for q in self])


class LocaleSorter(xapian.KeyMaker):
    """ Sort in a locale friendly way by using locale.xtrxfrm """
    def __init__(self, db):
        super(LocaleSorter, self).__init__()
        self.db = db

    def __call__(self, doc):
        return locale.strxfrm(
            doc.get_value(self.db._axi_values["display_name"]))


class TopRatedSorter(xapian.KeyMaker):
    """ Sort using the top rated data """
    def __init__(self, db, review_loader):
        super(TopRatedSorter, self).__init__()
        self.db = db
        self.review_loader = review_loader

    def __call__(self, doc):
        app = Application(self.db.get_appname(doc),
                          self.db.get_pkgname(doc))
        stats = self.review_loader.get_review_stats(app)
        import xapian
        if stats:
            return xapian.sortable_serialise(stats.dampened_rating)
        return xapian.sortable_serialise(0)


class StoreDatabase(GObject.GObject):
    """thin abstraction for the xapian database with convenient functions"""

    # TRANSLATORS: List of "grey-listed" words separated with ";"
    # Do not translate this list directly. Instead,
    # provide a list of words in your language that people are likely
    # to include in a search but that should normally be ignored in
    # the search.
    SEARCH_GREYLIST_STR = _("app;application;package;program;programme;"
                            "suite;tool")

    # signal emitted
    __gsignals__ = {"reopen": (GObject.SIGNAL_RUN_FIRST,
                               GObject.TYPE_NONE,
                               ()),
                    "open": (GObject.SIGNAL_RUN_FIRST,
                             GObject.TYPE_NONE,
                             (GObject.TYPE_STRING,)),
                    }

    def __init__(self, pathname=None, cache=None):
        GObject.GObject.__init__(self)
        # initialize at creation time to avoid spurious AttributeError
        self._use_agent = False
        self._use_axi = False

        if pathname is None:
            pathname = softwarecenter.paths.XAPIAN_PATH
        self._db_pathname = pathname
        if cache is None:
            cache = get_pkg_info()
        self._aptcache = cache
        self._additional_databases = []
        # the xapian values as read from /var/lib/apt-xapian-index/values
        self._axi_values = {}
        # we open one db per thread, thread names are reused eventually
        # so no memory leak
        self._db_per_thread = {}
        self._parser_per_thread = {}
        self._axi_stamp_monitor = None

    @property
    def xapiandb(self):
        """ returns a per thread db """
        thread_name = threading.current_thread().name
        if not thread_name in self._db_per_thread:
            self._db_per_thread[thread_name] = self._get_new_xapiandb()
        return self._db_per_thread[thread_name]

    @property
    def xapian_parser(self):
        """ returns a per thread query parser """
        thread_name = threading.current_thread().name
        if not thread_name in self._parser_per_thread:
            xapian_parser = self._get_new_xapian_parser()
            self._parser_per_thread[thread_name] = xapian_parser
        return self._parser_per_thread[thread_name]

    def _get_new_xapiandb(self):
        xapiandb = xapian.Database(self._db_pathname)
        if self._use_axi:
            try:
                axi = xapian.Database(
                    softwarecenter.paths.APT_XAPIAN_INDEX_DB_PATH)
                xapiandb.add_database(axi)
            except:
                LOG.exception("failed to add apt-xapian-index")
        if (self._use_agent and
                os.path.exists(XAPIAN_BASE_PATH_SOFTWARE_CENTER_AGENT)):
            try:
                sca = xapian.Database(XAPIAN_BASE_PATH_SOFTWARE_CENTER_AGENT)
                xapiandb.add_database(sca)
            except Exception as e:
                logging.warn("failed to add sca db %s" % e)
        for db in self._additional_databases:
            xapiandb.add_database(db)
        return xapiandb

    def _get_new_xapian_parser(self):
        xapian_parser = xapian.QueryParser()
        xapian_parser.set_database(self.xapiandb)
        xapian_parser.add_boolean_prefix("pkg", "XP")
        xapian_parser.add_boolean_prefix("pkg", "AP")
        xapian_parser.add_boolean_prefix("mime", "AM")
        xapian_parser.add_boolean_prefix("section", "XS")
        xapian_parser.add_boolean_prefix("origin", "XOC")
        xapian_parser.add_prefix("pkg_wildcard", "XP")
        xapian_parser.add_prefix("pkg_wildcard", "XPM")
        xapian_parser.add_prefix("pkg_wildcard", "AP")
        xapian_parser.add_prefix("pkg_wildcard", "APM")
        xapian_parser.set_default_op(xapian.Query.OP_AND)
        return xapian_parser

    def open(self, pathname=None, use_axi=True, use_agent=True):
        """ open the database """
        LOG.debug("open() database: path=%s use_axi=%s "
                          "use_agent=%s" % (pathname, use_axi, use_agent))
        if pathname:
            self._db_pathname = pathname
        # clean existing DBs on open
        self._db_per_thread = {}
        self._parser_per_thread = {}
        # add the apt-xapian-database for here (we don't do this
        # for now as we do not have a good way to integrate non-apps
        # with the UI)
        self.nr_databases = 0
        self._use_axi = use_axi
        self._axi_values = {}
        self._use_agent = use_agent
        if use_axi:
            if self._axi_stamp_monitor:
                self._axi_stamp_monitor.disconnect_by_func(
                    self._on_axi_stamp_changed)
            self._axi_values = parse_axi_values_file()
            self.nr_databases += 1
            # mvo: we could monitor changes in
            #       softwarecenter.paths.APT_XAPIAN_INDEX_DB_PATH here too
            #       as its a text file that points to the current DB
            #       *if* we do that, we need to change the event == ATTRIBUTE
            #       change in _on_axi_stamp_changed too
            self._axi_stamp = Gio.File.new_for_path(
                softwarecenter.paths.APT_XAPIAN_INDEX_UPDATE_STAMP_PATH)
            self._timeout_id = None
            self._axi_stamp_monitor = self._axi_stamp.monitor_file(0, None)
            self._axi_stamp_monitor.connect(
                "changed", self._on_axi_stamp_changed)
        if use_agent:
            self.nr_databases += 1
        # additional dbs
        for db in self._additional_databases:
            self.nr_databases += 1
        self.emit("open", self._db_pathname)

    def _on_axi_stamp_changed(self, monitor, afile, otherfile, event):
        # we only care about the utime() update from update-a-x-i
        if not event == Gio.FileMonitorEvent.ATTRIBUTE_CHANGED:
            return
        LOG.debug("afile '%s' changed" % afile)
        if self._timeout_id:
            GLib.source_remove(self._timeout_id)
            self._timeout_id = None
        self._timeout_id = GLib.timeout_add(500, self.reopen)

    def add_database(self, database):
        self._additional_databases.append(database)
        self.xapiandb.add_database(database)
        self.reopen()

    def del_database(self, database):
        self._additional_databases.remove(database)
        self.reopen()

    def schema_version(self):
        """Return the version of the database layout

           This is useful to ensure we force a rebuild if it's
           older than what we expect
        """
        return self.xapiandb.get_metadata("db-schema-version")

    def reopen(self):
        """ reopen the database """
        LOG.debug("reopen() database")
        self.open(use_axi=self._use_axi, use_agent=self._use_agent)
        self.emit("reopen")

    @property
    def popcon_max(self):
        popcon_max = xapian.sortable_unserialise(self.xapiandb.get_metadata(
            "popcon_max_desktop"))
        assert popcon_max > 0
        return popcon_max

    def get_query_for_pkgnames(self, pkgnames):
        """ return a xapian query that matches exactly the list of pkgnames """
        enquire = xapian.Enquire(self.xapiandb)
        query = xapian.Query()
        for pkgname in pkgnames:
            # even on the raspberry pi this query is super quick (~0.003s)
            with ExecutionTime("de-dup query_for_pkgnames"):
                tmp_query = xapian.Query("AP" + pkgname)
                enquire.set_query(tmp_query)
                result = enquire.get_mset(0, 1)
            # see bug #1043159, we need to ensure that we de-duplicate
            # when there is a pkg and a app (e.g. from the s-c-agent) in the db
            if len(result) == 1:
                query = xapian.Query(xapian.Query.OP_OR,
                                     query,
                                     xapian.Query("AP" + pkgname))
            else:
                query = xapian.Query(xapian.Query.OP_OR,
                                     query,
                                     xapian.Query("XP" + pkgname))
        return query

    def get_query_list_from_search_entry(self, search_term,
                                         category_query=None):
        """ get xapian.Query from a search term string and a limit the
            search to the given category
        """
        def _add_category_to_query(query):
            """ helper that adds the current category to the query"""
            if not category_query:
                return query
            return xapian.Query(xapian.Query.OP_AND,
                                category_query,
                                query)
        # empty query returns a query that matches nothing (for performance
        # reasons)
        if search_term == "" and category_query is None:
            return SearchQuery(xapian.Query())
        # we cheat and return a match-all query for single letter searches
        if len(search_term) < 2:
            return SearchQuery(_add_category_to_query(xapian.Query("")))

        # check if there is a ":" in the search, if so, it means the user
        # is using a xapian prefix like "pkg:" or "mime:" and in this case
        # we do not want to alter the search term (as application is in the
        # greylist but a common mime-type prefix)
        if not ":" in search_term:
            # filter query by greylist (to avoid overly generic search terms)
            orig_search_term = search_term
            for item in self.SEARCH_GREYLIST_STR.split(";"):
                (search_term, n) = re.subn('\\b%s\\b' % item, '', search_term)
                if n:
                    LOG.debug("greylist changed search term: '%s'" %
                        search_term)
        # restore query if it was just greylist words
        if search_term == '':
            LOG.debug("grey-list replaced all terms, restoring")
            search_term = orig_search_term
        # we have to strip the leading and trailing whitespaces to avoid having
        # different results for e.g. 'font ' and 'font' (LP: #506419)
        search_term = search_term.strip()
        # get a pkg query
        if "," in search_term:
            pkg_query = self.get_query_for_pkgnames(search_term.split(","))
        else:
            pkg_query = xapian.Query()
            for term in search_term.split():
                pkg_query = xapian.Query(xapian.Query.OP_OR,
                                         xapian.Query("XP" + term),
                                         pkg_query)
        pkg_query = _add_category_to_query(pkg_query)

        # get a search query
        if not ':' in search_term:  # ie, not a mimetype query
            # we need this to work around xapian oddness
            search_term = search_term.replace('-', '_')
        fuzzy_query = self.xapian_parser.parse_query(search_term,
                                           xapian.QueryParser.FLAG_PARTIAL |
                                           xapian.QueryParser.FLAG_BOOLEAN)
        # if the query size goes out of hand, omit the FLAG_PARTIAL
        # (LP: #634449)
        if fuzzy_query.get_length() > 1000:
            fuzzy_query = self.xapian_parser.parse_query(search_term,
                                            xapian.QueryParser.FLAG_BOOLEAN)
        # now add categories
        fuzzy_query = _add_category_to_query(fuzzy_query)
        return SearchQuery([pkg_query, fuzzy_query])

    def get_matches_from_query(self, query, start=0, end=-1, category=None):
        enquire = xapian.Enquire(self.xapiandb)
        if isinstance(query, str):
            if query == "":
                query = xapian.Query("")
            else:
                query = self.xapian_parser.parse_query(query)
        if category:
            query = xapian.Query(xapian.Query.OP_AND, category.query, query)
        enquire.set_query(query)
        if end == -1:
            end = len(self)
        return enquire.get_mset(start, end)

    def get_docs_from_query(self, query, start=0, end=-1, category=None):
        matches = self.get_matches_from_query(query, start, end, category)
        return [m.document for m in matches]

    def get_spelling_correction(self, search_term):
        # get a search query
        if not ':' in search_term:  # ie, not a mimetype query
            # we need this to work around xapian oddness
            search_term = search_term.replace('-', '_')
        self.xapian_parser.parse_query(
            search_term, xapian.QueryParser.FLAG_SPELLING_CORRECTION)
        return self.xapian_parser.get_corrected_query_string()

    def get_most_popular_applications_for_mimetype(self, mimetype,
                                                   only_uninstalled=True,
                                                   num=3):
        """ return a list of the most popular applications for the given
            mimetype
        """
        # sort by popularity by default
        enquire = xapian.Enquire(self.xapiandb)
        enquire.set_sort_by_value_then_relevance(XapianValues.POPCON)
        # query mimetype
        query = xapian.Query("AM%s" % mimetype)
        enquire.set_query(query)
        # mset just needs to be "big enough"
        matches = enquire.get_mset(0, 100)
        apps = []
        for match in matches:
            doc = match.document
            app = Application(self.get_appname(doc), self.get_pkgname(doc),
                              popcon=self.get_popcon(doc))
            if only_uninstalled:
                if app.get_details(self).pkg_state == PkgStates.UNINSTALLED:
                    apps.append(app)
            else:
                apps.append(app)
            if len(apps) == num:
                break
        return apps

    def get_summary(self, doc):
        """ get human readable summary of the given document """
        summary = doc.get_value(XapianValues.SUMMARY)
        channel = doc.get_value(XapianValues.ARCHIVE_CHANNEL)
        # if we do not have the summary in the xapian db, get it
        # from the apt cache
        if not summary and self._aptcache.ready:
            pkgname = self.get_pkgname(doc)
            if (pkgname in self._aptcache and
                    self._aptcache[pkgname].candidate):
                return self._aptcache[pkgname].candidate.summary
            elif channel:
                # FIXME: print something if available for our arch
                pass
        return summary

    def get_application(self, doc):
        """ Return a application from a xapian document """
        appname = self.get_appname(doc)
        pkgname = self.get_pkgname(doc)
        iconname = self.get_iconname(doc)
        return Application(appname, pkgname, iconname)

    def get_pkgname(self, doc):
        """ Return a packagename from a xapian document """
        pkgname = doc.get_value(XapianValues.PKGNAME)
        # if there is no value it means we use the apt-xapian-index
        # that stores the pkgname in the data field or as a value
        if not pkgname:
            # the doc says that get_value() is quicker than get_data()
            # so we use that if we have a updated DB, otherwise
            # fallback to the old way (the xapian DB may not yet be rebuilt)
            if self._axi_values and "pkgname" in self._axi_values:
                pkgname = doc.get_value(self._axi_values["pkgname"])
            else:
                pkgname = doc.get_data()
        return pkgname

    def get_appname(self, doc):
        """ Return a appname from a xapian document, or None if
            a value for appname cannot be found in the document
         """
        return doc.get_value(XapianValues.APPNAME)

    def get_iconname(self, doc):
        """ Return the iconname from the xapian document """
        iconname = doc.get_value(XapianValues.ICON)
        return iconname

    def get_desktopfile(self, doc):
        """ Return the desktopfile from the xapian document """
        desktopf = doc.get_value(XapianValues.DESKTOP_FILE)
        return desktopf

    def pkg_in_category(self, pkgname, cat_query):
        """ Return True if the given pkg is in the given category """
        pkg_query1 = xapian.Query("AP" + pkgname)
        pkg_query2 = xapian.Query("XP" + pkgname)
        pkg_query = xapian.Query(xapian.Query.OP_OR, pkg_query1, pkg_query2)
        pkg_and_cat_query = xapian.Query(xapian.Query.OP_AND, pkg_query,
            cat_query)
        enquire = xapian.Enquire(self.xapiandb)
        enquire.set_query(pkg_and_cat_query)
        matches = enquire.get_mset(0, len(self))
        if matches:
            return True
        return False

    def get_apps_for_pkgname(self, pkgname):
        """ Return set of docids with the matching applications for the
            given pkgname """
        result = set()
        for m in self.xapiandb.postlist("AP" + pkgname):
            result.add(m.docid)
        return result

    def get_icon_download_url(self, doc):
        """ Return the url of the icon or None """
        url = doc.get_value(XapianValues.ICON_URL)
        return url

    def get_popcon(self, doc):
        """ Return a popcon value from a xapian document """
        popcon_raw = doc.get_value(XapianValues.POPCON)
        if popcon_raw:
            popcon = xapian.sortable_unserialise(popcon_raw)
        else:
            popcon = 0
        return popcon

    def get_xapian_document(self, appname, pkgname):
        """Get the matching xapian document for appname, pkgname.

        If no document is found, raise a IndexError.

        """
        #LOG.debug("get_xapian_document app='%s' pkg='%s'" % (appname,
        #    pkgname))
        # first search for appname in the app-install-data namespace
        for m in self.xapiandb.postlist("AA" + appname):
            doc = self.xapiandb.get_document(m.docid)
            if doc.get_value(XapianValues.PKGNAME) == pkgname:
                return doc
        # then search for pkgname in the app-install-data namespace
        for m in self.xapiandb.postlist("AP" + pkgname):
            doc = self.xapiandb.get_document(m.docid)
            if doc.get_value(XapianValues.PKGNAME) == pkgname:
                return doc
        # then look for matching packages from a-x-i
        for m in self.xapiandb.postlist("XP" + pkgname):
            doc = self.xapiandb.get_document(m.docid)
            return doc
        # no matching document found
        raise IndexError("No app '%s' for '%s' in database" % (appname,
            pkgname))

    def is_pkgname_known(self, pkgname):
        """Check if 'pkgname' is known to this database.

        Note that even if this function returns True, it may mean that the
        package needs to be purchased first or is available in a
        not-yet-enabled source.

        """
        # check cache first, then our own database
        return (pkgname in self._aptcache or
                any(self.xapiandb.postlist("AP" + pkgname)))

    def is_appname_duplicated(self, appname):
        """Check if the given appname is stored multiple times in the db.

        This can happen for generic names like "Terminal".

        """
        for (i, m) in enumerate(self.xapiandb.postlist("AA" + appname)):
            if i > 0:
                return True
        return False

    def get_installed_purchased_packages(self):
        """ return a set() of packagenames of purchased apps that are
            currently installed
        """
        for_purchase_query = xapian.Query(
            "AH" + AVAILABLE_FOR_PURCHASE_MAGIC_CHANNEL_NAME)
        enquire = xapian.Enquire(self.xapiandb)
        enquire.set_query(for_purchase_query)
        matches = enquire.get_mset(0, self.xapiandb.get_doccount())
        installed_purchased_packages = set()
        for m in matches:
            pkgname = self.get_pkgname(m.document)
            if (pkgname in self._aptcache and
                    self._aptcache[pkgname].is_installed):
                installed_purchased_packages.add(pkgname)
        return installed_purchased_packages

    def get_origins_from_db(self):
        """ return all origins available in the current database """
        origins = set()
        for term in self.xapiandb.allterms("XOO"):
            if term.term[3:]:
                origins.add(term.term[3:])
        return list(origins)

    def get_exact_matches(self, pkgnames=[]):
        """Returns a list of fake MSetItems. If the pkgname is available, then
           MSetItem.document is pkgnames proper xapian document. If the pkgname
           is not available, then MSetItem is actually an Application.
        """
        matches = []
        for pkgname in pkgnames:
            app = Application('', pkgname.split('?')[0])
            if '?' in pkgname:
                app.request = pkgname.split('?')[1]
            match = app
            for m in self.xapiandb.postlist("XP" + app.pkgname):
                match = self.xapiandb.get_document(m.docid)
            for m in self.xapiandb.postlist("AP" + app.pkgname):
                match = self.xapiandb.get_document(m.docid)
            matches.append(FakeMSetItem(match))
        return matches

    def __len__(self):
        """return the doc count of the database"""
        return self.xapiandb.get_doccount()

    def __iter__(self):
        """ support iterating over the documents """
        for it in self.xapiandb.postlist(""):
            doc = self.xapiandb.get_document(it.docid)
            yield doc


class FakeMSetItem(object):
    def __init__(self, doc):
        self.document = doc

if __name__ == "__main__":
    import apt
    import sys

    db = StoreDatabase("/var/cache/software-center/xapian", apt.Cache())
    db.open()

    if len(sys.argv) < 2:
        search = "apt,apport"
    else:
        search = sys.argv[1]
    query = db.get_query_list_from_search_entry(search)
    print(query)
    enquire = xapian.Enquire(db.xapiandb)
    enquire.set_query(query)
    matches = enquire.get_mset(0, len(db))
    for m in matches:
        doc = m.document
        print(doc.get_data())

    # test origin
    query = xapian.Query("XOL" + "Ubuntu")
    enquire = xapian.Enquire(db.xapiandb)
    enquire.set_query(query)
    matches = enquire.get_mset(0, len(db))
    print("Ubuntu origin: %s" % len(matches))