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)