zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
205 lines
8.0 KiB
Python
205 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
# GUI 工具名 → CLI 子命令
|
|
TOOL_TO_COMMAND: dict[str, str] = {
|
|
"check_login_status": "check-login",
|
|
"browser_login": "login",
|
|
"delete_cookies": "delete-cookies",
|
|
"list_feeds": "list-feeds",
|
|
"search_feeds": "search-feeds",
|
|
"get_feed_detail": "get-feed-detail",
|
|
"publish_content": "publish",
|
|
"publish_with_video": "publish-video",
|
|
"post_comment_to_feed": "post-comment",
|
|
"reply_comment_in_feed": "reply-comment",
|
|
"like_feed": "like-feed",
|
|
"favorite_feed": "favorite-feed",
|
|
"user_profile": "user-profile",
|
|
"my_account_feeds": "my-account-feeds",
|
|
}
|
|
|
|
_FILTER_KEYS = (
|
|
("sort_by", "--sort-by"),
|
|
("note_type", "--note-type"),
|
|
("publish_time", "--publish-time"),
|
|
("search_scope", "--search-scope"),
|
|
("location", "--location"),
|
|
)
|
|
|
|
|
|
def build_cli_argv(tool_name: str, arguments: dict[str, Any] | None) -> list[str]:
|
|
"""将 GUI 工具参数转为 cli.py 参数列表（不含 python/cli 路径）。"""
|
|
args = dict(arguments or {})
|
|
command = TOOL_TO_COMMAND.get(tool_name)
|
|
if not command:
|
|
raise ValueError(f"未知工具: {tool_name}")
|
|
|
|
argv = [command]
|
|
|
|
if tool_name == "search_feeds":
|
|
keyword = str(args.pop("keyword", "")).strip()
|
|
if not keyword:
|
|
raise ValueError("search_feeds 需要 keyword")
|
|
argv.extend(["--keyword", keyword])
|
|
filters = args.pop("filters", None) or {}
|
|
if isinstance(filters, dict):
|
|
for key, flag in _FILTER_KEYS:
|
|
val = filters.get(key) or args.pop(key, None)
|
|
if val:
|
|
argv.extend([flag, str(val)])
|
|
for key, flag in _FILTER_KEYS:
|
|
val = args.pop(key, None)
|
|
if val:
|
|
argv.extend([flag, str(val)])
|
|
|
|
elif tool_name in ("get_feed_detail",):
|
|
argv.extend(["--feed-id", str(args.pop("feed_id", ""))])
|
|
argv.extend(["--xsec-token", str(args.pop("xsec_token", ""))])
|
|
if args.pop("load_all_comments", False):
|
|
argv.append("--load-all-comments")
|
|
limit = args.pop("limit", None)
|
|
if limit is not None:
|
|
argv.extend(["--max-comment-items", str(int(limit))])
|
|
|
|
elif tool_name in ("post_comment_to_feed",):
|
|
argv.extend(["--feed-id", str(args.pop("feed_id", ""))])
|
|
argv.extend(["--xsec-token", str(args.pop("xsec_token", ""))])
|
|
argv.extend(["--content", str(args.pop("content", ""))])
|
|
elif tool_name in ("reply_comment_in_feed",):
|
|
argv.extend(["--feed-id", str(args.pop("feed_id", ""))])
|
|
argv.extend(["--xsec-token", str(args.pop("xsec_token", ""))])
|
|
argv.extend(["--content", str(args.pop("content", ""))])
|
|
if args.get("comment_id"):
|
|
argv.extend(["--comment-id", str(args.pop("comment_id"))])
|
|
if args.get("user_id"):
|
|
argv.extend(["--user-id", str(args.pop("user_id"))])
|
|
elif tool_name in ("like_feed", "favorite_feed"):
|
|
argv.extend(["--feed-id", str(args.pop("feed_id", ""))])
|
|
argv.extend(["--xsec-token", str(args.pop("xsec_token", ""))])
|
|
if tool_name == "like_feed" and args.pop("unlike", False):
|
|
argv.append("--unlike")
|
|
if tool_name == "favorite_feed" and args.pop("unfavorite", False):
|
|
argv.append("--unfavorite")
|
|
|
|
elif tool_name == "user_profile":
|
|
argv.extend(["--user-id", str(args.pop("user_id", ""))])
|
|
argv.extend(["--xsec-token", str(args.pop("xsec_token", ""))])
|
|
|
|
elif tool_name == "publish_content":
|
|
title = str(args.pop("title", "")).strip()
|
|
content = str(args.pop("content", "")).strip()
|
|
images = args.pop("images", None) or []
|
|
if not images:
|
|
raise ValueError("publish_content 需要至少一张图片")
|
|
title_path, content_path = _write_publish_text_files(title, content)
|
|
argv.extend(["--title-file", str(title_path), "--content-file", str(content_path)])
|
|
argv.append("--images")
|
|
argv.extend([str(p) for p in images])
|
|
tags = args.pop("tags", None)
|
|
if tags:
|
|
argv.append("--tags")
|
|
argv.extend([str(t) for t in tags])
|
|
visibility = args.pop("visibility", None)
|
|
if visibility:
|
|
argv.extend(["--visibility", str(visibility)])
|
|
|
|
elif tool_name == "publish_with_video":
|
|
title = str(args.pop("title", "")).strip()
|
|
content = str(args.pop("content", "")).strip()
|
|
video = str(args.pop("video", "")).strip()
|
|
if not video:
|
|
raise ValueError("publish_with_video 需要 video 路径")
|
|
title_path, content_path = _write_publish_text_files(title, content)
|
|
argv.extend(
|
|
[
|
|
"--title-file",
|
|
str(title_path),
|
|
"--content-file",
|
|
str(content_path),
|
|
"--video",
|
|
video,
|
|
]
|
|
)
|
|
tags = args.pop("tags", None)
|
|
if tags:
|
|
argv.append("--tags")
|
|
argv.extend([str(t) for t in tags])
|
|
visibility = args.pop("visibility", None)
|
|
if visibility:
|
|
argv.extend(["--visibility", str(visibility)])
|
|
|
|
scroll_speed = args.pop("scroll_speed", None)
|
|
if scroll_speed and tool_name in ("list_feeds", "search_feeds", "my_account_feeds"):
|
|
argv.extend(["--scroll-speed", str(scroll_speed)])
|
|
|
|
return argv
|
|
|
|
|
|
def _write_publish_text_files(title: str, content: str) -> tuple[Path, Path]:
|
|
tmp = Path(tempfile.mkdtemp(prefix="xhs_publish_"))
|
|
title_path = tmp / "title.txt"
|
|
content_path = tmp / "content.txt"
|
|
title_path.write_text(title, encoding="utf-8")
|
|
content_path.write_text(content, encoding="utf-8")
|
|
return title_path, content_path
|
|
|
|
|
|
def format_tool_result_text(tool_name: str, data: dict[str, Any]) -> str:
|
|
"""将 CLI JSON 转为 GUI 习惯的文本（含 ✅/❌）。"""
|
|
if data.get("success") is False:
|
|
return f"❌ {data.get('error') or data.get('message') or '操作失败'}"
|
|
|
|
if tool_name == "check_login_status":
|
|
if data.get("logged_in"):
|
|
return "✅ 已登录"
|
|
hint = data.get("hint") or data.get("message") or "未登录"
|
|
return f"❌ {hint}"
|
|
|
|
if tool_name == "browser_login":
|
|
if data.get("logged_in"):
|
|
return f"✅ {data.get('message') or '登录成功'}"
|
|
return f"❌ {data.get('message') or '登录未完成或超时'}"
|
|
|
|
if tool_name == "delete_cookies":
|
|
return f"✅ {data.get('message') or '已退出登录'}"
|
|
|
|
if tool_name in ("list_feeds", "search_feeds"):
|
|
count = data.get("count", len(data.get("feeds") or []))
|
|
return f"✅ 获取到 {count} 条笔记"
|
|
|
|
if tool_name == "my_account_feeds":
|
|
count = data.get("count", len(data.get("feeds") or []))
|
|
nick = (data.get("basicInfo") or {}).get("nickname") or data.get("nickname") or ""
|
|
prefix = f"{nick} · " if nick else ""
|
|
return f"✅ {prefix}我的笔记 {count} 条"
|
|
|
|
if tool_name == "get_feed_detail":
|
|
note = data.get("note") or {}
|
|
title = note.get("title") or "（无标题）"
|
|
return f"✅ 笔记详情: {title}\n\n{json.dumps(data, ensure_ascii=False, indent=2)}"
|
|
|
|
if tool_name in ("publish_content", "publish_with_video"):
|
|
return f"✅ {data.get('status') or data.get('message') or '发布完成'}"
|
|
|
|
if tool_name in ("post_comment_to_feed", "reply_comment_in_feed"):
|
|
return f"✅ {data.get('message') or '评论成功'}"
|
|
|
|
if tool_name in ("like_feed", "favorite_feed"):
|
|
return f"✅ {data.get('message') or '操作成功'}"
|
|
|
|
if tool_name == "user_profile":
|
|
basic = (data.get("basicInfo") or {}) if isinstance(data.get("basicInfo"), dict) else {}
|
|
nick = basic.get("nickname") or "用户"
|
|
return f"✅ 用户主页: {nick}\n\n{json.dumps(data, ensure_ascii=False, indent=2)}"
|
|
|
|
if data.get("success") is True or data.get("logged_in"):
|
|
return f"✅ {data.get('message') or '成功'}"
|
|
|
|
return json.dumps(data, ensure_ascii=False, indent=2)
|