"""浏览器内人类行为参数（对应 scripts/xhs/human.py）。""" from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path from typing import Any @dataclass class MsRange: min_ms: int max_ms: int def to_tuple(self) -> tuple[int, int]: return (self.min_ms, self.max_ms) def validate(self, name: str) -> list[str]: errors: list[str] = [] if self.min_ms < 0 or self.max_ms < 0: errors.append(f"{name} 不能为负数") if self.max_ms < self.min_ms: errors.append(f"{name} 上限不能小于下限") return errors @dataclass class CountRange: min_count: int max_count: int def validate(self, name: str) -> list[str]: errors: list[str] = [] if self.min_count < 1: errors.append(f"{name} 下限至少为 1") if self.max_count < self.min_count: errors.append(f"{name} 上限不能小于下限") return errors def to_dict(self) -> dict[str, int]: return {"min": self.min_count, "max": self.max_count} @dataclass class SecRange: min_sec: int max_sec: int def validate(self, name: str) -> list[str]: errors: list[str] = [] if self.min_sec < 15: errors.append(f"{name} 下限至少为 15 秒") if self.max_sec < self.min_sec: errors.append(f"{name} 上限不能小于下限") if self.max_sec > 7200: errors.append(f"{name} 上限不能超过 7200 秒（2 小时）") return errors def to_dict(self) -> dict[str, int]: return {"min": self.min_sec, "max": self.max_sec} def sample_target_sec(self) -> int: import random lo = max(15, self.min_sec) hi = max(lo, self.max_sec) return random.randint(lo, hi) @dataclass class ScrollSpeedProfile: interval_base_ms: int interval_jitter_ms: int scroll_ratio: float @dataclass class BrowseFlowPlan: """列表采集拟人流程（不含防封号/打开页，由运行时按目标秒数另算）。""" stable_sec_lo: float = 2.0 stable_sec_hi: float = 3.0 scroll_lo: int = 8 scroll_hi: int = 15 click_lo: int = 1 click_hi: int = 3 sprint_sec_lo: float = 4.0 sprint_sec_hi: float = 6.0 parse_sec_lo: float = 1.5 parse_sec_hi: float = 2.5 detail_stay_sec_lo: float = 10.0 detail_stay_sec_hi: float = 20.0 detail_scroll_lo: int = 3 detail_scroll_hi: int = 6 def to_dict(self) -> dict[str, float | int]: return { "stable_sec_lo": self.stable_sec_lo, "stable_sec_hi": self.stable_sec_hi, "scroll_lo": self.scroll_lo, "scroll_hi": self.scroll_hi, "click_lo": self.click_lo, "click_hi": self.click_hi, "sprint_sec_lo": self.sprint_sec_lo, "sprint_sec_hi": self.sprint_sec_hi, "parse_sec_lo": self.parse_sec_lo, "parse_sec_hi": self.parse_sec_hi, "detail_stay_sec_lo": self.detail_stay_sec_lo, "detail_stay_sec_hi": self.detail_stay_sec_hi, "detail_scroll_lo": self.detail_scroll_lo, "detail_scroll_hi": self.detail_scroll_hi, } def sample_run_targets(self) -> dict[str, float | int]: """按保存区间随机本次执行的滑动/点开/等待目标（供 GUI 与 CLI 同步）。""" import random scroll_lo = max(0, self.scroll_lo) scroll_hi = max(scroll_lo, self.scroll_hi) click_lo = max(0, self.click_lo) click_hi = max(click_lo, self.click_hi) scroll_target = ( random.randint(scroll_lo, scroll_hi) if scroll_hi > 0 else 0 ) click_target = random.randint(click_lo, click_hi) if click_hi > 0 else 0 return { "scroll_target": scroll_target, "click_target": click_target, "stable_sec": round( random.uniform(self.stable_sec_lo, self.stable_sec_hi), 2 ), "sprint_sec": round( random.uniform(self.sprint_sec_lo, self.sprint_sec_hi), 2 ), "parse_sec": round( random.uniform(self.parse_sec_lo, self.parse_sec_hi), 2 ), "detail_stay_sec": round( random.uniform(self.detail_stay_sec_lo, self.detail_stay_sec_hi), 2 ), "detail_scroll_target": random.randint( max(1, self.detail_scroll_lo), max(self.detail_scroll_lo, self.detail_scroll_hi), ), } def _parse_browse_flow_plan(raw: Any, default: BrowseFlowPlan) -> BrowseFlowPlan: if not isinstance(raw, dict): return default return BrowseFlowPlan( stable_sec_lo=float(raw.get("stable_sec_lo", default.stable_sec_lo)), stable_sec_hi=float(raw.get("stable_sec_hi", default.stable_sec_hi)), scroll_lo=int(raw.get("scroll_lo", default.scroll_lo)), scroll_hi=int(raw.get("scroll_hi", default.scroll_hi)), click_lo=int(raw.get("click_lo", default.click_lo)), click_hi=int(raw.get("click_hi", default.click_hi)), sprint_sec_lo=float(raw.get("sprint_sec_lo", default.sprint_sec_lo)), sprint_sec_hi=float(raw.get("sprint_sec_hi", default.sprint_sec_hi)), parse_sec_lo=float(raw.get("parse_sec_lo", default.parse_sec_lo)), parse_sec_hi=float(raw.get("parse_sec_hi", default.parse_sec_hi)), detail_stay_sec_lo=float( raw.get("detail_stay_sec_lo", default.detail_stay_sec_lo) ), detail_stay_sec_hi=float( raw.get("detail_stay_sec_hi", default.detail_stay_sec_hi) ), detail_scroll_lo=int(raw.get("detail_scroll_lo", default.detail_scroll_lo)), detail_scroll_hi=int(raw.get("detail_scroll_hi", default.detail_scroll_hi)), ) @dataclass class HumanBehaviorConfig: """与 scripts/xhs/human.py 一一对应的可配置项。""" default_max_attempts: int = 500 stagnant_limit: int = 20 min_scroll_delta: int = 10 max_click_per_round: int = 3 stagnant_check_threshold: int = 2 large_scroll_trigger: int = 5 button_click_interval: int = 3 final_sprint_push_count: int = 15 human_delay: MsRange = field(default_factory=lambda: MsRange(300, 700)) reaction_time: MsRange = field(default_factory=lambda: MsRange(300, 800)) hover_time: MsRange = field(default_factory=lambda: MsRange(100, 300)) read_time: MsRange = field(default_factory=lambda: MsRange(500, 1200)) short_read: MsRange = field(default_factory=lambda: MsRange(600, 1200)) scroll_wait: MsRange = field(default_factory=lambda: MsRange(100, 200)) post_scroll: MsRange = field(default_factory=lambda: MsRange(300, 500)) navigation_delay: MsRange = field(default_factory=lambda: MsRange(1000, 2500)) scroll_slow: ScrollSpeedProfile = field( default_factory=lambda: ScrollSpeedProfile(1200, 300, 0.5) ) scroll_normal: ScrollSpeedProfile = field( default_factory=lambda: ScrollSpeedProfile(600, 200, 0.7) ) scroll_fast: ScrollSpeedProfile = field( default_factory=lambda: ScrollSpeedProfile(300, 100, 0.9) ) home_feed_collect_time: SecRange = field( default_factory=lambda: SecRange(90, 240) ) search_feed_collect_time: SecRange = field( default_factory=lambda: SecRange(60, 180) ) home_browse_plan: BrowseFlowPlan = field(default_factory=BrowseFlowPlan) search_browse_plan: BrowseFlowPlan = field(default_factory=BrowseFlowPlan) inaccessible_keywords: list[str] = field( default_factory=lambda: [ "当前笔记暂时无法浏览", "该内容因违规已被删除", "该笔记已被删除", "内容不存在", "笔记不存在", "已失效", "私密笔记", "仅作者可见", "因用户设置，你无法查看", "因违规无法查看", ] ) def validate(self) -> list[str]: errors: list[str] = [] for name, val, lo in [ ("default_max_attempts", self.default_max_attempts, 1), ("stagnant_limit", self.stagnant_limit, 1), ("min_scroll_delta", self.min_scroll_delta, 0), ("max_click_per_round", self.max_click_per_round, 1), ("stagnant_check_threshold", self.stagnant_check_threshold, 1), ("large_scroll_trigger", self.large_scroll_trigger, 1), ("button_click_interval", self.button_click_interval, 0), ("final_sprint_push_count", self.final_sprint_push_count, 1), ]: if val < lo: errors.append(f"{name} 不能小于 {lo}") for label, r in [ ("human_delay", self.human_delay), ("reaction_time", self.reaction_time), ("hover_time", self.hover_time), ("read_time", self.read_time), ("short_read", self.short_read), ("scroll_wait", self.scroll_wait), ("post_scroll", self.post_scroll), ("navigation_delay", self.navigation_delay), ]: errors.extend(r.validate(label)) for label, sp in [ ("scroll_slow", self.scroll_slow), ("scroll_normal", self.scroll_normal), ("scroll_fast", self.scroll_fast), ]: if sp.interval_base_ms < 0 or sp.interval_jitter_ms < 0: errors.append(f"{label} 间隔不能为负") if not (0 < sp.scroll_ratio <= 1.0): errors.append(f"{label} 滚动比例须在 (0, 1]") errors.extend(self.home_feed_collect_time.validate("首页推荐采集时间")) errors.extend(self.search_feed_collect_time.validate("搜索内容采集时间")) errors.extend(self._validate_browse_plan(self.home_browse_plan, "首页拟人流程")) errors.extend(self._validate_browse_plan(self.search_browse_plan, "搜索拟人流程")) if not self.inaccessible_keywords: errors.append("不可访问关键词列表不能为空") return errors @staticmethod def _validate_browse_plan(plan: BrowseFlowPlan, name: str) -> list[str]: errors: list[str] = [] if plan.stable_sec_hi < plan.stable_sec_lo or plan.stable_sec_lo < 0: errors.append(f"{name} · 页面稳定等待秒数无效") if plan.scroll_hi < plan.scroll_lo or plan.scroll_lo < 1: errors.append(f"{name} · 滑动次数无效") if plan.click_hi < plan.click_lo or plan.click_lo < 0: errors.append(f"{name} · 点开次数无效") if plan.sprint_sec_hi < plan.sprint_sec_lo or plan.sprint_sec_lo < 0: errors.append(f"{name} · 收尾滚动秒数无效") if plan.parse_sec_hi < plan.parse_sec_lo or plan.parse_sec_lo < 0: errors.append(f"{name} · 解析秒数无效") if ( plan.detail_stay_sec_hi < plan.detail_stay_sec_lo or plan.detail_stay_sec_lo < 10 ): errors.append(f"{name} · 笔记详情停留秒数无效（下限至少 10 秒）") if plan.detail_scroll_hi < plan.detail_scroll_lo or plan.detail_scroll_lo < 1: errors.append(f"{name} · 笔记详情滚动次数无效") return errors def _parse_ms_range(raw: Any, default: MsRange) -> MsRange: if not isinstance(raw, dict): return default return MsRange( min_ms=int(raw.get("min_ms", default.min_ms)), max_ms=int(raw.get("max_ms", default.max_ms)), ) def _parse_scroll_profile(raw: Any, default: ScrollSpeedProfile) -> ScrollSpeedProfile: if not isinstance(raw, dict): return default return ScrollSpeedProfile( interval_base_ms=int(raw.get("interval_base_ms", default.interval_base_ms)), interval_jitter_ms=int(raw.get("interval_jitter_ms", default.interval_jitter_ms)), scroll_ratio=float(raw.get("scroll_ratio", default.scroll_ratio)), ) def _parse_count_range(raw: Any, default: CountRange) -> CountRange: if not isinstance(raw, dict): return default lo = int(raw.get("min", raw.get("min_count", default.min_count))) hi = int(raw.get("max", raw.get("max_count", default.max_count))) return CountRange(min_count=lo, max_count=hi) def _parse_sec_range( raw: Any, default: SecRange, *, legacy_count_raw: Any = None, ) -> SecRange: if isinstance(raw, dict): if "min_sec" in raw or "max_sec" in raw: lo = int(raw.get("min_sec", default.min_sec)) hi = int(raw.get("max_sec", default.max_sec)) return SecRange(min_sec=lo, max_sec=hi) lo = int(raw.get("min", default.min_sec)) hi = int(raw.get("max", default.max_sec)) if hi >= lo >= 15: return SecRange(min_sec=lo, max_sec=hi) if isinstance(legacy_count_raw, dict): lo = int(legacy_count_raw.get("min", 10)) hi = int(legacy_count_raw.get("max", 30)) # 旧版「条数」里若填的是 60 以上，多为误填的秒数，按秒直接沿用 if lo >= 60 and hi >= lo: return SecRange(min_sec=max(15, lo), max_sec=hi) return SecRange(min_sec=max(15, lo * 6), max_sec=max(lo * 6, hi * 8)) return default def parse_human_behavior_config(raw: dict[str, Any] | None) -> HumanBehaviorConfig: base = HumanBehaviorConfig() if not raw: return base def _range(key: str, attr: str) -> MsRange: return _parse_ms_range(raw.get(key), getattr(base, attr)) keywords = raw.get("inaccessible_keywords") if isinstance(keywords, list): kw_list = [str(k).strip() for k in keywords if str(k).strip()] else: kw_list = base.inaccessible_keywords return HumanBehaviorConfig( default_max_attempts=int(raw.get("default_max_attempts", base.default_max_attempts)), stagnant_limit=int(raw.get("stagnant_limit", base.stagnant_limit)), min_scroll_delta=int(raw.get("min_scroll_delta", base.min_scroll_delta)), max_click_per_round=int(raw.get("max_click_per_round", base.max_click_per_round)), stagnant_check_threshold=int( raw.get("stagnant_check_threshold", base.stagnant_check_threshold) ), large_scroll_trigger=int(raw.get("large_scroll_trigger", base.large_scroll_trigger)), button_click_interval=int(raw.get("button_click_interval", base.button_click_interval)), final_sprint_push_count=int( raw.get("final_sprint_push_count", base.final_sprint_push_count) ), human_delay=_range("human_delay", "human_delay"), reaction_time=_range("reaction_time", "reaction_time"), hover_time=_range("hover_time", "hover_time"), read_time=_range("read_time", "read_time"), short_read=_range("short_read", "short_read"), scroll_wait=_range("scroll_wait", "scroll_wait"), post_scroll=_range("post_scroll", "post_scroll"), navigation_delay=_range("navigation_delay", "navigation_delay"), scroll_slow=_parse_scroll_profile(raw.get("scroll_slow"), base.scroll_slow), scroll_normal=_parse_scroll_profile(raw.get("scroll_normal"), base.scroll_normal), scroll_fast=_parse_scroll_profile(raw.get("scroll_fast"), base.scroll_fast), home_feed_collect_time=_parse_sec_range( raw.get("home_feed_collect_time"), base.home_feed_collect_time, legacy_count_raw=raw.get("home_feed_collect_count"), ), search_feed_collect_time=_parse_sec_range( raw.get("search_feed_collect_time"), base.search_feed_collect_time, legacy_count_raw=raw.get("search_feed_collect_count"), ), home_browse_plan=_parse_browse_flow_plan( raw.get("home_browse_plan"), base.home_browse_plan ), search_browse_plan=_parse_browse_flow_plan( raw.get("search_browse_plan"), base.search_browse_plan ), inaccessible_keywords=kw_list or base.inaccessible_keywords, ) def human_behavior_config_to_dict(cfg: HumanBehaviorConfig) -> dict[str, Any]: def _r(r: MsRange) -> dict[str, int]: return {"min_ms": r.min_ms, "max_ms": r.max_ms} def _sp(p: ScrollSpeedProfile) -> dict[str, Any]: return { "interval_base_ms": p.interval_base_ms, "interval_jitter_ms": p.interval_jitter_ms, "scroll_ratio": p.scroll_ratio, } return { "default_max_attempts": cfg.default_max_attempts, "stagnant_limit": cfg.stagnant_limit, "min_scroll_delta": cfg.min_scroll_delta, "max_click_per_round": cfg.max_click_per_round, "stagnant_check_threshold": cfg.stagnant_check_threshold, "large_scroll_trigger": cfg.large_scroll_trigger, "button_click_interval": cfg.button_click_interval, "final_sprint_push_count": cfg.final_sprint_push_count, "human_delay": _r(cfg.human_delay), "reaction_time": _r(cfg.reaction_time), "hover_time": _r(cfg.hover_time), "read_time": _r(cfg.read_time), "short_read": _r(cfg.short_read), "scroll_wait": _r(cfg.scroll_wait), "post_scroll": _r(cfg.post_scroll), "navigation_delay": _r(cfg.navigation_delay), "scroll_slow": _sp(cfg.scroll_slow), "scroll_normal": _sp(cfg.scroll_normal), "scroll_fast": _sp(cfg.scroll_fast), "home_feed_collect_time": cfg.home_feed_collect_time.to_dict(), "search_feed_collect_time": cfg.search_feed_collect_time.to_dict(), "home_browse_plan": cfg.home_browse_plan.to_dict(), "search_browse_plan": cfg.search_browse_plan.to_dict(), "inaccessible_keywords": list(cfg.inaccessible_keywords), } def human_behavior_json_path(state_dir: Path) -> Path: return state_dir / "human_behavior.json" def write_human_behavior_json(cfg: HumanBehaviorConfig, state_dir: Path) -> Path: path = human_behavior_json_path(state_dir) state_dir.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(human_behavior_config_to_dict(cfg), ensure_ascii=False, indent=2), encoding="utf-8", ) return path def apply_human_behavior_to_skills_module(cfg: HumanBehaviorConfig) -> None: """将配置注入已加载的 skills.xhs.human 模块（同进程调试用）。""" try: from xhs import human as h # type: ignore[import-not-found] except ImportError: return _inject_into_human_module(h, cfg) def _inject_into_human_module(h: Any, cfg: HumanBehaviorConfig) -> None: h.DEFAULT_MAX_ATTEMPTS = cfg.default_max_attempts h.STAGNANT_LIMIT = cfg.stagnant_limit h.MIN_SCROLL_DELTA = cfg.min_scroll_delta h.MAX_CLICK_PER_ROUND = cfg.max_click_per_round h.STAGNANT_CHECK_THRESHOLD = cfg.stagnant_check_threshold h.LARGE_SCROLL_TRIGGER = cfg.large_scroll_trigger h.BUTTON_CLICK_INTERVAL = cfg.button_click_interval h.FINAL_SPRINT_PUSH_COUNT = cfg.final_sprint_push_count h.HUMAN_DELAY = cfg.human_delay.to_tuple() h.REACTION_TIME = cfg.reaction_time.to_tuple() h.HOVER_TIME = cfg.hover_time.to_tuple() h.READ_TIME = cfg.read_time.to_tuple() h.SHORT_READ = cfg.short_read.to_tuple() h.SCROLL_WAIT = cfg.scroll_wait.to_tuple() h.POST_SCROLL = cfg.post_scroll.to_tuple() h.INACCESSIBLE_KEYWORDS = list(cfg.inaccessible_keywords) h._SCROLL_PROFILES = { "slow": ( cfg.scroll_slow.interval_base_ms, cfg.scroll_slow.interval_jitter_ms, cfg.scroll_slow.scroll_ratio, ), "normal": ( cfg.scroll_normal.interval_base_ms, cfg.scroll_normal.interval_jitter_ms, cfg.scroll_normal.scroll_ratio, ), "fast": ( cfg.scroll_fast.interval_base_ms, cfg.scroll_fast.interval_jitter_ms, cfg.scroll_fast.scroll_ratio, ), } h._NAVIGATION_DELAY = cfg.navigation_delay.to_tuple() h.HOME_FEED_COLLECT_TIME = cfg.home_feed_collect_time.to_dict() h.SEARCH_FEED_COLLECT_TIME = cfg.search_feed_collect_time.to_dict() h.HOME_BROWSE_PLAN = cfg.home_browse_plan.to_dict() h.SEARCH_BROWSE_PLAN = cfg.search_browse_plan.to_dict()