#!/bin/python #coding: utf-8 # +------------------------------------------------------------------- # | django-vue-lyadmin # +------------------------------------------------------------------- # | Author: lybbn # +------------------------------------------------------------------- # | QQ: 1042594286 # +------------------------------------------------------------------- # | Date: 2023-11-07 # +------------------------------------------------------------------- # 抖音电商*精选联盟*开放平台-抖音开放平台接口（资质管理-抖音应用-撮合类目） # -------------------------------------------------------------------- #官方文档：https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2329 # -------------------------------------------------------------------- import json import requests import logging from django.core.cache import cache from urllib import parse logger = logging.getLogger(__name__) class douyinUnion: def __init__(self, client_key: str, client_secret: str): self._client_key = client_key # 应用唯一标识 self._client_secret = client_secret # 应用密钥 self._gate_way = "https://open.douyinec.com" def _request(self, path: str, params: dict,access_token="",headers=None,isPath=True) -> json: """ 请求（联盟请求方法） :param path: 接口path路径 如： /alliance/kol/store/list/ :param params: 业务参数字典，示例：{'client_secret':'xxxx','aaa':'xxxx'} :param method: post 请求、 get请求 """ try: if isPath: url = self._gate_way + path else: url = path if not headers: headers = {} headers.update({'access-token': access_token}) 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'}) param_json = json.dumps(params) response = requests.post(url=url, data=param_json, headers=headers) if response.status_code != 200: logger.error('请求地址：{}错误，错误内容：{}'.format(url,response)) return None return json.loads(response.content) except Exception as e: logger.error('请求地址：{}错误，错误内容：{}'.format(url,e)) return None def _request2(self, url: str, params: dict, method: str = "post",headers=None,isJson=True) -> json: """ 请求(抖音开放平台请求方法) :param url: 接口url :param params: 业务参数字典，示例：{'client_secret':'xxxx','aaa':'xxxx'} :param method: post 请求、 get请求 """ try: if not headers: headers = {} if isJson: headers.update({'Content-Type': 'application/json'}) headers.update({'Accept': 'application/json'}) else: headers.update({'Content-Type': 'application/x-www-form-urlencoded'}) 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'}) if method == "post": if isJson: param_json = json.dumps(params) else: param_json = parse.urlencode(params) response = requests.post(url=url, data=param_json, headers=headers) else: response = requests.get(url=url, headers=headers) if response.status_code != 200: return None return json.loads(response.content) except Exception as e: logger.error('请求地址：{}报错,{}'.format(url,e)) return None def request(self, path: str, params: dict,access_token: str) -> json: """通用请求接口 :param path: 调用的API接口path地址，示例：'/alliance/kol/store/list/' :param params: 业务参数字典，示例：{'a':'123123123','b':'http://www.demo.com/demo.jpg','c':'123'} :param access_token: 用户access_token或者client_token """ return self._request(path=path, params=params,access_token=access_token) def create_client_token(self): """ 该接口用于获取接口调用的凭证 client_token client_token 用于不需要用户授权就可以调用的接口。 client_token 的有效时间为 2 个小时，重复获取 client_token 后会使上次的 client_token 失效（但有 5 分钟的缓冲时间，连续多次获取 client_token 只会保留最新的两个 client_token）。 """ url = "https://open.douyin.com/oauth/client_token/" param = { "grant_type":"client_credential", "client_key":self._client_key, "client_secret":self._client_secret, } res = self._request2(url=url,params=param,method="post") if res and "error_code" in res["data"] and res["data"]["error_code"] == 0: access_token = res.get('data').get("access_token") expires_in = res.get('data').get("expires_in") cache.set("douyinunion_client_token"+self._client_key, access_token,expires_in - 3000) return access_token logger.error("获取抖音client_token失败，client_key:%s,返回信息：%s"%(self._client_key,res)) return None def get_client_token(self): """ 获取client_token """ client_token = cache.get("douyinunion_client_token"+self._client_key) if client_token: return client_token return self.create_client_token() def get_buyin_id(self,access_token="",open_id=""): """ 支持获取达人的百应ID，需要用户授权。 """ path = "/alliance/kol/buyin_id/?open_id="+open_id return self._request(path=path,params={},access_token=access_token) def daren_storelist(self,access_token="",open_id="",pageinfo={"page":1,"limit":20}): """ 获取达人的 橱窗商品 列表(需分页) https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2320 """ path = "/alliance/kol/store/list/?open_id="+open_id param = { "page":int(pageinfo['page']), "page_size":int(pageinfo['limit']) } return self._request(path=path,params=param,access_token=access_token) def daren_storeadd(self,access_token="",open_id="",params={"products":[{"product_id":""}],"need_hide":False,"pick_extra":"","keep_picksource":True}): """ 添加商品(精选联盟商品)到达人橱窗，需要用户授权 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2319 """ path = "/alliance/store/add/?open_id="+open_id param = params return self._request(path=path,params=param,access_token=access_token) def daren_storeremove(self,access_token="",open_id="",params={"product_id":"","promotion_id":""}): """ 下架达人的 橱窗商品 ，需要用户授权 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2322 """ path = "/alliance/kol/store/remove/?open_id="+open_id param = params return self._request(path=path,params=param,access_token=access_token) def daren_reputation(self,access_token="",open_id=""): """ 用于查询达人的带货口碑和橱窗销量，需要用户授权 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2328 """ path = "/alliance/kol/reputation/?open_id="+open_id res = self._request(path=path,params={},access_token=access_token) if res and "error_code" in res["data"] and res["data"]["error_code"] == 0: return res["data"] return None def daren_orders(self,access_token="",open_id="",params={}): """ 查询达人联盟订单，需要达人授权。 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2172 """ path = "/alliance/kol/orders/?open_id="+open_id res = self._request(path=path,params=params,access_token=access_token) if res and "error_code" in res["data"] and res["data"]["error_code"] == 0: return res["data"] return None def daren_award_task_apply(self,access_token="",task_id=""): """ 达人报名团长赏金任务，需要达人授权 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=3570 """ path = "/alliance/colonel/award_task/apply/" params = {"task_id":task_id} res = self._request(path=path,params=params,access_token=access_token) if res and "err_no" in res and res["err_no"] == 0: return True logger.info("达人报名团长赏金任务task_id=%s,错误内容：%s"%(task_id,res)) return False def daren_award_task_detail(self,task_id=""): """ 获取赏金任务详情，不需要认证（也不需要token）（从赏金任务领取链接页面中获取的接口） """ url = "https://lianmengapi.snssdk.com/ecom/captain/mobile/task/detail?task_id="+task_id headers = {} 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'}) response = requests.get(url=url,headers=headers) if response.status_code != 200: return None res = json.loads(response.content) if res and "code" in res and res["code"] == 0: return res['data'] return None def daren_award_task_product(self,task_id="",product_info="",order_field=0,order_by=0,page=1,page_size=20): """ 赏金任务关联商品列表,不需要认证（也不需要token）（从赏金任务领取链接页面中获取的接口） product_info:搜索关键字（任务关联商品搜索关键字） order_field:排序方式 0按销量 、 1按价格、2按佣金率 """ url = "https://lianmengapi.snssdk.com/ecom/captain/mobile/task/promotion/product/list?product_info="+product_info+"&task_id="+task_id+"&page="+page+"&page_size="+page_size+"&order_field="+order_field+"&order_by="+order_by headers = {} 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'}) response = requests.get(url=url,headers=headers) if response.status_code != 200: return None res = json.loads(response.content) if res and "code" in res and res["code"] == 0: return res['data'] return None def product_category(self,parent_id=0): """ 返回商品类目列表：类目层级、类目名称、类目ID（无需授权，但需要client_token） 参数：parent_id为父类的id，0表示一级类目 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2326 """ path = "/alliance/materials/product/category/" param = {"parent_id":parent_id} client_token = self.get_client_token() return self._request(path=path,params=param,access_token=client_token) def product_list_search(self,access_token="",open_id="",params={}): """ 检索小店联盟商品。可根据筛选条件过滤出满足条件的商品，也可根据排序条件对检索出来的商品进行排序。需要用户授权 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2324 """ path = "/alliance/kol/materials/products/search/?open_id="+open_id return self._request(path=path,params=params,access_token=access_token) def product_detail(self,access_token="",open_id="",params={}): """ 查询某个/批商品的详情。需要用户授权 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2325 """ path = "/alliance/kol/materials/products/details/?open_id="+open_id return self._request(path=path,params=params,access_token=access_token) def product_sku(self,product_id=""): """ 查询商品 SKU 信息 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2338 """ path = "/alliance/product/skus/" param = {"product_id":product_id} client_token = self.get_client_token() return self._request(path=path,params=param,access_token=client_token) def product_picksource_convert(self,product_url="",pick_extra=""): """ 三方选品平台商品转链。支持自定义入参，可用于多业务团队业绩归因。在抖音、百应搜索转链链接加橱窗后产生的订单会归属为第三方选品来源订单。 https://buyin.jinritemai.com/dashboard/service-provider/doc-center?docId=2329 """ path = "/alliance/product/picksource/convert/" param = {"product_url":product_url,"pick_extra":pick_extra} client_token = self.get_client_token() return self._request(path=path,params=param,access_token=client_token)