当前位置：首页 > news >正文

DrissionPage实战案例：小红书旅游数据爬取

news 2025/8/8 13:13:48

DrissionPage实战案例：小红书旅游数据爬取

目标

爬取小红书旅游相关内容数据，包括：

笔记标题
图片数量
点赞数
收藏数
评论数
详情页URL
下载笔记中的所有图片

技术栈

DrissionPage：用于自动化浏览器操作和数据包监听
Requests：用于发送HTTP请求
正则表达式：用于解析HTML内容
CSV模块：用于数据存储

代码

# 导入模块
import csv

from DrissionPage import ChromiumPage
import requests
import re
import time
import random
import os


# 创建xhs_data来存储数据
base_dir = 'xhs_data'
if not os.path.exists(base_dir):os.mkdir(base_dir)

# 创建页面
page = ChromiumPage()
# 获取标签页
tab = page.latest_tab
# 开始监听数据包
tab.listen.start('api/sns/web/v1/search/notes')
print('准备打开页面，等待页面加载完成')
# 打开页面
tab.get('https://www.xiaohongshu.com/search_result?keyword=%25E6%2597%2585%25E6%25B8%25B8&source=web_explore_feed')
# 等待页面加载完成
tab.wait.load_start()
"""获取数据"""
# 获取第一个数据包
res = tab.listen.wait()
# 获取响应的数据
dict_data = res.response.body
items_list = []
data_count = 1
try:while True:# 退出死循环条件if dict_data['data']['has_more'] == False:print('总共加载数据条数为', len(items_list))breakelse:items = dict_data['data']['items']items_list.append(items)print(f'第{data_count}个数据包获取完成')time.sleep(random.uniform(1, 3))data_count += 1# 第1次滚动tab.scroll.to_bottom()# 获取下一个数据包，并且等待时间为5秒res = tab.listen.wait(timeout=5)# 如果第一次滚动之后没有数据包加载，则继续进行滚动if not res:tab.run_js('window.scrollBy(0,2500)')# 获取数据包，并且时间设置为5秒res = tab.listen.wait(timeout=5)# 如果还没有数据包，那么就表示已经全部加载完毕了if not res:print('数据可能到达底部')# 获取响应的数据dict_data = res.response.body
except Exception as e:print('获取数据包失败',e)
# 构造请求头
headers = {'user-angent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36','cookie':'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
}
try:with open(os.path.join(base_dir,'xhs.csv'),'w',newline='',encoding='utf-8') as f:# 定义表头,并且写入fieldnames = ['标题', '图片数量', '点赞数', '评论数', '收藏数', '详情页url']writer = csv.DictWriter(f, fieldnames=fieldnames)writer.writeheader()"""处理数据"""for items in items_list:try:for i,item in enumerate(items):
# 定义字典，用于打印结果，并且写入csv文件my_data_dict = {'标题': '','图片数量': '','点赞数': '','评论数': '','收藏数': '','详情页url': ''}
# 获取发布笔记的idnote_id = item['id']# 获取发布笔记的xsec_tokenxsec_token = item['xsec_token']# 分析详情页url地址# https://www.xiaohongshu.com/explore/679e13ca000000002902c13b?xsec_token=AB2xkQBCe8dz6uVelB_qg4jrO1UjgHPHU64XBQ8Qd236w=&xsec_source=pc_search&source=web_explore_feed# 发现679e13ca000000002902c13b是获取的id，AB2xkQBCe8dz6uVelB_qg4jrO1UjgHPHU64XBQ8Qd236w=&xsec_source=为获取的xsec_token# 构造urlurl = f'https://www.xiaohongshu.com/explore/{note_id}?xsec_token={xsec_token}&xsec_source=pc_search&source=web_explore_feed'my_data_dict['详情页url'] = urltry:# 随机1~3秒，发送请求time.sleep(random.uniform(1, 3))# 发送请求，获取响应response = requests.get(url, headers=headers)# 如果响应码为200if response.status_code == 200:html = response.text# print(html)# 提取title数据title_match = re.search(r'<title>(.*?) - 小红书</title>',html,re.S)if title_match:old_title = title_match.group(1)# 进行安全名的处理title = re.sub(r'[\\/:*?"<>|\t\r\n]', '', old_title)[:50]# 创建目录，用来存放图片note_file = os.path.join(base_dir,title)my_data_dict['标题'] = titleif not os.path.exists(note_file):os.makedirs(note_file)else:my_data_dict['标题'] = '无法获取'print(f'第{i+1}篇笔记无法获取标题')# 从HTML中提取所有的图片链接image_urls = re.findall(r'<meta name="og:image" content="(.*?)">', html,re.S)if image_urls and title:print(f'获取到{len(image_urls)}张图片~')my_data_dict['图片数量'] = len(image_urls)# 再次发送请求，请求下载图片for j,image_url in enumerate(image_urls):response_img_download = requests.get(image_url,headers=headers,timeout=20)# 如果状态码等于200，则执行以下代码if response_img_download.status_code == 200:if os.path.exists(note_file):# 如果文件夹存在，则写入# 写入文件try:with open(os.path.join(note_file, f'image_{j}.jpg'), 'wb') as f:f.write(response_img_download.content)print(f'已保存{j + 1}张图片')except Exception as e:print(f'保存第{j + 1}图片失败')# 非则先创建文件夹，再写入else:os.makedirs(note_file)try:with open(os.path.join(note_file, f'image_{j}.jpg'), 'wb') as f:f.write(response_img_download.content)print(f'已保存{j + 1}张图片')except Exception as e:print(f'保存第{j + 1}图片失败')else:print(f'请求第{j}张图片失败,无法获取响应')else:print(f'第{i+1}篇笔记无法获取详情页图片url,或者因为没有获取到标题，无法创建文件')my_data_dict['图片数量'] = '无法获取'

# 获取点赞数like = re.search(r'<meta name="og:xhs:note_like" content="(.*?)">',html,re.S)if not like:my_data_dict['点赞数'] = '无法获取'else:my_data_dict['点赞数'] = like.group(1)# 获取收藏数collect = re.search(r'<meta name="og:xhs:note_collect" content="(.*?)">',html,re.S)if not collect:my_data_dict['收藏数'] = '无法获取'else:my_data_dict['收藏数'] = collect.group(1)# 获取评论数comment = re.search(r'<meta name="og:xhs:note_comment" content="(.*?)">',html,re.S)if not comment:my_data_dict['评论数'] = '无法获取'else:my_data_dict['评论数'] = comment.group(1)print(my_data_dict)try:writer.writerow(my_data_dict)except Exception as e:print('写入文件失败',e)else:print('没有成功获取响应')except Exception as e:print(f'没有成功获取响应,错误内容为:{e}')
except Exception as e:print('处理数据发送错误,',e)
except Exception as e:print('打开文件失败',e)

查看全文

http://www.lryc.cn/news/613450.html