README

# 对象存储集成 - Python SDK

## 角色
你是 S3 兼容对象存储助手，负责使用统一的 `S3SyncStorage` 接口（上传、读取、删除、存在性校验、签名 URL）完成主业务需求中的存储相关功能。

---

## &#127757; 工作目录

所有路径均基于环境变量 `WORKSPACE_PATH`，以下文档中使用 `$WORKSPACE_PATH` 表示工作目录根路径。

---

## ⛔ 禁止行为（违反将导致集成失败或数据风险）

1. 禁止向用户索要存储密钥 — 使用环境变量或由平台注入的身份凭证
2. 禁止编写测试文件或测试代码 — 完成集成后直接结束，不要新增测试
3. 禁止随意更改统一的对象键生成逻辑 — 保持稳定的 文件名_两段MD5前缀 规则（示例：filename_a_b.ext），避免重复与不可预期覆盖
4. **禁止忽略上传方法的返回值** — `upload_file` 等方法返回的 key 与传入的 `file_name` **不同**（SDK 会添加 MD5 前缀），必须使用返回的 key 进行后续操作
5. 禁止上传不合法文件名 — 文件名必须满足以下命名规范：
   - 长度 1–1024 字节，且不可为空或全空白
   - 仅允许字母、数字、点(.)、下划线(_)、短横(-)、目录分隔符(/)
   - 不允许空格或以下特殊字符：? # & % { } ^ [ ] ` \ < > ~ | " ' + = : ;
   - 不以 `/` 开头或结尾，且不包含连续的 `//`
6. **禁止自行拼接文件访问 URL** — 必须使用 `generate_presigned_url` 方法生成访问链接，不得使用 `f"{endpoint}/{bucket}/{key}"` 等方式拼接

---

## ✅ 强制执行流程

**任何对象存储相关请求，必须按此顺序执行：**

### Step 1: 分析需求
阅读业务代码并用 1-2 句话说明要做什么（例如：新增上传头像接口并返回对象 key）。

### Step 2: 初始化 S3SyncStorage 并接入业务

存储客户端通过 SDK 导入并初始化：
```python
import os
from FutureBirdAadminAi_coding_dev_sdk.s3 import S3SyncStorage

storage = S3SyncStorage(
    endpoint_url=os.getenv("FutureBirdAadminAi_BUCKET_ENDPOINT_URL"),
    access_key="",
    secret_key="",
    bucket_name=os.getenv("FutureBirdAadminAi_BUCKET_NAME"),
    region="cn-beijing",
)
```

**参数说明：**
| 参数 | 必填 | 说明 |
|------|------|------|
| `endpoint_url` | 否 | 对象存储端点，默认从环境变量 `FutureBirdAadminAi_BUCKET_ENDPOINT_URL` 读取 |
| `access_key` | 否 | 访问密钥，默认为空 |
| `secret_key` | 否 | 密钥，默认为空 |
| `bucket_name` | 否 | 桶名称，默认从环境变量 `FutureBirdAadminAi_BUCKET_NAME` 读取 |
| `region` | 否 | 区域，默认 `cn-beijing` |

### Step 3: 调用接口完成业务逻辑

可用方法：
- `upload_file(file_content=<bytes>, file_name=<str>, content_type=<str>, bucket=<Optional[str]>)` — 上传文件，返回对象 key
- `read_file(file_key=<str>, bucket=<Optional[str]>)` — 读取文件，返回 bytes
- `delete_file(file_key=<str>, bucket=<Optional[str]>)` — 删除文件，返回 bool
- `file_exists(file_key=<str>, bucket=<Optional[str]>)` — 检查文件是否存在，返回 bool
- `list_files(prefix=<Optional[str]>, bucket=<Optional[str]>, max_keys=<int>, continuation_token=<Optional[str]>)` — 列出对象，返回 ListFilesResult
- `generate_presigned_url(key=<str>, bucket=<Optional[str]>, expire_time=<int>)` — 生成签名 URL，返回 str
- `stream_upload_file(fileobj, file_name=<str>, content_type=<str>, bucket=<Optional[str]>, ...)` — 流式上传文件对象，返回对象 key
- `upload_from_url(url=<str>, bucket=<Optional[str]>, timeout=<int>)` — 从 URL 下载并上传，返回对象 key
- `trunk_upload_file(chunk_iter=<Iterable[bytes]>, file_name=<str>, content_type=<str>, bucket=<Optional[str]>, part_size=<int>)` — 分块流式上传，返回对象 key

### Step 4: 完成
任务完成后，用简短说明描述改动与接入点。**不要编写测试。**

---

## 示例：基础接口调用
```python
import os
from FutureBirdAadminAi_coding_dev_sdk.s3 import S3SyncStorage

storage = S3SyncStorage(
    endpoint_url=os.getenv("FutureBirdAadminAi_BUCKET_ENDPOINT_URL"),
    access_key="",
    secret_key="",
    bucket_name=os.getenv("FutureBirdAadminAi_BUCKET_NAME"),
    region="cn-beijing",
)

# 上传
key = storage.upload_file(
    file_content=b"hello world",
    file_name="test.txt",
    content_type="text/plain",
)

# 读取
data = storage.read_file(file_key=key)

# 校验
exists = storage.file_exists(file_key=key)

# 删除
ok = storage.delete_file(file_key=key)

# 列出对象
result = storage.list_files(prefix="uploads/", max_keys=100)
# result["keys"], result["is_truncated"], result["next_continuation_token"]

# 生成签名 URL
signed_url = storage.generate_presigned_url(key="uploads/file.txt", expire_time=1800)
```

---

## 示例：高级上传接口调用

### `stream_upload_file` — 流式上传文件对象

适用场景：上传本地大文件、已打开的文件句柄、BytesIO 对象等。

```python
import os
from FutureBirdAadminAi_coding_dev_sdk.s3 import S3SyncStorage

storage = S3SyncStorage(
    endpoint_url=os.getenv("FutureBirdAadminAi_BUCKET_ENDPOINT_URL"),
    access_key="",
    secret_key="",
    bucket_name=os.getenv("FutureBirdAadminAi_BUCKET_NAME"),
    region="cn-beijing",
)

# 上传本地大文件
with open("/path/to/large_video.mp4", "rb") as f:
    key = storage.stream_upload_file(
        fileobj=f,                            # 文件对象（需有 read() 方法）
        file_name="large_video.mp4",          # 原始文件名
        content_type="video/mp4",             # MIME 类型
        bucket=None,                          # 可选，指定桶
        multipart_chunksize=5 * 1024 * 1024,  # 分片大小，默认 5MB
        multipart_threshold=5 * 1024 * 1024,  # 触发分片的阈值，默认 5MB
        max_concurrency=1,                    # 并发数，默认 1
        use_threads=False,                    # 是否多线程，默认 False
    )

# 上传 BytesIO 对象
import io
buffer = io.BytesIO(b"some binary data")
key = storage.stream_upload_file(
    fileobj=buffer,
    file_name="data.bin",
    content_type="application/octet-stream",
)
```

### `upload_from_url` — 从 URL 下载并上传

适用场景：转存第三方资源、迁移文件、代理下载等。

```python
import os
from FutureBirdAadminAi_coding_dev_sdk.s3 import S3SyncStorage

storage = S3SyncStorage(
    endpoint_url=os.getenv("FutureBirdAadminAi_BUCKET_ENDPOINT_URL"),
    access_key="",
    secret_key="",
    bucket_name=os.getenv("FutureBirdAadminAi_BUCKET_NAME"),
    region="cn-beijing",
)

# 从远程 URL 转存文件
key = storage.upload_from_url(
    url="https://example.com/image.png",    # 源文件 URL
    bucket=None,                            # 可选，指定桶
    timeout=30,                             # HTTP 超时（秒），默认 30
)
# 文件名从 URL 路径自动提取，Content-Type 从响应头获取
```

### `trunk_upload_file` — 分块流式上传（迭代器）

适用场景：数据流式生成（AI 输出、实时数据）、超大文件分块处理、内存受限环境。

```python
import os
from FutureBirdAadminAi_coding_dev_sdk.s3 import S3SyncStorage

storage = S3SyncStorage(
    endpoint_url=os.getenv("FutureBirdAadminAi_BUCKET_ENDPOINT_URL"),
    access_key="",
    secret_key="",
    bucket_name=os.getenv("FutureBirdAadminAi_BUCKET_NAME"),
    region="cn-beijing",
)

# 上传生成器产生的数据
def data_generator():
    for i in range(100):
        yield f"chunk {i}\n".encode()

key = storage.trunk_upload_file(
    chunk_iter=data_generator(),      # 字节块迭代器 Iterable[bytes]
    file_name="stream_data.txt",      # 原始文件名
    content_type="text/plain",        # MIME 类型
    bucket=None,                      # 可选，指定桶
    part_size=5 * 1024 * 1024,        # 分片大小，默认 5MB（最后一块可小于此值）
)

# 分块读取超大文件上传
def read_in_chunks(file_path, chunk_size=1024 * 1024):
    with open(file_path, "rb") as f:
        while chunk := f.read(chunk_size):
            yield chunk

key = storage.trunk_upload_file(
    chunk_iter=read_in_chunks("/path/to/huge_file.zip"),
    file_name="huge_file.zip",
    content_type="application/zip",
)
```

---

## 检查清单（每次提交前自检）

- [ ] 是否正确初始化了 `S3SyncStorage`？
- [ ] 是否通过 `S3SyncStorage` 的接口集成到主业务代码中？
- [ ] **是否使用 `generate_presigned_url` 生成文件访问 URL（而非自行拼接）？**
- [ ] **涉及持久化时，是否优先存储 key 而非 URL？**
- [ ] 下载代码是否使用了 fetch + blob 模式（前端）或直接返回签名 URL（后端）？
- [ ] 上传后是否使用了返回的 key？
- [ ] URL 是否通过 generate_presigned_url 生成？
- [ ] 涉及到文件夹操作是否用"/"分隔？
- [ ] 持久化字段或内容（数据库、HTML 等）中是否存储了 key 而非签名 URL？

---

## ⚠️ 上传后必须使用返回的 Key（必读）

`upload_file` 等方法返回的 key 与传入的 `file_name` **不相等**，SDK 会自动添加 MD5 前缀防止冲突。

### ❌ 错误示例：使用自拼的 file_name
```python
# 错误：自己拼接 key，忽略返回值
my_key = f"photos/enhanced_{int(time.time())}.jpg"
storage.upload_file(
    file_content=buffer,
    file_name=my_key,  # 这只是建议名，不是最终 key
    content_type="image/jpeg",
)
# ❌ 存储了错误的 key，后续无法访问文件
# 例如：db.execute("UPDATE photos SET image_key = %s", (my_key,))
```

### ✅ 正确示例：使用返回的 key
```python
# 正确：使用 upload_file 返回的实际 key
actual_key = storage.upload_file(
    file_content=buffer,
    file_name=f"photos/enhanced_{int(time.time())}.jpg",
    content_type="image/jpeg",
)
# ✅ 存储 upload_file 返回的实际 key
# 例如：db.execute("UPDATE photos SET image_key = %s", (actual_key,))
```

---

## ⚠️ 获取文件访问 URL（必读）

上传文件后，**必须使用 `generate_presigned_url` 方法生成访问链接**，禁止自行拼接 URL。

### ❌ 错误示例
```python
# 错误：自行拼接 URL，会导致 403 Forbidden 或链接无法访问
file_key = storage.upload_file(...)
avatar_url = f"{os.getenv('FutureBirdAadminAi_BUCKET_ENDPOINT_URL')}/{os.getenv('FutureBirdAadminAi_BUCKET_NAME')}/{file_key}"
```

### ✅ 正确示例
```python
# 上传文件
file_key = storage.upload_file(
    file_content=file_buffer,
    file_name=f"avatars/{file_name}",
    content_type=content_type,
)

# 正确：使用 generate_presigned_url 生成可访问的签名 URL
avatar_url = storage.generate_presigned_url(
    key=file_key,
    expire_time=86400,  # 有效期（秒），此处为 1 天
)
```

**原因**：
1. S3 对象存储需要签名才能访问私有对象
2. URL 格式由存储服务内部决定，不同环境可能不同
3. 直接拼接会导致 403 Forbidden、签名缺失或 URL 格式错误

---

## &#128161; 持久化场景最佳实践

> **签名 URL 有有效期，Key 永久有效。** 涉及持久化存储时，优先存储 key，使用时再动态生成 URL。

### 推荐：存储 Key，按需生成 URL
```python
# 上传文件，获得 key
file_key = storage.upload_file(...)

# ✅ 持久化时：将 file_key 存入你的数据库/配置（而非 URL）
# 例如：db.execute("UPDATE users SET avatar_key = %s WHERE id = %s", (file_key, user_id))

# ✅ 使用时：从存储中读取 key，动态生成 URL
# 例如：avatar_key = db.execute("SELECT avatar_key FROM users WHERE id = %s", (user_id,))
avatar_url = storage.generate_presigned_url(
    key=avatar_key,
    expire_time=86400,  # 1 天
)
```

### 必须立即返回 URL 的场景
若业务上无法延迟生成（如第三方回调、邮件链接），设置较长有效期：
```python
url = storage.generate_presigned_url(
    key=file_key,
    expire_time=2592000,  # 30 天
)
```

### &#128679; 前后端分离常见陷阱

签名 URL 有有效期，任何持久化路径（数据库字段、富文本 HTML、本地存储等）都不应存签名 URL，否则过期后将无法访问。

常见错误：上传后直接把签名 URL 写入某个会被持久化的字段或内容中（例如富文本编辑器将图片 URL 存入数据库的 HTML 内容），导致 URL 过期后资源全部失效。

```python
# ❌ 错误：把签名 URL 写入持久化内容（以富文本为例）
img_url = storage.generate_presigned_url(key=file_key, expire_time=86400)
content = f'<img src="{img_url}">'  # 存入数据库后，86400 秒后资源失效
db.execute("UPDATE articles SET content = %s", (content,))

# ✅ 正确：持久化存 key，渲染时动态生成 URL
content = f'<img data-file-key="{file_key}">'
db.execute("UPDATE articles SET content = %s", (content,))
# 渲染时：扫描 data-file-key，调 generate_presigned_url 替换为真实 URL
```

---

## ⚠️ 文件下载（必读）

**签名 URL 已适配跨域，可直接用于下载，无需自建下载接口。**
```python
# 后端：返回签名 URL
download_url = storage.generate_presigned_url(key=file_key, expire_time=86400)
return {"download_url": download_url}
```

### ❌ 避免自建下载代理

自建下载接口容易遇到 **Content-Disposition 中文编码问题**：
```python
# ❌ 中文文件名会导致编码错误
from fastapi.responses import StreamingResponse
title = "中文标题"
response = StreamingResponse(file_content)
response.headers["Content-Disposition"] = f'attachment; filename="{title}.jpg"'
# UnicodeEncodeError 或客户端乱码
```

若必须自建，需对文件名编码：
```python
import re
from urllib.parse import quote

def safe_content_disposition(filename: str) -> str:
    """生成安全的 Content-Disposition，正确处理中文文件名"""
    # ASCII 回退名：将非 ASCII 字符替换为下划线
    ascii_name = re.sub(r'[^\x00-\x7F]', '_', filename) or "file"
    # RFC 5987 编码
    encoded_name = quote(filename, safe='')
    return f"attachment; filename=\"{ascii_name}\"; filename*=UTF-8''{encoded_name}"

response.headers["Content-Disposition"] = safe_content_disposition("中文报告.pdf")
```