from __future__ import annotations from pathlib import Path from typing import Any from xhs_bridge.config import AppConfig, load_config from xhs_bridge.human_behavior import HumanBehaviorConfig, write_human_behavior_json from xhs_bridge.exceptions import HumanGuardError, LicenseRequiredError, SkillsError from xhs_bridge.human_guard import ( BROWSE_ACTIONS, EXEMPT_TOOLS, INTERACT_TOOLS, PUBLISH_TOOLS, HumanGuard, HumanGuardConfig, ProgressFn, ) from xhs_bridge.license import LicenseManager, LicenseRecord from xhs_bridge.skills_client import SkillsClient from xhs_bridge.skills_process_manager import SkillsProcessManager class XhsBridgeService: """LIKE AI 统一入口:Skills 引擎 + 授权 + 拟人化防护。""" def __init__(self, config: AppConfig | None = None, *, enable_http_log: bool = False) -> None: self.config = config or load_config() self.processes = SkillsProcessManager(self.config) self.license = LicenseManager(self.config.license) self.skills = SkillsClient(self.config) self.guard = HumanGuard(self.config.human_guard, self.config.human_guard_state_path()) self._status_cache: dict[str, Any] | None = None self._status_cache_at: float = 0.0 self._enable_http_log = enable_http_log # 保留参数,日志改为 CLI 记录 def invalidate_status_cache(self) -> None: self._status_cache = None self._status_cache_at = 0.0 def cached_status(self) -> dict[str, Any] | None: if self._status_cache is None: return None return dict(self._status_cache) def start_engine(self, force: bool = False) -> int: pid = self.processes.start_engine(force=force) self.invalidate_status_cache() return pid def stop_engine(self) -> None: self.processes.stop_engine() self.skills.close() self.invalidate_status_cache() def reset_login(self) -> dict[str, Any]: return self.clear_cookies(require_engine=True) def clear_cookies(self, *, require_engine: bool = False) -> dict[str, Any]: from xhs_bridge.extension_cookies import clear_cookie_cache skills_result: Any = None skills_error: str | None = None if require_engine: self.ensure_engine_running() skills_result = self.skills.delete_cookies() elif self.processes.is_bridge_running(): try: skills_result = self.skills.delete_cookies() except Exception as exc: skills_error = str(exc) clear_cookie_cache() self.invalidate_status_cache() return { "skills_result": skills_result, "skills_error": skills_error, } def browser_login(self, *, clear_first: bool = False) -> dict[str, Any]: """打开 Chrome,在小红书页面扫码登录(必须在浏览器中完成)。""" self.ensure_engine_running() self.processes.open_chrome() if clear_first: self.skills.delete_cookies() result = self.skills.browser_login(timeout=180) self.invalidate_status_cache() return result def relogin(self, *, clear_first: bool = True) -> dict[str, Any]: return self.browser_login(clear_first=clear_first) def ensure_engine_running(self) -> None: if not self.processes.is_engine_healthy(): self.start_engine() if not self.processes.is_engine_healthy(): self.processes.open_chrome() if not self.processes.wait_engine_ready(timeout_sec=25): ping = self.processes.ping_bridge() if not ping["extension_connected"]: raise SkillsError( "Chrome 扩展未连接。请在 chrome://extensions/ 加载 " f"{self.config.skills.extension_dir()} 并启用 XHS Bridge。" ) def activate(self, code: str) -> LicenseRecord: return self.license.activate(code) def require_license(self) -> LicenseRecord: return self.license.ensure_valid() def status( self, *, use_cache: bool = False, cache_ttl: float = 5.0, probe_login: bool = False, ) -> dict[str, Any]: import time if use_cache and self._status_cache is not None: if (time.time() - self._status_cache_at) < cache_ttl: return dict(self._status_cache) ping = self.processes.ping_bridge() engine_ok = ping["server_running"] and ping["extension_connected"] login_ok = bool(engine_ok) login_probed = False login_detail: dict[str, Any] | str = {} if engine_ok and probe_login: login_probed = True try: login_detail = self.skills.login_status() login_ok = bool(login_detail.get("is_logged_in")) except Exception as exc: login_detail = str(exc) login_ok = False elif engine_ok: login_detail = { "message": "未探测账号(在 Chrome 打开小红书登录即可)", } lic = self.license.load_cached() result = { "engine_running": engine_ok, "bridge_running": ping["server_running"], "extension_connected": ping["extension_connected"], "engine_url": self.config.skills.bridge_ws_url, "login_ok": login_ok, "login_probed": login_probed, "login_detail": login_detail, "license": None if lic is None else { "plan": lic.plan, "expires_at": lic.expires_at, "days_remaining": lic.days_remaining(), "expired": lic.is_expired(), }, "license_text": self.license.status_text(), "device_id": self.license.device_id, } self._status_cache = result self._status_cache_at = time.time() return dict(result) def check_login(self, use_rest: bool = True) -> dict[str, Any] | str: _ = use_rest self.ensure_engine_running() return self.skills.login_status() def human_guard_status(self) -> dict[str, Any]: return self.guard.status_summary() def human_guard_status_text(self) -> str: return self.guard.format_status_text() def apply_human_guard_config(self, cfg: HumanGuardConfig) -> None: self.config.human_guard = cfg self.guard.config = cfg def apply_human_behavior_config(self, cfg: HumanBehaviorConfig) -> None: self.config.human_behavior = cfg write_human_behavior_json(cfg, self.config.state_dir) def _tool_timeout(self, name: str) -> int: if name == "publish_with_video": return 900 if name in PUBLISH_TOOLS: return 600 return 300 def call_tool( self, name: str, arguments: dict[str, Any] | None = None, *, check_license: bool = True, on_progress: ProgressFn | None = None, skip_human_guard: bool = False, record_history: bool = True, ) -> str: if check_license and name not in EXEMPT_TOOLS: self.require_license() self.ensure_engine_running() args = arguments or {} if not skip_human_guard: if name in PUBLISH_TOOLS: return self._call_publish_tool( name, args, on_progress=on_progress, record_history=record_history, ) if name in INTERACT_TOOLS: self.guard.wait_before_interact(name, on_progress) elif name in BROWSE_ACTIONS: self.guard.wait_before_browse(name, on_progress) timeout = self._tool_timeout(name) text = self.skills.call_tool( name, args, timeout=timeout, on_progress=on_progress, ) if not skip_human_guard: if name in INTERACT_TOOLS: self.guard.record_interact(name) elif name in BROWSE_ACTIONS: self.guard.record_browse(name) return text def _call_publish_tool( self, name: str, arguments: dict[str, Any], *, on_progress: ProgressFn | None = None, record_history: bool = True, ) -> str: warnings = self.guard.validate_publish_content(arguments) for w in warnings: if on_progress: on_progress(f"⚠️ {w}") def warmup() -> None: self.skills.list_feeds() prelude_log = self.guard.run_publish_prelude(warmup_fn=warmup, on_progress=on_progress) timeout = self._tool_timeout(name) if on_progress: on_progress("正在通过浏览器发布…") text = self.skills.call_tool(name, arguments, timeout=timeout) success = "❌" not in text and "失败" not in text if success and record_history: self._record_publish_history(name, arguments, text) if not success: return text self.guard.record_publish() post_log = self.guard.run_publish_postlude(on_progress=on_progress) header = "【拟人化发布流程】\n" + "\n".join(prelude_log + post_log) if warnings: header += "\n\n【内容提示】\n" + "\n".join(f"- {w}" for w in warnings) return f"{header}\n\n{text}" @staticmethod def _record_publish_history( tool_name: str, arguments: dict[str, Any], result_text: str, ) -> None: try: from xhs_bridge.publish_history_store import PublishHistoryStore PublishHistoryStore().add_record( tool_name, arguments, result_text, success=True, ) except OSError: pass def publish_content( self, *, on_progress: ProgressFn | None = None, **kwargs: Any, ) -> str: self.require_license() self.ensure_engine_running() return self._call_publish_tool("publish_content", kwargs, on_progress=on_progress) def list_tools(self) -> list[dict[str, Any]]: self.ensure_engine_running() return self.skills.list_tools() def request_cancel_browse(self) -> None: """停止进行中的首页/搜索采集。""" self.skills.request_cancel_browse() def _browse_cancelled_result( self, data: dict[str, Any] | None = None, *, filters: dict[str, Any] | None = None, ) -> dict[str, Any]: from xhs_bridge.feed_filters import apply_client_feed_filters from xhs_bridge.feed_parser import parse_feeds_payload payload = data if data is not None else self.skills.browse_cancelled_payload() rows = parse_feeds_payload(payload) if filters: rows = apply_client_feed_filters(rows, filters) return {"feeds": rows, "cancelled": True} def fetch_feeds_list( self, filters: dict[str, Any] | None = None, *, on_progress: ProgressFn | None = None, browse_targets: dict[str, float | int] | None = None, ) -> dict[str, Any]: from xhs_bridge.feed_filters import apply_client_feed_filters from xhs_bridge.feed_parser import parse_feeds_payload self.require_license() self.ensure_engine_running() self.skills.begin_browse_task(browse_targets=browse_targets) cancel_check = self.skills.is_browse_cancel_requested if on_progress: on_progress("防封号节奏:检查中…") self.guard.wait_before_browse( "rest_list_feeds", on_progress, cancel_check=cancel_check ) if self.skills.is_browse_cancel_requested(): if on_progress: on_progress("已按您的请求停止采集") return self._browse_cancelled_result(filters=filters) data = self.skills.list_feeds(on_progress=on_progress) if self.skills.is_browse_cancel_requested() or data.get("cancelled"): rows = apply_client_feed_filters(parse_feeds_payload(data), filters) return {"feeds": rows, "cancelled": True} self.guard.record_browse("rest_list_feeds") rows = apply_client_feed_filters(parse_feeds_payload(data), filters) return { "feeds": rows, "cancelled": bool(data.get("cancelled")), } def fetch_feeds_search( self, keyword: str, filters: dict[str, Any] | None = None, *, on_progress: ProgressFn | None = None, browse_targets: dict[str, float | int] | None = None, ) -> dict[str, Any]: from xhs_bridge.feed_filters import apply_client_feed_filters from xhs_bridge.feed_parser import parse_feeds_payload self.require_license() self.ensure_engine_running() self.skills.begin_browse_task(browse_targets=browse_targets) cancel_check = self.skills.is_browse_cancel_requested if on_progress: on_progress("防封号节奏:检查中…") self.guard.wait_before_browse( "rest_search_feeds", on_progress, cancel_check=cancel_check ) if self.skills.is_browse_cancel_requested(): if on_progress: on_progress("已按您的请求停止采集") return self._browse_cancelled_result(filters=filters) data = self.skills.search_feeds(keyword, filters, on_progress=on_progress) if self.skills.is_browse_cancel_requested() or data.get("cancelled"): rows = apply_client_feed_filters(parse_feeds_payload(data), filters) return {"feeds": rows, "cancelled": True} self.guard.record_browse("rest_search_feeds") rows = apply_client_feed_filters(parse_feeds_payload(data), filters) return { "feeds": rows, "cancelled": bool(data.get("cancelled")), } def fetch_my_account_feeds( self, filters: dict[str, Any] | None = None, *, on_progress: ProgressFn | None = None, ) -> dict[str, Any]: from xhs_bridge.feed_filters import apply_client_feed_filters from xhs_bridge.feed_parser import parse_user_profile_feeds self.require_license() self.ensure_engine_running() self.skills.begin_browse_task() cancel_check = self.skills.is_browse_cancel_requested self.guard.wait_before_browse( "rest_my_account_feeds", on_progress, cancel_check=cancel_check ) if self.skills.is_browse_cancel_requested(): if on_progress: on_progress("已按您的请求停止采集") return self._browse_cancelled_result(filters=filters) if on_progress: on_progress("正在打开我的主页并加载笔记(图文/视频)…") data = self.skills.list_my_account_feeds(on_progress=on_progress) if self.skills.is_browse_cancel_requested() or data.get("cancelled"): rows = apply_client_feed_filters(parse_user_profile_feeds(data), filters) return {"feeds": rows, "cancelled": True} self.guard.record_browse("rest_my_account_feeds") rows = apply_client_feed_filters(parse_user_profile_feeds(data), filters) return { "feeds": rows, "cancelled": bool(data.get("cancelled")), "nickname": (data.get("basicInfo") or {}).get("nickname", ""), } def fetch_feed_detail( self, feed_id: str, xsec_token: str, *, on_progress: ProgressFn | None = None, ) -> str: self.require_license() self.ensure_engine_running() return self.call_tool( "get_feed_detail", {"feed_id": feed_id, "xsec_token": xsec_token}, check_license=False, on_progress=on_progress, ) def fetch_feed_detail_data( self, feed_id: str, xsec_token: str, *, load_all_comments: bool = False, limit: int = 20, on_progress: ProgressFn | None = None, ) -> dict[str, Any]: from xhs_bridge.feed_detail_parser import parse_feed_detail_payload self.require_license() self.ensure_engine_running() self.guard.wait_before_browse("rest_feed_detail", on_progress) resp = self.skills.feed_detail( { "feed_id": feed_id, "xsec_token": xsec_token, "load_all_comments": load_all_comments, "limit": limit, }, on_progress=on_progress, ) self.guard.record_browse("rest_feed_detail") return parse_feed_detail_payload(resp) def close(self) -> None: self.skills.close()