技术路线
本项目基于 DrissionPage
库中的 ChromiumPage
模块,使用无头浏览器加载抖音用户主页,并通过监听接口 aweme/post
获取页面返回的 JSON 数据,从中提取视频播放地址并批量下载视频。
核心流程如下:
- 创建无头 Chromium 浏览器实例:
使用ChromiumOptions()
配置无头模式并初始化ChromiumPage
实例。 - 监听 aweme/post 接口数据:
启动监听后加载目标用户主页 URL,等待拦截到包含作品列表的 JSON 响应。 - 解析视频信息:
从响应中提取aweme_list
字段,获取视频标题与播放地址。 - 清洗标题与下载视频:
使用正则表达式去除非法文件名字符,并通过requests
以指定的 cookie 和 UA 请求视频地址,保存为本地.mp4
文件。
另外一种技术路线
直接用 requests
循环分页采集全部作品,只需要递归/循环处理/aweme/post/
内容API
,直到拿完所有 aweme,下面具体说明内容API
中变化的字段。
程序内的"has_more"
表示后续是否还有更多的页面,一般的瀑布流加在页面都会有一个这样的标识。且注意到每次刷新新的页面的时候会有一个post
请求,并且请求url
中有一个字段"max_cursor"
,每次请求目标网站后,都会返回一个新的"max_cursor"
用于下一次的请求url
。因此只需要在每次请求的url
里面修改"max_cursor"
的值即可拿到新的18个视频数据。
代码解析
调用库
import requests
from DrissionPage import ChromiumPage, ChromiumOptions # 自动化模块
import re # 正则表达式模块
import os # 操作系统模块
创建无头浏览器
# 创建无头浏览器实例
co = ChromiumOptions()
co.set_argument('--headless') # 设置为无头模式
co.set_argument('--disable-gpu') # 可选,防止部分渲染问题
Google = ChromiumPage(co)
Google.listen.start("aweme/post") # 启动监听
Google.get(url)
sjb = Google.listen.wait() # 等待页面加载
JSON = sjb.response.body # 获取页面JSON数据
文件获取
headers
包含三个字段:
headers = {
'cookie' : cookie,
'referer' : url,
'user-agent' : user_agent
}
data = JSON['aweme_list']
for i in data:
title = i['desc'].strip().split('\n')[0] # 获取标题
title_ulti = re.sub('[\\/:*?"<>|]~`+=', '', title)# 替换非法字符
获取视频代码片段
抖音视频的文件位于['aweme_list']['video']['play_addr']['url_list'][0-2]
# 视频
if i['images'] is None: #纯视频的值为none,确保不下载动图视频的音乐
try:
video_url = i['video']['play_addr']['url_list'][0] # 获取视频地址
res_video = requests.get(video_url, headers=headers).content # 发送请求获取视频内容
with open(f"media/videos/{title_ulti}.mp4", 'wb') as f:
f.write(res_video)
print(f"视频: {title_ulti} 下载完成") # 打印下载完成信息
except Exception as e:
print(f"{title_ulti} 视频下载出错: {e}")
添加if
判断语句,确保当视频为动图视频的时候,video_url
为对应的BGM,而不是动图。
获取动图代码片段
抖音动图的文件位于['aweme_list']['images']['video']['play_addr']['url_list'][0-2]
# 动图
try:
for idx, j in enumerate(i['images']):
gif_url = j['video']['play_addr']['url_list'][0]
res_gif = requests.get(gif_url, headers=headers).content
filename = f"{title_ulti}_{idx}.mp4"
with open(f"media/videos/{filename}", 'wb') as f:
f.write(res_gif)
print(f"动图:{filename} 下载完成")
except:
print(f"{title_ulti} 不是图片视频。")
获取图片代码片段
抖音动图的文件位于['aweme_list']['images']['url_list'][0-2]
# 图片
try:
for idx, j in enumerate(i['images']):
image_url = j['url_list'][0]
res_image = requests.get(image_url, headers=headers).content
filename = f"{title_ulti}_{idx}.webp"
with open(f"media/images/{filename}", 'wb') as f:
f.write(res_image)
print(f"图片:{filename} 下载完成")
except:
print(f"{title_ulti} 不是图片视频。")
cookies结构化
由于requests.get
内置需要的cookies需要结构化,而不是str形式,因此需要对获取的cookies进行结构化。
def cookie_str_to_dict(cookie):
result = {}
for item in cookie.split("; "):
if "=" in item:
k, v = item.split("=", 1) # 仅分一次,避免后续value出现"="
result[k] = v
return result
完整代码
版本1.0
完整代码
import requests
from DrissionPage import ChromiumPage, ChromiumOptions # 自动化模块
import re # 正则表达式模块
import os # 操作系统模块
url = "" # 爬取用户主页地址
cookie = ""
user_agent = ""
os.makedirs('videos', exist_ok=True) # 创建文件夹
# 创建无头浏览器实例
co = ChromiumOptions()
co.set_argument('--headless') # 设置为无头模式
co.set_argument('--disable-gpu') # 可选,防止部分渲染问题
Google = ChromiumPage(co)
Google.listen.start("aweme/post") # 启动监听
Google.get(url)
sjb = Google.listen.wait() # 等待页面加载
JSON = sjb.response.body # 获取页面JSON数据
# print(JSON) # 打印JSON数据
headers = {
'cookie' : cookie,
'referer' : url,
'user-agent' : user_agent
}
data = JSON['aweme_list']
for i in data:
video_url = i['video']['play_addr']['url_list'][0] # 获取视频地址
title = i['desc'] # 获取视频标题
title_re = re.sub('[\\/:*?"<>|]~`+=', '', title) # 替换非法字符
res = requests.get(video_url, headers=headers).content # 发送请求获取视频内容
with open("videos\\" + title_re +".mp4", 'wb') as f:
f.write(res)
print(f"视频: {title_re} 下载完成") # 打印下载完成信息
print("所有视频下载完成")
Google.quit() # 关闭浏览器
版本2.0
完整代码
import requests
from DrissionPage import ChromiumPage, ChromiumOptions # 自动化模块
import re # 正则表达式模块
import os # 操作系统模块
os.makedirs('media', exist_ok=True) # 创建文件夹
os.makedirs('media/videos', exist_ok=True)
os.makedirs('media/images', exist_ok=True)
# 创建无头浏览器实例
co = ChromiumOptions()
co.set_argument('--headless') # 设置为无头模式
co.set_argument('--disable-gpu') # 可选,防止部分渲染问题
Google = ChromiumPage(co)
Google.listen.start("aweme/post") # 启动监听
Google.get(url)
sjb = Google.listen.wait() # 等待页面加载
JSON = sjb.response.body # 获取页面JSON数据
# print(JSON) # 打印JSON数据
headers = {
'cookie' : cookie,
'referer' : url,
'user-agent' : user_agent
}
data = JSON['aweme_list']
for i in data:
title = i['desc'].strip().split('\n')[0] # 获取标题
title_ulti = re.sub('[\\/:*?"<>|]~`+=', '', title)# 替换非法字符
# 视频
if i['images'] is None: #纯视频的值为none,确保不下载动图视频的音乐
try:
video_url = i['video']['play_addr']['url_list'][0] # 获取视频地址
res_video = requests.get(video_url, headers=headers).content # 发送请求获取视频内容
with open(f"media/videos/{title_ulti}.mp4", 'wb') as f:
f.write(res_video)
print(f"视频: {title_ulti} 下载完成") # 打印下载完成信息
except Exception as e:
print(f"{title_ulti} 视频下载出错: {e}")
# 动图
try:
for idx, j in enumerate(i['images']):
gif_url = j['video']['play_addr']['url_list'][0]
res_gif = requests.get(gif_url, headers=headers).content
filename = f"{title_ulti}_{idx}.mp4"
with open(f"media/videos/{filename}", 'wb') as f:
f.write(res_gif)
print(f"动图:{filename} 下载完成")
except:
print(f"{title_ulti} 不是图片视频。")
# 图片
try:
for idx, j in enumerate(i['images']):
image_url = j['url_list'][0]
res_image = requests.get(image_url, headers=headers).content
filename = f"{title_ulti}_{idx}.webp"
with open(f"media/images/{filename}", 'wb') as f:
f.write(res_image)
print(f"图片:{filename} 下载完成")
except:
print(f"{title_ulti} 不是图片视频。")
print("所有文件下载完成")
Google.quit() # 关闭浏览器
版本3.0
func.download_aweme_list.py
import re # 正则表达式模块
import requests
import os
def download_aweme_list(data, headers, already_download_nums, max_download_num):
if not data:
print("无内容,跳过。")
return
author = data[0]['author']['nickname']# 获取作者ID
os.makedirs(f"./media/{author}/images", exist_ok=True) # 创建作者视频文件夹
os.makedirs(f"./media/{author}/videos", exist_ok=True) # 创建作者图片文件夹
nums = 0 # 初始化下载计数
for i in data:
title = i['desc'].strip().split('\n')[0] # 获取标题
title_ulti = re.sub('[\/:*?"<>|]', '', title)# 替换非法字符
# 视频
if i['images'] is None: #纯视频的值为none,确保不下载动图视频的音乐
try:
video_url = i['video']['play_addr']['url_list'][0] # 获取视频地址
res_video = requests.get(video_url, headers=headers).content # 发送请求获取视频内容
nums += 1 # 增加下载计数
if nums + already_download_nums >= max_download_num:
print(f"已达到最大下载数量 {already_download_nums},停止下载。")
break
with open(f"media/{author}/videos/{title_ulti}.mp4", 'wb') as f:
f.write(res_video)
print(f"视频: {title_ulti} 下载完成") # 打印下载完成信息
continue
except Exception as e:
print(f"{title_ulti} 视频下载出错: {e}")
# 动图
for idx, j in enumerate(i['images']):
try:
gif_url = j['video']['play_addr']['url_list'][0]
res_gif = requests.get(gif_url, headers=headers).content
filename = f"{title_ulti}_{idx}.mp4"
with open(f"media/{author}/videos/{filename}", 'wb') as f:
f.write(res_gif)
print(f"动图:{filename} 下载完成")
except Exception as e:
print(f"{title_ulti} 不是图片视频。报错信息:{e}")
# 图片
for idx, j in enumerate(i['images']):
try:
image_url = j['url_list'][0]
res_image = requests.get(image_url, headers=headers).content
filename = f"{title_ulti}_{idx}.webp"
with open(f"media/{author}/images/{filename}", 'wb') as f:
f.write(res_image)
print(f"图片:{filename} 下载完成")
except:
print(f"{title_ulti} 不是图片视频。报错信息:{e}")
nums += 1 # 增加下载计数
if nums + already_download_nums >= max_download_num:
print(f"已达到最大下载数量 {already_download_nums},停止下载。")
break
return nums # 返回下载的视频数量
main.py
import requests
import time
from func.download_aweme_list import download_aweme_list # 下载函数
# ====== 配置区 ======
sec_user_id = ""
cookie_str = ""
user_agent = ""
max_download_num = 40 # 可自定义
# ====================
# 将 cookie 字符串转换为结构化列表
def cookie_str_to_dict(cookie):
result = {}
for item in cookie.split("; "):
if "=" in item:
k, v = item.split("=", 1)
result[k] = v
return result
headers = {
"user-agent": user_agent,
"accept": "application/json, text/plain, */*",
"referer": f"https://www.douyin.com/user/{sec_user_id}",
}
cookies = cookie_str_to_dict(cookie_str)
max_cursor = 0
has_more = True
already_download_nums = 0
while has_more and already_download_nums < max_download_num:
url = (
f"https://www.douyin.com/aweme/v1/web/aweme/post/"
f"?device_platform=webapp"
f"&aid=6383"
f"&channel=channel_pc_web"
f"&sec_user_id={sec_user_id}"
f"&max_cursor={max_cursor}"
f"&locate_query=false"
f"&show_live_replay_strategy=1"
f"&need_time_list=0"
f"&time_list_query=0"
f"&whale_cut_token="
f"&cut_version=1"
f"&count=18"
f"&publish_video_strategy_type=2"
f"&from_user_page=1"
f"&update_version_code=170400"
f"&pc_client_type=1"
f"&pc_libra_divert=Windows"
f"&support_h265=1"
f"&support_dash=1"
f"&cpu_core_num=20"
f"&version_code=290100"
f"&version_name=29.1.0"
f"&cookie_enabled=true"
f"&screen_width=1920"
f"&screen_height=1080"
f"&browser_language=zh-CN"
f"&browser_platform=Win32"
f"&browser_name=Chrome"
f"&browser_version=138.0.0.0"
f"&browser_online=true"
f"&engine_name=Blink"
f"&engine_version=138.0.0.0"
f"&os_name=Windows"
f"&os_version=10"
f"&device_memory=8"
f"&platform=PC"
f"&downlink=10"
f"&effective_type=4g"
f"&round_trip_time=0"
# 下面参数建议每次抓包更新,能用多久看实际测试(校验很强时需每次刷新)
f"&webid="
f"&uifid="
f"&verifyFp="
f"&fp="
f"&msToken="
f"&a_bogus="
)
try:
r = requests.get(url, headers=headers, cookies=cookies, timeout=10)
r.raise_for_status()
data = r.json()
except Exception as e:
print("请求失败:", e)
break
aweme_list = data.get("aweme_list", [])
if not aweme_list:
print("无更多作品")
break
# 下载当前页数据
nums = download_aweme_list(
aweme_list, headers, already_download_nums, max_download_num
)
already_download_nums += nums
print(f"累计下载:{already_download_nums} 个")
# 翻页判断
has_more = data.get("has_more", 0) == 1 and already_download_nums < max_download_num
max_cursor = data.get("max_cursor", 0)
time.sleep(1.5) # 防ban
print(f"全部下载完成,总计:{already_download_nums}个")
一些烧烤
requests.get 与 headers中cookies的结构
如果有:
headers = {
"Cookies": cookie_str,
..................
}
requests.get(headers = headers, cookies = cookies)
其中headers
里面的cookies
为字符串str,request.get
里面的cookies
为结构式,例如字典。
其实并不需要结构化cookies
后的字典形式cookies
,只需要在headers
里面传入cookies
。还有一点就是,假设headers
和request.get
中都存在cookies
,优先使用headers
里面的。