zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
350 lines
12 KiB
Python
350 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
from collections.abc import Callable
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from xhs_bridge.config import AppConfig
|
||
from xhs_bridge.exceptions import SkillsError
|
||
from xhs_bridge.human_behavior import human_behavior_json_path
|
||
from xhs_bridge.subprocess_win import popen_kwargs_hide_console
|
||
from xhs_bridge.tools_catalog import XHS_TOOLS
|
||
from xhs_bridge.skills_tool_adapter import build_cli_argv, format_tool_result_text
|
||
|
||
ProgressFn = Callable[[str], None] | None
|
||
|
||
|
||
class SkillsClient:
|
||
"""通过子进程调用内嵌 xiaohongshu-skills CLI。"""
|
||
|
||
_CANCELLABLE_COMMANDS = frozenset({
|
||
"list-feeds",
|
||
"search-feeds",
|
||
"my-account-feeds",
|
||
})
|
||
|
||
def __init__(self, config: AppConfig) -> None:
|
||
self.config = config
|
||
self._active_proc: subprocess.Popen[str] | None = None
|
||
self._cancel_requested = False
|
||
self._browse_targets: dict[str, float | int] | None = None
|
||
|
||
def _cli_script(self) -> Path:
|
||
return self.config.skills.cli_script()
|
||
|
||
def _browse_cancel_file(self) -> Path:
|
||
return self.config.state_dir / "browse.cancel"
|
||
|
||
def _browse_partial_file(self) -> Path:
|
||
return self.config.state_dir / "browse.partial.json"
|
||
|
||
def begin_browse_task(
|
||
self, *, browse_targets: dict[str, float | int] | None = None
|
||
) -> None:
|
||
"""新一次首页/搜索采集开始前调用，清除上一轮取消状态。"""
|
||
self._browse_targets = browse_targets
|
||
self._cancel_requested = False
|
||
self.config.state_dir.mkdir(parents=True, exist_ok=True)
|
||
cancel = self._browse_cancel_file()
|
||
if cancel.exists():
|
||
cancel.unlink()
|
||
partial = self._browse_partial_file()
|
||
partial.write_text(
|
||
json.dumps({"feeds": []}, ensure_ascii=False),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
def is_browse_cancel_requested(self) -> bool:
|
||
return self._cancel_requested
|
||
|
||
def browse_cancelled_payload(self) -> dict[str, Any]:
|
||
"""用户停止或子进程被中断时返回的最小结果。"""
|
||
partial = self._load_partial_payload()
|
||
if partial:
|
||
return partial
|
||
return {"feeds": [], "count": 0, "cancelled": True}
|
||
|
||
def _prepare_browse_session(self, env: dict[str, str]) -> None:
|
||
self.config.state_dir.mkdir(parents=True, exist_ok=True)
|
||
cancel = self._browse_cancel_file()
|
||
partial = self._browse_partial_file()
|
||
if not self._cancel_requested and cancel.exists():
|
||
cancel.unlink()
|
||
if not partial.exists():
|
||
partial.write_text(
|
||
json.dumps({"feeds": []}, ensure_ascii=False),
|
||
encoding="utf-8",
|
||
)
|
||
env["XHS_CANCEL_FILE"] = str(cancel)
|
||
env["XHS_PARTIAL_FEEDS_FILE"] = str(partial)
|
||
env["XHS_STATE_DIR"] = str(self.config.state_dir)
|
||
targets = getattr(self, "_browse_targets", None)
|
||
if targets:
|
||
env["XHS_BROWSE_TARGETS"] = json.dumps(targets, ensure_ascii=False)
|
||
|
||
def _load_partial_payload(self) -> dict[str, Any] | None:
|
||
partial = self._browse_partial_file()
|
||
if not partial.is_file():
|
||
return None
|
||
try:
|
||
data = json.loads(partial.read_text(encoding="utf-8"))
|
||
except (json.JSONDecodeError, OSError):
|
||
return None
|
||
if not isinstance(data, dict):
|
||
return None
|
||
feeds = data.get("feeds")
|
||
if not isinstance(feeds, list) or not feeds:
|
||
return None
|
||
return {"feeds": feeds, "count": len(feeds), "cancelled": True}
|
||
|
||
def request_cancel_browse(self) -> None:
|
||
"""停止当前 list-feeds / search-feeds 子进程并保留已落盘数据。"""
|
||
self._cancel_requested = True
|
||
cancel = self._browse_cancel_file()
|
||
cancel.parent.mkdir(parents=True, exist_ok=True)
|
||
cancel.write_text("1", encoding="utf-8")
|
||
proc = self._active_proc
|
||
if proc is not None and proc.poll() is None:
|
||
proc.kill()
|
||
|
||
def _run_cli(
|
||
self,
|
||
argv: list[str],
|
||
*,
|
||
timeout: int = 300,
|
||
on_progress: ProgressFn = None,
|
||
) -> tuple[dict[str, Any], int]:
|
||
env = dict(**__import__("os").environ)
|
||
if not getattr(sys, "frozen", False):
|
||
cli = self._cli_script()
|
||
if not cli.is_file():
|
||
raise SkillsError(f"CLI 脚本不存在: {cli}")
|
||
env["PYTHONPATH"] = str(self.config.skills.scripts_dir())
|
||
env["PYTHONUTF8"] = "1"
|
||
hb_json = human_behavior_json_path(self.config.state_dir)
|
||
if hb_json.is_file():
|
||
env["XHS_HUMAN_BEHAVIOR_FILE"] = str(hb_json)
|
||
cancellable = bool(argv) and argv[0] in self._CANCELLABLE_COMMANDS
|
||
if cancellable:
|
||
self._prepare_browse_session(env)
|
||
if self._cancel_requested:
|
||
return self.browse_cancelled_payload(), 0
|
||
|
||
cmd = self.config.skills.cli_launcher(argv)
|
||
if cmd and cmd[0].endswith(".py") and not Path(cmd[0]).is_file():
|
||
raise SkillsError(f"CLI 脚本不存在: {cmd[0]}")
|
||
if cmd and cmd[0].endswith(".exe") and not Path(cmd[0]).is_file():
|
||
raise SkillsError(f"CLI 引擎不存在: {cmd[0]}")
|
||
try:
|
||
proc = subprocess.Popen(
|
||
cmd,
|
||
cwd=str(self.config.skills.project_dir),
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
env=env,
|
||
**popen_kwargs_hide_console(),
|
||
)
|
||
except OSError as exc:
|
||
raise SkillsError(f"无法启动 CLI: {exc}") from exc
|
||
|
||
stderr_chunks: list[str] = []
|
||
stdout_chunks: list[str] = []
|
||
|
||
def _read_stderr() -> None:
|
||
stream = proc.stderr
|
||
if stream is None:
|
||
return
|
||
try:
|
||
for line in stream:
|
||
stderr_chunks.append(line)
|
||
stripped = line.strip()
|
||
if stripped.startswith("@@PROGRESS@@") and on_progress:
|
||
on_progress(stripped[len("@@PROGRESS@@") :])
|
||
except (ValueError, OSError):
|
||
# communicate / wait 结束后管道可能已关闭
|
||
pass
|
||
|
||
def _read_stdout() -> None:
|
||
stream = proc.stdout
|
||
if stream is None:
|
||
return
|
||
try:
|
||
stdout_chunks.append(stream.read())
|
||
except (ValueError, OSError):
|
||
pass
|
||
|
||
stderr_reader = threading.Thread(target=_read_stderr, daemon=True)
|
||
stdout_reader = threading.Thread(target=_read_stdout, daemon=True)
|
||
stderr_reader.start()
|
||
stdout_reader.start()
|
||
self._active_proc = proc
|
||
try:
|
||
deadline = time.monotonic() + timeout
|
||
while proc.poll() is None:
|
||
if self._cancel_requested:
|
||
proc.kill()
|
||
break
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
proc.kill()
|
||
proc.wait(timeout=5)
|
||
raise SkillsError(f"CLI 执行超时（{timeout}s）: {' '.join(argv)}")
|
||
try:
|
||
proc.wait(timeout=min(0.5, remaining))
|
||
except subprocess.TimeoutExpired:
|
||
continue
|
||
finally:
|
||
self._active_proc = None
|
||
stderr_reader.join(timeout=2)
|
||
stdout_reader.join(timeout=2)
|
||
|
||
stdout = stdout_chunks[0] if stdout_chunks else ""
|
||
stderr = "".join(stderr_chunks)
|
||
|
||
data = self._parse_json_output(stdout or "")
|
||
if data is None and cancellable and (self._cancel_requested or proc.returncode != 0):
|
||
partial = self._load_partial_payload()
|
||
if partial:
|
||
return partial, 0
|
||
if self._cancel_requested:
|
||
return self.browse_cancelled_payload(), 0
|
||
if data is None and proc.returncode != 0:
|
||
err = stderr.strip() or (stdout or "").strip() or f"退出码 {proc.returncode}"
|
||
raise SkillsError(err)
|
||
if data is None:
|
||
data = {"success": proc.returncode == 0, "raw": (stdout or "").strip()}
|
||
return data, proc.returncode
|
||
|
||
@staticmethod
|
||
def _parse_json_output(stdout: str) -> dict[str, Any] | None:
|
||
text = stdout.strip()
|
||
if not text:
|
||
return None
|
||
for candidate in (text, text.splitlines()[-1] if text else ""):
|
||
candidate = candidate.strip()
|
||
if not candidate.startswith("{"):
|
||
continue
|
||
try:
|
||
parsed = json.loads(candidate)
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
except json.JSONDecodeError:
|
||
continue
|
||
match = re.search(r"\{[\s\S]*\}\s*$", text)
|
||
if match:
|
||
try:
|
||
parsed = json.loads(match.group(0))
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return None
|
||
|
||
def call_tool(
|
||
self,
|
||
name: str,
|
||
arguments: dict[str, Any] | None = None,
|
||
timeout: int = 300,
|
||
on_progress: ProgressFn = None,
|
||
) -> str:
|
||
argv = build_cli_argv(name, arguments)
|
||
data, _code = self._run_cli(argv, timeout=timeout, on_progress=on_progress)
|
||
return format_tool_result_text(name, data)
|
||
|
||
def call_tool_json(
|
||
self,
|
||
name: str,
|
||
arguments: dict[str, Any] | None = None,
|
||
timeout: int = 300,
|
||
on_progress: ProgressFn = None,
|
||
) -> dict[str, Any]:
|
||
argv = build_cli_argv(name, arguments)
|
||
data, _code = self._run_cli(argv, timeout=timeout, on_progress=on_progress)
|
||
return data
|
||
|
||
def list_tools(self) -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"name": t["name"],
|
||
"description": t.get("description", ""),
|
||
"inputSchema": {"type": "object", "properties": {}},
|
||
}
|
||
for t in XHS_TOOLS
|
||
]
|
||
|
||
@staticmethod
|
||
def is_success_text(text: str) -> bool:
|
||
lowered = text.lower()
|
||
if "❌" in text or "失败" in text or "错误" in text:
|
||
return False
|
||
if "✅" in text or "成功" in text or "已登录" in text:
|
||
return True
|
||
return "未登录" not in text
|
||
|
||
def check_login_status(self) -> str:
|
||
return self.call_tool("check_login_status", timeout=90)
|
||
|
||
def list_feeds(self, *, on_progress: ProgressFn = None) -> dict[str, Any]:
|
||
return self.call_tool_json("list_feeds", timeout=300, on_progress=on_progress)
|
||
|
||
def list_my_account_feeds(self, *, on_progress: ProgressFn = None) -> dict[str, Any]:
|
||
return self.call_tool_json(
|
||
"my_account_feeds", timeout=360, on_progress=on_progress
|
||
)
|
||
|
||
def search_feeds(
|
||
self,
|
||
keyword: str,
|
||
filters: dict[str, Any] | None = None,
|
||
*,
|
||
on_progress: ProgressFn = None,
|
||
) -> dict[str, Any]:
|
||
args: dict[str, Any] = {"keyword": keyword}
|
||
if filters:
|
||
args["filters"] = filters
|
||
return self.call_tool_json("search_feeds", args, timeout=360, on_progress=on_progress)
|
||
|
||
def feed_detail(
|
||
self,
|
||
payload: dict[str, Any],
|
||
*,
|
||
on_progress: ProgressFn = None,
|
||
) -> dict[str, Any]:
|
||
return self.call_tool_json(
|
||
"get_feed_detail",
|
||
payload,
|
||
timeout=360,
|
||
on_progress=on_progress,
|
||
)
|
||
|
||
def login_status(self) -> dict[str, Any]:
|
||
data = self.call_tool_json("check_login_status", timeout=90)
|
||
logged_in = bool(data.get("logged_in"))
|
||
return {
|
||
"is_logged_in": logged_in,
|
||
"logged_in": logged_in,
|
||
"message": "已登录" if logged_in else (data.get("hint") or data.get("message") or "未登录"),
|
||
}
|
||
|
||
def browser_login(self, timeout: int = 180) -> dict[str, Any]:
|
||
"""在 Chrome 中打开小红书登录页并等待扫码（阻塞至完成或超时）。"""
|
||
data, _code = self._run_cli(["login"], timeout=timeout)
|
||
return data
|
||
|
||
def delete_cookies(self) -> dict[str, Any]:
|
||
return self.call_tool_json("delete_cookies", timeout=90)
|
||
|
||
def close(self) -> None:
|
||
if self._active_proc is not None and self._active_proc.poll() is None:
|
||
self._active_proc.kill()
|
||
self._active_proc = None
|