事件订阅接入指南

生图流程

生图主要包括三个阶段：

1. 页面渲染

2. 任务提交

3. 任务执行 & 结果展示

事件及回调流程

• sdImgGenControlConfig： 可用于展示自定义文案， 根据回调请求的token获取自有系统用户信息， 展示自由系统信息，比如用户额度。

• apiAccessPreInvoke：可用于自有系统的用户额度的实时扣减。

• apiAccessCommit：可用于自有系统的用户额度扣减状态维护。

• apiAccessRollback：可用于自由系统的用户额度扣减的回滚。

• sdTaskFinished：可用于自由系统的用户生图结果存储。

回调协议

回调请求通过HTTP协议发起，使用POST请求， 默认超时时间为5s， 请保证5s内响应。

回调类型

• 同步回调： 会阻塞平台下一步流程， 不会进行重试

• 异步回调： 不会阻塞平台下一步流程， 如响应状态码为4xx或者5xx系列， 会触发重试

接收回调

回调请求包括两部分

• 请求上下文：请求发起人、请求目标接口、事件类型等，在url参数中。

• 请求详情：用户发送到平台的业务参数内容，在请求体中，请求体可能为空

请求上下文参数

响应回调

状态码

• 2xx：表示请求已成功接收处理， 回调完成。

• 4xx/5xx: 表示请求未成功接收处理，回调失败，平台会发起重试。重试间隔为10s/30s/1分钟/2分钟/3分钟/4分钟/5分钟/6分钟/7分钟/8分钟/9分钟/10分钟/20分钟/30分钟/1小时/2小时。

事件请求参数、响应格式

apiAccessPreInvoke

用户发起请求后， 发起预校验， 判断当前用户是否有足够的权益

• 请求时机： 用户请求实际执行前

• 数据内容： 用户提交的请求

• 请求类型： 同步

• 请求格式： 用户实际请求内容， 无统一格式

响应体格式

sdPreInvoke

SD生图请求预校验， 判断当前用户是否具有足够的权益

• 请求时机： 用户发起生图请求前

• 请求内容： 用户生图原始请求

• 请求类型： 同步

请求详情

• SD WebUI中，用户生成一张图，内部可能对应多次文生图/图生图请求。sdPreInvoke和apiAccessPreInvoke都是在请求前发起。apiAccessPreInvoke对应了每次后端接收到的子请求，入参为接口实际请求；sdPreInvoke对应SD WebUI用户发起的原始请求。

响应体

sdTaskFinished

生图子任务执行完成后， 通知生图结果

• 请求时机： 生图子任务完成后

• 请求内容： 用户生图结果

• 请求类型： 异步

请求详情

响应体

无，成功返回HTTP状态码200。如果返回500，系统会进行重试调用

sdJobFinished

生图任务执行完成后， 通知生图结果

• 请求时机： 生图任务完成后

• 请求内容： 用户生图结果

• 请求类型： 异步

请求详情

• SD WebUI中，用户生成一张图，内部可能对应多次文生图/图生图请求。sdTaskFinished和sdJobFinished都是在请求完成后发起。sdTaskFinished表示单个子任务的结束；sdJobFinished表示用户视角的任务结束，结果可以是由插件对多个生图结果进行后置处理后产生的，是用户实际看到的图片。

响应体

无，成功返回HTTP状态码200。如果返回500，系统会进行重试调用

sdImgGenControlConfig

生图提交区块控制

• 请求时机： 用户打开生图页时

• 请求内容： 当前请求上下文

• 请求类型： 同步

请求详情

响应体

apiAccessRollback

生图提交失败回调

• 请求时机： 用于生图请求提交失败时

• 数据内容： 当前请求上下文

• 请求类型： 同步

响应体格式

解密 Token、验证签名（以Python接入为例）

安装依赖

pip install pycryptodome

样例

import hmac
import hashlib
import base64
from Crypto.Hash import SHA256
from Crypto.Cipher import AES


def decrypt_api_token(
        encrypted_api_token: str,
        sk: str
):
    """
    解密 api token，返回解密结果
    :param encrypted_api_token:
    :param sk:
    :return:
    """

    sha256 = SHA256.new()
    sha256.update(sk.encode('utf-8'))
    key = sha256.digest()[:16]

    encrypted = base64.b64decode(encrypted_api_token)
    iv = encrypted[:16]

    cipher = AES.new(key, AES.MODE_CBC, iv)
    decrypted = cipher.decrypt(encrypted[16:])
    pad_size = decrypted[-1]
    decrypted = decrypted[:-pad_size]
    return decrypted.decode('utf-8')


def check_sign_equal(
        sign: str,
        ak: str,
        sk: str,
        nonce: str,
        timestamp: str,
        body: str = None,
        biz_type: str = None,
        api_token: str = None,
        api_id: str = None,
        invoke_id: str = None,
):
    """
    检查 signature 是否一致
    :param sign:
    :param ak:
    :param sk:
    :param nonce:
    :param timestamp:
    :param body:
    :param biz_type:
    :param api_token: 解密之后
    :param api_id:
    :param invoke_id:
    :return
    """

    params = [ak, nonce, body, timestamp]
    if biz_type and len(biz_type.strip()) > 0:
        params.append(api_token)
        params.append(biz_type)
        params.append(api_id)
        params.append(invoke_id)
    sign_str = "".join(param for param in params if params is not None)

    h = hmac.new(bytes(sk, 'utf-8'), bytes(sign_str, 'utf-8'), hashlib.sha256)
    expect_sign = base64.b64encode(h.digest()).decode('utf-8')
    return sign == expect_sign