229 lines
12 KiB
Python
229 lines
12 KiB
Python
#!/bin/python
|
||
#coding: utf-8
|
||
# +-------------------------------------------------------------------
|
||
# | django-vue-lyadmin 专业版
|
||
# +-------------------------------------------------------------------
|
||
# | Author: lybbn
|
||
# +-------------------------------------------------------------------
|
||
# | QQ: 1042594286
|
||
# +-------------------------------------------------------------------
|
||
# | Date: 2023-10-21
|
||
# +-------------------------------------------------------------------
|
||
# | version: 1.0
|
||
# +-------------------------------------------------------------------
|
||
# 抖音开放平台接口
|
||
# ------------------------------
|
||
import json
|
||
import requests
|
||
import logging
|
||
from urllib import parse
|
||
from django.core.cache import cache
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class douyin:
|
||
def __init__(self, client_key: str, client_secret: str):
|
||
self._client_key = client_key # 应用唯一标识
|
||
self._client_secret = client_secret # 应用密钥
|
||
self._gate_way = "https://open.douyin.com/"
|
||
|
||
def _request(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 create_code(self,scope,redirect_uri="",optionalScope="",state="state") -> None:
|
||
"""
|
||
生成授权链接,获取授权码（打开该url后，页面会出现一个二维码，用户扫描该二维码即可授权,在抖音 APP 支持端内唤醒的版本内打开的话会弹出客户端原生授权页面）
|
||
生成的url链接给前端用户pc浏览器访问，或ur链接地址转换成二维码用抖音app扫码(注意该URL不是用来请求的, 需要展示给用户用于扫码)
|
||
用户允许授权后，将会重定向到redirect_uri的网址上，并且带上code和state参数
|
||
scope: 应用授权作用域,多个授权作用域以英文逗号（,）分隔
|
||
state: 用于保持请求和回调的状态
|
||
redirect_uri:需要配置再抖音的授权回调白名单，否则报错,回调地址一定要采用https协议
|
||
"""
|
||
url = self._gate_way+"platform/oauth/connect/?client_key={}&response_type=code&scope={}&optionalScope={}&redirect_uri={}&state={}".format(self._client_key,scope,optionalScope,redirect_uri,state)
|
||
return url
|
||
|
||
def callback_code(self,data):
|
||
"""
|
||
验证处理抖音开放平台code授权回调返回的code信息
|
||
用户允许授权后，将会重定向到redirect_uri的网址上，并且带上code和state参数
|
||
使用apiview获取回调请求数据
|
||
"""
|
||
code = data.get("code")
|
||
state = data.get("state")
|
||
return code,state
|
||
|
||
def get_access_token(self,code):
|
||
"""
|
||
获取access_token（过期时间 15 天）
|
||
当 access_token 过期（过期时间 15 天）后，可以通过该接口使用 refresh_token（过期时间 30 天）进行刷新。刷新后获得一个有效期为15天的 access_token，但是 refresh_token 的有效期保持不变。
|
||
若 refresh_token 过期，获取 access_token 会报错（error_code=10010），此时需要重新引导用户授权。
|
||
注意：获取到 access_token 后授权临时票据 (code) 不要再授权刷新，否则会导致上一次获取的 code 过期。
|
||
"""
|
||
url = self._gate_way+"oauth/access_token/"
|
||
param = {
|
||
"client_secret":self._client_secret,
|
||
"code":code,
|
||
"grant_type":"authorization_code",
|
||
"client_key":self._client_key
|
||
}
|
||
return self._request(url=url,params=param,method="post")
|
||
|
||
def refresh_refresh_token(self,refresh_token):
|
||
"""
|
||
刷新 refresh_token（过期时间 30 天）
|
||
该接口用于刷新 refresh_token 的有效期。使用前提:client_key 必须拥有 renew_refresh_token 权限。
|
||
刷新操作需要在 refresh_token 过期前进行。
|
||
通过旧的 refresh_token 获取新的 refresh_token，调用后旧 refresh_token 会失效，新 refresh_token 有 30 天有效期。最多只能获取 5 次新的 refresh_token，5 次过后需要用户重新授权。
|
||
"""
|
||
url = self._gate_way+"oauth/renew_refresh_token/"
|
||
param = {
|
||
"refresh_token":refresh_token,
|
||
"client_key":self._client_key
|
||
}
|
||
return self._request(url=url,params=param,method="post",isJson=False)
|
||
|
||
def create_client_token(self):
|
||
"""
|
||
该接口用于获取接口调用的凭证 client_token
|
||
client_token 用于不需要用户授权就可以调用的接口。
|
||
client_token 的有效时间为 2 个小时，重复获取 client_token 后会使上次的 client_token 失效（但有 5 分钟的缓冲时间，连续多次获取 client_token 只会保留最新的两个 client_token）。
|
||
"""
|
||
url = self._gate_way+"oauth/client_token/"
|
||
param = {
|
||
"grant_type":"client_credential",
|
||
"client_key":self._client_key,
|
||
"client_secret":self._client_secret,
|
||
}
|
||
res = self._request(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("douyin_client_token"+self._client_key, access_token,expires_in - 3000)
|
||
return access_token
|
||
return None
|
||
|
||
|
||
def get_client_token(self):
|
||
"""
|
||
获取client_token
|
||
"""
|
||
client_token = cache.get("douyin_client_token"+self._client_key)
|
||
if client_token:
|
||
return client_token
|
||
return self.create_client_token()
|
||
|
||
def refresh_access_token(self,refresh_token):
|
||
"""
|
||
该接口用于刷新 access_token 的有效期；
|
||
当 access_token 过期（过期时间 15 天）后，可以通过该接口使用 refresh_token（过期时间 30 天）进行刷新。刷新后获得一个有效期为15天的 access_token，但是 refresh_token 的有效期保持不变。
|
||
若 refresh_token 过期，获取 access_token 会报错（error_code=10010），此时需要重新引导用户授权。
|
||
用户可以在抖音-我-设置（右上角）-帐号与安全-授权管理 中取消对应用的授权，取消授权后原有 access_token 会立即失效。
|
||
抖音开放平台会定期对用户授权进行检查，取消不合规的 access_token 授权。
|
||
"""
|
||
url = self._gate_way+"oauth/refresh_token/"
|
||
param = {
|
||
"grant_type":"refresh_token",
|
||
"client_key":self._client_key,
|
||
"refresh_token":refresh_token,
|
||
}
|
||
return self._request(url=url,params=param,method="post",isJson=False)
|
||
|
||
def get_userinfo(self,access_token,open_id):
|
||
"""
|
||
该接口获取用户的抖音公开信息，包含昵称和头像，适用于抖音、抖音极速版
|
||
若需要获取用户手机号，需要用户额外授权 mobile_alert 权限，本接口会额外返回 encrypt_mobile 字段
|
||
open_id:通过 /oauth/access_token/ 获取的，用户唯一标识
|
||
"""
|
||
url = self._gate_way+"oauth/userinfo/"
|
||
params = {
|
||
"access_token":access_token,
|
||
"open_id":open_id
|
||
}
|
||
return self._request(url=url,params=params,method="post",isJson=False)
|
||
|
||
def get_usermobile(self):
|
||
"""
|
||
该接口在用户公开信息的基础上，额外获取用户的手机号。
|
||
需要抖音开放平台申请权限。路径：抖音开放平台控制台 > 应用详情 > 能力管理 > 用户权限 > 获取用户手机号
|
||
注意事项：
|
||
1、用户授权后，获取用户公开信息接口会额外返回 encrypt_mobile 字段。
|
||
2、解密手机号，使用 AES 算法解密，秘钥是 client_secret， 向量 lv 是 client_secret 前 16 字节。
|
||
"""
|
||
pass
|
||
|
||
def get_userfans(self,open_id="",date_type=7):
|
||
"""
|
||
获取用户粉丝数
|
||
open_id:用户openid
|
||
date_type：数字型 近7/15天；输入7代表7天、15代表15天、30代表30天
|
||
https://developer.open-douyin.com/docs/resource/zh-CN/dop/develop/openapi/data-open-service/user-data/get-user-fans-count
|
||
"""
|
||
url = self._gate_way+"data/external/user/fans/"
|
||
params = {
|
||
"date_type":date_type,
|
||
"open_id":open_id
|
||
}
|
||
res = self._request(url=url,params=params,method="post",isJson=True)
|
||
if res and "error_code" in res["data"] and res["data"]["error_code"] == 0:
|
||
return res["data"]["result_list"]
|
||
return None
|
||
|
||
def get_billboard_live(self):
|
||
"""
|
||
获取直播榜数据,不需要用户授权(使用client_token)
|
||
https://developer.open-douyin.com/docs/resource/zh-CN/dop/develop/openapi/data-open-service/tops-data/live-list-data/live-list
|
||
"""
|
||
url = self._gate_way+"data/extern/billboard/live/"
|
||
client_token = self.get_access_token()
|
||
headers = {}
|
||
headers.update({'Content-Type': 'application/json'})
|
||
headers.update({'access-token': client_token})
|
||
res = self._request(url=url,method="get",headers=headers)
|
||
if res and "error_code" in res["data"] and res["data"]["error_code"] == 0:
|
||
return res["data"]["list"]
|
||
return None
|
||
|
||
def video_search(self,access_token,open_id,cursor=0,count=10,keyword=""):
|
||
"""
|
||
通过关键词搜索全站视频,类似抖音端上搜索。使用前请到 管理中心-应用详情-关键词视频管理-关键词管理 创建关键词
|
||
该接口只返回最近 1 天的视频
|
||
cursor:分页游标, 第一页请求cursor是0, response中会返回下一页请求用到的cursor, 同时response还会返回has_more来表明是否有更多的数据。
|
||
count:每页数量
|
||
keyword:关键词
|
||
https://developer.open-douyin.com/docs/resource/zh-CN/dop/develop/openapi/search-management/keywords-video-list/keywords-video
|
||
"""
|
||
url = self._gate_way+"video/search/?open_id="+open_id+"&cursor="+cursor+"&count="+count+"&keyword="+keyword
|
||
headers = {}
|
||
headers.update({'Content-Type': 'application/json'})
|
||
headers.update({'access-token': access_token})
|
||
res = self._request(url=url,method="get",headers=headers)
|
||
if res and "error_code" in res["data"] and res["data"]["error_code"] == 0:
|
||
return res["data"]
|
||
return None |