用 Python 写一个护照识别程序,第一次大约二十行;要在生产环境里稳定运行,就会变成大约一百行。多出来的八十行与 OCR 无关,而是在回答每个付费 API 都会带来的三个问题:照片质量差时会怎样,请求超时后再发一次会怎样,以及你自己的记录如何知道哪些调用花了钱。

本教程只用 requests 来构建这个脚本。示例使用 doc.cheap,一个护照和身份证 OCR API,而这里是 doc.cheap 自己的博客,所以在产品选择上请自行权衡。文中的模式(每张图片一个幂等键、按稳定的错误码分支、为每次调用保存费用标志)适用于你从 Python 调用的任何付费 API。

用沙盒密钥发送一次请求

文档中公开了一个沙盒密钥 sk_sandbox_public。它运行的识别与付费密钥完全相同,而且无需账户:每个 IP 地址总共可免费识别 10 份文档,无论结果如何,每小时最多 10 个请求。免费护照 OCR API 页面用一条 curl 命令给出了同样的第一次调用。之后注册账户,每月还可额外获得 100 份免费文档,无需绑卡。

import base64
import requests

with open("specimen.jpg", "rb") as f:
    image = base64.b64encode(f.read()).decode("ascii")

response = requests.post(
    "https://api.doc.cheap/v1/scans",
    headers={"Authorization": "Bearer sk_sandbox_public"},
    json={"image": image},
    timeout=60,
)
scan = response.json()
print(response.status_code, scan["meta"]["status"], scan["meta"]["billed"])

json= 会自动为你设置 Content-Type: application/json。图片以 base64 形式放在请求体中,格式为 JPEG 或 PNG。调用是同步的:字段直接在这次响应中返回,不需要轮询任务 ID,也不需要部署 webhook。

请用合成的样本测试,切勿使用你自己的护照。ICAO 虚构的“Utopia”证件以及许多签发机构公布的样本页,正是为此而存在的。

先缩小照片。 手机拍的照片可能有好几兆,base64 还会再增加三分之一。护照指南建议长边约 1600 px、JPEG 质量 85;超过上限的请求体会在任何处理开始前以 payload_too_large 被拒绝。

读取响应

响应中的每个键始终存在,未知的值在 json() 之后是 None,绝不会缺少键。决定代码下一步的有四个部分:

  • meta.status 表示文档是否被读出。它是五个字符串之一:recognized、no_document_found、unreadable、unsupported_document、rejected。只有第一个带有数据。
  • document 和 holder 存放整理后的值:document.kind(passport、身份证等)、ISO 3166-1 alpha-3 国家代码、证件号码、ISO YYYY-MM-DD 格式的日期;以及 holder.surname、holder.given_names、holder.birth_date。若扫描没有为某组得到任何内容,整组为 None。
  • mrz.status 为 passed、failed 或 absent:表示是否找到了机读区,以及校验位是否一致。mrz.text 是读出的机读区原文,你可以自己验证校验位。
  • meta.billed 表示这次调用是否从余额中扣费。

决定整个客户端结构的一点是:无法读出的照片返回 HTTP 200,而不是错误。 因此代码要分支两次:对拒绝按 HTTP 错误码分支,对结果按 meta.status 分支。如果在 no_document_found 时抛出异常,重试循环就会不断重发一张永远读不出的照片;如果把它当作成功,你就会存下一份没有任何字段的文档。

billed 标志

在正式密钥上,只有文档真正被识别时 billed 才为 True。什么都没找到、图片无法读取、不支持的类型、服务端故障:这些都不收费。每份计费文档统一 $0.01,与用量无关;护照 OCR API 对比页面把这个价格与其他服务公布的价格放在一起。

沙盒完全不收费。在 sk_sandbox_public 上,billed 仍会告诉你同样的扫描在正式密钥上是否会被计费,这正是用它来测试的价值所在。把这个标志和每条结果一起保存:这样每月只需汇总自己的 billed 记录,无需与账单对账。

不会重复扣费的重试

有风险的重试是超时之后的重试。你不知道第一次请求是否已到达服务器,而在付费 API 上盲目重试,可能为同一张图片付两次钱。

解决办法是在 POST /v1/scans 上加 Idempotency-Key 请求头,长度 1 到 255 个字符。每张图片只生成一次,并在每次尝试中发送相同的值。在正式密钥上,使用相同键和相同请求体的重试会拿回第一次的结果,而不会再识别一次,重放也不收费。参考文档中的三条规则决定了代码的写法:

  • 键与请求体绑定。 同一个键配上不同的图片或不同的选项,会得到 409 idempotency_conflict。请在循环外只构建一次请求体。
  • 第二次尝试可能在第一次仍在运行时到达。 这时返回 409 idempotency_in_progress:等待后用同一个键重试。
  • 没有保存结果,就无法重放。 设置 retain_hours: 0 时什么都不保存,所以 24 小时内用同一个键重试会得到 409 idempotency_replay_unavailable,而不是再运行一次。什么都不保留与重放结果无法兼得,请按使用场景选择。

沙盒密钥也接受这个请求头,但由于不计费,它在那里不起任何作用。不过这条代码路径仍值得测试。

哪些错误值得重试。 所有错误响应体结构相同(code、message、docs_url、request_id、event_id),应当按错误码分支,因为同一个 HTTP 状态可能对应需要完全相反处理的错误码。错误处理指南将它们分为几类:

错误码 怎么做
rate_limited, document_repeated, internal_error, engine_unavailable, service_unavailable, maintenance 等待(遵守 Retry-After),然后重试
idempotency_in_progress 等待,用同一个键重试
validation_failed, invalid_request, payload_too_large, unsupported_media_type 修正请求;重试仍会失败
unauthorized, registration_required, insufficient_credits 修正密钥或账户;等待不会改变任何事

Retry-After 是整数秒。沙盒的每小时限额可能要求等待将近一小时,没有调用方希望一次函数调用睡这么久,所以下面的客户端在等待超过一分钟时会放弃。

完整客户端

# scan.py
import base64
import os
import time
import uuid

import requests

API = "https://api.doc.cheap/v1/scans"
KEY = os.environ.get("DOC_CHEAP_API_KEY", "sk_sandbox_public")
RETRY = {
    "rate_limited", "document_repeated", "internal_error", "engine_unavailable",
    "service_unavailable", "maintenance", "idempotency_in_progress",
}
MAX_WAIT = 60  # 秒;更长的 Retry-After 直接报告,不睡眠等待


class ScanError(Exception):
    def __init__(self, status, error):
        super().__init__(f"{error['code']} ({status}): {error['message']}")
        self.code = error["code"]
        self.docs_url = error["docs_url"]
        self.request_id = error["request_id"]


def retry_after(response, attempt):
    try:
        seconds = int(response.headers.get("Retry-After", ""))
    except ValueError:
        seconds = 0
    return seconds if seconds > 0 else 2 ** attempt


def scan(path, reference=None, attempts=4, session=None):
    session = session or requests.Session()
    with open(path, "rb") as f:
        image = base64.b64encode(f.read()).decode("ascii")
    body = {"image": image, "reference": reference, "options": {"return_portrait": False}}
    headers = {
        "Authorization": f"Bearer {KEY}",
        "Idempotency-Key": str(uuid.uuid4()),  # 每张图片一个键,每次尝试都复用
    }

    for attempt in range(1, attempts + 1):
        last = attempt == attempts
        try:
            response = session.post(API, headers=headers, json=body, timeout=60)
            payload = response.json()
        except (requests.RequestException, ValueError):
            # 超时、连接中断,或代理返回了 HTML。
            if last:
                raise
            time.sleep(2 ** attempt)
            continue

        if response.ok:
            return payload
        error = payload["error"]
        if error["code"] not in RETRY or last:
            raise ScanError(response.status_code, error)
        wait = retry_after(response, attempt)
        if wait > MAX_WAIT:
            raise ScanError(response.status_code, error)
        time.sleep(wait)


def summarise(scan):
    meta, document, holder, mrz = scan["meta"], scan["document"], scan["holder"], scan["mrz"]
    if meta["status"] != "recognized":
        return {"ok": False, "status": meta["status"], "billed": meta["billed"]}
    document, holder = document or {}, holder or {}
    return {
        "ok": True,
        "billed": meta["billed"],
        "kind": document.get("kind"),
        "country": document.get("country"),
        "expiry_date": document.get("expiry_date"),
        "surname": holder.get("surname"),
        "birth_date": holder.get("birth_date"),
        "mrz": mrz["status"],
    }


if __name__ == "__main__":
    import sys

    try:
        print(summarise(scan(sys.argv[1], reference="demo-1")))
    except ScanError as err:
        print(err, err.docs_url, err.request_id, file=sys.stderr)
        sys.exit(1)

用 python scan.py specimen.jpg 运行。有几个选择值得说明:

  • requests 遇到 4xx 或 5xx 不会抛异常,除非你调用 raise_for_status()。客户端无论哪种情况都读取 JSON 响应体,因为错误响应体里带着错误码;raise_for_status() 会把它丢掉。
  • 请求体和键在循环之前只构建一次。 这样在服务器看来,每次尝试都是同一个请求。
  • Session 会复用连接,无论是在重试之间,还是在批量处理多张图片时。扫描大量文件时请传入一个 session。
  • return_portrait: False 会省略持证人照片的裁剪图。你仍会得到页面裁剪图和各字段,日志和存储里也少了一张人脸。
  • reference 会以 meta.reference(最多 128 个字符)返回:这是把扫描与你自己的订单或用户关联起来的简单办法。身份证的正反两面是两次调用,请给它们相同的 reference。
  • 切勿记录请求体。 那是一份身份证件。请记录 code、request_id 和 docs_url,这就是客服所需的全部信息。

少保留数据

上传的图片只在请求期间保存在内存中,从不写入持久存储。结果会被保留,以便你通过 GET /v1/scans/{id} 再次获取,保留时长由你决定:每个请求的 options.retain_hours 可取 0 到 8760,0 表示什么都不保存。如果你只需要一次 JSON,就发送 retain_hours: 0,并接受上文所说的重放取舍。处理在欧盟境内进行。

正式上线

把 DOC_CHEAP_API_KEY 设置为你自己的密钥,其他什么都不用改:同样的端点、同样的响应结构、同样的客户端。注册密钥每分钟允许 60 个请求,因此遵守 Retry-After 的批处理任务不需要自己的限流器。积分用加密货币(BTC、ETH、TRX,或 Ethereum、Tron 上的 USDT)购买,$1 起;目前不支持银行卡支付。

如果响应中有什么在 Python 里不好处理,请写信至 admin@doc.cheap。