mirror of
https://github.com/imgyh/tiktok.git
synced 2025-04-05 11:42:43 +08:00
399 lines
15 KiB
Python
399 lines
15 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
'''
|
|
@FileName : douyinapi.py
|
|
@Project : apiproxy
|
|
@Description:
|
|
@Author : imgyh
|
|
@Mail : admin@imgyh.com
|
|
@Github : https://github.com/imgyh
|
|
@Site : https://www.imgyh.com
|
|
@Date : 2023/5/20 22:13
|
|
@Version : v1.0
|
|
@ChangeLog
|
|
------------------------------------------------
|
|
|
|
------------------------------------------------
|
|
'''
|
|
|
|
import re
|
|
import requests
|
|
import json
|
|
import time
|
|
import copy
|
|
|
|
from apiproxy.douyin import douyin_headers
|
|
from apiproxy.douyin.urls import Urls
|
|
from apiproxy.douyin.result import Result
|
|
from apiproxy.common import utils
|
|
|
|
class DouyinApi(object):
|
|
def __init__(self):
|
|
self.urls = Urls()
|
|
self.result = Result()
|
|
# 用于设置重复请求某个接口的最大时间
|
|
self.timeout = 10
|
|
|
|
# 从分享链接中提取网址
|
|
def getShareLink(self, string):
|
|
# findall() 查找匹配正则表达式的字符串
|
|
return re.findall('http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', string)[0]
|
|
|
|
# 得到 作品id 或者 用户id
|
|
# 传入 url 支持 https://www.iesdouyin.com 与 https://v.douyin.com
|
|
def getKey(self, url):
|
|
key = None
|
|
key_type = None
|
|
|
|
try:
|
|
r = requests.get(url=url, headers=douyin_headers)
|
|
except Exception as e:
|
|
print('[ 错误 ]:输入链接有误！\r')
|
|
return key_type, key
|
|
|
|
# 抖音把图集更新为note
|
|
# 作品 第一步解析出来的链接是share/video/{aweme_id}
|
|
# https://www.iesdouyin.com/share/video/7037827546599263488/?region=CN&mid=6939809470193126152&u_code=j8a5173b&did=MS4wLjABAAAA1DICF9-A9M_CiGqAJZdsnig5TInVeIyPdc2QQdGrq58xUgD2w6BqCHovtqdIDs2i&iid=MS4wLjABAAAAomGWi4n2T0H9Ab9x96cUZoJXaILk4qXOJlJMZFiK6b_aJbuHkjN_f0mBzfy91DX1&with_sec_did=1&titleType=title&schema_type=37&from_ssr=1&utm_source=copy&utm_campaign=client_share&utm_medium=android&app=aweme
|
|
# 用户 第一步解析出来的链接是share/user/{sec_uid}
|
|
# https://www.iesdouyin.com/share/user/MS4wLjABAAAA06y3Ctu8QmuefqvUSU7vr0c_ZQnCqB0eaglgkelLTek?did=MS4wLjABAAAA1DICF9-A9M_CiGqAJZdsnig5TInVeIyPdc2QQdGrq58xUgD2w6BqCHovtqdIDs2i&iid=MS4wLjABAAAAomGWi4n2T0H9Ab9x96cUZoJXaILk4qXOJlJMZFiK6b_aJbuHkjN_f0mBzfy91DX1&with_sec_did=1&sec_uid=MS4wLjABAAAA06y3Ctu8QmuefqvUSU7vr0c_ZQnCqB0eaglgkelLTek&from_ssr=1&u_code=j8a5173b×tamp=1674540164&ecom_share_track_params=%7B%22is_ec_shopping%22%3A%221%22%2C%22secuid%22%3A%22MS4wLjABAAAA-jD2lukp--I21BF8VQsmYUqJDbj3FmU-kGQTHl2y1Cw%22%2C%22enter_from%22%3A%22others_homepage%22%2C%22share_previous_page%22%3A%22others_homepage%22%7D&utm_source=copy&utm_campaign=client_share&utm_medium=android&app=aweme
|
|
# 合集
|
|
# https://www.douyin.com/collection/7093490319085307918
|
|
urlstr = str(r.request.path_url)
|
|
|
|
if "/user/" in urlstr:
|
|
# 获取用户 sec_uid
|
|
if '?' in r.request.path_url:
|
|
for one in re.finditer(r'user\/([\d\D]*)([?])', str(r.request.path_url)):
|
|
key = one.group(1)
|
|
else:
|
|
for one in re.finditer(r'user\/([\d\D]*)', str(r.request.path_url)):
|
|
key = one.group(1)
|
|
key_type = "user"
|
|
elif "/video/" in urlstr:
|
|
# 获取作品 aweme_id
|
|
key = re.findall('video/(\d+)?', urlstr)[0]
|
|
key_type = "aweme"
|
|
elif "/note/" in urlstr:
|
|
# 获取note aweme_id
|
|
key = re.findall('note/(\d+)?', urlstr)[0]
|
|
key_type = "aweme"
|
|
elif "/mix/detail/" in urlstr:
|
|
# 获取合集 id
|
|
key = re.findall('/mix/detail/(\d+)?', urlstr)[0]
|
|
key_type = "mix"
|
|
elif "/collection/" in urlstr:
|
|
# 获取合集 id
|
|
key = re.findall('/collection/(\d+)?', urlstr)[0]
|
|
key_type = "mix"
|
|
elif "/music/" in urlstr:
|
|
# 获取原声 id
|
|
key = re.findall('music/(\d+)?', urlstr)[0]
|
|
key_type = "music"
|
|
elif "/webcast/reflow/" in urlstr:
|
|
key1 = re.findall('reflow/(\d+)?', urlstr)[0]
|
|
url = self.urls.LIVE2 + utils.getXbogus(
|
|
f'live_id=1&room_id={key1}&app_id=1128')
|
|
res = requests.get(url, headers=douyin_headers)
|
|
resjson = json.loads(res.text)
|
|
key = resjson['data']['room']['owner']['web_rid']
|
|
key_type = "live"
|
|
elif "live.douyin.com" in r.url:
|
|
key = r.url.replace('https://live.douyin.com/', '')
|
|
key_type = "live"
|
|
|
|
if key is None or key_type is None:
|
|
print('[ 错误 ]:输入链接有误！无法获取 id\r')
|
|
return key_type, key
|
|
|
|
return key_type, key
|
|
|
|
def getAwemeInfoApi(self, aweme_id):
|
|
if aweme_id is None:
|
|
return None
|
|
start = time.time() # 开始时间
|
|
while True:
|
|
try:
|
|
jx_url = self.urls.POST_DETAIL + utils.getXbogus(
|
|
f'aweme_id={aweme_id}&device_platform=webapp&aid=6383')
|
|
|
|
raw = requests.get(url=jx_url, headers=douyin_headers).text
|
|
datadict = json.loads(raw)
|
|
if datadict is not None and datadict["status_code"] == 0:
|
|
break
|
|
except Exception as e:
|
|
end = time.time() # 结束时间
|
|
if end - start > self.timeout:
|
|
return None
|
|
|
|
# 清空self.awemeDict
|
|
self.result.clearDict(self.result.awemeDict)
|
|
|
|
# 默认为视频
|
|
awemeType = 0
|
|
try:
|
|
if datadict['aweme_detail']["images"] is not None:
|
|
awemeType = 1
|
|
except Exception as e:
|
|
pass
|
|
|
|
# 转换成我们自己的格式
|
|
self.result.dataConvert(awemeType, self.result.awemeDict, datadict['aweme_detail'])
|
|
|
|
return self.result.awemeDict, datadict
|
|
|
|
def getUserInfoApi(self, sec_uid, mode="post", count=35, max_cursor=0):
|
|
if sec_uid is None:
|
|
return None
|
|
|
|
awemeList = []
|
|
|
|
start = time.time() # 开始时间
|
|
while True:
|
|
try:
|
|
if mode == "post":
|
|
url = self.urls.USER_POST + utils.getXbogus(
|
|
f'sec_user_id={sec_uid}&count={count}&max_cursor={max_cursor}&device_platform=webapp&aid=6383')
|
|
elif mode == "like":
|
|
url = self.urls.USER_FAVORITE_A + utils.getXbogus(
|
|
f'sec_user_id={sec_uid}&count={count}&max_cursor={max_cursor}&device_platform=webapp&aid=6383')
|
|
else:
|
|
return None
|
|
|
|
res = requests.get(url=url, headers=douyin_headers)
|
|
datadict = json.loads(res.text)
|
|
if datadict is not None and datadict["status_code"] == 0:
|
|
break
|
|
except Exception as e:
|
|
end = time.time() # 结束时间
|
|
if end - start > self.timeout:
|
|
return None
|
|
|
|
for aweme in datadict["aweme_list"]:
|
|
# 清空self.awemeDict
|
|
self.result.clearDict(self.result.awemeDict)
|
|
|
|
# 默认为视频
|
|
awemeType = 0
|
|
try:
|
|
if aweme["images"] is not None:
|
|
awemeType = 1
|
|
except Exception as e:
|
|
pass
|
|
|
|
# 转换成我们自己的格式
|
|
self.result.dataConvert(awemeType, self.result.awemeDict, aweme)
|
|
|
|
if self.result.awemeDict is not None and self.result.awemeDict != {}:
|
|
awemeList.append(copy.deepcopy(self.result.awemeDict))
|
|
|
|
return awemeList, datadict, datadict["max_cursor"], datadict["has_more"]
|
|
|
|
def getLiveInfoApi(self, web_rid: str):
|
|
start = time.time() # 开始时间
|
|
while True:
|
|
try:
|
|
live_api = self.urls.LIVE + utils.getXbogus(
|
|
f'aid=6383&device_platform=web&web_rid={web_rid}')
|
|
|
|
response = requests.get(live_api, headers=douyin_headers)
|
|
live_json = json.loads(response.text)
|
|
if live_json != {} and live_json['status_code'] == 0:
|
|
break
|
|
except Exception as e:
|
|
end = time.time() # 结束时间
|
|
if end - start > self.timeout:
|
|
return None
|
|
|
|
# 清空字典
|
|
self.result.clearDict(self.result.liveDict)
|
|
|
|
# 类型
|
|
self.result.liveDict["awemeType"] = 2
|
|
# 是否在播
|
|
self.result.liveDict["status"] = live_json['data']['data'][0]['status']
|
|
|
|
if self.result.liveDict["status"] == 4:
|
|
return self.result.liveDict, live_json
|
|
|
|
# 直播标题
|
|
self.result.liveDict["title"] = live_json['data']['data'][0]['title']
|
|
|
|
# 直播cover
|
|
self.result.liveDict["cover"] = live_json['data']['data'][0]['cover']['url_list'][0]
|
|
|
|
# 头像
|
|
self.result.liveDict["avatar"] = live_json['data']['data'][0]['owner']['avatar_thumb']['url_list'][0].replace(
|
|
"100x100", "1080x1080")
|
|
|
|
# 观看人数
|
|
self.result.liveDict["user_count"] = live_json['data']['data'][0]['user_count_str']
|
|
|
|
# 昵称
|
|
self.result.liveDict["nickname"] = live_json['data']['data'][0]['owner']['nickname']
|
|
|
|
# sec_uid
|
|
self.result.liveDict["sec_uid"] = live_json['data']['data'][0]['owner']['sec_uid']
|
|
|
|
# 直播间观看状态
|
|
self.result.liveDict["display_long"] = live_json['data']['data'][0]['room_view_stats']['display_long']
|
|
|
|
# 推流
|
|
self.result.liveDict["flv_pull_url"] = live_json['data']['data'][0]['stream_url']['flv_pull_url']
|
|
|
|
try:
|
|
# 分区
|
|
self.result.liveDict["partition"] = live_json['data']['partition_road_map']['partition']['title']
|
|
self.result.liveDict["sub_partition"] = \
|
|
live_json['data']['partition_road_map']['sub_partition']['partition']['title']
|
|
except Exception as e:
|
|
self.result.liveDict["partition"] = '无'
|
|
self.result.liveDict["sub_partition"] = '无'
|
|
|
|
flv = []
|
|
|
|
for i, f in enumerate(self.result.liveDict["flv_pull_url"].keys()):
|
|
flv.append(f)
|
|
|
|
self.result.liveDict["flv_pull_url0"] = self.result.liveDict["flv_pull_url"][flv[0]]
|
|
|
|
return self.result.liveDict, live_json
|
|
|
|
def getMixInfoApi(self, mix_id: str, count=35, cursor=0):
|
|
if mix_id is None:
|
|
return None
|
|
|
|
awemeList = []
|
|
|
|
start = time.time() # 开始时间
|
|
while True:
|
|
try:
|
|
url = self.urls.USER_MIX + utils.getXbogus(
|
|
f'mix_id={mix_id}&cursor={cursor}&count={count}&device_platform=webapp&aid=6383')
|
|
|
|
res = requests.get(url=url, headers=douyin_headers)
|
|
datadict = json.loads(res.text)
|
|
if datadict is not None:
|
|
break
|
|
except Exception as e:
|
|
end = time.time() # 结束时间
|
|
if end - start > self.timeout:
|
|
return None
|
|
|
|
for aweme in datadict["aweme_list"]:
|
|
|
|
# 清空self.awemeDict
|
|
self.result.clearDict(self.result.awemeDict)
|
|
|
|
# 默认为视频
|
|
awemeType = 0
|
|
try:
|
|
if aweme["images"] is not None:
|
|
awemeType = 1
|
|
except Exception as e:
|
|
pass
|
|
|
|
# 转换成我们自己的格式
|
|
self.result.dataConvert(awemeType, self.result.awemeDict, aweme)
|
|
|
|
if self.result.awemeDict is not None and self.result.awemeDict != {}:
|
|
awemeList.append(copy.deepcopy(self.result.awemeDict))
|
|
|
|
return awemeList, datadict, datadict["cursor"], datadict["has_more"]
|
|
|
|
def getUserAllMixInfoApi(self, sec_uid, count=35, cursor=0):
|
|
|
|
if sec_uid is None:
|
|
return None
|
|
|
|
mixIdlist = []
|
|
|
|
start = time.time() # 开始时间
|
|
while True:
|
|
try:
|
|
url = self.urls.USER_MIX_LIST + utils.getXbogus(
|
|
f'sec_user_id={sec_uid}&count={count}&cursor={cursor}&device_platform=webapp&aid=6383')
|
|
|
|
res = requests.get(url=url, headers=douyin_headers)
|
|
datadict = json.loads(res.text)
|
|
if datadict is not None and datadict["status_code"] == 0:
|
|
break
|
|
except Exception as e:
|
|
end = time.time() # 结束时间
|
|
if end - start > self.timeout:
|
|
return None
|
|
|
|
for mix in datadict["mix_infos"]:
|
|
mixIdNameDict = {}
|
|
mixIdNameDict["https://www.douyin.com/collection/" + mix["mix_id"]] = mix["mix_name"]
|
|
mixIdlist.append(mixIdNameDict)
|
|
|
|
return mixIdlist, datadict, datadict["cursor"], datadict["has_more"]
|
|
|
|
def getMusicInfoApi(self, music_id: str, count=35, cursor=0):
|
|
if music_id is None:
|
|
return None
|
|
|
|
awemeList = []
|
|
|
|
start = time.time() # 开始时间
|
|
while True:
|
|
try:
|
|
url = self.urls.MUSIC + utils.getXbogus(
|
|
f'music_id={music_id}&cursor={cursor}&count={count}&device_platform=webapp&aid=6383')
|
|
|
|
res = requests.get(url=url, headers=douyin_headers)
|
|
datadict = json.loads(res.text)
|
|
if datadict is not None and datadict["status_code"] == 0:
|
|
break
|
|
except Exception as e:
|
|
end = time.time() # 结束时间
|
|
if end - start > self.timeout:
|
|
return None
|
|
|
|
for aweme in datadict["aweme_list"]:
|
|
# 清空self.awemeDict
|
|
self.result.clearDict(self.result.awemeDict)
|
|
|
|
# 默认为视频
|
|
awemeType = 0
|
|
try:
|
|
if aweme["images"] is not None:
|
|
awemeType = 1
|
|
except Exception as e:
|
|
pass
|
|
|
|
# 转换成我们自己的格式
|
|
self.result.dataConvert(awemeType, self.result.awemeDict, aweme)
|
|
|
|
if self.result.awemeDict is not None and self.result.awemeDict != {}:
|
|
awemeList.append(copy.deepcopy(self.result.awemeDict))
|
|
|
|
return awemeList, datadict, datadict["cursor"], datadict["has_more"]
|
|
|
|
def getUserDetailInfoApi(self, sec_uid):
|
|
if sec_uid is None:
|
|
return None
|
|
|
|
start = time.time() # 开始时间
|
|
while True:
|
|
# 接口不稳定, 有时服务器不返回数据, 需要重新获取
|
|
try:
|
|
url = self.urls.USER_DETAIL + utils.getXbogus(
|
|
f'sec_user_id={sec_uid}&device_platform=webapp&aid=6383')
|
|
|
|
res = requests.get(url=url, headers=douyin_headers)
|
|
datadict = json.loads(res.text)
|
|
|
|
if datadict is not None and datadict["status_code"] == 0:
|
|
return datadict
|
|
except Exception as e:
|
|
end = time.time() # 结束时间
|
|
if end - start > self.timeout:
|
|
return None
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pass
|