本文实例为大家分享了python批量爬取下载抖音视频的具体代码,供大家参考,具体内容如下
进口操作系统
进口的要求
进口再保险
导入系统
进口asyncio
进口aiohttp
头={
“用户代理”:“Mozilla/5.0 (iPhone;CPU iPhone OS 11 _0像Mac OS X) AppleWebKit/604.1.38 (KHTML,像壁虎一样)”
的版本/移动/15 a372 Safari 11.0/604.1”
}
VIDEO_URLS、页面=[],1
def get_info (url):
”“”
:param url:用户的链接
返回:返回名称、dytk user_id参数
”“”
name=没有
dytk=没有
user_id=没有
试一试:=响应请求。get (url,标题=标题)
user_id=response.url.split (“/? [5] .split (“& # 63;”) [0]
name=re.search (r 'class="昵称"祝辞(* & # 63;)& lt; ', response.text) [1]
“dytk dytk=re.search (r:“(* & # 63;)”“, response.text) [1]
除了(TypeError IndexError):
sys.stdout.write(“华陵:输入的链接错误的)
除了requests.exceptions:
sys.stdout.write(“华陵:链接错误的)
最后:
返回名称、user_id dytk
def make_dir(名称):
”“”
建立文件夹
:参数名称:用户名称
返回:
”“”
如果不是os.path.isdir(名称):
os.mkdir(名字)
其他:
通过
def get_all_video (user_id max_cursor dytk):
”“”
获取视频的地址
:param user_id:
:param max_cursor:
:param dytk:
返回:
”“”
url=" https://www.amemv.com/aweme/v1/aweme/post/& # 63;“
params={user_id: user_id,
“数”:21日
“max_cursor”: max_cursor,
“dytk”: dytk}
试一试:=响应请求。get (url=url, params=params,头=标题)
如果响应。status_code==200:
数据=response.json ()
数据在数据(“aweme_list”):
name=data.get (share_info) . get (“share_desc”)
url=data.get(视频). get (play_addr) . get (“url_list”) [0]。替换(‘playwm’,‘玩’)
VIDEO_URLS。追加([名字,url])
如果数据(“has_more”)==1和datas.get (max_cursor) !=0:
全球页面
print (f '收集第{}页页视频”)
页+=1
返回get_all_video (user_id datas.get (max_cursor), dytk)
其他:
打印(“收集完成”)
返回VIDEO_URLS
其他:
打印(“状态码:“,response.status_code)
回来没有
除了例外e:
打印(“华陵:”,e)
返回
异步def download_video(指数、名称、video_name url):
”“”
下载视频
:param指数:视频id
:参数名称:用户名称
:param video_name:视频名称
:param url:下载url
返回:
”“”
print (f '正在下载第{指数}个视频:{video_name}’)
video_path={}/{} mp4。格式(名称、video_name)
如果不是os.path.isfile (video_path):
试一试:
异步与aiohttp.ClientSession会话():
异步会话。get (url=url,标题=头,ssl=False)作为回应:
张开(video_path wb) f:
而真正的:
块=等待response.content.read (1024)
f.write(块)
如果不是块:
打破
print (f '下载完成第{指数}个视频:{video_name}’)
除了例外e:
print(“华陵:下载faild”video_name e)
返回
其他:
打印(“文件已存在的)
def main ():
url=' http://v.douyin.com/dEorkn/'
名称、user_id dytk=get_info (url)
如果不是(名称、user_id dytk):
返回
make_dir(名字)
get_all_video (user_id 0 dytk)
打印(f“{name}:总共有{len (VIDEO_URLS)}个视频”)
任务=[]
指数,项列举(VIDEO_URLS 1):
video_name[0]=项目
url=项目[1]
tasks.append (asyncio。ensure_future (download_video(指数、名称、video_name url)))
循环=asyncio.get_event_loop ()
loop.run_until_complete (asyncio.wait(任务)
loop.run_until_complete (asyncio.sleep (0))
loop.close ()
打印(f“{name}视频下载完成!”)
if __name__==癬_main__”:
main ()
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
