Python 小红书数据采集工具实现指南 | 极客日志

Python 小红书数据采集工具实现指南 | 极客日志

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows
# 安装依赖
pip install -r requirements.txt

git clone https://gitcode.com/gh_mirrors/xh/xhs
cd xhs
python setup.py install

[request]
timeout = 15
retry_times = 3
proxy_pool = http://proxy1:port,http://proxy2:port

[auth]
cookie_storage_path = ./cookies
qrcode_login_timeout = 300

[spider]
crawl_interval = 1.2
max_concurrent = 10

from xhs import XHS

# 基础初始化
xhs = XHS(
    timeout=10,
    proxies={"http": "http://127.0.0.1:7890"},
    sign=lambda uri, data: custom_sign(uri, data)  # 自定义签名函数
)

# 加载 Cookie
xhs.cookie = "your_cookie_string"

# 二维码登录
qr_code = xhs.get_qrcode()
qr_code.save("qrcode.png")  # 保存二维码图片
print("请扫描二维码登录")
login_status = xhs.check_qrcode(qr_code['qr_id'], qr_code['code'])

# 手机验证码登录
xhs.send_code(phone="13800138000")
code = input("请输入验证码:")
xhs.login_code(phone="13800138000", mobile_token=code)

notes = xhs.get_note_by_keyword(
    keyword="美食推荐",
    page=1,
    page_size=20,
    sort=SearchSortType.HOT  # 按热度排序
)

user_notes = xhs.get_user_all_notes(
    user_id="user123456",
    crawl_interval=1  # 控制请求间隔
)

comments = xhs.get_note_all_comments(
    note_id="note67890",
    crawl_interval=0.5
)

# 解析笔记数据
for note in notes:
    structured_data = {
        "note_id": note["id"],
        "title": note["title"],
        "content": note["desc"],
        "author_id": note["user"]["id"],
        "stats": {
            "likes": note["stats"]["like_count"],
            "comments": note["stats"]["comment_count"],
            "shares": note["stats"]["share_count"]
        },
        "images": xhs.get_imgs_url_from_note(note)
    }
    # 存储到数据库或 JSON 文件

# 获取竞品所有笔记
competitor_notes = xhs.get_user_all_notes(user_id="competitor_id")

# 内容主题分析
from collections import Counter
topics = [note["topics"][0]["name"] for note in competitor_notes if note["topics"]]
topic_counts = Counter(topics)
print("热门主题分布:", topic_counts.most_common(10))

import pandas as pd

# 按时间维度采集数据
trend_data = []
for date in date_range(start_date, end_date):
    notes = xhs.get_note_by_keyword(
        keyword="夏季穿搭",
        sort=SearchSortType.NEWEST
    )
    trend_data.append({
        "date": date,
        "count": len(notes),
        "avg_likes": sum(n["stats"]["like_count"] for n in notes)/len(notes)
    })

# 生成趋势图表
pd.DataFrame(trend_data).plot(x="date", y="count")

# 监听特定关键词
while True:
    latest_notes = xhs.get_note_by_keyword(
        keyword="品牌名称",
        sort=SearchSortType.NEWEST,
        page_size=10
    )
    for note in latest_notes:
        comments = xhs.get_note_comments(note_id=note["id"])
        for comment in comments:
            if "负面" in comment["content"]:
                send_alert(comment)  # 触发预警机制
    time.sleep(300)  # 每 5 分钟检查一次

def custom_sign(uri, data):
    ctime = int(time.time() * 1000)  # 实现自定义签名逻辑
    return generate_signature(uri, data, ctime, a1, b1)

from itertools import cycle
proxy_list = ["http://proxy1:port", "http://proxy2:port"]
proxy_cycle = cycle(proxy_list)

# 每次请求更换代理
def rotate_proxy():
    return {"http": next(proxy_cycle)}

xhs = XHS(proxies=rotate_proxy())

# 随机 UA 生成
from fake_useragent import UserAgent
ua = UserAgent()
xhs = XHS(user_agent=ua.random)

# 随机请求间隔
import random
def random_interval():
    return random.uniform(0.8, 2.0)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def safe_request(note_id):
    return xhs.get_note_by_id(note_id)

def compliance_check(url):
    # 检查是否为允许采集的 URL
    allowed_domains = ["xiaohongshu.com/api", "xiaohongshu.com/notes"]
    return any(domain in url for domain in allowed_domains)