{Ninja-Shell}

Filename: //lib//python3/dist-packages///debian/debtags.py

""" Facilities to work with debtags - tags for Debian packages """

# Copyright (C) 2006-2007  Enrico Zini <enrico@enricozini.org>
#
# 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, either version 3 of the License, or
# (at your option) any later version.
#
# 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, see <http://www.gnu.org/licenses/>.

from __future__ import absolute_import, print_function

import re
import six

try:
    import cPickle as pickle    # type: ignore
except ImportError:
    import pickle               # type: ignore

try:
    # pylint: disable=unused-import
    from typing import (
        Callable,
        Dict,
        IO,
        Iterable,
        Iterator,
        List,
        Optional,
        Set,
        Text,
        Tuple,
    )
    PkgTagDbType = Dict[str, Set[str]]
    TagPkgDbType = Dict[str, Set[str]]
    PkgFilterType = Callable[[Text], bool]
    TagFilterType = Callable[[Text], bool]
    PkgTagFilterType = Callable[[Tuple[Text, Set[Text]]], bool]
except ImportError:
    # Lack of typing is not important at runtime
    pass

from debian.deprecation import function_deprecated_by

def parse_tags(input_data):
    # type: (Iterator[Text]) -> Iterator[Tuple[Set[str], Set[str]]]
    lre = re.compile(r"^(.+?)(?::?\s*|:\s+(.+?)\s*)$")
    for line in input_data:
        # Is there a way to remove the last character of a line that does not
        # make a copy of the entire line?
        m = lre.match(line)
        if not m:
            continue

pkgs = set(m.group(1).split(', '))
        if m.group(2):
            tags = set(m.group(2).split(', '))
        else:
            tags = set()
        yield pkgs, tags

parseTags = function_deprecated_by(parse_tags)

def read_tag_database(input_data):
    # type: (Iterator[Text]) -> PkgTagDbType
    """Read the tag database, returning a pkg->tags dictionary"""
    db = {}   # type: PkgTagDbType
    for pkgs, tags in parse_tags(input_data):
        # Create the tag set using the native set
        for p in pkgs:
            db[p] = tags.copy()
    return db

readTagDatabase = function_deprecated_by(read_tag_database)

def read_tag_database_reversed(input_data):
    # type: (Iterator[Text]) -> TagPkgDbType
    """Read the tag database, returning a tag->pkgs dictionary"""
    db = {}    # type: TagPkgDbType
    for pkgs, tags in parse_tags(input_data):
        # Create the tag set using the native set
        for tag in tags:
            if tag in db:
                db[tag] |= pkgs
            else:
                db[tag] = pkgs.copy()
    return db

readTagDatabaseReversed = function_deprecated_by(read_tag_database_reversed)

def read_tag_database_both_ways(
        input_data,  # type: Iterator[Text]
        tag_filter=None,  # type: Optional[TagFilterType]
    ):
    # type: (...) -> Tuple[PkgTagDbType, TagPkgDbType]
    "Read the tag database, returning a pkg->tags and a tag->pkgs dictionary"
    db = {}   # type: PkgTagDbType
    dbr = {}   # type: TagPkgDbType
    for pkgs, tags in parse_tags(input_data):
        # Create the tag set using the native set
        if tag_filter is None:
            tags = set(tags)
        else:
            tags = set(filter(tag_filter, tags))
        for pkg in pkgs:
            db[pkg] = tags.copy()
        for tag in tags:
            if tag in dbr:
                dbr[tag] |= pkgs
            else:
                dbr[tag] = pkgs.copy()
    return db, dbr

readTagDatabaseBothWays = function_deprecated_by(read_tag_database_both_ways)

def reverse(db):
    # type: (PkgTagDbType) -> TagPkgDbType
    """Reverse a tag database, from package -> tags to tag->packages"""
    res = {}   # type: Dict[str, Set[str]]
    for pkg, tags in db.items():
        for tag in tags:
            if tag not in res:
                res[tag] = set()
            res[tag].add(pkg)
    return res

def output(db):
    # type: (PkgTagDbType) -> None
    "Write the tag database"
    for pkg, tags in db.items():
        # Using % here seems awkward to me, but if I use calls to
        # sys.stdout.write it becomes a bit slower
        print("%s:" % (pkg), ", ".join(tags))

def relevance_index_function(full, sub):   #type: ignore
    #return (float(sub.card(tag)) / float(sub.tag_count())) / \
    #       (float(full.card(tag)) / float(full.tag_count()))
    #return sub.card(tag) * full.card(tag) / sub.tag_count()

# New cardinality divided by the old cardinality
    #return float(sub.card(tag)) / float(full.card(tag))

## Same as before, but weighted by the relevance the tag had in the
    ## full collection, to downplay the importance of rare tags
    #return float(sub.card(tag) * full.card(tag)) / float(full.card(tag) * full.tag_count())
    # Simplified version:
    # return float(sub.card(tag)) / float(full.tag_count())

# Weighted by the square root of the relevance, to downplay the very
    # common tags a bit
    # return lambda tag: float(sub.card(tag)) / float(full.card(tag)) *
    # math.sqrt(full.card(tag) / float(full.tag_count()))
    # return lambda tag: float(sub.card(tag)) / float(full.card(tag)) *
    # math.sqrt(full.card(tag) / float(full.package_count()))
    # One useless factor removed, and simplified further, thanks to Benjamin Mesing
    return lambda tag: float(sub.card(tag)**2) / float(full.card(tag))

# The difference between how many packages are in and how many packages are out
    # (problems: tags that mean many different things can be very much out
    # as well.  In the case of 'image editor', for example, there will be
    # lots of editors not for images in the outside group.
    # It is very, very good for nonambiguous keywords like 'image'.
    # return lambda tag: 2 * sub.card(tag) - full.card(tag)
    # Same but it tries to downplay the 'how many are out' value in the
    # case of popular tags, to mitigate the 'there will always be popular
    # tags left out' cases.  Does not seem to be much of an improvement.
    # return lambda tag: sub.card(tag) - float(full.card(tag) - sub.card(tag))/
    # (math.sin(float(full.card(tag))*3.1415/full.package_count())/4 + 0.75)

relevanceIndexFunction = function_deprecated_by(relevance_index_function)

class DB:
    """
    In-memory database mapping packages to tags and tags to packages.
    """

def __init__(self):
        # type: () -> None
        self.db = {}     # type: PkgTagDbType
        self.rdb = {}    # type: TagPkgDbType

def read(self,
             input_data,       # type: Iterator[Text]
             tag_filter=None,  # type: Optional[TagFilterType]
            ):
        # type: (...) -> None
        """
        Read the database from a file.

Example::
            # Read the system Debtags database
            db.read(open("/var/lib/debtags/package-tags", "r"))
        """
        self.db, self.rdb = read_tag_database_both_ways(input_data, tag_filter)

def qwrite(self, file):
        # type: (IO[bytes]) -> None
        """Quickly write the data to a pickled file"""
        pickle.dump(self.db, file)
        pickle.dump(self.rdb, file)

def qread(self, file):
        # type: (IO[bytes]) -> None
        """Quickly read the data from a pickled file"""
        self.db = pickle.load(file)
        self.rdb = pickle.load(file)

def insert(self, pkg, tags):
        # type: (str, Set[str]) -> None
        self.db[pkg] = tags.copy()
        for tag in tags:
            if tag in self.rdb:
                self.rdb[tag].add(pkg)
            else:
                self.rdb[tag] = set((pkg))

def dump(self):
        # type: () -> None
        output(self.db)

def dump_reverse(self):
        # type: () -> None
        output(self.rdb)

dumpReverse = function_deprecated_by(dump_reverse)

def reverse(self):
        # type: () -> DB
        "Return the reverse collection, sharing tagsets with this one"
        res = DB()
        res.db = self.rdb
        res.rdb = self.db
        return res

def facet_collection(self):
        # type: () -> DB
        """
        Return a copy of this collection, but replaces the tag names
        with only their facets.
        """
        fcoll = DB()
        tofacet = re.compile(r"^([^:]+).+")
        for pkg, tags in self.iter_packages_tags():
            ftags = {tofacet.sub(r"\1", t) for t in tags}
            fcoll.insert(pkg, ftags)
        return fcoll

facetCollection = function_deprecated_by(facet_collection)

def copy(self):
        # type: () -> DB
        """
        Return a copy of this collection, with the tagsets copied as
        well.
        """
        res = DB()
        res.db = self.db.copy()
        res.rdb = self.rdb.copy()
        return res

def reverse_copy(self):
        # type: () -> DB
        """
        Return the reverse collection, with a copy of the tagsets of
        this one.
        """
        res = DB()
        res.db = self.rdb.copy()
        res.rdb = self.db.copy()
        return res

reverseCopy = function_deprecated_by(reverse_copy)

def choose_packages(self, package_iter):
        # type: (Iterable[str]) -> DB
        """
        Return a collection with only the packages in package_iter,
        sharing tagsets with this one
        """
        res = DB()
        db = {}
        for pkg in package_iter:
            if pkg in self.db:
                db[pkg] = self.db[pkg]
        res.db = db
        res.rdb = reverse(db)
        return res

choosePackages = function_deprecated_by(choose_packages)

def choose_packages_copy(self, package_iter):
        # type: (Iterable[str]) -> DB
        """
        Return a collection with only the packages in package_iter,
        with a copy of the tagsets of this one
        """
        res = DB()
        db = {}
        for pkg in package_iter:
            db[pkg] = self.db[pkg]
        res.db = db
        res.rdb = reverse(db)
        return res

choosePackagesCopy = function_deprecated_by(choose_packages_copy)

def filter_packages(self, package_filter):
        # type: (PkgFilterType) -> DB
        """
        Return a collection with only those packages that match a
        filter, sharing tagsets with this one.  The filter will match
        on the package.
        """
        res = DB()
        db = {}
        for pkg in filter(package_filter, six.iterkeys(self.db)):
            db[pkg] = self.db[pkg]
        res.db = db
        res.rdb = reverse(db)
        return res

filterPackages = function_deprecated_by(filter_packages)

def filter_packages_copy(self, filter_data):
        # type: (PkgFilterType) -> DB
        """
        Return a collection with only those packages that match a
        filter, with a copy of the tagsets of this one.  The filter
        will match on the package.
        """
        res = DB()
        db = {}
        for pkg in filter(filter_data, six.iterkeys(self.db)):
            db[pkg] = self.db[pkg].copy()
        res.db = db
        res.rdb = reverse(db)
        return res

filterPackagesCopy = function_deprecated_by(filter_packages_copy)

def filter_packages_tags(self, package_tag_filter):
        # type: (PkgTagFilterType) -> DB
        """
        Return a collection with only those packages that match a
        filter, sharing tagsets with this one.  The filter will match
        on (package, tags).
        """
        res = DB()
        db = {}
        for pkg, _ in filter(package_tag_filter, six.iteritems(self.db)):
            db[pkg] = self.db[pkg]
        res.db = db
        res.rdb = reverse(db)
        return res

filterPackagesTags = function_deprecated_by(filter_packages_tags)

def filter_packages_tags_copy(self, package_tag_filter):
        # type: (PkgTagFilterType) -> DB
        """
        Return a collection with only those packages that match a
        filter, with a copy of the tagsets of this one.  The filter
        will match on (package, tags).
        """
        res = DB()
        db = {}
        for pkg, _ in filter(package_tag_filter, six.iteritems(self.db)):
            db[pkg] = self.db[pkg].copy()
        res.db = db
        res.rdb = reverse(db)
        return res

filterPackagesTagsCopy = function_deprecated_by(filter_packages_tags_copy)

def filter_tags(self, tag_filter):
        # type: (TagFilterType) -> DB
        """
        Return a collection with only those tags that match a
        filter, sharing package sets with this one.  The filter will match
        on the tag.
        """
        res = DB()
        rdb = {}
        for tag in filter(tag_filter, six.iterkeys(self.rdb)):
            rdb[tag] = self.rdb[tag]
        res.rdb = rdb
        res.db = reverse(rdb)
        return res

filterTags = function_deprecated_by(filter_tags)

def filter_tags_copy(self, tag_filter):
        # type: (TagFilterType) -> DB
        """
        Return a collection with only those tags that match a
        filter, with a copy of the package sets of this one.  The
        filter will match on the tag.
        """
        res = DB()
        rdb = {}
        for tag in filter(tag_filter, six.iterkeys(self.rdb)):
            rdb[tag] = self.rdb[tag].copy()
        res.rdb = rdb
        res.db = reverse(rdb)
        return res

filterTagsCopy = function_deprecated_by(filter_tags_copy)

def has_package(self, pkg):
        # type: (str) -> bool
        """Check if the collection contains the given package"""
        return pkg in self.db

hasPackage = function_deprecated_by(has_package)

def has_tag(self, tag):
        # type: (str) -> bool
        """Check if the collection contains packages tagged with tag"""
        return tag in self.rdb

hasTag = function_deprecated_by(has_tag)

def tags_of_package(self, pkg):
        # type: (str) -> Set[str]
        """Return the tag set of a package"""
        return self.db[pkg] if pkg in self.db else set()

tagsOfPackage = function_deprecated_by(tags_of_package)

def packages_of_tag(self, tag):
        # type: (str) -> Set[str]
        """Return the package set of a tag"""
        return self.rdb[tag] if tag in self.rdb else set()

packagesOfTag = function_deprecated_by(packages_of_tag)

def tags_of_packages(self, pkgs):
        # type: (Iterable[str]) -> Set[str]
        """Return the set of tags that have all the packages in ``pkgs``"""
        return set.union(*(self.tags_of_package(p) for p in pkgs))

tagsOfPackages = function_deprecated_by(tags_of_packages)

def packages_of_tags(self, tags):
        # type: (Iterable[str]) -> Set[str]
        """Return the set of packages that have all the tags in ``tags``"""
        return set.union(*(self.packages_of_tag(t) for t in tags))

packagesOfTags = function_deprecated_by(packages_of_tags)

def card(self, tag):
        # type: (str) -> int
        """
        Return the cardinality of a tag
        """
        return len(self.rdb[tag]) if tag in self.rdb else 0

def discriminance(self, tag):
        # type: (str) -> int
        """
        Return the discriminance index if the tag.

Th discriminance index of the tag is defined as the minimum
        number of packages that would be eliminated by selecting only
        those tagged with this tag or only those not tagged with this
        tag.
        """
        n = self.card(tag)
        tot = self.package_count()
        return min(n, tot - n)

def iter_packages(self):
        # type: () -> Iterator[str]
        """Iterate over the packages"""
        return six.iterkeys(self.db)

iterPackages = function_deprecated_by(iter_packages)

def iter_tags(self):
        # type: () -> Iterator[str]
        """Iterate over the tags"""
        return six.iterkeys(self.rdb)

iterTags = function_deprecated_by(iter_tags)

def iter_packages_tags(self):
        # type: () -> Iterator[Tuple[str, Set[str]]]
        """Iterate over 2-tuples of (pkg, tags)"""
        return six.iteritems(self.db)

iterPackagesTags = function_deprecated_by(iter_packages_tags)

def iter_tags_packages(self):
        # type: () -> Iterator[Tuple[str, Set[str]]]
        """Iterate over 2-tuples of (tag, pkgs)"""
        return six.iteritems(self.rdb)

iterTagsPackages = function_deprecated_by(iter_tags_packages)

def package_count(self):
        # type: () -> int
        """Return the number of packages"""
        return len(self.db)

packageCount = function_deprecated_by(package_count)

def tag_count(self):
        # type: () -> int
        """Return the number of tags"""
        return len(self.rdb)

tagCount = function_deprecated_by(tag_count)

def ideal_tagset(self, tags):
        # type: (List[str]) -> Set[str]
        """
        Return an ideal selection of the top tags in a list of tags.

Return the tagset made of the highest number of tags taken in
        consecutive sequence from the beginning of the given vector,
        that would intersect with the tagset of a comfortable amount
        of packages.

Comfortable is defined in terms of how far it is from 7.
        """

# TODO: the scoring function is quite ok, but may need more
        # tuning.  I also center it on 15 instead of 7 since we're
        # setting a starting point for the search, not a target point
        def score_fun(x): # type: (float) -> float
            return float((x-15)*(x-15))/x

tagset = set()   # type: Set[str]
        min_score = 3.
        for i in range(len(tags)):
            pkgs = self.packages_of_tags(tags[:i+1])
            card = len(pkgs)
            if card == 0:
                break
            score = score_fun(card)
            if score < min_score:
                min_score = score
                tagset = set(tags[:i+1])

# Return always at least the first tag
        if not tagset:
            return set(tags[:1])
        return tagset

idealTagset = function_deprecated_by(ideal_tagset)

def correlations(self):
        # type: () -> Iterator[Tuple[str, str, float]]
        """
        Generate the list of correlation as a tuple (hastag, hasalsotag, score).

Every tuple will indicate that the tag 'hastag' tends to also
        have 'hasalsotag' with a score of 'score'.
        """
        for pivot in self.iter_tags():
            # pylint: disable=cell-var-from-loop
            with_ = self.filter_packages_tags(lambda pt: pivot in pt[1])
            without = self.filter_packages_tags(lambda pt: pivot not in pt[1])
            for tag in with_.iter_tags():
                if tag == pivot:
                    continue
                has = float(with_.card(tag)) / float(with_.package_count())
                hasnt = float(without.card(tag)) / float(without.package_count())
                yield pivot, tag, has - hasnt

./Ninja\.