跳到主要内容

参考代码 - Python

安装依赖:

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.ezfuns.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()