---
title: "04-hwcloud_api"
created: 2026-04-02
tags:
  - 项目
aliases:
  - hwcloud_api
---
# hwcloud_api.py

### `hwcloud_api.py` — 华为云 OBS 上传

本地文件 / 二进制流两类上传接口（`upload_file_to_obs` / `upload_stream_to_obs`），招聘海报、公司 Logo 等多媒体持久化到 `quanfile` 桶并返回公共 URL。

## 代码

```python
# -*- 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)
```

---

**项目分区导航**：[[03-doubao_api_new|doubao_api_new]] ⬅️ | 04-hwcloud_api | ➡️ [[05-ocr_api|ocr_api]]