front/xhs-agent-desktop/engine/scripts/xhs/browse_plan_session.py

"""列表采集拟人计划：按保存的 5 项配置随机本次目标，不再按采集时长停止。"""

from __future__ import annotations

import json
import os
import random
import time
from dataclasses import dataclass
from typing import Any, Literal

from . import human

CollectKind = Literal["home", "search"]


@dataclass(frozen=True)
class BrowseRunTargets:
    scroll_target: int
    click_target: int
    stable_sec: float
    sprint_sec: float
    parse_sec: float
    detail_stay_sec: float = 15.0
    detail_scroll_target: int = 4

    def to_dict(self) -> dict[str, float | int]:
        return {
            "scroll_target": self.scroll_target,
            "click_target": self.click_target,
            "stable_sec": self.stable_sec,
            "sprint_sec": self.sprint_sec,
            "parse_sec": self.parse_sec,
            "detail_stay_sec": self.detail_stay_sec,
            "detail_scroll_target": self.detail_scroll_target,
        }

    @classmethod
    def from_dict(cls, raw: dict[str, Any]) -> BrowseRunTargets | None:
        try:
            scroll = int(raw["scroll_target"])
            click = int(raw["click_target"])
            stable = float(raw["stable_sec"])
            sprint = float(raw["sprint_sec"])
            parse = float(raw["parse_sec"])
            detail_stay = float(raw.get("detail_stay_sec", 15.0))
            detail_scroll = int(raw.get("detail_scroll_target", 4))
        except (KeyError, TypeError, ValueError):
            return None
        if scroll < 0 or click < 0:
            return None
        return cls(
            scroll_target=max(0, scroll),
            click_target=max(0, click),
            stable_sec=max(0.0, stable),
            sprint_sec=max(0.0, sprint),
            parse_sec=max(0.0, parse),
            detail_stay_sec=max(10.0, detail_stay),
            detail_scroll_target=max(1, detail_scroll),
        )


def _read_targets_from_env() -> BrowseRunTargets | None:
    raw = os.environ.get("XHS_BROWSE_TARGETS", "").strip()
    if not raw:
        return None
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return None
    if not isinstance(data, dict):
        return None
    return BrowseRunTargets.from_dict(data)


def _plan_dict(kind: CollectKind) -> dict[str, Any]:
    return human.HOME_BROWSE_PLAN if kind == "home" else human.SEARCH_BROWSE_PLAN


def sample_targets_from_plan(plan: dict[str, Any]) -> BrowseRunTargets:
    scroll_lo = max(0, int(plan.get("scroll_lo", 0)))
    scroll_hi = max(scroll_lo, int(plan.get("scroll_hi", scroll_lo)))
    click_lo = max(0, int(plan.get("click_lo", 0)))
    click_hi = max(click_lo, int(plan.get("click_hi", click_lo)))
    stable_lo = float(plan.get("stable_sec_lo", 2.0))
    stable_hi = max(stable_lo, float(plan.get("stable_sec_hi", stable_lo)))
    sprint_lo = float(plan.get("sprint_sec_lo", 4.0))
    sprint_hi = max(sprint_lo, float(plan.get("sprint_sec_hi", sprint_lo)))
    parse_lo = float(plan.get("parse_sec_lo", 1.5))
    parse_hi = max(parse_lo, float(plan.get("parse_sec_hi", parse_lo)))
    stay_lo = max(10.0, float(plan.get("detail_stay_sec_lo", 10.0)))
    stay_hi = max(stay_lo, float(plan.get("detail_stay_sec_hi", stay_lo)))
    dsc_lo = max(1, int(plan.get("detail_scroll_lo", 3)))
    dsc_hi = max(dsc_lo, int(plan.get("detail_scroll_hi", dsc_lo)))

    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 BrowseRunTargets(
        scroll_target=scroll_target,
        click_target=click_target,
        stable_sec=round(random.uniform(stable_lo, stable_hi), 2),
        sprint_sec=round(random.uniform(sprint_lo, sprint_hi), 2),
        parse_sec=round(random.uniform(parse_lo, parse_hi), 2),
        detail_stay_sec=round(random.uniform(stay_lo, stay_hi), 2),
        detail_scroll_target=random.randint(dsc_lo, dsc_hi),
    )


class BrowsePlanSession:
    """本次列表采集的拟人执行目标（滑动次数、点开次数等）。"""

    def __init__(self, kind: CollectKind) -> None:
        if kind == "home":
            self.label = "首页推荐"
        else:
            self.label = "搜索内容"
        env_targets = _read_targets_from_env()
        if env_targets is not None:
            self.targets = env_targets
        else:
            self.targets = sample_targets_from_plan(_plan_dict(kind))
        self._start = time.monotonic()

    @property
    def scroll_target(self) -> int:
        return self.targets.scroll_target

    @property
    def click_target(self) -> int:
        return self.targets.click_target

    def start_message(self) -> str:
        t = self.targets
        parts = [f"滑动 {t.scroll_target} 次"]
        if t.click_target > 0:
            parts.append(f"点开 {t.click_target} 次")
        return f"本次拟人计划（{self.label}）：{'，'.join(parts)}"

    def elapsed_sec(self) -> float:
        return time.monotonic() - self._start

    def summary(
        self,
        count: int,
        *,
        plan_completed: bool = True,
        user_cancelled: bool = False,
    ) -> str:
        elapsed = self.elapsed_sec()
        t = self.targets
        plan_hint = f"计划滑动 {t.scroll_target} 次"
        if t.click_target > 0:
            plan_hint += f"、点开 {t.click_target} 次"
        if user_cancelled:
            return f"已停止，返回 {count} 条；{plan_hint}，用时 {elapsed:.1f} 秒"
        if plan_completed:
            return f"拟人流程完成，共 {count} 条；{plan_hint}，用时 {elapsed:.1f} 秒"
        return f"共采集 {count} 条；{plan_hint}，用时 {elapsed:.1f} 秒"