5-1 抖音短视频抓取需求分析
1.需求分析
1.1 模拟滑动视频和发布者点击发布者
1.2 通过mitmproxy抓取并解析数据
import uiautomator2 as u2
import time
class Douyin(object):
def __init__(self, serial="192.168.0.101"):
# 连接
self.d = u2.connect(serial)
# 调用方法
self.start_app() # 启动
self.handle_watcher() # 监控器
self.size = self.get_windowsize()
# 获取初始时间
self.t0 = time.perf_counter()
def start_app(self):
self.d.app_start(package_name="com.ss.android.ugc.aweme")
def stop_app(self):
"""退出逻辑"""
self.d.watcher.stop()
self.d.app_stop("com.ss.android.ugc.aweme")
self.d.app_clear("com.ss.android.ugc.aweme")
def stop_time(self):
if time.perf_counter() - self.t0 > 20: # s
return True
def handle_watcher(self):
# 通知权限
self.d.watcher.when('//*[@resource-id="com.ss.android.ugc.aweme:id/a4r"]').click()
# 发现滑动查看更多
self.d.watcher.when('//*[@text="滑动查看更多"]').click()
self.d.watcher.when('//*[@text="快速进入TA的个人中心"]').click()
self.d.watcher.start(interval=1)
def get_windowsize(self):
return self.d.window_size()
def swipe_douyin(self):
'''滑动抖音短视频 点击视频发布者头像的操作'''
# 判断是否正常进入抖音界面
if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists(timeout=20):
while True:
if self.stop_time():
self.stop_app()
return
# 查看 是不是正常的发布者(不是广告)
if self.d(resourceId="com.ss.android.ugc.aweme:id/u0").exists:
# 是不是正常的 发布者,点击头像
self.d(resourceId="com.ss.android.ugc.aweme:id/tw").click()
# 返回
self.d(resourceId="com.ss.android.ugc.aweme:id/et").click()
# 判断是否正常进入抖音界面
if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists:
# 滑动: 由下向上滑动;
x1 = int(self.size[0]*0.5)
y1 = int(self.size[1]*0.9)
y2 = int(self.size[1]*0.15)
self.d.swipe(x1, y1, x1, y2)
if __name__ == '__main__':
d = Douyin()
d.swipe_douyin()
5-6 通过mitmproxy解析短视频App返回数据-接口分析
先保证WIN和手机IP地址的一致性
确保 手机上安装fiddle证书
先点掉 Capturing, 抓取的数据包就是连到手机上的数据包;
有两个JS文件,先看第一个
特别注意
在新版本的抖音里,我们是找到的是加密的数据返回;低版本在豌豆荚中可以下载,得到没有加密的数据;
# 个人信息页接口
# https://aweme-eagle.snssdk.com/aweme/v1/user/?user_id
# 滑动视频接口
# https://aweme-eagle.snssdk.com/aweme/v1/feed
1. 个人信息页接口
第一个JS文件
# json格式文件
"wx_nickname_replace":"开心锤锤♪","replaced":false},
"ky_only_predict":0,"school_auth":0,"is_star":false,"general_permission":{"following_follower_list_toast":1},
"share_weibo_desc":"在抖音,记录美好生活!","share_desc":"在抖音,记录美好生活!",
"share_title":"快来加入抖音,让你发现最有趣的我!","share_qrcode_url":{"uri":"671f001b2afba0ae78dc","url_list":["https://p9-dy.byteimg.com/obj/671f001b2afba0ae78dc","https://p6-dy-ipv6.byteimg.com/obj/671f001b2afba0ae78dc","https://p3-dy-ipv6.byteimg.com/obj/671f001b2afba0ae78dc"]}},"activity":{"use_music_count":0,"digg_count":0},"follower_status":0,"duet_setting":0,"unique_id":"2B2B2","geofencing":
2. 滑动视频接口
点击 滑动按钮,重新抓取数据;
5-7 通过mitmproxy解析短视频App返回数据-编写mitmdump解析文件
确保安装好了手机mitmproxy 以及证书(IP变化就要重新安装)
# decode_douyin.py
import json
def response(flow):
"""解析10版本抖音app返回数据"""
# 视频
if 'https://aweme-eagle.snssdk.com/aweme/v1/feed' in flow.request.url:
# 使用json来loads response.text
video_response = json.loads(flow.response.text)
video_list = video_response.get("aweme_list", [])
for item in video_list:
print(item.get("desc"), "") # 视频标题
# 发布者页面
if 'https://aweme-eagle.snssdk.com/aweme/v1/user/?user_id' in flow.request.url:
person_response = json.loads(flow.response.text)
person_info = person_response.get("user", "")
if person_info:
info = {
'nickname': person_info.get("nickname", ""), # 名称
'total_favorited': person_info.get("total_favorited", 0), # 点赞数
'following_count': person_info.get("following_count", 0), # 关注着数量
'douyin_id': person_info.get("unique_id", ""), # 抖音ID
'folllower_count': person_info.get("follower_count", 0) # 粉丝数量
}
print('--------------------------------------------------------------------')
print(info)
启动 mitmdump
(venv) user1@imooc:~/u2_project/douyin$ mitmdump -s decode_douyin.py -p 8889
Loading script decode_douyin.py
Proxy server listening at http://*:8889
# run handle_douyin.py
import uiautomator2 as u2
import time
class Douyin(object):
def __init__(self, serial="192.168.0.101"):
# 连接
self.d = u2.connect(serial)
# 调用方法
self.start_app() # 启动
self.handle_watcher() # 监控器
self.size = self.get_windowsize()
# 获取初始时间
self.t0 = time.perf_counter()
def start_app(self):
self.d.app_start(package_name="com.ss.android.ugc.aweme")
def stop_app(self):
"""退出逻辑"""
self.d.watcher.stop()
self.d.app_stop("com.ss.android.ugc.aweme")
self.d.app_clear("com.ss.android.ugc.aweme")
def stop_time(self):
if time.perf_counter() - self.t0 > 20: # s
return True
def handle_watcher(self):
# 通知权限
self.d.watcher.when('//*[@resource-id="com.ss.android.ugc.aweme:id/a4r"]').click()
# 发现滑动查看更多
self.d.watcher.when('//*[@text="滑动查看更多"]').click()
self.d.watcher.when('//*[@text="快速进入TA的个人中心"]').click()
self.d.watcher.start(interval=1)
def get_windowsize(self):
return self.d.window_size()
def swipe_douyin(self):
'''滑动抖音短视频 点击视频发布者头像的操作'''
# 判断是否正常进入抖音界面
if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists(timeout=20):
while True:
if self.stop_time():
self.stop_app()
return
# 查看 是不是正常的发布者(不是广告)
if self.d(resourceId="com.ss.android.ugc.aweme:id/u0").exists:
# 是不是正常的 发布者,点击头像
self.d(resourceId="com.ss.android.ugc.aweme:id/tw").click()
# 返回
self.d(resourceId="com.ss.android.ugc.aweme:id/et").click()
# 判断是否正常进入抖音界面
if self.d(resourceId="com.ss.android.ugc.aweme:id/yy", text="我").exists:
# 滑动: 由下向上滑动;
x1 = int(self.size[0]*0.5)
y1 = int(self.size[1]*0.9)
y2 = int(self.size[1]*0.15)
self.d.swipe(x1, y1, x1, y2)
if __name__ == '__main__':
d = Douyin()
d.swipe_douyin()
--------------------------------------------------------------------
{'nickname': '七猫探长', 'total_favorited': 4012853, 'following_count': 89, 'douyin_id': 'qimao2021', 'folllower_count': 550240}
实战2:小红书
1.抓取需求分析
抓取 推荐也文章基本信息(文章标题、类型、基本描述、用户信息等)
文章详细内容
文章评论
2. 滑动视频接口
import uiautomator2 as u2
import time, datetime
import adbutils, multiprocessing
from elasticsearch import Elasticsearch, helpers
es = Elasticsearch(hosts="192.168.0.101:9200")
class Xiaohongshu(object):
def __init__(self, serial="192.168.0.103:5555"):
'''
手机IP:192.168.0.103
Linux系统IP : 192.168.0.101
desc: V1:刷新新的内容, V2:下一版本功能:爬取文章详细内容和评论数
:param serial:
'''
self.d = u2.connect(serial)
self.size = self.get_window_size()
self.start_app()
def start_app(self, name='com.xingin.xhs'):
self.d.app_start(package_name=name)
def get_window_size(self):
return self.d.window_size()
def swipe_screen(self):
x1 = int(self.size[0] * 0.5)
y1 = int(self.size[1] * 0.9)
y2 = int(self.size[1] * 0.25)
self.d.swipe(x1, y2, x1, y1)
if __name__ == '__main__':
d = Xiaohongshu()
time.sleep(3)
flag = True
while flag:
d.swipe_screen()
# time.sleep(1)
print(es.search(index="xiaohongshu_2021-03-17"))
3.mitmproxy解析短视频App返回数据
import json
import time
import datetime
from elasticsearch import Elasticsearch, helpers
es = Elasticsearch(hosts="192.168.0.101:9200")
def handle_date(key):
"""处理时间方法"""
date_touple = time.localtime()
date = None
if key == "hms":
date = time.strftime("%Y-%m-%d %H:%M:%S", date_touple)
date = date.replace(' ', 'T')
elif key == "bir":
date = time.strftime("%Y-%m-%d", date_touple)
return date
def response(flow):
'''
解析 小红书 最新版本 app 的 目录页数据;
每次滑动拿到 10条数据;
'''
if 'https://edith.xiaohongshu.com/api/sns/v6/homefeed' in flow.request.url:
article_response = json.loads(flow.response.text)
article_list = article_response.get('data', [])
data_list = []
for item in article_list:
info ={
"_index": "xiaohongshu_%s" % handle_date(key="bir"),
"_source": {
"@timestamp": handle_date(key="hms"),
"birthday": handle_date(key="bir"),
"title": item.get("display_title", ""),
"name": item.get("user", "").get("nickname", ""),
"userid": item.get("user", "").get("userid", ""),
"type": item.get("type", ""),
"display_image":item.get("images_list", "").get("url_size_large", "")
"desc": item.get("desc", ""),
"likes": item.get("likes", "")
}
}
print(info)
data_list.append(info)
# 通过helpers插入数据
helpers.bulk(es, data_list)
5. 查看10分钟爬取效果
3284/10m
get xiaohongshu_2021-03-17/_search
#---------------------------------------------------------
{
"took" : 2,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"skipped" : 0,
"failed" : 0
},
"hits" : {
"total" : {
"value" : 3284,
"relation" : "eq"
},
"max_score" : 1.0,
"hits" : [
{
"_index" : "xiaohongshu_2021-03-17",
"_type" : "_doc",
"_id" : "4d3sPXgBy2dlyd-YuFIb",
"_score" : 1.0,
"_source" : {
"@timestamp" : "2021-03-17T10:03:47",
"birthday" : "2021-03-17",
"title" : "考研复试简历模板大放送,请查收㊙️",
"name" : "研学长",
"userid" : "5d8f85a600000000010036ac",
"type" : "normal",
"desc" : """🌈哈喽,各位21/22考研的小伙伴们
❤️在考研复试中,我们为了让导师更加快速全面的了解我们,通常都需要准备一份精美的简历进行自我介绍,一份好的模板通常能让导师对你有更好的印象与了解!也是你个人形象的""",
"likes" : 5813
}
},
注意
1.解决kibana 超过10000条数据无法查询的问题
get xiaohongshu_2021-03-17/_settings
{
"max_result_window" : 200000000
}
网友评论