# 示例代码

> Fileverity API 的完整 Node.js 和 Python 程序：检测图片、检测大型 PDF、检测 AI 撰写的文本以及校验 Webhook 签名。

来源： https://fileverity.com/zh/docs/sample-code

本页提供 Fileverity API 的四个完整程序，每个程序都有 Node.js 18 或更高版本和 Python 3.9 或更高版本两种实现：检测图片并等待结果、检测大型 PDF、检测 AI 撰写的文本，以及接收 Webhook 并校验签名。复制其中一个，设置 `FILEVERITY_API_KEY` 后运行即可。

这些程序直接调用 HTTPS API，Node.js 使用内置的 `fetch`，Python 使用 `requests` 包，因此您只需要一个 HTTP 客户端。其中的调用、字段和状态与[文件](https://fileverity.com/zh/docs/files.md)、[分析](https://fileverity.com/zh/docs/analyses.md)、[文本检测](https://fileverity.com/zh/docs/text-detections.md)和 [Webhooks](https://fileverity.com/zh/docs/webhooks.md) 中的描述一致。

| 程序 | 功能 | 调用 |
| --- | --- | --- |
| [检测图片](https://fileverity.com/zh/docs/sample-code.md#image) | 上传图片，等待结果，并输出摘要和报告 URL。 | `POST /files`, `POST /analyses`, `GET /analyses/{id}` |
| [检测大型 PDF](https://fileverity.com/zh/docs/sample-code.md#pdf) | 通过预签名 URL 上传文档，轮询直到检测完成，并保存 PDF 报告。 | `POST /files`, `GET /files/{id}`, `POST /analyses`, `GET /analyses/{id}`, `GET /analyses/{id}/report.pdf` |
| [检测 AI 撰写的文本](https://fileverity.com/zh/docs/sample-code.md#text) | 发送文本文件，并输出判定结果、分数以及每个 AI 撰写的段落。 | `POST /text/detections` |
| [接收 Webhook](https://fileverity.com/zh/docs/sample-code.md#webhooks) | 校验每次投递的签名，忽略重复投递，并处理事件。 | 您的端点，由 `analysis.completed`、`analysis.failed` 和 `balance.low` 事件调用 |

## 运行之前

- 在控制台的 [API 密钥](https://fileverity.com/console/keys?lang=zh)中创建一个 **Test** 密钥。测试密钥返回固定的示例结果，永不扣费，因此所有程序都可以免费运行。如需检测真实文件，请创建 **Live** 密钥，程序无需修改。
- Node.js：将每个程序保存为 `.mjs` 文件，这样即可使用 `import` 和顶层 `await`，然后用 `node` 运行。无需安装任何依赖。
- Python：运行 `pip install requests`。Webhook 接收程序还需要运行 `pip install flask`。

**设置密钥: Shell**

```bash
export FILEVERITY_API_KEY=fv_test_...
```

**设置密钥: PowerShell**

```powershell
$env:FILEVERITY_API_KEY = "fv_test_..."
```

## 程序如何处理密钥、错误和重试

调用 API 的三个程序共用相同的 `request` 和 `api` 函数。可以原样复制到您自己的代码中。

| 方面 | 程序的做法 |
| --- | --- |
| 密钥 | 从 `FILEVERITY_API_KEY` 读取，绝不写入代码。请从您的服务器调用 API，不要在浏览器代码中调用。 |
| 幂等性 | 每个 POST 请求都会发送 `Idempotency-Key`，每个请求使用一个新的 UUID，并在每次重试时复用它，因此重试不会让同一次检测运行或计费两次。若要让整个脚本重新运行时也安全，请改为根据您自己的记录 ID 生成该值。参见[幂等性](https://fileverity.com/zh/docs/idempotency.md)。 |
| 错误 | 非 `2xx` 的响应会包含 `error.type`、`error.code`、`error.message`、`error.param` 和 `error.request_id`。程序会以 `code: message (request request_id)` 的形式报错并停止。请以 `code` 为判断依据，并在联系技术支持时提供请求 ID。参见[错误](https://fileverity.com/zh/docs/errors.md)。 |
| 重试 | 网络故障、`5xx` 状态码（`internal_error`、`analysis_failed`、`service_unavailable`）、`rate_limited`（429）和 `idempotency_in_progress`（409）都会重试，最多尝试 5 次。若响应带有 `retry-after` 响应头，则等待该时长，否则依次等待 2、4、8 和 16 秒。其他所有 `4xx` 错误会立即停止。`pending_files_limit` 同样是 429，但不带 `retry-after`，因此不会重试。 |
| 速率限制 | 当响应头显示 `x-ratelimit-remaining-requests: 0` 时，程序会先等待 `x-ratelimit-reset-requests` 秒再发起下一次调用，从而避免触发 429。参见[速率限制与 HTTP 头](https://fileverity.com/zh/docs/rate-limits.md#headers)。 |
| 超时 | Python 程序的单次请求超时时间为 90 秒，长于 `Prefer: wait` 最长可保持请求的 60 秒。 |

## 检测图片并等待结果

程序以 multipart 方式上传图片，并通过 `Prefer: wait=60` 创建分析，使请求保持等待，直到结果就绪，然后输出摘要和报告 URL。如果等待超时而结果仍未就绪，API 会返回 `202` 和 `status: processing`，程序随后每 3 秒轮询一次 `GET /analyses/{id}`。参见[等待、轮询或使用 Webhook](https://fileverity.com/zh/docs/analyses.md#wait)。

将其保存为 `check-image.mjs` 或 `check_image.py`，然后运行 `node check-image.mjs photo.jpg` 或 `python check_image.py photo.jpg`。multipart 上传最大支持 4 MB 的图片。更大的图片（最大 10 MB）请使用下一个程序中的预签名上传，并指定该图片的内容类型，例如 `image/jpeg`。

**检测图片: JavaScript**

```javascript
import { randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";
import { basename } from "node:path";

const base = "https://api.fileverity.com/v1";
const key = process.env.FILEVERITY_API_KEY;
const path = process.argv[2];
if (!key || !path) throw new Error("Usage: FILEVERITY_API_KEY=... node check-image.mjs photo.jpg");

const retryable = ["rate_limited", "idempotency_in_progress"];
const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

async function request(url, init) {
  for (let attempt = 1; ; attempt++) {
    let response = null;
    let problem;
    try {
      response = await fetch(url, init);
      if (response.ok) {
        const remaining = response.headers.get("x-ratelimit-remaining-requests");
        const reset = Number(response.headers.get("x-ratelimit-reset-requests"));
        if (remaining === "0") await sleep(reset);
        return response;
      }
      problem = (await response.json().catch(() => ({}))).error ?? {};
    } catch (error) {
      problem = { code: "network_error", message: error.message };
    }
    const status = response?.status ?? 0;
    const retry = status === 0 || status >= 500 || retryable.includes(problem.code);
    if (!retry || attempt === 5) {
      const { code = status, message = "Request failed", request_id: id = "unknown" } = problem;
      throw new Error(`${code}: ${message} (request ${id})`);
    }
    await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt);
  }
}

async function api(method, route, { headers, body } = {}) {
  const auth = { Authorization: `Bearer ${key}` };
  const response = await request(base + route, { method, headers: { ...auth, ...headers }, body });
  return response.json();
}

const form = new FormData();
form.append("file", new Blob([await readFile(path)]), basename(path));
const file = await api("POST", "/files", {
  headers: { "Idempotency-Key": randomUUID() },
  body: form,
});

let analysis = await api("POST", "/analyses", {
  headers: {
    "Content-Type": "application/json",
    "Idempotency-Key": randomUUID(),
    Prefer: "wait=60",
  },
  body: JSON.stringify({ model: "fvimg-2", file: file.id }),
});
while (analysis.status === "processing") {
  await sleep(3);
  analysis = await api("GET", `/analyses/${analysis.id}`);
}

if (analysis.status !== "succeeded") {
  throw new Error(`The check ${analysis.status}: ${analysis.error?.message}`);
}
const { level, action, assessment, ai, edit } = analysis.summary;
console.log(`${analysis.id} ${analysis.status}`);
console.log(`level ${level}, action ${action}, assessment ${assessment}`);
console.log(`AI generation: ${ai}, editing: ${edit}`);
console.log(`Report: ${analysis.report_url}`);
```

**检测图片: Python**

```python
import os
import sys
import time
import uuid

import requests

BASE = "https://api.fileverity.com/v1"
KEY = os.environ.get("FILEVERITY_API_KEY")
if not KEY or len(sys.argv) < 2:
    sys.exit("Usage: FILEVERITY_API_KEY=... python check_image.py photo.jpg")

RETRYABLE = ("rate_limited", "idempotency_in_progress")


def request(method, url, **kwargs):
    for attempt in range(1, 6):
        try:
            response = requests.request(method, url, timeout=90, **kwargs)
        except requests.RequestException as error:
            response, problem = None, {"code": "network_error", "message": str(error)}
        else:
            if response.ok:
                if response.headers.get("x-ratelimit-remaining-requests") == "0":
                    time.sleep(int(response.headers["x-ratelimit-reset-requests"]))
                return response
            try:
                problem = response.json()["error"]
            except (ValueError, KeyError, TypeError):
                problem = {}
        status = response.status_code if response is not None else 0
        retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE
        if not retry or attempt == 5:
            code = problem.get("code", status)
            message = problem.get("message", "Request failed")
            request_id = problem.get("request_id", "unknown")
            raise RuntimeError(f"{code}: {message} (request {request_id})")
        wait = response.headers.get("retry-after") if response is not None else None
        time.sleep(int(wait or 2**attempt))


def api(method, route, headers=None, **kwargs):
    headers = {"Authorization": f"Bearer {KEY}", **(headers or {})}
    return request(method, BASE + route, headers=headers, **kwargs).json()


with open(sys.argv[1], "rb") as source:
    data = source.read()
upload = {"file": (os.path.basename(sys.argv[1]), data)}
file = api("POST", "/files", headers={"Idempotency-Key": str(uuid.uuid4())}, files=upload)

analysis = api(
    "POST",
    "/analyses",
    headers={"Idempotency-Key": str(uuid.uuid4()), "Prefer": "wait=60"},
    json={"model": "fvimg-2", "file": file["id"]},
)
while analysis["status"] == "processing":
    time.sleep(3)
    analysis = api("GET", f"/analyses/{analysis['id']}")

if analysis["status"] != "succeeded":
    problem = analysis["error"] or {}
    sys.exit(f"The check {analysis['status']}: {problem.get('message')}")
summary = analysis["summary"]
print(f"{analysis['id']} {analysis['status']}")
print(f"level {summary['level']}, action {summary['action']}, assessment {summary['assessment']}")
print(f"AI generation: {summary['ai']}, editing: {summary['edit']}")
print(f"Report: {analysis['report_url']}")
```

**图片检测的输出**

```text
an_8Hk2pQ4vN9sL1xR7cW3mY6zB succeeded
level L2, action accept, assessment authentic_untouched
AI generation: very_low, editing: none
Report: https://fileverity.com/verify/FV-7K2M-9QXD-4T8R
```

`summary` 是图片检测稳定且机器可读的判定结果，包含 `level`、`action`、`assessment`、`ai` 和 `edit`，详见[图片摘要](https://fileverity.com/zh/docs/analyses.md#summary-image)。`report_url` 是该报告的公开核验页。使用测试密钥时结果是固定的示例，其报告 URL 没有对应的页面。

## 检测大型 PDF

50 MB 以内的文档分三步上传：用 `POST /files` 和 JSON 请求体申请上传 URL，用 `PUT` 将文件内容发送到该 URL，并等待文件状态变为 `uploaded`。随后程序创建分析并每 5 秒轮询一次，因为文档可能需要排队，检测也可能需要一段时间。程序会输出进度、摘要和报告 URL，并把 PDF 报告保存在脚本所在目录。

将其保存为 `check-pdf.mjs` 或 `check_pdf.py`，然后运行 `node check-pdf.mjs contract.pdf` 或 `python check_pdf.py contract.pdf`。如需检测其他类型的文档，请将 `content_type` 改为[支持的内容类型](https://fileverity.com/zh/docs/files.md#content-types)之一。如果文档已加密，请在分析请求中添加 `options: { password }`。

**检测大型 PDF: JavaScript**

```javascript
import { randomUUID } from "node:crypto";
import { readFile, writeFile } from "node:fs/promises";
import { basename } from "node:path";

const base = "https://api.fileverity.com/v1";
const key = process.env.FILEVERITY_API_KEY;
const path = process.argv[2];
if (!key || !path) throw new Error("Usage: FILEVERITY_API_KEY=... node check-pdf.mjs contract.pdf");

const retryable = ["rate_limited", "idempotency_in_progress"];
const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

async function request(url, init) {
  for (let attempt = 1; ; attempt++) {
    let response = null;
    let problem;
    try {
      response = await fetch(url, init);
      if (response.ok) {
        const remaining = response.headers.get("x-ratelimit-remaining-requests");
        const reset = Number(response.headers.get("x-ratelimit-reset-requests"));
        if (remaining === "0") await sleep(reset);
        return response;
      }
      problem = (await response.json().catch(() => ({}))).error ?? {};
    } catch (error) {
      problem = { code: "network_error", message: error.message };
    }
    const status = response?.status ?? 0;
    const retry = status === 0 || status >= 500 || retryable.includes(problem.code);
    if (!retry || attempt === 5) {
      const { code = status, message = "Request failed", request_id: id = "unknown" } = problem;
      throw new Error(`${code}: ${message} (request ${id})`);
    }
    await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt);
  }
}

async function api(method, route, { headers, body } = {}) {
  const auth = { Authorization: `Bearer ${key}` };
  const response = await request(base + route, { method, headers: { ...auth, ...headers }, body });
  return response.json();
}

const bytes = await readFile(path);
const file = await api("POST", "/files", {
  headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() },
  body: JSON.stringify({
    filename: basename(path),
    content_type: "application/pdf",
    bytes: bytes.length,
  }),
});

if (file.status === "pending") {
  const { url, method, headers } = file.upload;
  await request(url, { method, headers, body: bytes });
  let ready = false;
  for (let attempt = 0; attempt < 15 && !ready; attempt++) {
    ready = (await api("GET", `/files/${file.id}`)).status === "uploaded";
    if (!ready) await sleep(2);
  }
  if (!ready) throw new Error("The upload was not confirmed.");
}
console.log(`${file.id} uploaded`);

let analysis = await api("POST", "/analyses", {
  headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() },
  body: JSON.stringify({ model: "fvdoc-3", file: file.id }),
});
const deadline = Date.now() + 15 * 60 * 1000;
while (analysis.status === "processing") {
  if (Date.now() > deadline) {
    throw new Error(`Still processing. Fetch it later with GET /analyses/${analysis.id}`);
  }
  console.log(`${analysis.id} processing: ${analysis.progress?.stage ?? "starting"}`);
  await sleep(5);
  analysis = await api("GET", `/analyses/${analysis.id}`);
}

if (analysis.status !== "succeeded") {
  throw new Error(`The check ${analysis.status}: ${analysis.error?.message}`);
}
const { level, action, reason, pages, images, text, findings } = analysis.summary;
console.log(`${analysis.id} ${analysis.status}`);
console.log(`level ${level}, action ${action}, reason ${reason ?? "none"}`);
console.log(`${pages} pages, ${images} photos examined, text: ${text ?? "not evaluated"}`);
console.log(`findings: ${findings.join(", ") || "none"}`);
console.log(`Report: ${analysis.report_url}`);

if (analysis.livemode) {
  const pdf = await request(`${base}/analyses/${analysis.id}/report.pdf`, {
    headers: { Authorization: `Bearer ${key}` },
  });
  const name = `Fileverity-${analysis.report_id}.pdf`;
  await writeFile(name, Buffer.from(await pdf.arrayBuffer()));
  console.log(`Saved ${name}`);
}
```

**检测大型 PDF: Python**

```python
import os
import sys
import time
import uuid

import requests

BASE = "https://api.fileverity.com/v1"
KEY = os.environ.get("FILEVERITY_API_KEY")
if not KEY or len(sys.argv) < 2:
    sys.exit("Usage: FILEVERITY_API_KEY=... python check_pdf.py contract.pdf")

RETRYABLE = ("rate_limited", "idempotency_in_progress")


def request(method, url, **kwargs):
    for attempt in range(1, 6):
        try:
            response = requests.request(method, url, timeout=90, **kwargs)
        except requests.RequestException as error:
            response, problem = None, {"code": "network_error", "message": str(error)}
        else:
            if response.ok:
                if response.headers.get("x-ratelimit-remaining-requests") == "0":
                    time.sleep(int(response.headers["x-ratelimit-reset-requests"]))
                return response
            try:
                problem = response.json()["error"]
            except (ValueError, KeyError, TypeError):
                problem = {}
        status = response.status_code if response is not None else 0
        retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE
        if not retry or attempt == 5:
            code = problem.get("code", status)
            message = problem.get("message", "Request failed")
            request_id = problem.get("request_id", "unknown")
            raise RuntimeError(f"{code}: {message} (request {request_id})")
        wait = response.headers.get("retry-after") if response is not None else None
        time.sleep(int(wait or 2**attempt))


def api(method, route, headers=None, **kwargs):
    headers = {"Authorization": f"Bearer {KEY}", **(headers or {})}
    return request(method, BASE + route, headers=headers, **kwargs).json()


with open(sys.argv[1], "rb") as source:
    data = source.read()
file = api(
    "POST",
    "/files",
    headers={"Idempotency-Key": str(uuid.uuid4())},
    json={
        "filename": os.path.basename(sys.argv[1]),
        "content_type": "application/pdf",
        "bytes": len(data),
    },
)

if file["status"] == "pending":
    request("PUT", file["upload"]["url"], headers=file["upload"]["headers"], data=data)
    for _ in range(15):
        if api("GET", f"/files/{file['id']}")["status"] == "uploaded":
            break
        time.sleep(2)
    else:
        sys.exit("The upload was not confirmed.")
print(f"{file['id']} uploaded")

analysis = api(
    "POST",
    "/analyses",
    headers={"Idempotency-Key": str(uuid.uuid4())},
    json={"model": "fvdoc-3", "file": file["id"]},
)
deadline = time.time() + 15 * 60
while analysis["status"] == "processing":
    if time.time() > deadline:
        sys.exit(f"Still processing. Fetch it later with GET /analyses/{analysis['id']}")
    stage = (analysis["progress"] or {}).get("stage", "starting")
    print(f"{analysis['id']} processing: {stage}")
    time.sleep(5)
    analysis = api("GET", f"/analyses/{analysis['id']}")

if analysis["status"] != "succeeded":
    problem = analysis["error"] or {}
    sys.exit(f"The check {analysis['status']}: {problem.get('message')}")
summary = analysis["summary"]
reason = summary["reason"] or "none"
text = summary["text"] or "not evaluated"
findings = ", ".join(summary["findings"]) or "none"
print(f"{analysis['id']} {analysis['status']}")
print(f"level {summary['level']}, action {summary['action']}, reason {reason}")
print(f"{summary['pages']} pages, {summary['images']} photos examined, text: {text}")
print(f"findings: {findings}")
print(f"Report: {analysis['report_url']}")

if analysis["livemode"]:
    url = f"{BASE}/analyses/{analysis['id']}/report.pdf"
    pdf = request("GET", url, headers={"Authorization": f"Bearer {KEY}"})
    name = f"Fileverity-{analysis['report_id']}.pdf"
    with open(name, "wb") as target:
        target.write(pdf.content)
    print(f"Saved {name}")
```

**PDF 检测的输出**

```text
file_Pw3mY6zB8Hk2pQ4vN9sL1xR7 uploaded
an_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: queued
an_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: analyzing
an_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: reporting
an_Zc4vB7nP1sK8tL3wR6yQm2Xa succeeded
level L2, action accept, reason clean
12 pages, 3 photos examined, text: human
findings: none
Report: https://fileverity.com/verify/FV-3PXH-6WQ2-8NDC
Saved Fileverity-FV-3PXH-6WQ2-8NDC.pdf
```

- `PUT` 请求直接发送到存储服务，不需要 `Authorization` 请求头，`upload.headers` 中的请求头按返回的内容原样发送。参见[预签名上传](https://fileverity.com/zh/docs/files.md#presigned)。
- 使用测试密钥时，文件直接返回 `uploaded` 状态，无需上传任何内容，因此程序会跳过上传。分析会立即返回固定的示例结果，并且没有示例 PDF，所以程序只在使用 Live 密钥时保存报告。
- 检测完成后，请立即下载 PDF。文档的页面图片和文本摘录只保留 3 小时，之后下载的 PDF 将不再包含这些内容。参见 [PDF 报告](https://fileverity.com/zh/docs/analyses.md#pdf)。
- 以 `failed` 或 `canceled` 结束的检测不会扣费。程序会以该分析的错误信息报错并停止。

## 检测 AI 撰写的文本

文本检测只需一次同步调用。程序读取文本文件，将其发送到 `POST /text/detections`，然后输出判定结果、分数、AI 撰写文本的占比，以及每个 AI 撰写的段落及其偏移量。AI 撰写文本检测支持英语、法语、德语、西班牙语和中文（`en`、`fr`、`de`、`es`、`zh`）。语言会自动识别；也可以将语言代码作为第二个参数传入来指定。文本不得少于 100 个词，且不得超过 100,000 个字符。

将其保存为 `detect-text.mjs` 或 `detect_text.py`，然后运行 `node detect-text.mjs essay.txt` 或 `python detect_text.py essay.txt fr`。[快速入门](https://fileverity.com/zh/docs/quickstart.md#text)中提供了生成示例 `essay.txt` 的命令。使用测试密钥时，响应是针对人类撰写文本的固定结果，因此程序不会输出任何段落。

**检测 AI 撰写的文本: JavaScript**

```javascript
import { randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";

const base = "https://api.fileverity.com/v1";
const key = process.env.FILEVERITY_API_KEY;
const [, , path, language] = process.argv;
if (!key || !path) {
  throw new Error("Usage: FILEVERITY_API_KEY=... node detect-text.mjs essay.txt [en|fr|de|es|zh]");
}

const retryable = ["rate_limited", "idempotency_in_progress"];
const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

async function request(url, init) {
  for (let attempt = 1; ; attempt++) {
    let response = null;
    let problem;
    try {
      response = await fetch(url, init);
      if (response.ok) {
        const remaining = response.headers.get("x-ratelimit-remaining-requests");
        const reset = Number(response.headers.get("x-ratelimit-reset-requests"));
        if (remaining === "0") await sleep(reset);
        return response;
      }
      problem = (await response.json().catch(() => ({}))).error ?? {};
    } catch (error) {
      problem = { code: "network_error", message: error.message };
    }
    const status = response?.status ?? 0;
    const retry = status === 0 || status >= 500 || retryable.includes(problem.code);
    if (!retry || attempt === 5) {
      const { code = status, message = "Request failed", request_id: id = "unknown" } = problem;
      throw new Error(`${code}: ${message} (request ${id})`);
    }
    await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt);
  }
}

async function api(method, route, { headers, body } = {}) {
  const auth = { Authorization: `Bearer ${key}` };
  const response = await request(base + route, { method, headers: { ...auth, ...headers }, body });
  return response.json();
}

const text = await readFile(path, "utf8");
const detection = await api("POST", "/text/detections", {
  headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() },
  body: JSON.stringify({ text, language }),
});

const { verdict, score, ai_fraction: share, usage, segments } = detection;
console.log(`verdict: ${verdict} (score ${score.toFixed(2)})`);
console.log(`AI-written share: ${Math.round(share * 100)}% of ${usage.characters} characters`);
console.log(`language: ${detection.language}, billed units: ${usage.billed_units}`);
const points = Array.from(text);
for (const { start, end, score: passage } of segments) {
  const excerpt = points.slice(start, Math.min(end, start + 80)).join("");
  console.log(`${start}-${end} score ${passage.toFixed(2)}: ${excerpt}`);
}
```

**检测 AI 撰写的文本: Python**

```python
import os
import sys
import time
import uuid

import requests

BASE = "https://api.fileverity.com/v1"
KEY = os.environ.get("FILEVERITY_API_KEY")
if not KEY or len(sys.argv) < 2:
    sys.exit("Usage: FILEVERITY_API_KEY=... python detect_text.py essay.txt [en|fr|de|es|zh]")
sys.stdout.reconfigure(encoding="utf-8")

RETRYABLE = ("rate_limited", "idempotency_in_progress")


def request(method, url, **kwargs):
    for attempt in range(1, 6):
        try:
            response = requests.request(method, url, timeout=90, **kwargs)
        except requests.RequestException as error:
            response, problem = None, {"code": "network_error", "message": str(error)}
        else:
            if response.ok:
                if response.headers.get("x-ratelimit-remaining-requests") == "0":
                    time.sleep(int(response.headers["x-ratelimit-reset-requests"]))
                return response
            try:
                problem = response.json()["error"]
            except (ValueError, KeyError, TypeError):
                problem = {}
        status = response.status_code if response is not None else 0
        retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE
        if not retry or attempt == 5:
            code = problem.get("code", status)
            message = problem.get("message", "Request failed")
            request_id = problem.get("request_id", "unknown")
            raise RuntimeError(f"{code}: {message} (request {request_id})")
        wait = response.headers.get("retry-after") if response is not None else None
        time.sleep(int(wait or 2**attempt))


def api(method, route, headers=None, **kwargs):
    headers = {"Authorization": f"Bearer {KEY}", **(headers or {})}
    return request(method, BASE + route, headers=headers, **kwargs).json()


with open(sys.argv[1], encoding="utf-8") as source:
    text = source.read()
language = sys.argv[2] if len(sys.argv) > 2 else None
detection = api(
    "POST",
    "/text/detections",
    headers={"Idempotency-Key": str(uuid.uuid4())},
    json={"text": text, "language": language},
)

usage = detection["usage"]
share = round(detection["ai_fraction"] * 100)
print(f"verdict: {detection['verdict']} (score {detection['score']:.2f})")
print(f"AI-written share: {share}% of {usage['characters']} characters")
print(f"language: {detection['language']}, billed units: {usage['billed_units']}")
for segment in detection["segments"]:
    start, end = segment["start"], segment["end"]
    excerpt = text[start : min(end, start + 80)]
    print(f"{start}-{end} score {segment['score']:.2f}: {excerpt}")
```

**快速入门示例的输出**

```text
verdict: mixed (score 0.97)
AI-written share: 50% of 889 characters
language: en, billed units: 1
267-495 score 0.97: In today's fast-paced logistics landscape, it is essential for businesses to pri
590-809 score 0.93: Moving forward, it is crucial to implement robust quality assurance measures and
```

`start` 和 `end` 按 Unicode 码点计数，而不是字节，也不是 UTF-16 码元。Python 字符串本身就按码点索引；Node.js 程序会先用 `Array.from` 拆分文本。参见[段落偏移量](https://fileverity.com/zh/docs/text-detections.md#offsets)。

> **注意。** 文本检测返回的是统计估计，不是作者身份的证明。请将其作为人工复核的参考依据之一，不要单独依据它作出影响个人的决定。

## 接收 Webhook 并校验签名

Webhook 会在检测完成时通知您的服务器，因此无需轮询。这个接收程序是一个最简 HTTP 服务器：Node.js 使用 `http` 模块，Python 使用 Flask。它会校验每次投递的签名，忽略已经处理过的投递，并输出事件。实现遵循 [Webhooks](https://fileverity.com/zh/docs/webhooks.md#signature) 页面的说明。

1. 让 3000 端口可通过公网 HTTPS 地址访问，例如通过隧道指向您的本机。端点 URL 必须使用 HTTPS 和默认端口 443，并指向公网地址。
2. 在控制台中打开 **Webhooks**，添加一个 URL 为 `https://your-host/webhooks/fileverity` 的端点，并选择其事件。复制以 `whsec_` 开头的签名密钥，它只显示一次。
3. 将程序保存为 `webhook-receiver.mjs` 或 `webhook_receiver.py`，并带上密钥启动：`FILEVERITY_WEBHOOK_SECRET=whsec_... node webhook-receiver.mjs` 或 `FILEVERITY_WEBHOOK_SECRET=whsec_... python webhook_receiver.py`。
4. 在该端点上选择**发送测试**。测试密钥不会发送事件，而**发送测试**会投递一个经过签名的示例事件。

**接收 Webhook: JavaScript**

```javascript
import { createHmac, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";

const secret = process.env.FILEVERITY_WEBHOOK_SECRET;
if (!secret) throw new Error("Set FILEVERITY_WEBHOOK_SECRET to the endpoint's signing secret.");
const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
const port = Number(process.env.PORT) || 3000;
const seen = new Set();

function verify(headers, body) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures) return false;
  if (!(Math.abs(Date.now() / 1000 - Number(timestamp)) <= 300)) return false;
  const expected = createHmac("sha256", key).update(`${id}.${timestamp}.`).update(body).digest();
  return signatures.split(" ").some((entry) => {
    const [version, value = ""] = entry.split(",");
    const given = Buffer.from(value, "base64");
    return version === "v1" && given.length === expected.length && timingSafeEqual(given, expected);
  });
}

function handleEvent(event) {
  const { type, data } = event;
  if (data.test) {
    console.log(`Test event ${event.id}`);
  } else if (type === "analysis.completed") {
    const { level, action } = data.summary;
    console.log(`${data.id} succeeded: level ${level}, action ${action}, ${data.report_url}`);
  } else if (type === "analysis.failed") {
    console.log(`${data.id} ${data.status}: ${data.error.code}`);
  } else if (type === "balance.low") {
    const money = (cents) => `US$${(cents / 100).toFixed(2)}`;
    console.log(`Balance ${money(data.balance)} is below your ${money(data.threshold)} alert`);
  }
}

async function receive(req, res) {
  if (req.method !== "POST" || req.url !== "/webhooks/fileverity") return res.writeHead(404).end();
  const chunks = [];
  let size = 0;
  for await (const chunk of req) {
    size += chunk.length;
    if (size > 1_000_000) return res.writeHead(413).end();
    chunks.push(chunk);
  }
  const body = Buffer.concat(chunks);
  if (!verify(req.headers, body)) return res.writeHead(400).end();
  const event = JSON.parse(body.toString("utf8"));
  try {
    if (!seen.has(event.id)) {
      handleEvent(event);
      seen.add(event.id);
    }
  } catch (error) {
    console.error(`Could not handle ${event.id}`, error);
    return res.writeHead(500).end();
  }
  res.writeHead(200).end();
}

createServer((req, res) => receive(req, res).catch(() => res.destroy())).listen(port, () => console.log(`Listening on port ${port}`));
```

**接收 Webhook: Python**

```python
import base64
import hashlib
import hmac
import json
import os
import threading
import time

from flask import Flask, request

SECRET = os.environ.get("FILEVERITY_WEBHOOK_SECRET")
if not SECRET:
    raise SystemExit("Set FILEVERITY_WEBHOOK_SECRET to the endpoint's signing secret.")
KEY = base64.b64decode(SECRET.removeprefix("whsec_"))
seen = set()
lock = threading.Lock()
app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 1_000_000


def verify(headers, body):
    message_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp", "")
    signatures = headers.get("webhook-signature", "")
    if not message_id or not timestamp.isdecimal() or abs(time.time() - int(timestamp)) > 300:
        return False
    signed = f"{message_id}.{timestamp}.".encode() + body
    expected = hmac.new(KEY, signed, hashlib.sha256).digest()
    for entry in signatures.split(" "):
        version, _, value = entry.partition(",")
        try:
            given = base64.b64decode(value, validate=True)
        except ValueError:
            continue
        if version == "v1" and hmac.compare_digest(given, expected):
            return True
    return False


def handle_event(event):
    data = event["data"]
    if data.get("test"):
        print(f"Test event {event['id']}")
    elif event["type"] == "analysis.completed":
        level, action = data["summary"]["level"], data["summary"]["action"]
        print(f"{data['id']} succeeded: level {level}, action {action}, {data['report_url']}")
    elif event["type"] == "analysis.failed":
        print(f"{data['id']} {data['status']}: {data['error']['code']}")
    elif event["type"] == "balance.low":
        balance, threshold = data["balance"] / 100, data["threshold"] / 100
        print(f"Balance US${balance:.2f} is below your US${threshold:.2f} alert")


@app.post("/webhooks/fileverity")
def receive():
    body = request.get_data()
    if not verify(request.headers, body):
        return "", 400
    event = json.loads(body)
    with lock:
        try:
            if event["id"] not in seen:
                handle_event(event)
                seen.add(event["id"])
        except Exception:
            app.logger.exception("Could not handle %s", event["id"])
            return "", 500
    return "", 200


if __name__ == "__main__":
    port = int(os.environ.get("PORT", "3000"))
    print(f"Listening on port {port}")
    app.run(port=port)
```

**接收程序的输出**

```text
Listening on port 3000
Test event evt_Qm4vB7nP1sK8tL3wR6yZ2cXa
an_8Hk2pQ4vN9sL1xR7cW3mY6zB succeeded: level L2, action accept, https://fileverity.com/verify/FV-7K2M-9QXD-4T8R
```

- **原始请求体**：签名覆盖实际发送的原始字节。程序以字节形式读取请求体，使用经 Base64 解码的签名密钥作为 HMAC 密钥，对 `webhook-id`、一个英文句点、`webhook-timestamp`、一个英文句点和这些字节计算 HMAC-SHA256，并且只在签名有效后才解析 JSON。
- **恒定时间比较**：Node.js 中的 `timingSafeEqual` 和 Python 中的 `hmac.compare_digest` 会将结果与 `webhook-signature` 请求头中的每个 `v1` 签名进行比较。
- **时间戳容差**：`webhook-timestamp` 与系统时钟相差超过五分钟，或者不是数字，都会被拒绝。
- **快速响应**：有效的投递在事件处理完成后立即返回 `200`；签名无效则返回 `400`。如果处理失败，程序返回 `500`，Fileverity 稍后会使用相同的事件 ID 重试。参见[投递与重试](https://fileverity.com/zh/docs/webhooks.md#delivery)。
- **重复投递**：投递至少发生一次，重试沿用同一个事件 ID。程序会记住已处理的事件 ID，对重复事件直接返回 `200`，不会重复处理。在生产环境中，请将内存中的集合替换为数据库中事件 ID 的唯一约束，并把实际工作放入队列，以保持响应迅速。
- **事件数据**：`analysis.completed` 和 `analysis.failed` 携带不含 `result` 的分析对象；需要完整报告时，请调用 `GET /analyses/{id}`。`balance.low` 中的 `balance` 和 `threshold` 以美分为单位。

## 后续步骤

- 参照[受理流程中的文档检测](https://fileverity.com/zh/docs/intake-workflow.md)，运行包含人工复核的完整受理流程。
- 在 [API 参考](https://fileverity.com/zh/docs/api-reference.md)中查阅任意字段，或将 OpenAPI 文档导入您自己的工具。
- 正式上线时，使用 Live 密钥，并设置每月支出上限和 `balance.low` Webhook。参见[价格与计费](https://fileverity.com/zh/docs/pricing.md#alerts)。
- 需要帮助？[联系技术支持](https://fileverity.com/zh/contact)时，请提供 `x-request-id` 响应头的值。