diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py new file mode 100644 index 00000000..afcbefb4 --- /dev/null +++ b/tests/test_pixi_to_conda_lock.py @@ -0,0 +1,396 @@ +"""Tests for the pixi_to_conda_lock.py script.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, mock_open, patch + +import pytest + +import unidep.pixi_to_conda_lock as ptcl + + +@pytest.fixture +def sample_pixi_lock() -> dict[str, Any]: + """Sample pixi.lock data for testing.""" + return { + "version": 6, + "environments": { + "default": { + "channels": [{"url": "https://conda.anaconda.org/conda-forge/"}], + "indexes": ["https://pypi.org/simple"], + "packages": { + "osx-arm64": [ + { + "conda": "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda", + }, + { + "pypi": "https://files.pythonhosted.org/packages/04/27/8739697a1d77f972feee90b844786b893217a133941477570d161de2750f/numthreads-0.5.0-py3-none-any.whl", + }, + ], + }, + }, + }, + "packages": [ + { + "conda": "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "depends": {"bzip2": ">=1.0.8,<2.0a0", "libexpat": ">=2.6.4,<3.0a0"}, + }, + { + "pypi": "https://files.pythonhosted.org/packages/04/27/8739697a1d77f972feee90b844786b893217a133941477570d161de2750f/numthreads-0.5.0-py3-none-any.whl", + "name": "numthreads", + "version": "0.5.0", + "sha256": "e56e83cbccef103901e678aa014d64b203cdb1b3a3be7cdedb2516ef62ec8fa1", + }, + ], + } + + +@pytest.fixture +def sample_repodata() -> dict[str, Any]: + """Sample repodata for testing.""" + return { + "repo1": { + "info": {"subdir": "osx-arm64"}, + "packages.conda": { + "python-3.13.2-hfd29fff_1_cp313t.conda": { + "name": "python", + "version": "3.13.2", + "build": "hfd29fff_1_cp313t", + "build_number": 1, + "depends": ["bzip2 >=1.0.8,<2.0a0", "libexpat >=2.6.4,<3.0a0"], + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + }, + }, + }, + } + + +def test_read_yaml_file() -> None: + """Test reading a YAML file.""" + mock_yaml_content = """ + key1: value1 + key2: value2 + """ + with patch("builtins.open", mock_open(read_data=mock_yaml_content)): + result = ptcl.read_yaml_file(Path("test.yaml")) + assert result == {"key1": "value1", "key2": "value2"} + + +def test_write_yaml_file() -> None: + """Test writing a YAML file.""" + data = {"key1": "value1", "key2": "value2"} + mock_file = mock_open() + with patch("builtins.open", mock_file): + ptcl.write_yaml_file(Path("test.yaml"), data) + mock_file.assert_called_once_with(Path("test.yaml"), "w") + mock_file().write.assert_called() + + +def test_find_repodata_cache_dir() -> None: + """Test finding the repodata cache directory.""" + # Simulate a valid repodata directory exists + with ( + patch("pathlib.Path.exists", return_value=True), + patch("pathlib.Path.is_dir", return_value=True), + patch("subprocess.check_output", return_value='{"cache_dir": "/dummy/path"}'), + patch("json.loads", return_value={"cache_dir": "/dummy/path"}), + # Create a dummy Path object that behaves like it exists + patch("unidep.pixi_to_conda_lock.Path", wraps=Path), + ): + result = ptcl.find_repodata_cache_dir() + assert result is not None + + # Simulate the repodata directory does not exist, and expect a ValueError + with ( + patch("pathlib.Path.exists", return_value=False), + patch("subprocess.check_output", return_value='{"cache_dir": "/dummy/path"}'), + patch("json.loads", return_value={"cache_dir": "/dummy/path"}), + pytest.raises(ValueError, match="Could not find repodata cache directory"), + ): + ptcl.find_repodata_cache_dir() + + +def test_load_json_file() -> None: + """Test loading a JSON file.""" + mock_json_content = '{"key1": "value1", "key2": "value2"}' + with patch("builtins.open", mock_open(read_data=mock_json_content)): + result = ptcl.load_json_file(Path("test.json")) + assert result == {"key1": "value1", "key2": "value2"} + + +def test_load_repodata_files() -> None: + """Test loading repodata files.""" + mock_dir = MagicMock() + mock_file1 = MagicMock() + mock_file1.name = "file1.json" + mock_file1.stem = "file1" + mock_file2 = MagicMock() + mock_file2.name = "file2.info.json" + + mock_dir.glob.return_value = [mock_file1, mock_file2] + + with patch("unidep.pixi_to_conda_lock.load_json_file") as mock_load: + mock_load.return_value = {"key": "value"} + result = ptcl.load_repodata_files(mock_dir) + + assert "file1" in result + assert "file2" not in result + assert result["file1"] == {"key": "value"} + + +def test_extract_filename_from_url() -> None: + """Test extracting filename from URL.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + result = ptcl.extract_filename_from_url(url) + assert result == "python-3.13.2-hfd29fff_1_cp313t.conda" + + +def test_find_package_in_repodata(sample_repodata: dict[str, Any]) -> None: + """Test finding a package in repodata.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + result = ptcl.find_package_in_repodata(sample_repodata, url) + assert result is not None + assert result["name"] == "python" + assert result["version"] == "3.13.2" + + # Test package not found + url_not_found = "https://conda.anaconda.org/conda-forge/osx-arm64/nonexistent-1.0.0-abc123.conda" + result_not_found = ptcl.find_package_in_repodata(sample_repodata, url_not_found) + assert result_not_found is None + + +def test_extract_platform_from_url() -> None: + """Test extracting platform from URL.""" + # Test different platforms + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/noarch/pkg-1.0.0.conda", + ) + == "noarch" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/osx-arm64/pkg-1.0.0.conda", + ) + == "osx-arm64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/osx-64/pkg-1.0.0.conda", + ) + == "osx-64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/linux-64/pkg-1.0.0.conda", + ) + == "linux-64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/linux-aarch64/pkg-1.0.0.conda", + ) + == "linux-aarch64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/win-64/pkg-1.0.0.conda", + ) + == "win-64" + ) + with pytest.raises(ValueError, match="Unknown platform"): + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/unknown/pkg-1.0.0.conda", + ) + + +def test_extract_name_version_from_url() -> None: + """Test extracting name and version from URL.""" + # Test standard package + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + name, version = ptcl.extract_name_version_from_url(url) + assert name == "python" + assert version == "3.13.2" + + # Test package with tar.bz2 extension + url_tar = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.tar.bz2" + name_tar, version_tar = ptcl.extract_name_version_from_url(url_tar) + assert name_tar == "python" + assert version_tar == "3.13.2" + + # Test with dash + url_with_dash = "https://conda.anaconda.org/conda-forge/osx-arm64/ca-certificates-2025.1.31-hf0a4a13_0.conda" + name_with_dash, version_with_dash = ptcl.extract_name_version_from_url( + url_with_dash, + ) + assert name_with_dash == "ca-certificates" + assert version_with_dash == "2025.1.31" + + +def test_parse_dependencies_from_repodata() -> None: + """Test parsing dependencies from repodata.""" + depends_list = ["python >=3.8", "numpy", "pandas >=1.0.0,<2.0.0"] + result = ptcl.parse_dependencies_from_repodata(depends_list) + assert result == {"python": ">=3.8", "numpy": "", "pandas": ">=1.0.0,<2.0.0"} + + +def test_create_conda_package_entry() -> None: + """Test creating a conda package entry.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + repodata_info = { + "name": "python", + "version": "3.13.2", + "build": "hfd29fff_1_cp313t", + "build_number": 1, + "depends": ["bzip2 >=1.0.8,<2.0a0", "libexpat >=2.6.4,<3.0a0"], + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + } + + result = ptcl.create_conda_package_entry(url, repodata_info) + + assert result["name"] == "python" + assert result["version"] == "3.13.2" + assert result["manager"] == "conda" + assert result["platform"] == "osx-arm64" + assert result["dependencies"] == { + "bzip2": ">=1.0.8,<2.0a0", + "libexpat": ">=2.6.4,<3.0a0", + } + assert result["url"] == url + assert result["hash"]["md5"] == "9d0ae3f3e43c192a992827c0abffe284" + assert ( + result["hash"]["sha256"] + == "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a" + ) + + +def test_create_conda_package_entry_fallback() -> None: + """Test creating a conda package entry using fallback.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + package_info = { + "depends": ["bzip2 >=1.0.8,<2.0a0", "libexpat >=2.6.4,<3.0a0"], + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + } + + result = ptcl.create_conda_package_entry_fallback(url, package_info) + + assert result["name"] == "python" + assert result["version"] == "3.13.2" + assert result["manager"] == "conda" + assert result["platform"] == "osx-arm64" + assert result["dependencies"] == { + "bzip2": ">=1.0.8,<2.0a0", + "libexpat": ">=2.6.4,<3.0a0", + } + assert result["url"] == url + assert result["hash"]["md5"] == "9d0ae3f3e43c192a992827c0abffe284" + assert ( + result["hash"]["sha256"] + == "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a" + ) + + +def test_noarch_package_expansion(sample_pixi_lock: dict[str, Any]) -> None: + """Test that a noarch package is expanded into entries for each platform.""" + # Modify sample_pixi_lock to include a noarch package and specific platforms + sample_pixi_lock["packages"] = [ + { + "conda": "https://conda.anaconda.org/conda-forge/noarch/cached-property-1.5.2-hd8ed1ab_1.tar.bz2", + }, + ] + sample_pixi_lock["environments"] = { + "default": { + "channels": [{"url": "https://conda.anaconda.org/conda-forge/"}], + "indexes": ["https://pypi.org/simple"], + # Define two platforms for testing + "packages": {"linux-64": [], "osx-arm64": []}, + }, + } + + # Create a sample repodata that contains info for the noarch package. + repodata = { + "repo1": { + "info": {"subdir": "noarch"}, + "packages": { + "cached-property-1.5.2-hd8ed1ab_1.tar.bz2": { + "name": "cached-property", + "version": "1.5.2", + "build": "hd8ed1ab_1", + "build_number": 1, + "depends": ["cached_property >=1.5.2,<1.5.3.0a0"], + "md5": "9b347a7ec10940d3f7941ff6c460b551", + "sha256": "561e6660f26c35d137ee150187d89767c988413c978e1b712d53f27ddf70ea17", + }, + }, + }, + } + + # Import the module under test. + from unidep import pixi_to_conda_lock as ptcl + + # Process the conda packages. + result = ptcl.process_conda_packages(sample_pixi_lock, repodata) + + # Expect an entry per platform (linux-64 and osx-arm64) + assert len(result) == 2, "Expected two package entries, one per platform" + + # Check that each entry has the expected properties. + for entry in result: + assert entry["name"] == "cached-property" + assert entry["version"] == "1.5.2" + assert entry["manager"] == "conda" + # Even though the URL is noarch, the entry should have the platform set to the target environment. + assert ( + entry["url"] + == "https://conda.anaconda.org/conda-forge/noarch/cached-property-1.5.2-hd8ed1ab_1.tar.bz2" + ) + assert entry["hash"]["md5"] == "9b347a7ec10940d3f7941ff6c460b551" + assert ( + entry["hash"]["sha256"] + == "561e6660f26c35d137ee150187d89767c988413c978e1b712d53f27ddf70ea17" + ) + + # Verify that the packages were duplicated for each of the two platforms. + platforms = {entry["platform"] for entry in result} + assert platforms == {"linux-64", "osx-arm64"}, ( + "Expected platforms to be linux-64 and osx-arm64" + ) + + +def test_missing_pip_exception() -> None: + """Test that convert_pixi_to_conda_lock raises a ValueError + when there are PyPI packages but no pip package in conda packages. + """ # noqa: D205 + # Create a pixi_data sample with a PyPI package and no pip package. + pixi_data = { + "environments": { + "default": { + "channels": [{"url": "https://conda.anaconda.org/conda-forge/"}], + # Define two target platforms. + "packages": {"linux-64": [], "osx-arm64": []}, + }, + }, + "packages": [ + { + # Only a PyPI package entry, no conda package for pip. + "pypi": "https://files.pythonhosted.org/packages/example/somepypi-1.0.0-py3-none-any.whl", + "name": "somepypi", + "version": "1.0.0", + }, + ], + } + # For this test, repodata can be empty since it's only used for conda packages. + repodata: dict[str, dict[str, Any]] = {} + + with pytest.raises( + ValueError, + match="PyPI packages are present but no pip package found in conda packages", + ): + ptcl.convert_pixi_to_conda_lock(pixi_data, repodata) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py new file mode 100755 index 00000000..651c7d09 --- /dev/null +++ b/unidep/pixi_to_conda_lock.py @@ -0,0 +1,550 @@ +#!/usr/bin/env python3 +"""Convert a pixi.lock file to a conda-lock.yml file using repodata. + +This script reads a pixi.lock file and generates a conda-lock.yml file with the same +package information, using repodata to extract accurate package metadata. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import subprocess +import sys +from pathlib import Path +from typing import Any + +import yaml + + +def setup_logging(verbose: bool = False) -> None: # noqa: FBT001, FBT002 + """Set up logging configuration. + + Args: + verbose: Whether to enable debug logging + + """ + try: + from rich.logging import RichHandler + + handlers = [RichHandler(rich_tracebacks=True)] + except ImportError: + handlers = [logging.StreamHandler()] + + log_level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=log_level, + format="%(asctime)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + handlers=handlers, + ) + + +def read_yaml_file(file_path: Path) -> dict[str, Any]: + """Read a YAML file and return its contents as a dictionary.""" + logging.debug("Reading YAML file: %s", file_path) + with open(file_path) as f: # noqa: PTH123 + data = yaml.safe_load(f) + logging.debug("Successfully read YAML file: %s", file_path) + return data + + +def write_yaml_file(file_path: Path, data: dict[str, Any]) -> None: + """Write data to a YAML file.""" + logging.debug("Writing YAML file: %s", file_path) + with open(file_path, "w") as f: # noqa: PTH123 + yaml.dump(data, f, sort_keys=False) + logging.debug("Successfully wrote YAML file: %s", file_path) + + +def find_repodata_cache_dir() -> Path | None: + """Find the repodata cache directory using 'pixi info --json' output. + + This function runs 'pixi info --json' and extract the 'cache_dir' + field, appending 'repodata' to it. + """ + cmd = ["pixi", "info", "--json"] + result = subprocess.check_output(cmd, text=True) + info = json.loads(result) + cache_dir = info.get("cache_dir") + if cache_dir: + repodata_path = Path(cache_dir) / "repodata" + logging.debug("Using cache_dir from pixi info: %s", repodata_path) + if repodata_path.exists() and repodata_path.is_dir(): + return repodata_path + msg = "Could not find repodata cache directory" + raise ValueError(msg) + + +def load_json_file(file_path: Path) -> dict[str, Any]: + """Load a JSON file and return its contents as a dictionary.""" + logging.debug("Loading JSON file: %s", file_path) + with open(file_path) as f: # noqa: PTH123 + data = json.load(f) + logging.debug("Successfully loaded JSON file: %s", file_path) + return data + + +def load_repodata_files(repodata_dir: Path) -> dict[str, dict[str, Any]]: + """Load all repodata files from the cache directory.""" + logging.info("Loading repodata files from: %s", repodata_dir) + repodata = {} + + # Load all .json files (not .info.json) + json_files = list(repodata_dir.glob("*.json")) + logging.debug("Found %d JSON files in repodata directory", len(json_files)) + + for file_path in json_files: + if not file_path.name.endswith(".info.json"): + try: + logging.debug("Loading repodata file: %s", file_path.name) + data = load_json_file(file_path) + repodata[file_path.stem] = data + logging.debug("Successfully loaded repodata file: %s", file_path.name) + except Exception as e: # noqa: BLE001 + logging.warning("Failed to load %s: %s", file_path, e) + + logging.info("Loaded %d repodata files", len(repodata)) + return repodata + + +def extract_filename_from_url(url: str) -> str: + """Extract the filename from a URL.""" + filename = url.split("/")[-1] + logging.debug("Extracted filename '%s' from URL: %s", filename, url) + return filename + + +def find_package_in_repodata( + repodata: dict[str, dict[str, Any]], + package_url: str, +) -> dict[str, Any] | None: + """Find a package in repodata based on its URL.""" + logging.debug("Searching for package in repodata: %s", package_url) + filename = extract_filename_from_url(package_url) + + # Check all repodata files + for repo_name, repo_data in repodata.items(): + # Check in packages (.tar.bz2) + if "packages" in repo_data and filename in repo_data["packages"]: + logging.debug( + "🔍 Found package '%s' in repository '%s'", + filename, + repo_name, + ) + return repo_data["packages"][filename] + + # Check in packages.conda (.conda) + if "packages.conda" in repo_data and filename in repo_data["packages.conda"]: + logging.debug( + "🔍 Found package '%s' in repository '%s' (packages.conda)", + filename, + repo_name, + ) + return repo_data["packages.conda"][filename] + + logging.debug("Package not found in repo") + return None + + +def extract_platform_from_url(url: str) -> str: + """Extract platform information from a conda package URL.""" + logging.debug("Extracting platform from URL: %s", url) + if "/noarch/" in url: + platform = "noarch" + elif "/osx-arm64/" in url: + platform = "osx-arm64" + elif "/osx-64/" in url: + platform = "osx-64" + elif "/linux-64/" in url: + platform = "linux-64" + elif "/linux-aarch64/" in url: + platform = "linux-aarch64" + elif "/win-64/" in url: + platform = "win-64" + else: + msg = f"Unknown platform in URL: {url}" + raise ValueError(msg) + + logging.debug("Extracted platform: %s", platform) + return platform + + +def extract_name_version_from_url(url: str) -> tuple[str, str]: + """Extract package name and version from a conda package URL as a fallback.""" + logging.debug("Extracting name and version from URL: %s", url) + filename = extract_filename_from_url(url) + + # Remove file extension (.conda or .tar.bz2) + if filename.endswith(".conda"): + filename_no_ext = filename[:-6] + elif filename.endswith(".tar.bz2"): + filename_no_ext = filename[:-8] + else: + filename_no_ext = filename + + # Split by hyphens to separate name, version, and build + name, version, _build_string = filename_no_ext.rsplit("-", 2) + + logging.debug("Extracted name: %s, version: %s", name, version) + return name, version + + +def parse_dependencies_from_repodata(depends_list: list[str]) -> dict[str, str]: + """Parse dependencies from repodata format to conda-lock format.""" + logging.debug("Parsing dependencies from repolist") + dependencies = {} + for dep in depends_list: + parts = dep.split() + if len(parts) > 1: + dependencies[parts[0]] = " ".join(parts[1:]) + else: + dependencies[dep] = "" + logging.debug("Parsed dependencies: %s", dependencies) + return dependencies + + +def create_conda_package_entry( + url: str, + repodata_info: dict[str, Any], +) -> dict[str, Any]: + """Create a conda package entry for conda-lock.yml from repodata.""" + logging.debug("Creating conda package entry from repodata for: %s", url) + platform = extract_platform_from_url(url) + + package_entry = { + "name": repodata_info["name"], + "version": repodata_info["version"], + "manager": "conda", + "platform": platform, + "dependencies": parse_dependencies_from_repodata( + repodata_info.get("depends", []), + ), + "url": url, + "hash": { + "md5": repodata_info["md5"], + "sha256": repodata_info["sha256"], + }, + "category": "main", + "optional": False, + } + + logging.debug( + "Created conda package entry: %s v%s", + package_entry["name"], + package_entry["version"], + ) + return package_entry + + +def create_conda_package_entry_fallback( + url: str, + package_info: dict[str, Any], +) -> dict[str, Any]: + """Create a conda package entry for conda-lock.yml using URL parsing as fallback.""" + logging.debug("Creating conda package entry using fallback for: %s", url) + platform = extract_platform_from_url(url) + name, version = extract_name_version_from_url(url) + print(package_info) + package_entry = { + "name": name, + "version": version, + "manager": "conda", + "platform": platform, + "dependencies": parse_dependencies_from_repodata( + package_info.get("depends", []), + ), + "url": url, + "hash": { + "md5": package_info.get("md5", ""), + "sha256": package_info.get("sha256", ""), + }, + "category": "main", + "optional": False, + } + + logging.debug("Created conda package entry (fallback): %s v%s", name, version) + return package_entry + + +def create_pypi_package_entry( + platform: str, + package_info: dict[str, Any], +) -> dict[str, Any]: + """Create a PyPI package entry for conda-lock.yml.""" + url = package_info["pypi"] + logging.debug("Creating PyPI package entry for: %s (platform: %s)", url, platform) + + package_entry = { + "name": package_info.get("name", ""), + "version": package_info.get("version", ""), + "manager": "pip", + "platform": platform, + "dependencies": {}, # PyPI dependencies are handled differently + "url": url, + "hash": { + "sha256": package_info.get("sha256", ""), + }, + "category": "main", + "optional": False, + } + + logging.debug( + "Created PyPI package entry: %s v%s", + package_entry["name"], + package_entry["version"], + ) + return package_entry + + +def extract_platforms_from_pixi(pixi_data: Any) -> list[str]: + """Extract platform information from pixi.lock data.""" + logging.debug("Extracting platforms from pixi.lock data") + platforms = [] + environments = pixi_data.get("environments", {}) + for env_name, env_data in environments.items(): + logging.debug("Processing environment: %s", env_name) + for platform in env_data.get("packages", {}): + if platform not in platforms and platform != "noarch": + platforms.append(platform) + logging.debug("Added platform: %s", platform) + + logging.info("Extracted platforms: %s", platforms) + return platforms + + +def _channel_url_to_name(url: str) -> str: + """Convert a channel URL to a channel name.""" + return url.replace("https://conda.anaconda.org/", "").rstrip("/") + + +def extract_channels_from_pixi( + pixi_data: dict[str, Any], +) -> list[dict[str, Any]]: + """Extract channel information from pixi.lock data.""" + logging.debug("Extracting channels from pixi.lock data") + channels_data = ( + pixi_data.get("environments", {}).get("default", {}).get("channels", []) + ) + channels = [ + {"url": _channel_url_to_name(channel["url"]), "used_env_vars": []} + for channel in channels_data + ] + + logging.info( + "Extracted %d channels: %s", + len(channels), + [c["url"] for c in channels], + ) + return channels + + +def create_conda_lock_metadata( + platforms: list[str], + channels: list[dict[str, str]], +) -> dict[str, Any]: + """Create metadata section for conda-lock.yml.""" + logging.debug("Creating conda-lock metadata") + metadata = { + "content_hash": { + platform: "generated-from-pixi-lock" for platform in platforms + }, + "channels": channels, + "platforms": platforms, + "sources": ["converted-from-pixi.lock"], + } + logging.debug("Created conda-lock metadata with %d platforms", len(platforms)) + return metadata + + +def process_conda_packages( + pixi_data: dict[str, Any], + repodata: dict[str, dict[str, Any]], +) -> list[dict[str, Any]]: + """Process conda packages from pixi.lock and convert to conda-lock format.""" + logging.info("Processing conda packages from pixi.lock") + package_entries = [] + conda_packages = [p for p in pixi_data.get("packages", []) if "conda" in p] + platforms = extract_platforms_from_pixi(pixi_data) + logging.debug("Found %d conda packages to process", len(conda_packages)) + + for package_info in conda_packages: + url = package_info["conda"] + logging.debug("Processing conda package: %s", url) + + # Try to find package in repodata + repodata_info = find_package_in_repodata(repodata, url) + + # Create a base package entry, either using repodata or fallback. + if repodata_info: + # Use the information from repodata + logging.debug("✅ Using repodata information for package") + base_entry = create_conda_package_entry(url, repodata_info) + else: + # Fallback to parsing the URL if repodata doesn't have the package + logging.warning("❌ Repodata not found, using fallback method") + base_entry = create_conda_package_entry_fallback(url, package_info) + + # If the package is noarch, replicate it for each platform. + if "noarch" in url: + for plat in platforms: + entry = base_entry.copy() + entry["platform"] = plat + package_entries.append(entry) + else: + package_entries.append(base_entry) + + logging.info("Processed %d conda packages", len(package_entries)) + return package_entries + + +def process_pypi_packages( + pixi_data: dict[str, Any], + platforms: list[str], +) -> list[dict[str, Any]]: + """Process PyPI packages from pixi.lock and convert to conda-lock format.""" + logging.info("Processing PyPI packages from pixi.lock") + package_entries = [] + pypi_packages = [p for p in pixi_data.get("packages", []) if "pypi" in p] + logging.debug("Found %d PyPI packages to process", len(pypi_packages)) + + for package_info in pypi_packages: + logging.debug( + "Processing PyPI package: %s v%s", + package_info.get("name", "unknown"), + package_info.get("version", "unknown"), + ) + + for platform in platforms: + logging.debug("Creating entry for platform: %s", platform) + package_entry = create_pypi_package_entry(platform, package_info) + package_entries.append(package_entry) + + logging.info( + "Processed %d PyPI package entries (across all platforms)", + len(package_entries), + ) + return package_entries + + +def convert_pixi_to_conda_lock( + pixi_data: dict[str, Any], + repodata: dict[str, dict[str, Any]], +) -> dict[str, Any]: + """Convert pixi.lock data structure to conda-lock.yml format using repodata.""" + logging.info("Converting pixi.lock to conda-lock.yml format") + + # Extract platforms and channels + platforms = extract_platforms_from_pixi(pixi_data) + channels = extract_channels_from_pixi(pixi_data) + + # Create basic conda-lock structure + conda_lock_data = { + "version": 1, + "metadata": create_conda_lock_metadata(platforms, channels), + "package": [], + } + + # Process conda packages + logging.info("Processing conda packages") + conda_packages = process_conda_packages(pixi_data, repodata) + conda_lock_data["package"].extend(conda_packages) # type: ignore[attr-defined] + logging.info("Added %d conda packages to conda-lock data", len(conda_packages)) + + # Process PyPI packages + logging.info("Processing PyPI packages") + pypi_packages = process_pypi_packages(pixi_data, platforms) + if pypi_packages: + _validate_pip_in_conda_packages(conda_packages) + + conda_lock_data["package"].extend(pypi_packages) # type: ignore[attr-defined] + logging.info("Added %d PyPI package entries to conda-lock data", len(pypi_packages)) + + logging.info( + "Conversion complete - conda-lock data contains %d package entries", + len(conda_lock_data["package"]), # type: ignore[arg-type] + ) # type: ignore[attr-defined] + return conda_lock_data + + +def _validate_pip_in_conda_packages(conda_packages: list[dict[str, Any]]) -> None: + pip_included = any( + pkg.get("name") == "pip" and pkg.get("manager") == "conda" + for pkg in conda_packages + ) + if not pip_included: + msg = ( + "PyPI packages are present but no pip package found in conda packages. " + "Please ensure that pip is included in your pixi.lock file." + ) + raise ValueError(msg) + + +def main() -> int: + """Main function to convert pixi.lock to conda-lock.yml.""" + parser = argparse.ArgumentParser(description="Convert pixi.lock to conda-lock.yml") + parser.add_argument("pixi_lock", type=Path, help="Path to pixi.lock file") + parser.add_argument( + "--output", + "-o", + type=Path, + default=Path("conda-lock.yml"), + help="Output conda-lock.yml file path", + ) + parser.add_argument( + "--repodata-dir", + type=Path, + help="Path to repodata cache directory", + ) + parser.add_argument( + "--verbose", + "-v", + action="store_true", + help="Enable verbose logging", + ) + + args = parser.parse_args() + + # Set up logging + setup_logging(args.verbose) + + logging.info("Starting pixi.lock to conda-lock.yml conversion") + logging.info("Input file: %s", args.pixi_lock) + logging.info("Output file: %s", args.output) + + if not args.pixi_lock.exists(): + logging.error("Error: %s does not exist", args.pixi_lock) + return 1 + + # Find repodata cache directory + repodata_dir = args.repodata_dir + if repodata_dir is None: + repodata_dir = find_repodata_cache_dir() + logging.info("Using repodata from: %s", repodata_dir) + repodata = load_repodata_files(repodata_dir) + else: + if not repodata_dir.exists(): + logging.error( + "Specified repodata directory %s does not exist", + repodata_dir, + ) + return 1 + logging.info("Using specified repodata directory: %s", repodata_dir) + repodata = load_repodata_files(repodata_dir) + + logging.info("Reading pixi.lock file") + pixi_data = read_yaml_file(args.pixi_lock) + + logging.info("Converting pixi.lock data to conda-lock format") + conda_lock_data = convert_pixi_to_conda_lock(pixi_data, repodata) + + logging.info("Writing conda-lock.yml file") + write_yaml_file(args.output, conda_lock_data) + + logging.info("Successfully converted %s to %s", args.pixi_lock, args.output) + return 0 + + +if __name__ == "__main__": + sys.exit(main())