front/xhs-agent-desktop/engine/xhs_bridge/human_behavior.py
zzh 5a26ae0048 feat: 小红书模块迁移、登录重构与 xhs-agent-desktop 桌面 Agent
- 新增小红书前端模块(启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手)

- 新增 xhs-agent-desktop 独立桌面 Agent,对接 OMS 任务调度

- 登录页重构,整合微信登录与 OMS JWT 认证

- 订单、渠道、用户中心等页面与 API 增强

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-30 16:13:13 +08:00

511 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""浏览器内人类行为参数(对应 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()