#!/usr/bin/env python3 """ 小红书笔记搜索工具 根据关键词搜索小红书笔记，支持多种筛选条件 """ import requests import json import os import argparse import time import logging from datetime import datetime from typing import Dict, Any logger = logging.getLogger(__name__) class XiaohongshuSearch: """小红书笔记搜索API封装类""" BASE_URL = "http://47.84.182.56:8001" TOOL_NAME = "xhs_note_search" PLATFORM = "xiaohongshu" def __init__(self, results_dir: str = None, cache_dir: str = "search_cache"): """ 初始化API客户端 Args: results_dir: 结果输出目录，默认为项目根目录下的 data/search 文件夹 cache_dir: 缓存目录，默认为 search_cache """ self.api_url = f"{self.BASE_URL}/tools/call/{self.TOOL_NAME}" # 设置结果输出目录 if results_dir: self.results_base_dir = results_dir else: # 默认使用项目根目录的 data/search 文件夹 script_dir = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(os.path.dirname(script_dir)) self.results_base_dir = os.path.join(project_root, "data", "search") # 设置缓存目录 self.cache_dir = cache_dir if cache_dir: os.makedirs(cache_dir, exist_ok=True) def _get_cache_key( self, keyword: str, content_type: str, sort_type: str, publish_time: str ) -> str: """ 生成缓存key Args: keyword: 搜索关键词 content_type: 内容类型 sort_type: 排序方式 publish_time: 发布时间 Returns: 缓存key字符串 """ return f"{keyword}_{content_type}_{sort_type}_{publish_time}" def _get_cache_path(self, cache_key: str) -> str: """ 获取缓存文件路径 Args: cache_key: 缓存key Returns: 缓存文件完整路径 """ # 清理文件名中的非法字符 safe_key = cache_key.replace('/', '_').replace('\\', '_').replace(' ', '_') return os.path.join(self.cache_dir, f"{safe_key}.json") def search( self, keyword: str, content_type: str = "不限", sort_type: str = "综合", publish_time: str = "不限", cursor: str = "", timeout: int = 30, max_retries: int = 5, retry_delay: int = 2, use_cache: bool = True ) -> Dict[str, Any]: """ 搜索小红书笔记（带重试机制和缓存） Args: keyword: 搜索关键词 content_type: 内容类型，可选值：不限、视频、图文，默认为'不限' sort_type: 排序方式，可选值：综合、最新、最多点赞、最多评论，默认为'综合' publish_time: 发布时间筛选，可选值：不限、一天内、一周内、半年内，默认为'不限' cursor: 翻页游标，第一页默认为空，下一页的游标在上一页的返回值中获取 timeout: 请求超时时间（秒），默认30秒 max_retries: 最大重试次数，默认3次 retry_delay: 重试间隔时间（秒），默认2秒 use_cache: 是否使用缓存，默认True Returns: API响应的JSON数据 Raises: requests.exceptions.RequestException: 所有重试都失败时抛出异常 """ # 检查缓存 if use_cache and self.cache_dir: cache_key = self._get_cache_key(keyword, content_type, sort_type, publish_time) cache_path = self._get_cache_path(cache_key) if os.path.exists(cache_path): try: with open(cache_path, 'r', encoding='utf-8') as f: cached_result = json.load(f) logger.info(f" ✓ 使用缓存: {keyword}") return cached_result except Exception as e: logger.warning(f" 读取缓存失败: {e}，将重新搜索") # 缓存未命中或未启用，执行实际搜索 payload = { "keyword": keyword, "content_type": '不限', # 使用映射后的参数 "sort_type": sort_type, "publish_time": publish_time, "cursor": cursor } last_exception = None # 重试循环：最多尝试 max_retries 次 for attempt in range(1, max_retries + 1): try: if attempt > 1: print(f" 重试第 {attempt - 1}/{max_retries - 1} 次: {keyword}") response = requests.post( self.api_url, json=payload, timeout=timeout, headers={"Content-Type": "application/json"} ) response.raise_for_status() api_response = response.json() # 解析API返回的result字段（是JSON字符串） if not api_response.get("success"): raise Exception(f"API返回失败: {api_response}") result_str = api_response.get("result", "{}") result = json.loads(result_str) # 预处理返回数据：提取 image_list 中的 URL 字符串 self._preprocess_response(result) if attempt > 1: print(f" ✓ 重试成功") # 保存到缓存 if use_cache and self.cache_dir: try: cache_key = self._get_cache_key(keyword, content_type, sort_type, publish_time) cache_path = self._get_cache_path(cache_key) with open(cache_path, 'w', encoding='utf-8') as f: json.dump(result, f, ensure_ascii=False, indent=2) logger.info(f" ✓ 已缓存: {keyword}") except Exception as e: logger.warning(f" 保存缓存失败: {e}") return result except requests.exceptions.RequestException as e: last_exception = e if attempt < max_retries: # 还有重试机会，等待后继续 print(f" ✗ 请求失败 (第{attempt}次尝试): {e}") print(f" 等待 {retry_delay} 秒后重试...") time.sleep(retry_delay) else: # 已达最大重试次数，抛出异常 print(f" ✗ 请求失败 (已达最大重试次数 {max_retries}): {e}") # 所有重试都失败，抛出最后一次的异常 raise last_exception def _preprocess_response(self, result: Dict[str, Any]) -> None: """ 预处理搜索结果，将 image_list 中的字典格式转换为 URL 字符串列表 Args: result: API返回的原始结果字典（会直接修改） """ # 获取帖子列表 notes = result.get("data", {}).get("data", []) for note in notes: note_card = note.get("note_card", {}) image_list_raw = note_card.get("image_list", []) # 提取 URL 字符串 image_list = [] for img in image_list_raw: if isinstance(img, dict) and "image_url" in img: image_list.append(img["image_url"]) elif isinstance(img, str): # 如果已经是字符串，直接使用 image_list.append(img) # 更新为预处理后的列表 note_card["image_list"] = image_list def save_result(self, keyword: str, result: Dict[str, Any], page: int = 1) -> str: """ 保存结果到文件 目录结构: results/xiaohongshu_search/关键词/时间戳_page{页码}.json Args: keyword: 搜索关键词 result: API返回的结果 page: 页码 Returns: 保存的文件路径 """ # 创建目录结构: results/xiaohongshu_search/关键词/ result_dir = os.path.join(self.results_base_dir, "xiaohongshu_search", keyword) os.makedirs(result_dir, exist_ok=True) # 文件名使用时间戳和页码 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{timestamp}_page{page}.json" filepath = os.path.join(result_dir, filename) # 保存结果 with open(filepath, 'w', encoding='utf-8') as f: json.dump(result, f, ensure_ascii=False, indent=2) return filepath def main(): """示例使用""" # 解析命令行参数 parser = argparse.ArgumentParser(description='小红书笔记搜索工具') parser.add_argument( '--results-dir', type=str, default='data/search', help='结果输出目录 (默认: data/search)' ) parser.add_argument( '--keyword', type=str, required=True, help='搜索关键词 (必填)' ) parser.add_argument( '--content-type', type=str, default='不限', choices=['不限', '视频', '图文'], help='内容类型 (默认: 不限)' ) parser.add_argument( '--sort-type', type=str, default='综合', choices=['综合', '最新', '最多点赞', '最多评论'], help='排序方式 (默认: 综合)' ) parser.add_argument( '--publish-time', type=str, default='不限', choices=['不限', '一天内', '一周内', '半年内'], help='发布时间筛选 (默认: 不限)' ) parser.add_argument( '--cursor', type=str, default='', help='翻页游标 (默认为空，即第一页)' ) parser.add_argument( '--page', type=int, default=1, help='页码标识，用于保存文件名 (默认: 1)' ) args = parser.parse_args() # 创建API客户端实例 client = XiaohongshuSearch(results_dir=args.results_dir) # 执行搜索并保存 try: result = client.search( args.keyword, args.content_type, args.sort_type, args.publish_time, args.cursor ) filepath = client.save_result(args.keyword, result, args.page) print(f"Output: {filepath}") except Exception as e: print(f"Error: {e}", file=__import__('sys').stderr) if __name__ == "__main__": main()