e79b0022d9
Instead of finding a match by examining upper bounds and lower bounds of individual specs use a scoring matcher that scores all combinations of candidate requirement sets and then finds the best match among all those potential candidates to select the final match from. This has the advantage of finding a requirement set that will best fit the input requirement set, removing requirements with higher scores (less compatible) and leaving those with lower scores (more compatible) as the better matches. Fixes bug 1288481 Change-Id: Ic95e4d607e04c7d7d4125bc5fbb5ebf205194c0c
357 lines
12 KiB
Python
Executable File
357 lines
12 KiB
Python
Executable File
#!/usr/bin/python
|
|
|
|
from __future__ import print_function
|
|
|
|
import collections
|
|
import itertools
|
|
import logging
|
|
import re
|
|
import sys
|
|
|
|
import argparse
|
|
import six
|
|
|
|
import pip.index
|
|
import pip.req
|
|
import pkg_resources
|
|
|
|
|
|
BAD_REQUIREMENTS = 2
|
|
INCOMPATIBLE_REQUIREMENTS = 3
|
|
logger = logging.getLogger()
|
|
|
|
|
|
class RequirementException(Exception):
|
|
pass
|
|
|
|
|
|
def create_parser():
|
|
parser = argparse.ArgumentParser()
|
|
|
|
parser.add_argument(
|
|
"-r", "--requirement",
|
|
dest="requirements",
|
|
nargs="*",
|
|
default=[],
|
|
metavar="<file>",
|
|
help="Install all the packages listed in the given requirements file")
|
|
parser.add_argument(
|
|
"requirement_specs",
|
|
nargs="*",
|
|
default=[],
|
|
metavar="<requirement specifier>",
|
|
help="Install specified package")
|
|
parser.add_argument(
|
|
# A regex to be used to skip requirements
|
|
"--skip-requirements-regex",
|
|
default="",
|
|
help=argparse.SUPPRESS)
|
|
parser.add_argument(
|
|
# The default version control system for editables, e.g. 'svn'
|
|
'--default-vcs',
|
|
dest='default_vcs',
|
|
default='',
|
|
help=argparse.SUPPRESS)
|
|
parser.add_argument(
|
|
"--debug", "-d",
|
|
action="store_true",
|
|
default=False,
|
|
help="Print debug information")
|
|
parser.add_argument(
|
|
"--ignore-packages",
|
|
nargs="*",
|
|
default=[],
|
|
metavar="<requirement specifier>",
|
|
help="Ignore listed packages")
|
|
return parser
|
|
|
|
|
|
def setup_logging(options):
|
|
level = logging.DEBUG if options.debug else logging.WARNING
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
logger.addHandler(handler)
|
|
logger.setLevel(level)
|
|
|
|
|
|
def install_requirement_ensure_req_field(req):
|
|
if not hasattr(req, 'req') or not req.req:
|
|
# pip 0.8 or so
|
|
link = pip.index.Link(req.url)
|
|
name = link.egg_fragment
|
|
if not name:
|
|
raise Exception("Cannot find package name from `%s'" % req.url)
|
|
req.req = pkg_resources.Requirement.parse(name)
|
|
return req
|
|
|
|
|
|
def install_requirement_str(req):
|
|
return req.url or str(req.req)
|
|
|
|
|
|
def install_requirement_parse(line, comes_from):
|
|
line = line.strip()
|
|
if line.startswith('-e') or line.startswith('--editable'):
|
|
if line.startswith('-e'):
|
|
line = line[2:].strip()
|
|
else:
|
|
line = line[len('--editable'):].strip().lstrip('=')
|
|
req = pip.req.InstallRequirement.from_editable(
|
|
line, comes_from=comes_from)
|
|
else:
|
|
req = pip.req.InstallRequirement.from_line(line, comes_from)
|
|
return install_requirement_ensure_req_field(req)
|
|
|
|
|
|
def iter_combinations(elements, include_empty=False):
|
|
"""Iterates over all combinations of the given elements list."""
|
|
if include_empty:
|
|
start = 0
|
|
else:
|
|
start = 1
|
|
for i in range(start, len(elements) + 1):
|
|
for c in itertools.combinations(elements, i):
|
|
yield c
|
|
|
|
|
|
def conflict_scorer(versioned):
|
|
"""Scores a list of (op, version) tuples, a higher score means more
|
|
conflicts while a lower score means less conflicts.
|
|
"""
|
|
if len(versioned) == 1:
|
|
return 0
|
|
op_versions = collections.defaultdict(list)
|
|
for (op, version) in versioned:
|
|
op_versions[op].append(version)
|
|
score = 0
|
|
for version in sorted(op_versions.get("==", [])):
|
|
for (op, version2) in versioned:
|
|
if version != version2:
|
|
score += 1
|
|
for version in sorted(op_versions.get("!=", [])):
|
|
for (op, version2) in versioned:
|
|
if op in ["!=", ">", "<"]:
|
|
continue
|
|
if version2 == version:
|
|
score += 1
|
|
for version in sorted(op_versions.get(">", [])):
|
|
for (op, version2) in versioned:
|
|
if (op, version2) == (">", version):
|
|
continue
|
|
if op in ["<", "<="] and version2 <= version:
|
|
score += 1
|
|
if op == "==" and version2 == version:
|
|
score += 1
|
|
for version in sorted(op_versions.get(">=", [])):
|
|
for (op, version2) in versioned:
|
|
if (op, version2) == (">=", version):
|
|
continue
|
|
if op in ["<", "<="] and version2 < version:
|
|
score += 1
|
|
for version in sorted(op_versions.get("<", [])):
|
|
for (op, version2) in versioned:
|
|
if (op, version2) == ("<", version):
|
|
continue
|
|
if op in [">", ">="] and version2 >= version:
|
|
score += 1
|
|
if op == "==" and version2 == version:
|
|
score += 1
|
|
for version in sorted(op_versions.get("<=", [])):
|
|
for (op, version2) in versioned:
|
|
if (op, version2) == ("<=", version):
|
|
continue
|
|
if op in [">", ">="] and version2 > version:
|
|
score += 1
|
|
return score
|
|
|
|
|
|
def find_best_match(versioned, scorer_func):
|
|
"""Iterates over all combinations of the given version and comparator in
|
|
the provided lists and finds the one with the best score (closest to zero
|
|
with the maximum number of elements).
|
|
"""
|
|
scored = []
|
|
for combo in iter_combinations(versioned):
|
|
scored.append((combo, scorer_func(combo)))
|
|
|
|
# Find the lowest score with the highest number of elements.
|
|
min_score = sys.maxint
|
|
for (combo, combo_score) in scored:
|
|
if combo_score < min_score:
|
|
min_score = combo_score
|
|
max_elems = -1
|
|
best_match = []
|
|
for (combo, combo_score) in scored:
|
|
if min_score == combo_score:
|
|
if len(combo) > max_elems:
|
|
best_match = combo
|
|
max_elems = len(combo)
|
|
|
|
incompatibles = set()
|
|
for (combo, combo_score) in scored:
|
|
for spec in combo:
|
|
if spec not in best_match:
|
|
incompatibles.add(spec)
|
|
return (best_match, incompatibles)
|
|
|
|
|
|
def best_match(req_key, req_list):
|
|
"""Attempts to find the versions which will work the best for the given
|
|
requirement specification list.
|
|
"""
|
|
all_specs = []
|
|
req_specs = []
|
|
for req in req_list:
|
|
if req.req.specs:
|
|
all_specs.extend(req.req.specs)
|
|
req_specs.append((req, tuple(req.req.specs)))
|
|
if not all_specs:
|
|
return (req_list[0], [])
|
|
|
|
def spec_sort(spec1, spec2):
|
|
(op1, version1) = spec1
|
|
(op2, version2) = spec2
|
|
c = cmp(version1, version2)
|
|
if c == 0:
|
|
c = cmp(op1, op2)
|
|
return c
|
|
|
|
def reform(specs, versions, default_source='compiled'):
|
|
# Covert the parsed versions back into the string versions so that
|
|
# we can return that as matches (instead of the comparable versions).
|
|
tmp_specs = []
|
|
for (op, version) in specs:
|
|
tmp_specs.append((op, versions[version]))
|
|
# Try to see if any of the requirements that we had actually had this
|
|
# exact spec, if so then just return that as the requirement, if not
|
|
# create a requirement instead.
|
|
specs = tuple(tmp_specs)
|
|
for (req, req_spec) in req_specs:
|
|
if specs == req_spec:
|
|
return req
|
|
spec_pieces = []
|
|
for (op, version) in specs:
|
|
spec_pieces.append("%s%s" % (op, version))
|
|
spec = "%s%s" % (req_key, ",".join(spec_pieces))
|
|
return pip.req.InstallRequirement.from_line(spec, default_source)
|
|
|
|
versions = {}
|
|
versioned = set()
|
|
for (op, version) in all_specs:
|
|
parsed_version = pkg_resources.parse_version(version)
|
|
versioned.add((op, parsed_version))
|
|
versions[parsed_version] = version
|
|
versioned = list(sorted(versioned, cmp=spec_sort))
|
|
initial_score = conflict_scorer(versioned)
|
|
if initial_score == 0:
|
|
return (reform(versioned, versions), [])
|
|
else:
|
|
match, incompatibles = find_best_match(versioned, conflict_scorer)
|
|
incompatibles = [reform([s], versions,
|
|
default_source='compiled conflict')
|
|
for s in incompatibles]
|
|
return (reform(match, versions), incompatibles)
|
|
|
|
|
|
def parse_requirements(options):
|
|
"""Parse package requirements from command line and files.
|
|
|
|
:return: tuple (all, ignored) of InstallRequirement
|
|
"""
|
|
all_requirements = {}
|
|
skip_match = None
|
|
if options.skip_requirements_regex:
|
|
skip_match = re.compile(options.skip_requirements_regex)
|
|
for req_spec in options.requirement_specs:
|
|
try:
|
|
req = install_requirement_parse(req_spec, "command line")
|
|
if skip_match and skip_match.search(req.req.key):
|
|
continue
|
|
all_requirements.setdefault(req.req.key, []).append(req)
|
|
except Exception as ex:
|
|
raise RequirementException("Cannot parse `%s': %s" % (req_spec, ex))
|
|
for filename in options.requirements:
|
|
try:
|
|
for req in pip.req.parse_requirements(filename):
|
|
req = install_requirement_ensure_req_field(req)
|
|
if skip_match and skip_match.search(req.req.key):
|
|
continue
|
|
all_requirements.setdefault(req.req.key, []).append(req)
|
|
except Exception as ex:
|
|
raise RequirementException("Cannot parse `%s': %s" % (filename, ex))
|
|
ignored_requirements = []
|
|
for req_spec in options.ignore_packages:
|
|
try:
|
|
req = install_requirement_parse(req_spec, "command line")
|
|
ignored_requirements.append(req)
|
|
except Exception as ex:
|
|
raise RequirementException("Cannot parse `%s': %s" % (req_spec, ex))
|
|
return (all_requirements, ignored_requirements)
|
|
|
|
|
|
def join_requirements(requirements, ignored_requirements):
|
|
skip_keys = set(pkg.req.key for pkg in ignored_requirements)
|
|
incompatibles = {}
|
|
joined_requirements = {}
|
|
for (req_key, req_list) in six.iteritems(requirements):
|
|
if req_key in skip_keys:
|
|
continue
|
|
match, req_incompatibles = best_match(req_key, req_list)
|
|
joined_requirements[req_key] = match
|
|
if req_incompatibles:
|
|
incompatibles[req_key] = req_incompatibles
|
|
return (joined_requirements, incompatibles)
|
|
|
|
|
|
def print_requirements(joined_requirements):
|
|
formatted_requirements = []
|
|
for req_key in sorted(six.iterkeys(joined_requirements)):
|
|
req = joined_requirements[req_key]
|
|
if req.url:
|
|
req = "%s#egg=%s" % (req.url, req.req)
|
|
else:
|
|
req = str(req.req)
|
|
formatted_requirements.append(req)
|
|
for req in formatted_requirements:
|
|
print(req)
|
|
|
|
|
|
def print_incompatibles(incompatibles, joined_requirements):
|
|
for req_key in sorted(six.iterkeys(incompatibles)):
|
|
req_incompatibles = incompatibles[req_key]
|
|
if not req_incompatibles:
|
|
continue
|
|
print("%s: incompatible requirements" % (req_key),
|
|
file=sys.stderr)
|
|
chosen = joined_requirements[req_key]
|
|
print("Choosing:", file=sys.stderr)
|
|
print("\t%s: %s" % (chosen.comes_from,
|
|
install_requirement_str(chosen)),
|
|
file=sys.stderr)
|
|
print("Conflicting:", file=sys.stderr)
|
|
for conflicting in req_incompatibles:
|
|
print("\t%s: %s" % (conflicting.comes_from,
|
|
install_requirement_str(conflicting)),
|
|
file=sys.stderr)
|
|
|
|
|
|
def main():
|
|
parser = create_parser()
|
|
options = parser.parse_args()
|
|
setup_logging(options)
|
|
try:
|
|
requirements, ignored_requirements = parse_requirements(options)
|
|
except RequirementException as ex:
|
|
logger.error("Requirement failure: %s", ex)
|
|
sys.exit(BAD_REQUIREMENTS)
|
|
else:
|
|
joined_requirements, incompatibles = join_requirements(requirements,
|
|
ignored_requirements)
|
|
print_incompatibles(incompatibles, joined_requirements)
|
|
print_requirements(joined_requirements)
|
|
if incompatibles:
|
|
sys.exit(INCOMPATIBLE_REQUIREMENTS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|