"""搜索功能"""
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 []
# 解析为 Feed 对象
feeds = [Feed.model_validate(feed) for feed in search_data]
# 过滤掉没有 note_card 的 Feed（可能是广告或其他类型）
feeds = [feed for feed in feeds if feed.note_card is not None]
logger.info(f"Found {len(feeds)} results")
return feeds
except Exception as e:
logger.error(f"Search failed: {e}")
raise
