managers/support/package_manager.py

"""
Package Manager for CodeAct Agent.
Handles Python package management and imports.
"""

from typing import Dict
from core.constants import LIBRARY_CONTENT_DICT


class PackageManager:
    """Manages Python packages available to the agent."""

    def __init__(self, default_packages: Dict[str, str] = None):
        self.default_packages = default_packages or LIBRARY_CONTENT_DICT.copy()
        self.custom_packages = {}

    def add_packages(self, packages: Dict[str, str]) -> bool:
        """Add new packages to the available packages."""
        try:
            if not isinstance(packages, dict):
                raise ValueError("Packages must be a dictionary with package name as key and description as value")

            for package_name, description in packages.items():
                if not isinstance(package_name, str) or not isinstance(description, str):
                    print("Warning: Skipping invalid package entry - package_name and description must be strings")
                    continue

                self.custom_packages[package_name] = description
                print(f"Added package '{package_name}': {description}")

            print(f"Successfully added {len(packages)} package(s) to the library")
            return True

        except Exception as e:
            print(f"Error adding packages: {e}")
            return False

    def get_all_packages(self) -> Dict[str, str]:
        """Get all available packages (default + custom)."""
        all_packages = self.default_packages.copy()
        all_packages.update(self.custom_packages)
        return all_packages

    def import_packages(self, python_executor) -> tuple:
        """Import all packages into the executor environment."""
        all_packages = self.get_all_packages()
        imported_packages = []
        failed_packages = []

        for package_name in all_packages.keys():
            try:
                import_code = f"import {package_name}"
                python_executor(import_code)
                imported_packages.append(package_name)
            except Exception as e:
                failed_packages.append((package_name, str(e)))

        return imported_packages, failed_packages