下载地址：http://m.pan38.com/download.php?code=SLWPSL 提取码：3332
这个是今年5月份给客户定制开发的一个工具，前前后后大概花费了半个月的时间，改了又写，写了又改，那么我为啥现在要给大家分享出来呢，因为这个源码其实还是存在一些值得大家学习的部分，毕竟留着也没啥用，干脆分享出来。
源码部分：【UI部分和控制部分用易语言 核心采集调用py代码，仅供学习】
易语言部分代码：
版本 2
.程序集 小红书爬虫工具
.程序集变量 python进程, 整数型
.程序集变量 输出管道, 整数型
.程序集变量 输入管道, 整数型
.程序集变量 安全属性, SECURITY_ATTRIBUTES
.子程序 _启动子程序, 整数型
.局部变量 用户主页链接, 文本型
.局部变量 输出文件, 文本型
.局部变量 命令行, 文本型
' 初始化安全属性
安全属性.nLength ＝ 12
安全属性.lpSecurityDescriptor ＝ 0
安全属性.bInheritHandle ＝ 真
' 创建管道
CreatePipe (输出管道, 输入管道, 安全属性, 0)
' 设置启动信息
.局部变量 启动信息, STARTUPINFO
启动信息.cb ＝ 68
启动信息.dwFlags ＝ 位或 (1, 256) ' STARTF_USESHOWWINDOW|STARTF_USESTDHANDLES
启动信息.wShowWindow ＝ 0 ' SW_HIDE
启动信息.hStdOutput ＝ 输出管道
启动信息.hStdError ＝ 输出管道
' 设置进程信息
.局部变量 进程信息, PROCESS_INFORMATION
' 构建Python命令行
命令行 ＝ "python xhs_spider.py " ＋ 用户主页链接 ＋ " > output.txt"
' 创建Python进程
CreateProcess (0, 命令行, 0, 0, 真, 0, 0, 0, 启动信息, 进程信息)
' 等待进程结束
WaitForSingleObject (进程信息.hProcess, -1)
' 读取输出
.局部变量 输出内容, 文本型
.局部变量 读取字节数, 整数型
.局部变量 缓冲区, 字节集
缓冲区 ＝ 取空白字节集 (4096)
ReadFile (输出管道, 缓冲区, 取字节集长度 (缓冲区), 读取字节数, 0)
输出内容 ＝ 到文本 (缓冲区)
' 关闭句柄
CloseHandle (输出管道)
CloseHandle (输入管道)
CloseHandle (进程信息.hThread)
CloseHandle (进程信息.hProcess)
' 显示结果
信息框 ("数据采集完成！结果已保存到output.txt", 0, "小红书爬虫工具")
返回 (0)
.子程序 采集小红书数据, 逻辑型
.参数 用户主页链接, 文本型
.局部变量 结果, 逻辑型
' 调用Python脚本
运行 ("python xhs_spider.py " ＋ 用户主页链接, 假, )
结果 ＝ 取反 (取反 (结果)) ' 简单模拟调用结果
返回 (结果)
.子程序 解析JSON数据, 整数型
.参数 JSON文件, 文本型
.局部变量 文件号, 整数型
.局部变量 JSON内容, 文本型
.局部变量 解析结果, 整数型
文件号 ＝ 打开文件 (JSON文件, , )
JSON内容 ＝ 读入文本 (文件号, )
关闭文件 (文件号)
' 这里应该添加JSON解析代码
' 由于易语言原生不支持JSON解析，需要自行实现或使用第三方模块
返回 (解析结果)
Python部分：
import requests
import json
import time
import random
from bs4 import BeautifulSoup
from urllib.parse import urlparse
class XHSSpider:
def init(self):
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Cookie': 'your_cookie_here'
}
self.session = requests.Session()
self.base_url = "https://www.xiaohongshu.com"
def get_note_links(self, user_url):
"""获取用户主页所有笔记链接"""
note_links = []
page = 1
while True:
params = {
'page': page,
'pageSize': 20
}
try:
response = self.session.get(user_url, headers=self.headers, params=params)
soup = BeautifulSoup(response.text, 'html.parser')
links = soup.find_all('a', {'class': 'note-item'})
if not links:
break
for link in links:
note_links.append(self.base_url + link['href'])
page += 1
time.sleep(random.uniform(1, 3))
except Exception as e:
print(f"获取笔记链接出错: {e}")
break
return note_links
def get_note_detail(self, note_url):
"""获取笔记详情"""
try:
response = self.session.get(note_url, headers=self.headers)
soup = BeautifulSoup(response.text, 'html.parser')
提取笔记基本信息
title = soup.find('h1', {'class': 'title'}).text.strip()
author = soup.find('a', {'class': 'author'}).text.strip()
likes = soup.find('span', {'class': 'like-count'}).text.strip()
提取评论
comments = []
comment_items = soup.find_all('div', {'class': 'comment-item'})
for item in comment_items:
user = item.find('a', {'class': 'user'}).text.strip()
content = item.find('div', {'class': 'content'}).text.strip()
time_str = item.find('span', {'class': 'time'}).text.strip()
comments.append({
'user': user,
'content': content,
'time': time_str
})
return {
'title': title,
'author': author,
'likes': likes,
'comments': comments,
'url': note_url
}
except Exception as e:
print(f"获取笔记详情出错: {e}")
return None
def save_to_json(self, data, filename):
"""保存数据到JSON文件"""
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=4)
def run(self, user_url):
"""主运行方法"""
print("开始采集小红书数据...")
note_links = self.get_note_links(user_url)
print(f"共获取到{len(note_links)}篇笔记")
all_data = []
for link in note_links:
note_data = self.get_note_detail(link)
if note_data:
all_data.append(note_data)
print(f"已采集: {note_data['title']}")
time.sleep(random.uniform(2, 5))
self.save_to_json(all_data, 'xhs_data.json')
print("数据采集完成，已保存到xhs_data.json")
