bdd-file-client 0.5.1

BDD 平台文件服务客户端

BDD File Client

BDD 平台文件服务客户端

安装方式

$ pip install bdd-file-client

快速开始

配置 BddFileClient

from bdd_file import BddFileClient, UploadMode

client = BddFileClient(
    profile="dev",            # 可选 dev、beta、prod，默认为 dev
    default_user_id="114514", # 默认的用户 ID，可在调用时覆盖
    default_biz="chat",       # 默认的业务类型，可在调用时覆盖
    default_mode=UploadMode.AUTO,  # 默认的上传模式，可在调用时覆盖
)

• profile 用于指定后端环境，会自动映射到对应的服务地址
• default_user_id、default_biz、default_mode 均为可选，若在调用时未显式提供则使用此处的默认值

上传文件

# 1. 通过文件路径上传（简单上传）
result = client.upload(
    path="example.png",
    mode=UploadMode.SIMPLE,   # 默认 auto，可选 simple、chunked
    biz="chat",
    biz_params={"chat_id": "191810"},
)
print("上传成功，file_id =", result.file_id)

# 2. 通过文件流上传
with open("example.png", "rb") as f:
    result = client.upload(
        stream=f,
        filename="example.png",           # 当使用 stream 时，必须显式指定文件名
        biz="chat",
        biz_params={"chat_id": "191810"},
    )

注意

• path 与 stream 只能二选一
• 分片上传（UploadMode.CHUNKED）暂未实现，设置后会抛出 BddFileError

下载文件

# 1. 保存到本地文件
client.download(
    file_id=result.file_id,
    path="downloaded.png",
)

# 2. 写入到自定义流
from io import BytesIO

buffer = BytesIO()
client.download(file_id=result.file_id, stream=buffer)
print(buffer.getvalue())

# 3. 获取下载链接（支持设置过期时间）
url = client.get_download_url(
    file_id=result.file_id,
    expires_seconds=600,  # 链接有效期，默认 10 分钟
)
print("下载链接：", url)

文件管理

# 1. 删除文件
client.delete(file_id=result.file_id)

# 2. 获取聊天文件列表
files = client.list_chat_files(chat_id="191810")
for file in files:
    print(f"文件ID: {file.id}, 文件名: {file.filename}")

同步聊天文件

# 1. 将聊天文件同步到本地
results = client.sync_chat_files_to_local(
    chat_id="191810",
    local_dir="./downloads",  # 本地目录路径
)
for file_id, result in results.items():
    if result.is_success:
        print(f"文件 {file_id} 同步成功：{result.local_path}")
    else:
        print(f"文件 {file_id} 同步失败：{result.error_message}")

# 2. 将本地文件同步到聊天
results = client.sync_chat_files_to_remote(
    chat_id="191810",
    local_dir="./uploads",  # 本地目录路径
)
for file_path, result in results.items():
    if result.status == "success":
        print(f"文件 {file_path} 上传成功，file_id = {result.file_id}")
    elif result.status == "skipped":
        print(f"文件 {file_path} 已存在，跳过")
    else:
        print(f"文件 {file_path} 上传失败：{result.error_message}")

错误处理

所有业务及网络错误都会抛出 bdd_file.BddFileError，可按需捕获：

from bdd_file import BddFileClient, BddFileError

client = BddFileClient(default_user_id="114514")
try:
    result = client.upload(path="not_exists.txt", biz="chat")
except BddFileError as e:
    print("操作失败：", e)

注意事项

1. 所有需要 user_id 的方法都支持在调用时覆盖默认值
2. 同步文件到远程时，会根据文件名判断是否需要同步：
   • 本地和远程都存在，跳过
   • 本地存在，远程不存在，上传
   • 本地不存在，远程存在，跳过
3. 文件列表接口支持分页，会自动获取所有数据
4. 同步文件操作支持并发处理，提高效率