138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
import subprocess
|
||
from typing import List, Optional
|
||
import re
|
||
import subprocess
|
||
import shlex
|
||
from typing import List, Optional, Dict, Any, Union
|
||
from dataclasses import dataclass
|
||
|
||
@dataclass
|
||
class PipResult:
|
||
"""Результат выполнения команды pip."""
|
||
exit_code: int
|
||
stdout: str
|
||
stderr: str
|
||
command: str
|
||
success: bool
|
||
|
||
|
||
class pip_api:
|
||
@staticmethod
|
||
def get_pkg_versions(package: str, index_url: Optional[str] = None) -> List[str]:
|
||
# Формируем базовую команду
|
||
command = [
|
||
'python3', '-m', 'pip',
|
||
'install',
|
||
'--use-deprecated=legacy-resolver', # Для совместимости со старыми репозиториями
|
||
'--dry-run',
|
||
'--no-deps',
|
||
f'{package}==0.0.0.0' # Специально несуществующая версия
|
||
]
|
||
|
||
if index_url:
|
||
command.extend(['--index-url', index_url])
|
||
|
||
# Запускаем процесс
|
||
result = subprocess.run(
|
||
command,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=30 # Таймаут для избежания зависаний
|
||
)
|
||
|
||
#print(result.stdout)
|
||
#print(result.stderr)
|
||
# Парсим вывод для получения версий
|
||
if result.stderr:
|
||
match = re.search(r'from versions: (.+?)\)', result.stderr)
|
||
if match:
|
||
versions = match.group(1).split(', ')
|
||
return [v.strip() for v in versions if v.strip()]
|
||
|
||
return []
|
||
|
||
@staticmethod
|
||
def run_pip_install(
|
||
pip_path: str,
|
||
packages: List[str],
|
||
index_url: Optional[str] = None,
|
||
extra_index_urls: Optional[List[str]] = None,
|
||
dry_run: bool = False
|
||
) -> PipResult:
|
||
"""
|
||
Выполняет установку пакетов через pip с указанными параметрами.
|
||
|
||
Args:
|
||
pip_path: Путь к исполняемому файлу pip
|
||
packages: Список пакетов для установки
|
||
index_url: Основной URL репозитория пакетов
|
||
extra_index_urls: Дополнительные URLs репозиториев пакетов
|
||
dry_run: Если True, команда только выводится, но не выполняется
|
||
|
||
Returns:
|
||
PipResult: Объект с результатами выполнения команды
|
||
|
||
Raises:
|
||
FileNotFoundError: Если pip_path не существует
|
||
ValueError: Если packages пуст
|
||
"""
|
||
# Проверка входных параметров
|
||
if not packages:
|
||
raise ValueError("Список пакетов не может быть пустым")
|
||
|
||
# Формирование команды
|
||
command = [pip_path, "install"]
|
||
|
||
if dry_run:
|
||
command.append('--dry-run')
|
||
|
||
# Добавление основного index-url
|
||
if index_url:
|
||
command.extend(["--index-url", index_url])
|
||
|
||
# Добавление дополнительных index-urls
|
||
if extra_index_urls:
|
||
for url in extra_index_urls:
|
||
command.extend(["--extra-index-url", url])
|
||
|
||
# Добавление пакетов
|
||
command.extend(packages)
|
||
|
||
# Преобразование команды в строку для вывода
|
||
command_str = " ".join(shlex.quote(arg) for arg in command)
|
||
|
||
# Выполнение команды
|
||
try:
|
||
result = subprocess.run(
|
||
command,
|
||
capture_output=True,
|
||
text=True,
|
||
check=False # Не вызываем исключение при ненулевом коде возврата
|
||
)
|
||
|
||
return PipResult(
|
||
exit_code=result.returncode,
|
||
stdout=result.stdout,
|
||
stderr=result.stderr,
|
||
command=command_str,
|
||
success=result.returncode == 0
|
||
)
|
||
|
||
except FileNotFoundError:
|
||
raise FileNotFoundError(f"Файл pip не найден по пути: {pip_path}")
|
||
except Exception as e:
|
||
return PipResult(
|
||
exit_code=-1,
|
||
stdout="",
|
||
stderr=str(e),
|
||
command=command_str,
|
||
success=False
|
||
)
|
||
|
||
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
versions = pip_api.get_pkg_versions("torch", "https://download.pytorch.org/whl/cu121")
|
||
print(versions) |