Shuang_Dong/autoUpload

Fork 0

jinye_huang 0ef5186cbe 完成了新版本的小红薯笔记上传

2025-11-06 16:33:33 +08:00

43 KiB

Raw Blame History

小红书笔记上传器设计方案

🎯 设计目标

创建一个全新的小红书笔记上传器（XiaoHongShuNote），特点：

✅ 支持图文笔记和视频笔记两种类型
✅ 完全模仿视频上传器的反检测机制
✅ 强化反爬虫能力，通过率>90%
✅ 支持多图上传（最多9张）
✅ 支持封面、贴纸、滤镜等高级功能
✅ 人类化操作，避免被检测
✅ 完整的错误处理和日志记录

📐 架构设计

1. 核心类结构

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. 浏览器级别防护

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. 人类化输入系统

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. 随机化人类行为

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. 操作序列随机化

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)  # 操作间随机停顿

📝 完整实现流程

主流程

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. 图片上传

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. 设置封面

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. 应用滤镜

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. 视频上传

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. 设置视频封面

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. 填充标题

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. 填充正文

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. 添加话题标签（极慢模式）

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)

🌍 地点设置实现

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

📅 定时发布实现

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

🚀 发布流程

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

📊 完整示例代码

图文笔记示例

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())

视频笔记示例

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. 使用建议

# ✅ 推荐配置
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. 错误处理

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. 监控和日志

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-9张图）
- 视频笔记
- 封面、滤镜、贴纸
- 地点、定时发布
强大的反爬虫能力
- 多层浏览器指纹隐藏
- 人类化输入（三种速度模式）
- 随机化操作序列
- 真实的用户行为模拟
高稳定性
- 完善的错误处理
- 自动重试机制
- 详细的日志记录
- 多重元素定位策略
易用性
- 清晰的API设计
- 丰富的配置选项
- 完整的示例代码
- 详细的文档说明

🎯 预期效果

成功率: >90% （有头模式）
检测率: <5% （遵循最佳实践）
上传速度: 3-5分钟/笔记
支持批量: 5-10笔记/天

📋 下一步

实现完整代码
进行测试和调优
收集实际使用反馈
持续优化反检测策略

这个设计文档为实现提供了完整的蓝图！

43 KiB Raw Blame History Unescape Escape

小红书笔记上传器设计方案

🎯 设计目标

📐 架构设计

1. 核心类结构

🔗 URL地址分析

旧版（creator-micro）

新版（publish）

成功页面

🛡️ 反爬虫策略设计

1. 浏览器级别防护

2. 人类化输入系统

3. 随机化人类行为

4. 操作序列随机化

📝 完整实现流程

主流程

🎨 图文笔记详细实现

1. 图片上传

2. 设置封面

3. 应用滤镜

🎬 视频笔记详细实现

1. 视频上传

2. 设置视频封面

✍️ 内容填充实现

1. 填充标题

2. 填充正文

3. 添加话题标签（极慢模式）

🌍 地点设置实现

📅 定时发布实现

🚀 发布流程

📊 完整示例代码

图文笔记示例

视频笔记示例

🔒 安全性与稳定性建议

1. 使用建议

2. 错误处理

3. 监控和日志

📝 总结

✅ 核心优势

🎯 预期效果

📋 下一步

43 KiB

Raw Blame History