# 实现指南
本文档提供详细的实现指南,帮助你完善 xiaohongshu-python 项目中的功能。
## 目录
1. [项目概览](#项目概览)
2. [已完成功能](#已完成功能)
3. [待完善功能](#待完善功能)
4. [实现步骤](#实现步骤)
5. [调试技巧](#调试技巧)
6. [常见问题](#常见问题)
## 项目概览
### 架构设计
```
用户请求
↓
HTTP API / MCP 协议
↓
Service 层 (service.py)
↓
Action 层 (xiaohongshu/*.py)
↓
Browser 层 (browser/browser.py)
↓
Playwright → 小红书网站
```
### 核心组件
1. **Browser Manager** (`browser/browser.py`)
- 管理 Playwright 浏览器实例
- 处理 Cookie 加载和保存
- 提供页面创建接口
2. **Action 层** (`xiaohongshu/*.py`)
- 每个文件对应一个功能模块
- 封装具体的页面操作逻辑
- 使用 Playwright API 与页面交互
3. **Service 层** (`service.py`)
- 整合所有 Action
- 提供统一的业务接口
- 管理页面生命周期
4. **API 层** (`routes.py`, `mcp_server.py`)
- HTTP REST API
- MCP 协议工具
## 已完成功能
### ✅ 基础设施
- [x] 项目结构
- [x] 配置管理(环境变量、设置)
- [x] 浏览器管理(Playwright 集成)
- [x] Cookie 管理(加载、保存、删除)
- [x] 错误处理
- [x] 日志系统
- [x] Docker 部署
### ✅ 核心模块
- [x] 数据模型(Pydantic)
- [x] 页面导航
- [x] 登录功能(完整实现)
- [x] 业务模块骨架(8 个模块)
- [x] 服务层
- [x] MCP 服务器
- [x] HTTP 路由
- [x] 主程序入口
## 待完善功能
### 🔨 需要完善的模块
#### 1. Feed 列表 (`xiaohongshu/feeds.py`)
**当前状态**:骨架已实现,JavaScript 数据提取逻辑已完成
**需要完善**:
- 测试数据提取是否正确
- 处理页面加载失败的情况
- 添加重试机制
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/feeds.go`
#### 2. 搜索功能 (`xiaohongshu/search.py`)
**当前状态**:骨架已实现
**需要完善**:
- 实现搜索过滤器转换逻辑
- 处理搜索结果分页
- 添加搜索类型支持(笔记、用户等)
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/search.go`
**关键实现**:
```python
def convert_filters(filters: Optional[Dict[str, Any]]) -> str:
"""
将过滤器转换为小红书 API 格式
Go 代码参考:
- sort: 综合、最新、最热
- noteType: 全部、视频、图文
"""
if not filters:
return ""
# TODO: 实现过滤器转换逻辑
pass
```
#### 3. Feed 详情 (`xiaohongshu/feed_detail.py`)
**当前状态**:基础实现已完成
**需要完善**:
- 实现评论加载逻辑(滚动加载)
- 实现"更多回复"展开
- 添加评论数量限制
- 优化滚动算法
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/feed_detail.go`
**关键实现**:
```python
async def load_all_comments(self, page: Page, limit: int = 100):
"""
滚动加载所有评论
Go 代码参考:
1. 检测页面高度变化
2. 智能滚动(避免过快)
3. 检测停滞状态
4. 展开"更多回复"按钮
"""
previous_height = 0
stall_count = 0
while True:
# 滚动到底部
await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
await asyncio.sleep(1)
# 检查高度变化
current_height = await page.evaluate("document.body.scrollHeight")
if current_height == previous_height:
stall_count += 1
if stall_count >= 3:
break
else:
stall_count = 0
previous_height = current_height
# 检查评论数量
comments = await page.query_selector_all(".comment-item")
if len(comments) >= limit:
break
```
#### 4. 评论功能 (`xiaohongshu/comment_feed.py`)
**当前状态**:基础骨架
**需要完善**:
- 找到正确的评论输入框选择器
- 实现评论提交逻辑
- 实现回复评论逻辑
- 添加评论成功验证
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/comment_feed.go`
**调试步骤**:
1. 在浏览器中打开小红书详情页
2. 打开开发者工具
3. 找到评论输入框的 CSS 选择器
4. 找到发送按钮的选择器
5. 更新代码中的选择器
#### 5. 点赞和收藏 (`xiaohongshu/like_favorite.py`)
**当前状态**:基础骨架
**需要完善**:
- 找到点赞按钮选择器
- 找到收藏按钮选择器
- 实现点赞/取消点赞逻辑
- 实现收藏/取消收藏逻辑
- 添加状态检测(已点赞/未点赞)
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/like_favorite.go`
#### 6. 用户主页 (`xiaohongshu/user_profile.py`)
**当前状态**:基础实现
**需要完善**:
- 验证数据提取逻辑
- 处理私密账号
- 添加 Feed 分页加载
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/user_profile.go`
#### 7. 发布图文 (`xiaohongshu/publish.py`)
**当前状态**:骨架
**需要完善**:
- 实现图片上传逻辑
- 实现标签添加逻辑
- 实现发布确认
- 添加发布状态检测
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/publish.go`
**关键实现**:
```python
async def upload_images(self, page: Page, images: List[str]):
"""
上传图片
步骤:
1. 找到文件上传输入框
2. 使用 set_input_files 上传
3. 等待上传完成
"""
# 找到上传按钮
upload_input = await page.wait_for_selector("input[type='file']")
# 上传所有图片
await upload_input.set_input_files(images)
# 等待上传完成
await page.wait_for_selector(".upload-success", timeout=30000)
```
#### 8. 发布视频 (`xiaohongshu/publish_video.py`)
**当前状态**:骨架
**需要完善**:
- 实现视频上传逻辑
- 实现封面上传逻辑
- 处理视频处理进度
- 添加发布状态检测
**参考 Go 代码**:`xiaohongshu-mcp/xiaohongshu/publish_video.go`
## 实现步骤
### 步骤 1:设置开发环境
```bash
cd xiaohongshu-python
# 创建虚拟环境
python -m venv venv
source venv/bin/activate
# 安装依赖
pip install -r requirements.txt
# 安装 Playwright
playwright install chromium
```
### 步骤 2:测试基础功能
```bash
# 启动服务(非无头模式,方便调试)
export HEADLESS=false
python main.py
```
访问 `http://localhost:18060/docs` 查看 API 文档。
### 步骤 3:调试页面选择器
创建一个测试脚本 `test_selectors.py`:
```python
import asyncio
from playwright.async_api import async_playwright
async def test_selectors():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False)
page = await browser.new_page()
# 导航到小红书
await page.goto("https://www.xiaohongshu.com/explore")
# 等待用户手动操作
input("按回车继续...")
# 测试选择器
element = await page.query_selector(".your-selector")
if element:
print("找到元素!")
else:
print("未找到元素")
await browser.close()
asyncio.run(test_selectors())
```
### 步骤 4:逐个完善模块
按以下顺序完善模块:
1. **Feed 列表** - 最简单,先测试数据提取
2. **搜索功能** - 类似 Feed 列表
3. **Feed 详情** - 需要处理评论加载
4. **点赞收藏** - 需要找到正确的按钮
5. **评论功能** - 需要处理输入和提交
6. **用户主页** - 类似 Feed 列表
7. **发布功能** - 最复杂,需要处理文件上传
### 步骤 5:编写测试
为每个模块编写测试:
```python
# tests/test_feeds.py
import pytest
from browser.browser import create_browser
from xiaohongshu.feeds import FeedsListAction
@pytest.mark.asyncio
async def test_get_feeds_list():
browser = await create_browser(headless=False)
page = await browser.new_page()
action = FeedsListAction(page)
feeds = await action.get_feeds_list()
assert len(feeds) > 0
assert feeds[0].note_card.display_title
await page.close()
await browser.close()
```
## 调试技巧
### 1. 使用非无头模式
```python
# 在 .env 中设置
HEADLESS=false
```
这样可以看到浏览器的实际操作。
### 2. 添加调试日志
```python
from loguru import logger
logger.debug(f"Current URL: {page.url}")
logger.debug(f"Page title: {await page.title()}")
```
### 3. 截图调试
```python
await page.screenshot(path="debug.png")
```
### 4. 打印页面 HTML
```python
html = await page.content()
logger.debug(html)
```
### 5. 使用 Playwright Inspector
```bash
PWDEBUG=1 python main.py
```
### 6. 检查元素是否存在
```python
element = await page.query_selector(".selector")
if element:
logger.info("Element found")
text = await element.text_content()
logger.info(f"Text: {text}")
else:
logger.warning("Element not found")
```
## 常见问题
### Q1: 找不到元素
**原因**:
- 页面未完全加载
- CSS 选择器错误
- 元素在 iframe 中
- 需要登录
**解决方案**:
```python
# 等待元素出现
await page.wait_for_selector(".selector", timeout=10000)
# 等待网络空闲
await page.wait_for_load_state("networkidle")
# 检查 iframe
frames = page.frames
for frame in frames:
element = await frame.query_selector(".selector")
if element:
break
```
### Q2: 数据提取失败
**原因**:
- JavaScript 未执行完成
- 数据结构变化
- 需要滚动加载
**解决方案**:
```python
# 等待 JavaScript 执行
await page.wait_for_function("window.__INITIAL_STATE__ !== undefined")
# 打印数据结构
data = await page.evaluate("() => window.__INITIAL_STATE__")
logger.debug(f"Data structure: {data}")
```
### Q3: 上传文件失败
**原因**:
- 文件路径错误
- 文件类型不支持
- 文件大小超限
**解决方案**:
```python
from pathlib import Path
# 使用绝对路径
file_path = Path(image_path).absolute()
await input_element.set_input_files(str(file_path))
# 检查文件是否存在
if not file_path.exists():
raise FileNotFoundError(f"File not found: {file_path}")
```
### Q4: Cookie 未保存
**原因**:
- 浏览器未正常关闭
- 权限问题
- 路径错误
**解决方案**:
```python
# 确保浏览器正常关闭
try:
# ... 操作
finally:
await browser_manager.close()
# 检查 Cookie 文件
import os
cookie_path = settings.cookies_path
if os.path.exists(cookie_path):
logger.info(f"Cookie file exists: {cookie_path}")
else:
logger.warning(f"Cookie file not found: {cookie_path}")
```
## 下一步
1. **完善核心功能**:按优先级完善各个模块
2. **添加测试**:为每个模块编写单元测试和集成测试
3. **优化性能**:实现页面池、请求缓存等
4. **完善文档**:添加更多使用示例和 API 文档
5. **错误处理**:添加更完善的错误处理和重试机制
## 参考资源
- **Playwright 文档**:https://playwright.dev/python/
- **FastAPI 文档**:https://fastapi.tiangolo.com/
- **Pydantic 文档**:https://docs.pydantic.dev/
- **原 Go 版本**:`xiaohongshu-mcp/` 目录
## 联系与支持
如有问题,请:
1. 查看本文档
2. 查看 Go 版本的实现
3. 使用 Playwright Inspector 调试
4. 提交 Issue
祝开发顺利!🚀