Python

本页使用 requests，不依赖专用 SDK。

安装

pip install requests

完整示例

from __future__ import annotations

import re
import time
import uuid
from pathlib import Path
from urllib.parse import unquote

import requests

API_KEY = "sk_your_key"
BASE = "https://omnidoc.top/api/v1"
API_HEADERS = {"X-API-Key": API_KEY}


def upload_document(file_path: Path) -> dict:
    size = file_path.stat().st_size
    response = requests.post(
        f"{BASE}/upload/presign",
        headers=API_HEADERS,
        json={
            "filename": file_path.name,
            "file_size": size,
        },
        timeout=20,
    )
    response.raise_for_status()
    ticket = response.json()

    with file_path.open("rb") as stream:
        put_headers = {**ticket["upload_headers"], "Content-Length": str(size)}
        uploaded = requests.put(
            ticket["upload_url"],
            headers=put_headers,
            data=stream,
            timeout=300,
        )
    uploaded.raise_for_status()

    finalized = requests.post(
        f"{BASE}/upload/finalize",
        headers=API_HEADERS,
        json={"finalize_token": ticket["finalize_token"]},
        timeout=30,
    )
    finalized.raise_for_status()
    return finalized.json()["document"]


def create_job(document_id: str, target_lang: str, engine: str) -> dict:
    response = requests.post(
        f"{BASE}/jobs",
        headers={**API_HEADERS, "Idempotency-Key": str(uuid.uuid4())},
        json={
            "document_id": document_id,
            "target_lang": target_lang,
            "engine": engine,
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()


def wait_for_job(job_id: str) -> dict:
    while True:
        response = requests.get(
            f"{BASE}/jobs/{job_id}", headers=API_HEADERS, timeout=20
        )
        response.raise_for_status()
        job = response.json()
        if job["status"] == "completed":
            return job
        if job["status"] == "failed":
            raise RuntimeError(job.get("error_msg") or "translation failed")
        print(f"{job.get('stage', 'processing')}: {job.get('stage_progress', 0)}%")
        time.sleep(3)


def response_filename(response: requests.Response, fallback: str) -> str:
    disposition = response.headers.get("Content-Disposition", "")
    utf8 = re.search(r"filename\*=UTF-8''([^;]+)", disposition, re.I)
    if utf8:
        return Path(unquote(utf8.group(1))).name or fallback
    ascii_name = re.search(r'filename="([^"]+)"', disposition, re.I)
    if ascii_name:
        return Path(ascii_name.group(1)).name or fallback
    return fallback


def download_artifacts(job_id: str, output_dir: Path) -> list[Path]:
    output_dir.mkdir(parents=True, exist_ok=True)
    response = requests.get(
        f"{BASE}/artifacts",
        headers=API_HEADERS,
        params={"job_id": job_id},
        timeout=20,
    )
    response.raise_for_status()

    results: list[Path] = []
    for artifact in response.json():
        download = requests.get(
            f"{BASE}/artifacts/{artifact['id']}/download",
            headers=API_HEADERS,
            timeout=300,
        )
        download.raise_for_status()
        filename = response_filename(
            download, f"{artifact['id']}.{artifact['format']}"
        )
        destination = output_dir / filename
        destination.write_bytes(download.content)
        results.append(destination)
    return results


def translate_document(
    source: str,
    *,
    target_lang: str = "en",
    engine: str = "xr-397b",
) -> list[Path]:
    document = upload_document(Path(source))
    job = create_job(document["id"], target_lang, engine)
    wait_for_job(job["id"])
    return download_artifacts(job["id"], Path("omnidoc-output"))


if __name__ == "__main__":
    for result in translate_document("report.docx", target_lang="en"):
        print(result)

生产建议

• 连接、读取和总任务等待分别设置超时，不用一个无限超时覆盖全部阶段
• 上传 PUT 失败时重新申请上传授权；过期地址立即丢弃
• 每次新提交生成新的 Idempotency-Key，重试同一次提交时沿用原来的键
• 429 按 Retry-After 有界退避
• 大产物使用流式下载，避免一次性把全部字节放进内存