sais-pyobs 0.1.4

A high-performance OBS transfer tool for AI datasets.

sais_pyobs

一个为AI数据集设计的高性能、支持断点续传的华为云OBS流式上传工具。

该库的核心功能是实现 从一个数据流（如HTTP URL）到OBS的直接传输，整个过程无需将文件完整保存在本地磁盘，非常适合大文件的迁移和备份。

核心特性

• 流式上传 (Stream Upload): 直接将网络响应流、文件流等数据以流的形式上传至OBS，内存占用极低。
• 断点续传 (Breakpoint Resume): 如果上传过程意外中断，再次运行时能自动从上次的断点处继续上传，无需重头再来。
• 并发分片 (Concurrent Upload): 内部使用多线程并发上传数据分片，大幅提升大文件上传速度。
• 配置灵活 (Flexible Configuration): 支持通过构造函数参数或环境变量来设置OBS的访问凭证和配置。
• 自动修复 (Monkey Patch): 内置了对 huaweicloud-sdk-python-obs 的猴子补丁，修复了在某些环境下因SSL参数导致的连接问题。

安装

首先，请确保你安装了 Python 3.10 或更高版本。

1. 克隆本项目到本地：

   git clone <your-repo-url>
   cd sais_pyobs
   

2. 使用 pip 从本地源码安装：

   pip install .
   

   该命令会自动处理并安装所有在 pyproject.toml 中声明的依赖项，包括：

   • requests
   • huaweicloud-sdk-python-obs
   • tqdm

如何使用

以下是一个典型的用例：将一个网络上的大文件直接传输到你的OBS桶中。

import requests
from pyobs import StreamUploader

# --- 1. 初始化上传器 ---
# 建议使用环境变量配置，更安全
# os.environ["OBS_AK"] = "YOUR_ACCESS_KEY"
# os.environ["OBS_SK"] = "YOUR_SECRET_KEY"
# os.environ["OBS_SERVER"] = "obs.cn-east-3.myhuaweicloud.com"
# os.environ["OBS_BUCKET"] = "your-bucket-name"
# uploader = StreamUploader()

# 或者直接通过参数传入
uploader = StreamUploader(
    ak="YOUR_ACCESS_KEY",
    sk="YOUR_SECRET_KEY",
    server="obs.cn-east-3.myhuaweicloud.com",
    bucket_name="your-bucket-name"
)

# --- 2. 初始化上传任务，获取上传上下文 ---
# 这一步会检查OBS上是否有未完成的任务，并返回断点位置
object_key = "my-remote-folder/large_file.zip"
context = uploader.init_upload(object_key=object_key)

print(f"-> SDK 建议从 {context.offset} 字节处开始下载")

# --- 3. 根据断点位置发起流式下载 ---
# 使用 Range 请求头，告诉服务器只发送我们需要的数据
headers = {
    "Range": f"bytes={context.offset}-"
}
source_url = "https://example.com/path/to/your/large_file.zip"
response = requests.get(source_url, headers=headers, stream=True)
response.raise_for_status()  # 确保请求成功

# 获取剩余文件总大小，用于显示进度条
total_size = int(response.headers.get('content-length', 0))

# --- 4. 将数据流送入上传器 ---
# SDK内部会自动处理分片、并发和重试
try:
    uploader.upload_stream(
        context=context,
        stream_iterator=response.iter_content(chunk_size=10 * 1024 * 1024), # 10MB的块大小
        total_size=total_size
    )
    print("✅ 文件上传成功！")
except Exception as e:
    print(f"❌ 上传过程中断: {e}")
    print("程序已中断。下次运行时，将自动从断点续传。")

配置方式

你可以通过两种方式来配置 StreamUploader：

1. 构造函数参数（推荐用于测试）

在创建 StreamUploader 实例时直接传入所有必要信息。

uploader = StreamUploader(
    ak="YOUR_ACCESS_KEY",
    sk="YOUR_SECRET_KEY",
    server="obs.cn-east-3.myhuaweicloud.com",
    bucket_name="your-bucket-name"
)

2. 环境变量（推荐用于生产）

将凭证信息存储在环境变量中，代码无需硬编码任何敏感信息。

export OBS_AK="YOUR_ACCESS_KEY"
export OBS_SK="YOUR_SECRET_KEY"
export OBS_SERVER="obs.cn-east-3.myhuaweicloud.com"
export OBS_BUCKET="your-bucket-name"

然后在代码中无参初始化即可：

uploader = StreamUploader()

工作原理解析

本工具的工作流程分为两个核心步骤，以实现高效的断点续传：

1. init_upload(object_key):

   • 目的: 准备上传并检查断点。
   • 过程:
     1. 向OBS查询指定 object_key 是否存在一个尚未完成的“分段上传”（Multipart Upload）任务。
     2. 如果存在，则列出所有已成功上传的分片（Part），并计算这些分片的总字节数。这个总和就是 offset。
     3. 如果没有找到任务，就为这个 object_key 初始化一个新的分段上传任务，此时 offset 为 0。
     4. 返回一个包含任务ID、offset 等信息的 UploadContext 对象。

2. upload_stream(context, stream_iterator, ...):

   • 目的: 接收数据流并执行上传。
   • 过程:
     1. 用户根据上一步返回的 context.offset 从数据源（如URL）请求一个从 offset 开始的数据流。
     2. upload_stream 接收这个流，并开始读取数据。
     3. 当读取的数据达到预设的分片大小（默认为20MB）时，就将这个数据块提交到一个线程池中进行异步上传。
     4. 所有数据流读取完毕且所有分片都上传成功后，发送一个“完成”请求给OBS，OBS会将所有分片按顺序合并成一个完整的文件。
     5. 如果在中途失败，已上传的分片会保留在OBS上，等待下一次 init_upload 时被识别，从而实现续传。