Image playground

import json
import requests

api_url = "https://v2.api.piktid.com"
access_token = "your_access_token"
job_id = "job_abc123..."

headers = {"Authorization": "Bearer " + access_token}

with requests.get(
    api_url + "/notifications/events",
    headers=headers,
    stream=True,
    timeout=600,
) as response:
    response.raise_for_status()

    for raw_line in response.iter_lines(decode_unicode=True):
        if not raw_line or raw_line.startswith(":"):
            continue
        if not raw_line.startswith("data: "):
            continue

        notification = json.loads(raw_line[6:])
        if notification["name"] != "image_result":
            continue

        data = notification.get("data", {})
        if data.get("id_task") != job_id:
            continue

        status = data.get("status")
        if status == "completed":
            print(f"Image ready: {data['output_link']}")
            break
        if status == "failed":
            raise RuntimeError(data.get("error_message", "Generation failed"))

import hashlib
import hmac

import requests
from flask import Flask, abort, request

api_url = "https://v2.api.piktid.com"
access_token = "your_access_token"
job_id = "job_abc123..."
public_webhook_url = "https://example.com/webhooks/piktid"

setup = requests.put(
    api_url + "/webhooks",
    headers={"Authorization": "Bearer " + access_token},
    json={"url": public_webhook_url},
).json()
webhook_secret = setup["secret"]

app = Flask(__name__)


def verify_signature(secret: str, body: bytes, signature_header: str) -> bool:
    expected = "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)


@app.post("/webhooks/piktid")
def handle_piktid_webhook():
    body = request.get_data()
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_signature(webhook_secret, body, signature):
        abort(401)

    notification = request.get_json()
    if notification["name"] == "image_result":
        data = notification.get("data", {})
        if data.get("id_task") == job_id and data.get("status") == "completed":
            print(f"Image ready: {data['output_link']}")

    return "", 204


app.run(port=8000)