"""人类行为模拟参数（延迟、滚动、悬停），对应 Go feed_detail.go 中的常量。""" from __future__ import annotations import json import os import random import time from pathlib import Path from typing import Any # ========== 配置常量（可被 human_behavior.json 覆盖） ========== DEFAULT_MAX_ATTEMPTS = 500 STAGNANT_LIMIT = 20 MIN_SCROLL_DELTA = 10 MAX_CLICK_PER_ROUND = 3 STAGNANT_CHECK_THRESHOLD = 2 LARGE_SCROLL_TRIGGER = 5 BUTTON_CLICK_INTERVAL = 3 FINAL_SPRINT_PUSH_COUNT = 15 # ========== 延迟范围（毫秒） ========== HUMAN_DELAY = (300, 700) REACTION_TIME = (300, 800) HOVER_TIME = (100, 300) READ_TIME = (500, 1200) SHORT_READ = (600, 1200) SCROLL_WAIT = (100, 200) POST_SCROLL = (300, 500) _NAVIGATION_DELAY = (1000, 2500) _SCROLL_PROFILES: dict[str, tuple[int, int, float]] = { "slow": (1200, 300, 0.5), "normal": (600, 200, 0.7), "fast": (300, 100, 0.9), } # 列表采集时间区间（秒，仅兼容旧配置，执行由 browse_plan 控制） HOME_FEED_COLLECT_TIME = {"min": 90, "max": 240} SEARCH_FEED_COLLECT_TIME = {"min": 60, "max": 180} _DEFAULT_BROWSE_PLAN: dict[str, float | int] = { "stable_sec_lo": 2.0, "stable_sec_hi": 3.0, "scroll_lo": 8, "scroll_hi": 15, "click_lo": 1, "click_hi": 3, "sprint_sec_lo": 4.0, "sprint_sec_hi": 6.0, "parse_sec_lo": 1.5, "parse_sec_hi": 2.5, "detail_stay_sec_lo": 10.0, "detail_stay_sec_hi": 20.0, "detail_scroll_lo": 3, "detail_scroll_hi": 6, } HOME_BROWSE_PLAN: dict[str, float | int] = dict(_DEFAULT_BROWSE_PLAN) SEARCH_BROWSE_PLAN: dict[str, float | int] = dict(_DEFAULT_BROWSE_PLAN) # 页面不可访问关键词 INACCESSIBLE_KEYWORDS = [ "当前笔记暂时无法浏览", "该内容因违规已被删除", "该笔记已被删除", "内容不存在", "笔记不存在", "已失效", "私密笔记", "仅作者可见", "因用户设置，你无法查看", "因违规无法查看", ] def _config_path() -> Path | None: env = os.environ.get("XHS_HUMAN_BEHAVIOR_FILE", "").strip() if env: p = Path(env) return p if p.is_file() else None root = Path(__file__).resolve().parents[2] p = root / ".xhs_bridge" / "human_behavior.json" return p if p.is_file() else None def _apply_tuple(key: str, raw: Any, target: str) -> None: global _NAVIGATION_DELAY if not isinstance(raw, dict): return lo = int(raw.get("min_ms", 0)) hi = int(raw.get("max_ms", 0)) if hi >= lo >= 0: if key == "navigation_delay": _NAVIGATION_DELAY = (lo, hi) else: globals()[target] = (lo, hi) def _apply_scroll_profile(key: str, raw: Any) -> None: if not isinstance(raw, dict): return base = int(raw.get("interval_base_ms", 0)) jitter = int(raw.get("interval_jitter_ms", 0)) ratio = float(raw.get("scroll_ratio", 0)) if base >= 0 and jitter >= 0 and 0 < ratio <= 1.0: _SCROLL_PROFILES[key] = (base, jitter, ratio) def reload_human_config(path: Path | None = None) -> bool: """从 JSON 加载人类行为参数；成功返回 True。""" cfg_path = path or _config_path() if cfg_path is None or not cfg_path.is_file(): return False try: data = json.loads(cfg_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False if not isinstance(data, dict): return False int_keys = { "default_max_attempts": "DEFAULT_MAX_ATTEMPTS", "stagnant_limit": "STAGNANT_LIMIT", "min_scroll_delta": "MIN_SCROLL_DELTA", "max_click_per_round": "MAX_CLICK_PER_ROUND", "stagnant_check_threshold": "STAGNANT_CHECK_THRESHOLD", "large_scroll_trigger": "LARGE_SCROLL_TRIGGER", "button_click_interval": "BUTTON_CLICK_INTERVAL", "final_sprint_push_count": "FINAL_SPRINT_PUSH_COUNT", } for json_key, attr in int_keys.items(): if json_key in data: globals()[attr] = int(data[json_key]) range_map = { "human_delay": "HUMAN_DELAY", "reaction_time": "REACTION_TIME", "hover_time": "HOVER_TIME", "read_time": "READ_TIME", "short_read": "SHORT_READ", "scroll_wait": "SCROLL_WAIT", "post_scroll": "POST_SCROLL", } for json_key, attr in range_map.items(): _apply_tuple(json_key, data.get(json_key), attr) _apply_tuple("navigation_delay", data.get("navigation_delay"), "navigation_delay") for speed in ("slow", "normal", "fast"): _apply_scroll_profile(speed, data.get(f"scroll_{speed}")) kw = data.get("inaccessible_keywords") if isinstance(kw, list): cleaned = [str(x).strip() for x in kw if str(x).strip()] if cleaned: globals()["INACCESSIBLE_KEYWORDS"] = cleaned for json_key, attr, default, legacy_key in ( ( "home_feed_collect_time", "HOME_FEED_COLLECT_TIME", HOME_FEED_COLLECT_TIME, "home_feed_collect_count", ), ( "search_feed_collect_time", "SEARCH_FEED_COLLECT_TIME", SEARCH_FEED_COLLECT_TIME, "search_feed_collect_count", ), ): raw = data.get(json_key) if not isinstance(raw, dict): raw = data.get(legacy_key) if isinstance(raw, dict): if json_key.endswith("_time") or "min_sec" in raw: lo = int(raw.get("min", raw.get("min_sec", default["min"]))) hi = int(raw.get("max", raw.get("max_sec", default["max"]))) else: lo = int(raw.get("min", default["min"])) hi = int(raw.get("max", default["max"])) if lo >= 60 and hi >= lo: lo = max(15, lo) else: lo = max(15, lo * 6) hi = max(lo, hi * 8) if lo >= 15 and hi >= lo: globals()[attr] = {"min": lo, "max": hi} def _merge_browse_plan(key: str, attr: str) -> None: raw_plan = data.get(key) if not isinstance(raw_plan, dict): return base = dict(globals()[attr]) for field, default in base.items(): if field in raw_plan: val = raw_plan[field] if isinstance(default, int): base[field] = int(val) else: base[field] = float(val) globals()[attr] = base _merge_browse_plan("home_browse_plan", "HOME_BROWSE_PLAN") _merge_browse_plan("search_browse_plan", "SEARCH_BROWSE_PLAN") return True def sleep_random(min_ms: int, max_ms: int) -> None: """随机延迟；浏览采集会话中若收到停止信号则立即中断。""" if max_ms <= min_ms: total = min_ms / 1000.0 else: total = random.randint(min_ms, max_ms) / 1000.0 _sleep_interruptible(total) def _sleep_interruptible(seconds: float) -> None: if seconds <= 0: return try: from .collect_buffer import is_cancelled from .errors import BrowseCancelledError except ImportError: time.sleep(seconds) return deadline = time.monotonic() + seconds while time.monotonic() < deadline: if is_cancelled(): raise BrowseCancelledError() time.sleep(min(0.25, max(0.0, deadline - time.monotonic()))) def navigation_delay() -> None: """页面导航后的随机等待，模拟人类阅读。""" sleep_random(_NAVIGATION_DELAY[0], _NAVIGATION_DELAY[1]) def get_scroll_interval(speed: str) -> float: """根据速度获取滚动间隔（秒）。""" prof = _SCROLL_PROFILES.get(speed) or _SCROLL_PROFILES["normal"] base, jitter, _ratio = prof return (base + random.randint(0, jitter)) / 1000.0 def get_scroll_ratio(speed: str) -> float: """根据速度获取滚动比例。""" prof = _SCROLL_PROFILES.get(speed) or _SCROLL_PROFILES["normal"] return prof[2] def calculate_scroll_delta(viewport_height: int, base_ratio: float) -> float: """计算滚动距离。""" scroll_delta = viewport_height * (base_ratio + random.random() * 0.2) if scroll_delta < 400: scroll_delta = 400.0 return scroll_delta + random.randint(-50, 50) def human_scroll_page( page, speed: str, large_mode: bool = False, push_count: int = 1, ) -> tuple[int, int]: """人类化页面滚动，返回 (实际滚动像素, 当前 scrollTop)。""" before_top = page.get_scroll_top() viewport_height = page.get_viewport_height() base_ratio = get_scroll_ratio(speed) if large_mode: base_ratio *= 2.0 actual_delta = 0 current_scroll_top = before_top for i in range(max(1, push_count)): scroll_delta = calculate_scroll_delta(viewport_height, base_ratio) page.scroll_by(0, int(scroll_delta)) sleep_random(*SCROLL_WAIT) current_scroll_top = page.get_scroll_top() delta_this = current_scroll_top - before_top actual_delta += delta_this before_top = current_scroll_top if i < push_count - 1: sleep_random(*HUMAN_DELAY) if actual_delta < MIN_SCROLL_DELTA and push_count > 0: page.scroll_to_bottom() sleep_random(*POST_SCROLL) current_scroll_top = page.get_scroll_top() actual_delta = current_scroll_top - (before_top - actual_delta) return actual_delta, current_scroll_top reload_human_config()