use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
#[cfg(unix)]
use std::borrow::Cow;
use color_eyre::Result;
use color_eyre::eyre::{self, WrapErr};
use serde_json::Value;
use crate::io::read_json_file;
use memchr::memmem;
#[cfg(unix)]
use crate::shebang::{is_valid_shebang_length, convert_shebang_to_env};
use crate::plan::InstallationPlan;
use crate::models::LinkType;
use crate::link::mirror_file;
#[cfg(unix)]
use crate::utils;
use crate::lfs;
#[cfg(windows)]
fn normalize_conda_path(path: &str) -> PathBuf {
PathBuf::from(path.replace('/', "\\"))
}
#[cfg(not(windows))]
fn normalize_conda_path(path: &str) -> PathBuf {
PathBuf::from(path)
}
use log;
const DEFAULT_PYTHON_VERSION: (u64, u64) = (3, 13);
const DEFAULT_PYTHON_VERSION_STR: &str = "3.13";
#[derive(Debug, Clone)]
pub struct PythonInfo {
#[allow(dead_code)]
pub short_version: (u64, u64),
pub path: PathBuf,
pub site_packages_path: PathBuf,
pub bin_dir: PathBuf,
}
#[derive(Debug, Clone)]
pub struct EntryPoint {
pub command: String,
pub module: String,
pub function: String,
}
#[derive(Debug, Clone)]
pub struct IndexJson {
#[allow(dead_code)]
pub name: String,
pub noarch: Option<String>,
pub python_site_packages_path: Option<String>,
}
#[derive(Debug, Clone)]
pub struct PathsEntry {
pub relative_path: PathBuf,
pub path_type: String,
#[allow(dead_code)]
pub sha256: Option<String>,
#[allow(dead_code)]
pub size_in_bytes: Option<u64>,
pub prefix_placeholder: Option<PrefixPlaceholder>,
pub no_link: bool,
}
#[derive(Debug, Clone)]
pub struct PrefixPlaceholder {
pub placeholder: String,
pub file_mode: FileMode,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FileMode {
Text,
Binary,
}
fn read_index_json(package_dir: &Path) -> Result<IndexJson> {
let index_path = crate::dirs::path_join(package_dir, &["info", "conda", "index.json"]);
let index_data: Value = crate::io::read_json_file(&index_path)
.wrap_err_with(|| format!("Failed to read index.json from {}", package_dir.display()))?;
let name = index_data.get("name")
.and_then(|v| v.as_str())
.ok_or_else(|| eyre::eyre!("Missing 'name' in index.json"))?
.to_string();
let noarch = index_data.get("noarch")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let python_site_packages_path = index_data.get("python_site_packages_path")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
Ok(IndexJson {
name,
noarch,
python_site_packages_path,
})
}
fn parse_paths_entry(path_entry: &Value) -> Result<PathsEntry> {
let relative_path = path_entry.get("_path")
.or_else(|| path_entry.get("path"))
.and_then(|v| v.as_str())
.ok_or_else(|| eyre::eyre!("Missing path in paths.json entry"))?;
let path_type = path_entry.get("path_type")
.and_then(|v| v.as_str())
.unwrap_or("file")
.to_string();
let sha256 = path_entry.get("sha256")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let size_in_bytes = path_entry.get("size_in_bytes")
.and_then(|v| v.as_u64());
let prefix_placeholder = path_entry.get("prefix_placeholder")
.and_then(|v| {
match v {
Value::String(s) => {
let file_mode_str = path_entry
.get("file_mode")
.and_then(|m| m.as_str())
.unwrap_or("text");
let file_mode = if file_mode_str == "binary" {
FileMode::Binary
} else {
FileMode::Text
};
Some(PrefixPlaceholder {
placeholder: s.to_string(),
file_mode,
})
}
Value::Object(map) => {
let placeholder = map.get("placeholder")?.as_str()?.to_string();
let file_mode_str = map
.get("file_mode")
.and_then(|m| m.as_str())
.unwrap_or("text");
let file_mode = if file_mode_str == "binary" {
FileMode::Binary
} else {
FileMode::Text
};
Some(PrefixPlaceholder {
placeholder,
file_mode,
})
}
_ => None,
}
});
let no_link = path_entry.get("no_link")
.and_then(|v| v.as_bool())
.unwrap_or(false);
Ok(PathsEntry {
relative_path: normalize_conda_path(relative_path),
path_type,
sha256,
size_in_bytes,
prefix_placeholder,
no_link,
})
}
fn read_paths_json(package_dir: &Path) -> Result<Vec<PathsEntry>> {
let paths_path = crate::dirs::path_join(package_dir, &["info", "conda", "paths.json"]);
if !lfs::exists_on_host(&paths_path) {
log::info!("paths.json not found: {}", paths_path.display());
return Ok(Vec::new());
}
let paths_data: Value = read_json_file(&paths_path)
.wrap_err_with(|| format!("Failed to read paths.json from {}", package_dir.display()))?;
let mut entries = Vec::new();
if let Some(paths_array) = paths_data.get("paths").and_then(|v| v.as_array()) {
for path_entry in paths_array {
entries.push(parse_paths_entry(path_entry)?);
}
}
Ok(entries)
}
fn parse_entry_point(ep: &Value) -> Result<EntryPoint> {
let ep_obj = ep.as_object()
.ok_or_else(|| eyre::eyre!("Entry point is not an object"))?;
let command = ep_obj.get("command")
.and_then(|v| v.as_str())
.ok_or_else(|| eyre::eyre!("Missing 'command' in entry point"))?
.to_string();
let func_str = ep_obj.get("func")
.or_else(|| ep_obj.get("function"))
.and_then(|v| v.as_str())
.ok_or_else(|| eyre::eyre!("Missing 'func' or 'function' in entry point"))?;
let (module, function) = if let Some((m, f)) = func_str.split_once(':') {
(m.to_string(), f.to_string())
} else {
let parts: Vec<&str> = func_str.split('.').collect();
if parts.len() >= 2 {
let mod_parts = &parts[..parts.len() - 1];
let func = parts.last().unwrap();
(mod_parts.join("."), func.to_string())
} else {
return Err(eyre::eyre!("Invalid function format: {}", func_str));
}
};
Ok(EntryPoint {
command,
module,
function,
})
}
fn read_link_json(package_dir: &Path) -> Result<Option<Vec<EntryPoint>>> {
let link_path = crate::dirs::path_join(package_dir, &["info", "conda", "link.json"]);
if !lfs::exists_on_host(&link_path) {
return Ok(None);
}
let link_data: Value = crate::io::read_json_file(&link_path)
.wrap_err_with(|| format!("Failed to read link.json from {}", package_dir.display()))?;
let noarch = link_data.get("noarch")
.and_then(|v| v.as_str());
if noarch != Some("python") {
return Ok(None);
}
let entry_points = link_data.get("entry_points")
.and_then(|v| v.as_array())
.ok_or_else(|| eyre::eyre!("Missing entry_points in link.json"))?;
let mut result = Vec::new();
for ep in entry_points {
result.push(parse_entry_point(ep)?);
}
Ok(Some(result))
}
fn get_python_version_from_installed() -> Option<(u64, u64)> {
use crate::models::PACKAGE_CACHE;
use crate::package;
let pkgkey_opt = {
let installed = PACKAGE_CACHE.installed_packages.read().unwrap();
installed.iter()
.find(|(pkgkey, _pkg)| {
if let Ok(pkgname) = package::pkgkey2pkgname(pkgkey) {
pkgname == "python"
} else {
false
}
})
.map(|(pkgkey, _)| pkgkey.clone())
};
if let Some(pkgkey) = pkgkey_opt {
if let Ok(version_str) = package::pkgkey2version(&pkgkey) {
let version_clean = version_str.split('-').next().unwrap_or(&version_str);
return extract_python_version_from_string(version_clean);
}
}
None
}
fn get_python_version_from_index(index_json: &IndexJson) -> Option<(u64, u64)> {
if let Some(sp_path) = &index_json.python_site_packages_path {
return extract_python_version_from_path(sp_path);
}
None
}
fn get_python_info(index_json: &IndexJson) -> Result<Option<PythonInfo>> {
let (major, minor) = get_python_version_from_installed()
.or_else(|| get_python_version_from_index(index_json))
.unwrap_or_else(|| {
log::debug!("Could not determine Python version, defaulting to {}", DEFAULT_PYTHON_VERSION_STR);
DEFAULT_PYTHON_VERSION
});
#[cfg(unix)]
let path = PathBuf::from(format!("bin/python{major}.{minor}"));
#[cfg(windows)]
let path = PathBuf::from(format!("Scripts/python{major}.{minor}.exe"));
let site_packages_path = index_json.python_site_packages_path
.as_ref()
.map(|s| normalize_conda_path(s))
.unwrap_or_else(|| {
#[cfg(unix)]
{ PathBuf::from(format!("lib/python{major}.{minor}/site-packages")) }
#[cfg(windows)]
{ PathBuf::from("Lib\\site-packages") }
});
#[cfg(unix)]
let bin_dir = PathBuf::from("bin");
#[cfg(windows)]
let bin_dir = PathBuf::from("Scripts");
Ok(Some(PythonInfo {
short_version: (major, minor),
path,
site_packages_path,
bin_dir,
}))
}
fn extract_python_version_from_string(version: &str) -> Option<(u64, u64)> {
let parts: Vec<&str> = version.split('.').collect();
if parts.len() >= 2 {
let major = parts[0].parse().ok()?;
let minor = parts[1].parse().ok()?;
return Some((major, minor));
}
None
}
fn extract_python_version_from_path(path: &str) -> Option<(u64, u64)> {
let re = regex::Regex::new(r"python(\d+)\.(\d+)").ok()?;
if let Some(captures) = re.captures(path) {
let major = captures.get(1)?.as_str().parse().ok()?;
let minor = captures.get(2)?.as_str().parse().ok()?;
return Some((major, minor));
}
None
}
fn compute_paths(
index_json: &IndexJson,
paths_entries: &[PathsEntry],
python_info: Option<&PythonInfo>,
) -> Vec<(PathsEntry, PathBuf)> {
let mut final_paths = Vec::new();
let is_noarch_python = index_json.noarch.as_ref()
.map(|n| n == "python")
.unwrap_or(false);
for entry in paths_entries {
let path = if is_noarch_python {
if let Some(py_info) = python_info {
remap_noarch_path(&entry.relative_path, py_info)
} else {
entry.relative_path.clone()
}
} else {
entry.relative_path.clone()
};
final_paths.push((entry.clone(), path));
}
final_paths
}
fn remap_noarch_path(relative_path: &Path, python_info: &PythonInfo) -> PathBuf {
#[cfg(unix)]
let (site_packages_prefix, python_scripts_prefix) = ("site-packages/", "python-scripts/");
#[cfg(windows)]
let (site_packages_prefix, python_scripts_prefix) = (r"site-packages\", r"python-scripts\");
if let Ok(rest) = relative_path.strip_prefix(site_packages_prefix) {
return python_info.site_packages_path.join(rest);
}
if let Ok(rest) = relative_path.strip_prefix(python_scripts_prefix) {
return python_info.bin_dir.join(rest);
}
relative_path.to_path_buf()
}
#[cfg(unix)]
fn replace_shebang<'a>(
shebang: Cow<'a, str>,
old_prefix: &str,
new_prefix: &str,
) -> Cow<'a, str> {
assert!(
shebang.starts_with("#!"),
"Shebang does not start with #! ({})",
shebang
);
if new_prefix.contains(' ') {
if !shebang.contains(old_prefix) {
return shebang;
}
let new_shebang = convert_shebang_to_env(shebang).replace(old_prefix, new_prefix);
return Cow::Owned(new_shebang);
}
let shebang: Cow<'_, str> = shebang.replace(old_prefix, new_prefix).into();
if !shebang.starts_with("#!") {
log::warn!("Shebang does not start with #! ({})", shebang);
return shebang;
}
if is_valid_shebang_length(&shebang) {
shebang
} else {
convert_shebang_to_env(shebang)
}
}
fn copy_replace_textual_placeholder(
source_path: &Path,
target_path: &Path,
prefix_placeholder: &str,
target_prefix: &str,
) -> Result<()> {
let source_bytes = fs::read(source_path)
.wrap_err_with(|| format!("Failed to read source file: {}", source_path.display()))?;
let mut target_file = lfs::file_create(target_path)?;
let old_prefix = prefix_placeholder.as_bytes();
let new_prefix = target_prefix.as_bytes();
#[allow(unused_mut)]
let mut source_bytes = source_bytes.as_slice();
#[cfg(unix)]
{
if source_bytes.starts_with(b"#!") {
let newline_pos = source_bytes.iter().position(|&c| c == b'\n').unwrap_or(source_bytes.len());
let (first, rest) = source_bytes.split_at(newline_pos);
let first_line = String::from_utf8_lossy(first);
let new_shebang = replace_shebang(
first_line,
prefix_placeholder,
target_prefix,
);
target_file.write_all(new_shebang.as_bytes())
.wrap_err_with(|| format!("Failed to write shebang to {}", target_path.display()))?;
source_bytes = rest;
}
}
let mut last_match = 0;
for index in memmem::find_iter(source_bytes, old_prefix) {
target_file.write_all(&source_bytes[last_match..index])
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
target_file.write_all(new_prefix)
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
last_match = index + old_prefix.len();
}
if last_match < source_bytes.len() {
target_file.write_all(&source_bytes[last_match..])
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
}
Ok(())
}
fn copy_replace_cstring_placeholder(
source_path: &Path,
target_path: &Path,
prefix_placeholder: &str,
target_prefix: &str,
) -> Result<()> {
let source_bytes = fs::read(source_path)
.wrap_err_with(|| format!("Failed to read binary file: {}", source_path.display()))?;
let mut target_file = lfs::file_create(target_path)?;
let old_prefix = prefix_placeholder.as_bytes();
let new_prefix = target_prefix.as_bytes();
let mut source_bytes = source_bytes.as_slice();
let finder = memmem::Finder::new(old_prefix);
loop {
if let Some(index) = finder.find(source_bytes) {
target_file.write_all(&source_bytes[..index])
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
let mut end = index + old_prefix.len();
while end < source_bytes.len() && source_bytes[end] != b'\0' {
end += 1;
}
let mut out = Vec::new();
let mut old_bytes = &source_bytes[index..end];
let old_len = old_bytes.len();
while let Some(sub_index) = finder.find(old_bytes) {
out.write_all(&old_bytes[..sub_index])
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
out.write_all(new_prefix)
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
old_bytes = &old_bytes[sub_index + old_prefix.len()..];
}
out.write_all(old_bytes)
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
if out.len() > old_len {
target_file.write_all(&out[..old_len])
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
} else {
target_file.write_all(&out)
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
}
let padding = old_len.saturating_sub(out.len());
if padding > 0 {
target_file.write_all(&vec![0u8; padding])
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
}
source_bytes = &source_bytes[end..];
} else {
target_file.write_all(source_bytes)
.wrap_err_with(|| format!("Failed to write to {}", target_path.display()))?;
return Ok(());
}
}
}
#[cfg(unix)]
fn create_unix_python_entry_point(
target_dir: &Path,
target_prefix: &str,
entry_point: &EntryPoint,
python_info: &PythonInfo,
) -> Result<PathBuf> {
let relative_path = python_info.bin_dir.join(&entry_point.command);
let script_path = target_dir.join(&relative_path);
if let Some(parent) = script_path.parent() {
lfs::create_dir_all(parent)?;
}
let python_path = Path::new(target_prefix).join(&python_info.path);
let python_path_str = python_path.to_string_lossy().replace('\\', "/");
let shebang = if python_path_str.len() > 125 || python_path_str.contains(' ') {
// Use exec wrapper for long shebangs or paths with spaces
format!("#!/bin/sh\n'''exec' \"{}\" \"$0\" \"$@\" #'''", python_path_str)
} else {
format!("#!{}", python_path_str)
};
// Generate entry point script content
let (import_name, _) = entry_point.function.split_once('.')
.unwrap_or((&entry_point.function, ""));
let script_content = format!(
"{}\n\
# -*- coding: utf-8 -*-\n\
import re\n\
import sys\n\n\
from {} import {}\n\n\
if __name__ == '__main__':\n\
\tsys.argv[0] = re.sub(r'(-script\\.pyw?|\\.exe)?$', '', sys.argv[0])\n\
\tsys.exit({}())\n",
shebang,
entry_point.module,
import_name,
entry_point.function
);
lfs::write(&script_path, script_content)?;
// Make executable
utils::set_executable_permissions(&script_path, 0o775)?;
log::debug!("Created Python entry point: {}", script_path.display());
Ok(relative_path)
}
/// Create Windows Python entry point script for conda packages.
///
/// On Windows, Python entry points are typically `.exe` files bundled with the package.
/// For noarch packages, we create a `.exe` wrapper using Python's `-c` approach.
#[cfg(windows)]
fn create_windows_python_entry_point(
target_dir: &Path,
target_prefix: &str,
entry_point: &EntryPoint,
python_info: &PythonInfo,
) -> Result<PathBuf> {
let relative_path = python_info.bin_dir.join(format!("{}.exe", entry_point.command));
let script_path = target_dir.join(&relative_path);
if let Some(parent) = script_path.parent() {
lfs::create_dir_all(parent)?;
}
let python_path = Path::new(target_prefix).join(&python_info.path);
let bat_path = target_dir.join(python_info.bin_dir.join(format!("{}.bat", entry_point.command)));
let bat_content = format!(
"@echo off\n\"{}\" -c \"from {} import {}; {}()\"\n",
python_path.display(),
entry_point.module,
entry_point.function,
entry_point.function
);
lfs::write(&bat_path, bat_content)?;
log::debug!("Created Python entry point batch file: {}", bat_path.display());
Ok(relative_path)
}
fn prepare_conda_package_metadata(
package_dir: &Path,
) -> Result<(IndexJson, Vec<PathsEntry>, Option<PythonInfo>)> {
let index_json = read_index_json(package_dir)
.wrap_err_with(|| format!("Failed to read index.json from {}", package_dir.display()))?;
let paths_entries = read_paths_json(package_dir)
.wrap_err_with(|| format!("Failed to read paths.json from {}", package_dir.display()))?;
let python_info = if index_json.noarch.as_ref().map(|n| n == "python").unwrap_or(false) {
get_python_info(&index_json)
.wrap_err_with(|| "Failed to get Python info for noarch package")?
} else {
None
};
Ok((index_json, paths_entries, python_info))
}
fn copy_file_with_prefix_replacement(
source_path: &Path,
target_path: &Path,
placeholder_info: &PrefixPlaceholder,
target_prefix: &str,
) -> Result<()> {
let target_path = lfs::resolve_ancestor_symlink(target_path);
match placeholder_info.file_mode {
FileMode::Text => {
copy_replace_textual_placeholder(
source_path,
&target_path,
&placeholder_info.placeholder,
target_prefix,
)?;
}
FileMode::Binary => {
copy_replace_cstring_placeholder(
source_path,
&target_path,
&placeholder_info.placeholder,
target_prefix,
)?;
}
}
crate::utils::preserve_file_permissions(source_path, &target_path)?;
Ok(())
}
fn link_file_without_prefix_replacement(
plan: &InstallationPlan,
source_path: &Path,
target_path: &Path,
path_type: &str,
fhs_file: &Path,
) -> Result<()> {
let is_link = lfs::is_symlink(source_path);
let link_type = if path_type == "hardlink" && plan.can_hardlink {
LinkType::Hardlink
} else if path_type == "softlink" && plan.can_symlink {
LinkType::Symlink
} else {
plan.link
};
let target_path = lfs::resolve_ancestor_symlink(target_path);
mirror_file(source_path, &target_path, fhs_file, is_link, link_type, plan.can_reflink)?;
Ok(())
}
fn link_conda_files(
plan: &InstallationPlan,
store_fs_dir: &PathBuf,
final_paths: Vec<(PathsEntry, PathBuf)>,
target_prefix: &str,
) -> Result<()> {
for (entry, computed_path) in final_paths {
if entry.path_type == "directory" || entry.no_link {
continue;
}
let source_path = store_fs_dir.join(&entry.relative_path);
let target_path = plan.env_root.join(&computed_path);
if let Some(parent) = target_path.parent() {
let resolved_parent = lfs::resolve_ancestor_symlink(parent);
lfs::create_dir_all(&resolved_parent)?;
}
if let Some(placeholder_info) = &entry.prefix_placeholder {
copy_file_with_prefix_replacement(
&source_path,
&target_path,
placeholder_info,
target_prefix,
)?;
} else {
link_file_without_prefix_replacement(
plan,
&source_path,
&target_path,
&entry.path_type,
&computed_path,
)?;
}
}
Ok(())
}
fn create_conda_entry_points(
plan: &InstallationPlan,
package_dir: &Path,
python_info: &PythonInfo,
target_prefix: &str,
) -> Result<()> {
if let Ok(Some(entry_points)) = read_link_json(package_dir) {
for entry_point in entry_points {
#[cfg(unix)]
create_unix_python_entry_point(
&plan.env_root,
target_prefix,
&entry_point,
python_info,
)?;
#[cfg(windows)]
create_windows_python_entry_point(
&plan.env_root,
target_prefix,
&entry_point,
python_info,
)?;
}
}
Ok(())
}
pub fn link_conda_package(plan: &InstallationPlan, store_fs_dir: &PathBuf) -> Result<()> {
let package_dir = store_fs_dir.parent()
.ok_or_else(|| eyre::eyre!("Invalid store_fs_dir path: {}", store_fs_dir.display()))?;
if plan.link == crate::models::LinkType::Move {
if let Some(_fs_files) = crate::link::handle_move_link_type(package_dir, store_fs_dir, &plan.env_root)? {
return Ok(());
}
crate::store::create_consumed_marker(package_dir, &plan.env_root.display().to_string(), &plan.env_root)
.with_context(|| format!("Failed to create consumed marker for {}", package_dir.display()))?;
}
let (index_json, paths_entries, python_info) = prepare_conda_package_metadata(package_dir)?;
if paths_entries.is_empty() {
log::info!("paths.json missing or empty for {}, falling back to generic linking", package_dir.display());
return crate::link::link_package_generic(plan, store_fs_dir);
}
let final_paths = compute_paths(&index_json, &paths_entries, python_info.as_ref());
let target_prefix = plan.env_root.to_string_lossy().to_string();
link_conda_files(plan, store_fs_dir, final_paths, &target_prefix)?;
if let Some(py_info) = python_info {
create_conda_entry_points(plan, package_dir, &py_info, &target_prefix)?;
}
Ok(())
}