Agent/examples/cloud_xhs.py

import os
import time
import random
from dotenv import load_dotenv
from browser_use import BrowserUse
from fake_useragent import UserAgent

# 加载环境变量（建议将敏感信息存在.env文件中）
load_dotenv()
API_KEY = os.getenv("BROWSER_USE_API_KEY")  # 你的BrowserUse API密钥
PROFILE_ID = os.getenv("XHS_PROFILE_ID")    # 你的小红书专属Profile ID

# 小红书关键Cookie（替换为你从浏览器导出的完整Cookie）
XHS_COOKIES = [
    {"name": "web_session", "value": "你的web_session值"},
    {"name": "a1", "value": "你的a1值"},
    {"name": "webId", "value": "你的webId值"},
    {"name": "gid", "value": "你的gid值"},
    {"name": "xhsTrackerId", "value": "你的xhsTrackerId值"},
    {"name": "timestamp", "value": str(int(time.time()))}
]

# 真实浏览器UA池（匹配小红书常用设备）
ua = UserAgent()
XHS_UA = ua.random if ua.random else "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"

def init_xhs_browser():
    """初始化小红书专用浏览器（防指纹+IP+Cookie配置）"""
    try:
        browser = BrowserUse(
            api_key=API_KEY,
            # 核心反指纹配置
            stealth=True,                # 开启Stealth模式，清除自动化特征
            browser_profile_id=PROFILE_ID, # 绑定专属Profile，持久化Cookie/存储
            # 真实浏览器指纹配置
            user_agent=XHS_UA,
            screen_resolution="1920,1080", # 匹配真实屏幕分辨率
            timezone="Asia/Shanghai",      # 时区与IP归属地一致
            language="zh-CN,zh;q=0.9",     # 中文语言环境
            # 禁用自动化标识（关键）
            extra_args=[
                "--disable-blink-features=AutomationControlled",
                "--disable-dev-shm-usage",
                "--no-sandbox",
                "--disable-extensions",    # 禁用扩展，减少特征暴露
                "--disable-images=false"   # 加载图片，模拟真实访问
            ],
            # 代理配置（替换为你的专属静态IP）
            proxy={
                "type": "http",           # 推荐http/socks5
                "host": "你的代理IP",
                "port": 你的代理端口,
                "username": "代理账号（如有）",
                "password": "代理密码（如有）"
            }
        )
        print("✅ 浏览器初始化成功")
        return browser
    except Exception as e:
        print(f"❌ 浏览器初始化失败: {e}")
        return None

def inject_xhs_cookies(browser, url="https://www.xiaohongshu.com"):
    """注入小红书完整Cookie并验证登录状态"""
    try:
        # 先访问小红书首页（建立会话）
        browser.get(url)
        time.sleep(random.uniform(2, 4))  # 随机延迟，模拟人类加载
        
        # 注入Cookie（逐个添加，确保完整）
        for cookie in XHS_COOKIES:
            browser.add_cookie(cookie)
        print("✅ Cookie注入完成")
        
        # 刷新页面，验证登录状态
        browser.refresh()
        time.sleep(random.uniform(3, 5))
        
        # 检查是否登录成功（通过页面元素判断）
        page_source = browser.page_source
        if "未登录" not in page_source and "我的主页" in page_source:
            print("✅ 小红书登录验证成功")
            return True
        else:
            print("❌ Cookie无效或登录状态未维持")
            return False
    except Exception as e:
        print(f"❌ Cookie注入失败: {e}")
        return False

def simulate_human_behavior(browser):
    """模拟人类行为（滚动、延迟、点击），规避行为风控"""
    try:
        # 模拟滚动浏览（随机步长）
        scroll_steps = random.randint(3, 8)
        for _ in range(scroll_steps):
            scroll_height = random.randint(300, 800)
            browser.execute_script(f"window.scrollBy(0, {scroll_height});")
            time.sleep(random.uniform(1.5, 3.5))  # 随机滚动间隔
        
        # 模拟随机停留
        time.sleep(random.uniform(5, 10))
        print("✅ 人类行为模拟完成")
    except Exception as e:
        print(f"❌ 行为模拟失败: {e}")

def get_xhs_data(browser, target_url):
    """获取小红书目标页面数据（笔记/用户信息）"""
    try:
        # 访问目标页面（非直接请求，模拟点击跳转）
        browser.get(target_url)
        time.sleep(random.uniform(4, 6))  # 页面加载延迟
        
        # 模拟人类行为后再提取数据
        simulate_human_behavior(browser)
        
        # 提取页面核心数据（可根据需求修改）
        page_title = browser.title
        page_html = browser.page_source
        
        # 示例：提取笔记标题（小红书笔记页专属）
        note_title = browser.execute_script("""
            return document.querySelector('h1[class*="note-title"]')?.innerText || '未找到标题';
        """)
        
        print(f"\n📊 数据提取结果：")
        print(f"页面标题: {page_title}")
        print(f"笔记标题: {note_title}")
        
        # 可扩展：提取点赞、评论、内容等信息
        return {
            "title": note_title,
            "page_html": page_html,
            "status": "success"
        }
    except Exception as e:
        print(f"❌ 数据提取失败: {e}")
        return {"status": "failed", "error": str(e)}

def main():
    """主流程：初始化→注入Cookie→获取数据→清理会话"""
    # 1. 初始化浏览器
    browser = init_xhs_browser()
    if not browser:
        return
    
    try:
        # 2. 注入Cookie并验证登录
        login_success = inject_xhs_cookies(browser)
        if not login_success:
            return
        
        # 3. 访问目标小红书页面（替换为你要爬取的URL）
        target_url = "https://www.xiaohongshu.com/explore/65a1b2c3d4e5f67890abcdef"
        data = get_xhs_data(browser, target_url)
        
        if data["status"] == "success":
            print("\n🎉 小红书数据获取成功！")
        else:
            print("\n❌ 小红书数据获取失败，请检查Cookie/IP/指纹配置")
    
    finally:
        # 4. 安全关闭会话（避免资源浪费）
        time.sleep(random.uniform(2, 3))
        browser.quit()
        print("\n🔚 浏览器会话已关闭")

if __name__ == "__main__":
    main()