Metadata-Version: 2.4
Name: fileapi-sdk
Version: 0.1.1
Summary: FileApi Open API 的 Python 客户端：请求签名、代理上传、直传与分片续传、秒传、游标列表、下载、访问链接与 Webhook 验签
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: OS Independent
Classifier: Typing :: Typed
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# FileApi Python SDK

FileApi Open API 的官方 Python 客户端：请求签名、代理上传、直传与分片续传、秒传、游标列表、下载、访问链接与 Webhook 验签。

- Python 3.9+，带类型标注（`py.typed`）
- 零依赖，只用标准库（`http.client`、`hashlib`、`hmac`、`concurrent.futures`）
- 客户端可在多个线程中共用；分片直传用线程池并发上传
- 与服务端、管理后台、Go / Java / Node.js SDK 共用同一组签名测试向量，保证算法一致

## 安装

```bash
pip install fileapi-sdk
```

PyPI 上的项目名为 `fileapi-sdk`，导入名为 `fileapi`（`from fileapi import FileApiClient`）。
也可以从源码目录 `sdk/python` 直接安装：`pip install /path/to/FileApi/sdk/python`。

## 快速开始

```python
import os
from fileapi import FileApiClient

client = FileApiClient("https://files.example.com", os.environ["FILEAPI_CLIENT_ID"], os.environ["FILEAPI_CLIENT_SECRET"])

# 代理上传（经 FileApi 写入存储），关联业务对象
f = client.upload_file("./avatar.png", object_type="user_avatar", object_id="42")

# 统一入口链接（与底层存储解耦，迁移存储后仍然有效）
print(client.access_url(f.id, expires_in=3600).url)
```

`base_url` 可以带反向代理前缀（如 `https://example.com/api`），前提是代理转发时去掉该前缀——签名只覆盖服务端看到的 `/open/v1/...` 路径。

调用参数为关键字参数；返回值是 dataclass，字段与服务端 JSON 同名（snake_case），时间解析为带时区的 `datetime`，与 OpenAPI 文档和 Webhook 载荷保持同一形状。
时长参数（`expires_in`）单位为秒。

## 能力一览

| 方法 | 说明 |
|---|---|
| `upload` / `upload_file` | 代理上传。流式发送不占内存，边发边算 SHA-256 附在表单末尾，服务端校验不一致时拒绝；bytes、文件或指定了 `size` 的流带 `Content-Length`，其余分块发送 |
| `upload_direct` / `upload_file_direct` | 直传存储平台（不经过 FileApi）。达到 `multipart_threshold`（默认 64MiB）自动分片并发上传：分片地址不足或将过期时自动补签，单个分片失败重试 3 次；失败时删除待确认记录以释放配额（设置 `resumable` 时保留，可断点续传） |
| `multipart_status` / `resume_upload` | 分片直传进度；从中断处继续分片直传 |
| `upload_instant` / `hash_sha256` | 秒传：按内容 SHA-256 复用本客户端已有的相同文件，不传输内容；`hit=False` 时需正常上传 |
| `get` / `list` / `list_all` | 详情、单页查询（游标或页码）、按游标遍历全部（生成器） |
| `batch_get` / `batch_delete` / `resolve_names` | 批量接口，超过服务端单次上限时自动分批 |
| `open` / `download` / `download_file` | 代理下载（流）、区间下载、整体下载到内存；`download_file` 先写临时文件，完整后再改名 |
| `access_url` | 签发访问链接：统一入口（默认）或存储直链、有效期、附件下载、缩略图、按需图片处理与水印（`image`） |
| `rename` / `set_expiration` | 修改原始文件名、设置或取消过期时间 |
| `batch_renew` / `batch_set_expiration` | 批量续期 / 批量设置或取消过期时间（最多 1000 个） |
| `similar_images` / `similar_image_groups` | 相似图片：与某张图片相似的图片 / 按相似度分组找出重复图片（见下文"相似图片"） |
| `update_metadata` | 修改元数据（合并设置、删除键；列表可按元数据筛选） |
| `delete` / `batch_delete` | 删除（服务端开启回收站时移入回收站）；`permanent=True` 彻底删除 |
| `list_trash` / `restore` / `empty_trash` | 回收站：列表、恢复、清空（见下文"回收站"） |
| `create_archive` | 批量打包下载链接（ZIP，见下文"批量打包下载"） |
| `link_status` / `revoke_link` | 限次 / 可吊销 / 带提取码链接的状态与吊销（见下文"限次 / 可吊销链接"） |
| `list_versions` / `open_version` / `restore_version` / `delete_version` | 文件版本历史（上传新版本用 `version_of`，见下文"文件版本历史"） |
| `video` / `transcode` / `wait_video` / `cancel_transcode` / `delete_video` / `play` | 视频转码为多清晰度 HLS 与播放地址（上传时用 `transcode`，见下文"视频转码与播放"） |
| `list_folders` / `move_files` / `move_file` / `rename_folder` / `delete_folder` | 文件夹（虚拟目录）：列出子目录、移动文件、重命名 / 删除目录（见下文"文件夹"） |
| `hot_files` | 热门文件：最近 N 天访问次数最多的文件；`get` / `list` 返回的 `access_count` / `last_accessed_at` 为累计访问次数与最近访问时间 |
| `usage` | 存储用量与本月流量 |
| `create_upload_token` | 本地签发前端上传令牌，浏览器 / 小程序凭令牌直接上传（见下文"前端上传令牌"） |
| `verify_webhook` | 校验 FileApi 推送的事件签名 |

### 上传的内容

- `upload(filename, data)`：`data` 可以是 `bytes`、`ByteSource`、二进制文件对象（如 `open(path, "rb")`、`io.BytesIO`）或产出 `bytes` 的迭代器。
  后两者的长度用 `size` 指定（只发送前 `size` 字节），不指定时以分块编码发送。
- `upload_direct` / `resume_upload` 需要按区间重复读取内容，接受 `bytes` 或 `ByteSource`；本地文件用 `upload_file_direct`，或传 `file_source(path)`（按分片区间打开文件，不整体读入内存）。
- `ByteSource` 是带 `size` 属性与 `read(start, end)`（产出该区间内容的迭代器）的对象：内置 `bytes_source(data)` 与 `file_source(path)`，
  也可以自行实现（如从其它存储按区间读取）；分片并发上传时 `read` 会被多个线程同时调用。

### 直传与分片直传

```python
f = client.upload_file_direct(
    "./backup.tar.gz",
    part_size=16 << 20,  # 默认由服务端决定
    concurrency=8,       # 默认 4
    on_progress=lambda sent, total: print(f"\r{sent * 100 // total}%", end=""),
)
```

直传需要调用方能访问存储平台的 Endpoint；存储只在内网可达时用代理上传。进度回调在上传线程中调用，分片失败重试时进度会回退。

### 断点续传

设置 `resumable=True` 后分片直传失败时保留已上传的分片，抛出 `ResumableError`；保存其中的 `file_id`（如写入数据库），
进程重启后用 `resume_upload` 继续：SDK 查询已上传的分片，用其 MD5 ETag 校验本地内容（不一致的重新上传），只上传其余分片后合并。
须在服务端清理未完成上传之前完成（`multipart_status(id).expires_at`，默认创建后 24 小时）。

```python
from fileapi import ResumableError, file_source

try:
    client.upload_file_direct("./backup.tar.gz", resumable=True)
except ResumableError as err:
    # 保存 err.file_id，稍后（可在另一个进程中）继续
    client.resume_upload(err.file_id, file_source("./backup.tar.gz"))
```

### 秒传

服务端开启去重（`upload.dedup`，默认开启）时，本客户端已上传过相同内容的文件可以秒传：设置 `try_instant=True` 后 SDK 先读一遍内容计算 SHA-256，
命中则直接登记新文件、不再上传（`instant=True`），未命中再按原方式上传。内容须可重复读取（bytes、文件或 `ByteSource`）。

```python
res = client.upload_file("./report.pdf", try_instant=True)
f = client.upload_file_direct("./backup.tar.gz", try_instant=True)

# 也可以自己算好摘要直接秒传
src = file_source("./report.pdf")
hit = client.upload_instant("report.pdf", hash_sha256(src), src.size)
```

只会命中本客户端经代理上传或秒传的文件（服务端开启 `upload.dedup_direct_max_size` 后也包括直传）。即使不秒传，代理上传的内容与已有文件相同时服务端也会复用已有对象（`deduplicated=True`）。

### 图片处理与水印

`image` 让服务端按宽高、缩放方式、格式与质量处理图片（JPEG / PNG / GIF / WebP），处理结果由服务端缓存，同一参数只处理一次：

```python
from fileapi import FIT_COVER, FORMAT_WEBP, ImageOptions

url = client.access_url(file_id, image=ImageOptions(width=300, height=300, fit=FIT_COVER, format=FORMAT_WEBP, quality=80)).url
```

`ImageOptions.watermark` 加文字或图片水印（二选一，可不缩放单独使用）。图片水印先像普通文件一样上传，再用它的文件 ID：

```python
from fileapi import POSITION_SE, POSITION_TILE, ImageOptions, Watermark

logo = client.upload_file("./logo.png")
url = client.access_url(file_id, image=ImageOptions(width=1080, watermark=Watermark(image_id=logo.id, position=POSITION_SE, opacity=80))).url
# 倾斜平铺的文字水印
tiled = client.access_url(file_id, image=ImageOptions(watermark=Watermark(text="内部资料", position=POSITION_TILE, opacity=30)))
```

每组不同的参数（包括不同的水印文字）服务端各缓存一份结果，不要为每个访问者生成不同的文字。

### 过期时间

上传参数 `expires_in`（秒）设置多久后过期，到期后服务端自动删除；文件的 `expires_at` 为过期时间。已上传的文件用 `set_expiration` 修改，传 `None` 取消过期：

```python
from datetime import datetime, timedelta, timezone

res = client.upload_file("./export.csv", expires_in=24 * 3600)
client.set_expiration(res.id, datetime.now(timezone.utc) + timedelta(days=7))
client.set_expiration(res.id, None)  # 取消过期
```

即将到期的文件：`list(expiring_days=7)`（7 天内到期删除）/ `list(retention_expiring_days=7)`（7 天内保留锁定到期），批量续期：

```python
page = client.list(expiring_days=7)
client.batch_renew(ids, timedelta(days=30))   # 从现在起 30 天后过期，返回修改的文件数
client.batch_set_expiration(ids, None)        # 取消过期
```

到期前（服务端系统设置 `expiry_alert.days_before`，默认 3 天）推送 `EventType.FILE_EXPIRING` / `FILE_RETENTION_EXPIRING`，`event.expiring()` 取出即将到期的文件。

### 相似图片

服务端后台为图片计算感知哈希，`distance`（0 ~ 64）越小越相似，缩放、重新压缩后仍能识别；新上传的图片稍后才能参与比较（`hashed` 为 False 时稍后再查）：

```python
res = client.similar_images(file_id)                            # 默认阈值 8，按距离从近到远
groups = client.similar_image_groups(folder="/相册/")           # 重复图片分组，每组第一张为最新的
```

### 列表

```python
from fileapi import CATEGORY_IMAGE

# 遍历（按游标，不做 OFFSET 翻页，数据量大也不变慢）
for f in client.list_all(category=CATEGORY_IMAGE):
    print(f.id, f.original_filename)

# 手动翻页
page = client.list(page_size=50, with_total=True)
nxt = client.list(page_size=50, cursor=page.next_cursor)
```

### 下载

```python
data = client.download(file_id)                  # 小文件
client.download_file(file_id, "./out.pdf")       # 写入文件
with client.open(file_id, offset=0, length=1024) as d:  # 流 + 区间
    for chunk in d:                               # d.size 为文件总大小，d.filename 为原始文件名
        sys.stdout.buffer.write(chunk)
```

### 元数据检索与修改

按元数据筛选用 `list` 的 `metadata` / `metadata_keys`（文件须包含全部键值 / 须包含这些键）；`update_metadata(id, set=..., remove=...)` 与已有元数据合并修改，remove 中的键删除，其余键不变（最多 20 项）。

### 视频转码与播放

上传（`upload` / `upload_direct` / `upload_instant` / `create_upload_token`）时设置 `transcode=True` 或 `profiles`，服务端的 transcoder 把视频转为多档 HLS；
`wait_video` 等待完成，`play` 签发 HLS 主清单地址（hls.js、Safari / iOS 可直接播放）。

```python
f = client.upload_file("movie.mp4", transcode=True)          # 或 profiles=["720p", "1080p"]
v = client.wait_video(f.id, timeout=600)                      # v.status == "ready"，v.renditions 为各档
p = client.play(f.id, expires_in=7200)                        # p.playable 时 p.url 为 .../v/{token}/master.m3u8，否则 p.fallback_url
client.transcode(f.id, ["360p", "720p"])                      # 换档位重新转码（完成前仍可播放旧产物）
```

字幕：`put_subtitle(id, "zh-CN", srt_or_vtt, label="简体中文", default=True)` 上传或替换某一语言的字幕（SRT 自动转为 WebVTT，每个文件最多 20 条），`subtitles` / `subtitle` / `delete_subtitle` 查询与删除；有可播放产物时主清单自动带上字幕分组，`play` 返回 `subtitles`。

播放地址提取码：`play(id, password="Ab12")` 签发的地址 `password_protected` 为 True，不能直接播放；播放页收集提取码后 `POST {url 所在目录}/unlock`（`{"password": "…"}`）换取带凭据、可直接播放的地址。

HLS 加密：服务端开启 `video.encrypt`（或客户端 `video_encrypt`）时分片经 AES-128 加密，`video.encrypted` 为 True；播放器按媒体清单的 `EXT-X-KEY` 自动从播放地址同一令牌的 `/key` 取密钥，接入方无需改动。

多编码、水印与 DASH：`transcode(id, ["720p"], codecs=["h264", "hevc"], watermark=VideoWatermark(text="版权所有", position="ne"))`
按编码各输出一套档位（`720p` / `720p-hevc` / `720p-av1`，`rendition.codec`），烧录文字或图片（`image_file_id`）水印（不传用客户端默认水印，`False` 不加）；
新产物音视频分离（音频档名为 `audio`），`video.dash` 为 True 时 `play` 返回 `dash_url`（DASH 清单，dash.js / Shaka / ExoPlayer 可播放；加密的产物只有 HLS）。

拖动预览：服务端开启 `video.thumbnails`（默认开启）时转码顺带生成雪碧图，`video.thumbnails` 为间隔、单格尺寸与数量，`play.thumbnails_url` 为 WebVTT 索引（每条为雪碧图地址 `#xywh=x,y,w,h`，可交给 Video.js / Plyr / Vidstack 等播放器的缩略图预览）。

转码完成 / 失败推送 `EventType.VIDEO_READY` / `EventType.VIDEO_FAILED`（`event.video_event()` 取出数据）；`FileInfo.video` 带转码状态，`usage().video_bytes` 为产物占用。

### 文件夹

文件可以归入虚拟目录（如 `/合同/2026`）：`upload` / `upload_direct` / `upload_instant` / `create_upload_token` 的 `folder` 参数指定目录，
`FileInfo.folder` 为所在目录（以 `/` 开头和结尾，根目录为 `/`）。目录随文件存在，移动与重命名只修改目录字段，不改动存储对象与链接。

```python
client.upload_file("/tmp/甲方.pdf", folder="/合同/2026")
folders = client.list_folders("/合同").folders               # 子目录（含其中的文件数与大小）
page = client.list(folder="/合同", recursive=True)           # 目录中的文件（含下级目录）
client.move_files([1, 2], "/归档")
client.rename_folder("/合同", "/法务/合同")                    # 下级目录与回收站中的文件一并修改
arch = client.create_archive([], folder="/法务")              # 打包整个目录
res = client.delete_folder("/临时")                           # res.remaining > 0 时再次调用
```

文件清单导出：`export_files("files.csv", folder="/合同", recursive=True)` 按与 list 相同的条件流式导出 CSV（带 BOM，dest 可为路径或二进制文件对象），返回写入的字节数（服务端一次最多 100 万行）。

文件标签：`create_tag("合同", "blue")` 新建标签，`set_file_tags(file_id, [tag.id])` 设置文件的标签（替换），`batch_tag(ids, add=…, remove=…)` 批量添加 / 移除，`list_tags` / `update_tag` / `delete_tag` 管理标签库；`FileInfo.tags` 为文件的标签，`list(tag_ids=…, tag_any=True)` 按标签筛选。自动打标签规则：`create_tag_rule("合同 PDF", [tag.id], extensions=["pdf"], folder="/合同/", recursive=True)` 之后上传的新文件满足条件时自动打标签，`preview_tag_rule` 试运行、`apply_tag_rule` 给已有文件补打（返回的 remaining 为 True 时再次调用），`list_tag_rules` / `update_tag_rule` / `delete_tag_rule` 管理规则。上传时直接打标签：`upload` / `upload_direct` / `upload_instant` / `create_upload_token` 的 `tag_ids`（最多 20 个，不能与 `version_of` 同时使用）；`share_tags([tag.id])` 分享带有全部这些标签的文件（创建时取当时的文件，之后改标签不影响该分享）。

月度用量（月度账单）：`usage_report(month="2026-09")` 返回该月的存储（月末、峰值、按天平均）、上传与删除、下载流量与转码用量，`from_month` / `to_month` 查询多个月（最多 12 个），`usage_report_csv` 返回 CSV（带 BOM，Excel 可直接打开）。

压缩包预览：`zip_entries(id)` 列出 ZIP 文件中的条目（服务端只读取文件末尾的目录，不下载整个文件，GBK 文件名自动转换），`open_zip_entry(id, entry.index)` 打开其中一个文件的解压内容流（计入下载流量）。

保留锁定（合规保留 / 法律保留）：`set_retention(id, retain_until=...)` 设置或延长保留期限，`legal_hold=True` 设置或解除法律保留；锁定期内（`FileInfo.locked`）删除、上传新版本、恢复或删除历史版本抛出 `Codes.FILE_LOCKED`（42302），`list(locked=True)` 按是否锁定筛选。客户端只能延长保留期限，缩短或清除需要管理员操作；客户端配置了 `retention_days` 时上传的文件自动锁定。

Office 文档预览：上传的 Word / Excel / PPT 由服务端转为 PDF（系统设置 `preview.auto`，或 `request_preview` 按需生成），`wait_preview` 等待完成，`FileInfo.preview` 为状态；`access_url(id, preview=True)` 签发预览 PDF 的地址，浏览器内直接打开。

### 分享链接

```python
s = client.share_files([f.id], password="Ab12", expires_in=7 * 86400, max_downloads=10)
# s.url 为 FileApi 自带的分享页（{base}/s/{code}）：文件列表、在线预览、下载与打包下载；访问者输入提取码后可见
client.share_folder("/合同/2026/", recursive=True, preview_only=True)  # 分享目录（访问时实时读取），只能预览
page = client.shares(page=1)                                          # 下载与浏览次数在 downloads / views 中
client.delete_share(s.code)                                           # 取消分享，立即失效
```

自建分享页时可用 `GET {s.url}/info`（JSON，提取码放在 `X-Link-Password` 请求头）取得文件列表，文件地址为 `{s.url}/f/{id}`（`?download=1` 下载）。

### 文件收集

```python
col = client.create_collection("第 3 周作业", tag_ids=[tag.id], extensions=["pdf"], max_files=100, group_by_uploader=True, expires_in=7 * 86400)
# col.url 为 FileApi 自带的收集页（{base}/c/{code}）：外部人员无需登录即可上传，文件放入 col.folder 并打上标签
files = client.collection_files(col.code)  # 收到的文件（metadata["uploader"] 为提交者姓名）
client.update_collection(col.code, enabled=False)  # 关闭收集
```

每次提交推送 `EventType.COLLECTION_SUBMITTED`，`event.collection_submission()` 取出数据（收集码、姓名、备注、上传成功的文件与失败数）；`Collection.submissions` 为提交次数。

### 批量打包下载

`create_archive` 签发多个文件的 ZIP 下载链接：访问时服务端边读边写流式返回，包内路径可含目录、同名自动加序号，有效期内可多次下载，按实际字节计入下载流量。

```python
arch = client.create_archive([(1, "合同/甲方.pdf"), 2], filename="合同资料", expires_in=3600)
# arch.url 交给浏览器下载
```

### 限次 / 可吊销链接

`access_url` 的 `max_visits` 限定访问次数、`revocable=True` 签发可吊销的链接（仅统一入口链接，须有有效期），返回的 `link_id` 用于查询与吊销。
完整下载或从头开始的分段请求计一次，视频拖动、断点续传与 HEAD 不计；用完返回 410 / `Codes.LINK_EXHAUSTED`，吊销后返回 410 / `Codes.LINK_REVOKED`。

```python
u = client.access_url(file_id, expires_in=86400, max_visits=3)
link = client.link_status(u.link_id)  # link.visits / link.max_visits
client.revoke_link(u.link_id)         # 立即失效
```

`access_url` 的 `password` 设置提取码（4 ~ 16 位字母或数字，不区分大小写）：浏览器打开链接时先输入提取码，程序访问在 `LINK_PASSWORD_HEADER`（`X-Link-Password`）请求头中携带；
返回的 `password_protected` 为 True。缺少 / 错误 / 错误次数过多分别为 `Codes.LINK_PASSWORD_REQUIRED` / `LINK_PASSWORD_WRONG` / `LINK_PASSWORD_LOCKED`。

### 文件版本历史

上传时传 `version_of`（文件 ID，`upload` 与 `upload_direct` 均可）即作为该文件的新版本：文件 ID、业务对象、元数据、过期时间与已签发的链接不变，
旧内容保留为历史版本，返回文件的最新信息（`version` 加 1）。每个文件保留的历史版本数按客户端设置（默认 10 个），超出时删除最旧的；历史版本字节计入配额，
文件彻底删除时一并删除。新版本不参与秒传与去重。

```python
f = client.upload_file("合同-修订.pdf", version_of=file_id)
versions = client.list_versions(file_id)                     # 版本号大的在前
with client.open_version(file_id, versions[0].id) as obj:    # 下载历史版本
    data = obj.read()
client.restore_version(file_id, versions[0].id)              # 恢复为当前内容（当前内容转为历史版本）
client.delete_version(file_id, versions[0].id)
```

### 媒体信息

文件信息带 `width` / `height` / `duration_ms`：图片（按 EXIF 方向校正后）与视频的宽高、音视频时长（毫秒），未知时为空。生成了缩略图的图片上传时即返回，
其余由服务端在上传后补齐（音视频需服务端配置 ffprobe）。按媒体信息筛选用 `list` 的 `min_width` / `max_width` / `min_height` / `max_height` / `min_duration_ms` / `max_duration_ms`（0 表示不限）。

### 回收站

服务端默认开启回收站（系统设置 `trash.retention_days`，客户端可单独设置）：`delete` / `batch_delete` 把文件移入回收站，
保留期内查询、下载、访问链接都如同已删除，也不占用配额；`restore` 恢复后之前签发且未过期的链接重新可用。

```python
client.delete(file_id)                         # 移入回收站（服务端关闭回收站时直接彻底删除）
trash = client.list_trash(object_type="avatar")  # 文件带 trashed_at、purge_at
f = client.restore(file_id)                    # 超出配额时抛出业务码 41302
client.delete(file_id, permanent=True)         # 不进回收站；也用于彻底删除回收站中的文件
scheduled = client.empty_trash()               # 清空回收站（服务端在几分钟内彻底删除）
```

Webhook 事件：移入回收站触发 `file.deleted`（`WebhookFile.trashed` 为 True），恢复触发 `file.restored`，彻底删除触发 `file.purged`。

### 错误处理

服务端错误为 `FileApiError`（`status_code`、业务码 `code`、`message`、`request_id`）：

```python
from fileapi import Codes, FileApiError, is_code, is_not_found

try:
    client.get(file_id)
except FileApiError as err:
    if is_not_found(err):
        ...  # 不存在、已删除或属于其它客户端
    elif is_code(err, Codes.QUOTA_EXCEEDED):
        ...  # 配额不足
    else:
        log.error("fileapi %s request_id=%s", err.code, err.request_id)
```

直传时存储平台拒绝上传抛出 `StorageError`；网络错误与超时抛出 `OSError`（幂等请求重试用尽后为 `ConnectionError`）；下载内容不完整抛出 `IOError`。

服务端开启病毒扫描时：代理上传发现病毒返回 `Codes.FILE_INFECTED`（42201，消息带病毒名）；直传的文件确认后 `scan_status` 为 `pending`，
后台扫描发现病毒的文件会被隔离（之后查询返回不存在），可订阅 Webhook `file.quarantined`。服务端开启 `scan.block_pending` 时，
扫描完成前下载、签发直链返回 `Codes.SCAN_PENDING`（42301）。

`client.usage()` 查询存储用量与本月流量（下载流量 `traffic.egress_bytes` = 代理返回的实际字节 + 跳转与直链按对象大小估算的字节）。
客户端设置了月流量配额时，达到后下载、统一入口访问与签发直链返回 `Codes.TRAFFIC_EXCEEDED`（42903），下月 1 日恢复。
设置了月转码分钟配额（`usage().transcode_quota_minutes`，本月已用 `transcode_used_ms`）时，用完后 `transcode` 抛出 `Codes.TRANSCODE_EXCEEDED`（42905），剩余不足本次所需（源时长 × 档位数）时转码任务失败。

### 超时、重试与时钟

- `timeout`（秒，默认 60）作用于建立连接与每次读写；上传大文件时只要数据持续流动就不会超时。
- 查询类请求（GET、批量查询、确认上传等幂等请求）在网络错误与 429/502/503/504 时重试，默认 2 次，指数退避并遵循 `Retry-After`；每次重试重新签名。`retries=0` 关闭。上传与删除不自动重试。
- 本机时钟与服务端偏差超过 5 分钟时，服务端返回 40103，SDK 按响应中的服务端时间校准后立即重试一次，此后的请求都使用校准后的时间（`client.clock_skew`）。

### 选项

```python
import ssl

FileApiClient(
    base_url, client_id, client_secret,
    retries=3,
    retry_backoff=0.2,
    timeout=30,
    user_agent="order-service/1.4",
    ssl_context=ssl.create_default_context(cafile="/etc/ssl/internal-ca.pem"),
    proxy="http://127.0.0.1:3128",  # HTTP 代理，https 经 CONNECT 隧道；存储直传也经过它
)
```

SDK 不读取 `HTTP_PROXY` 等环境变量，需要代理时显式传 `proxy`。

## 前端上传令牌

浏览器、小程序直接上传时，不要把客户端密钥交给前端：业务后端用 `create_upload_token` 在本地签发短时令牌（不请求 FileApi），
前端在请求头 `X-Upload-Token`（`HEADER_UPLOAD_TOKEN`）中携带令牌调用 `POST {base_url}/upload/v1/files`（multipart，文件字段 `file`）。

```python
tok = client.create_upload_token(
    expires_in=600,                 # 秒，默认 900，最长 7 天
    max_size=5 * 1024 * 1024,       # 不超过客户端自身的上限
    content_types=["image/*"],
    object_type="user_avatar", object_id=user_id, replace_existing=True,  # 上传方不能修改
    max_uploads=1,                  # 默认 1
)
# 把 tok.token、tok.expires_at 返回给前端
```

前端推荐使用浏览器 / 小程序 SDK（源码仓库的 `sdk/browser` 目录）：小文件经 FileApi 代理上传，大文件自动分片直传到存储，带进度、取消与断点续传。上传完成照常触发 `file.uploaded` Webhook。
传 `version_of`（文件 ID）签发只能上传该文件新版本的令牌：前端凭它上传的文件作为新版本（文件 ID 与链接不变，旧内容保留为历史版本），不能与业务对象、元数据、`replace_existing`、`file_expires_in` 同时使用。

令牌格式、限制与前端示例见服务端 README 的"前端上传令牌"。

## 接收 Webhook

FileApi 推送事件时带签名头 `X-FileApi-Signature`，用 `verify_webhook` 校验（防伪造、防重放）；须传入原始请求体（未经重新序列化）。

Flask：

```python
from flask import Flask, request
from fileapi import SIGNATURE_HEADER, EventType, WebhookSignatureError, WebhookTimestampError, verify_webhook

app = Flask(__name__)

@app.post("/fileapi/events")
def events():
    try:
        event = verify_webhook(SECRET, request.headers.get(SIGNATURE_HEADER), request.get_data())
    except (WebhookSignatureError, WebhookTimestampError) as err:
        return str(err), 401
    if event.type == EventType.FILE_UPLOADED:
        on_uploaded(event.id, event.file())  # 按事件 ID 去重：同一事件可能重复推送
    return "", 204  # 返回 5xx 时 FileApi 按退避重试
```

用量告警：设置了配额的客户端存储用量 / 本月下载流量达到预警比例（服务端系统设置，默认 80%）与配额时各推送一次 `EventType.QUOTA_STORAGE` / `EventType.QUOTA_TRAFFIC` / `EventType.QUOTA_TRANSCODE`（转码数据为 `used_minutes` / `quota_minutes`），
`event.usage_alert()` 取出数据（`level` 为 `warning` / `exceeded`，含用量、配额与百分比）；当前级别也在 `usage().alerts` 中返回。

FastAPI：

```python
@app.post("/fileapi/events", status_code=204)
async def events(request: Request):
    try:
        event = verify_webhook(SECRET, request.headers.get(SIGNATURE_HEADER), await request.body())
    except (WebhookSignatureError, WebhookTimestampError) as err:
        raise HTTPException(401, str(err))
    ...
```

## 开发与测试

```bash
PYTHONPATH=src python3 -m unittest discover -s tests                                      # 单元测试（模拟服务端）
FILEAPI_ADMIN_API_KEY=... PYTHONPATH=src python3 -m unittest discover -s tests_integration  # 对运行中的 FileApi 做端到端测试
```

端到端测试的环境变量：`FILEAPI_BASE_URL`（默认 `http://127.0.0.1:8080`）、`FILEAPI_ADMIN_API_KEY`、`FILEAPI_TEST_STORAGE_ID`（可选，客户端绑定的存储）。
