from __future__ import annotations import json import random import time from dataclasses import dataclass, field from datetime import date, datetime, timedelta from pathlib import Path from typing import Any, Callable from xhs_bridge.exceptions import HumanGuardError ProgressFn = Callable[[str], None] # 发布类工具 PUBLISH_TOOLS = frozenset({"publish_content", "publish_with_video"}) # 浏览类动作键 BROWSE_ACTIONS = frozenset({ "list_feeds", "search_feeds", "my_account_feeds", "get_feed_detail", "rest_list_feeds", "rest_search_feeds", "rest_my_account_feeds", "rest_feed_detail", }) # 互动类 INTERACT_TOOLS = frozenset({ "post_comment_to_feed", "reply_comment_in_feed", "like_feed", "favorite_feed", }) # 不参与限流 EXEMPT_TOOLS = frozenset({ "check_login_status", "browser_login", "delete_cookies", }) EXTREME_WORDS = ( "最好", "第一", "全网最低", "绝对", "100%", "必看", "震惊", "秒杀", "顶级", "最强", "唯一", "国家级", "世界级", ) @dataclass class PublishGuardConfig: max_per_day: int = 2 min_interval_hours: float = 6.0 pre_delay_min_sec: int = 60 pre_delay_max_sec: int = 180 warmup_enabled: bool = True warmup_pause_min_sec: int = 30 warmup_pause_max_sec: int = 90 post_stay_min_sec: int = 0 post_stay_max_sec: int = 0 @dataclass class BrowseGuardConfig: min_interval_sec: int = 25 max_interval_sec: int = 45 max_per_hour: int = 20 @dataclass class InteractGuardConfig: min_interval_sec: int = 45 max_per_day: int = 10 @dataclass class QuietHoursConfig: enabled: bool = True start: str = "02:00" end: str = "07:00" @dataclass class ContentGuardConfig: max_title_len: int = 20 warn_extreme_words: bool = True @dataclass class HumanGuardConfig: enabled: bool = True publish: PublishGuardConfig = field(default_factory=PublishGuardConfig) browse: BrowseGuardConfig = field(default_factory=BrowseGuardConfig) interact: InteractGuardConfig = field(default_factory=InteractGuardConfig) quiet_hours: QuietHoursConfig = field(default_factory=QuietHoursConfig) content: ContentGuardConfig = field(default_factory=ContentGuardConfig) def parse_human_guard_config(raw: dict[str, Any] | None) -> HumanGuardConfig: raw = raw or {} pub = raw.get("publish") or {} br = raw.get("browse") or {} inter = raw.get("interact") or {} quiet = raw.get("quiet_hours") or {} content = raw.get("content") or {} return HumanGuardConfig( enabled=bool(raw.get("enabled", True)), publish=PublishGuardConfig( max_per_day=int(pub.get("max_per_day", 2)), min_interval_hours=float(pub.get("min_interval_hours", 6.0)), pre_delay_min_sec=int(pub.get("pre_delay_min_sec", 60)), pre_delay_max_sec=int(pub.get("pre_delay_max_sec", 180)), warmup_enabled=bool(pub.get("warmup_enabled", True)), warmup_pause_min_sec=int(pub.get("warmup_pause_min_sec", 30)), warmup_pause_max_sec=int(pub.get("warmup_pause_max_sec", 90)), post_stay_min_sec=int(pub.get("post_stay_min_sec", 0)), post_stay_max_sec=int(pub.get("post_stay_max_sec", 0)), ), browse=BrowseGuardConfig( min_interval_sec=int(br.get("min_interval_sec", 25)), max_interval_sec=int(br.get("max_interval_sec", 45)), max_per_hour=int(br.get("max_per_hour", 20)), ), interact=InteractGuardConfig( min_interval_sec=int(inter.get("min_interval_sec", 45)), max_per_day=int(inter.get("max_per_day", 10)), ), quiet_hours=QuietHoursConfig( enabled=bool(quiet.get("enabled", True)), start=str(quiet.get("start", "02:00")), end=str(quiet.get("end", "07:00")), ), content=ContentGuardConfig( max_title_len=int(content.get("max_title_len", 20)), warn_extreme_words=bool(content.get("warn_extreme_words", True)), ), ) def human_guard_config_to_dict(cfg: HumanGuardConfig) -> dict[str, Any]: """序列化为 config.yaml 的 human_guard 段。""" return { "enabled": cfg.enabled, "publish": { "max_per_day": cfg.publish.max_per_day, "min_interval_hours": cfg.publish.min_interval_hours, "pre_delay_min_sec": cfg.publish.pre_delay_min_sec, "pre_delay_max_sec": cfg.publish.pre_delay_max_sec, "warmup_enabled": cfg.publish.warmup_enabled, "warmup_pause_min_sec": cfg.publish.warmup_pause_min_sec, "warmup_pause_max_sec": cfg.publish.warmup_pause_max_sec, "post_stay_min_sec": cfg.publish.post_stay_min_sec, "post_stay_max_sec": cfg.publish.post_stay_max_sec, }, "browse": { "min_interval_sec": cfg.browse.min_interval_sec, "max_interval_sec": cfg.browse.max_interval_sec, "max_per_hour": cfg.browse.max_per_hour, }, "interact": { "min_interval_sec": cfg.interact.min_interval_sec, "max_per_day": cfg.interact.max_per_day, }, "quiet_hours": { "enabled": cfg.quiet_hours.enabled, "start": cfg.quiet_hours.start, "end": cfg.quiet_hours.end, }, "content": { "max_title_len": cfg.content.max_title_len, "warn_extreme_words": cfg.content.warn_extreme_words, }, } def validate_human_guard_config(cfg: HumanGuardConfig) -> list[str]: """校验配置合法性，返回错误信息列表（空=通过）。""" errors: list[str] = [] pub = cfg.publish if pub.pre_delay_max_sec < pub.pre_delay_min_sec: errors.append("发布前等待：上限不能小于下限") if pub.warmup_pause_max_sec < pub.warmup_pause_min_sec: errors.append("预热停留：上限不能小于下限") if pub.post_stay_max_sec < pub.post_stay_min_sec: errors.append("发布后停留：上限不能小于下限") if cfg.browse.max_interval_sec < cfg.browse.min_interval_sec: errors.append("浏览间隔：上限不能小于下限") for label, value in ( ("每日发布上限", pub.max_per_day), ("发布最短间隔(小时)", pub.min_interval_hours), ): if value <= 0: errors.append(f"{label} 须大于 0") if cfg.browse.max_per_hour <= 0: errors.append("每小时浏览上限须大于 0") if cfg.interact.max_per_day <= 0: errors.append("每日互动上限须大于 0") for label, text in (("静默开始", cfg.quiet_hours.start), ("静默结束", cfg.quiet_hours.end)): parts = text.strip().split(":") if len(parts) != 2: errors.append(f"{label} 须为 HH:MM 格式") continue try: h, m = int(parts[0]), int(parts[1]) if not (0 <= h <= 23 and 0 <= m <= 59): raise ValueError except ValueError: errors.append(f"{label} 须为有效时间（00:00～23:59）") return errors class HumanGuard: """单账号拟人化节奏与频率控制（本机浏览器引擎，不涉及多 IP / 多账号）。""" def __init__(self, config: HumanGuardConfig, state_path: Path) -> None: self.config = config self.state_path = state_path self._state: dict[str, Any] = {} self._load_state() def _load_state(self) -> None: if self.state_path.is_file(): try: with self.state_path.open("r", encoding="utf-8") as f: self._state = json.load(f) except (json.JSONDecodeError, OSError): self._state = {} if not isinstance(self._state, dict): self._state = {} def _save_state(self) -> None: self.state_path.parent.mkdir(parents=True, exist_ok=True) with self.state_path.open("w", encoding="utf-8") as f: json.dump(self._state, f, ensure_ascii=False, indent=2) @staticmethod def _today_key() -> str: return date.today().isoformat() def _daily_count(self, key: str) -> int: daily = self._state.setdefault("daily", {}) day = daily.get(self._today_key(), {}) return int(day.get(key, 0)) def _inc_daily(self, key: str) -> None: daily = self._state.setdefault("daily", {}) today = self._today_key() day = daily.setdefault(today, {}) day[key] = int(day.get(key, 0)) + 1 # 清理 7 天前的记录 cutoff = (date.today() - timedelta(days=7)).isoformat() for old in list(daily.keys()): if old < cutoff: del daily[old] self._save_state() def _last_ts(self, key: str) -> float | None: val = self._state.get(key) return float(val) if val is not None else None def _set_last_ts(self, key: str) -> None: self._state[key] = time.time() self._save_state() @staticmethod def _parse_hhmm(text: str) -> tuple[int, int]: parts = text.strip().split(":") h = int(parts[0]) if parts else 0 m = int(parts[1]) if len(parts) > 1 else 0 return h, m def _in_quiet_hours(self, now: datetime | None = None) -> bool: qh = self.config.quiet_hours if not qh.enabled: return False now = now or datetime.now() sh, sm = self._parse_hhmm(qh.start) eh, em = self._parse_hhmm(qh.end) cur = now.hour * 60 + now.minute start = sh * 60 + sm end = eh * 60 + em if start <= end: return start <= cur < end return cur >= start or cur < end def _browse_count_last_hour(self) -> int: events = self._state.get("browse_events") or [] cutoff = time.time() - 3600 recent = [t for t in events if float(t) >= cutoff] self._state["browse_events"] = recent[-200:] return len(recent) def _record_browse_event(self) -> None: events = self._state.setdefault("browse_events", []) events.append(time.time()) self._state["browse_events"] = events[-200:] self._save_state() @staticmethod def _sleep_random(min_sec: float, max_sec: float, on_progress: ProgressFn | None, label: str) -> float: if max_sec <= 0: return 0.0 duration = random.uniform(min_sec, max(max_sec, min_sec)) if on_progress: on_progress(f"{label}（约 {duration:.0f} 秒）…") time.sleep(duration) return duration def validate_publish_content(self, arguments: dict[str, Any]) -> list[str]: """内容合规提示（不阻断，仅警告）。""" warnings: list[str] = [] cfg = self.config.content title = str(arguments.get("title") or "").strip() content = str(arguments.get("content") or "").strip() if title and len(title) > cfg.max_title_len: warnings.append( f"标题 {len(title)} 字，建议 ≤ {cfg.max_title_len} 字（过长可能影响推荐）" ) if content and len(content) > 1000: warnings.append("正文较长，建议控制在 200～800 字更易阅读") if cfg.warn_extreme_words: blob = title + content hits = [w for w in EXTREME_WORDS if w in blob] if hits: warnings.append(f"含营销敏感词：{', '.join(hits[:5])}") return warnings def check_can_publish(self, now: datetime | None = None) -> tuple[bool, str]: if not self.config.enabled: return True, "" now = now or datetime.now() pub = self.config.publish if self._in_quiet_hours(now): return False, f"静默时段（{self.config.quiet_hours.start}～{self.config.quiet_hours.end}）不建议发布" today_count = self._daily_count("publish") if today_count >= pub.max_per_day: return False, f"今日已发布 {today_count} 篇，达到每日上限 {pub.max_per_day}" last = self._last_ts("last_publish") if last is not None: min_gap = pub.min_interval_hours * 3600 elapsed = time.time() - last if elapsed < min_gap: wait_min = int((min_gap - elapsed) / 60) + 1 return False, f"距上次发布未满 {pub.min_interval_hours:g} 小时，建议再等约 {wait_min} 分钟" return True, "" def check_can_browse(self) -> tuple[bool, str]: if not self.config.enabled: return True, "" br = self.config.browse if self._browse_count_last_hour() >= br.max_per_hour: return False, f"近 1 小时浏览请求已达 {br.max_per_hour} 次，请稍后再试" last = self._last_ts("last_browse") if last is not None: gap = random.uniform(br.min_interval_sec, br.max_interval_sec) elapsed = time.time() - last if elapsed < gap: wait = int(gap - elapsed) + 1 return False, f"浏览过快，建议 {wait} 秒后再操作" return True, "" def check_can_interact(self, tool_name: str) -> tuple[bool, str]: if not self.config.enabled: return True, "" inter = self.config.interact today = self._daily_count("interact") if today >= inter.max_per_day: return False, f"今日互动已达 {inter.max_per_day} 次上限" last = self._last_ts("last_interact") if last is not None: elapsed = time.time() - last if elapsed < inter.min_interval_sec: wait = int(inter.min_interval_sec - elapsed) + 1 return False, f"互动过快（{tool_name}），建议 {wait} 秒后再试" return True, "" @staticmethod def _sleep_interruptible( seconds: float, on_progress: ProgressFn | None, label: str, cancel_check: Callable[[], bool] | None, ) -> None: if seconds <= 0: return if on_progress: on_progress(f"{label} {seconds:.0f} 秒…") deadline = time.monotonic() + seconds while time.monotonic() < deadline: if cancel_check and cancel_check(): return time.sleep(min(0.25, max(0.0, deadline - time.monotonic()))) def wait_before_browse( self, action: str, on_progress: ProgressFn | None = None, *, cancel_check: Callable[[], bool] | None = None, ) -> None: if not self.config.enabled: if on_progress: on_progress("防封号节奏：完成") return ok, reason = self.check_can_browse() if not ok: raise HumanGuardError(reason) last = self._last_ts("last_browse") if last is not None: br = self.config.browse gap = random.uniform(br.min_interval_sec, br.max_interval_sec) elapsed = time.time() - last if elapsed < gap: wait = gap - elapsed if on_progress: on_progress("防封号节奏：等待中…") self._sleep_interruptible( wait, on_progress, "浏览节奏控制：等待", cancel_check, ) if on_progress: on_progress("防封号节奏：完成") def record_browse(self, action: str) -> None: self._set_last_ts("last_browse") self._record_browse_event() self._inc_daily("browse") def wait_before_interact(self, tool_name: str, on_progress: ProgressFn | None = None) -> None: if not self.config.enabled: return ok, reason = self.check_can_interact(tool_name) if not ok: raise HumanGuardError(reason) last = self._last_ts("last_interact") if last is not None: gap = self.config.interact.min_interval_sec elapsed = time.time() - last if elapsed < gap: wait = gap - elapsed if on_progress: on_progress(f"互动节奏：等待 {wait:.0f} 秒…") time.sleep(wait) def record_interact(self, tool_name: str) -> None: self._set_last_ts("last_interact") self._inc_daily("interact") def run_publish_prelude( self, warmup_fn: Callable[[], None] | None, on_progress: ProgressFn | None = None, ) -> list[str]: """ 发布前：校验 → 可选预热浏览 → 随机等待。 返回各阶段日志行。 """ log: list[str] = [] if not self.config.enabled: log.append("拟人化防护已关闭") return log ok, reason = self.check_can_publish() if not ok: raise HumanGuardError(reason) pub = self.config.publish if pub.warmup_enabled and warmup_fn is not None: if on_progress: on_progress("发布前预热：浏览首页推荐…") log.append("预热：拉取首页推荐") try: warmup_fn() except Exception as exc: log.append(f"预热浏览失败（已跳过）：{exc}") else: pause = self._sleep_random( pub.warmup_pause_min_sec, pub.warmup_pause_max_sec, on_progress, "预热后停留", ) log.append(f"预热停留 {pause:.0f}s") pre = self._sleep_random( pub.pre_delay_min_sec, pub.pre_delay_max_sec, on_progress, "发布前缓冲", ) log.append(f"发布前等待 {pre:.0f}s") return log def record_publish(self) -> None: self._set_last_ts("last_publish") self._inc_daily("publish") def run_publish_postlude(self, on_progress: ProgressFn | None = None) -> list[str]: log: list[str] = [] pub = self.config.publish if not self.config.enabled: return log if pub.post_stay_max_sec <= 0: return log stay = self._sleep_random( pub.post_stay_min_sec, pub.post_stay_max_sec, on_progress, "发布后停留", ) log.append(f"发布后停留 {stay:.0f}s") return log def status_summary(self) -> dict[str, Any]: """供 GUI / CLI 展示当前配额与下次可发布时间。""" pub = self.config.publish today_pub = self._daily_count("publish") today_interact = self._daily_count("interact") browse_hour = self._browse_count_last_hour() next_publish_hint = "" last_pub = self._last_ts("last_publish") if last_pub is not None and pub.min_interval_hours > 0: next_at = last_pub + pub.min_interval_hours * 3600 if next_at > time.time(): next_publish_hint = datetime.fromtimestamp(next_at).strftime("%H:%M") can_pub, pub_reason = self.check_can_publish() return { "enabled": self.config.enabled, "quiet_hours_active": self._in_quiet_hours(), "publish_today": today_pub, "publish_max_per_day": pub.max_per_day, "publish_can": can_pub, "publish_block_reason": pub_reason, "next_publish_after": next_publish_hint, "interact_today": today_interact, "interact_max_per_day": self.config.interact.max_per_day, "browse_last_hour": browse_hour, "browse_max_per_hour": self.config.browse.max_per_hour, "last_publish_at": ( datetime.fromtimestamp(last_pub).isoformat(timespec="minutes") if last_pub else None ), } def format_status_text(self) -> str: s = self.status_summary() if not s["enabled"]: return "拟人化防护：已关闭" lines = [ f"今日发布 {s['publish_today']}/{s['publish_max_per_day']}", f"今日互动 {s['interact_today']}/{s['interact_max_per_day']}", f"近1h浏览 {s['browse_last_hour']}/{s['browse_max_per_hour']}", ] if s["quiet_hours_active"]: lines.append("当前处于静默时段") if s["publish_can"]: lines.append("当前可以发布") else: lines.append(f"发布受限：{s['publish_block_reason']}") if s["next_publish_after"]: lines.append(f"建议下次发布不早于 {s['next_publish_after']}") return " · ".join(lines)