"""
业务服务层
整合所有业务模块，提供统一的服务接口
"""
from typing import List, Optional, Dict, Any
from playwright.async_api import Page
from loguru import logger
from browser.browser import BrowserManager
from xiaohongshu.login import LoginAction
from xiaohongshu.feeds import FeedsListAction
from xiaohongshu.search import SearchAction
from xiaohongshu.feed_detail import FeedDetailAction
from xiaohongshu.comment_feed import CommentFeedAction
from xiaohongshu.like_favorite import LikeAction, FavoriteAction
from xiaohongshu.user_profile import UserProfileAction
from xiaohongshu.publish import PublishAction
from xiaohongshu.publish_video import PublishVideoAction
from xiaohongshu.types import (
Feed,
FeedDetailResponse,
UserProfileResponse,
)
from cookies.cookies import create_cookie_manager
class XiaohongshuService:
"""小红书服务"""
def __init__(self, browser_manager: BrowserManager):
self.browser_manager = browser_manager
self.cookie_manager = create_cookie_manager()
async def _get_page(self) -> Page:
"""获取或创建页面"""
return await self.browser_manager.new_page()
async def _save_cookies(self) -> None:
"""保存当前 cookies"""
try:
await self.browser_manager.save_cookies()
except Exception as e:
logger.warning(f"Failed to save cookies: {e}")
# ========== 登录相关 ==========
async def check_login_status(self) -> bool:
"""检查主站登录状态"""
page = await self._get_page()
try:
action = LoginAction(page)
return await action.check_login_status()
finally:
await page.close()
await self._save_cookies()
async def check_creator_login_status(self) -> bool:
"""检查创作中心登录状态"""
page = await self._get_page()
try:
# 访问创作中心
await page.goto("https://creator.xiaohongshu.com", timeout=60000)
await page.wait_for_load_state("load", timeout=60000)
import asyncio
await asyncio.sleep(2)
# 检查是否在登录页
current_url = page.url
is_logged_in = 'login' not in current_url.lower()
logger.info(f"Creator login status: {is_logged_in}, url: {current_url}")
return is_logged_in
finally:
await page.close()
await self._save_cookies()
async def init_creator_login(self) -> Dict[str, Any]:
"""初始化创作中心登录 - 打开登录页面供用户登录"""
page = await self._get_page()
try:
import asyncio
logger.info("Opening creator center login page...")
# 访问创作中心
await page.goto("https://creator.xiaohongshu.com", timeout=60000)
await page.wait_for_load_state("load", timeout=60000)
await asyncio.sleep(2)
current_url = page.url
if 'login' in current_url.lower():
# 需要登录 - 保持页面打开60秒供用户登录
logger.info("Creator center requires login, waiting for user to login...")
# 等待最多60秒，检查是否登录成功
max_wait = 60
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < max_wait:
current_url = page.url
if 'login' not in current_url.lower():
logger.info("Login detected, verifying...")
await asyncio.sleep(2)
# 验证发布页面
await page.goto("https://creator.xiaohongshu.com/publish/publish", timeout=60000)
await page.wait_for_load_state("load", timeout=60000)
await asyncio.sleep(2)
current_url = page.url
if 'login' not in current_url.lower():
return {
"success": True,
"logged_in": True,
"message": "创作中心登录成功！",
"url": current_url
}
break
await asyncio.sleep(2)
# 超时或仍未登录
return {
"success": False,
"logged_in": False,
"message": "等待登录超时，请在浏览器窗口中完成登录后重试",
"url": current_url,
"hint": "服务运行在非无头模式，你应该能看到浏览器窗口"
}
else:
# 已经登录
logger.info("Creator center already logged in")
# 验证发布页面
await page.goto("https://creator.xiaohongshu.com/publish/publish", timeout=60000)
await page.wait_for_load_state("load", timeout=60000)
await asyncio.sleep(2)
current_url = page.url
publish_accessible = 'login' not in current_url.lower()
return {
"success": True,
"logged_in": publish_accessible,
"message": "创作中心已登录" if publish_accessible else "发布页面需要登录",
"url": current_url
}
finally:
await page.close()
await self._save_cookies()
async def get_login_qrcode(self) -> Dict[str, Any]:
"""获取登录二维码"""
page = await self._get_page()
try:
action = LoginAction(page)
qr_url, is_logged_in = await action.fetch_qrcode_image()
if is_logged_in:
return {"logged_in": True, "qr_url": None}
return {"logged_in": False, "qr_url": qr_url}
finally:
await page.close()
await self._save_cookies()
async def delete_cookies(self) -> None:
"""删除 cookies"""
await self.cookie_manager.delete_cookies()
# ========== Feed 相关 ==========
async def list_feeds(self) -> List[Feed]:
"""获取 Feed 列表"""
page = await self._get_page()
try:
action = FeedsListAction(page)
return await action.get_feeds_list()
finally:
await page.close()
await self._save_cookies()
async def search_feeds(
self,
keyword: str,
filters: Optional[Dict[str, Any]] = None
) -> List[Feed]:
"""搜索内容"""
page = await self._get_page()
try:
action = SearchAction(page)
return await action.search(keyword, filters)
finally:
await page.close()
await self._save_cookies()
async def get_feed_detail(
self,
feed_id: str,
xsec_token: str,
load_all_comments: bool = False
) -> FeedDetailResponse:
"""获取 Feed 详情"""
page = await self._get_page()
try:
action = FeedDetailAction(page)
return await action.get_feed_detail(feed_id, xsec_token, load_all_comments)
finally:
await page.close()
await self._save_cookies()
# ========== 评论相关 ==========
async def post_comment_to_feed(
self,
feed_id: str,
xsec_token: str,
content: str
) -> bool:
"""发表评论"""
page = await self._get_page()
try:
action = CommentFeedAction(page)
return await action.post_comment(feed_id, xsec_token, content)
finally:
await page.close()
await self._save_cookies()
async def reply_comment_to_feed(
self,
feed_id: str,
xsec_token: str,
comment_id: str,
content: str
) -> bool:
"""回复评论"""
page = await self._get_page()
try:
action = CommentFeedAction(page)
return await action.reply_to_comment(feed_id, xsec_token, comment_id, content)
finally:
await page.close()
await self._save_cookies()
# ========== 点赞和收藏 ==========
async def like_feed(self, feed_id: str, xsec_token: str) -> bool:
"""点赞"""
page = await self._get_page()
try:
action = LikeAction(page)
return await action.like(feed_id, xsec_token)
finally:
await page.close()
await self._save_cookies()
async def unlike_feed(self, feed_id: str, xsec_token: str) -> bool:
"""取消点赞"""
page = await self._get_page()
try:
action = LikeAction(page)
return await action.unlike(feed_id, xsec_token)
finally:
await page.close()
await self._save_cookies()
async def favorite_feed(self, feed_id: str, xsec_token: str) -> bool:
"""收藏"""
page = await self._get_page()
try:
action = FavoriteAction(page)
return await action.favorite(feed_id, xsec_token)
finally:
await page.close()
await self._save_cookies()
async def unfavorite_feed(self, feed_id: str, xsec_token: str) -> bool:
"""取消收藏"""
page = await self._get_page()
try:
action = FavoriteAction(page)
return await action.unfavorite(feed_id, xsec_token)
finally:
await page.close()
await self._save_cookies()
# ========== 用户主页 ==========
async def user_profile(
self,
user_id: str,
xsec_token: str
) -> UserProfileResponse:
"""获取用户主页"""
page = await self._get_page()
try:
action = UserProfileAction(page)
return await action.user_profile(user_id, xsec_token)
finally:
await page.close()
await self._save_cookies()
# ========== 发布内容 ==========
async def publish_content(
self,
title: str,
content: str,
images: List[str],
tags: Optional[List[str]] = None
) -> bool:
"""发布图文内容"""
page = await self._get_page()
try:
action = PublishAction(page)
return await action.publish(title, content, images, tags)
finally:
await page.close()
await self._save_cookies()
async def publish_video(
self,
title: str,
content: str,
video_path: str,
cover_path: Optional[str] = None,
tags: Optional[List[str]] = None
) -> bool:
"""发布视频内容"""
page = await self._get_page()
try:
action = PublishVideoAction(page)
return await action.publish_video(title, content, video_path, cover_path, tags)
finally:
await page.close()
await self._save_cookies()
