huace-aigc-oss-proxy-client 0.1.5

华策 AIGC OSS Proxy Client - 统一 RustFS / 阿里云 OSS / 七牛云，支持下载降级与上传路由

AIGC OSS Proxy Python SDK

面向华策 AIGC 插件的统一对象存储 SDK：内网 RustFS（S3 兼容）、阿里云 OSS、七牛云，支持读路径按优先级降级、写路径按内外网指定后端。

详细设计见 DESIGN.md。

Python 版本：>= 3.10（与 pyproject.toml 中 requires-python 一致；已声明兼容 3.10 / 3.11 / 3.12）。

安装

# 全量后端（RustFS + 阿里云 + 七牛）
pip install huace-aigc-oss-proxy-client[all]

# 仅阿里云 OSS
pip install huace-aigc-oss-proxy-client[aliyun]

# 仅 RustFS / S3 兼容内网（boto3）
pip install huace-aigc-oss-proxy-client[rustfs]

# 仅七牛云
pip install huace-aigc-oss-proxy-client[qiniu]

环境变量

OssClient.from_env() 从进程环境读取配置，变量名统一加前缀 HUACE_AIGC_（例如 HUACE_AIGC_OSS_DOWNLOAD_PRIORITY）。仍兼容旧版无前缀的 OSS_*。

可复制 env.example 为 .env 后填入真实值。

路由与策略（常用）

变量	说明	示例
HUACE_AIGC_OSS_FILE_PREFIX	对象键统一前缀（全客户端共用，唯一前缀变量）	tmp
HUACE_AIGC_OSS_DOWNLOAD_PRIORITY	下载尝试顺序，逗号分隔	rustfs,aliyun
HUACE_AIGC_OSS_UPLOAD_INTRANET_BACKEND	内网上传后端	rustfs
HUACE_AIGC_OSS_UPLOAD_EXTRANET_BACKEND	外网上传后端	aliyun
HUACE_AIGC_OSS_FALLBACK_ENABLED	第一优先级无法访问时是否改用第二优先级	true
HUACE_AIGC_OSS_CIRCUIT_BREAKER_ENABLED	是否启用熔断	false
HUACE_AIGC_OSS_MAX_OBJECT_SIZE_MB	单对象大小上限（MB）	5120

RustFS / S3 兼容内网

变量	说明
HUACE_AIGC_OSS_RUSTFS_ENABLED	是否启用
HUACE_AIGC_OSS_RUSTFS_ENDPOINTS	内网多地址，逗号分隔、按优先级（优先于单 ENDPOINT）
HUACE_AIGC_OSS_RUSTFS_ENDPOINT	单 S3 API 地址（未配 ENDPOINTS 时使用）
HUACE_AIGC_OSS_RUSTFS_ACCESS_KEY	Access Key
HUACE_AIGC_OSS_RUSTFS_SECRET_KEY	Secret Key
HUACE_AIGC_OSS_RUSTFS_BUCKET	桶名
HUACE_AIGC_OSS_RUSTFS_REGION	区域，默认 us-east-1
HUACE_AIGC_OSS_RUSTFS_USE_SSL	是否 HTTPS

阿里云 OSS

变量	说明
HUACE_AIGC_OSS_ALIYUN_ENABLED	是否启用
HUACE_AIGC_OSS_ALIYUN_ENDPOINT	Endpoint（兼容 OSS_ENDPOINT）
HUACE_AIGC_OSS_ALIYUN_ACCESS_KEY_ID	AK（兼容 OSS_ACCESS_KEY_ID）
HUACE_AIGC_OSS_ALIYUN_ACCESS_KEY_SECRET	SK（兼容 OSS_ACCESS_KEY_SECRET）
HUACE_AIGC_OSS_ALIYUN_BUCKET	桶名（兼容 OSS_BUCKET_NAME）

七牛云（可选）

变量	说明
HUACE_AIGC_OSS_QINIU_ENABLED	是否启用
HUACE_AIGC_OSS_QINIU_ACCESS_KEY	Access Key
HUACE_AIGC_OSS_QINIU_SECRET_KEY	Secret Key
HUACE_AIGC_OSS_QINIU_BUCKET	空间名
HUACE_AIGC_OSS_QINIU_DOMAIN	访问域名

最小示例

from huace_aigc_oss import OssClient

client = OssClient.from_env()

超时、重试等进阶项见 env.example。

路由原理

下载 / exists（动态探测 + 优先级链）

HUACE_AIGC_OSS_DOWNLOAD_PRIORITY 定义尝试顺序，例如 rustfs,aliyun 表示 第一优先级 RustFS，第二优先级阿里云。

未指定 backend / mode 的 download（默认） 分两阶段：

1. 动态探测（exists）：按优先级询问对象是否存在
   • RustFS：对 HUACE_AIGC_OSS_RUSTFS_ENDPOINTS 中每个内网 IP 依次 head_object
   • 内网全部「连不上」或「文件不存在」→ 再探测下一后端（如阿里云）
2. 下载：仅从探测命中的后端拉取文件；若探测命中但下载失败，再对其余后端降级重试

可用 client.locate(key) 单独查看探测结果（返回 rustfs / aliyun 等，不存在为 None）。

在 HUACE_AIGC_OSS_FALLBACK_ENABLED=true（默认）时才会跨后端探测；为 false 时只探测第一优先级。

exists() 与探测使用同一套优先级链；指定 backend / mode 的 download 不做跨后端探测，只在指定范围内尝试。

上传（单后端，不降级）

调用方式	使用的后端
upload(..., mode=INTRANET)（默认）	HUACE_AIGC_OSS_UPLOAD_INTRANET_BACKEND
upload(..., mode=EXTRANET)	HUACE_AIGC_OSS_UPLOAD_EXTRANET_BACKEND

上传 只写一个后端；该后端失败时 不会 自动切换到另一个后端。

调用时指定目标（可选）

未指定时：上传默认内网，下载默认先内网再外网（与上表一致）。

参数	适用	说明
（不传）	upload / download	上传走内网；下载/exists 走完整优先级链
mode=UploadMode.INTRANET	upload / download	仅内网后端
mode=UploadMode.EXTRANET	upload / download	仅外网后端
backend="rustfs"	upload / download / exists	强制指定后端名：rustfs / aliyun / qiniu

client.download("k", "/tmp/a.wav")                      # 默认：内网 RustFS 多端点 → 外网
client.download("k", "/tmp/a.wav", mode=UploadMode.INTRANET)  # 仅内网
client.download("k", "/tmp/a.wav", backend="aliyun")  # 仅阿里云
client.upload("k", "/tmp/a.wav")                        # 默认内网
client.upload("k", "/tmp/a.wav", backend="aliyun")    # 指定上传阿里云

示例（HUACE_AIGC_OSS_DOWNLOAD_PRIORITY=rustfs,aliyun）

• 下载：先 RustFS 各内网 IP → 全部不可用 再 阿里云。
• 内网上传：固定 RustFS。
• 外网上传（UploadMode.EXTRANET）：固定阿里云。

各后端需使用相同逻辑对象键（含统一前缀 HUACE_AIGC_OSS_FILE_PREFIX）；路由只负责选路，不跨后端复制数据。

快速开始

from huace_aigc_oss import OssClient, UploadMode

client = OssClient.from_env()

# 下载：先第一优先级，无法访问时用第二优先级（见「路由原理」）
local = client.download("task-1/output.wav", "/tmp/output.wav")

# 上传：仅内网指定后端，不切换第二后端
client.upload("task-1/output.wav", "/tmp/output.wav")

# 上传：外网
client.upload("task-1/output.wav", "/tmp/output.wav", mode=UploadMode.EXTRANET)

# 指定后端
client.download("task-1/output.wav", "/tmp/output.wav", backend="aliyun")

client.exists("task-1/output.wav")

插件集成示例

from huace_aigc_oss import OssClient, UploadMode

_client = None

def get_client() -> OssClient:
    global _client
    if _client is None:
        _client = OssClient.from_env()
    return _client

def upload_local_file(local_path: str, oss_key: str, extranet: bool = False) -> str:
    mode = UploadMode.EXTRANET if extranet else UploadMode.INTRANET
    result = get_client().upload(oss_key, local_path, mode=mode)
    return result.url or oss_key

核心行为

操作	行为
download（默认）	先 exists 动态探测（内网多 IP → 外网），命中后再下载
locate	仅探测对象所在后端名，不下载
upload	内网/外网各固定一个后端，失败不切换
exists	与 download 相同：第一优先级不可用时用第二优先级