blackteay/tiktok
1
0
Fork 0
mirror of https://github.com/imgyh/tiktok.git synced 2025-04-05 23:25:45 +08:00
Code Issues Projects Releases Wiki Activity

feat(tiktok): 删除多余文件

Browse Source
This commit is contained in:
imgyh 2023-05-21 16:25:02 +08:00
parent 54e50598de
commit 2114bc925e
9 changed files with 0 additions and 3041 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1214
TikTok.py
View File

File diff suppressed because it is too large Load Diff

353
TikTokCommand.py
View File

@ -1,353 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@Description:TikTok.py
@Date :2023/01/27 19:36:18
@Author :imgyh
@version :1.0
@Github :https://github.com/imgyh
@Mail :admin@imgyh.com
-------------------------------------------------
Change Log :
-------------------------------------------------
'''
import argparse
import os
import sys
import json
import yaml
import time
from TikTok import TikTok
from TikTokUtils import Utils
configModel = {
"link": [],
"path": os.getcwd(),
"music": True,
"cover": True,
"avatar": True,
"json": True,
"mode": ["post"],
"number": {
"post": 0,
"like": 0,
"allmix": 0,
"mix": 0,
"music": 0,
},
"increase": {
"post": False,
"like": False,
"allmix": False,
"mix": False,
"music": False,
},
"thread": 5,
"cookie": None
}
def argument():
parser = argparse.ArgumentParser(description='抖音批量下载工具 使用帮助')
parser.add_argument("--cmd", "-C", help="使用命令行(True)或者配置文件(False), 默认为False",
type=Utils().str2bool, required=False, default=False)
parser.add_argument("--link", "-l",
help="作品(视频或图集)、直播、合集、音乐集合、个人主页的分享链接或者电脑浏览器网址, 可以设置多个链接(删除文案, 保证只有URL, https://v.douyin.com/kcvMpuN/ 或者 https://www.douyin.com/开头的)",
type=str, required=False, default=[], action="append")
parser.add_argument("--path", "-p", help="下载保存位置, 默认当前文件位置",
type=str, required=False, default=os.getcwd())
parser.add_argument("--music", "-m", help="是否下载视频中的音乐(True/False), 默认为True",
type=Utils().str2bool, required=False, default=True)
parser.add_argument("--cover", "-c", help="是否下载视频的封面(True/False), 默认为True, 当下载视频时有效",
type=Utils().str2bool, required=False, default=True)
parser.add_argument("--avatar", "-a", help="是否下载作者的头像(True/False), 默认为True",
type=Utils().str2bool, required=False, default=True)
parser.add_argument("--json", "-j", help="是否保存获取到的数据(True/False), 默认为True",
type=Utils().str2bool, required=False, default=True)
parser.add_argument("--mode", "-M", help="link是个人主页时, 设置下载发布的作品(post)或喜欢的作品(like)或者用户所有合集(mix), 默认为post, 可以设置多种模式",
type=str, required=False, default=[], action="append")
parser.add_argument("--postnumber", help="主页下作品下载个数设置, 默认为0 全部下载",
type=int, required=False, default=0)
parser.add_argument("--likenumber", help="主页下喜欢下载个数设置, 默认为0 全部下载",
type=int, required=False, default=0)
parser.add_argument("--allmixnumber", help="主页下合集下载个数设置, 默认为0 全部下载",
type=int, required=False, default=0)
parser.add_argument("--mixnumber", help="单个合集下作品下载个数设置, 默认为0 全部下载",
type=int, required=False, default=0)
parser.add_argument("--musicnumber", help="音乐(原声)下作品下载个数设置, 默认为0 全部下载",
type=int, required=False, default=0)
parser.add_argument("--postincrease", help="是否开启主页作品增量下载(True/False), 默认为False",
type=Utils().str2bool, required=False, default=False)
parser.add_argument("--likeincrease", help="是否开启主页喜欢增量下载(True/False), 默认为False",
type=Utils().str2bool, required=False, default=False)
parser.add_argument("--allmixincrease", help="是否开启主页合集增量下载(True/False), 默认为False",
type=Utils().str2bool, required=False, default=False)
parser.add_argument("--mixincrease", help="是否开启单个合集下作品增量下载(True/False), 默认为False",
type=Utils().str2bool, required=False, default=False)
parser.add_argument("--musicincrease", help="是否开启音乐(原声)下作品增量下载(True/False), 默认为False",
type=Utils().str2bool, required=False, default=False)
parser.add_argument("--thread", "-t",
help="设置线程数, 默认5个线程",
type=int, required=False, default=5)
parser.add_argument("--cookie", help="设置cookie, 格式: \"name1=value1; name2=value2;\" 注意要加冒号",
type=str, required=False, default='')
args = parser.parse_args()
if args.thread <= 0:
args.thread = 5
return args
def yamlConfig():
curPath = os.path.dirname(os.path.realpath(sys.argv[0]))
yamlPath = os.path.join(curPath, "config.yml")
f = open(yamlPath, 'r', encoding='utf-8')
cfg = f.read()
configDict = yaml.load(stream=cfg, Loader=yaml.FullLoader)
try:
if configDict["link"] != None:
configModel["link"] = configDict["link"]
except Exception as e:
print("[ 警告 ]:link未设置, 程序退出...\r\n")
try:
if configDict["path"] != None:
configModel["path"] = configDict["path"]
except Exception as e:
print("[ 警告 ]:path未设置, 使用当前路径...\r\n")
try:
if configDict["music"] != None:
configModel["music"] = configDict["music"]
except Exception as e:
print("[ 警告 ]:music未设置, 使用默认值True...\r\n")
try:
if configDict["cover"] != None:
configModel["cover"] = configDict["cover"]
except Exception as e:
print("[ 警告 ]:cover未设置, 使用默认值True...\r\n")
try:
if configDict["avatar"] != None:
configModel["avatar"] = configDict["avatar"]
except Exception as e:
print("[ 警告 ]:avatar未设置, 使用默认值True...\r\n")
try:
if configDict["json"] != None:
configModel["json"] = configDict["json"]
except Exception as e:
print("[ 警告 ]:json未设置, 使用默认值True...\r\n")
try:
if configDict["mode"] != None:
configModel["mode"] = configDict["mode"]
except Exception as e:
print("[ 警告 ]:mode未设置, 使用默认值post...\r\n")
try:
if configDict["number"]["post"] != None:
configModel["number"]["post"] = configDict["number"]["post"]
except Exception as e:
print("[ 警告 ]:post number未设置, 使用默认值0...\r\n")
try:
if configDict["number"]["like"] != None:
configModel["number"]["like"] = configDict["number"]["like"]
except Exception as e:
print("[ 警告 ]:like number未设置, 使用默认值0...\r\n")
try:
if configDict["number"]["allmix"] != None:
configModel["number"]["allmix"] = configDict["number"]["allmix"]
except Exception as e:
print("[ 警告 ]:allmix number未设置, 使用默认值0...\r\n")
try:
if configDict["number"]["mix"] != None:
configModel["number"]["mix"] = configDict["number"]["mix"]
except Exception as e:
print("[ 警告 ]:mix number未设置, 使用默认值0...\r\n")
try:
if configDict["number"]["music"] != None:
configModel["number"]["music"] = configDict["number"]["music"]
except Exception as e:
print("[ 警告 ]:music number未设置, 使用默认值0...\r\n")
try:
if configDict["increase"]["post"] != None:
configModel["increase"]["post"] = configDict["increase"]["post"]
except Exception as e:
print("[ 警告 ]:post 增量更新未设置, 使用默认值False...\r\n")
try:
if configDict["increase"]["like"] != None:
configModel["increase"]["like"] = configDict["increase"]["like"]
except Exception as e:
print("[ 警告 ]:like 增量更新未设置, 使用默认值False...\r\n")
try:
if configDict["increase"]["allmix"] != None:
configModel["increase"]["allmix"] = configDict["increase"]["allmix"]
except Exception as e:
print("[ 警告 ]:allmix 增量更新未设置, 使用默认值False...\r\n")
try:
if configDict["increase"]["mix"] != None:
configModel["increase"]["mix"] = configDict["increase"]["mix"]
except Exception as e:
print("[ 警告 ]:mix 增量更新未设置, 使用默认值False...\r\n")
try:
if configDict["increase"]["music"] != None:
configModel["increase"]["music"] = configDict["increase"]["music"]
except Exception as e:
print("[ 警告 ]:music 增量更新未设置, 使用默认值False...\r\n")
try:
if configDict["thread"] != None:
configModel["thread"] = configDict["thread"]
except Exception as e:
print("[ 警告 ]:thread未设置, 使用默认值5...\r\n")
try:
if configDict["cookies"] != None:
cookiekey = configDict["cookies"].keys()
cookieStr = ""
for i in cookiekey:
cookieStr = cookieStr + i + "=" + configDict["cookies"][i] + "; "
configModel["cookie"] = cookieStr
except Exception as e:
pass
try:
if configDict["cookie"] != None:
configModel["cookie"] = configDict["cookie"]
except Exception as e:
pass
def main():
start = time.time() # 开始时间
utils = Utils()
args = argument()
if args.cmd:
configModel["link"] = args.link
configModel["path"] = args.path
configModel["music"] = args.music
configModel["cover"] = args.cover
configModel["avatar"] = args.avatar
configModel["json"] = args.json
if args.mode == None or args.mode == []:
args.mode = []
args.mode.append("post")
configModel["mode"] = list(set(args.mode))
configModel["number"]["post"] = args.postnumber
configModel["number"]["like"] = args.likenumber
configModel["number"]["allmix"] = args.allmixnumber
configModel["number"]["mix"] = args.mixnumber
configModel["number"]["music"] = args.musicnumber
configModel["increase"]["post"] = args.postincrease
configModel["increase"]["like"] = args.likeincrease
configModel["increase"]["allmix"] = args.allmixincrease
configModel["increase"]["mix"] = args.mixincrease
configModel["increase"]["music"] = args.musicincrease
configModel["thread"] = args.thread
configModel["cookie"] = args.cookie
else:
yamlConfig()
if configModel["link"] == []:
return
tk = TikTok()
if configModel["cookie"] is not None and configModel["cookie"] != "":
tk.headers["Cookie"] = configModel["cookie"]
configModel["path"] = os.path.abspath(configModel["path"])
print("[ 提示 ]:数据保存路径 " + configModel["path"])
if not os.path.exists(configModel["path"]):
os.mkdir(configModel["path"])
for link in configModel["link"]:
print("--------------------------------------------------------------------------------")
print("[ 提示 ]:正在请求的链接: " + link + "\r\n")
url = tk.getShareLink(link)
key_type, key = tk.getKey(url)
if key_type == "user":
print("[ 提示 ]:正在请求用户主页下作品\r\n")
userPath = os.path.join(configModel["path"], "user_" + key)
if not os.path.exists(userPath):
os.mkdir(userPath)
for mode in configModel["mode"]:
print("--------------------------------------------------------------------------------")
print("[ 提示 ]:正在请求用户主页模式: " + mode + "\r\n")
if mode == 'post' or mode == 'like':
datalist = tk.getUserInfo(key, mode, 35, configModel["number"][mode], configModel["increase"][mode])
if datalist is not None and datalist != []:
modePath = os.path.join(userPath, mode)
if not os.path.exists(modePath):
os.mkdir(modePath)
tk.userDownload(awemeList=datalist, music=configModel["music"], cover=configModel["cover"],
avatar=configModel["avatar"], resjson=configModel["json"],
savePath=modePath, thread=configModel["thread"])
elif mode == 'mix':
mixIdNameDict = tk.getUserAllMixInfo(key, 35, configModel["number"]["allmix"])
if mixIdNameDict is not None and mixIdNameDict != {}:
for mix_id in mixIdNameDict:
print(f'[ 提示 ]:正在下载合集 [{mixIdNameDict[mix_id]}] 中的作品\r\n')
mix_file_name = utils.replaceStr(mixIdNameDict[mix_id])
datalist = tk.getMixInfo(mix_id, 35, 0, configModel["increase"]["allmix"], key)
if datalist is not None and datalist != []:
modePath = os.path.join(userPath, mode)
if not os.path.exists(modePath):
os.mkdir(modePath)
tk.userDownload(awemeList=datalist, music=configModel["music"],
cover=configModel["cover"],
avatar=configModel["avatar"], resjson=configModel["json"],
savePath=os.path.join(modePath, mix_file_name),
thread=configModel["thread"])
print(f'[ 提示 ]:合集 [{mixIdNameDict[mix_id]}] 中的作品下载完成\r\n')
elif key_type == "mix":
print("[ 提示 ]:正在请求单个合集下作品\r\n")
datalist = tk.getMixInfo(key, 35, configModel["number"]["mix"], configModel["increase"]["mix"], "")
if datalist is not None and datalist != []:
mixPath = os.path.join(configModel["path"], "mix_" + key)
if not os.path.exists(mixPath):
os.mkdir(mixPath)
tk.userDownload(awemeList=datalist, music=configModel["music"], cover=configModel["cover"],
avatar=configModel["avatar"], resjson=configModel["json"],
savePath=mixPath, thread=configModel["thread"])
elif key_type == "music":
print("[ 提示 ]:正在请求音乐(原声)下作品\r\n")
datalist = tk.getMusicInfo(key, 35, configModel["number"]["music"], configModel["increase"]["music"])
if datalist is not None and datalist != []:
musicPath = os.path.join(configModel["path"], "music_" + key)
if not os.path.exists(musicPath):
os.mkdir(musicPath)
tk.userDownload(awemeList=datalist, music=configModel["music"], cover=configModel["cover"],
avatar=configModel["avatar"], resjson=configModel["json"],
savePath=musicPath, thread=configModel["thread"])
elif key_type == "aweme":
print("[ 提示 ]:正在请求单个作品\r\n")
datanew, dataraw = tk.getAwemeInfo(key)
if datanew is not None and datanew != {}:
datalist = []
datalist.append(datanew)
awemePath = os.path.join(configModel["path"], "aweme")
if not os.path.exists(awemePath):
os.mkdir(awemePath)
tk.userDownload(awemeList=datalist, music=configModel["music"], cover=configModel["cover"],
avatar=configModel["avatar"], resjson=configModel["json"],
savePath=awemePath, thread=configModel["thread"])
elif key_type == "live":
print("[ 提示 ]:正在进行直播解析\r\n")
live_json = tk.getLiveInfo(key)
if configModel["json"]:
livePath = os.path.join(configModel["path"], "live")
if not os.path.exists(livePath):
os.mkdir(livePath)
live_file_name = utils.replaceStr(key + live_json["nickname"])
# 保存获取到json
print("[ 提示 ]:正在保存获取到的信息到result.json\r\n")
with open(os.path.join(livePath, live_file_name + ".json"), "w", encoding='utf-8') as f:
f.write(json.dumps(live_json, ensure_ascii=False, indent=2))
f.close()
end = time.time() # 结束时间
print('\n' + '[下载完成]:总耗时: %d分钟%d秒\n' % (int((end - start) / 60), ((end - start) % 60))) # 输出下载用时时间
if __name__ == "__main__":
main()

172
TikTokDataBase.py
View File

@ -1,172 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
@FileName : TikTokDataBase.py
@Project : tiktok
@Description:
@Author : imgyh
@Mail : admin@imgyh.com
@Github : https://github.com/imgyh
@Site : https://www.imgyh.com
@Date : 2023/4/24 10:05
@Version : v1.0
@ChangeLog
------------------------------------------------
使用数据库保存获取的状态信息
------------------------------------------------
'''
import sqlite3
import json
class db(object):
def __init__(self):
self.conn = sqlite3.connect('data.db')
self.cursor = self.conn.cursor()
self.create_user_post_table()
self.create_user_like_table()
self.create_mix_table()
self.create_music_table()
def create_user_post_table(self):
sql = """CREATE TABLE if not exists t_user_post (
id integer primary key autoincrement,
sec_uid varchar(200),
aweme_id integer unique,
rawdata json
);"""
try:
self.cursor.execute(sql)
self.conn.commit()
except Exception as e:
pass
def get_user_post(self, sec_uid: str, aweme_id: int):
sql = """select id, sec_uid, aweme_id, rawdata from t_user_post where sec_uid=? and aweme_id=?;"""
try:
self.cursor.execute(sql, (sec_uid, aweme_id))
self.conn.commit()
res = self.cursor.fetchone()
return res
except Exception as e:
pass
def insert_user_post(self, sec_uid: str, aweme_id: int, data: dict):
insertsql = """insert into t_user_post (sec_uid, aweme_id, rawdata) values(?,?,?);"""
try:
self.cursor.execute(insertsql, (sec_uid, aweme_id, json.dumps(data)))
self.conn.commit()
except Exception as e:
pass
def create_user_like_table(self):
sql = """CREATE TABLE if not exists t_user_like (
id integer primary key autoincrement,
sec_uid varchar(200),
aweme_id integer unique,
rawdata json
);"""
try:
self.cursor.execute(sql)
self.conn.commit()
except Exception as e:
pass
def get_user_like(self, sec_uid: str, aweme_id: int):
sql = """select id, sec_uid, aweme_id, rawdata from t_user_like where sec_uid=? and aweme_id=?;"""
try:
self.cursor.execute(sql, (sec_uid, aweme_id))
self.conn.commit()
res = self.cursor.fetchone()
return res
except Exception as e:
pass
def insert_user_like(self, sec_uid: str, aweme_id: int, data: dict):
insertsql = """insert into t_user_like (sec_uid, aweme_id, rawdata) values(?,?,?);"""
try:
self.cursor.execute(insertsql, (sec_uid, aweme_id, json.dumps(data)))
self.conn.commit()
except Exception as e:
pass
def create_mix_table(self):
sql = """CREATE TABLE if not exists t_mix (
id integer primary key autoincrement,
sec_uid varchar(200),
mix_id varchar(200),
aweme_id integer,
rawdata json
);"""
try:
self.cursor.execute(sql)
self.conn.commit()
except Exception as e:
pass
def get_mix(self, sec_uid: str, mix_id: str, aweme_id: int):
sql = """select id, sec_uid, mix_id, aweme_id, rawdata from t_mix where sec_uid=? and mix_id=? and aweme_id=?;"""
try:
self.cursor.execute(sql, (sec_uid, mix_id, aweme_id))
self.conn.commit()
res = self.cursor.fetchone()
return res
except Exception as e:
pass
def insert_mix(self, sec_uid: str, mix_id: str, aweme_id: int, data: dict):
insertsql = """insert into t_mix (sec_uid, mix_id, aweme_id, rawdata) values(?,?,?,?);"""
try:
self.cursor.execute(insertsql, (sec_uid, mix_id, aweme_id, json.dumps(data)))
self.conn.commit()
except Exception as e:
pass
def create_music_table(self):
sql = """CREATE TABLE if not exists t_music (
id integer primary key autoincrement,
music_id varchar(200),
aweme_id integer unique,
rawdata json
);"""
try:
self.cursor.execute(sql)
self.conn.commit()
except Exception as e:
pass
def get_music(self, music_id: str, aweme_id: int):
sql = """select id, music_id, aweme_id, rawdata from t_music where music_id=? and aweme_id=?;"""
try:
self.cursor.execute(sql, (music_id, aweme_id))
self.conn.commit()
res = self.cursor.fetchone()
return res
except Exception as e:
pass
def insert_music(self, music_id: str, aweme_id: int, data: dict):
insertsql = """insert into t_music (music_id, aweme_id, rawdata) values(?,?,?);"""
try:
self.cursor.execute(insertsql, (music_id, aweme_id, json.dumps(data)))
self.conn.commit()
except Exception as e:
pass
if __name__ == '__main__':
pass

284
TikTokResult.py
View File

@ -1,284 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@Description:TikTok.py
@Date :2023/02/11 13:06:23
@Author :imgyh
@version :1.0
@Github :https://github.com/imgyh
@Mail :admin@imgyh.com
-------------------------------------------------
Change Log :
-------------------------------------------------
'''
import time
import copy
class Result(object):
def __init__(self):
# 作者信息
self.authorDict = {
"avatar_thumb": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"avatar": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"cover_url": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
# 喜欢的作品数
"favoriting_count": "",
# 粉丝数
"follower_count": "",
# 关注数
"following_count": "",
# 昵称
"nickname": "",
# 是否允许下载
"prevent_download": "",
# 用户 url id
"sec_uid": "",
# 是否私密账号
"secret": "",
# 短id
"short_id": "",
# 签名
"signature": "",
# 总获赞数
"total_favorited": "",
# 用户id
"uid": "",
# 用户自定义唯一id 抖音号
"unique_id": "",
# 年龄
"user_age": "",
}
# 图片信息
self.picDict = {
"height": "",
"mask_url_list": "",
"uri": "",
"url_list": [],
"width": ""
}
# 音乐信息
self.musicDict = {
"cover_hd": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"cover_large": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"cover_medium": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"cover_thumb": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
# 音乐作者抖音号
"owner_handle": "",
# 音乐作者id
"owner_id": "",
# 音乐作者昵称
"owner_nickname": "",
"play_url": {
"height": "",
"uri": "",
"url_key": "",
"url_list": [],
"width": ""
},
# 音乐名字
"title": "",
}
# 视频信息
self.videoDict = {
"play_addr": {
"uri": "",
"url_list": [],
},
"cover_original_scale": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"dynamic_cover": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"origin_cover": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
},
"cover": {
"height": "",
"uri": "",
"url_list": [],
"width": ""
}
}
# 作品信息
self.awemeDict = {
# 作品创建时间
"create_time": "",
# awemeType=0 视频, awemeType=1 图集, awemeType=2 直播
"awemeType": "",
# 作品 id
"aweme_id": "",
# 作者信息
"author": self.authorDict,
# 作品描述
"desc": "",
# 图片
"images": [],
# 音乐
"music": self.musicDict,
# 视频
"video": self.videoDict,
# 作品信息统计
"statistics": {
"admire_count": "",
"collect_count": "",
"comment_count": "",
"digg_count": "",
"play_count": "",
"share_count": ""
}
}
# 用户作品信息
self.awemeList = []
# 直播信息
self.liveDict = {
# awemeType=0 视频, awemeType=1 图集, awemeType=2 直播
"awemeType": "",
# 是否在播
"status": "",
# 直播标题
"title": "",
# 直播cover
"cover": "",
# 头像
"avatar": "",
# 观看人数
"user_count": "",
# 昵称
"nickname": "",
# sec_uid
"sec_uid": "",
# 直播间观看状态
"display_long": "",
# 推流
"flv_pull_url": "",
# 分区
"partition": "",
"sub_partition": "",
# 最清晰的地址
"flv_pull_url0": "",
}
# 将得到的json数据（dataRaw）精简成自己定义的数据（dataNew）
# 转换得到的数据
def dataConvert(self, awemeType, dataNew, dataRaw):
for item in dataNew:
try:
# 作品创建时间
if item == "create_time":
dataNew['create_time'] = time.strftime(
"%Y-%m-%d %H.%M.%S", time.localtime(dataRaw['create_time']))
continue
# 设置 awemeType
if item == "awemeType":
dataNew["awemeType"] = awemeType
continue
# 当 解析的链接 是图片时
if item == "images":
if awemeType == 1:
for image in dataRaw[item]:
for i in image:
self.picDict[i] = image[i]
# 字典要深拷贝
self.awemeDict["images"].append(copy.deepcopy(self.picDict))
continue
# 当 解析的链接 是视频时
if item == "video":
if awemeType == 0:
self.dataConvert(awemeType, dataNew[item], dataRaw[item])
continue
# 将小头像放大
if item == "avatar":
for i in dataNew[item]:
if i == "url_list":
for j in self.awemeDict["author"]["avatar_thumb"]["url_list"]:
dataNew[item][i].append(j.replace("100x100", "1080x1080"))
elif i == "uri":
dataNew[item][i] = self.awemeDict["author"]["avatar_thumb"][i].replace("100x100",
"1080x1080")
else:
dataNew[item][i] = self.awemeDict["author"]["avatar_thumb"][i]
continue
# 原来的json是[{}] 而我们的是 {}
if item == "cover_url":
self.dataConvert(awemeType, dataNew[item], dataRaw[item][0])
continue
# 根据 uri 获取 1080p 视频
if item == "play_addr":
dataNew[item]["uri"] = dataRaw["bit_rate"][0]["play_addr"]["uri"]
# 使用 这个api 可以获得1080p
# dataNew[item]["url_list"] = "https://aweme.snssdk.com/aweme/v1/play/?video_id=%s&ratio=1080p&line=0" \
# % dataNew[item]["uri"]
dataNew[item]["url_list"] = copy.deepcopy(dataRaw["bit_rate"][0]["play_addr"]["url_list"])
continue
# 常规 递归遍历 字典
if isinstance(dataNew[item], dict):
self.dataConvert(awemeType, dataNew[item], dataRaw[item])
else:
# 赋值
dataNew[item] = dataRaw[item]
except Exception as e:
# 删除这个警告, 总是让人误会出错了
# print("[ 警告 ]:转换数据时在接口中未找到 %s\r" % (item))
pass
def clearDict(self, data):
for item in data:
# 常规 递归遍历 字典
if isinstance(data[item], dict):
self.clearDict(data[item])
elif isinstance(data[item], list):
data[item] = []
else:
data[item] = ""

109
TikTokTest.py
View File

@ -1,109 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@Description:TikTok.py
@Date :2023/02/11 13:06:23
@Author :imgyh
@version :1.0
@Github :https://github.com/imgyh
@Mail :admin@imgyh.com
-------------------------------------------------
Change Log :
-------------------------------------------------
'''
import TikTokUtils
from TikTok import TikTok
def getAwemeInfo():
share_link_video = "3.56 uSy:/ 复制打开抖音，看看【小透明的作品】没有女朋友就用我的吧哈哈哈哈 # 表情包锁屏 https://v.douyin.com/BugmVVD/"
share_link_pic = "8.20 MJI:/ 复制打开抖音，看看【舍溪的图文作品】我又来放图集啦～还有你们要的小可爱大图也放啦～# ... https://v.douyin.com/BugrFTN/"
tk = TikTok()
url = tk.getShareLink(share_link_pic)
key_type, key = tk.getKey(url)
datanew, dataraw = tk.getAwemeInfo(key)
print(datanew)
def getUserInfo():
share_link_post = "1- 长按复制此条消息，打开抖音搜索，查看TA的更多作品。 https://v.douyin.com/BupCppt/"
share_link_like = "2- 长按复制此条消息，打开抖音搜索，查看TA的更多作品。 https://v.douyin.com/BusJrfr/"
tk = TikTok()
url = tk.getShareLink(share_link_like)
key_type, key = tk.getKey(url)
awemeList = tk.getUserInfo(key, mode="like", count=35)
print(awemeList)
def getLiveInfo():
live_link = "https://live.douyin.com/40768897856"
tk = TikTok()
url = tk.getShareLink(live_link)
key_type, key = tk.getKey(url)
live_json = tk.getLiveInfo(key)
print(live_json)
def getMixInfo():
mix_link = 'https://v.douyin.com/B3J63Le/'
tk = TikTok()
url = tk.getShareLink(mix_link)
key_type, key = tk.getKey(url)
awemeList = tk.getMixInfo(key, count=35)
print(len(awemeList))
def getUserAllMixInfo():
user_all_mix_link = 'https://v.douyin.com/B38oovu/'
tk = TikTok()
url = tk.getShareLink(user_all_mix_link)
key_type, key = tk.getKey(url)
mixIdNameDict = tk.getUserAllMixInfo(key, count=35)
print(mixIdNameDict)
def getMusicInfo():
music_link = 'https://v.douyin.com/S6YMNXs/'
tk = TikTok()
url = tk.getShareLink(music_link)
key_type, key = tk.getKey(url)
awemeList = tk.getMusicInfo(key, count=35)
print(len(awemeList))
def test():
utils = TikTokUtils.Utils()
user_all_mix_link = 'https://www.douyin.com/aweme/v1/web/aweme/favorite/?' + \
utils.getXbogus(
url='device_platform=webapp&aid=6383&sec_user_id=MS4wLjABAAAAjQn6ONfaGgUpk0Q1ep8dPiD3W4T_lxTJmemfy3MTJ64&max_cursor=1676441180000&count=10')
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
'referer': 'https://www.douyin.com/',
'accept-encoding': None,
'Cookie': 'ttwid=1|oLudm-Hi5ikxQQhmAnv4Km4LjwwvLa4Qk_JGrKffuYU|1681878460|b0f581d97797bb67d2260bf92d65e8808b90713afc08bf5d8100f571fd70a275; passport_csrf_token=570d671dcd8d8598fcde4c3f7c99664d; passport_csrf_token_default=570d671dcd8d8598fcde4c3f7c99664d; s_v_web_id=verify_lgn70gzw_sAVtjJdD_Clyk_43UL_8M9L_6JiTtPC2TyU0; n_mh=vrLGYVtwqutbPLOGNTDUGahwaD9AyYjn4iVvAO2Xt0s; sso_uid_tt=92e014dfb6653bdc319ecc6a6ceea870; sso_uid_tt_ss=92e014dfb6653bdc319ecc6a6ceea870; toutiao_sso_user=d0bd8d5cc0f75420799903572941ce83; toutiao_sso_user_ss=d0bd8d5cc0f75420799903572941ce83; passport_auth_status=5c576b088f4a19448eb4efd7aaeb7c5e,; passport_auth_status_ss=5c576b088f4a19448eb4efd7aaeb7c5e,; uid_tt=564215aecc985785d033c9e4c9d00fc4; uid_tt_ss=564215aecc985785d033c9e4c9d00fc4; sid_tt=f14390d924c81856fde84b1cd534bf09; sessionid=f14390d924c81856fde84b1cd534bf09; sessionid_ss=f14390d924c81856fde84b1cd534bf09; odin_tt=0aa1c100d17bf3541dce9e8dd62c6353302b8abaaa0a5998d5c437313feb245d8f7b6391a69772e4afa7d3c718878837; passport_assist_user=CjwC3fi9JyApG4DN-HiFI6n5FcGAdzNDT29nR-nSDJYAVfqh2BYI77qdTN2GRfgagkYLRi1wxNp1akHBwGcaSAo8XDkNT-UWnkFmc_0eXMYCb8OIh4G_YnH8pylwwdfS-7PCTekX3trj0JyENUVWLWFxnKuG5HhSS4FB2CtxEJ7yrg0Yia_WVCIBA9bY7IU=; sid_ucp_sso_v1=1.0.0-KDc0NmY3NzA5ZWFhNzI2YzMyOWMxMDMwNjAwMDg3YzRjYzg3ZjlhODcKHQjG-7zT9QEQ5dv9oQYY7zEgDDCL_7nMBTgGQPQHGgJscSIgZDBiZDhkNWNjMGY3NTQyMDc5OTkwMzU3Mjk0MWNlODM; ssid_ucp_sso_v1=1.0.0-KDc0NmY3NzA5ZWFhNzI2YzMyOWMxMDMwNjAwMDg3YzRjYzg3ZjlhODcKHQjG-7zT9QEQ5dv9oQYY7zEgDDCL_7nMBTgGQPQHGgJscSIgZDBiZDhkNWNjMGY3NTQyMDc5OTkwMzU3Mjk0MWNlODM; publish_badge_show_info="0,0,0,1681878505305"; LOGIN_STATUS=1; store-region=cn-sc; store-region-src=uid; sid_guard=f14390d924c81856fde84b1cd534bf09|1681878504|5183998|Sun,+18-Jun-2023+04:28:22+GMT; sid_ucp_v1=1.0.0-KGY5OGQ3OTQ4YmVkYzczODgzYzM0MmJmOWYxMzhkMDliMTc5NGI3NjMKGQjG-7zT9QEQ6Nv9oQYY7zEgDDgGQPQHSAQaAmxmIiBmMTQzOTBkOTI0YzgxODU2ZmRlODRiMWNkNTM0YmYwOQ; ssid_ucp_v1=1.0.0-KGY5OGQ3OTQ4YmVkYzczODgzYzM0MmJmOWYxMzhkMDliMTc5NGI3NjMKGQjG-7zT9QEQ6Nv9oQYY7zEgDDgGQPQHSAQaAmxmIiBmMTQzOTBkOTI0YzgxODU2ZmRlODRiMWNkNTM0YmYwOQ; download_guide="3/20230419"; pwa2="3|0"; FOLLOW_NUMBER_YELLOW_POINT_INFO="MS4wLjABAAAA-jD2lukp--I21BF8VQsmYUqJDbj3FmU-kGQTHl2y1Cw/1681920000000/0/0/1681900423333"; __ac_nonce=0644234d60042cacebbb6; __ac_signature=_02B4Z6wo00f01eZN0dAAAIDAhUcRuQz7dNHmbdVAAB3TRUC7i8VqXQejR8jv-D8UggBx1MBLH564PE.cCZf00m7Cw640CNUvcc5jJfhgn8u5FhvVndykvwbQb.HEpSfGN-8eqql7GpuGZJ8f1d; strategyABtestKey="1682060514.261"; passport_fe_beating_status=true; csrf_session_id=405b0ec4f6338a5d893a5f14f6fd1a64; bd_ticket_guard_client_data=eyJiZC10aWNrZXQtZ3VhcmQtdmVyc2lvbiI6MiwiYmQtdGlja2V0LWd1YXJkLWl0ZXJhdGlvbi12ZXJzaW9uIjoxLCJiZC10aWNrZXQtZ3VhcmQtY2xpZW50LWNlcnQiOiItLS0tLUJFR0lOIENFUlRJRklDQVRFLS0tLS1cbk1JSUNGRENDQWJxZ0F3SUJBZ0lVR1ljQ3FuQUh0UUJBZm5WWkYxQW84cUtjY2Zrd0NnWUlLb1pJemowRUF3SXdcbk1URUxNQWtHQTFVRUJoTUNRMDR4SWpBZ0JnTlZCQU1NR1hScFkydGxkRjluZFdGeVpGOWpZVjlsWTJSellWOHlcbk5UWXdIaGNOTWpNd016STNNRE15T0RBeldoY05Nek13TXpJM01URXlPREF6V2pBbk1Rc3dDUVlEVlFRR0V3SkRcblRqRVlNQllHQTFVRUF3d1BZbVJmZEdsamEyVjBYMmQxWVhKa01Ga3dFd1lIS29aSXpqMENBUVlJS29aSXpqMERcbkFRY0RRZ0FFNmZ1Z3d0MEJnZUh5akVub1FvNWtXUS9qc2daTTV1YXBiNTQ4KytTV0dRSjMwb2lSTHNtYlVFSUZcblFIYzh3UEthZzZmdXNPTm91WncxOEdNYm5vTStwNk9CdVRDQnRqQU9CZ05WSFE4QkFmOEVCQU1DQmFBd01RWURcblZSMGxCQ293S0FZSUt3WUJCUVVIQXdFR0NDc0dBUVVGQndNQ0JnZ3JCZ0VGQlFjREF3WUlLd1lCQlFVSEF3UXdcbktRWURWUjBPQkNJRUlPV3Y3d01ZUGhoeUNPL2ZwenJGNDJNeEQ4ZGIzN0YyTDgxaW8zVTVlVFpaTUNzR0ExVWRcbkl3UWtNQ0tBSURLbForcU9aRWdTamN4T1RVQjdjeFNiUjIxVGVxVFJnTmQ1bEpkN0lrZURNQmtHQTFVZEVRUVNcbk1CQ0NEbmQzZHk1a2IzVjVhVzR1WTI5dE1Bb0dDQ3FHU000OUJBTUNBMGdBTUVVQ0lCbm9xRDBRbVdUSlNLOVNcbkFZRjJ6YkljYzBsZjFRMDVTUGxXMURDQ0FMVUpBaUVBazJFRWpKdkdFRnl0YzBWbXRoRTA5bFpGeFFkUmlGN21cbk9pdE5IZzBOZUJrPVxuLS0tLS1FTkQgQ0VSVElGSUNBVEUtLS0tLVxuIn0=; msToken=qSfWwAdwksuBS5wmQAvpzUsf2ovkFFKefOLSvZDKA1Z_FX7ith-wCknpVQB08kft4ISWp00GHeQBaPwV9tcWJq6xBC-mPnQKNjBVINeOQGvFtSdsfacWMtWpa8x1RJE=; FOLLOW_LIVE_POINT_INFO="MS4wLjABAAAA-jD2lukp--I21BF8VQsmYUqJDbj3FmU-kGQTHl2y1Cw/1682092800000/0/0/1682061497980"; msToken=jDwC5gjTRXV6hrFvGMG-AkOBXHGrt_Mp5NaltB1upOUm0aQnZ7sy7qlSEn2tQbGHShp2X7ayNMDQQlPekSTV9MkxBv56LR9zepTlYNOoqbH_RdjDvbl-MZDrmui3OEE=; tt_scid=hERl4ibL-B89BGXb9wUfsmVkQh-G2dvL2wI0QvEDYLooFsvwoz.q6J4ZA0WErn0Tb9fb; home_can_add_dy_2_desktop="1"'}
import requests
res = requests.get(user_all_mix_link, headers=headers).text
import json
datadict = json.loads(res)
print(datadict["aweme_list"][0]["video"]["bit_rate"])
print(len(datadict["aweme_list"][0]["video"]["bit_rate"]))
if __name__ == "__main__":
# test()
# getMusicInfo()
# getUserAllMixInfo()
# getMixInfo()
# getAwemeInfo()
# getUserInfo()
# getLiveInfo()
pass

90
TikTokUrls.py
View File

@ -1,90 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@Description:TikTok.py
@Date :2023/02/11 13:06:23
@Author :imgyh
@version :1.0
@Github :https://github.com/imgyh
@Mail :admin@imgyh.com
-------------------------------------------------
Change Log :
-------------------------------------------------
'''
class Urls(object):
def __init__(self):
# https://langyue.cc/APIdocV1.0.html
######################################### WEB #########################################
# 首页推荐
self.TAB_FEED = 'https://www.douyin.com/aweme/v1/web/tab/feed/?'
# 用户短信息（给多少个用户secid就返回多少的用户信息）
self.USER_SHORT_INFO = 'https://www.douyin.com/aweme/v1/web/im/user/info/?'
# 用户详细信息
self.USER_DETAIL = 'https://www.douyin.com/aweme/v1/web/user/profile/other/?'
# 用户作品
# cookies 暂时只需要 __ac_signature, s_v_web_id两个参数, 好像会过期
# url 暂时不需要携带 msToken, X-Bogus, _signature
# 每次返回数据很少
# self.USER_POST = 'https://m.douyin.com/web/api/v2/aweme/post/?'
# 2023/02/19 失效
self.USER_POST = 'https://www.douyin.com/aweme/v1/web/aweme/post/?'
# 作品信息
self.POST_DETAIL = 'https://www.douyin.com/aweme/v1/web/aweme/detail/?'
# 用户喜欢A
# 需要 odin_tt
self.USER_FAVORITE_A = 'https://www.douyin.com/aweme/v1/web/aweme/favorite/?'
# 用户喜欢B
self.USER_FAVORITE_B = 'https://www.iesdouyin.com/web/api/v2/aweme/like/?'
# 用户历史
self.USER_HISTORY = 'https://www.douyin.com/aweme/v1/web/history/read/?'
# 用户收藏
self.USER_COLLECTION = 'https://www.douyin.com/aweme/v1/web/aweme/listcollection/?'
# 用户评论
self.COMMENT = 'https://www.douyin.com/aweme/v1/web/comment/list/?'
# 首页朋友作品
self.FRIEND_FEED = 'https://www.douyin.com/aweme/v1/web/familiar/feed/?'
# 关注用户作品
self.FOLLOW_FEED = 'https://www.douyin.com/aweme/v1/web/follow/feed/?'
# 合集下所有作品
# 只需要X-Bogus
self.USER_MIX = 'https://www.douyin.com/aweme/v1/web/mix/aweme/?'
# 用户所有合集列表
# 需要 ttwid
self.USER_MIX_LIST = 'https://www.douyin.com/aweme/v1/web/mix/list/?'
# 直播
self.LIVE = 'https://live.douyin.com/webcast/room/web/enter/?'
self.LIVE2 = 'https://webcast.amemv.com/webcast/room/reflow/info/?'
# 音乐
self.MUSIC = 'https://www.douyin.com/aweme/v1/web/music/aweme/?'
# X-Bogus Path
# 60 秒内,请求同一URI累计超过 600 次,封锁IP 300 秒
# 两个都可以用
# 服务器在国外
# self.GET_XB_PATH = 'https://tiktok.199933.xyz/xb'
# 服务器在国内
self.GET_XB_PATH = 'http://47.115.208.101:9090/xb'
#######################################################################################
if __name__ == '__main__':
Urls()

105
TikTokUtils.py
View File

@ -1,105 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@Description:TikTok.py
@Date :2023/01/27 19:36:18
@Author :imgyh
@version :1.0
@Github :https://github.com/imgyh
@Mail :admin@imgyh.com
-------------------------------------------------
Change Log :
-------------------------------------------------
'''
import random
import re
import requests
import execjs
import os
import sys
import json
from TikTokUrls import Urls
class Utils(object):
def __init__(self):
pass
def generate_random_str(self, randomlength=16):
"""
根据传入长度产生随机字符串
"""
random_str = ''
base_str = 'ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789='
length = len(base_str) - 1
for _ in range(randomlength):
random_str += base_str[random.randint(0, length)]
return random_str
def replaceStr(self, filenamestr: str):
"""
替换非法字符，缩短字符长度，使其能成为文件名
"""
# 匹配 汉字 字母 数字 空格
match = "([0-9A-Za-z\u4e00-\u9fa5]+)"
result = re.findall(match, filenamestr)
result = "".join(result).strip()
if len(result) > 20:
result = result[:20]
# 去除前后空格
return result
def resource_path(self, relative_path):
if getattr(sys, 'frozen', False): # 是否Bundle Resource
base_path = sys._MEIPASS
else:
base_path = os.path.dirname(os.path.abspath(__file__))
return os.path.join(base_path, relative_path)
def getXbogus(self, url, headers=None):
# getXbogus算法开源地址https://github.com/B1gM8c/tiktok
user_agent = headers.get(
'User-Agent') if headers else "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36"
try:
xbogus = execjs.compile(open(self.resource_path(os.path.join("X-Bogus.js"))).read()).call('sign', url,
user_agent)
params = url + "&X-Bogus=" + xbogus
except Exception as e:
# print('[ 错误 ]:X-Bogus算法异常或者本地没有JS环境')
try:
# print('[ 提示 ]:尝试远程调用X-Bogus接口')
response = json.loads(requests.post(
url=Urls().GET_XB_PATH, data={"param": url}, headers=headers).text)
params = response["param"]
xbogus = response["X-Bogus"]
except Exception as e:
print('[ 错误 ]:X-Bogus获取异常')
return
return params
def str2bool(self, v):
if isinstance(v, bool):
return v
if v.lower() in ('yes', 'true', 't', 'y', '1'):
return True
elif v.lower() in ('no', 'false', 'f', 'n', '0'):
return False
else:
return True
# https://www.52pojie.cn/thread-1589242-1-1.html
def getttwid(self):
url = 'https://ttwid.bytedance.com/ttwid/union/register/'
data = '{"region":"cn","aid":1768,"needFid":false,"service":"www.ixigua.com","migrate_info":{"ticket":"","source":"node"},"cbUrlProtocol":"https","union":true}'
res = requests.post(url=url, data=data)
for i, j in res.cookies.items():
return j
if __name__ == "__main__":
pass

150
TikTokWeb.py
View File

@ -1,150 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@Description:TikTok.py
@Date :2023/01/27 19:36:18
@Author :imgyh
@version :1.0
@Github :https://github.com/imgyh
@Mail :admin@imgyh.com
-------------------------------------------------
Change Log :
-------------------------------------------------
'''
from flask import *
from TikTok import TikTok
import argparse
def work(share_link, max_cursor, mode, cookie):
tk = TikTok()
if cookie is not None and cookie != "":
tk.headers["Cookie"] = cookie
url = tk.getShareLink(share_link)
key_type, key = tk.getKey(url)
datalist = None
rawdatalist = None
cursor = None
has_more = None
if key_type == "user":
if mode == 'post' or mode == 'like':
datalist, rawdatalist, cursor, has_more = tk.getUserInfoApi(sec_uid=key, mode=mode, count=35,
max_cursor=max_cursor)
elif mode == 'mix':
datalist, rawdatalist, cursor, has_more = tk.getUserAllMixInfoApi(sec_uid=key, count=35, cursor=max_cursor)
elif key_type == "mix":
datalist, rawdatalist, cursor, has_more = tk.getMixInfoApi(mix_id=key, count=35, cursor=max_cursor)
elif key_type == "music":
datalist, rawdatalist, cursor, has_more = tk.getMusicInfoApi(music_id=key, count=35, cursor=max_cursor)
elif key_type == "aweme":
datalist, rawdatalist = tk.getAwemeInfoApi(aweme_id=key)
elif key_type == "live":
datalist, rawdatalist = tk.getLiveInfoApi(web_rid=key)
datadict = {}
if datalist is not None and datalist != []:
datadict["data"] = datalist
datadict["rawdata"] = rawdatalist
datadict["cursor"] = cursor
datadict["has_more"] = has_more
datadict["status_code"] = 200
else:
datadict["status_code"] = 500
return datadict
def deal(mode=None):
usefuldict = {}
if request.headers.get("content_type") == "application/json":
result = request.get_json(force=True)
else:
result = request.form
share_link = None
cursor = 0
cookie = None
try:
share_link = result["share_link"]
cursor = result["cursor"]
cookie = result["cookie"]
except Exception as e:
usefuldict["status_code"] = 500
try:
if share_link is not None and share_link != "":
usefuldict = work(share_link, cursor, mode, cookie)
usefuldict["status_code"] = 200
except Exception as e:
usefuldict["status_code"] = 500
return jsonify(usefuldict)
app = Flask(__name__)
# 设置编码
app.config['JSON_AS_ASCII'] = False
def argument():
parser = argparse.ArgumentParser(description='抖音去水印工具 使用帮助')
parser.add_argument("--port", "-p", help="Web端口",
type=int, required=False, default=5000)
args = parser.parse_args()
return args
@app.route("/douyin/music", methods=["POST"])
def douyinMusic():
return deal()
@app.route("/douyin/mix", methods=["POST"])
def douyinMix():
return deal()
@app.route("/douyin/user/mix", methods=["POST"])
def douyinUserMix():
return deal(mode="mix")
@app.route("/douyin/user/like", methods=["POST"])
def douyinUserLike():
return deal(mode="like")
@app.route("/douyin/user/post", methods=["POST"])
def douyinUserPost():
return deal(mode="post")
@app.route("/douyin/aweme", methods=["POST"])
def douyinAweme():
return deal()
@app.route("/douyin/live", methods=["POST"])
def douyinLive():
return deal()
@app.route("/douyin", methods=["POST"])
def douyin():
return deal()
@app.route("/", methods=["GET"])
def index():
return render_template("index.html")
if __name__ == "__main__":
args = argument()
app.run(debug=False, host="0.0.0.0", port=args.port)

564
X-Bogus.js
View File

File diff suppressed because one or more lines are too long