เรียก API ตรวจสลิปจาก Python ด้วย requests

Python เหมาะกับงานทั้งสคริปต์หลังบ้าน บอท และเว็บเฟรมเวิร์กอย่าง Flask หรือ Django การเรียก API ตรวจสลิปด้วยไลบรารี requests ทำได้สั้นและอ่านง่าย บทความนี้ใช้ตัวอย่างพื้นฐานจากคู่มือ API ของ PPay แล้วต่อยอดเป็นฟังก์ชันที่จัดการผลลัพธ์และข้อผิดพลาดได้ครบ เพื่อให้นำไปใช้ได้จริง

เตรียมการ

ติดตั้ง requests ด้วย pip เก็บ API key (ขึ้นต้น ppk_) ไว้ใน environment variable เช่น PPAY_API_KEY และเก็บ Base URL ตามคู่มือไว้ใน PPAY_API_BASE อย่าฝัง key ในโค้ดหรือ commit ลง Git การเรียกควรทำจากเซิร์ฟเวอร์หรือสคริปต์ของคุณเท่านั้น รูปต้องเป็น PNG หรือ JPG ขนาดไม่เกิน 5MB ใช้รูปต้นฉบับจากแอปธนาคาร เพราะรูปที่ถูกครอปหรือบีบอัดมากอาจอ่าน QR ไม่ได้

ตัวอย่างพื้นฐาน

import os, requests

r = requests.post(
    os.environ["PPAY_API_BASE"] + "/api/v1/verify",
    headers={"Authorization": "Bearer " + os.environ["PPAY_API_KEY"]},
    files={"file": ("slip.jpg", open("slip.jpg", "rb"), "image/jpeg")},
    timeout=30,
)
data = r.json()
if r.status_code == 202:
    job_id = data["jobId"]            # queued
elif r.ok and data["valid"]:
    print(data["slip"]["amount"])

พารามิเตอร์ files ทำให้ requests ส่งเป็น multipart/form-data ให้เอง ส่วน timeout=30 สำคัญมาก เพราะค่าเริ่มต้นของ requests คือไม่มี timeout หากเครือข่ายค้าง โปรแกรมของคุณจะค้างตามไปด้วย

ฟังก์ชันที่ใช้ซ้ำได้

ตัวอย่างนี้แยกผลลัพธ์เป็นกลุ่มชัดเจน และปิดไฟล์ให้ถูกต้องด้วย with

import os, requests

def verify_slip(path):
    url = os.environ["PPAY_API_BASE"] + "/api/v1/verify"
    headers = {"Authorization": "Bearer " + os.environ["PPAY_API_KEY"]}
    try:
        with open(path, "rb") as f:
            r = requests.post(url, headers=headers,
                              files={"file": ("slip.jpg", f, "image/jpeg")}, timeout=30)
    except requests.RequestException as e:
        return {"kind": "network-error", "message": str(e)}   # outcome unknown

    data = r.json()
    if r.status_code == 202:
        return {"kind": "queued", "job_id": data["jobId"]}
    if not r.ok:
        return {"kind": "request-error", "status": r.status_code, "code": data.get("code"),
                "retry_after": r.headers.get("Retry-After")}
    if data.get("valid"):
        return {"kind": "valid", "slip": data["slip"]}
    if data.get("result") == "error":
        return {"kind": "system-error", "code": data.get("code")}
    return {"kind": "rejected", "code": data.get("code")}

กรณี network-error หมายถึงไม่รู้ผล ไม่ควรส่งสลิปใบเดิมซ้ำทันที เพราะส่งซ้ำอาจได้ DUPLICATE_SLIP และถูกหักเพิ่ม ให้ตรวจประวัติการตรวจก่อน ดูหลักคิดที่ การ retry โดยไม่เกิดผลซ้ำ

ส่งเป็น base64 และแบบคิว

ถ้าระบบของคุณได้รูปเป็น base64 อยู่แล้ว เช่นจากบอทแชต ส่งเป็น JSON ที่ฟิลด์ image ได้เลย

import base64, os, requests

def verify_b64(raw_bytes):
    b64 = base64.b64encode(raw_bytes).decode("ascii")
    r = requests.post(
        os.environ["PPAY_API_BASE"] + "/api/v1/verify",
        headers={"Authorization": "Bearer " + os.environ["PPAY_API_KEY"]},
        json={"image": b64},
        timeout=30,
    )
    return r.json()

ส่วนการส่งเข้าคิวใช้ /api/v1/verify/async พร้อมฟิลด์ฟอร์ม reference เช่นเลขออเดอร์ (1-64 ตัวอักษร A-Z a-z 0-9 . _ : -) จะตอบ 202 พร้อม jobId แล้วรับผลทาง Webhook หรือดึงผลด้วย GET /api/v1/jobs/{jobId} โดยเว้น 2-5 วินาทีต่อครั้งหากต้อง poll

def queue_slip(path, order_no):
    with open(path, "rb") as f:
        r = requests.post(
            os.environ["PPAY_API_BASE"] + "/api/v1/verify/async",
            headers={"Authorization": "Bearer " + os.environ["PPAY_API_KEY"]},
            files={"file": ("slip.jpg", f, "image/jpeg")},
            data={"reference": order_no},
            timeout=30,
        )
    if r.status_code != 202:
        raise RuntimeError(r.json().get("code"))
    return r.json()["jobId"]

เทียบยอดก่อนอนุมัติ

ผล valid หมายถึงสลิปจริงและไม่ซ้ำ แต่ยังไม่ได้ตัดสินว่ายอดตรงออเดอร์หรือไม่ ให้ทำเองในโค้ดของคุณ

def can_approve(order, result):
    if result.get("kind") != "valid":
        return False
    amount = float(result["slip"]["amount"])
    return amount >= float(order["total"])   # also compare the receiver in your own logic

สำหรับเงินจริง ควรใช้ชนิดข้อมูลทศนิยมที่เหมาะสม (เช่น Decimal) ในระบบบัญชีของคุณ แทน float เพื่อเลี่ยงปัญหาปัดเศษ ตัวอย่างข้างบนเขียนแบบสั้นเพื่ออธิบายแนวคิดเท่านั้น ส่วนการอนุมัติควรอัปเดตสถานะเฉพาะออเดอร์ที่ยังรอชำระ เพื่อให้เรียกซ้ำแล้วไม่เกิดผลซ้ำ

รับ Webhook ด้วย Flask

เมื่อใช้แบบคิว ให้รับด้วย Flask โดยอ่านไบต์ดิบด้วย request.get_data() ก่อน parse แล้วคำนวณ HMAC-SHA256 ของ t + "." + raw ด้วยคีย์ whsec_ เทียบด้วย hmac.compare_digest และปฏิเสธถ้าเวลา t ต่างจากปัจจุบันเกิน 300 วินาที ตามตัวอย่างในคู่มือ แล้วตอบ 200 ก่อนทำงานต่อ รายละเอียดอยู่ที่ การตรวจสอบ webhook

เช็กลิสต์

1. ตั้ง timeout ทุกคำขอ
2. เก็บ key ใน environment variable
3. แยกกรณีเครือข่ายล้ม คำขอผิด ระบบขัดข้อง และสลิปไม่ผ่าน
4. รองรับ 202 ทุกครั้ง
5. เทียบยอดและผู้รับเองก่อนอนุมัติ
6. ไม่ retry อัตโนมัติเมื่อผล invalid
7. เก็บ id หรือ jobId ลง log

ดูรายละเอียดทุก endpoint ที่ คู่มือ API และถ้าอยากเริ่มทดลอง สมัครสมาชิก แล้วสร้าง API key ได้ในพื้นที่สมาชิก