python抖音批量下载作者所有视频源码,纯官方api,无第三方接口,直接放源码文件已重构,旧版本主代码依旧可用,新版本代码,改为多文件了,可以直接压缩包下载。
直接运行,然后输入分享复制的链接就好
2022.5.20日更新
1. 增加重试,如果下载失败,会重试3次,每次会随机一个新的user-agent,如果还是失败会保存失败的链接到作者目录的"下载失败视频.txt"
2. 评论区指出的作者名称有非法字符文件夹建立失败问题,增加了正则过滤,非法字符串替换为_
3. 下载模式添加了关键字匹配,匹配视频的简介文字
4. 添加下载进度条和视频序号
2022.5.21日更新
1. 添加下载成功的视频不会二次下载,原理是下载完成后会在作者同级文件夹写入下载成功的视频唯一id,下载前检查是否有此id,有的话就跳过。
2. 添加图集下载,会自己判断作品是否为图集,如果是的话,会在作者同级目录下建立一个新的文件夹,以序号存储图片
3. 修复了当文件名不合法时,存储为0byte文件的问题
2022.5.22日更新
1. 增加批量下载模式,请在代码或者单文件程序同级目录下放置一个 ”作者主页链接.txt“ 的文件,格式为每行一个作者主页链接,批量模式是检测到该文件有内容时自动触发,如果想再次使用单作者批量下载请清空文件内容或者删除文件即可
2022.5.23日更新
1. 根据 话痨司机啊 的建议,修改存储已下载视频id的方式改为数据库存储,文件为同级目录下的 "douyin.db" 。如果不想重复下载视频,请不要删除这个文件,或者你想下载之前下载成功的视频,请删除这个文件
2. 解决了部分作者无法下载的问题,原因为小概率会出现视频不存在,丢失视频vid,然后直接报错keyerror
3. 解决了当部分视频简介出现"\n"换行符时导致文件无法创建的问题
2022.5.25日更新
1. 感谢 话痨司机啊 的推荐,使用 gooey 库重构为图形界面,由于界面的原因,代码分为了三个文件,帖子的代码依旧可用,新版本代码改为压缩包分享,下面可下载
2. 修复了部分作者拉取报错的问题,控制台版本也修复了这个问题,可以直接复制使用
2022.5.26日更新
1. 再再再次更新,修复了一个老bug了,这个bug前前后后修了三四次了,这次应该是彻底修复了。就是大家下载的时候需要一个sec_uid参数,这个参数的来源是短链接跳转后的长链接里面的一个参数,前前后后我用了正则,数组,都有bug导致提取不到该参数,今天我才发现有个自带的urllib.parse可以直接把path转为字典,这样就不会解析错误了,我是傻逼....
import linecache import os import re from faker import Faker from concurrent.futures import ThreadPoolExecutor, as_completed from requests.adapters import HTTPAdapter from tqdm import tqdm import requests import sqlite3 import urllib.parse def creat_table(table_name): conn = sqlite3.connect('douyin.db') c = conn.cursor() c.execute(f'''CREATE TABLE IF NOT EXISTS t_{table_name} (ID INTEGER PRIMARY KEY AUTOINCREMENT,VID TEXT NOT NULL);''') conn.commit() conn.close() def insert_data(table_name, vid): conn = sqlite3.connect('douyin.db') c = conn.cursor() cursor = c.execute(f"SELECT vid from t_{table_name}") already_have = False for row in cursor: if vid in row: already_have = True if already_have is False: c.execute(f"INSERT INTO t_{table_name} (ID,VID) VALUES (null,{vid})") conn.commit() conn.close() def selet_data(table_name): conn = sqlite3.connect('douyin.db') c = conn.cursor() cursor = c.execute(f"SELECT VID from t_{table_name}") vid_list = [out_exp[0] for out_exp in cursor] return vid_list class Douyin: def __init__(self, url): self.share_url = url self.headers = { 'User-Agent': "Mozilla/5.0 (iPhone; U; CPU like Mac OS X; en) AppleWebKit/420+ (KHTML, like Gecko) Version/3.0 Mobile/1C28 Safari/419.3" } self.sec_uid = None self.uid = None self.nick_name = None def get_user_info(self): resp = requests.get(self.share_url, headers=self.headers) self.sec_uid = 'sec_uid=' + urllib.parse.parse_qs(urllib.parse.urlparse(resp.url).query)['sec_uid'][0] user_info = f'https://www.iesdouyin.com/web/api/v2/user/info/?{self.sec_uid}' resp = requests.get(user_info, headers=self.headers) user_data = { 'signature': resp.json()['user_info']['signature'], 'nickname': resp.json()['user_info']['nickname'], 'aweme_count': resp.json()['user_info']['aweme_count'], 'following_count': resp.json()['user_info']['following_count'], 'total_favorited': resp.json()['user_info']['total_favorited'], 'avatar': resp.json()['user_info']['avatar_larger']['url_list'][0], } self.uid = resp.json()['user_info']['uid'] self.nick_name = re.sub(r'[<|>\/:"*?]', '_', resp.json()['user_info']['nickname']) creat_table(self.uid) return user_data def get_all_video(self): max_cursor = 0 video_has_more = True all_video_list = [] if self.sec_uid is None: self.get_user_info() while video_has_more is True: json_url = f'https://www.iesdouyin.com/web/api/v2/aweme/post/?{self.sec_uid}&' \ f'count=21&max_cursor={max_cursor}' resp = requests.get(json_url, headers=self.headers) video_has_more = resp.json()['has_more'] max_cursor = resp.json()['max_cursor'] video_list = resp.json()['aweme_list'] for i in video_list: try: all_video_list.append({'desc': i['desc'], 'vid': i['video']['vid'], 'aweme_id': i['aweme_id']}) except KeyError: all_video_list.append({'desc': i['desc'], 'vid': None, 'aweme_id': i['aweme_id']}) return all_video_list def down_video(self, down_info, index): alreday_down = False retry = 0 session = requests.session() session.mount('https://', HTTPAdapter(max_retries=5)) alreday_down_video = selet_data(self.uid) for n in alreday_down_video: if down_info['aweme_id'] in n: alreday_down = True break if alreday_down is False: if os.path.exists(self.nick_name) is False: try: os.makedirs(self.nick_name) except FileExistsError: pass if down_info['desc'] == '': down_info['desc'] = down_info['aweme_id'] down_info['desc'] = re.sub(r'[<|>\/:"*?\n]', '_', down_info['desc']) save_name = f'{self.nick_name}/{index.zfill(2)}_{down_info["desc"]}.mp4' if down_info["vid"]: download_url = f'https://aweme.snssdk.com/aweme/v1/play/?video_id={down_info["vid"]}&ratio=1080p' response = session.get(download_url, headers=self.headers, stream=True) while response.content == b'' and retry < 3: self.headers = { 'User-Agent': Faker().chrome() } response = session.get(download_url, headers=self.headers) retry += 1 if response.content: with open(save_name, 'wb') as file: file.write(response.content) insert_data(self.uid, down_info['aweme_id']) else: with open(f'{self.nick_name}/下载失败视频.txt', 'a+') as file: file.write(download_url + '\n') else: n = 0 img_json_url = f'https://www.douyin.com/web/api/v2/aweme/iteminfo/?item_ids={down_info["aweme_id"]}' if os.path.exists(f'{self.nick_name}/{down_info["desc"]}') is False: try: os.makedirs(f'{self.nick_name}/{down_info["desc"]}') except FileExistsError: pass response = session.get(img_json_url, headers=self.headers) for i in response.json()["item_list"][0]["images"]: n += 1 img_url = i["url_list"][0] img_content = session.get(img_url, headers=self.headers).content img_save_name = f'{self.nick_name}/{down_info["desc"]}/{n}.jpg' with open(img_save_name, 'wb') as file: file.write(img_content) def main(share_url, down_all=False): share_url = re.search(r'[a-zA-z]+://[^\s]*', share_url).group() douyin = Douyin(share_url) info = douyin.get_user_info() print(f'作者:{info["nickname"]}\n视频数:{info["aweme_count"]}\n{"-" * 20}\n拉取作者所有作品中...') down_list = douyin.get_all_video() down_mode = '1' if not down_all: down_mode = input(f'{"-" * 20}\n选择下载模式:\n1.全部下载\n2.关键词匹配下载\n') def down_task(down_load): with ThreadPoolExecutor(max_workers=10) as t: obj_list = [] for i in range((len(down_load))): obj = t.submit(douyin.down_video, down_load[i], str(i)) obj_list.append(obj) with tqdm(total=len(down_load), ncols=100) as bar: for x in as_completed(obj_list): bar.update(1) if down_mode == '1': if down_list: down_task(down_list) else: print(f'无视频可下载') elif down_mode == '2': k = 0 filter_down_list = [] keyword = input('请输入关键词:') for v in down_list: if keyword in v['desc']: filter_down_list.append(v) if len(filter_down_list) == 0: print('无匹配记录') else: print(f'共找到 {len(filter_down_list)} 条匹配记录, 开始下载') down_task(filter_down_list) else: print('输入错误') if __name__ == '__main__': share_url_list = linecache.getlines('作者主页链接.txt') if share_url_list: print('已检测到有批量下载文件,进入批量下载模式\n') if input('请选择批量下载模式:\n1.下载所有作者所有视频\n2.手动选择每个作者下载模式\n') == '1': down_all = True else: down_all = False for i in share_url_list: main(i, down_all) print(f'当前任务完成\n{"*" * 30}\n') input('所有任务已完成') else: url = input('输入作者主页分享链接:') main(url) input('任务已完成')
版权声明
文章来源: https://www.uihtm.com/python/19007.html
版权说明:仅限用于学习和研究目的;不得将上述内容用于商业或者非法用途,否则,一切后果请用户自负。我们非常重视版权问题,如有侵权请邮件(44784009#qq.com)与我们联系处理。敬请谅解!