#!/bin/python #coding: utf-8 # +------------------------------------------------------------------- # | django-vue-lyadmin 专业版 # +------------------------------------------------------------------- # | Author: lybbn # +------------------------------------------------------------------- # | QQ: 1042594286 # +------------------------------------------------------------------- # | Date: 2023-10-21 # +------------------------------------------------------------------- # | version: 1.0 # +------------------------------------------------------------------- # 整理自：https://gitee.com/minibear2021/doudian/blob/master/doudian/core.py # ------------------------------ # 抖店开放平台接口 # ------------------------------ import json import requests import datetime import time from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.hashes import MD5, SHA256, Hash from cryptography.hazmat.primitives.hmac import HMAC from django.core.cache import cache import logging logger = logging.getLogger(__name__) class doudian: def __init__(self, app_key: str, app_secret: str, app_type="self", code=None, shop_id: str = None, token_file: str = None, proxy: str = None, test_mode=False): """ 1、初始化DouDian实例，自用型应用传入shop_id用于初始化access token（可选），工具型应用传入code换取access token（如初始化时未传入，可以在访问抖店API之前调用init_token(code)进行token的初始化。 2、精选联盟自研应用和普通电商后台自研应用换取token的方式不同（普通电商后台自研应用采用authorization_self形式），精选联盟自研应用使用授权code换取token，也即使用authorization_code授权形式 """ self._app_key = app_key # 应用key，长度19位数字字符串 self._app_secret = app_secret # 应用密钥 字符串 self._app_type = app_type # App类型，self=自用型应用, tool=工具型应用 # if self._app_type == "self" and not shop_id: # raise ValueError('shop_id is not assigned.') self._token_file = token_file # token缓存文件 self._shop_id = shop_id # 店铺ID，自用型应用必传。 self._code = code # 工具型应用需要传的授权码code self._proxy = proxy self._test_mode = test_mode if self._test_mode: self._gate_way = 'https://openapi-sandbox.jinritemai.com' else: self._gate_way = 'https://openapi-fxg.jinritemai.com' self._version = 2 self._token = None self._sign_method = 'hmac-sha256' if self._token_file: try: with open(self._token_file) as f: self._token = json.load(f) except Exception as e: logger.error('{}'.format(e)) # if self._app_type == "self": # self.init_token() # elif self._app_type == "tool" and code: # self.init_token(code) def getCurrentAppKey(self): return self._app_key def _sign(self, method: str, param_json: str, timestamp: str) -> str: param_pattern = 'app_key{}method{}param_json{}timestamp{}v{}'.format(self._app_key, method, param_json, timestamp, self._version) sign_pattern = '{}{}{}'.format(self._app_secret, param_pattern, self._app_secret) return self._hash_hmac(sign_pattern) def _hash_hmac(self, pattern: str) -> str: try: hmac = HMAC(key=self._app_secret.encode('UTF-8'), algorithm=SHA256(), backend=default_backend()) hmac.update(pattern.encode('UTF-8')) signature = hmac.finalize() return signature.hex() except Exception as e: return None def get_access_token(self): return self._access_token() def _access_token(self) -> str: cache_token = self.get_cache_access_token() if cache_token: return cache_token refresh_token = self.get_cache_refresh_token() if refresh_token: access_token = self.refresh_token() return access_token return None def init_token(self, code: str = '') -> bool: """初始化access token :param code: 工具型应用从授权url回调中获取到的code，自用型应用无需传入。 """ if self._app_type == "tool" and not code: raise ValueError('code is not assigned.') path = '/token/create' grant_type = 'authorization_self' if self._app_type == "self" else 'authorization_code' params = {} params.update({'code': code if code else ''}) params.update({'grant_type': grant_type}) if self._app_type == "self": if self._test_mode: params.update({'test_shop': '1'}) elif self._shop_id: params.update({'shop_id': self._shop_id}) # else: # raise ValueError('shop_id is not assigned.') result = self._request(path=path, params=params, token_request=True) if result and result.get('code') == 10000 and result.get('data'): logger.error("初始化token成功appkey=%s：返回内容：%s" % (self._app_key, result)) self._token = result.get('data') expires_in_int = result.get('data').get('expires_in') self._token.update({'expires_in': int(time.time()) + expires_in_int}) expires_times = expires_in_int - 3000 self.set_cache_access_token(access_token = result.get('data').get('access_token'),expires = expires_times) self.set_cache_buyin_id(buyin_id=result.get('data').get('authority_id'), expires=expires_times) #authority_id其实为百应ID self.set_cache_refresh_token(refresh_token = result.get('data').get('refresh_token')) if self._token_file: with open(self._token_file, mode='w') as f: f.write(json.dumps(self._token)) return True logger.error("初始化token失败appkey=%s：返回内容：%s"%(self._app_key,result)) return False # 获取机构各角色的信息，如百应ID、角色名称等 # https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2305 def getBuyinInstitutionInfo(self): path = '/buyin/institutionInfo' params = {} result = self.request(path=path, params=params) return result def set_cache_buyin_id(self,buyin_id="",expires = 7*86400): cache.set("doudian_buyin_id"+self._app_key, buyin_id,expires) def set_cache_access_token(self,access_token="",expires = 7*86400): cache.set("doudian_access_token"+self._app_key, access_token,expires) def set_cache_refresh_token(self,refresh_token="",expires = 14*86400): cache.set("doudian_refresh_token"+self._app_key, refresh_token,expires) def get_cache_access_token_ttl(self): expire =cache.ttl("doudian_access_token"+self._app_key) return expire def get_cache_refresh_access_token_ttl(self): expire =cache.ttl("doudian_refresh_token"+self._app_key) return expire def get_cache_buyin_id(self): buyin_id =cache.get("doudian_buyin_id"+self._app_key) if buyin_id: return buyin_id return None def get_cache_access_token(self): access_token =cache.get("doudian_access_token"+self._app_key) if access_token: return access_token return None def get_cache_refresh_token(self): refresh_token =cache.get("doudian_refresh_token"+self._app_key) if refresh_token: return refresh_token return None #刷新access_token和refresh_token def refresh_token(self) -> None: path = '/token/refresh' refresh_token = self.get_cache_refresh_token() if refresh_token: grant_type = 'refresh_token' params = {} params.update({'grant_type': grant_type}) params.update({'refresh_token': refresh_token}) result = self._request(path=path, params=params, token_request=True) if result and result.get('code') == 10000 and result.get('data'): new_access_token = result.get('data').get('access_token') new_refresh_token = result.get('data').get('refresh_token') self._token = result.get('data') expires_in_int = result.get('data').get('expires_in') self._token.update({'expires_in': int(time.time()) + expires_in_int}) expires_times = expires_in_int - 3000 self.set_cache_access_token(access_token=new_access_token, expires=expires_times) self.set_cache_buyin_id(buyin_id=result.get('data').get('authority_id'), expires=expires_times) #authority_id其实为百应ID self.set_cache_refresh_token(refresh_token=new_refresh_token) if self._token_file: with open(self._token_file, mode='w') as f: f.write(json.dumps(self._token)) # return result.get('data').get('access_token') return True logger.error("刷新token失败:appkey=%s,返回：%s"%(self._app_key,result)) return False return False # 判断token是否过期 def is_token_expire(self): if self._access_token(): return False return True def _request(self, path: str, params: dict, token_request: bool = False,c_access_token = None) -> json: """ c_access_token :用户侧access_token，默认使用系统的access_token """ try: headers = {} headers.update({'Content-Type': 'application/json'}) headers.update({'Accept': 'application/json'}) headers.update({'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36'}) timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') param_json = json.dumps(params, sort_keys=True, separators=(',', ':')) method = path[1:].replace('/', '.') sign = self._sign(method=method, param_json=param_json, timestamp=timestamp) if token_request: url = self._gate_way + '{}?app_key={}&method={}¶m_json={}×tamp={}&v={}&sign_method={}&sign={}'.format( path, self._app_key, method, param_json, timestamp, self._version, self._sign_method, sign) else: if c_access_token: access_token = c_access_token else: access_token = self._access_token() url = self._gate_way + '{}?app_key={}&method={}&access_token={}×tamp={}&v={}&sign_method={}&sign={}'.format( path, self._app_key, method, access_token, timestamp, self._version, self._sign_method, sign) response = requests.post(url=url, data=param_json, headers=headers, proxies=self._proxy) if response.status_code != 200: logger.error("抖店应用请求失败：请求地址:%s,错误内容：%s"%(path,response)) return None return json.loads(response.content) except Exception as e: logger.error("抖店应用请求失败：请求地址:%s,错误内容：%s"%(path,e)) return None def request(self, path: str, params: dict, access_token=None) -> json: """请求抖店API接口 :param path: 调用的API接口地址，示例：'/material/uploadImageSync' :param params: 业务参数字典，示例：{'folder_id':'123123123','url':'http://www.demo.com/demo.jpg','material_name':'demo.jpg'} """ return self._request(path=path, params=params,c_access_token=access_token) def callback(self, headers: dict, body: bytes) -> json: """验证处理消息推送服务收到信息 """ data: str = body.decode('UTF-8') if not data: return None if headers.get('app-id') != self._app_key: return None event_sign: str = headers.get('event-sign') if not event_sign: return None h = Hash(algorithm=MD5(), backend=default_backend()) h.update('{}{}{}'.format(self._app_key, data, self._app_secret).encode('UTF-8')) if h.finalize().hex() != event_sign: return None return json.loads(data) def build_auth_url(self,id="", state="state",type=1) -> str: """拼接授权URL，引导商家、机构/团长、抖客/达人点击完成授权 type：类型 1 商家、 2 机构/团长 3、抖客/达人 id： type 类型为 1 和2时 id为service_id , 为3时为app_id(未上架应用填app_key) """ if self._app_type == "tool": if type == 1: return 'https://fuwu.jinritemai.com/authorize?service_id={}&state={}'.format(id, state) if type == 2: return 'https://market.jinritemai.com/authorize?service_id={}&state={}'.format(id, state) if type == 3: if not id: id = self._app_key return 'https://buyin.jinritemai.com/dashboard/institution/through-power?app_id={}&state={}'.format(id, state) return 'https://fuwu.jinritemai.com/authorize?service_id={}&state={}'.format(id, state) else: return None