"""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)
logger.info(f"Navigating to: {url}")
# 增加超时时间到 60 秒
await self.page.goto(url, timeout=60000)
# 等待页面加载，使用更宽松的超时
try:
await self.page.wait_for_load_state("networkidle", timeout=60000)
except Exception as e:
logger.warning(f"Wait for networkidle timeout, trying to get data anyway: {e}")
# 即使 networkidle 超时，也尝试获取数据
await self.page.wait_for_load_state("domcontentloaded", timeout=10000)
# 等待一下让 JavaScript 执行
import asyncio
await asyncio.sleep(2)
# 获取详情数据
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
