zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
299 lines
9.3 KiB
Python
299 lines
9.3 KiB
Python
"""人类行为模拟参数（延迟、滚动、悬停），对应 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()
|