crazy-tools 1.0.1
Crazy Tools SDK
Requires Python: >3.10

# python-sdk

文件上传 python 客户端

2025-02-19


# 从源码构建

编译前安装依赖

```shell
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pipreqs build

```

```shell
git clone https://dev.aminer.cn/crazy-team/python-sdk
cd python-sdk

make build

pip install dist/[crazy_tools-1.0.0-py3-none-any.whl](dist/crazy_tools-1.0.0-py3-none-any.whl)
```


# 一键上传

```python
import asyncio
from datetime import datetime
from crazy_tools.uploader import SimpleClient, ISimpleClient, UploadOptions, UploadCompletedEvent, UploadFailedEvent
from crazy_tools.uploader.facade.helper import Helper
from crazy_tools.uploader.facade.config import ConfigBuilder
from tqdm import tqdm


# ############################ 配置参数
sem = asyncio.Semaphore(10)

now_timestamp = datetime.now().timestamp()
output_path = f"logs/{now_timestamp}"
success_path = f"{output_path}_success.log"
failure_path = f"{output_path}_failure.log"

app_key = "your app key"
app_secret = "your app secret"
api_endpoint = "http://open.annot.xhanz.cn"

scan_path = "/Users/xlh/Downloads/video_clip_3"
scan_ext = "*.jsonl"
upload_group = "test_group_25_3"

##### ############################

client: ISimpleClient = SimpleClient(
    ConfigBuilder()
    .with_app_key(app_key)
    .with_app_secret(app_secret)
    .with_url(api_endpoint)
)


## 成功回调事件
@client.event_bus().on(UploadCompletedEvent)
async def on_completed(ev: UploadCompletedEvent):
    with open(success_path, "a") as f:
        f.write(ev.model_dump_json() + '\n')


## 失败回调事件
@client.event_bus().on(UploadFailedEvent)
async def on_failure(ev: UploadFailedEvent):
    with open(failure_path, "a") as f:
        f.write(str(ev.model_dump()) + '\n')


async def process(file_path: str):
    async with sem:
        await client.upload(file_path, opts=UploadOptions(
            group=upload_group,
        ))


async def main():
    files = list(Helper.scan_local_files(scan_path, scan_ext))  # 转换为列表以获取总数
    pbar = tqdm(total=len(files), desc="Uploading files")    # 初始化进度条

    tasks = []
    for file in files:
        task = asyncio.create_task(process(file))
        task.add_done_callback(lambda _: pbar.update(1))
        tasks.append(task)

    await asyncio.gather(*tasks)


if __name__ == '__main__':
    asyncio.run(main())
```


## 二次开发

```python
import asyncio
from datetime import datetime, timedelta
from crazy_tools.uploader import UploadClient, IUploadClient, UploadCompletedEvent, UploadFailedEvent, UploadPartCompletedEvent, UploadPartFailedEvent
from crazy_tools.uploader.facade import UploadOptions
from crazy_tools.common import EventBus, IEventBus

# 获取事件总线实例
bus: IEventBus = EventBus.get_or_init()

# 订阅上传成功事件


@bus.on(UploadCompletedEvent)
async def on_success(ev: UploadCompletedEvent):
    """
    处理上传成功事件。
    :param ev: 上传成功事件，包含上传任务的详细信息。
    """
    print("Upload succeeded:", ev.model_dump())

# 订阅上传失败事件


@bus.on(UploadFailedEvent)
async def on_failed(ev: UploadFailedEvent):
    """
    处理上传失败事件。
    :param ev: 上传失败事件，包含失败原因和相关信息。
    """
    print("Upload failed:", ev.model_dump())

# 订阅分片上传完成事件


@bus.on(UploadPartCompletedEvent)
async def on_part_success(ev: UploadPartCompletedEvent):
    """
    处理分片上传完成事件。
    :param ev: 分片完成事件，包含分片ID、大小、进度等信息。
    """
    print("Part completed:", ev.model_dump())

# 订阅分片上传失败事件


@bus.on(UploadPartFailedEvent)
async def on_part_failed(ev: UploadPartFailedEvent):
    """
    处理分片上传失败事件。
    :param ev: 分片失败事件，包含失败原因和分片信息。
    """
    print("Part failed:", ev.model_dump())


async def main():
    """
    主函数：演示如何使用 UploadClient 进行分片上传，并设置上传选项。
    """
    try:
        # 初始化上传客户端，传入配置文件路径
        client: IUploadClient = UploadClient(
            "/Users/xlh/Downloads/config_1.json")

        # 设置上传文件的过期时间（当前时间 + 30天）
        expire_at = (datetime.now() + timedelta(days=30)).timestamp()

        # 设置上传选项，指定过期时间
        opts = UploadOptions(expire_at=int(expire_at))

        # 执行分片上传，传入文件路径、组名和上传选项
        await client.multipart_upload(
            "/Users/xlh/Downloads/PopClip_2024_12_TNT.dmg",  # 文件路径
            "lzf_test",  # 组名
            opts=opts  # 上传选项
        )
    except Exception as e:
        print(f"Upload error: {e}")

if __name__ == "__main__":
    asyncio.run(main())

```