Python构建企业级离线包仓库的实战指南与脚本

import pandas as pd
import subprocess
from tqdm import tqdm

# 读取热门包列表 https://hugovk.github.io/top-pypi-packages/
data = pd.read_csv("top-pypi-packages.csv")
data = data.loc[data["project"].apply(lambda x: isinstance(x, str)), "project"].values.tolist()

i = 0
for package in tqdm(data):
    i += 1
    if i <= 8578:
        continue
        
    command = "C:/Users/dfy918/AppData/Local/Programs/Python/Python312/Scripts/pip download -i https://mirrors.aliyun.com/pypi/simple/ --python-version 313 --only-binary=:all: --platform manylinux_2_17_x86_64 --no-cache-dir -d ./packages {} ".format(package)

    result = subprocess.run(command, shell=True)

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def download_package(package):
    command = f"pip download -i https://mirrors.aliyun.com/pypi/simple/ --python-version 313 --only-binary=:all: --platform manylinux_2_17_x86_64 --no-cache-dir -d ./packages {package}"
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"下载失败: {package}, 错误信息: {result.stderr}")
        raise Exception(f"下载失败: {result.stderr}")
    return result

# 每5个包为一组进行下载
batch_size = 5
for i in tqdm(range(0, len(data), batch_size)):
    batch_packages = data[i:i+batch_size]
    command = f"pip download -i https://mirrors.aliyun.com/pypi/simple/ --python-version 313 --only-binary=:all: --platform manylinux_2_17_x86_64 -d ./packages {' '.join(batch_packages)}"
    result = subprocess.run(command, shell=True)

pypiserver run -p 8080 -P .htpasswd -a update,download E:\python_packages

pip install package_name -i http://localhost:8080/simple/ --trusted-host localhost

[global]
index-url = http://localhost:8080/simple/
trusted-host = localhost

pip install package_name -i http://username:password@localhost:8080/simple/ --trusted-host localhost

#!/usr/bin/env python3
"""
企业级Python离线包仓库构建工具
增强版本：包含错误处理、重试机制和进度跟踪
"""

import pandas as pd
import subprocess
import os
from tqdm import tqdm
from tenacity import retry, stop_after_attempt, wait_fixed

class OfflinePackageManager:
    def __init__(self, python_version="313", platform="manylinux_2_17_x86_64"):
        self.python_version = python_version
        self.platform = platform
        self.package_dir = "./packages"
        
        # 创建包存储目录
        os.makedirs(self.package_dir, exist_ok=True)
    
    @retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
    def download_package(self, package):
        """下载单个包，包含重试机制"""
        command = (f"pip download -i https://mirrors.aliyun.com/pypi/simple/ "
                  f"--python-version {self.python_version} --only-binary=:all: "
                  f"--platform {self.platform} --no-cache-dir "
                  f"-d {self.package_dir} {package}")
        
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        if result.returncode != 0:
            print(f"下载失败: {package}, 错误信息: {result.stderr}")
            raise Exception(f"下载失败: {result.stderr}")
        return result
    
    def download_from_checkpoint(self, checkpoint_file="checkpoint.txt"):
        """从检查点继续下载"""
        # 实现检查点逻辑
        pass
    
    def generate_requirements(self):
        """生成requirements.txt文件"""
        # 实现需求文件生成逻辑
        pass

if __name__ == "__main__":
    manager = OfflinePackageManager()
    # 这里可以添加更复杂的逻辑