Python で書くパスポートリーダーは、最初は 20 行ほどで済みますが、本番環境で耐えられるようにすると 100 行ほどになります。増えた 80 行は OCR とは関係ありません。有料 API なら必ず出てくる 3 つの問いへの答えです。写真の質が悪いときにどうなるか、リクエストがタイムアウトして再送したときにどうなるか、そしてどの呼び出しに費用がかかったかを自分の記録でどう把握するか、です。
このチュートリアルでは、requests だけを使ってそのスクリプトを組み立てます。使うのは パスポート・身分証 OCR API の doc.cheap で、これは doc.cheap 自身のブログです。製品選びについてはその点を踏まえて判断してください。ここで紹介するパターン(画像ごとに 1 つの冪等性キー、安定したエラーコードでの分岐、呼び出しごとの課金フラグの保存)は、Python から呼び出すどの有料 API にも応用できます。
サンドボックスキーで 1 回リクエストする
ドキュメントには公開サンドボックスキー sk_sandbox_public が掲載されています。有料キーと同じ認識処理が動き、アカウントは不要です。IP アドレスごとに合計 10 件まで無料で認識でき、結果にかかわらず 1 時間あたり最大 10 リクエストです。無料パスポート OCR API のページには、同じ最初の呼び出しが curl 1 行で載っています。後でアカウントを作ると、カード登録なしで毎月 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 が自動で設定されます。画像は JPEG または PNG を base64 にしてボディに入れて送ります。呼び出しは同期的で、フィールドはこのレスポンスでそのまま返ってきます。ポーリングするジョブ ID も、用意すべき Webhook もありません。
テストには合成した見本を使い、自分のパスポートは決して使わないでください。ICAO の架空の「Utopia」文書や、多くの発行機関が公開している見本ページは、まさにこの用途のためにあります。
まず写真を縮小しましょう。 スマートフォンの写真は数メガバイトになることがあり、base64 にするとさらに 3 分の 1 増えます。パスポートガイド では、長辺を約 1600 px、JPEG 品質 85 にすることを推奨しています。上限を超えるボディは、処理が始まる前に payload_too_large で拒否されます。
レスポンスを読む
レスポンスのキーは常にすべて存在し、値が不明な場合は json() の後で None になります。キーが欠けることはありません。コードの次の動きを決めるのは次の 4 つです。
meta.statusは文書が読み取れたかどうかを示します。recognized、no_document_found、unreadable、unsupported_document、rejectedの 5 つの文字列のいずれかで、データを伴うのは最初のものだけです。documentとholderには整理済みの値が入ります。document.kind(passport、身分証など)、ISO 3166-1 alpha-3 の国コード、番号、ISOYYYY-MM-DD形式の日付、そしてholder.surname、holder.given_names、holder.birth_dateです。スキャンでそのグループの値が何も得られなかった場合、グループ全体がNoneになります。mrz.statusはpassed、failed、absentのいずれかで、機械読み取り領域が見つかったか、チェックディジットが一致したかを示します。mrz.textは読み取った領域そのものなので、チェックディジットを自分で検証できます。meta.billedは、この呼び出しが残高から引き落とされたかどうかを示します。
クライアント全体の形を決めるポイントはこれです。読み取れなかった写真は HTTP 200 であり、エラーではありません。 そのためコードは 2 回分岐します。拒否については HTTP のエラーコードで、結果については meta.status で分岐します。no_document_found で例外を投げると、リトライループは決して読めない写真を再送し続けます。成功として扱えば、フィールドが空の文書を保存してしまいます。
billed フラグ
本番キーでは、billed が True になるのは文書が実際に認識されたときだけです。何も見つからない、画像が読めない、未対応の種類、サービス側の障害、いずれも課金されません。課金される文書は量にかかわらず一律 $0.01 です。パスポート OCR API 比較 では、この価格を他サービスの公開価格と並べています。
サンドボックスでは一切課金されません。sk_sandbox_public でも billed は、同じスキャンが本番キーなら課金されたかどうかを示すので、テストする価値があります。このフラグを結果ごとに保存しておけば、1 か月分の billed 行を自分で合計するだけで済み、請求書との突き合わせは不要になります。
二重課金にならないリトライ
危険なのはタイムアウト後のリトライです。最初のリクエストがサーバーに届いたかどうかが分からず、有料 API では無闇にリトライすると 1 枚の画像に 2 回支払うことになりかねません。
解決策は、POST /v1/scans に付ける Idempotency-Key ヘッダー(1〜255 文字)です。画像ごとに 1 回だけ生成し、すべての試行で同じ値を送ります。本番キーでは、同じキーと同じボディでのリトライには 2 回目の認識ではなく最初の結果が返り、再送に費用はかかりません。リファレンスにある 3 つのルールがコードの形を決めます。
- キーはボディに紐づきます。 同じキーで別の画像や別のオプションを送ると
409 idempotency_conflictになります。ボディはループの外で 1 回だけ組み立てましょう。 - 最初の試行がまだ処理中のうちに 2 回目が届くことがあります。 その場合は
409 idempotency_in_progressです。待ってから同じキーでリトライします。 - 保存された結果がなければ再送はできません。
retain_hours: 0では何も保存されないため、24 時間以内に同じキーでリトライすると、2 回目の処理ではなく409 idempotency_replay_unavailableが返ります。何も残さないことと結果を再送することは両立しないので、用途ごとに選んでください。
サンドボックスキーもこのヘッダーを受け付けますが、何も課金されないため意味を持ちません。それでもこのコードパスはテストしておく価値があります。
リトライすべきエラー。 エラーボディはすべて同じ形(code、message、docs_url、request_id、event_id)をしており、分岐にはコードを使います。1 つの 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 は整数の秒数です。サンドボックスの時間あたりの制限では 1 時間近く待つよう求められることがあり、関数呼び出し 1 回がそんなに長くスリープするのは誰も望みません。そこで以下のクライアントは、待ち時間が 1 分を超える場合はあきらめます。
クライアント全体
# 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()), # 画像ごとに 1 つのキーを、すべての試行で再利用
}
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()を使うとその情報が失われます。- ボディとキーはループの前に 1 回だけ組み立てます。 これにより、サーバーから見てすべての試行が同じリクエストになります。
Sessionは接続を再利用します。 リトライの間でも、複数画像のバッチ処理の間でも同じです。多くのファイルをスキャンするときは、セッションを渡してください。return_portrait: Falseにすると、所持者の顔写真の切り抜きが省かれます。ページの切り抜きとフィールドは受け取れ、ログやストレージに残る顔が 1 つ減ります。referenceはmeta.reference(最大 128 文字)として返ってきます。スキャンを自分の注文やユーザーと結びつける簡単な方法です。身分証の表と裏は 2 回の呼び出しになるので、同じ reference を付けてください。- リクエストボディは決してログに残さないでください。 それは身分証明書そのものです。ログには
code、request_id、docs_urlを残しましょう。サポートに必要なのはそれだけです。
保持するデータを減らす
アップロードされた画像はリクエストの間だけメモリに保持され、永続ストレージには一切書き込まれません。結果は GET /v1/scans/{id} で再取得できるよう、指定した期間だけ保持されます。リクエストごとに options.retain_hours に 0〜8760 を指定でき、0 なら何も保存しません。JSON が 1 回だけ必要なら retain_hours: 0 を送り、前述の再送とのトレードオフを受け入れてください。処理は EU 内で行われます。
本番に移行する
DOC_CHEAP_API_KEY に自分のキーを設定するだけで、他は何も変わりません。同じエンドポイント、同じレスポンス形式、同じクライアントです。登録済みキーでは 1 分あたり 60 リクエストまで送れるため、Retry-After を守るバッチジョブなら独自のレートリミッターは不要です。クレジットは暗号資産(BTC、ETH、TRX、または Ethereum か Tron 上の USDT)で $1 から購入できます。現在、カード決済はありません。
レスポンスの中に Python から扱いにくい点があれば、admin@doc.cheap までご連絡ください。