Python

pip install flask requests

import hashlib, hmac, os, threading, time
import requests
from flask import Flask, abort, request

BASE = "https://office.takkub.com/api/v1"
AUTH = {"Authorization": f"Bearer {os.environ['TAKKUB_API_KEY']}"}
SECRET = os.environ["TAKKUB_WEBHOOK_SECRET"]
app = Flask(__name__)


def ask_ai(history):
    """เรียก AI ของคุณ รับประวัติ คืนข้อความตอบ"""
    raise NotImplementedError("ใส่โค้ดเรียก AI ตรงนี้")


def verify(raw, timestamp, signature):
    if abs(time.time() - int(timestamp or 0)) > 300:
        return False
    message = (timestamp or "").encode() + b"." + raw
    expected = "v1=" + hmac.new(SECRET.encode(), message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature or "")


def handle(event):
    thread_id = event["thread"]["id"]
    thread = requests.get(f"{BASE}/threads/{thread_id}", headers=AUTH, timeout=10).json()["data"]
    # คนในร้านกำลังดูแลอยู่ บอทไม่แทรก
    if thread["state"] == "working" or not thread["can_reply"]:
        return
    messages = requests.get(
        f"{BASE}/threads/{thread_id}/messages", params={"limit": 30}, headers=AUTH, timeout=10
    ).json()["data"]["messages"]
    history = [
        {"role": "user" if m["from"] == "customer" else "assistant", "content": m["text"]}
        for m in messages if m["text"]
    ]
    requests.post(
        f"{BASE}/threads/{thread_id}/messages",
        json={"text": ask_ai(history), "client_id": f"reply-{event['id']}"},
        headers=AUTH, timeout=10,
    )


@app.post("/takkub")
def webhook():
    raw = request.get_data()
    if not verify(raw, request.headers.get("X-Takkub-Timestamp"), request.headers.get("X-Takkub-Signature")):
        abort(401)
    event = request.get_json()
    # ตอบก่อน แล้วค่อยเรียก AI เพราะต้องตอบภายใน 5 วินาที
    if event["event"] == "message.received":
        threading.Thread(target=handle, args=(event,), daemon=True).start()
    return "", 200