"""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", timeout=60000)
# 等待页面基本加载完成
await self.page.wait_for_load_state("load", timeout=60000)
# 等待一下让 JavaScript 执行
import asyncio
await asyncio.sleep(5) # 增加到 5 秒
# 执行 JavaScript 获取 feeds 数据
feeds_data = await self.page.evaluate("""
() => {
const state = window.__INITIAL_STATE__;
console.log('State exists:', !!state);
if (state) {
console.log('State keys:', Object.keys(state));
console.log('feed exists:', !!state.feed);
if (state.feed) {
console.log('feed keys:', Object.keys(state.feed));
console.log('feeds exists:', !!state.feed.feeds);
if (state.feed.feeds) {
console.log('feeds keys:', Object.keys(state.feed.feeds));
console.log('_value exists:', !!state.feed.feeds._value);
if (state.feed.feeds._value) {
console.log('_value length:', state.feed.feeds._value.length);
return state.feed.feeds._value;
}
}
}
}
return null;
}
""")
logger.info(f"Feeds data result: {feeds_data is not None}, count: {len(feeds_data) if feeds_data else 0}")
if not feeds_data:
raise NoFeedsError()
# 解析为 Feed 对象
feeds = [Feed.model_validate(feed) for feed in feeds_data]
# 过滤掉没有 note_card 的 Feed（可能是广告或其他类型）
feeds = [feed for feed in feeds if feed.note_card is not None]
logger.info(f"Got {len(feeds)} feeds")
return feeds
except Exception as e:
logger.error(f"Failed to get feeds list: {e}")
raise
