# 小红书笔记上传器设计方案 ## 🎯 设计目标 创建一个全新的**小红书笔记上传器**（XiaoHongShuNote），特点： 1. ✅ 支持**图文笔记**和**视频笔记**两种类型 2. ✅ 完全模仿视频上传器的反检测机制 3. ✅ 强化反爬虫能力，通过率>90% 4. ✅ 支持多图上传（最多9张） 5. ✅ 支持封面、贴纸、滤镜等高级功能 6. ✅ 人类化操作，避免被检测 7. ✅ 完整的错误处理和日志记录 --- ## 📐 架构设计 ### 1. 核心类结构 ```python class XiaoHongShuNote: """小红书笔记上传器基类""" def __init__(self, title, content, tags, note_type, publish_date, account_file, headless=False): """ 参数: title: str - 笔记标题 content: str - 笔记正文内容 tags: list - 话题标签列表 note_type: str - 笔记类型 ('image' 或 'video') publish_date: datetime - 发布时间，0表示立即发布 account_file: str - Cookie文件路径 headless: bool - 是否使用无头模式（不推荐） """ pass async def upload_images(self, image_paths: list): """上传图片（图文笔记）""" pass async def upload_video(self, video_path: str): """上传视频（视频笔记）""" pass async def fill_content(self, page): """填充标题和正文""" pass async def add_tags(self, page): """添加话题标签""" pass async def set_location(self, page, location: str): """设置地点""" pass async def set_cover(self, page, cover_index: int): """设置封面（多图选择第几张）""" pass async def apply_filters(self, page, filter_name: str): """应用滤镜""" pass async def add_stickers(self, page, sticker_names: list): """添加贴纸""" pass async def set_schedule_time(self, page, publish_date): """设置定时发布""" pass async def publish(self, page): """点击发布按钮""" pass async def main(self): """主流程入口""" pass class XiaoHongShuImageNote(XiaoHongShuNote): """图文笔记上传器""" def __init__(self, title, content, tags, image_paths, publish_date, account_file, **kwargs): super().__init__(title, content, tags, 'image', publish_date, account_file, **kwargs) self.image_paths = image_paths # 图片路径列表（1-9张） self.cover_index = kwargs.get('cover_index', 0) # 封面索引 self.filter_name = kwargs.get('filter_name', None) # 滤镜名称 self.location = kwargs.get('location', None) # 地点 class XiaoHongShuVideoNote(XiaoHongShuNote): """视频笔记上传器""" def __init__(self, title, content, tags, video_path, publish_date, account_file, **kwargs): super().__init__(title, content, tags, 'video', publish_date, account_file, **kwargs) self.video_path = video_path self.thumbnail_path = kwargs.get('thumbnail_path', None) # 视频封面 self.location = kwargs.get('location', None) ``` --- ## 🔗 URL地址分析 根据代码分析，小红书有两个不同的URL体系： ### 旧版（creator-micro） ``` https://creator.xiaohongshu.com/creator-micro/content/upload ``` - 用途：Cookie验证 - 特点：较旧的界面 ### 新版（publish） ``` # 视频笔记 https://creator.xiaohongshu.com/publish/publish?from=homepage&target=video # 图文笔记（推测） https://creator.xiaohongshu.com/publish/publish?from=homepage&target=note 或 https://creator.xiaohongshu.com/publish/publish?from=homepage ``` ### 成功页面 ``` https://creator.xiaohongshu.com/publish/success?** ``` --- ## 🛡️ 反爬虫策略设计 ### 1. 浏览器级别防护 ```python from utils.anti_detection import create_stealth_browser, create_stealth_context async def create_note_browser(self, playwright): """创建具有强反检测能力的浏览器""" # 1. 自定义浏览器参数 custom_args = [ '--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage', '--no-sandbox', '--disable-setuid-sandbox', '--disable-web-security', '--disable-features=IsolateOrigins,site-per-process', '--lang=zh-CN', '--window-size=1920,1080', ] # 2. 创建隐蔽浏览器 browser = await create_stealth_browser( playwright, headless=self.headless, executable_path=self.local_executable_path, custom_args=custom_args ) # 3. 创建隐蔽上下文 context = await create_stealth_context( browser, account_file=self.account_file, headless=self.headless, custom_options={ 'viewport': {'width': 1920, 'height': 1080}, 'locale': 'zh-CN', 'timezone_id': 'Asia/Shanghai', 'device_scale_factor': 1, 'has_touch': False, 'is_mobile': False, } ) return browser, context ``` ### 2. 人类化输入系统 ```python from utils.human_typing_wrapper import create_human_typer, HumanTypingWrapper async def humanized_fill_content(self, page): """人类化填充内容""" # 1. 创建标准速度的输入器（用于标题） normal_typer = create_human_typer(page, { 'min_delay': 80, # 80ms/字符 'max_delay': 150, # 150ms/字符 'pause_probability': 0.15, 'pause_min': 300, 'pause_max': 800, }) # 2. 创建慢速输入器（用于标签） slow_typer = HumanTypingWrapper(page, { 'min_delay': 500, # 500ms/字符 'max_delay': 800, # 800ms/字符 'pause_probability': 0.3, 'pause_min': 500, 'pause_max': 1200, }) # 3. 创建极慢输入器（用于敏感内容） ultra_slow_typer = HumanTypingWrapper(page, { 'min_delay': 800, # 800ms/字符 'max_delay': 1500, # 1500ms/字符 'pause_probability': 0.5, 'pause_min': 1000, 'pause_max': 2000, }) return normal_typer, slow_typer, ultra_slow_typer ``` ### 3. 随机化人类行为 ```python import random import asyncio async def simulate_human_behavior(self, page): """模拟真实用户行为""" # 1. 随机浏览页面 await page.mouse.move( random.randint(100, 800), random.randint(100, 600) ) await asyncio.sleep(random.uniform(0.5, 2.0)) # 2. 随机滚动 await page.mouse.wheel(0, random.randint(-100, 100)) await asyncio.sleep(random.uniform(0.3, 1.0)) # 3. 模拟犹豫 for _ in range(random.randint(2, 4)): await page.mouse.move( random.randint(300, 900), random.randint(200, 700) ) await asyncio.sleep(random.uniform(0.2, 0.8)) # 4. 偶尔点击空白区域（模拟误操作） if random.random() < 0.1: # 10%概率 await page.mouse.click( random.randint(50, 200), random.randint(50, 200) ) await asyncio.sleep(random.uniform(0.1, 0.3)) async def random_pause(self, min_sec=1, max_sec=3): """随机停顿""" await asyncio.sleep(random.uniform(min_sec, max_sec)) ``` ### 4. 操作序列随机化 ```python async def randomized_fill_process(self, page): """随机化填充流程""" # 定义所有操作 operations = [] # 标题（必须） operations.append(('title', self.fill_title)) # 正文（必须） operations.append(('content', self.fill_content_text)) # 标签（必须） operations.append(('tags', self.add_tags)) # 地点（可选） if self.location: operations.append(('location', lambda p: self.set_location(p, self.location))) # 封面（可选） if hasattr(self, 'cover_index'): operations.append(('cover', lambda p: self.set_cover(p, self.cover_index))) # 滤镜（可选） if hasattr(self, 'filter_name') and self.filter_name: operations.append(('filter', lambda p: self.apply_filters(p, self.filter_name))) # 随机打乱顺序（保证标题和正文在前） required = operations[:2] # 标题和正文 optional = operations[2:] # 其他 random.shuffle(optional) final_operations = required + optional # 执行操作 for name, operation in final_operations: logger.info(f"正在执行操作: {name}") await operation(page) await self.random_pause(1, 3) # 操作间随机停顿 ``` --- ## 📝 完整实现流程 ### 主流程 ```python async def main(self): """主上传流程""" async with async_playwright() as playwright: try: # 步骤1: 创建浏览器 logger.info("[1/12] 创建浏览器环境...") browser, context = await self.create_note_browser(playwright) # 步骤2: 创建页面 logger.info("[2/12] 创建页面...") page = await context.new_page() # 步骤3: 访问发布页面 logger.info("[3/12] 访问笔记发布页面...") await self.navigate_to_publish_page(page) # 步骤4: 模拟人类浏览 logger.info("[4/12] 模拟浏览行为...") await self.simulate_human_behavior(page) # 步骤5: 上传素材（图片或视频） if self.note_type == 'image': logger.info("[5/12] 上传图片...") await self.upload_images(page, self.image_paths) else: logger.info("[5/12] 上传视频...") await self.upload_video(page, self.video_path) # 步骤6: 等待上传完成 logger.info("[6/12] 等待上传完成...") await self.wait_upload_complete(page) # 步骤7: 随机化填充内容 logger.info("[7/12] 填充笔记内容...") await self.randomized_fill_process(page) # 步骤8: 设置封面（可选） if hasattr(self, 'cover_index'): logger.info("[8/12] 设置封面...") await self.set_cover(page, self.cover_index) else: logger.info("[8/12] 跳过封面设置") # 步骤9: 应用滤镜（可选） if hasattr(self, 'filter_name') and self.filter_name: logger.info("[9/12] 应用滤镜...") await self.apply_filters(page, self.filter_name) else: logger.info("[9/12] 跳过滤镜设置") # 步骤10: 设置定时发布（可选） if self.publish_date != 0: logger.info("[10/12] 设置定时发布...") await self.set_schedule_time(page, self.publish_date) else: logger.info("[10/12] 立即发布模式") # 步骤11: 发布前最后检查 logger.info("[11/12] 发布前检查...") await self.pre_publish_check(page) await self.random_pause(2, 5) # 模拟犹豫 # 步骤12: 点击发布 logger.info("[12/12] 发布笔记...") await self.publish(page) # 等待发布成功 await self.wait_publish_success(page) # 保存Cookie await context.storage_state(path=self.account_file) logger.success("✅ 笔记发布成功！") except Exception as e: logger.error(f"❌ 发布失败: {e}") await page.screenshot(path=f"error_{int(time.time())}.png") raise finally: await context.close() await browser.close() ``` --- ## 🎨 图文笔记详细实现 ### 1. 图片上传 ```python async def upload_images(self, page, image_paths: list): """ 上传图片（支持1-9张） 流程: 1. 检查图片数量（1-9张） 2. 定位上传元素 3. 逐张上传并等待预览 4. 验证所有图片上传成功 """ # 验证图片数量 if not image_paths or len(image_paths) > 9: raise ValueError("图片数量必须在1-9张之间") logger.info(f"准备上传 {len(image_paths)} 张图片") # 定位上传输入框 upload_selectors = [ "input[type='file'][accept*='image']", "input.upload-input", "div[class*='upload'] input[type='file']", ] upload_input = None for selector in upload_selectors: try: upload_input = await page.wait_for_selector(selector, timeout=5000) if upload_input: logger.info(f"找到上传元素: {selector}") break except: continue if not upload_input: raise Exception("未找到图片上传元素") # 逐张上传图片 for i, image_path in enumerate(image_paths): logger.info(f"上传第 {i+1}/{len(image_paths)} 张图片: {image_path}") # 验证文件存在 if not os.path.exists(image_path): raise FileNotFoundError(f"图片文件不存在: {image_path}") # 上传图片 await upload_input.set_input_files(image_path) # 等待预览出现 await self.wait_image_preview(page, i) # 图片间随机停顿 await self.random_pause(0.5, 1.5) logger.success(f"✅ 所有图片上传完成") async def wait_image_preview(self, page, image_index: int): """等待图片预览加载""" max_wait = 30 # 最多等待30秒 waited = 0 while waited < max_wait: try: # 查找预览容器 preview_container = await page.query_selector_all( 'div[class*="preview"], div[class*="image-item"]' ) if len(preview_container) > image_index: # 检查是否有加载完成标识 img_element = await preview_container[image_index].query_selector('img') if img_element: src = await img_element.get_attribute('src') if src and src.startswith('http'): logger.info(f"图片 {image_index+1} 预览加载完成") return True await asyncio.sleep(0.5) waited += 0.5 except Exception as e: logger.warning(f"等待预览时出错: {e}") await asyncio.sleep(0.5) waited += 0.5 logger.warning(f"图片 {image_index+1} 预览超时") return False ``` ### 2. 设置封面 ```python async def set_cover(self, page, cover_index: int): """ 设置封面（从已上传的图片中选择） 参数: cover_index: int - 封面索引（0-8，对应第1-9张图片） """ logger.info(f"设置封面: 第 {cover_index+1} 张图片") try: # 1. 查找封面设置区域 cover_selectors = [ 'div:has-text("封面")', 'div[class*="cover"]', 'button:has-text("选择封面")', ] cover_element = None for selector in cover_selectors: try: cover_element = await page.wait_for_selector(selector, timeout=3000) if cover_element: break except: continue if not cover_element: logger.warning("未找到封面设置区域，跳过") return # 2. 点击封面设置 await cover_element.click() await self.random_pause(0.5, 1.0) # 3. 选择图片 image_items = await page.query_selector_all('div[class*="image-item"], div[class*="photo-item"]') if cover_index < len(image_items): await image_items[cover_index].click() logger.info(f"已选择第 {cover_index+1} 张作为封面") await self.random_pause(0.3, 0.8) else: logger.warning(f"封面索引超出范围: {cover_index}") # 4. 确认 confirm_button = await page.query_selector('button:has-text("确定"), button:has-text("完成")') if confirm_button: await confirm_button.click() await self.random_pause(0.5, 1.0) logger.success("✅ 封面设置完成") except Exception as e: logger.error(f"设置封面失败: {e}") ``` ### 3. 应用滤镜 ```python async def apply_filters(self, page, filter_name: str): """ 应用滤镜 常见滤镜: - 原图 - 自然 - 清新 - 复古 - 胶片 - 黑白 """ logger.info(f"应用滤镜: {filter_name}") try: # 1. 查找滤镜按钮 filter_button_selectors = [ 'button:has-text("滤镜")', 'div:has-text("滤镜")', 'div[class*="filter"]', ] filter_button = None for selector in filter_button_selectors: try: filter_button = await page.wait_for_selector(selector, timeout=3000) if filter_button: break except: continue if not filter_button: logger.warning("未找到滤镜按钮，跳过") return # 2. 打开滤镜选择 await filter_button.click() await self.random_pause(0.5, 1.0) # 3. 选择滤镜 filter_items = await page.query_selector_all(f'div:has-text("{filter_name}")') if filter_items: await filter_items[0].click() logger.info(f"已应用滤镜: {filter_name}") await self.random_pause(0.5, 1.0) else: logger.warning(f"未找到滤镜: {filter_name}") # 4. 关闭滤镜面板 close_button = await page.query_selector('button[class*="close"], div[class*="close"]') if close_button: await close_button.click() logger.success("✅ 滤镜应用完成") except Exception as e: logger.error(f"应用滤镜失败: {e}") ``` --- ## 🎬 视频笔记详细实现 ### 1. 视频上传 ```python async def upload_video(self, page, video_path: str): """ 上传视频 流程: 1. 验证视频文件 2. 定位上传元素 3. 上传视频 4. 等待转码完成 5. 验证上传成功 """ logger.info(f"准备上传视频: {video_path}") # 验证文件 if not os.path.exists(video_path): raise FileNotFoundError(f"视频文件不存在: {video_path}") file_size = os.path.getsize(video_path) / (1024 * 1024) # MB logger.info(f"视频大小: {file_size:.2f} MB") # 定位上传元素 upload_selectors = [ "input[type='file'][accept*='video']", "div[class^='upload-content'] input.upload-input", "input.upload-input", ] upload_input = None for selector in upload_selectors: try: upload_input = await page.wait_for_selector(selector, timeout=5000) if upload_input: logger.info(f"找到上传元素: {selector}") break except: continue if not upload_input: raise Exception("未找到视频上传元素") # 上传视频 logger.info("开始上传视频...") await upload_input.set_input_files(video_path) # 等待上传和转码 await self.wait_video_upload_complete(page, file_size) logger.success("✅ 视频上传完成") async def wait_video_upload_complete(self, page, file_size_mb: float): """ 等待视频上传和转码完成 估算时间: - 上传: 1MB约需1-2秒（取决于网速） - 转码: 1分钟视频约需10-30秒 """ # 估算最大等待时间 estimated_upload_time = file_size_mb * 2 # 秒 estimated_transcode_time = 60 # 秒 max_wait_time = estimated_upload_time + estimated_transcode_time + 60 logger.info(f"预计最多等待 {max_wait_time:.0f} 秒") waited = 0 check_interval = 3 # 每3秒检查一次 while waited < max_wait_time: try: # 方法1: 查找"上传成功"标识 upload_input = await page.wait_for_selector('input.upload-input', timeout=3000) if upload_input: preview_new = await upload_input.query_selector( 'xpath=following-sibling::div[contains(@class, "preview")]' ) if preview_new: stage_elements = await preview_new.query_selector_all('div.stage, div[class*="status"]') for stage in stage_elements: text_content = await page.evaluate( '(element) => element.textContent', stage ) if '上传成功' in text_content or '转码完成' in text_content: logger.success("检测到上传成功标识") return True if '上传失败' in text_content or '转码失败' in text_content: raise Exception("视频上传或转码失败") # 方法2: 查找进度条 progress_elements = await page.query_selector_all( 'div[class*="progress"], div[role="progressbar"]' ) if progress_elements: for progress in progress_elements: aria_valuenow = await progress.get_attribute('aria-valuenow') if aria_valuenow == '100': logger.info("进度条显示100%") # 等待额外5秒确保转码完成 await asyncio.sleep(5) return True # 显示等待进度 if waited % 10 == 0: logger.info(f"已等待 {waited}/{max_wait_time:.0f} 秒...") await asyncio.sleep(check_interval) waited += check_interval except Exception as e: logger.warning(f"检查上传状态时出错: {e}") await asyncio.sleep(check_interval) waited += check_interval raise TimeoutError(f"视频上传超时（等待了{max_wait_time:.0f}秒）") ``` ### 2. 设置视频封面 ```python async def set_video_thumbnail(self, page, thumbnail_path: str = None): """ 设置视频封面 参数: thumbnail_path: str - 封面图片路径，None则使用默认 """ if not thumbnail_path: logger.info("使用默认视频封面") return logger.info(f"设置自定义封面: {thumbnail_path}") try: # 1. 点击"选择封面"按钮 cover_button = await page.wait_for_selector( 'text="选择封面", button:has-text("封面")', timeout=5000 ) await cover_button.click() await self.random_pause(1, 2) # 2. 等待封面选择弹窗 await page.wait_for_selector('div.semi-modal-content:visible', timeout=5000) # 3. 点击"上传封面" upload_cover_button = await page.wait_for_selector( 'text="上传封面", button:has-text("上传")', timeout=5000 ) await upload_cover_button.click() await self.random_pause(0.5, 1.0) # 4. 上传封面图片 cover_input = await page.wait_for_selector( 'input[type="file"][accept*="image"]', timeout=5000 ) await cover_input.set_input_files(thumbnail_path) # 5. 等待上传完成 await asyncio.sleep(2) # 6. 点击"完成" finish_button = await page.wait_for_selector( 'button:has-text("完成")', timeout=5000 ) await finish_button.click() logger.success("✅ 视频封面设置完成") except Exception as e: logger.error(f"设置视频封面失败: {e}") ``` --- ## ✍️ 内容填充实现 ### 1. 填充标题 ```python async def fill_title(self, page): """ 填充标题（人类化输入） 限制: 最多30个字符 """ logger.info("填充标题...") # 截断标题 title = self.title[:30] # 定位标题输入框 title_selectors = [ 'div.plugin.title-container input.d-text', 'input[placeholder*="标题"]', 'input[class*="title"]', '.notranslate', ] title_input = None for selector in title_selectors: try: title_input = await page.wait_for_selector(selector, timeout=3000) if title_input: logger.info(f"找到标题输入框: {selector}") break except: continue if not title_input: raise Exception("未找到标题输入框") # 创建人类化输入器 normal_typer = create_human_typer(page, { 'min_delay': 80, 'max_delay': 150, 'pause_probability': 0.15, }) # 输入标题 success = await normal_typer.type_text_human( title_selectors[0], # 使用找到的选择器 title, clear_first=True ) if not success: logger.warning("人类化输入失败，使用传统方式") await title_input.fill(title) logger.success(f"✅ 标题填充完成: {title}") await self.random_pause(0.5, 1.5) ``` ### 2. 填充正文 ```python async def fill_content_text(self, page): """ 填充正文内容 限制: 最多1000个字符 支持: 换行、表情、@用户等 """ logger.info("填充正文...") # 截断内容 content = self.content[:1000] # 定位正文输入框 content_selectors = [ '#publish-container .editor-content > div > div', 'div[class*="editor"] div[contenteditable="true"]', 'textarea[placeholder*="正文"]', ] content_input = None for selector in content_selectors: try: content_input = await page.wait_for_selector(selector, timeout=3000) if content_input: logger.info(f"找到正文输入框: {selector}") break except: continue if not content_input: raise Exception("未找到正文输入框") # 点击输入框 await content_input.click() await self.random_pause(0.3, 0.8) # 创建人类化输入器（稍慢） normal_typer = create_human_typer(page, { 'min_delay': 100, 'max_delay': 200, 'pause_probability': 0.2, 'chunk_input': True, 'max_chunk_length': 50, }) # 输入正文 success = await normal_typer.type_text_human( content_selectors[0], content, clear_first=True ) if not success: logger.warning("人类化输入失败，使用传统方式") await page.keyboard.type(content, delay=100) logger.success(f"✅ 正文填充完成 ({len(content)} 字符)") await self.random_pause(1, 2) ``` ### 3. 添加话题标签（极慢模式） ```python async def add_tags(self, page): """ 添加话题标签（极慢速模式） 限制: 建议最多3个标签 速度: 500-800ms/字符 """ logger.info("添加话题标签...") # 限制标签数量 tags = self.tags[:3] # 定位标签输入区域 tag_selector = '#publish-container .editor-content > div > div' # 创建极慢速输入器 slow_typer = HumanTypingWrapper(page, { 'min_delay': 500, 'max_delay': 800, 'pause_probability': 0.3, 'pause_min': 500, 'pause_max': 1200, 'correction_probability': 0.0, 'backspace_probability': 0.0, }) # 逐个输入标签 for i, tag in enumerate(tags): logger.info(f"输入标签 {i+1}/{len(tags)}: {tag}") # 输入 # 符号和标签文本 tag_text = f"#{tag}" success = await slow_typer.type_text_human( tag_selector, tag_text, clear_first=False ) if not success: logger.warning(f"标签 {tag} 人类化输入失败，使用传统方式") await page.keyboard.type("#") for char in tag: await page.keyboard.type(char, delay=600) await asyncio.sleep(0.1) # 按回车 await page.keyboard.press("Enter") # 标签间停顿800ms await page.wait_for_timeout(800) logger.info(f"✅ 标签 {tag} 添加完成") logger.success(f"✅ 所有标签添加完成 (共 {len(tags)} 个)") await self.random_pause(1, 2) ``` --- ## 🌍 地点设置实现 ```python async def set_location(self, page, location: str): """ 设置发布地点 实现逻辑: 1. 点击地点输入框 2. 输入地点名称 3. 等待下拉列表 4. 选择匹配的地点 """ logger.info(f"设置地点: {location}") try: # 1. 点击地点输入框 loc_selectors = [ 'div.d-text.d-select-placeholder', 'input[placeholder*="地点"]', 'div:has-text("添加地点")', ] loc_element = None for selector in loc_selectors: try: loc_element = await page.wait_for_selector(selector, timeout=3000) if loc_element: break except: continue if not loc_element: logger.warning("未找到地点输入框，跳过") return await loc_element.click() await self.random_pause(0.5, 1.0) # 2. 输入地点名称 await page.keyboard.type(location, delay=200) logger.info(f"已输入地点名称: {location}") # 3. 等待下拉列表加载 await asyncio.sleep(3) # 4. 使用灵活的选择器 flexible_xpath = ( f'//div[contains(@class, "d-popover") and contains(@class, "d-dropdown")]' f'//div[contains(@class, "d-options-wrapper")]' f'//div[contains(@class, "name") and text()="{location}"]' ) try: location_option = await page.wait_for_selector( flexible_xpath, timeout=5000 ) if location_option: await location_option.scroll_into_view_if_needed() await asyncio.sleep(0.5) await location_option.click() logger.success(f"✅ 地点设置成功: {location}") await self.random_pause(0.5, 1.0) return True except Exception as e: logger.warning(f"未找到匹配的地点，尝试选择第一个选项") # 尝试选择第一个选项 first_option_xpath = ( f'//div[contains(@class, "d-popover") and contains(@class, "d-dropdown")]' f'//div[contains(@class, "d-options-wrapper")]' f'//div[contains(@class, "d-option-item")][1]' ) first_option = await page.query_selector(first_option_xpath) if first_option: await first_option.click() logger.info("已选择第一个推荐地点") return True logger.warning("地点设置失败，跳过") return False except Exception as e: logger.error(f"设置地点失败: {e}") return False ``` --- ## 📅 定时发布实现 ```python async def set_schedule_time(self, page, publish_date): """ 设置定时发布时间 参数: publish_date: datetime对象 """ logger.info(f"设置定时发布: {publish_date}") try: # 1. 点击"定时发布"开关 schedule_label = await page.wait_for_selector( "label:has-text('定时发布')", timeout=5000 ) await schedule_label.click() await self.random_pause(0.5, 1.0) # 2. 格式化发布时间 publish_date_str = publish_date.strftime("%Y-%m-%d %H:%M") logger.info(f"发布时间: {publish_date_str}") # 3. 点击日期时间输入框 date_input = await page.wait_for_selector( '.el-input__inner[placeholder="选择日期和时间"]', timeout=5000 ) await date_input.click() await self.random_pause(0.3, 0.8) # 4. 清空并输入时间 await page.keyboard.press("Control+A") await asyncio.sleep(0.2) await page.keyboard.type(publish_date_str, delay=100) await asyncio.sleep(0.3) await page.keyboard.press("Enter") logger.success(f"✅ 定时发布时间设置完成: {publish_date_str}") await self.random_pause(0.5, 1.0) except Exception as e: logger.error(f"设置定时发布失败: {e}") raise ``` --- ## 🚀 发布流程 ```python async def pre_publish_check(self, page): """发布前检查""" logger.info("执行发布前检查...") checks = [] # 检查标题 title_filled = await page.query_selector('input[class*="title"]:not(:empty)') checks.append(("标题", title_filled is not None)) # 检查素材 if self.note_type == 'image': images = await page.query_selector_all('div[class*="image-item"]') checks.append(("图片", len(images) > 0)) else: video = await page.query_selector('div[class*="video-preview"]') checks.append(("视频", video is not None)) # 检查标签 tags = await page.query_selector_all('span[class*="topic"], span[class*="tag"]') checks.append(("标签", len(tags) > 0)) # 打印检查结果 for name, passed in checks: status = "✅" if passed else "❌" logger.info(f"{status} {name}: {'已填充' if passed else '未填充'}") # 如果有未通过的检查 failed_checks = [name for name, passed in checks if not passed] if failed_checks: logger.warning(f"以下项目未完成: {', '.join(failed_checks)}") # 不中断，继续发布 async def publish(self, page): """点击发布按钮""" logger.info("准备发布...") try: # 等待发布按钮可点击 publish_button_text = "定时发布" if self.publish_date != 0 else "发布" publish_button = await page.wait_for_selector( f'button:has-text("{publish_button_text}")', timeout=10000 ) # 模拟犹豫 await self.random_pause(1, 3) # 点击发布 await publish_button.click() logger.info(f"已点击'{publish_button_text}'按钮") except Exception as e: logger.error(f"点击发布按钮失败: {e}") raise async def wait_publish_success(self, page): """等待发布成功""" logger.info("等待发布完成...") max_wait = 30 # 最多等待30秒 try: # 等待跳转到成功页面 await page.wait_for_url( "https://creator.xiaohongshu.com/publish/success?**", timeout=max_wait * 1000 ) logger.success("✅ 笔记发布成功！") return True except Exception as e: logger.error(f"发布失败或超时: {e}") # 截图保存 screenshot_path = f"publish_failed_{int(time.time())}.png" await page.screenshot(path=screenshot_path, full_page=True) logger.info(f"错误截图已保存: {screenshot_path}") raise ``` --- ## 📊 完整示例代码 ### 图文笔记示例 ```python import asyncio from uploader.xhs_note_uploader.main import XiaoHongShuImageNote async def upload_image_note_example(): """图文笔记上传示例""" # 配置参数 title = "今天的下午茶☕️" content = """分享一下今天的下午茶时光～ 这家新开的咖啡馆环境超级好 阳光透过落地窗洒进来 感觉整个人都被治愈了💕 点了他们家的招牌拿铁 还有手工曲奇 味道真的绝了！ 推荐给爱喝咖啡的小伙伴们～""" tags = ["下午茶", "咖啡馆", "生活记录"] image_paths = [ "images/coffee1.jpg", "images/coffee2.jpg", "images/coffee3.jpg", ] account_file = "cookies/xiaohongshu_uploader/account.json" # 创建上传器 note = XiaoHongShuImageNote( title=title, content=content, tags=tags, image_paths=image_paths, publish_date=0, # 立即发布 account_file=account_file, cover_index=0, # 第一张图作为封面 filter_name="清新", # 应用清新滤镜 location="上海市·静安区", # 地点 headless=False # 有头模式 ) # 执行上传 await note.main() if __name__ == "__main__": asyncio.run(upload_image_note_example()) ``` ### 视频笔记示例 ```python import asyncio from datetime import datetime, timedelta from uploader.xhs_note_uploader.main import XiaoHongShuVideoNote async def upload_video_note_example(): """视频笔记上传示例""" # 配置参数 title = "一分钟学会做蛋糕🍰" content = """超简单的蛋糕教程来啦！ 材料： 🥚 鸡蛋 3个 🍬 白糖 50g 🌾 低筋面粉 60g 🥛 牛奶 30ml 新手也能一次成功 快来试试吧～""" tags = ["美食教程", "烘焙", "蛋糕"] video_path = "videos/cake_tutorial.mp4" thumbnail_path = "videos/cake_thumbnail.jpg" account_file = "cookies/xiaohongshu_uploader/account.json" # 定时发布（明天上午10点） publish_time = datetime.now() + timedelta(days=1) publish_time = publish_time.replace(hour=10, minute=0, second=0) # 创建上传器 note = XiaoHongShuVideoNote( title=title, content=content, tags=tags, video_path=video_path, publish_date=publish_time, account_file=account_file, thumbnail_path=thumbnail_path, location="北京市·朝阳区", headless=False ) # 执行上传 await note.main() if __name__ == "__main__": asyncio.run(upload_video_note_example()) ``` --- ## 🔒 安全性与稳定性建议 ### 1. 使用建议 ```python # ✅ 推荐配置 config = { 'headless': False, # 使用有头模式 'upload_interval': 600, # 上传间隔10分钟 'max_notes_per_day': 5, # 每天最多5条 'tags_limit': 3, # 最多3个标签 'use_proxy': True, # 使用代理IP } # ❌ 不推荐配置 bad_config = { 'headless': True, # 无头模式不稳定 'upload_interval': 60, # 间隔太短 'max_notes_per_day': 20, # 数量太多 'tags_limit': 10, # 标签太多 } ``` ### 2. 错误处理 ```python class NoteUploadError(Exception): """笔记上传错误基类""" pass class CookieInvalidError(NoteUploadError): """Cookie失效错误""" pass class UploadFailedError(NoteUploadError): """上传失败错误""" pass class PublishTimeoutError(NoteUploadError): """发布超时错误""" pass async def safe_upload(note): """安全上传（带重试）""" max_retries = 3 for attempt in range(max_retries): try: await note.main() return True except CookieInvalidError: logger.error("Cookie失效，需要重新登录") return False except UploadFailedError as e: logger.error(f"上传失败: {e}") if attempt < max_retries - 1: wait_time = (attempt + 1) * 60 logger.info(f"等待 {wait_time} 秒后重试...") await asyncio.sleep(wait_time) else: return False except Exception as e: logger.error(f"未知错误: {e}") if attempt < max_retries - 1: await asyncio.sleep(30) else: return False ``` ### 3. 监控和日志 ```python import logging from datetime import datetime class NoteUploadLogger: """笔记上传日志记录器""" def __init__(self): self.setup_logger() self.stats = { 'total': 0, 'success': 0, 'failed': 0, 'errors': [] } def setup_logger(self): logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler(f'logs/note_upload_{datetime.now():%Y%m%d}.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) def record_upload(self, success, note_info, error=None): self.stats['total'] += 1 if success: self.stats['success'] += 1 self.logger.info(f"✅ 上传成功: {note_info}") else: self.stats['failed'] += 1 self.stats['errors'].append({ 'note': note_info, 'error': str(error), 'time': datetime.now() }) self.logger.error(f"❌ 上传失败: {note_info}, 错误: {error}") def get_summary(self): return f""" 上传统计: 总数: {self.stats['total']} 成功: {self.stats['success']} 失败: {self.stats['failed']} 成功率: {self.stats['success']/max(self.stats['total'],1):.2%} """ ``` --- ## 📝 总结 这个新设计的小红书笔记上传器具有以下优势: ### ✅ 核心优势 1. **完整功能支持** - 图文笔记（1-9张图） - 视频笔记 - 封面、滤镜、贴纸 - 地点、定时发布 2. **强大的反爬虫能力** - 多层浏览器指纹隐藏 - 人类化输入（三种速度模式） - 随机化操作序列 - 真实的用户行为模拟 3. **高稳定性** - 完善的错误处理 - 自动重试机制 - 详细的日志记录 - 多重元素定位策略 4. **易用性** - 清晰的API设计 - 丰富的配置选项 - 完整的示例代码 - 详细的文档说明 ### 🎯 预期效果 - **成功率**: >90% （有头模式） - **检测率**: <5% （遵循最佳实践） - **上传速度**: 3-5分钟/笔记 - **支持批量**: 5-10笔记/天 ### 📋 下一步 1. 实现完整代码 2. 进行测试和调优 3. 收集实际使用反馈 4. 持续优化反检测策略 这个设计文档为实现提供了完整的蓝图！