346 lines
17 KiB
Python
346 lines
17 KiB
Python
import pkg_resources
|
|
from packaging import version
|
|
from collections import defaultdict
|
|
import subprocess
|
|
|
|
class PackageDependencyChecker:
|
|
"""Utility functions for unspaghettifying version hell."""
|
|
def __init__(self):
|
|
self.installed_packages = {dist.project_name: dist for dist in pkg_resources.working_set}
|
|
|
|
def find_dependents(self, package_name):
|
|
dependents = []
|
|
for dist in self.installed_packages.values():
|
|
requires = [str(req) for req in dist.requires()]
|
|
if any(package_name in req for req in requires):
|
|
dependents.append((dist.project_name, dist.version, requires))
|
|
return dependents
|
|
|
|
def check_version_discrepancy_for_line(self, line: str) -> dict:
|
|
line = line.strip()
|
|
if not line or line.startswith('#'):
|
|
return None
|
|
|
|
try:
|
|
requirement = pkg_resources.Requirement.parse(line)
|
|
package_name = requirement.name
|
|
version_spec = str(requirement.specifier)
|
|
return self._check_version(package_name, version_spec)
|
|
except pkg_resources.extern.packaging.requirements.InvalidRequirement:
|
|
return None
|
|
|
|
def _check_version(self, package_name, version_spec):
|
|
if package_name in self.installed_packages:
|
|
installed_version = self.installed_packages[package_name].version
|
|
#print(f"Checking version: {package_name} {version_spec}")
|
|
requirement = pkg_resources.Requirement.parse(f"{package_name} {version_spec}")
|
|
if not requirement.specifier.contains(installed_version):
|
|
dependents = self.find_dependents(package_name)
|
|
return {
|
|
'package_name': package_name,
|
|
'installed_version': installed_version,
|
|
'required_version': version_spec,
|
|
'dependents': dependents
|
|
}
|
|
else:
|
|
return {
|
|
'package_name': package_name,
|
|
'installed_version': None,
|
|
'required_version': version_spec,
|
|
'dependents': []
|
|
}
|
|
return None
|
|
|
|
def check_version_discrepancies(self, requirements_file):
|
|
discrepancies = []
|
|
with open(requirements_file, 'r') as file:
|
|
for line in file:
|
|
discrepancy = self.check_version_discrepancy_for_line(line)
|
|
if discrepancy:
|
|
discrepancies.append(discrepancy)
|
|
return discrepancies
|
|
|
|
def generate_user_instructions(self, discrepancies):
|
|
instructions = []
|
|
for discrepancy in discrepancies:
|
|
if discrepancy['installed_version'] is None:
|
|
instructions.append(f"Package '{discrepancy['package_name']}' is missing. Please install version {discrepancy['required_version']}.")
|
|
else:
|
|
instructions.append(f"Package '{discrepancy['package_name']}' has version {discrepancy['installed_version']} installed, but version {discrepancy['required_version']} is required.")
|
|
if discrepancy['dependents']:
|
|
instructions.append("Dependent packages:")
|
|
for dependent in discrepancy['dependents']:
|
|
instructions.append(f" - {dependent[0]} (version {dependent[1]}) requires {discrepancy['package_name']}.")
|
|
return "\n".join(instructions)
|
|
|
|
def full_check(self, requirements_file):
|
|
discrepancies = self.check_version_discrepancies(requirements_file)
|
|
instructions = self.generate_user_instructions(discrepancies)
|
|
return instructions
|
|
|
|
def check_dependents_discrepancies(self, package_name):
|
|
discrepancies = []
|
|
dependents = self.find_dependents(package_name)
|
|
for dependent in dependents:
|
|
for requirement in dependent[2]:
|
|
if package_name in requirement:
|
|
try:
|
|
req = pkg_resources.Requirement.parse(requirement)
|
|
if req.name == package_name:
|
|
requirement_version = str(req.specifier)
|
|
discrepancy = self._check_version(package_name, requirement_version)
|
|
if discrepancy:
|
|
discrepancies.append(discrepancy)
|
|
except pkg_resources.extern.packaging.requirements.InvalidRequirement:
|
|
continue
|
|
return discrepancies
|
|
|
|
def analyze_discrepancies(self, discrepancies):
|
|
"""Analyzes discrepancies and finds possible version ranges."""
|
|
print("Analyzing package discrepancies...")
|
|
requirements = defaultdict(list)
|
|
for discrepancy in discrepancies:
|
|
requirements[discrepancy['package_name']].append(discrepancy['required_version'])
|
|
|
|
solutions = {}
|
|
for package, version_specs in requirements.items():
|
|
solutions[package] = self.find_version_solutions(version_specs)
|
|
return solutions
|
|
|
|
def find_version_solutions(self, version_specs):
|
|
"""Finds version ranges that satisfy the given version specifications."""
|
|
ranges = []
|
|
for spec in version_specs:
|
|
requirement = pkg_resources.Requirement.parse(f"dummy {spec}")
|
|
ranges.append(requirement.specifier)
|
|
|
|
return self.find_common_version_range(ranges)
|
|
|
|
def find_common_version_range(self, ranges):
|
|
"""Finds common version range for the given specifiers."""
|
|
min_version = version.parse("0")
|
|
max_version = version.parse("9999")
|
|
excluded_versions = set()
|
|
|
|
for r in ranges:
|
|
for spec in r:
|
|
if spec.operator == '>=':
|
|
min_version = max(min_version, version.parse(spec.version))
|
|
elif spec.operator == '>':
|
|
new_min_version = version.parse(spec.version)
|
|
min_version = max(min_version, version.parse(f"{new_min_version.major}.{new_min_version.minor}.{new_min_version.micro}") + version.parse("0.0.1"))
|
|
elif spec.operator == '<=':
|
|
max_version = min(max_version, version.parse(spec.version))
|
|
elif spec.operator == '<':
|
|
max_version = min(max_version, version.parse(spec.version))
|
|
elif spec.operator == '!=':
|
|
excluded_versions.add(version.parse(spec.version))
|
|
|
|
if min_version <= max_version:
|
|
return (min_version, max_version, excluded_versions)
|
|
else:
|
|
return None
|
|
|
|
def fetch_available_versions(self, package_name):
|
|
"""Fetches all available versions of a package from pip."""
|
|
result = subprocess.run(['pip', 'index', 'versions', package_name], capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
return []
|
|
|
|
lines = result.stdout.splitlines()
|
|
versions = []
|
|
for line in lines:
|
|
if line.startswith('Available versions:'):
|
|
versions = [ver.strip() for ver in line.split(':')[1].split(',')]
|
|
break
|
|
return versions
|
|
|
|
def analyze_possible_upgrades(self, discrepancies):
|
|
possible_upgrades = {}
|
|
for discrepancy in discrepancies:
|
|
package_name = discrepancy['package_name']
|
|
current_version = discrepancy['installed_version']
|
|
# Fetch possible upgrades from pip
|
|
available_versions = self.fetch_available_versions(package_name)
|
|
for version in available_versions:
|
|
if version > current_version:
|
|
impact = self.calculate_upgrade_impact(package_name, version, discrepancies)
|
|
possible_upgrades[(package_name, version)] = impact
|
|
return possible_upgrades
|
|
|
|
def calculate_upgrade_impact(self, package_name, version, discrepancies):
|
|
# Simulate upgrading the package and recheck discrepancies
|
|
new_discrepancies = self.simulate_upgrade(package_name, version, discrepancies)
|
|
return len(discrepancies) - len(new_discrepancies)
|
|
|
|
def simulate_upgrade(self, package_name, new_version, discrepancies):
|
|
original_version = self.installed_packages[package_name].version
|
|
self.installed_packages[package_name] = pkg_resources.Distribution(project_name=package_name, version=new_version)
|
|
new_discrepancies = []
|
|
for discrepancy in discrepancies:
|
|
if discrepancy['package_name'] != package_name:
|
|
new_discrepancy = self._check_version(discrepancy['package_name'], discrepancy['required_version'])
|
|
if new_discrepancy:
|
|
new_discrepancies.append(new_discrepancy)
|
|
self.installed_packages[package_name] = pkg_resources.Distribution(project_name=package_name, version=original_version)
|
|
return new_discrepancies
|
|
|
|
def suggest_solutions(self, solutions):
|
|
"""Generates solution suggestions based on the analyzed discrepancies."""
|
|
suggestions = []
|
|
for package, version_range in solutions.items():
|
|
dependents = self.find_dependents(package)
|
|
|
|
if version_range:
|
|
min_version, max_version, excluded_versions = version_range
|
|
excluded_versions_str = ", ".join(str(v) for v in sorted(excluded_versions))
|
|
|
|
if excluded_versions:
|
|
suggestions.append(f"Package '{package}' can be installed in the version range {min_version} - {max_version}, excluding versions: {excluded_versions_str}.")
|
|
else:
|
|
suggestions.append(f"Package '{package}' can be installed in the version range {min_version} - {max_version}.")
|
|
|
|
suggestions.append("This version range is required by the following packages:")
|
|
for dependent in dependents:
|
|
suggestions.append(f" - {dependent[0]} (version {dependent[1]})")
|
|
|
|
# Fetch available versions from pip
|
|
available_versions = self.fetch_available_versions(package)
|
|
valid_versions = [
|
|
version.parse(ver) for ver in available_versions
|
|
if min_version <= version.parse(ver) <= max_version and version.parse(ver) not in excluded_versions
|
|
]
|
|
|
|
# Filter to keep only the latest version in each subversion
|
|
latest_versions = self.filter_latest_versions(valid_versions)
|
|
|
|
if latest_versions:
|
|
suggestions.append("Possible Commands:")
|
|
for ver in sorted(latest_versions, reverse=True):
|
|
suggestions.append(f" - pip install {package}=={ver}")
|
|
else:
|
|
suggestions.append(f"No valid versions found for '{package}' within the specified range.")
|
|
else:
|
|
suggestions.append(f"No common version range found for package '{package}'. However, you can try the following versions required by other packages:")
|
|
package_commands = defaultdict(list)
|
|
for dependent in dependents:
|
|
for requirement in dependent[2]:
|
|
if package in requirement:
|
|
suggestions.append(f" - {requirement} required by {dependent[0]} (version {dependent[1]})")
|
|
available_versions = self.fetch_available_versions(package)
|
|
try:
|
|
req = pkg_resources.Requirement.parse(requirement)
|
|
for ver in available_versions:
|
|
ver_parsed = version.parse(ver)
|
|
if str(ver_parsed) in str(req.specifier):
|
|
package_commands[(dependent[0], dependent[1])].append(f" - pip install {package}=={ver}")
|
|
except (ValueError, IndexError):
|
|
continue
|
|
|
|
if package_commands:
|
|
for packages, commands in package_commands.items():
|
|
package_list = f"{packages[0]} (version {packages[1]})"
|
|
suggestions.append(f"To get the package {package_list} working, you can use the following commands:")
|
|
suggestions.extend(commands)
|
|
else:
|
|
suggestions.append(f"No valid versions found for '{package}' based on individual requirements.")
|
|
return suggestions
|
|
|
|
def filter_latest_versions(self, versions):
|
|
"""Filters the versions to keep only the latest version in each subversion."""
|
|
latest_versions = {}
|
|
for ver in versions:
|
|
major_minor = (ver.major, ver.minor)
|
|
if major_minor not in latest_versions or ver > latest_versions[major_minor]:
|
|
latest_versions[major_minor] = ver
|
|
return latest_versions.values()
|
|
|
|
def rank_solutions(self, solutions):
|
|
"""Ranks version ranges based on the number of discrepancies they resolve."""
|
|
version_range_counts = {}
|
|
for package, version_range in solutions.items():
|
|
if version_range:
|
|
min_version, max_version, _ = version_range
|
|
range_key = (min_version, max_version)
|
|
if range_key not in version_range_counts:
|
|
version_range_counts[range_key] = 0
|
|
version_range_counts[range_key] += 1
|
|
|
|
ranked_solutions = sorted(version_range_counts.items(), key=lambda x: -x[1])
|
|
return ranked_solutions
|
|
|
|
def rank_upgrades(self, possible_upgrades):
|
|
"""Ranks upgrades based on the number of discrepancies they resolve."""
|
|
ranked_upgrades = sorted(possible_upgrades.items(), key=lambda x: -x[1])
|
|
return ranked_upgrades
|
|
|
|
def find_best_upgrade_path(self, discrepancies):
|
|
"""Finds the best upgrade path to resolve the most discrepancies."""
|
|
solutions = self.analyze_discrepancies(discrepancies)
|
|
possible_upgrades = self.analyze_possible_upgrades(discrepancies)
|
|
ranked_solutions = self.rank_solutions(solutions)
|
|
ranked_upgrades = self.rank_upgrades(possible_upgrades)
|
|
|
|
best_path = []
|
|
|
|
# Combine ranked solutions and upgrades to find the optimal path
|
|
for upgrade, impact in ranked_upgrades:
|
|
package, version = upgrade
|
|
best_path.append(f"Upgrading {package} to version {version} resolves {impact} discrepancies.")
|
|
|
|
for version_range, count in ranked_solutions:
|
|
best_path.append(f"Installing packages in the version range {version_range} resolves {count} discrepancies.")
|
|
|
|
return best_path
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
# Usage example:
|
|
checker = PackageDependencyChecker()
|
|
|
|
# Find dependents of a specific package
|
|
package_name = sys.argv[1]
|
|
dependents = checker.find_dependents(package_name)
|
|
print(f"Dependents of '{package_name}':\n")
|
|
for dependent in dependents:
|
|
print(f"Package: {dependent[0]}, Version: {dependent[1]}")
|
|
print(f" Requires: {dependent[2]}")
|
|
|
|
# Check version discrepancies based on requirements.txt
|
|
# discrepancies = checker.check_version_discrepancies('requirements.txt')
|
|
# print("\nVersion discrepancies:")
|
|
# for discrepancy in discrepancies:
|
|
# print(f"Package: {discrepancy['package_name']}, Installed: {discrepancy['installed_version']}, Required: {discrepancy['required_version']}")
|
|
# print("Dependents:")
|
|
# for dependent in discrepancy['dependents']:
|
|
# print(f" - Dependent Package: {dependent[0]}, Version: {dependent[1]}")
|
|
# print(f" Requires: {dependent[2]}")
|
|
|
|
# instructions = checker.generate_user_instructions(discrepancies)
|
|
# print("\nVersion discrepancies and instructions:")
|
|
# print(instructions)
|
|
|
|
# Check version discrepancies among dependents of the specific package
|
|
dependent_discrepancies = checker.check_dependents_discrepancies(package_name)
|
|
# if dependent_discrepancies:
|
|
# print(f"\nDiscrepancies in dependents of '{package_name}':")
|
|
# for discrepancy in dependent_discrepancies:
|
|
# print(f"Package: {discrepancy['package_name']}, Installed: {discrepancy['installed_version']}, Required: {discrepancy['required_version']}")
|
|
# print("Dependents:")
|
|
# for dependent in discrepancy['dependents']:
|
|
# print(f" - Dependent Package: {dependent[0]}, Version: {dependent[1]}")
|
|
# print(f" Requires: {dependent[2]}")
|
|
# else:
|
|
# print(f"No discrepancies found among the dependents of '{package_name}'.")
|
|
|
|
# Analyze discrepancies and suggest solutions
|
|
solutions = checker.analyze_discrepancies(dependent_discrepancies)
|
|
solution_suggestions = checker.suggest_solutions(solutions)
|
|
print("\nSuggested solutions:")
|
|
for suggestion in solution_suggestions:
|
|
print(suggestion)
|
|
|
|
best_upgrade_path = checker.find_best_upgrade_path(dependent_discrepancies)
|
|
for up in best_upgrade_path:
|
|
print(up) |