如何使用Python从Nexus仓库下载文件并实现自动化处理？

当然可以用Python实现从Nexus仓库下载文件，核心是利用Nexus的REST API结合HTTP请求库（比如requests）完成操作。下面是具体的实现步骤和代码示例：

1. 前置准备 #

先安装requests库用于发送HTTP请求：

pip install requests

2. 核心实现步骤 #

2.1 认证处理 #

Nexus仓库大多需要身份验证，通常采用Basic Auth，准备好你的Nexus用户名和密码即可。

2.2 获取目标文件列表 #

通过Nexus的API或目录页面获取仓库内的文件路径：

• 若为Maven仓库，可访问路径如http://你的Nexus地址/repository/{仓库名}/maven2/{分组路径}/{工件ID}/{版本号}/
• 若为Raw仓库，直接访问对应目录路径即可获取文件列表

2.3 下载文件 #

遍历文件列表，发送GET请求下载每个文件，保存到本地指定目录。

2.4 处理文件内容 #

读取下载后的文件，根据需求处理内容，再生成输出文件。

3. 完整代码示例 #

import os
import requests
from requests.auth import HTTPBasicAuth
from bs4 import BeautifulSoup

# Nexus配置信息
NEXUS_URL = "http://your-nexus-server-url"
REPO_NAME = "your-repo-name"
USERNAME = "your-nexus-username"
PASSWORD = "your-nexus-password"
# 目标文件在Nexus中的路径
TARGET_PATH = "com/example/demo/1.0.0/"
# 本地保存下载文件的目录
LOCAL_DOWNLOAD_DIR = "./downloaded_files"
# 处理后输出文件的目录
OUTPUT_DIR = "./output_files"

# 创建本地目录
os.makedirs(LOCAL_DOWNLOAD_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)

def get_nexus_file_list():
    """获取Nexus目标路径下的文件列表"""
    url = f"{NEXUS_URL}/repository/{REPO_NAME}/{TARGET_PATH}"
    response = requests.get(url, auth=HTTPBasicAuth(USERNAME, PASSWORD))
    response.raise_for_status()  # 抛出请求失败的异常
    
    # 解析Nexus目录页面提取文件链接
    soup = BeautifulSoup(response.text, 'html.parser')
    file_links = []
    for link in soup.find_all('a'):
        href = link.get('href')
        # 过滤目录链接和上级目录链接
        if href and not href.endswith('/') and href != '../':
            file_links.append(href)
    return file_links

def download_file(file_name):
    """下载单个文件到本地"""
    file_url = f"{NEXUS_URL}/repository/{REPO_NAME}/{TARGET_PATH}{file_name}"
    local_file_path = os.path.join(LOCAL_DOWNLOAD_DIR, file_name)
    
    with requests.get(file_url, auth=HTTPBasicAuth(USERNAME, PASSWORD), stream=True) as r:
        r.raise_for_status()
        with open(local_file_path, 'wb') as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
    print(f"已下载: {file_name}")
    return local_file_path

def process_file(input_file_path):
    """处理文件内容并生成输出文件"""
    file_name = os.path.basename(input_file_path)
    output_file_path = os.path.join(OUTPUT_DIR, f"processed_{file_name}")
    
    # 替换为你的实际处理逻辑，这里示例为在内容开头添加标识
    with open(input_file_path, 'r', encoding='utf-8') as f:
        content = f.read()
    processed_content = f"[PROCESSED]\n{content}"
    
    with open(output_file_path, 'w', encoding='utf-8') as f:
        f.write(processed_content)
    print(f"已生成处理文件: {output_file_path}")

if __name__ == "__main__":
    try:
        file_list = get_nexus_file_list()
        for file_name in file_list:
            local_path = download_file(file_name)
            process_file(local_path)
        print("所有文件下载和处理完成！")
    except Exception as e:
        print(f"执行出错: {str(e)}")

注意事项 #

• 若使用Nexus 3，可调用官方APIhttp://你的Nexus地址/service/rest/v1/components?repository={仓库名}获取组件信息，直接解析JSON提取下载链接，无需解析HTML
• 确保你的Nexus账号拥有对应仓库的读取权限
• 处理二进制文件时，需调整文件读写的编码模式（如使用rb/wb）

内容的提问来源于stack exchange，提问作者Vignesh G