现在人们都喜欢看小视频，我平时没事的时候也喜欢看些小视频，所以今天我就带大家用python爬取一些小视频数据。
分析网页打开目标网站，我们在搜索栏里面搜索一下自己感兴趣的关键字，比如我搜索 “手机” 这个关键字，网站会给我们跳转到另一个网页，我们就在跳转的这个页面进行分析，打开开发者工具，定位到XHR选项，找一下有没有和网页相似的数据内容。
如图所示，可以看到，这个里面就有和网页相似的内容，那么代表我们需要的数据就在这个地址里面，
可以看到图中标箭头的就是我们需要拿到的数据，其中有一个photo-id的数据是最重要的，因为除了拿到这些数据，我们还需要拿到每个视频的评论数据，评论数据需要用到这个id。
因为是post请求，所以必须要传递参数才能返回数据，上图的是我们需要获取视频信息的参数数据
我们随便点击一个视频进入，定位到评论信息的网页，如图所示，这个是我们获取评论信息需要传递的参数数据，可以看到，需要的网址都是一样的，但是传递的post请求参数是不一样的，这个请求参数是需要传递一个photoid，这个id正是我们需要获取视频信息里面的id数据，也就是说我们先获取到视频信息，然后把视频信息里面的这个id给拿出来，传递给评论信息这个参数里面，就可以获取到视频的评论信息了。
案例模块import httpx import json import pandas as pd import wget import os构建视频信息post请求参数，并发送请求
def get_video_post_data(self):
data = {
"operationName":
"visionSearchPhoto",
"variables": {
"keyword": self.search_name,
"pcursor": "",
"page": "search"
},
"query":
"fragment photoContent on PhotoEntity {n idn durationn captionn likeCountn viewCountn realLikeCountn coverUrln photoUrln photoH265Urln manifestn manifestH265n videoResourcen coverUrls {n urln __typenamen }n timestampn expTagn animatedCoverUrln distancen videoRation likedn stereoTypen profileUserTopPhoton __typenamen}nnfragment feedContent on Feed {n typen author {n idn namen headerUrln followingn headerUrls {n urln __typenamen }n __typenamen }n photo {n ...photoContentn __typenamen }n canAddCommentn llsidn statusn currentPcursorn __typenamen}nnquery visionSearchPhoto($keyword: String, $pcursor: String, $searchSessionId: String, $page: String, $webPageArea: String) {n visionSearchPhoto(keyword: $keyword, pcursor: $pcursor, searchSessionId: $searchSessionId, page: $page, webPageArea: $webPageArea) {n resultn llsidn webPageArean feeds {n ...feedContentn __typenamen }n searchSessionIdn pcursorn aladdinBanner {n imgUrln linkn __typenamen }n __typenamen }n}n"
}
return json.dumps(data)
def parse_post_url(self, url, data):
r = httpx.post(url, headers=self.HEADERS, data=data)
return r.content.decode()
解析出需要视频的数据
def get_video_datas(self, json_str):
json_datas = json.loads(json_str)
for json_data in json_datas.get('data').get('visionSearchPhoto').get('feeds'):
yield {
#提取视频作者名字
'ks_user_name': json_data.get('author').get('name'),
#提取视频作者id
'ks_user_id': json_data.get('author').get('id'),
#提取视频id
'ks_video_id': json_data.get('photo').get('id'),
#提取视频名字
'ks_video_name': json_data.get('photo').get('caption'),
#提取视频的url链接
'ks_video_url': json_data.get('photo').get('photoUrl'),
}
构建评论信息post请求参数，并发送请求
def get_comment_post_data(self, photoId):
data = {
"operationName":
"commentListQuery",
"variables": {
"photoId": f"{photoId}",
"pcursor": ""
},
"query":
"query commentListQuery($photoId: String, $pcursor: String) {n visionCommentList(photoId: $photoId, pcursor: $pcursor) {n commentCountn pcursorn rootComments {n commentIdn authorIdn authorNamen contentn headurln timestampn likedCountn realLikedCountn likedn statusn subCommentCountn subCommentsPcursorn subComments {n commentIdn authorIdn authorNamen contentn headurln timestampn likedCountn realLikedCountn likedn statusn replyToUserNamen replyTon __typenamen }n __typenamen }n __typenamen }n}n"
}
return json.dumps(data)
提取评论信息
def get_comment_datas(self, json_str2):
json_datas = json.loads(json_str2)
comment_pcursor = json_datas.get('data').get('visionCommentList').get('pcursor')
for json_data in json_datas.get('data').get('visionCommentList').get('rootComments'):
yield {
#提取评论参数
'comment_pcursor': comment_pcursor,
#提取评论者的名字
'comment_user_name': json_data.get('authorName'),
#提取评论者的id
'comment_user_id': json_data.get('authorId'),
#提取评论者的评论内容
'comment_user_content': json_data.get('content'),
}
保存视频信息
def save_data(self, ks_list):
df = pd.DataFrame(ks_list)
df.to_csv(f'快手{self.search_name}.csv', index=False,encoding='utf-8-sig')
print('数据保存完成！！！！！！！！！！！！！！')
下载视频
def down_video(self, ks_data):
video_name = ks_data.get('ks_video_name')
path = os.path.join('video', f'{video_name}.mp4')
file_name = wget.download(ks_data.get('ks_video_url'), out=path)
print(file_name)
实现程序的主要逻辑思路
def run(self):
"""实现程序主要逻辑思路
"""
# 1.构建一个空列表，来储存视频的信息数据
ks_list = []
# 2.构建视频信息的post请求参数
video_post_data = self.get_video_post_data()
# 3.发送请求，获取响应数据
json_str = self.parse_post_url(self.API_URL, video_post_data)
# 4.提取需要的视频数据
ks_datas = self.get_video_datas(json_str)
for ks_data in ks_datas:
# 5.下载视频
# self.down_video(ks_data)
# 6.构建评论信息的post请求参数
comment_post_data = self.get_comment_post_data(ks_data.get('ks_video_id'))
# 7.再次发送请求，获取响应数据
json_str2 = self.parse_post_url(self.API_URL, comment_post_data)
# 8.提取需要的评论信息数据
ks_comment_datas = self.get_comment_datas(json_str2)
for ks_comment_data in ks_comment_datas:
ks_list.append(ks_data)
ks_list.append(ks_comment_data)
# 9.保存数据
self.save_data(ks_list)
效果展示
