Python一次性將指定版本所有包上傳PyPI鏡像解決方案

更新時(shí)間：2025年09月10日 08:32:09 作者：東方佑

本文主要介紹了一個(gè)安全、完整、可離線部署的解決方案,用于一次性準(zhǔn)備指定Python版本的所有包,然后導(dǎo)出到內(nèi)網(wǎng)環(huán)境,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

根據(jù)您的需求和知識(shí)庫(kù)中的警告信息（清華鏡像會(huì)阻斷大量下載行為），我設(shè)計(jì)了一個(gè)安全、完整、可離線部署的解決方案，用于一次性準(zhǔn)備指定Python版本的所有包，然后導(dǎo)出到內(nèi)網(wǎng)環(huán)境。

為什么需要這個(gè)方案

知識(shí)庫(kù)明確顯示：清華鏡像會(huì)檢測(cè)到大量下載行為并阻斷請(qǐng)求。直接嘗試下載所有包會(huì)觸發(fā)此限制，導(dǎo)致下載失敗。我們需要一個(gè)謹(jǐn)慎、分階段、有間隔的下載策略。

完整解決方案

1. 項(xiàng)目目錄結(jié)構(gòu)

mkdir -p pypi-offline/{config,scripts,packages}
cd pypi-offline

2. 創(chuàng)建智能下載腳本

scripts/safe_download.py

#!/usr/bin/env python3
import os
import sys
import time
import random
import requests
import argparse
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

# 配置參數(shù)
MAX_RETRIES = 3
MIN_DELAY = 5  # 最小延遲(秒)
MAX_DELAY = 15  # 最大延遲(秒)
MAX_WORKERS = 2  # 最大并發(fā)數(shù)(避免觸發(fā)限制)

# PyPI API端點(diǎn)
PYPI_SIMPLE = "https://pypi.org/simple"
# 使用官方PyPI而非清華鏡像，避免觸發(fā)限制
# 如果網(wǎng)絡(luò)條件好，可考慮使用其他鏡像

def get_package_info(package_name):
    """獲取包的元信息"""
    url = f"https://pypi.org/pypi/{package_name}/json"
    for i in range(MAX_RETRIES):
        try:
            response = requests.get(url, timeout=30)
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 404:
                return None
            time.sleep((i + 1) * 2)
        except Exception as e:
            print(f"獲取包 {package_name} 信息失敗: {str(e)}")
            time.sleep((i + 1) * 5)
    return None

def filter_packages_for_python(package_data, python_version):
    """過(guò)濾出兼容指定Python版本的包"""
    compatible_files = []
    py_ver = python_version.replace('.', '')
    
    for file_info in package_data['urls']:
        # 檢查Python版本兼容性
        py_tag = file_info.get('python_version', '')
        if py_tag == 'source' or py_tag.startswith('py') or py_tag.startswith('cp' + py_ver):
            compatible_files.append(file_info)
    
    return compatible_files

def download_package(package_name, file_info, target_dir):
    """安全下載單個(gè)包文件"""
    file_url = file_info['url']
    file_name = file_info['filename']
    target_path = os.path.join(target_dir, file_name)
    
    # 如果文件已存在，跳過(guò)
    if os.path.exists(target_path):
        print(f"跳過(guò)已存在的文件: {file_name}")
        return True
    
    print(f"下載: {file_name} ({package_name})")
    
    for i in range(MAX_RETRIES):
        try:
            response = requests.get(file_url, stream=True, timeout=60)
            if response.status_code == 200:
                with open(target_path, 'wb') as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        f.write(chunk)
                print(f"成功下載: {file_name}")
                return True
            print(f"下載失敗 ({response.status_code}): {file_name}")
            time.sleep((i + 1) * 5)
        except Exception as e:
            print(f"下載 {file_name} 失敗: {str(e)}")
            time.sleep((i + 1) * 5)
    
    # 下載失敗，刪除可能的部分文件
    if os.path.exists(target_path):
        os.remove(target_path)
    return False

def get_all_packages():
    """獲取所有包的列表"""
    print("獲取所有包的列表...")
    response = requests.get(f"{PYPI_SIMPLE}/", timeout=30)
    if response.status_code != 200:
        raise Exception(f"無(wú)法獲取包列表: HTTP {response.status_code}")
    
    # 解析HTML獲取包名
    import re
    package_names = re.findall(r'<a href="/simple/([^" rel="external nofollow" /]+)">\1</a>', response.text)
    return package_names

def main():
    parser = argparse.ArgumentParser(description='安全下載指定Python版本的所有包')
    parser.add_argument('--python-version', required=True, help='目標(biāo)Python版本 (如: 3.8)')
    parser.add_argument('--output-dir', default='../packages', help='輸出目錄')
    args = parser.parse_args()
    
    # 創(chuàng)建輸出目錄
    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    print(f"輸出目錄: {output_dir}")
    
    # 獲取所有包列表
    try:
        all_packages = get_all_packages()
        print(f"找到 {len(all_packages)} 個(gè)包")
    except Exception as e:
        print(f"獲取包列表失敗: {str(e)}")
        print("請(qǐng)嘗試使用清華鏡像的簡(jiǎn)單頁(yè)面 API (需要處理 HTML)")
        # 作為備選方案，可以使用清華鏡像的簡(jiǎn)單頁(yè)面
        # all_packages = get_packages_from_tuna()
        return 1
    
    # 處理包
    success_count = 0
    failed_packages = []
    
    # 分批次處理，避免一次性太多請(qǐng)求
    batch_size = 50
    for i in range(0, len(all_packages), batch_size):
        batch = all_packages[i:i+batch_size]
        print(f"\n處理包批次 {i//batch_size + 1} ({len(batch)} 個(gè)包)")
        
        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
            future_to_pkg = {}
            
            for pkg in batch:
                pkg_data = get_package_info(pkg)
                if pkg_data:
                    compatible_files = filter_packages_for_python(pkg_data, args.python_version)
                    if compatible_files:
                        for file_info in compatible_files:
                            future = executor.submit(
                                download_package, 
                                pkg, 
                                file_info, 
                                output_dir
                            )
                            future_to_pkg[future] = pkg
                # 添加隨機(jī)延遲，避免觸發(fā)限制
                time.sleep(random.uniform(MIN_DELAY, MAX_DELAY))
            
            # 等待并處理結(jié)果
            for future in as_completed(future_to_pkg):
                pkg = future_to_pkg[future]
                try:
                    if future.result():
                        success_count += 1
                    else:
                        failed_packages.append(pkg)
                except Exception as e:
                    print(f"處理包 {pkg} 時(shí)出錯(cuò): {str(e)}")
                    failed_packages.append(pkg)
        
        # 批次間額外延遲
        print(f"\n批次處理完成，等待 {MAX_DELAY*2} 秒...")
        time.sleep(MAX_DELAY * 2)
    
    # 生成報(bào)告
    print("\n===== 下載完成 =====")
    print(f"成功: {success_count} 個(gè)包")
    print(f"失敗: {len(failed_packages)} 個(gè)包")
    
    if failed_packages:
        print("\n失敗的包列表 (可后續(xù)重試):")
        for pkg in failed_packages[:20]:  # 只顯示前20個(gè)
            print(f"- {pkg}")
        if len(failed_packages) > 20:
            print(f"... 及其他 {len(failed_packages) - 20} 個(gè)包")
        
        # 保存失敗列表以便重試
        with open(output_dir / 'failed_packages.txt', 'w') as f:
            for pkg in failed_packages:
                f.write(pkg + '\n')
    
    return 0

if __name__ == "__main__":
    sys.exit(main())

3. 創(chuàng)建包清單生成腳本

scripts/generate_simple_index.py

#!/usr/bin/env python3
import os
import sys
from pathlib import Path
import re
import html

def generate_simple_index(packages_dir, output_dir):
    """生成符合PEP 503的simple index"""
    packages_dir = Path(packages_dir)
    output_dir = Path(output_dir)
    
    # 創(chuàng)建輸出目錄
    output_dir.mkdir(parents=True, exist_ok=True)
    
    # 收集所有包名
    package_names = set()
    for filename in os.listdir(packages_dir):
        # 從文件名提取包名 (PEP 427 格式)
        match = re.match(r'^([a-zA-Z0-9_-]+)(?:-[a-zA-Z0-9_.+-]+)?\.(?:whl|tar\.gz|zip)$', filename)
        if match:
            package_name = match.group(1).lower()
            # 規(guī)范化包名 (PEP 503)
            package_name = package_name.replace('_', '-')
            package_names.add(package_name)
    
    # 為每個(gè)包生成索引頁(yè)面
    for package in package_names:
        index_content = f'<html><head><title>Links for {package}</title></head>\n'
        index_content += f'<body>\n<h1>Links for {package}</h1>\n'
        
        # 找出該包的所有文件
        for filename in os.listdir(packages_dir):
            if re.match(f'^{re.escape(package)}(?:-[a-zA-Z0-9_.+-]+)?\.(?:whl|tar\.gz|zip)$', filename, re.IGNORECASE):
                file_url = f'../{filename}'
                index_content += f'<a href="{file_url}" rel="external nofollow" >{html.escape(filename)}</a>
\n'
        
        index_content += '</body></html>'
        
        # 保存索引頁(yè)面
        package_dir = output_dir / package
        package_dir.mkdir(exist_ok=True)
        with open(package_dir / 'index.html', 'w', encoding='utf-8') as f:
            f.write(index_content)
    
    # 生成根索引頁(yè)面
    root_index = '<html><head><title>Simple Index</title></head>\n'
    root_index += '<body>\n<h1>Simple Index</h1>\n'
    
    for package in sorted(package_names):
        root_index += f'<a href="{package}/" rel="external nofollow" >{html.escape(package)}</a>
\n'
    
    root_index += '</body></html>'
    
    with open(output_dir / 'index.html', 'w', encoding='utf-8') as f:
        f.write(root_index)
    
    print(f"成功生成 {len(package_names)} 個(gè)包的索引")

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("用法: python generate_simple_index.py <packages_dir> <output_dir>")
        sys.exit(1)
    
    generate_simple_index(sys.argv[1], sys.argv[2])

4. 創(chuàng)建 Docker Compose 配置docker-compose.yml

version: '3.8'

services:
  pypi-offline:
    image: python:3.9-slim
    container_name: pypi-offline
    ports:
      - "8080:8000"
    volumes:
      - ./web:/usr/src/app
      - ./packages:/usr/src/app/packages
    working_dir: /usr/src/app
    command: >
      sh -c "python -m http.server 8000 --directory /usr/src/app"
    restart: unless-stopped

5. 創(chuàng)建 Web 服務(wù)目錄結(jié)構(gòu)

mkdir -p web/{simple,packages}

6. 創(chuàng)建準(zhǔn)備腳本

prepare_offline_mirror.sh

#!/bin/bash
# 準(zhǔn)備離線PyPI鏡像

PYTHON_VERSION=$1
if [ -z "$PYTHON_VERSION" ]; then
    echo "用法: $0 <python版本>"
    echo "示例: $0 3.8"
    exit 1
fi

# 檢查依賴
if ! command -v python3 &> /dev/null; then
    echo "錯(cuò)誤: 需要安裝Python 3"
    exit 1
fi

# 1. 安全下載所有兼容指定Python版本的包
echo "步驟1: 安全下載所有兼容Python $PYTHON_VERSION 的包"
python3 scripts/safe_download.py --python-version "$PYTHON_VERSION" --output-dir packages

# 2. 生成simple index
echo "步驟2: 生成PEP 503兼容的simple index"
python3 scripts/generate_simple_index.py packages web/simple

# 3. 創(chuàng)建README
cat > web/README.md << EOF
# 離線PyPI鏡像

此目錄包含Python $PYTHON_VERSION 的完整PyPI鏡像。

## 使用方法

1. 啟動(dòng)服務(wù):
   docker-compose up -d

2. 客戶端使用:
   pip install 包名 -i http://your-server:8080/simple
EOF

echo "離線PyPI鏡像準(zhǔn)備完成!"
echo "現(xiàn)在可以將整個(gè)pypi-offline目錄復(fù)制到內(nèi)網(wǎng)環(huán)境"
echo "在內(nèi)網(wǎng)環(huán)境中執(zhí)行: docker-compose up -d 啟動(dòng)服務(wù)"

完整操作流程

1. 在可聯(lián)網(wǎng)的環(huán)境中準(zhǔn)備鏡像

# 克隆項(xiàng)目
git clone https://github.com/your-repo/pypi-offline.git
cd pypi-offline

# 賦予腳本執(zhí)行權(quán)限
chmod +x scripts/*.py prepare_offline_mirror.sh

# 準(zhǔn)備Python 3.8的完整鏡像 (這將需要較長(zhǎng)時(shí)間)
./prepare_offline_mirror.sh 3.8

# 準(zhǔn)備Python 3.9的完整鏡像 (可選)
# ./prepare_offline_mirror.sh 3.9

2. 驗(yàn)證下載結(jié)果

# 檢查包數(shù)量
ls packages | wc -l

# 檢查simple index
ls web/simple | wc -l

3. 將整個(gè)目錄復(fù)制到內(nèi)網(wǎng)環(huán)境

# 壓縮整個(gè)目錄
tar -czvf pypi-offline-3.8.tar.gz pypi-offline

# 將壓縮文件傳輸?shù)絻?nèi)網(wǎng)環(huán)境

4. 在內(nèi)網(wǎng)環(huán)境中部署

# 解壓
tar -xzvf pypi-offline-3.8.tar.gz
cd pypi-offline

# 啟動(dòng)服務(wù)
docker-compose up -d

5. 內(nèi)網(wǎng)客戶端使用

# 臨時(shí)使用
pip install numpy -i http://pypi-offline-server:8080/simple

# 永久配置
pip config set global.index-url http://pypi-offline-server:8080/simple