zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
"""搜索 Feeds，对应 Go xiaohongshu/search.go。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from .browse_feeds import human_browse_feed_list
|
||
from .browse_plan_session import BrowsePlanSession
|
||
from .cdp import Page
|
||
from .collect_buffer import is_cancelled, persist_collected, raise_if_cancelled
|
||
from .errors import BrowseCancelledError, NoFeedsError
|
||
from .extract_feeds_state import extract_feeds_list, merge_feed_dicts
|
||
from .human import _sleep_interruptible as sleep_interruptible_sec, sleep_random
|
||
from .search_filters import apply_search_filters, filter_option_to_labels, wait_search_results_ready
|
||
from .types import Feed, FilterOption
|
||
from .urls import make_search_url
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _feeds_from_collected(collected: dict[str, dict]) -> list[Feed]:
|
||
return [Feed.from_dict(f) for f in collected.values()]
|
||
|
||
|
||
def search_feeds(
|
||
page: Page,
|
||
keyword: str,
|
||
filter_option: FilterOption | None = None,
|
||
*,
|
||
scroll_speed: str = "normal",
|
||
on_progress=None,
|
||
) -> tuple[list[Feed], bool, dict]:
|
||
"""搜索并采集笔记：先等列表 → 应用筛选 → 按拟人计划滚动采集。"""
|
||
keyword = (keyword or "").strip()
|
||
if not keyword:
|
||
raise ValueError("搜索关键词不能为空")
|
||
|
||
session = BrowsePlanSession("search")
|
||
|
||
def progress(msg: str) -> None:
|
||
logger.info(msg)
|
||
if on_progress:
|
||
on_progress(msg)
|
||
|
||
search_url = make_search_url(keyword)
|
||
collected: dict[str, dict] = {}
|
||
meta = {
|
||
"browse_scroll_target": session.scroll_target,
|
||
"browse_click_target": session.click_target,
|
||
"browse_targets": session.targets.to_dict(),
|
||
"collect_elapsed_sec": 0.0,
|
||
"plan_completed": False,
|
||
}
|
||
|
||
def absorb(label: str) -> int:
|
||
items = extract_feeds_list(page)
|
||
added = merge_feed_dicts(collected, items)
|
||
if added:
|
||
logger.info("%s: 新增 %d 条，累计 %d 条", label, added, len(collected))
|
||
persist_collected(collected)
|
||
raise_if_cancelled()
|
||
return added
|
||
|
||
try:
|
||
progress(session.start_message())
|
||
progress(f"打开搜索页：{keyword}")
|
||
page.navigate(search_url)
|
||
page.wait_for_load()
|
||
page.wait_dom_stable()
|
||
|
||
wait_search_results_ready(page, timeout=22.0)
|
||
absorb("搜索页初次加载")
|
||
|
||
labels = filter_option_to_labels(filter_option)
|
||
if labels:
|
||
progress(f"应用筛选：{'、'.join(labels)}")
|
||
result = apply_search_filters(page, labels)
|
||
clicked = result.get("clicked") or []
|
||
missed = result.get("missed") or []
|
||
if clicked:
|
||
progress(f"已选筛选：{'、'.join(clicked)}")
|
||
if missed:
|
||
progress(f"部分筛选未生效：{'、'.join(missed)}")
|
||
if not result.get("ok"):
|
||
progress("筛选面板可能未打开，将使用当前列表继续采集")
|
||
if is_cancelled():
|
||
raise BrowseCancelledError()
|
||
wait_search_results_ready(page, timeout=18.0)
|
||
absorb("应用筛选后")
|
||
sleep_random(600, 1000)
|
||
|
||
list_url = page.evaluate("window.location.href") or search_url
|
||
progress("打开搜索页完成")
|
||
progress("滚动加载更多结果…")
|
||
|
||
def on_scroll_collect() -> None:
|
||
absorb("滚动中")
|
||
|
||
human_browse_feed_list(
|
||
page,
|
||
list_url,
|
||
scroll_speed=scroll_speed,
|
||
click_notes=session.click_target > 0,
|
||
browse_mode="search",
|
||
on_progress=on_progress,
|
||
after_scroll=on_scroll_collect,
|
||
run_targets=session.targets,
|
||
)
|
||
if not is_cancelled():
|
||
parse_sec = session.targets.parse_sec
|
||
if parse_sec > 0:
|
||
progress(
|
||
f"模拟浏览：提取并解析列表数据（约 {parse_sec:.1f} 秒）…"
|
||
)
|
||
sleep_interruptible_sec(parse_sec)
|
||
progress("模拟浏览：提取并解析列表数据…")
|
||
absorb("滚动结束后")
|
||
progress("模拟浏览：解析完成")
|
||
|
||
if not collected:
|
||
raise NoFeedsError()
|
||
|
||
meta["plan_completed"] = True
|
||
meta["collect_elapsed_sec"] = round(session.elapsed_sec(), 1)
|
||
progress(session.summary(len(collected), plan_completed=True))
|
||
return _feeds_from_collected(collected), False, meta
|
||
except BrowseCancelledError:
|
||
meta["collect_elapsed_sec"] = round(session.elapsed_sec(), 1)
|
||
if collected:
|
||
progress(
|
||
session.summary(
|
||
len(collected),
|
||
plan_completed=False,
|
||
user_cancelled=True,
|
||
)
|
||
)
|
||
return _feeds_from_collected(collected), True, meta
|
||
raise
|