spider/Douyin_TikTok_Download_API
1
0
Fork 0
mirror of https://github.com/Evil0ctal/Douyin_TikTok_Download_API.git synced 2025-04-22 13:26:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

Delete Web directory

Browse Source
This commit is contained in:
Evil0ctal 2022-04-21 04:13:19 -07:00 committed by GitHub
parent e6ed45a223
commit 9e2a16f05a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 0 additions and 911 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Web/API_logs.txt
View File

@ -1,2 +0,0 @@
# @Author: https://github.com/Evil0ctal/
# @Time: 2021/11/06

2
Web/logs.txt
View File

@ -1,2 +0,0 @@
# @Author: https://github.com/Evil0ctal/
# @Time: 2021/11/06

18
Web/requirements.txt
View File

@ -1,18 +0,0 @@
certifi==2021.10.8
charset-normalizer==2.0.12
click==8.1.2
colorama==0.4.4
Flask==2.1.1
idna==3.3
itsdangerous==2.1.2
Jinja2==3.1.1
MarkupSafe==2.1.1
pywebio==1.6.0
requests==2.27.1
retrying==1.3.3
six==1.16.0
tornado==6.1
ua-parser==0.10.0
urllib3==1.26.9
user-agents==2.2.0
Werkzeug==2.1.1

410
Web/scraper.py
View File

@ -1,410 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# @Author: https://github.com/Evil0ctal/
# @Time: 2021/11/06
# @Update: 2022/04/06
# @Function:
# 核心代码，估值1块(๑•̀ㅂ•́)و✧
# 用于爬取Douyin/TikTok数据并以字典形式返回。
import re
import json
import time
import requests
from retrying import retry
class Scraper:
"""
Scraper.douyin():抖音视频/图集解析，返回字典。
Scraper.tiktok():TikTok视频解析，返回字典。
"""
def __init__(self):
self.headers = {
'user-agent': 'Mozilla/5.0 (Linux; Android 8.0; Pixel 2 Build/OPD3.170816.012) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Mobile Safari/537.36 Edg/87.0.664.66'
}
self.tiktok_headers = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
"authority": "www.tiktok.com",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
"Host": "www.tiktok.com",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) coc_coc_browser/86.0.170 Chrome/80.0.3987.170 Safari/537.36",
}
@retry(stop_max_attempt_number=6)
def douyin(self, original_url):
"""
利用官方接口解析抖音链接信息
:param original_url: 抖音/TikTok链接(支持长/短链接)
:return:包含信息的字典
"""
headers = self.headers
try:
# 开始时间
start = time.time()
# 原视频链接
r = requests.get(url=original_url, headers=headers, allow_redirects=False)
try:
# 2021/12/11 发现抖音做了限制，会自动重定向网址，但是可以从回执头中获取
long_url = r.headers['Location']
except:
# 报错后判断为长链接，直接截取视频id
long_url = original_url
# 正则匹配出视频ID
key = re.findall('video/(\d+)?', long_url)[0]
# 构造抖音API链接
api_url = f'https://www.iesdouyin.com/web/api/v2/aweme/iteminfo/?item_ids={key}'
print("正在请求抖音API链接: " + '\n' + api_url)
# 将回执以JSON格式处理
js = json.loads(requests.get(url=api_url, headers=headers).text)
# 判断是否为图集
if js['item_list'][0]['images'] is not None:
print("类型 = 图集")
# 类型为图集
url_type = 'album'
# 图集标题
album_title = str(js['item_list'][0]['desc'])
# 图集作者昵称
album_author = str(js['item_list'][0]['author']['nickname'])
# 图集作者签名
album_author_signature = str(js['item_list'][0]['author']['signature'])
# 图集作者UID
album_author_uid = str(js['item_list'][0]['author']['uid'])
# 图集作者抖音号
album_author_id = str(js['item_list'][0]['author']['unique_id'])
if album_author_id == "":
# 如果作者未修改过抖音号，应使用此值以避免无法获取其抖音ID
album_author_id = str(js['item_list'][0]['author']['short_id'])
# 尝试获取图集BGM信息
try:
# 图集BGM链接
album_music = str(js['item_list'][0]['music']['play_url']['url_list'][0])
# 图集BGM标题
album_music_title = str(js['item_list'][0]['music']['title'])
# 图集BGM作者
album_music_author = str(js['item_list'][0]['music']['author'])
# 图集BGM ID
album_music_id = str(js['item_list'][0]['music']['id'])
# 图集BGM MID
album_music_mid = str(js['item_list'][0]['music']['mid'])
except:
# 报错后代表无背景音乐
# 图集BGM链接
album_music = 'No BGM found'
# 图集BGM标题
album_music_title = 'No BGM found'
# 图集BGM作者
album_music_author = 'No BGM found'
# 图集BGM ID
album_music_id = 'No BGM found'
# 图集BGM MID
album_music_mid = 'No BGM found'
# 图集ID
album_aweme_id = str(js['item_list'][0]['statistics']['aweme_id'])
# 评论数量
album_comment_count = str(js['item_list'][0]['statistics']['comment_count'])
# 获赞数量
album_digg_count = str(js['item_list'][0]['statistics']['digg_count'])
# 播放次数
album_play_count = str(js['item_list'][0]['statistics']['play_count'])
# 分享次数
album_share_count = str(js['item_list'][0]['statistics']['share_count'])
# 上传时间戳
album_create_time = str(js['item_list'][0]['create_time'])
# 将话题保存在列表中
album_hashtags = []
for tag in js['item_list'][0]['text_extra']:
album_hashtags.append(tag['hashtag_name'])
# 将无水印图片链接保存在列表中
images_list = []
for data in js['item_list'][0]['images']:
images_list.append(data['url_list'][0])
# 结束时间
end = time.time()
# 解析时间
analyze_time = format((end - start), '.4f')
# 将信息储存在字典中
album_data = {'status': 'success',
'analyze_time': (analyze_time + 's'),
'url_type': url_type,
'platform': 'douyin',
'original_url': original_url,
'api_url': api_url,
'album_aweme_id': album_aweme_id,
'album_title': album_title,
'album_author': album_author,
'album_author_signature': album_author_signature,
'album_author_uid': album_author_uid,
'album_author_id': album_author_id,
'album_music': album_music,
'album_music_title': album_music_title,
'album_music_author': album_music_author,
'album_music_id': album_music_id,
'album_music_mid': album_music_mid,
'album_comment_count': album_comment_count,
'album_digg_count': album_digg_count,
'album_play_count': album_play_count,
'album_share_count': album_share_count,
'album_create_time': album_create_time,
'album_list': images_list,
'album_hashtags': album_hashtags}
return album_data
else:
print("类型 = 视频")
# 类型为视频
url_type = 'video'
# 视频标题
video_title = str(js['item_list'][0]['desc'])
# 视频作者昵称
video_author = str(js['item_list'][0]['author']['nickname'])
# 视频作者抖音号
video_author_id = str(js['item_list'][0]['author']['unique_id'])
if video_author_id == "":
# 如果作者未修改过抖音号，应使用此值以避免无法获取其抖音ID
video_author_id = str(js['item_list'][0]['author']['short_id'])
# 有水印视频链接
wm_video_url = str(js['item_list'][0]['video']['play_addr']['url_list'][0])
# 无水印视频链接 (在回执JSON中将关键字'playwm'替换为'play'即可获得无水印地址)
nwm_video_url = str(js['item_list'][0]['video']['play_addr']['url_list'][0]).replace('playwm', 'play')
# 去水印后视频链接(2022年1月1日抖音APi获取到的URL会进行跳转，需要在Location中获取直链)
r = requests.get(url=nwm_video_url, headers=headers, allow_redirects=False)
video_url = r.headers['Location']
# 视频作者签名
video_author_signature = str(js['item_list'][0]['author']['signature'])
# 视频作者UID
video_author_uid = str(js['item_list'][0]['author']['uid'])
# 尝试获取视频背景音乐
try:
# 视频BGM链接
video_music = str(js['item_list'][0]['music']['play_url']['url_list'][0])
# 视频BGM标题
video_music_title = str(js['item_list'][0]['music']['title'])
# 视频BGM作者
video_music_author = str(js['item_list'][0]['music']['author'])
# 视频BGM ID
video_music_id = str(js['item_list'][0]['music']['id'])
# 视频BGM MID
video_music_mid = str(js['item_list'][0]['music']['mid'])
except:
# 出错代表无背景音乐
# 视频BGM链接
video_music = 'No BGM found'
# 视频BGM标题
video_music_title = 'No BGM found'
# 视频BGM作者
video_music_author = 'No BGM found'
# 视频BGM ID
video_music_id = 'No BGM found'
# 视频BGM MID
video_music_mid = 'No BGM found'
# 视频ID
video_aweme_id = str(js['item_list'][0]['statistics']['aweme_id'])
# 评论数量
video_comment_count = str(js['item_list'][0]['statistics']['comment_count'])
# 获赞数量
video_digg_count = str(js['item_list'][0]['statistics']['digg_count'])
# 播放次数
video_play_count = str(js['item_list'][0]['statistics']['play_count'])
# 分享次数
video_share_count = str(js['item_list'][0]['statistics']['share_count'])
# 上传时间戳
video_create_time = str(js['item_list'][0]['create_time'])
# 将话题保存在列表中
video_hashtags = []
for tag in js['item_list'][0]['text_extra']:
video_hashtags.append(tag['hashtag_name'])
# 结束时间
end = time.time()
# 解析时间
analyze_time = format((end - start), '.4f')
# 返回包含数据的字典
video_data = {'status': 'success',
'analyze_time': (analyze_time + 's'),
'url_type': url_type,
'platform': 'douyin',
'original_url': original_url,
'api_url': api_url,
'video_title': video_title,
'nwm_video_url': video_url,
'wm_video_url': wm_video_url,
'video_aweme_id': video_aweme_id,
'video_author': video_author,
'video_author_signature': video_author_signature,
'video_author_uid': video_author_uid,
'video_author_id': video_author_id,
'video_music': video_music,
'video_music_title': video_music_title,
'video_music_author': video_music_author,
'video_music_id': video_music_id,
'video_music_mid': video_music_mid,
'video_comment_count': video_comment_count,
'video_digg_count': video_digg_count,
'video_play_count': video_play_count,
'video_share_count': video_share_count,
'video_create_time': video_create_time,
'video_hashtags': video_hashtags}
return video_data
except Exception as e:
# 返回异常
return {'status': 'failed', 'reason': e, 'function': 'Scraper.douyin()', 'value': original_url}
@retry(stop_max_attempt_number=6)
def tiktok(self, original_url):
"""
解析TikTok链接
:param original_url:TikTok链接
:return:包含信息的字典
"""
headers = self.headers
# 开始时间
start = time.time()
# 校验TikTok链接
if original_url[:12] == "https://www.":
original_url = original_url
print("目标链接: ", original_url)
else:
# 从请求头中获取原始链接
response = requests.get(url=original_url, headers=headers, allow_redirects=False)
true_link = response.headers['Location'].split("?")[0]
original_url = true_link
# TikTok请求头返回的第二种链接类型
if '.html' in true_link:
response = requests.get(url=true_link, headers=headers, allow_redirects=False)
original_url = response.headers['Location'].split("?")[0]
print("目标链接: ", original_url)
try:
# 开始获取TikTok数据
tiktok_headers = self.tiktok_headers
html = requests.get(url=original_url, headers=tiktok_headers)
res = re.search('<script id="sigi-persisted-data">(.*)</script><script', html.text).group(1)
resp = re.findall(r'^window\[\'SIGI_STATE\']=(.*)?;window', res)[0]
result = json.loads(resp)
author_id = result["ItemList"]["video"]["list"][0]
# 从网页中获得的视频JSON数据
video_info = result["ItemModule"][author_id]
# print(video_info)
# 使用第三方API获取无水印视频链接（不保证稳定）
s = requests.Session()
api_url = "https://ttdownloader.com/req/"
source = s.get("https://ttdownloader.com/")
token = re.findall(r'value=\"([0-9a-z]+)\"', source.text)
result = s.post(
api_url,
data={'url': original_url, 'format': '', 'token': token[0]}
)
nwm, wm, audio = re.findall(
r'(https?://.*?.php\?v\=.*?)\"', result.text
)
r = requests.get(nwm, allow_redirects=False)
# 整理数据
print("类型 = 视频")
# 无水印视频链接
nwm_video_url = r.headers['Location']
# 有水印视频链接
wm_video_url = video_info['video']['playAddr']
# 类型为视频
url_type = 'video'
# 视频标题
video_title = video_info['desc']
# 视频作者昵称
video_author = video_info['author']
# 视频作者别名
video_author_nickname = video_info['nickname']
# 视频作者抖音号
video_author_id = video_info['authorId']
# 视频作者secid
video_author_SecId = video_info['authorSecId']
# 视频背景音频
video_music = video_info['music']['playUrl']
# 上传时间戳
video_create_time = video_info['createTime']
# 视频ID
video_aweme_id = video_info['video']['id']
# 视频分辨率
video_ratio = video_info['video']['ratio']
# 视频BGM标题
video_music_title = video_info['music']['title']
# 视频BGM作者
video_music_author = video_info['music']['authorName']
# 视频BGM ID
video_music_id = video_info['music']['id']
# 视频BGM链接
video_music_url = video_info['music']['playUrl']
# 评论数量
video_comment_count = video_info['stats']['commentCount']
# 获赞数量
video_digg_count = video_info['stats']['diggCount']
# 播放次数
video_play_count = video_info['stats']['playCount']
# 分享次数
video_share_count = video_info['stats']['shareCount']
# 作者粉丝数量
video_author_followerCount = video_info['authorStats']['followerCount']
# 作者关注数量
video_author_followingCount = video_info['authorStats']['followingCount']
# 作者获赞数量
video_author_heartCount = video_info['authorStats']['heartCount']
# 作者视频数量
video_author_videoCount = video_info['authorStats']['videoCount']
# 作者已赞作品数量
video_author_diggCount = video_info['authorStats']['diggCount']
# 将话题保存在列表中
video_hashtags = []
for tag in video_info['challenges']:
video_hashtags.append(tag['title'])
# 结束时间
end = time.time()
# 解析时间
analyze_time = format((end - start), '.4f')
# 储存数据
video_date = {'status': 'success',
'analyze_time': (analyze_time + 's'),
'url_type': url_type,
'original_url': original_url,
'platform': 'tiktok',
'video_title': video_title,
'nwm_video_url': nwm_video_url,
'wm_video_url': wm_video_url,
'video_author': video_author,
'video_author_nickname': video_author_nickname,
'video_author_id': video_author_id,
'video_author_SecId': video_author_SecId,
'video_music': video_music,
'video_create_time': video_create_time,
'video_aweme_id': video_aweme_id,
'video_ratio': video_ratio,
'video_music_title': video_music_title,
'video_music_author': video_music_author,
'video_music_id': video_music_id,
'video_music_url': video_music_url,
'video_comment_count': video_comment_count,
'video_digg_count': video_digg_count,
'video_play_count': video_play_count,
'video_share_count': video_share_count,
'video_author_followerCount': video_author_followerCount,
'video_author_followingCount': video_author_followingCount,
'video_author_heartCount': video_author_heartCount,
'video_author_videoCount': video_author_videoCount,
'video_author_diggCount': video_author_diggCount,
'video_hashtags': video_hashtags
}
# 返回包含数据的字典
return video_date
except Exception as e:
# 异常捕获
return {'status': 'failed', 'reason': e, 'function': 'Scraper.tiktok()', 'value': original_url}
if __name__ == '__main__':
# 测试类
scraper = Scraper()
tiktok_url = "https://www.tiktok.com/@oregonzoo/video/7074995215647477034"
tiktok_date = scraper.tiktok(tiktok_url)
print(tiktok_date)
print('')
douyin_url = "https://www.douyin.com/video/7055581212840086817"
douyin_date = scraper.douyin(douyin_url)
print(douyin_date)

479
Web/web_zh.py
View File

@ -1,479 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# @Author: https://github.com/Evil0ctal/
# @Time: 2021/11/06
# @Update: 2022/04/17
# @Function:
# 用于在线批量解析Douyin/TikTok的无水印视频/图集。
# 基于 PyWebIO、Flask, 将scraper.py返回的内容显示在网页上。
# 默认运行端口5000, 请自行在文件底部修改。
import os
import re
import time
import json
import tarfile
import requests
from scraper import Scraper
from pywebio import config, session
from pywebio.input import *
from pywebio.output import *
from pywebio.platform.flask import webio_view
from flask import Flask, request, jsonify
app = Flask(__name__)
title = "抖音/TikTok无水印在线解析"
description = "支持在线批量解析下载无水印抖音/TikTok的无水印视频/图集。支持API调用，开源，免费，无广告。"
headers = {
'user-agent': 'Mozilla/5.0 (Linux; Android 8.0; Pixel 2 Build/OPD3.170816.012) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Mobile Safari/537.36 Edg/87.0.664.66'
}
def loading():
# 写一个进度条装装样子吧 :)
set_scope('bar', position=3)
with use_scope('bar'):
put_processbar('bar')
for i in range(1, 4):
set_processbar('bar', i / 3)
time.sleep(0.1)
def find_url(string):
# 解析抖音分享口令中的链接并返回列表
url = re.findall('http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', string)
return url
def valid_check(kou_ling):
# 校验输入的内容
url_list = find_url(kou_ling)
# 对每一个链接进行校验
if url_list:
total_urls = len(url_list)
# 最大接受提交URL的数量
max_urls = 10
if total_urls > max_urls:
return '为了避免资源占用过多请确保每次提交的链接少于10个，如需大量解析请自行部署。'
else:
for i in url_list:
if 'douyin.com' in i[:31]:
if i == url_list[-1]:
return None
elif 'tiktok.com' in i[:31]:
if i == url_list[-1]:
return None
else:
return '请确保输入链接均为有效的抖音/TikTok链接!'
elif kou_ling == 'wyn':
return None
else:
return '抖音分享口令有误!'
def error_do(reason, function, value):
# 输出一个毫无用处的信息
put_html("<hr>")
put_error("发生了了意料之外的错误，输入值已被记录。")
put_html('<h3>⚠详情</h3>')
put_table([
['函数名', '原因', '输入值'],
[function, str(reason), value]])
put_markdown('可能的原因:')
put_markdown('服务器可能被目标主机的防火墙限流(稍等片刻后再次尝试)')
put_markdown('输入了错误的链接(暂不支持主页链接解析)')
put_markdown('该视频已经被删除或屏蔽(你看的都是些啥(⊙_⊙)?)')
put_markdown('你可以在右上角的关于菜单中查看本站错误日志。')
put_markdown('[点击此处在GayHub上进行反馈](https://github.com/Evil0ctal/Douyin_TikTok_Download_API/issues)')
put_html("<hr>")
# 将错误记录在logs.txt中
error_date = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
with open('logs.txt', 'a') as f:
f.write(error_date + ":\n" + function + ': ' + str(reason) + '\n' + "Input value: " + value + '\n')
def clean_filename(string, author_name):
# 替换不能用于文件名的字符('/ \ : * ? " < > |')
rstr = r"[\/\\\:\*\?\"\<\>\|]"
# 将上述字符替换为下划线
new_title = re.sub(rstr, "_", string)
# 新文件名
filename = (author_name + '_' + new_title).replace('\n', '')
return filename
def compress_file(tar_file, target_file):
# tar_file是输出压缩包名字以及目录("./output/mp4.tar")，target_file是要打包的目录或文件名("./files")
if os.path.isfile(target_file):
with tarfile.open(tar_file, 'w') as tar:
tar.add(target_file)
return 'finished'
else:
with tarfile.open(tar_file, 'w') as tar:
for root, dirs, files in os.walk(target_file):
for single_file in files:
filepath = os.path.join(root, single_file)
tar.add(filepath)
return 'finished'
def clean_file(path):
# 清理下载文件夹
while True:
for root, dirs, files in os.walk(path, topdown=False):
for name in files:
os.remove(os.path.join(root, name))
# print("%s文件删除成功 %s" % (name, (time.strftime("%d/%m/%Y%H:%M:%S"))))
for name in dirs:
os.rmdir(os.path.join(root, name))
# print("%s子文件夹下文件删除成功 %s" % (name, (time.strftime("%d/%m/%Y%H:%M:%S"))))
# 每30分钟(1800秒)清理一次
time.sleep(1800)
def video_download_window(result_dict):
try:
# result_dict = {'文件名': '链接'}
total_amount = len(result_dict)
download_time = (time.strftime("%Y_%m_%d_%H_%M_%S", time.localtime()))
# 存储根目录
save_path = './web/saved_videos/' + (download_time + '_total_' + str(total_amount) + '_videos')
# 判断目录是否存在
if not os.path.exists(save_path):
os.makedirs(save_path)
# 弹出窗口
with popup("正在服务器后台下载视频(共{}个下载任务)".format(str(len(result_dict)))):
# 下载索引计数
download_count = 0
# 遍历字典的键和值
for file_name, url in result_dict.items():
try:
download_count += 1
put_info('正在下载第{}个视频:\n{}'.format(download_count, file_name))
response = requests.get(url, headers=headers)
data = response.content
if data:
file_path = '{}/{}.{}'.format(save_path, file_name, 'mp4')
if not os.path.exists(file_path):
with open(file_path, 'wb') as f:
f.write(data)
f.close()
put_success('{}下载成功'.format(file_name))
except Exception as e:
download_count += 1
put_error('视频下载失败，将跳过该视频。')
continue
if download_count == total_amount:
put_html('<hr>')
put_html('<h3>💾结果页视频合集下载完成</h3>')
output_path = save_path + '/output'
tarfile_name = download_time + '_total_' + str(total_amount) + '_videos.tar'
output_file = output_path + '/' + tarfile_name
put_info('正在压缩视频文件，请勿关闭当前弹窗，完成后会在下方显示按钮...')
# 判断目录是否存在
if not os.path.exists(output_path):
os.mkdir(output_path)
if compress_file(tar_file=output_file, target_file=save_path) == 'finished':
tar = open(output_file, "rb").read()
put_file(tarfile_name, tar, '点击下载视频合集压缩包')
except Exception as e:
print(str(e))
def put_douyin_result(item):
# 向前端输出表格
api = Scraper()
# 抖音数据
douyin_date = api.douyin(item)
# API链接
short_api_url = 'https://api.douyin.wtf/api?url=' + item
download_video = 'https://api.douyin.wtf/video?url=' + item
download_bgm = 'https://api.douyin.wtf/music?url=' + item
if douyin_date['status'] == 'success':
if douyin_date['url_type'] == 'video':
put_table([
['类型', '内容'],
['格式:', douyin_date['url_type']],
['视频直链: ', put_link('点击打开视频', douyin_date['nwm_video_url'], new_window=True)],
['视频下载：', put_link('点击下载', download_video, new_window=True)],
['背景音乐直链: ', put_link('点击打开音频', douyin_date['video_music'], new_window=True)],
['背景音乐下载：', put_link('点击下载', download_bgm, new_window=True)],
['视频标题: ', douyin_date['video_title']],
['作者昵称: ', douyin_date['video_author']],
['作者抖音ID: ', douyin_date['video_author_id']],
['原视频链接: ', put_link('点击打开原视频', item, new_window=True)],
['当前视频API链接: ', put_link('点击浏览API数据', douyin_date['api_url'], new_window=True)],
['当前视频精简API链接: ', put_link('点击浏览API数据', short_api_url, new_window=True)]
])
return {'status': 'success',
'type': 'video',
'video_title': douyin_date['video_title'],
'video_author': douyin_date['video_author'],
'nwm_video_url': douyin_date['nwm_video_url'],
'video_music': douyin_date['video_music'],
'original_url': douyin_date['original_url']}
else:
put_table([
['类型', '内容'],
['格式:', douyin_date['url_type']],
['背景音乐直链: ', put_link('点击打开音频', douyin_date['album_music'], new_window=True)],
['背景音乐下载：', put_link('点击下载', download_bgm, new_window=True)],
['视频标题: ', douyin_date['album_title']],
['作者昵称: ', douyin_date['album_author']],
['作者抖音ID: ', douyin_date['album_author_id']],
['原视频链接: ', put_link('点击打开原视频', douyin_date['original_url'], new_window=True)],
['当前视频API链接: ', put_link('点击浏览API数据', douyin_date['api_url'], new_window=True)],
['当前视频精简API链接: ', put_link('点击浏览API数据', 'short_api_url', new_window=True)]
])
for i in douyin_date['album_list']:
put_table([
['图片直链: ', put_link('点击打开图片', i, new_window=True), put_image(i)]
])
return {'status': 'success',
'type': 'album',
'album_title': douyin_date['album_title'],
'video_author': douyin_date['video_author'],
'album_list': douyin_date['album_list'],
'album_music': douyin_date['album_music'],
'original_url': douyin_date['original_url']}
else:
# {'status': 'failed', 'reason': e, 'function': 'API.tiktok()', 'value': original_url}
reason = douyin_date['reason']
function = douyin_date['function']
value = douyin_date['value']
error_do(reason, function, value)
return 'failed'
def put_tiktok_result(item):
# 将TikTok结果显示在前端
api = Scraper()
# TikTok数据
tiktok_date = api.tiktok(item)
if tiktok_date['status'] == 'success':
# API链接
short_api_url = 'https://api.douyin.wtf/api?url=' + item
download_video = 'https://api.douyin.wtf/video?url=' + item
download_bgm = 'https://api.douyin.wtf/music?url=' + item
put_table([
['类型', '内容'],
['视频标题: ', tiktok_date['video_title']],
['视频直链(有水印): ', put_link('点击打开视频', tiktok_date['wm_video_url'], new_window=True)],
['视频直链(无水印): ', put_link('点击打开视频', tiktok_date['nwm_video_url'], new_window=True)],
['视频下载(无水印)：', put_link('点击下载', download_video, new_window=True)],
['音频(名称-作者)：', tiktok_date['video_music_title'] + " - " + tiktok_date['video_music_author']],
['音频播放：', put_link('点击播放', tiktok_date['video_music_url'], new_window=True)],
['作者昵称: ', tiktok_date['video_author_nickname']],
['作者ID: ', tiktok_date['video_author']],
['粉丝数量: ', tiktok_date['video_author_followerCount']],
['关注他人数量: ', tiktok_date['video_author_followingCount']],
['获赞总量: ', tiktok_date['video_author_heartCount']],
['视频总量: ', tiktok_date['video_author_videoCount']],
['原视频链接: ', put_link('点击打开原视频', item, new_window=True)],
['当前视频API链接: ', put_link('点击浏览API数据', short_api_url, new_window=True)]
])
return {'status': 'success',
'type': 'video',
'video_title': tiktok_date['video_title'],
'video_author': tiktok_date['video_author'],
'nwm_video_url': tiktok_date['nwm_video_url'],
'video_music_url': tiktok_date['video_music_url'],
'original_url': item}
else:
# {'status': 'failed', 'reason': e, 'function': 'API.tiktok()', 'value': original_url}
reason = tiktok_date['reason']
function = tiktok_date['function']
value = tiktok_date['value']
error_do(reason, function, value)
return 'failed'
def ios_pop_window():
with popup("iOS快捷指令"):
try:
shortcut = json.loads(requests.get(url='https://api.douyin.wtf/ios', headers=headers).text)
shortcut_link = shortcut['link']
shortcut_note = shortcut['note']
shortcut_update = shortcut['update']
shortcut_version = shortcut['version']
except Exception as e:
shortcut_link = '无法获取快捷指令信息,请到Github上进行反馈。'
shortcut_note = '无法获取快捷指令信息,请到Github上进行反馈。'
shortcut_update = '无法获取快捷指令信息,请到Github上进行反馈。'
shortcut_version = '无法获取快捷指令信息,请到Github上进行反馈。'
put_text('快捷指令需要在抖音或TikTok的APP内，浏览你想要无水印保存的视频或图集。')
put_text('然后点击右下角分享按钮，选择更多，然后下拉找到 "抖音TikTok无水印下载" 这个选项。')
put_text('如遇到通知询问是否允许快捷指令访问xxxx (域名或服务器)，需要点击允许才可以正常使用。')
put_text('该快捷指令会在你相册创建一个新的相薄方便你浏览保存的内容。')
put_html('<hr>')
put_text('最新快捷指令版本: {}'.format(shortcut_version))
put_text('快捷指令更新时间: {}'.format(shortcut_update))
put_text('快捷指令更新内容: {}'.format(shortcut_note))
put_link('[点击获取快捷指令]', shortcut_link, new_window=True)
def api_document_pop_window():
with popup("API文档"):
put_markdown("💽API文档")
put_markdown("API可将请求参数转换为需要提取的无水印视频/图片直链，配合IOS捷径可实现应用内下载。")
put_link('[中文文档]', 'https://github.com/Evil0ctal/Douyin_TikTok_Download_API#%EF%B8%8Fapi%E4%BD%BF%E7%94%A8',
new_window=True)
put_html('<br>')
put_link('[English doc]',
'https://github.com/Evil0ctal/Douyin_TikTok_Download_API/blob/main/README.en.md#%EF%B8%8Fapi-usage',
new_window=True)
put_html('<hr>')
put_markdown("🛰️API参考")
put_markdown('抖音/TikTok解析请求参数')
put_code('https://api.douyin.wtf/api?url="复制的(抖音/TikTok)的(分享文本/链接)"\n#返回JSON')
put_markdown('抖音/TikTok视频下载请求参数')
put_code('https://api.douyin.wtf/video?url="复制的抖音/TikTok链接"\n'
'# 返回mp4文件下载请求\n'
'# 大量请求时很吃服务器内存，容易崩，慎用。')
put_markdown('抖音视频/图集音频下载请求参数')
put_code('https://api.douyin.wtf/music?url="复制的抖音/TikTok链接"\n'
'# 返回mp3文件下载请求\n'
'# 大量请求时很吃服务器内存，容易崩，慎用。')
def log_popup_window():
with popup('错误日志'):
put_html('<h3>⚠️关于解析失败可能的原因</h3>')
put_markdown('服务器可能被目标主机的防火墙限流(稍等片刻后再次尝试)')
put_markdown('输入了错误的链接(暂不支持主页链接解析)')
put_markdown('该视频已经被删除或屏蔽(你看的都是些啥(⊙_⊙)?)')
put_markdown('[点击此处在GayHub上进行反馈](https://github.com/Evil0ctal/Douyin_TikTok_Download_API/issues)')
put_html('<hr>')
put_text('点击logs.txt可下载日志:')
content = open(r'./logs.txt', 'rb').read()
put_file('logs.txt', content=content)
with open('./logs.txt', 'r') as f:
content = f.read()
put_text(str(content))
def about_popup_window():
with popup('更多信息'):
put_html('<h3>👀访问记录</h3>')
put_image('https://views.whatilearened.today/views/github/evil0ctal/TikTokDownload_PyWebIO.svg',
title='访问记录')
put_html('<hr>')
put_html('<h3>⭐Github</h3>')
put_markdown('[Douyin_TikTok_Download_API](https://github.com/Evil0ctal/Douyin_TikTok_Download_API)')
put_html('<hr>')
put_html('<h3>🎯反馈</h3>')
put_markdown('提交：[issues](https://github.com/Evil0ctal/Douyin_TikTok_Download_API/issues)')
put_html('<hr>')
put_html('<h3>🌐视频/图集批量下载</h3>')
put_markdown('可以使用[IDM](https://www.zhihu.com/topic/19746283/hot)之类的工具对结果页面的链接进行嗅探。')
put_markdown('如果你有更好的想法欢迎PR')
put_html('<hr>')
put_html('<h3>💖WeChat</h3>')
put_markdown('微信：[Evil0ctal](https://mycyberpunk.com/)')
put_html('<hr>')
@config(title=title, description=description)
def main():
# 设置favicon
favicon_url = "https://raw.githubusercontent.com/Evil0ctal/Douyin_TikTok_Download_API/main/favicon/android-chrome-512x512.png"
session.run_js("""
$('#favicon32,#favicon16').remove();
$('head').append('<link rel="icon" type="image/png" href="%s">')
""" % favicon_url)
# 修改footer
session.run_js("""$('footer').remove()""")
# 访问记录
view_amount = requests.get("https://views.whatilearened.today/views/github/evil0ctal/TikTokDownload_PyWebIO.svg")
put_markdown("""<div align='center' ><font size='20'>😼抖音/TikTok无水印在线解析</font></div>""")
put_html('<hr>')
put_row([put_button("快捷指令", onclick=lambda: ios_pop_window(), link_style=True, small=True),
put_button("API", onclick=lambda: api_document_pop_window(), link_style=True, small=True),
put_button("日志", onclick=lambda: log_popup_window(), link_style=True, small=True),
put_button("关于", onclick=lambda: about_popup_window(), link_style=True, small=True)
])
placeholder = "批量解析请直接粘贴多个口令或链接，无需使用符号分开，支持抖音和TikTok链接混合，暂时不支持作者主页链接批量解析。"
kou_ling = textarea('请将抖音或TikTok的分享口令或网址粘贴于此', type=TEXT, validate=valid_check, required=True,
placeholder=placeholder,
position=0)
if kou_ling:
if kou_ling == 'wyn':
# 好想你(小彩蛋)
with popup('给 WYN💖'):
put_text('我大约真的没有什么才华，只是因为有幸见着了你，于是这颗庸常的心中才凭空生出好些浪漫。')
put_text('真的好爱你呀！')
put_link('WYN&THB', 'https://www.wynthb.com/')
else:
url_lists = find_url(kou_ling)
total_urls = len(url_lists)
# 解析开始时间
start = time.time()
# 放一个毫无意义的进度条
loading()
# 成功/失败统计
success_count = 0
failed_count = 0
# 解析成功的url
success_list = []
# 解析失败的url
failed_list = []
# 成功解析的视频标题/视频直链
nwm_success_list = {}
# 遍历链接
for url in url_lists:
if 'douyin.com' in url:
result = put_douyin_result(url)
if result == 'failed':
failed_count += 1
# 将url添加到失败列表内
failed_list.append(url)
continue
else:
success_count += 1
# 将url添加到成功列表内
success_list.append(url)
if result['type'] == 'video':
filename = clean_filename(string=result['video_title'], author_name=result['video_author'])
nwm_success_list.update({filename: result['nwm_video_url']})
else:
result = put_tiktok_result(url)
if result == 'failed':
failed_count += 1
# 将url添加到失败列表内
failed_list.append(url)
continue
else:
success_count += 1
# 将url添加到成功列表内
success_list.append(url)
if result['type'] == 'video':
filename = clean_filename(string=result['video_title'], author_name=result['video_author'])
nwm_success_list.update({filename: result['nwm_video_url']})
clear('bar')
# 解析结束时间
end = time.time()
put_html("<br><hr>")
put_text('总共收到' + str(total_urls) + '个链接')
put_text('成功: ' + str(success_count) + ' ' + '失败: ' + str(failed_count))
put_text('解析共耗时: %.4f秒' % (end - start))
put_button("下载结果页中的所有视频", onclick=lambda: video_download_window(nwm_success_list))
put_link('返回主页', '/')
time.sleep(300)
# 清理文件夹
clean_file('./web/saved_videos')
if __name__ == "__main__":
# 初始化logs.txt
date = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
with open('logs.txt', 'a') as f:
f.write("时间: " + date + " " + "程序重载完毕!" + '\n')
app.add_url_rule('/', 'webio_view', webio_view(main), methods=['GET', 'POST', 'OPTIONS'])
# 获取空闲端口
if os.environ.get('PORT'):
port = int(os.environ.get('PORT'))
else:
# 在这里修改默认端口(记得在防火墙放行该端口)
port = 5000
app.run(host='0.0.0.0', port=port)