Metadata-Version: 2.1
Name: cdt_data_uploader
Version: 1.1.0
Summary: Python SDK for uploading CDT datasets (VLA, TXT, MV-MF Fusion)
Home-page: https://github.com/yourorg/cdt-data-uploader
Author: CDT Team
Author-email: cdt@example.com
License: UNKNOWN
Project-URL: Bug Reports, https://github.com/yourorg/cdt-data-uploader/issues
Project-URL: Source, https://github.com/yourorg/cdt-data-uploader
Keywords: cdt data uploader sdk machine learning ai vla txt fusion stereo pointcloud
Platform: UNKNOWN
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Requires-Python: >=3.6
Description-Content-Type: text/markdown
Provides-Extra: dev
License-File: LICENSE

# CDT Data Uploader SDK

[![Python 3.6+](https://img.shields.io/badge/python-3.6+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

一个用于上传CDT数据集的Python SDK，支持VLA(Visual Language Action)、TXT(文本数据集)、MV-MF Fusion(多目多帧融合)等多种数据格式，提供并发上传、自动重试和失败恢复功能。

## 特性

- ✅ **并发上传** - 支持多线程并发上传,提高效率
- ✅ **自动重试** - 上传失败时自动重试,支持可配置的重试次数和延迟
- ✅ **失败恢复** - 记录失败的上传任务,支持后续重试
- ✅ **灵活配置** - 支持多种配置参数,满足不同场景需求
- ✅ **完整日志** - 提供详细的日志输出,便于调试和监控
- ✅ **类型提示** - 完整的类型注解,提高代码可读性
- ✅ **Python 3.6+兼容** - 使用attrs库兼容Python 3.6及以上版本

## 安装

```bash
pip install cdt_data_uploader
```

## 依赖

- Python 3.6+
- `requests>=2.20.0` - HTTP请求库
- `tos>=2.5.0` - 腾讯云对象存储SDK
- `attrs>=21.0.0` - 数据类库(兼容Python 3.6)

**兼容性:** Windows/Linux/macOS

## 模块说明

SDK包含以下数据上传模块：

- **vla_uploader** - VLA (Visual Language Action) 数据上传
- **txt_uploader** - TXT 文本数据集(JSONL格式)上传
- **mv_mf_fusion_uploader** - 多目多帧融合数据集(立体视觉/点云/标定)上传

## 快速开始

### 1. 导入SDK

```python
from vla_uploader import VLAUploader
```

### 2. 初始化上传器

```python
uploader = VLAUploader(
    api_key="your_api_key",              # 从系统管理员获取
    dataset_id="your_dataset_id",         # 数据集ID
    api_base="https://nebula.core-dt.com/api/v1",  # API地址
    workers=4                             # 并发线程数
)
```

### 3. 上传数据

#### 上传目录

```python
# 上传整个目录
result = uploader.upload_directory("./dataset")

# 查看结果
print(f"总数: {result.total_rows}")
print(f"成功: {result.success_count}")
print(f"失败: {result.failure_count}")
print(f"成功率: {result.success_rate:.2f}%")
```

#### 上传ZIP文件

```python
# 上传ZIP压缩包
result = uploader.upload_zip(
    zip_path="./dataset.zip",
    workers=4
)
```

#### 重试失败的上传

```python
# SDK自动记录失败的上传，可以重试
result = uploader.retry_failed("./.vla_upload_work/run-xxx/failed_rows.json")
```

## 多模块快速开始

SDK提供三个独立模块,根据数据类型选择:

### VLA 数据上传

```python
from vla_uploader import VLAUploader

uploader = VLAUploader(
    api_key="your_api_key",
    dataset_id="your_dataset_id",
    workers=4
)
result = uploader.upload_directory("./dataset")
```

**数据结构:**
```
dataset/
├── episode_0001/
│   ├── manifest.json
│   ├── data/video/front.mp4
│   └── data/actions/actions.jsonl
├── episode_0002/
```

### TXT 文本数据集上传

```python
from txt_uploader import TXTUploader

uploader = TXTUploader(
    api_key="your_api_key",
    dataset_id="your_dataset_id",
    workers=4,
    row_key_field="id"  # 默认即 "id"
)
result = uploader.upload_zip("./text_dataset.zip")
```

**数据结构:**
```
text_dataset.zip
├── manifest.json
└── data/
    └── rows.jsonl   # JSON Lines,每行一条记录
```

**rows.jsonl 示例:**
```json
{"id":"r1","text":"第一条记录","metadata":{"src":"test"}}
{"id":"r2","text":"第二条记录"}
```

### MV-MF Fusion 多目多帧融合数据上传

```python
from mv_mf_fusion_uploader import MVMFFusionUploader

uploader = MVMFFusionUploader(
    api_key="your_api_key",
    dataset_id="your_dataset_id",
    workers=3,
    row_key_field="id"
)
result = uploader.upload_zip("./stereo_dataset.zip")
```

**数据结构:**
```
stereo_dataset.zip
├── manifest.json
└── data/
    ├── images/
    │   ├── left/     # 左目图像
    │   └── right/    # 右目图像
    ├── pointcloud/   # Potree点云
    ├── calib/        # 标定参数
    ├── pre_annotation/
    └── annotation/
```

**资源类型自动识别:**
- `data/images/left/` → image_left
- `data/images/right/` → image_right
- `data/pointcloud/` → pointcloud
- `data/calib/` → calib
- `data/pre_annotation/` → pre_annotation
- `data/annotation/` → annotation

## 使用场景

### 场景1: 首次上传VLA数据集

```python
from vla_uploader import VLAUploader

# 1. 初始化
uploader = VLAUploader(
    api_key="your_api_key",
    dataset_id="your_dataset_id"
)

# 2. 准备数据目录结构
# dataset/
# ├── episode_0001/
# │   ├── data/
# │   │   ├── video/front.mp4
# │   │   ├── actions/actions.jsonl
# │   │   └── depth/frame_000001.png
# └── episode_0002/

# 3. 上传
result = uploader.upload_directory("./dataset")

# 4. 检查结果
if result.success_rate == 100.0:
    print("✅ 所有数据上传成功!")
else:
    print(f"⚠️ 部分失败: {result.failure_count} 行")
    for failed in result.failed_rows:
        print(f"  失败: {failed.row_name} - {failed.error}")
```

### 场景2: 上传ZIP压缩包并记录源路径

```python
# 适用于数据集已经打包成ZIP的情况
result = uploader.upload_zip(
    zip_path="./my_dataset.zip",
    source_path="D:/raw/my_dataset.zip",           # 原始路径
    cloud_source_path="tos://bucket/raw/my_dataset.zip"  # 云端路径
)
```

### 场景3: 并发上传大型数据集

```python
# 增加并发数以提高上传速度
uploader = VLAUploader(
    api_key="your_api_key",
    dataset_id="your_dataset_id",
    workers=8  # 8个并发线程
)

result = uploader.upload_directory("./large_dataset")
```

### 场景4: 重试失败的上传

```python
import glob

# 查找最新的失败记录
failed_files = glob.glob("./.vla_upload_work/run-*/failed_rows.json")
if failed_files:
    latest_failed = max(failed_files)
    print(f"重试: {latest_failed}")

    result = uploader.retry_failed(latest_failed)
    print(f"重试结果: {result.success_count}/{result.total_rows}")
```

### 场景5: 批量处理多个数据集

```python
datasets = [
    "./dataset1",
    "./dataset2.zip",
    "./dataset3"
]

for dataset_path in datasets:
    print(f"\n处理: {dataset_path}")

    try:
        if dataset_path.endswith(".zip"):
            result = uploader.upload_zip(dataset_path)
        else:
            result = uploader.upload_directory(dataset_path)

        if result.success_rate == 100.0:
            print(f"✅ 成功: {result.success_count} 行")
        else:
            print(f"⚠️ 部分失败: {result.failure_count} 行")

    except Exception as e:
        print(f"❌ 失败: {e}")
```

## API参考

### 主要方法

#### `upload_directory(data_path, workers=None, ...)`

上传目录中的VLA数据

**参数:**

- `data_path` - 数据目录路径
- `workers` - 并发线程数（默认3）
- `source_path` - 原始路径（可选）
- `cloud_source_path` - 云端路径（可选）

**返回:** `BatchUploadResult` 对象

#### `upload_zip(zip_path, workers=None, ...)`

上传ZIP压缩包

**参数:** 同上

#### `retry_failed(failed_json_path)`

重试失败的上传

**参数:**

- `failed_json_path` - 失败记录文件路径

### 返回结果

#### `BatchUploadResult`

```python
result.total_rows       # 总行数
result.success_count    # 成功数量
result.failure_count    # 失败数量
result.success_rate     # 成功率(%)
result.succeeded_rows   # 成功的行列表
result.failed_rows      # 失败的行列表
```

#### `UploadResult`

```python
row.success        # 是否成功
row.row_name       # 行名称
row.datarow_id     # 数据行ID
row.error          # 错误信息（如果失败）
```

## 数据格式要求

SDK要求VLA数据集遵循以下目录结构:

```
dataset/
├── episode_0001/
│   ├── manifest.json
│   ├── data/
│   │   ├── video/
│   │   │   └── front.mp4
│   │   ├── actions/
│   │   │   └── actions.jsonl
│   │   ├── depth/
│   │   │   └── frame_000001.png
│   │   └── camera_params.json
├── episode_0002/
│   └── ...
```

**必需项(全部不可缺失):**
- `manifest.json` - 数据清单(含 `id` 字段作为 row_key)
- `data/video/` - 视频资源目录(每相机一个视频文件)
- `data/actions/` - 动作片段 JSON Lines 目录

**可选:**
- `data/depth/` - 深度图序列(按相机分组)
- `data/camera_params.json` - 相机参数

**支持的格式:**

- 视频: `.mp4`, `.mov`, `.avi`, `.mkv`, `.webm`
- 图像: `.jpg`, `.jpeg`, `.png`, `.bmp`, `.webp`, `.tif`, `.tiff`
- 文本: `.json`, `.jsonl`

### TXT 文本数据集格式

```
text_dataset/
├── manifest.json               # 数据清单
└── data/
    └── rows.jsonl              # JSON Lines,每行一条记录
```

**必需项(全部不可缺失):**
- `manifest.json` - 数据集清单
- `data/rows.jsonl` - JSONL格式记录文件

**记录格式:**
每行是一个独立的JSON对象,`id` 字段作为 row_key(可通过 `row_key_field` 配置):

```json
{"id":"r1","text":"记录内容","metadata":{"key":"value"}}
```

### MV-MF Fusion 融合数据集格式

```
stereo_dataset/
├── manifest.json
└── data/
    ├── images/
    │   ├── left/               # 左目图像 (.jpg/.png)
    │   └── right/              # 右目图像 (.jpg/.png)
    ├── pointcloud/             # Potree single-octree 点云
    │   ├── metadata.json
    │   ├── hierarchy.bin
    │   └── octree.bin
    ├── calib/                  # 标定参数 (.json)
    ├── pre_annotation/         # 预标注 (.json)
    └── annotation/             # 标注 (.json)
```

**必需项(全部不可缺失):**
- `manifest.json` - 数据集清单(含 `id` 字段作为 row_key)
- `data/images/` - 图像目录(含 left/ 和 right/)
- `data/pointcloud/` - 点云目录
- `data/calib/` - 标定参数目录
- `data/pre_annotation/` - 预标注目录
- `data/annotation/` - 标注目录

**单zip = 单DataRow:** 每个zip视为一个完整的数据行,所有文件上传到同一TOS前缀下。

## 配置参数

SDK支持以下配置:

```python
uploader = VLAUploader(
    api_key="...",                    # 必需：API密钥
    dataset_id="...",                  # 必需：数据集ID

    # 可选配置
    api_base="https://...",           # API地址
    workers=4,                         # 并发线程数(默认3)

    # 重试配置
    max_upload_attempts=5,            # 上传重试次数(默认5)
    max_sync_attempts=3,              # 同步重试次数(默认3)

    # 其他配置
    dedupe_by_filename=True            # 按文件名去重(默认True)
)
```

### media_type 说明

`media_type` 用于标识数据类型,后端据此分类处理。**每个模块的 `media_type` 由后台写死,不允许用户修改**:

| 模块 | 固定 media_type | 说明 |
|---|---|---|
| vla_uploader | `video` | VLA视频动作数据 |
| txt_uploader | `text` | JSONL文本数据 |
| mv_mf_fusion_uploader | `point_cloud` | 多目多帧融合数据(点云) |

> ⚠️ 注意:即使向构造函数传入 `media_type` 参数,也会被忽略。各模块的 `media_type` 完全由模块自身决定,以确保后端数据分类正确。

## 故障排查

### 常见问题

**Q: 上传失败怎么办？**

```python
# SDK自动保存失败记录到 .vla_upload_work/run-*/failed_rows.json
# 可以重试:
result = uploader.retry_failed("./.vla_upload_work/run-xxx/failed_rows.json")
```

**Q: 如何查看详细日志？**

```python
import logging
logging.basicConfig(level=logging.DEBUG)
```

**Q: 如何调整并发数？**

```python
# 根据网络带宽和服务器性能调整
uploader = VLAUploader(..., workers=8)  # 增加
uploader = VLAUploader(..., workers=1)  # 减少
```

**Q: 如何获取API密钥？**
联系系统管理员获取API密钥和数据集ID。

### 错误处理

```python
from vla_uploader import VLAUploader, UploadError, CredentialError

try:
    result = uploader.upload_directory("./dataset")
except CredentialError as e:
    print(f"凭证错误: {e}")
    print("请检查API密钥是否正确")
except UploadError as e:
    print(f"上传错误: {e}")
    print("请检查网络连接和数据格式")
```

## 更多信息

- **项目地址**: https://github.com/yourorg/cdt-data-uploader
- **问题反馈**: https://github.com/yourorg/cdt-data-uploader/issues
- **许可证**: MIT License

## 版本历史

- **v1.1.0** (2026-08-13): 新增TXT和Fusion模块 + 结构校验加强
  - 新增 txt_uploader 模块:支持JSONL文本数据集上传
  - 新增 mv_mf_fusion_uploader 模块:支持多目多帧融合数据(立体视觉/点云/标定)上传
  - 三个模块共享一致的API设计(并发上传/自动重试/失败恢复)
  - 加强压缩包结构校验:缺失必需文件/目录时,明确提示缺失项
  - `media_type` 由各模块后台写死,不允许用户自定义
  - 更新测试套件覆盖三个模块


