#!/bin/bash
# Ensure we are in the project root
cd "$(dirname "$0")/.."
# 创建所有业务模块的骨架文件
# Feeds 模块
cat > xiaohongshu/feeds.py << 'PYEOF'
"""Feed 列表功能"""
from typing import List
from playwright.async_api import Page
from loguru import logger
from xiaohongshu.types import Feed
from errors.errors import NoFeedsError
class FeedsListAction:
"""Feed 列表操作"""
def __init__(self, page: Page):
self.page = page
async def get_feeds_list(self) -> List[Feed]:
"""获取 Feed 列表"""
logger.info("Getting feeds list...")
try:
await self.page.goto("https://www.xiaohongshu.com/explore")
await self.page.wait_for_load_state("networkidle")
# 执行 JavaScript 获取 feeds 数据
feeds_data = await self.page.evaluate("""
() => {
const state = window.__INITIAL_STATE__;
if (state && state.feed && state.feed.feeds && state.feed.feeds._value) {
return state.feed.feeds._value;
}
return null;
}
""")
if not feeds_data:
raise NoFeedsError()
# 解析为 Feed 对象
feeds = [Feed.model_validate(feed) for feed in feeds_data]
logger.info(f"Got {len(feeds)} feeds")
return feeds
except Exception as e:
logger.error(f"Failed to get feeds list: {e}")
raise
PYEOF
# Search 模块
cat > xiaohongshu/search.py << 'PYEOF'
"""搜索功能"""
from typing import List, Optional, Dict, Any
from playwright.async_api import Page
from loguru import logger
from xiaohongshu.types import Feed
class SearchAction:
"""搜索操作"""
def __init__(self, page: Page):
self.page = page
async def search(
self,
keyword: str,
filters: Optional[Dict[str, Any]] = None
) -> List[Feed]:
"""搜索内容"""
logger.info(f"Searching for: {keyword}")
try:
# 构造搜索 URL
search_url = f"https://www.xiaohongshu.com/search_result?keyword={keyword}"
await self.page.goto(search_url)
await self.page.wait_for_load_state("networkidle")
# 获取搜索结果
search_data = await self.page.evaluate("""
() => {
const state = window.__INITIAL_STATE__;
if (state && state.search && state.search.feeds && state.search.feeds._value) {
return state.search.feeds._value;
}
return null;
}
""")
if not search_data:
logger.warning("No search results found")
return []
feeds = [Feed.model_validate(feed) for feed in search_data]
logger.info(f"Found {len(feeds)} results")
return feeds
except Exception as e:
logger.error(f"Search failed: {e}")
raise
PYEOF
# Feed Detail 模块
cat > xiaohongshu/feed_detail.py << 'PYEOF'
"""Feed 详情功能"""
from playwright.async_api import Page
from loguru import logger
from xiaohongshu.types import FeedDetailResponse
from xiaohongshu.navigate import make_feed_detail_url
from errors.errors import NoFeedDetailError
class FeedDetailAction:
"""Feed 详情操作"""
def __init__(self, page: Page):
self.page = page
async def get_feed_detail(
self,
feed_id: str,
xsec_token: str,
load_all_comments: bool = False
) -> FeedDetailResponse:
"""获取 Feed 详情"""
logger.info(f"Getting feed detail: {feed_id}")
try:
# 导航到详情页
url = make_feed_detail_url(feed_id, xsec_token)
await self.page.goto(url)
await self.page.wait_for_load_state("networkidle")
# 获取详情数据
detail_data = await self.page.evaluate("""
() => {
const state = window.__INITIAL_STATE__;
if (state && state.note) {
return {
note: state.note.noteDetailMap[Object.keys(state.note.noteDetailMap)[0]].note,
comments: state.note.noteDetailMap[Object.keys(state.note.noteDetailMap)[0]].comments || {list: [], cursor: "", hasMore: false}
};
}
return null;
}
""")
if not detail_data:
raise NoFeedDetailError()
detail = FeedDetailResponse.model_validate(detail_data)
logger.info(f"Got feed detail with {len(detail.comments.list)} comments")
return detail
except Exception as e:
logger.error(f"Failed to get feed detail: {e}")
raise
PYEOF
# Comment 模块
cat > xiaohongshu/comment_feed.py << 'PYEOF'
"""评论功能"""
from playwright.async_api import Page
from loguru import logger
class CommentFeedAction:
"""评论操作"""
def __init__(self, page: Page):
self.page = page
async def post_comment(
self,
feed_id: str,
xsec_token: str,
content: str
) -> bool:
"""发表评论"""
logger.info(f"Posting comment to feed: {feed_id}")
try:
from xiaohongshu.navigate import make_feed_detail_url
# 导航到详情页
url = make_feed_detail_url(feed_id, xsec_token)
await self.page.goto(url)
await self.page.wait_for_load_state("networkidle")
# 找到评论输入框
comment_input = await self.page.wait_for_selector("textarea.comment-input")
await comment_input.fill(content)
# 点击发送按钮
send_button = await self.page.wait_for_selector("button.send-btn")
await send_button.click()
logger.info("Comment posted successfully")
return True
except Exception as e:
logger.error(f"Failed to post comment: {e}")
return False
async def reply_to_comment(
self,
feed_id: str,
xsec_token: str,
comment_id: str,
content: str
) -> bool:
"""回复评论"""
logger.info(f"Replying to comment: {comment_id}")
try:
# TODO: 实现回复评论逻辑
logger.warning("Reply to comment not fully implemented yet")
return False
except Exception as e:
logger.error(f"Failed to reply to comment: {e}")
return False
PYEOF
# Like and Favorite 模块
cat > xiaohongshu/like_favorite.py << 'PYEOF'
"""点赞和收藏功能"""
from playwright.async_api import Page
from loguru import logger
class LikeAction:
"""点赞操作"""
def __init__(self, page: Page):
self.page = page
async def like(self, feed_id: str, xsec_token: str) -> bool:
"""点赞"""
logger.info(f"Liking feed: {feed_id}")
try:
from xiaohongshu.navigate import make_feed_detail_url
url = make_feed_detail_url(feed_id, xsec_token)
await self.page.goto(url)
await self.page.wait_for_load_state("networkidle")
# 找到点赞按钮并点击
like_button = await self.page.wait_for_selector(".like-btn")
await like_button.click()
logger.info("Liked successfully")
return True
except Exception as e:
logger.error(f"Failed to like: {e}")
return False
async def unlike(self, feed_id: str, xsec_token: str) -> bool:
"""取消点赞"""
logger.info(f"Unliking feed: {feed_id}")
# TODO: 实现取消点赞逻辑
return False
class FavoriteAction:
"""收藏操作"""
def __init__(self, page: Page):
self.page = page
async def favorite(self, feed_id: str, xsec_token: str) -> bool:
"""收藏"""
logger.info(f"Favoriting feed: {feed_id}")
try:
from xiaohongshu.navigate import make_feed_detail_url
url = make_feed_detail_url(feed_id, xsec_token)
await self.page.goto(url)
await self.page.wait_for_load_state("networkidle")
# 找到收藏按钮并点击
favorite_button = await self.page.wait_for_selector(".favorite-btn")
await favorite_button.click()
logger.info("Favorited successfully")
return True
except Exception as e:
logger.error(f"Failed to favorite: {e}")
return False
async def unfavorite(self, feed_id: str, xsec_token: str) -> bool:
"""取消收藏"""
logger.info(f"Unfavoriting feed: {feed_id}")
# TODO: 实现取消收藏逻辑
return False
PYEOF
# User Profile 模块
cat > xiaohongshu/user_profile.py << 'PYEOF'
"""用户主页功能"""
from playwright.async_api import Page
from loguru import logger
from xiaohongshu.types import UserProfileResponse
class UserProfileAction:
"""用户主页操作"""
def __init__(self, page: Page):
self.page = page
async def user_profile(
self,
user_id: str,
xsec_token: str
) -> UserProfileResponse:
"""获取用户主页信息"""
logger.info(f"Getting user profile: {user_id}")
try:
# 导航到用户主页
url = f"https://www.xiaohongshu.com/user/profile/{user_id}?xsec_token={xsec_token}"
await self.page.goto(url)
await self.page.wait_for_load_state("networkidle")
# 获取用户数据
profile_data = await self.page.evaluate("""
() => {
const state = window.__INITIAL_STATE__;
if (state && state.user) {
return {
userBasicInfo: state.user.userPageData._rawValue.basicInfo,
interactions: state.user.userPageData._rawValue.interactions,
feeds: state.user.feeds._value || []
};
}
return null;
}
""")
if not profile_data:
raise Exception("Failed to get user profile data")
profile = UserProfileResponse.model_validate(profile_data)
logger.info(f"Got user profile with {len(profile.feeds)} feeds")
return profile
except Exception as e:
logger.error(f"Failed to get user profile: {e}")
raise
PYEOF
# Publish 模块
cat > xiaohongshu/publish.py << 'PYEOF'
"""发布图文功能"""
from typing import List, Optional
from pathlib import Path
from playwright.async_api import Page
from loguru import logger
class PublishAction:
"""发布操作"""
def __init__(self, page: Page):
self.page = page
async def publish(
self,
title: str,
content: str,
images: List[str],
tags: Optional[List[str]] = None
) -> bool:
"""发布图文内容"""
logger.info(f"Publishing content: {title}")
try:
# 导航到发布页面
await self.page.goto("https://creator.xiaohongshu.com/publish/publish")
await self.page.wait_for_load_state("networkidle")
# 上传图片
for image_path in images:
logger.info(f"Uploading image: {image_path}")
# TODO: 实现图片上传逻辑
# 填写标题
title_input = await self.page.wait_for_selector("input.title-input")
await title_input.fill(title)
# 填写内容
content_input = await self.page.wait_for_selector("textarea.content-input")
await content_input.fill(content)
# 添加标签
if tags:
for tag in tags:
logger.info(f"Adding tag: {tag}")
# TODO: 实现添加标签逻辑
# 点击发布按钮
publish_button = await self.page.wait_for_selector("button.publish-btn")
await publish_button.click()
logger.info("Content published successfully")
return True
except Exception as e:
logger.error(f"Failed to publish content: {e}")
return False
PYEOF
# Publish Video 模块
cat > xiaohongshu/publish_video.py << 'PYEOF'
"""发布视频功能"""
from typing import List, Optional
from playwright.async_api import Page
from loguru import logger
class PublishVideoAction:
"""发布视频操作"""
def __init__(self, page: Page):
self.page = page
async def publish_video(
self,
title: str,
content: str,
video_path: str,
cover_path: Optional[str] = None,
tags: Optional[List[str]] = None
) -> bool:
"""发布视频内容"""
logger.info(f"Publishing video: {title}")
try:
# 导航到发布页面
await self.page.goto("https://creator.xiaohongshu.com/publish/publish?type=video")
await self.page.wait_for_load_state("networkidle")
# 上传视频
logger.info(f"Uploading video: {video_path}")
# TODO: 实现视频上传逻辑
# 上传封面（如果提供）
if cover_path:
logger.info(f"Uploading cover: {cover_path}")
# TODO: 实现封面上传逻辑
# 填写标题和内容
title_input = await self.page.wait_for_selector("input.title-input")
await title_input.fill(title)
content_input = await self.page.wait_for_selector("textarea.content-input")
await content_input.fill(content)
# 添加标签
if tags:
for tag in tags:
logger.info(f"Adding tag: {tag}")
# TODO: 实现添加标签逻辑
# 点击发布按钮
publish_button = await self.page.wait_for_selector("button.publish-btn")
await publish_button.click()
logger.info("Video published successfully")
return True
except Exception as e:
logger.error(f"Failed to publish video: {e}")
return False
PYEOF
echo "All module skeletons created successfully!"
