在业务系统里集成 PDF 翻译能力，最常见的需求不是"翻译一句话"，而是"上传一个 PDF，异步拿到翻译后的 PDF"。这篇文章用 Python + FastAPI 搭建一个最小可用的 PDF 翻译微服务，覆盖文件上传、任务队列、翻译调用、结果回调四个核心环节。

一、系统架构

整个服务分为三层：

• API 层：FastAPI 提供 /upload 和 /status/{task_id} 接口。
• 任务层：后台线程池处理翻译任务，避免阻塞请求。
• 存储层：本地目录保存上传文件和翻译结果。

Client → FastAPI → Task Queue → Translator → Callback/Status API

为了简化演示，翻译环节调用一个模拟的翻译接口。生产环境中可以替换为真实的 PDF 翻译 API 或本地模型。

二、环境准备

• Python 3.10+
• FastAPI + Uvicorn
• aiofiles（异步文件操作）
• requests（调用翻译 API）

pip install fastapi uvicorn aiofiles requests python-multipart

三、项目结构

pdf_translate_service/
├── main.py          # FastAPI 主服务
├── worker.py        # 后台翻译任务
├── storage/         # 上传文件和结果存储
└── callbacks/       # 回调日志

四、核心代码实现

1. 任务模型与内存状态

# main.py
import os
import uuid
import shutil
from datetime import datetime
from typing import Optional
from fastapi import FastAPI, File, UploadFile, BackgroundTasks
from pydantic import BaseModel
import worker

app = FastAPI(title="PDF Translation Microservice")

UPLOAD_DIR = "storage/uploads"
RESULT_DIR = "storage/results"
os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(RESULT_DIR, exist_ok=True)

# 内存中的任务状态，生产环境建议用 Redis
tasks = {}

class TaskStatus(BaseModel):
    task_id: str
    status: str  # pending / processing / completed / failed
    original: Optional[str] = None
    result: Optional[str] = None
    message: Optional[str] = None
    created_at: str
    updated_at: str

2. 文件上传接口

@app.post("/upload", response_model=TaskStatus)
async def upload_pdf(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...),
    callback_url: Optional[str] = None
):
    if not file.filename.endswith(".pdf"):
        return {"error": "Only PDF files are supported"}

    task_id = str(uuid.uuid4())
    upload_path = os.path.join(UPLOAD_DIR, f"{task_id}.pdf")

    with open(upload_path, "wb") as f:
        shutil.copyfileobj(file.file, f)

    now = datetime.utcnow().isoformat()
    tasks[task_id] = {
        "task_id": task_id,
        "status": "pending",
        "original": upload_path,
        "result": None,
        "message": "File uploaded, waiting for processing",
        "created_at": now,
        "updated_at": now,
    }

    background_tasks.add_task(worker.process_translation, task_id, upload_path, callback_url)
    return TaskStatus(**tasks[task_id])

3. 状态查询接口

@app.get("/status/{task_id}", response_model=TaskStatus)
async def get_status(task_id: str):
    if task_id not in tasks:
        return {"error": "Task not found"}
    return TaskStatus(**tasks[task_id])

4. 后台翻译任务 worker.py

# worker.py
import os
import time
import requests
from datetime import datetime

RESULT_DIR = "storage/results"

def process_translation(task_id: str, file_path: str, callback_url: str = None):
    """模拟异步翻译任务"""
    from main import tasks

    tasks[task_id]["status"] = "processing"
    tasks[task_id]["message"] = "Translating PDF..."
    tasks[task_id]["updated_at"] = datetime.utcnow().isoformat()

    try:
        # 模拟翻译耗时
        time.sleep(3)

        # 生产环境：替换为真实翻译 API
        # response = requests.post(
        #     "https://api.pdftranslator.org/v1/translate",
        #     files={"file": open(file_path, "rb")},
        #     data={"target_lang": "zh"}
        # )
        # response.raise_for_status()

        result_path = os.path.join(RESULT_DIR, f"{task_id}_translated.pdf")
        # 这里用原文件占位，真实场景保存翻译后的 PDF
        with open(file_path, "rb") as src, open(result_path, "wb") as dst:
            dst.write(src.read())

        tasks[task_id]["status"] = "completed"
        tasks[task_id]["result"] = result_path
        tasks[task_id]["message"] = "Translation completed"

        if callback_url:
            send_callback(callback_url, tasks[task_id])

    except Exception as e:
        tasks[task_id]["status"] = "failed"
        tasks[task_id]["message"] = str(e)

    tasks[task_id]["updated_at"] = datetime.utcnow().isoformat()


def send_callback(url: str, payload: dict):
    try:
        requests.post(url, json=payload, timeout=10)
    except requests.RequestException as e:
        print(f"Callback failed: {e}")

五、运行与测试

启动服务：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

上传 PDF：

curl -X POST "http://localhost:8000/upload" \
  -F "file=@sample.pdf" \
  -F "callback_url=http://your-callback-server.com/notify"

查询状态：

curl "http://localhost:8000/status/{task_id}"

六、生产环境建议

1. 任务队列：内存 tasks 仅适合演示，生产环境用 Celery + Redis 或 RQ。
2. 文件存储：上传到对象存储（S3/OSS），避免本地磁盘容量限制。
3. 并发控制：翻译 API 通常有 QPS 限制，用 Semaphore 或令牌桶限流。
4. 错误重试：网络抖动时自动重试，记录失败任务便于人工介入。
5. 回调可靠性：回调失败时持久化到数据库，支持手动补偿。

七、总结

这套 FastAPI 微服务的核心思路是把"翻译一个 PDF"拆成"接收文件 → 异步处理 → 查询/回调结果"三个标准化步骤。接口简单、易于集成，也便于后续替换不同的翻译后端。

如果你正在把 PDF 翻译能力接入自己的业务系统，这个骨架应该能帮你快速跑通第一个版本。

标签：FastAPI、Python、PDF翻译、微服务、AI翻译