参考代码 - Python

安装依赖:

• https://github.com/jpadilla/pyjwt
• https://github.com/psf/requests
• https://github.com/benjaminp/six

pip install pyjwt requests six

示例代码:

import logging
import secrets
import time

import jwt
import requests
import six

LOG = logging.getLogger(__name__)

_TOKEN_ALGORITHM = "HS256"
_TOKEN_FIELDS = ["exp", "nonce"]
_TOKEN_EXPIRES_IN = 30 * 60

_BASE_URL = 'https://revenue.ezboti.com/api/v1/server'


class EzrevenueClient:
    def __init__(
        self,
        *,
        project_id: str,
        project_secret: str,
        base_url: str = None,
    ) -> None:
        self.project_id = project_id
        self.project_secret = project_secret
        self.base_url = base_url or _BASE_URL

    def decode_token(self, token: str) -> dict:
        try:
            payload = jwt.decode(
                token,
                key=self.project_secret,
                options={'require': _TOKEN_FIELDS},
                algorithms=[_TOKEN_ALGORITHM],
            )
        except jwt.InvalidTokenError as ex:
            raise ex
        return payload

    def encode_token(self, payload: dict) -> str:
        exp = int(time.time()) + int(_TOKEN_EXPIRES_IN)
        nonce = secrets.token_urlsafe(10)
        payload = dict(payload, exp=exp, nonce=nonce)
        token = jwt.encode(
            payload,
            key=self.project_secret,
            algorithm=_TOKEN_ALGORITHM,
            headers=dict(project_id=self.project_id),
        )
        # Note: pyjwt 2.0+ 返回 str 类型, 1.0+ 返回 bytes 类型
        return six.ensure_str(token)

    def call(self, api: str, params: dict):
        payload = {"method": api, "params": params}
        content = self.encode_token(payload)
        url = self.base_url + '/' + api
        response = requests.post(url, data=content)
        try:
            response.raise_for_status()
        except requests.HTTPError:
            err_msg = '%s failed status=%s, body=%s'
            LOG.warning(err_msg, api, response.status_code, response.text)
            raise
        result = self.decode_token(response.text)
        return result['result']


def main():
    client = EzrevenueClient(
        project_id='project-id',          # 艺爪付费项目ID
        project_secret='project-secret',  # 艺爪付费项目密钥
    )
    # 接口名称和参数，可参考具体API页面的说明
    result = client.call(
        'customer.info',  # 接口名称
        {
            "paywall_id": "xxx",          # 付费界面ID，ID和别名传一个即可
            "paywall_alias": "",          # 付费界面别名，ID和别名传一个即可
            "customer": {
                "external_id": 'test-user-id',  # 商户系统用户ID
                "nickname": None,               # 商户系统用户用户名/昵称，可选
                "external_dt_created": None,    # 商户系统用户创建时间，可选
            },
            "include_balance": True,            # 是否返回用户余额
        },
    )
    print(result)


if __name__ == '__main__':
    main()