hwcloud_api.py
hwcloud_api.py — 华为云 OBS 上传
本地文件 / 二进制流两类上传接口（upload_file_to_obs / upload_stream_to_obs），招聘海报、公司 Logo 等多媒体持久化到 quanfile 桶并返回公共 URL。
代码
# -*- coding: utf-8 -*-
# @Time : 2024/12/17 15:33
# @Author : chang
"""
华为云 OBS 文件上传工具类
功能：提供本地文件上传、二进制流上传到华为云OBS存储的功能
依赖：华为云 obs SDK (pip install esdk-obs-python)
"""
# 导入华为云OBS客户端
from obs import ObsClient
# 导入系统文件操作库
import os
# ======================== 华为云OBS 基础配置 ========================
# OBS 访问终端节点（地域节点），这里使用北京4区
endpoint = 'http://obs.cn-north-4.myhuaweicloud.com'
# 初始化 OBS 客户端（密钥认证）
obs_client = ObsClient(
access_key_id="DTU7ZXZJBFUCO7K274JL", # 华为云 Access Key
secret_access_key="vPXWwWoug8y2CKkQPtXjir9OpwA1TH965qXFSQuU", # 华为云 Secret Key
server=endpoint # 绑定地域节点
)
# ======================== 默认常量配置 ========================
# 默认上传的 OBS 桶名（Bucket）
DEFAULT_BUCKET = "quanfile"
# 默认上传到 OBS 的存储路径（文件夹）
DEFAULT_PATH = 'ann/test'
# 默认文件访问域名（拼接桶域名 + 默认路径，用于生成文件访问URL）
DEFAULT_URL = f'https://quanfile.obs.cn-north-4.myhuaweicloud.com/{DEFAULT_PATH}'
# ======================== 上传函数 ========================
def upload_file_to_obs(filename, local_file_path,headers={},bucket_name=DEFAULT_BUCKET):
"""
【本地文件上传】将本地磁盘文件上传到华为云OBS
:param filename: 上传到OBS后保存的文件名
:param local_file_path: 本地文件的完整路径（绝对路径）
:param headers: 上传请求头（如文件类型、权限等）
:param bucket_name: OBS桶名，不传则使用默认桶 quanfile
:return: (上传成功标记, 访问URL/错误信息)
"""
try:
# 拼接 OBS 上的完整存储路径 = 默认目录 + 文件名
obs_file_name = f'{DEFAULT_PATH}/{filename}'
# 打印上传日志
print(f"Uploading {local_file_path} to {obs_file_name} in bucket {bucket_name}")
# ======================== 核心：执行文件上传 ========================
# putFile：上传本地文件到OBS
# 参数：桶名、OBS存储路径、本地文件路径、元数据、请求头
r = obs_client.putFile(bucket_name, obs_file_name, local_file_path, {}, headers)
# HTTP 状态码 < 300 表示上传成功
if r.status < 300:
print("文件上传成功！", local_file_path)
# 返回成功 + 文件可访问的URL
return True, f'{DEFAULT_URL}/{filename}'
else:
# 上传失败，打印错误原因
print(r.reason, local_file_path, "上传失败")
return False, r.reason
# 捕获所有异常（网络错误、文件不存在、权限错误等）
except Exception as e:
print("文件上传失败：", e)
return False, str(e)
def upload_stream_to_obs(objectKey, content, metadata={},headers={},bucket_name=DEFAULT_BUCKET):
"""
【二进制流上传】将内存中的二进制流（如图片、文件字节流）直接上传到OBS
适用场景：不落地本地磁盘，直接上传网络流、内存文件
:param objectKey: OBS存储路径（含文件名）
:param content: 二进制流数据（bytes）
:param metadata: 文件元数据（如文件名）
:param headers: 请求头
:param bucket_name: OBS桶名
:return: (上传成功标记, 访问URL/错误信息)
"""
try:
# ======================== 核心：执行流上传 ========================
# putObject：上传二进制流到OBS
r = obs_client.putObject(bucket_name, objectKey, content, metadata, headers)
# 状态码 ≤300 判定为成功
if r.status <= 300:
print("文件上传成功！", objectKey)
# 异常捕获
except Exception as e:
print("文件上传失败：", e, objectKey)
return False, str(e)
# 返回成功状态 + 访问URL（从元数据取文件名拼接）
return True, f'{DEFAULT_URL}/{metadata["filename"]}'
# ======================== 测试入口 ========================
if __name__ == '__main__':
# 测试：本地图片路径（请替换为你自己的文件路径）
filepath = "/Users/ziguangchu/Downloads/1111.jpeg"
# ======================== 配置上传头信息 ========================
headers = {
"Content-Type": "image/jpeg", # 文件类型：图片
"x-obs-acl": "public-read" # 设置文件权限：公共读（可直接通过URL访问）
}
# 获取本地文件的文件名（如 1111.jpeg）
filename = os.path.basename(filepath)
# 调用【本地文件上传】函数
_ok, _s = upload_file_to_obs(filename, filepath, headers)
# 打印结果：True/False + 访问URL 或 错误信息
print(_ok, _s)
# ======================== 流上传示例（已注释，可直接启用） ========================
# with open(filepath,"rb") as f:
# content=f.read() # 读取为二进制流
# filename = os.path.basename(filepath)
# # 调用流上传
# _ok,url = upload_stream_to_obs(DEFAULT_PATH, f"{DEFAULT_PATH}/{filename}", content, metadata={"filename": filename})
# print(_ok,url)
项目分区导航：doubao_api_new ⬅️ | 04-hwcloud_api | ➡️ ocr_api
💬 评论