zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
309 lines
9.3 KiB
Python
309 lines
9.3 KiB
Python
"""首页 / 搜索列表页的人类化浏览（滚动、停留、点开笔记弹窗养号）。
|
||
|
||
养号点开笔记：随机点击封面打开弹窗（note_detail_modal），不走路由。
|
||
按 feed_id 拉取详情：见 feed_detail.py（路由）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import sys
|
||
import time
|
||
from collections.abc import Callable
|
||
from typing import TYPE_CHECKING, Literal
|
||
|
||
from .browse_plan_session import BrowseRunTargets
|
||
from .collect_buffer import is_cancelled
|
||
from .note_detail_modal import BrowseScene, browse_note_modal_session
|
||
from .human import (
|
||
BUTTON_CLICK_INTERVAL,
|
||
DEFAULT_MAX_ATTEMPTS,
|
||
HUMAN_DELAY,
|
||
HOVER_TIME,
|
||
LARGE_SCROLL_TRIGGER,
|
||
MIN_SCROLL_DELTA,
|
||
POST_SCROLL,
|
||
STAGNANT_CHECK_THRESHOLD,
|
||
STAGNANT_LIMIT,
|
||
get_scroll_interval,
|
||
human_scroll_page,
|
||
navigation_delay,
|
||
sleep_random,
|
||
)
|
||
from .human import _sleep_interruptible as sleep_interruptible_sec
|
||
|
||
if TYPE_CHECKING:
|
||
from .cdp import Page
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
ProgressFn = Callable[[str], None] | None
|
||
BrowseMode = Literal["list", "search"]
|
||
|
||
def _click_label(mode: BrowseMode) -> str:
|
||
return "点开详情" if mode == "search" else "点开笔记"
|
||
|
||
|
||
def _emit_progress(on_progress: ProgressFn, msg: str) -> None:
|
||
logger.info(msg)
|
||
if on_progress:
|
||
on_progress(msg)
|
||
print(f"@@PROGRESS@@{msg}", file=sys.stderr, flush=True)
|
||
|
||
|
||
def _browse_scene(mode: BrowseMode) -> BrowseScene:
|
||
return "search" if mode == "search" else "home"
|
||
|
||
|
||
def _run_final_sprint(
|
||
page: Page,
|
||
scroll_speed: str,
|
||
sprint_sec: float,
|
||
on_progress: ProgressFn,
|
||
) -> None:
|
||
"""在配置的秒数内收尾滚动。"""
|
||
_emit_progress(on_progress, "模拟浏览：收尾冲刺…")
|
||
budget = max(0.5, float(sprint_sec))
|
||
deadline = time.monotonic() + budget
|
||
while time.monotonic() < deadline:
|
||
if is_cancelled():
|
||
return
|
||
human_scroll_page(page, scroll_speed, True, 1)
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
break
|
||
sleep_random(200, 400)
|
||
time.sleep(min(get_scroll_interval(scroll_speed), remaining))
|
||
sleep_random(*HUMAN_DELAY)
|
||
_emit_progress(on_progress, "模拟浏览：收尾冲刺完成")
|
||
|
||
|
||
def _open_one_note(
|
||
page: Page,
|
||
list_url: str,
|
||
scroll_speed: str,
|
||
*,
|
||
mode: BrowseMode,
|
||
run_targets: BrowseRunTargets | None,
|
||
on_progress: ProgressFn,
|
||
index: int,
|
||
total: int,
|
||
) -> bool:
|
||
"""养号：随机点击笔记封面打开弹窗，拟人后关闭（不走路由、不拉详情）。"""
|
||
label = _click_label(mode)
|
||
_emit_progress(on_progress, f"模拟浏览：{label} ({index}/{total})…")
|
||
if run_targets:
|
||
stay = run_targets.detail_stay_sec
|
||
detail_scrolls = run_targets.detail_scroll_target
|
||
else:
|
||
stay = 15.0
|
||
detail_scrolls = 4
|
||
ok = browse_note_modal_session(
|
||
page,
|
||
list_url,
|
||
scroll_speed,
|
||
scene=_browse_scene(mode),
|
||
stay_sec=stay,
|
||
scroll_count=detail_scrolls,
|
||
)
|
||
if ok:
|
||
navigation_delay()
|
||
else:
|
||
logger.debug("%s: 弹窗拟人未完成", label)
|
||
return ok
|
||
|
||
|
||
def _fulfill_click_quota(
|
||
page: Page,
|
||
list_url: str,
|
||
scroll_speed: str,
|
||
*,
|
||
mode: BrowseMode,
|
||
run_targets: BrowseRunTargets | None,
|
||
on_progress: ProgressFn,
|
||
clicks_done: int,
|
||
click_target: int,
|
||
) -> int:
|
||
"""滚动结束后补足未完成的点开次数（搜索页链接常较晚出现）。"""
|
||
label = _click_label(mode)
|
||
done = clicks_done
|
||
max_tries = max(click_target - done, 0) * 4 + 2
|
||
tries = 0
|
||
while done < click_target and tries < max_tries:
|
||
if is_cancelled():
|
||
break
|
||
tries += 1
|
||
if _open_one_note(
|
||
page,
|
||
list_url,
|
||
scroll_speed,
|
||
mode=mode,
|
||
run_targets=run_targets,
|
||
on_progress=on_progress,
|
||
index=done + 1,
|
||
total=click_target,
|
||
):
|
||
done += 1
|
||
if done >= click_target:
|
||
_emit_progress(
|
||
on_progress,
|
||
f"模拟浏览：{label}完成（{done}/{click_target}）",
|
||
)
|
||
else:
|
||
sleep_random(400, 800)
|
||
return done
|
||
|
||
|
||
def human_browse_feed_list(
|
||
page: Page,
|
||
list_url: str,
|
||
*,
|
||
scroll_speed: str = "normal",
|
||
click_notes: bool = True,
|
||
browse_mode: BrowseMode = "list",
|
||
on_progress: ProgressFn = None,
|
||
max_attempts: int | None = None,
|
||
stagnant_limit: int | None = None,
|
||
after_scroll: Callable[[], None] | None = None,
|
||
run_targets: BrowseRunTargets | None = None,
|
||
) -> None:
|
||
"""在 Feed 列表页模拟真人：按拟人计划滑动、点开、收尾。"""
|
||
scroll_target = run_targets.scroll_target if run_targets else 0
|
||
click_target = run_targets.click_target if run_targets else 0
|
||
if run_targets and run_targets.stable_sec > 0:
|
||
_emit_progress(
|
||
on_progress,
|
||
f"模拟浏览：等待页面稳定（约 {run_targets.stable_sec:.1f} 秒）…",
|
||
)
|
||
sleep_interruptible_sec(run_targets.stable_sec)
|
||
else:
|
||
_emit_progress(on_progress, "模拟浏览：等待页面稳定…")
|
||
navigation_delay()
|
||
sleep_random(300, 800)
|
||
_emit_progress(on_progress, "模拟浏览：页面稳定完成")
|
||
|
||
stagnant = 0
|
||
clicks_done = 0
|
||
scrolls_done = 0
|
||
large_mode = False
|
||
push_count = 0
|
||
last_scroll_top = page.get_scroll_top()
|
||
click_label = _click_label(browse_mode)
|
||
|
||
use_plan = run_targets is not None and scroll_target > 0
|
||
effective_click = click_target if use_plan else 0
|
||
if click_notes and not use_plan:
|
||
effective_click = 3
|
||
|
||
stop_stagnant = stagnant_limit if stagnant_limit is not None else STAGNANT_LIMIT
|
||
if use_plan:
|
||
cap = scroll_target + max(effective_click, 1) * 8 + 16
|
||
else:
|
||
cap = max_attempts if max_attempts is not None else DEFAULT_MAX_ATTEMPTS
|
||
max_attempts = min(cap, stop_stagnant + 25)
|
||
|
||
scroll_finished = False
|
||
for attempt in range(max_attempts):
|
||
if is_cancelled():
|
||
_emit_progress(on_progress, "模拟浏览：已停止")
|
||
return
|
||
|
||
if (
|
||
click_notes
|
||
and effective_click > 0
|
||
and clicks_done < effective_click
|
||
and scrolls_done < scroll_target
|
||
and attempt > 0
|
||
and attempt % max(1, BUTTON_CLICK_INTERVAL) == 0
|
||
):
|
||
if _open_one_note(
|
||
page,
|
||
list_url,
|
||
scroll_speed,
|
||
mode=browse_mode,
|
||
run_targets=run_targets,
|
||
on_progress=on_progress,
|
||
index=clicks_done + 1,
|
||
total=effective_click,
|
||
):
|
||
clicks_done += 1
|
||
stagnant = 0
|
||
push_count = 0
|
||
large_mode = False
|
||
last_scroll_top = page.get_scroll_top()
|
||
if clicks_done >= effective_click:
|
||
_emit_progress(
|
||
on_progress,
|
||
f"模拟浏览：{click_label}完成（{clicks_done}/{effective_click}）",
|
||
)
|
||
|
||
if stagnant >= STAGNANT_CHECK_THRESHOLD:
|
||
large_mode = True
|
||
if push_count >= LARGE_SCROLL_TRIGGER:
|
||
large_mode = True
|
||
|
||
if use_plan and scrolls_done >= scroll_target:
|
||
scroll_finished = True
|
||
break
|
||
|
||
scroll_label = (
|
||
f"模拟浏览：滚动 ({scrolls_done + 1}/{scroll_target})…"
|
||
if use_plan
|
||
else f"模拟浏览：滚动 ({attempt + 1})…"
|
||
)
|
||
_emit_progress(on_progress, scroll_label)
|
||
actual_delta, current_top = human_scroll_page(
|
||
page,
|
||
scroll_speed,
|
||
large_mode,
|
||
push_count if large_mode and push_count > 0 else 1,
|
||
)
|
||
scrolls_done += 1
|
||
sleep_random(*POST_SCROLL)
|
||
time.sleep(get_scroll_interval(scroll_speed))
|
||
|
||
if after_scroll:
|
||
after_scroll()
|
||
|
||
if actual_delta < MIN_SCROLL_DELTA or current_top == last_scroll_top:
|
||
stagnant += 1
|
||
else:
|
||
stagnant = 0
|
||
push_count = 0
|
||
large_mode = False
|
||
last_scroll_top = current_top
|
||
push_count += 1
|
||
|
||
if stagnant >= stop_stagnant:
|
||
_emit_progress(on_progress, "模拟浏览：已滑到底部")
|
||
scroll_finished = True
|
||
break
|
||
|
||
if attempt % 5 == 4:
|
||
sleep_random(*HOVER_TIME)
|
||
|
||
if use_plan and scrolls_done > 0:
|
||
_emit_progress(
|
||
on_progress,
|
||
f"模拟浏览：滑动完成（{scrolls_done}/{scroll_target}）",
|
||
)
|
||
elif not scroll_finished and scrolls_done > 0:
|
||
_emit_progress(
|
||
on_progress,
|
||
f"模拟浏览：滑动完成（{scrolls_done} 次）",
|
||
)
|
||
|
||
if click_notes and effective_click > 0 and clicks_done < effective_click:
|
||
clicks_done = _fulfill_click_quota(
|
||
page,
|
||
list_url,
|
||
scroll_speed,
|
||
mode=browse_mode,
|
||
run_targets=run_targets,
|
||
on_progress=on_progress,
|
||
clicks_done=clicks_done,
|
||
click_target=effective_click,
|
||
)
|
||
|
||
sprint_sec = run_targets.sprint_sec if run_targets else 5.0
|
||
_run_final_sprint(page, scroll_speed, sprint_sec, on_progress)
|