Angiin
Committed by GitHub

Merge pull request #20 from autoclaw-cc/feat/multi-account-port-isolation

- 增加多账号功能,使用不同的端口对应不同的账号
- 多账号操作:执行任务前,和用户确认在什么账号上操作
- 修复手机验证码体验:使用手机验证码登录时,流程过长,会导致验证码失败
... ... @@ -21,7 +21,7 @@ uv run pytest # 运行测试
双层结构:`scripts/` 是 Python CDP 自动化引擎,`skills/` 是 Claude Code Skills 定义(SKILL.md 格式)。
- `scripts/xhs/` — 核心自动化库(模块化,每个功能一个文件)
- `scripts/cli.py` — 统一 CLI 入口,19 个子命令,JSON 结构化输出
- `scripts/cli.py` — 统一 CLI 入口,23 个子命令,JSON 结构化输出
- `scripts/publish_pipeline.py` — 发布编排器(含图片下载和登录检查)
- `skills/*/SKILL.md` — 指导 Claude 如何调用 scripts/
... ... @@ -74,3 +74,7 @@ python scripts/publish_pipeline.py --title-file t.txt --content-file c.txt --ima
| `long-article` | — | 长文发布(填写+排版) |
| `select-template` | — | 长文发布(选择模板) |
| `next-step` | — | 长文发布(下一步+描述) |
| `add-account` | — | 账号管理(添加,自动分配端口) |
| `list-accounts` | — | 账号管理(列出所有) |
| `remove-account` | — | 账号管理(删除) |
| `set-default-account` | — | 账号管理(设置默认) |
... ...
... ... @@ -13,6 +13,9 @@ logger = logging.getLogger(__name__)
_CONFIG_DIR = Path.home() / ".xhs"
_ACCOUNTS_FILE = _CONFIG_DIR / "accounts.json"
# 命名账号端口起始值(默认账号使用 9222)
_NAMED_PORT_START = 9223
def _load_config() -> dict:
"""加载账号配置。"""
... ... @@ -41,20 +44,25 @@ def list_accounts() -> list[dict]:
"name": name,
"description": info.get("description", ""),
"is_default": name == default,
"profile_dir": _get_profile_dir(name),
"profile_dir": get_profile_dir(name),
"port": info.get("port", _NAMED_PORT_START),
}
)
return result
def add_account(name: str, description: str = "") -> None:
"""添加账号。"""
"""添加账号,自动分配独立端口(从 _NAMED_PORT_START 递增)。"""
config = _load_config()
accounts = config.setdefault("accounts", {})
if name in accounts:
raise ValueError(f"账号 '{name}' 已存在")
accounts[name] = {"description": description}
# 自动分配端口:取已有端口的最大值(至少 _NAMED_PORT_START - 1)加 1
existing_ports = {info.get("port", _NAMED_PORT_START) for info in accounts.values()}
port = max(existing_ports | {_NAMED_PORT_START - 1}) + 1
accounts[name] = {"description": description, "port": port}
# 如果是第一个账号,设为默认
if not config.get("default"):
... ... @@ -63,10 +71,10 @@ def add_account(name: str, description: str = "") -> None:
_save_config(config)
# 创建 Profile 目录
profile_dir = _get_profile_dir(name)
profile_dir = get_profile_dir(name)
os.makedirs(profile_dir, exist_ok=True)
logger.info("添加账号: %s", name)
logger.info("添加账号: %s (port=%d)", name, port)
def remove_account(name: str) -> None:
... ... @@ -98,12 +106,37 @@ def set_default_account(name: str) -> None:
logger.info("默认账号设置为: %s", name)
def update_account_description(name: str, description: str) -> None:
"""更新账号描述(通常用于存储平台昵称)。"""
config = _load_config()
accounts = config.get("accounts", {})
if name not in accounts:
raise ValueError(f"账号 '{name}' 不存在")
accounts[name]["description"] = description
_save_config(config)
logger.info("账号 %s 描述已更新: %s", name, description)
def get_default_account() -> str:
"""获取默认账号名称。"""
config = _load_config()
return config.get("default", "")
def _get_profile_dir(account: str) -> str:
def get_profile_dir(account: str) -> str:
"""获取账号的 Chrome Profile 目录。"""
return str(_CONFIG_DIR / "accounts" / account / "chrome-profile")
def _get_profile_dir(account: str) -> str:
"""获取账号的 Chrome Profile 目录(别名,向后兼容)。"""
return get_profile_dir(account)
def get_account_port(name: str) -> int:
"""获取指定账号的 Chrome 调试端口。"""
config = _load_config()
accounts = config.get("accounts", {})
if name not in accounts:
raise ValueError(f"账号 '{name}' 不存在")
return accounts[name].get("port", _NAMED_PORT_START)
... ...
... ... @@ -15,47 +15,52 @@ import os
import sys
import tempfile
# 记录登录用 tab 的 target_id,确保 verify-code / wait-login 连回精确的那个 tab
_LOGIN_TAB_FILE = os.path.join(tempfile.gettempdir(), "xhs", "login_tab_id.txt")
def _session_tab_file(port: int) -> str:
"""返回指定端口的 session tab 文件路径(每账号独立隔离)。"""
return os.path.join(tempfile.gettempdir(), "xhs", f"session_tab_{port}.txt")
# 记录上次命令使用的 tab,供下次命令复用,避免重复开新 tab
_SESSION_TAB_FILE = os.path.join(tempfile.gettempdir(), "xhs", "session_tab_id.txt")
def _login_tab_file(port: int) -> str:
"""返回指定端口的 login tab 文件路径(每账号独立隔离)。"""
return os.path.join(tempfile.gettempdir(), "xhs", f"login_tab_{port}.txt")
def _save_login_tab(target_id: str) -> None:
os.makedirs(os.path.dirname(_LOGIN_TAB_FILE), exist_ok=True)
with open(_LOGIN_TAB_FILE, "w") as f:
def _save_login_tab(target_id: str, port: int) -> None:
path = _login_tab_file(port)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
f.write(target_id)
def _load_login_tab() -> str | None:
def _load_login_tab(port: int) -> str | None:
with contextlib.suppress(FileNotFoundError):
data = open(_LOGIN_TAB_FILE).read().strip()
data = open(_login_tab_file(port)).read().strip()
return data or None
return None
def _clear_login_tab() -> None:
def _clear_login_tab(port: int) -> None:
with contextlib.suppress(FileNotFoundError):
os.remove(_LOGIN_TAB_FILE)
os.remove(_login_tab_file(port))
def _save_session_tab(target_id: str) -> None:
os.makedirs(os.path.dirname(_SESSION_TAB_FILE), exist_ok=True)
with open(_SESSION_TAB_FILE, "w") as f:
def _save_session_tab(target_id: str, port: int) -> None:
path = _session_tab_file(port)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
f.write(target_id)
def _load_session_tab() -> str | None:
def _load_session_tab(port: int) -> str | None:
with contextlib.suppress(FileNotFoundError):
data = open(_SESSION_TAB_FILE).read().strip()
data = open(_session_tab_file(port)).read().strip()
return data or None
return None
def _clear_session_tab() -> None:
def _clear_session_tab(port: int) -> None:
with contextlib.suppress(FileNotFoundError):
os.remove(_SESSION_TAB_FILE)
os.remove(_session_tab_file(port))
# Windows 控制台默认编码(如 cp1252)不支持中文,强制 UTF-8
if sys.stdout and hasattr(sys.stdout, "reconfigure"):
... ... @@ -76,16 +81,51 @@ def _output(data: dict, exit_code: int = 0) -> None:
sys.exit(exit_code)
def _update_account_nickname(args: argparse.Namespace, page) -> None:
"""登录成功后,将平台昵称写入账号描述(best-effort,失败不影响登录结果)。"""
if not getattr(args, "account", ""):
return
import sys as _sys
_sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
import account_manager
from xhs.login import get_current_user_nickname
try:
nickname = get_current_user_nickname(page)
if nickname:
account_manager.update_account_description(args.account, nickname)
logger.info("账号 %s 昵称已更新: %s", args.account, nickname)
except Exception as e:
logger.warning("更新账号昵称失败: %s", e)
def _resolve_account(args: argparse.Namespace) -> str | None:
"""解析 --account 参数,更新 args.port,返回 user_data_dir(无账号时返回 None)。"""
if not getattr(args, "account", ""):
return None
import sys as _sys
_sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
import account_manager
name = args.account
args.port = account_manager.get_account_port(name)
return account_manager.get_profile_dir(name)
def _connect(args: argparse.Namespace):
"""连接到 Chrome 并返回 (browser, page)。
优先复用上次命令留下的 tab(通过 _SESSION_TAB_FILE 记录),
优先复用上次命令留下的 tab(通过端口隔离的 session tab 文件记录),
避免每次命令都新建 tab 导致 Chrome 中 tab 堆积。
"""
from chrome_launcher import ensure_chrome, has_display
from xhs.cdp import Browser
if not ensure_chrome(port=args.port, headless=not has_display()):
user_data_dir = _resolve_account(args)
if not ensure_chrome(port=args.port, headless=not has_display(), user_data_dir=user_data_dir):
_output(
{"success": False, "error": "无法启动 Chrome,请检查 Chrome 是否已安装"},
exit_code=2,
... ... @@ -95,32 +135,34 @@ def _connect(args: argparse.Namespace):
browser.connect()
# 优先复用上次命令留下的 tab
saved_id = _load_session_tab()
saved_id = _load_session_tab(args.port)
if saved_id:
page = browser.get_page_by_target_id(saved_id)
if page:
logger.debug("复用会话 tab: %s", saved_id)
_save_session_tab(page.target_id)
_save_session_tab(page.target_id, args.port)
return browser, page
logger.warning("会话 tab (target_id=%s) 已失效,重新获取", saved_id)
page = browser.get_or_create_page()
_save_session_tab(page.target_id)
_save_session_tab(page.target_id, args.port)
return browser, page
def _connect_saved_tab(args: argparse.Namespace):
"""连接到登录流程中记录的精确 tab(via _LOGIN_TAB_FILE),回退到第一个非空白 tab。"""
"""连接到登录流程中记录的精确 tab,回退到第一个非空白 tab。"""
from chrome_launcher import ensure_chrome, has_display
from xhs.cdp import Browser
if not ensure_chrome(port=args.port, headless=not has_display()):
user_data_dir = _resolve_account(args)
if not ensure_chrome(port=args.port, headless=not has_display(), user_data_dir=user_data_dir):
_output({"success": False, "error": "无法连接到 Chrome"}, exit_code=2)
browser = Browser(host=args.host, port=args.port)
browser.connect()
target_id = _load_login_tab()
target_id = _load_login_tab(args.port)
if target_id:
page = browser.get_page_by_target_id(target_id)
if page:
... ... @@ -141,7 +183,9 @@ def _connect_existing(args: argparse.Namespace):
from chrome_launcher import ensure_chrome, has_display
from xhs.cdp import Browser
if not ensure_chrome(port=args.port, headless=not has_display()):
user_data_dir = _resolve_account(args)
if not ensure_chrome(port=args.port, headless=not has_display(), user_data_dir=user_data_dir):
_output(
{"success": False, "error": "无法连接到 Chrome"},
exit_code=2,
... ... @@ -244,6 +288,8 @@ def cmd_login(args: argparse.Namespace) -> None:
)
)
success = wait_for_login(page, timeout=120)
if success:
_update_account_nickname(args, page)
_output(
{"logged_in": success, "message": "登录成功" if success else "登录超时"},
exit_code=0 if success else 2,
... ... @@ -293,7 +339,7 @@ def cmd_phone_login(args: argparse.Namespace) -> None:
exit_code=0 if success else 2,
)
finally:
browser.close_page(page)
# 不关闭 tab——与 verify-code 一致,保留页面供重试
browser.close()
... ... @@ -318,8 +364,10 @@ def cmd_get_qrcode(args: argparse.Namespace) -> None:
qrcode_path = save_qrcode_to_file(png_bytes)
# 记录 tab,供 wait-login 精确reconnect
_save_login_tab(page.target_id)
# 记录 login tab,供 wait-login 精确 reconnect
_save_login_tab(page.target_id, args.port)
# 清除 session tab 引用——隔离登录表单,防止其他命令复用并关闭/导航该 tab
_clear_session_tab(args.port)
# 只断开 CDP 连接,不关闭 tab——QR 会话保持,用户可继续扫码
browser.close()
... ... @@ -340,7 +388,8 @@ def cmd_wait_login(args: argparse.Namespace) -> None:
try:
success = wait_for_login(page, timeout=args.timeout)
if success:
_clear_login_tab()
_clear_login_tab(args.port)
_update_account_nickname(args, page)
_output(
{
"logged_in": success,
... ... @@ -366,8 +415,10 @@ def cmd_send_code(args: argparse.Namespace) -> None:
_output({"logged_in": True, "message": "已登录,无需重新登录"})
return
# 记录 tab,供 verify-code 精确 reconnect
_save_login_tab(page.target_id)
# 记录 login tab,供 verify-code 精确 reconnect
_save_login_tab(page.target_id, args.port)
# 清除 session tab 引用——隔离登录表单,防止其他命令复用并关闭/导航该 tab
_clear_session_tab(args.port)
_output({
"status": "code_sent",
"message": f"验证码已发送至 {args.phone[:3]}****{args.phone[-4:]},请运行 verify-code --code <验证码>",
... ... @@ -393,13 +444,14 @@ def cmd_verify_code(args: argparse.Namespace) -> None:
try:
success = submit_phone_code(page, args.code)
if success:
_clear_login_tab()
_clear_login_tab(args.port)
_update_account_nickname(args, page)
_output(
{"logged_in": success, "message": "登录成功" if success else "验证码错误或超时"},
exit_code=0 if success else 2,
)
finally:
browser.close_page(page)
# 不关闭 tab——成功后供后续命令复用,失败后用户可再次运行 verify-code 重试
browser.close()
... ... @@ -420,7 +472,7 @@ def cmd_delete_cookies(args: argparse.Namespace) -> None:
path = get_cookies_file_path(args.account)
delete_cookies(path)
_clear_session_tab() # 退出登录后清除会话 tab 记录
_clear_session_tab(args.port) # 退出登录后清除会话 tab 记录
msg = "已退出登录并删除 cookies" if logged_out else "未登录,已删除 cookies 文件"
_output({"success": True, "message": msg, "cookies_path": path})
... ... @@ -817,6 +869,55 @@ def cmd_publish_video(args: argparse.Namespace) -> None:
browser.close()
# ========== 账号管理子命令 ==========
def cmd_add_account(args: argparse.Namespace) -> None:
"""添加命名账号,自动分配独立端口和 Chrome Profile。"""
import sys as _sys
_sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
import account_manager
account_manager.add_account(args.name, description=args.description or "")
port = account_manager.get_account_port(args.name)
profile = account_manager.get_profile_dir(args.name)
_output({"success": True, "name": args.name, "port": port, "profile_dir": profile})
def cmd_list_accounts(args: argparse.Namespace) -> None:
"""列出所有命名账号。"""
import sys as _sys
_sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
import account_manager
accounts = account_manager.list_accounts()
_output({"accounts": accounts, "count": len(accounts)})
def cmd_remove_account(args: argparse.Namespace) -> None:
"""删除命名账号。"""
import sys as _sys
_sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
import account_manager
account_manager.remove_account(args.name)
_output({"success": True, "name": args.name})
def cmd_set_default_account(args: argparse.Namespace) -> None:
"""设置默认账号。"""
import sys as _sys
_sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
import account_manager
account_manager.set_default_account(args.name)
_output({"success": True, "default": args.name})
# ========== 参数解析 ==========
... ... @@ -1001,6 +1102,26 @@ def build_parser() -> argparse.ArgumentParser:
sub = subparsers.add_parser("save-draft", help="保存为草稿(取消发布时使用)")
sub.set_defaults(func=cmd_save_draft)
# add-account(添加命名账号)
sub = subparsers.add_parser("add-account", help="添加命名账号,自动分配独立端口")
sub.add_argument("--name", required=True, help="账号名称")
sub.add_argument("--description", default="", help="账号描述(可选)")
sub.set_defaults(func=cmd_add_account)
# list-accounts(列出所有账号)
sub = subparsers.add_parser("list-accounts", help="列出所有命名账号")
sub.set_defaults(func=cmd_list_accounts)
# remove-account(删除账号)
sub = subparsers.add_parser("remove-account", help="删除命名账号")
sub.add_argument("--name", required=True, help="账号名称")
sub.set_defaults(func=cmd_remove_account)
# set-default-account(设置默认账号)
sub = subparsers.add_parser("set-default-account", help="设置默认账号")
sub.add_argument("--name", required=True, help="账号名称")
sub.set_defaults(func=cmd_set_default_account)
return parser
... ...
... ... @@ -28,12 +28,28 @@ from .selectors import (
PHONE_INPUT,
PHONE_LOGIN_SUBMIT,
QRCODE_IMG,
USER_NICKNAME,
USER_PROFILE_NAV_LINK,
)
from .urls import EXPLORE_URL
logger = logging.getLogger(__name__)
def _wait_for_countdown(page: Page, timeout: float = 5.0) -> None:
"""等待"获取验证码"按钮出现倒计时数字,确认验证码已发送。
轮询按钮文字直到包含数字(如 "60s"),超时则抛出 RateLimitError。
"""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
btn_text = page.get_element_text(GET_CODE_BUTTON) or ""
if any(ch.isdigit() for ch in btn_text):
return
time.sleep(0.3)
raise RateLimitError()
def _wait_for_auth_ui(page: Page, timeout: float = 8.0) -> None:
"""等待认证 UI 出现,替代固定延迟。
... ... @@ -47,6 +63,40 @@ def _wait_for_auth_ui(page: Page, timeout: float = 8.0) -> None:
time.sleep(0.2)
def get_current_user_nickname(page: Page) -> str:
"""获取当前登录用户的真实昵称,失败时返回空字符串(best-effort)。
流程:首页导航栏取个人主页 href → 导航过去 → 读 .user-name 文字。
"""
try:
page.navigate(EXPLORE_URL)
page.wait_for_load()
_wait_for_auth_ui(page)
if not page.has_element(LOGIN_STATUS):
return ""
# 从导航栏"我"的链接取个人主页 URL(含 /user/profile/<user_id>)
profile_href = page.evaluate(
f"document.querySelector({json.dumps(USER_PROFILE_NAV_LINK)})?.getAttribute('href') || ''"
)
if not profile_href:
return ""
# 导航到个人主页读取真实昵称
profile_url = f"https://www.xiaohongshu.com{profile_href}"
page.navigate(profile_url)
page.wait_for_load()
page.wait_dom_stable()
nickname = page.evaluate(
f"document.querySelector({json.dumps(USER_NICKNAME)})?.innerText?.trim() || ''"
)
return nickname or ""
except Exception:
logger.warning("获取用户昵称失败")
return ""
def check_login_status(page: Page) -> bool:
"""检查登录状态。
... ... @@ -135,20 +185,26 @@ def send_phone_code(page: Page, phone: str) -> bool:
"""
page.navigate(EXPLORE_URL)
page.wait_for_load()
sleep_random(1500, 2500)
# 直接等待登录容器出现(合并了 _wait_for_auth_ui 的逻辑,避免重复等待)
try:
page.wait_for_element(LOGIN_CONTAINER, timeout=10.0)
except Exception as exc:
# 可能已登录(没有登录容器),检查登录状态
if page.has_element(LOGIN_STATUS):
return False
raise RuntimeError("找不到登录表单") from exc
# 等待登录弹窗出现
page.wait_for_element(LOGIN_CONTAINER, timeout=15.0)
sleep_random(500, 800)
if page.has_element(LOGIN_STATUS):
return False
sleep_random(200, 400)
# 点击手机号输入框并逐字输入
page.click_element(PHONE_INPUT)
sleep_random(200, 400)
page.type_text(phone, delay_ms=80)
sleep_random(500, 800)
sleep_random(200, 400)
# 先勾选用户协议,再点获取验证码
if not page.has_element(AGREE_CHECKBOX_CHECKED):
... ... @@ -157,12 +213,9 @@ def send_phone_code(page: Page, phone: str) -> bool:
# 点击"获取验证码"
page.click_element(GET_CODE_BUTTON)
sleep_random(2000, 2500)
# 检测按钮是否变为倒计时(成功发送后按钮文字会包含数字秒数)
btn_text = page.get_element_text(GET_CODE_BUTTON) or ""
if not any(ch.isdigit() for ch in btn_text):
raise RateLimitError()
# 事件驱动:轮询按钮文字直到出现倒计时数字,替代固定 2-2.5s 等待
_wait_for_countdown(page)
logger.info("验证码已发送至 %s", phone[:3] + "****" + phone[-4:])
return True
... ... @@ -178,15 +231,27 @@ def submit_phone_code(page: Page, code: str) -> bool:
Returns:
True 登录成功,False 失败(超时或验证码错误)。
"""
# 点击验证码输入框并逐字输入
# 点击验证码输入框,先清空再用 CDP 键盘事件逐字输入(isTrusted=true,React 能识别)
page.click_element(CODE_INPUT)
sleep_random(300, 500)
page.type_text(code, delay_ms=100)
sleep_random(500, 800)
sleep_random(100, 200)
page.evaluate(
f"""(() => {{
const el = document.querySelector({json.dumps(CODE_INPUT)});
if (el && el.value) {{
const setter = Object.getOwnPropertyDescriptor(
window.HTMLInputElement.prototype, 'value'
).set;
setter.call(el, '');
el.dispatchEvent(new Event('input', {{ bubbles: true }}));
}}
}})()"""
)
page.type_text(code, delay_ms=0)
sleep_random(100, 200)
# 点击登录按钮
page.click_element(PHONE_LOGIN_SUBMIT)
sleep_random(1000, 2000)
sleep_random(500, 1000)
# 检查是否有错误提示
err = page.get_element_text(LOGIN_ERR_MSG)
... ...
... ... @@ -91,3 +91,7 @@ LOGOUT_MENU_ITEM = 'div.menu-item[data-name="退出登录"]'
# ========== 用户主页 ==========
SIDEBAR_PROFILE = "div.main-container li.user.side-bar-component a.link-wrapper span.channel"
# 登录后导航栏"我"的链接(href 含 /user/profile/<user_id>)
USER_PROFILE_NAV_LINK = ".main-container .user .link-wrapper a.link-wrapper"
# 个人主页真实昵称
USER_NICKNAME = ".user-name"
... ...
... ... @@ -39,6 +39,29 @@ metadata:
| `send-code --phone` | 发送手机验证码 |
| `verify-code --code` | 提交验证码完成登录 |
| `delete-cookies` | 退出登录并清除 cookies |
| `add-account --name` | 添加命名账号(自动分配端口) |
| `list-accounts` | 列出所有命名账号及端口 |
| `remove-account --name` | 删除命名账号 |
| `set-default-account --name` | 设置默认账号 |
---
## 账号选择(前置步骤)
> **例外**:用户要求"添加账号 / 列出账号 / 删除账号 / 设置默认账号"时,**跳过此步骤**,直接执行对应管理命令。
其余操作(检查登录、登录、退出登录)先运行:
```bash
python scripts/cli.py list-accounts
```
根据返回的 `count`
- **0 个命名账号**:直接使用默认账号(后续命令不加 `--account`)。
- **1 个命名账号**:告知用户"将对账号 X 执行操作",直接加 `--account <名称>` 执行。
- **多个命名账号**:向用户展示列表,询问操作哪个账号,用 `--account <选择的名称>` 执行后续命令。
账号选定后,本次操作全程固定该账号,**不重复询问**
---
... ... @@ -102,14 +125,18 @@ python scripts/cli.py wait-login
#### 方式 B:手机验证码登录(无界面服务器,分两步)
**执行前必须先向用户索取手机号,不得自行假设或跳过此步。**
**⚠️ 强制要求:必须先向用户确认手机号,即使上下文中已有手机号也不得跳过。**
- 用户可能要登录不同账号,手机号可能已变更。
- **禁止从历史对话、记忆或上下文中自动填入手机号。**
- **每次登录都必须明确向用户询问并得到确认后才能执行 `send-code`。**
**第一步** — 向用户询问手机号,然后发送验证码:
**第一步** — 向用户确认手机号,然后发送验证码:
> 请先问用户:"请提供您的手机号(不含国家码,如 13800138000)",获得回复后再执行以下命令。
> **必须先问用户**:"请提供您要登录的手机号(不含国家码,如 13800138000)"。
> 收到用户明确回复手机号后,才能执行以下命令。**不得跳过此步。**
```bash
python scripts/cli.py send-code --phone <用户提供的手机号>
python scripts/cli.py send-code --phone <用户确认的手机号>
```
- 自动填写手机号、勾选用户协议、点击"获取验证码"。
- Chrome 页面保持打开,等待下一步。
... ... @@ -132,6 +159,41 @@ python scripts/cli.py delete-cookies
python scripts/cli.py --account work delete-cookies # 指定账号
```
## 多账号工作流
每个命名账号拥有独立端口(从 9223 起递增)和独立 Chrome Profile,账号之间完全隔离。
### 添加账号
```bash
python scripts/cli.py add-account --name work --description "工作号"
# 输出: {"success": true, "name": "work", "port": 9223, "profile_dir": "..."}
python scripts/cli.py add-account --name personal
# 输出: {"success": true, "name": "personal", "port": 9224, "profile_dir": "..."}
```
### 使用指定账号执行操作
通过全局 `--account` 参数指定账号,CLI 自动切换到对应端口和 Chrome Profile:
```bash
python scripts/cli.py --account work check-login
python scripts/cli.py --account work get-qrcode
python scripts/cli.py --account personal check-login
python scripts/cli.py check-login # 不指定账号,使用默认端口 9222
```
### 管理账号
```bash
python scripts/cli.py list-accounts # 列出所有账号及端口
python scripts/cli.py set-default-account --name work # 设置默认账号
python scripts/cli.py remove-account --name personal # 删除账号
```
---
## 失败处理
- **Chrome 未找到**:提示用户安装 Google Chrome 或设置 `CHROME_BIN` 环境变量。
... ...
... ... @@ -46,6 +46,23 @@ metadata:
---
## 账号选择(前置步骤)
每次 skill 触发后,先运行:
```bash
python scripts/cli.py list-accounts
```
根据返回的 `count`
- **0 个命名账号**:直接使用默认账号(后续命令不加 `--account`)。
- **1 个命名账号**:告知用户"将使用账号 X",直接加 `--account <名称>` 执行。
- **多个命名账号**:向用户展示列表,询问选择哪个,再用 `--account <选择的名称>` 执行所有后续命令。
账号选定后,本次操作全程固定该账号,**不重复询问**
---
## 输入判断
按优先级判断:
... ...
... ... @@ -40,6 +40,23 @@ metadata:
---
## 账号选择(前置步骤)
每次 skill 触发后,先运行:
```bash
python scripts/cli.py list-accounts
```
根据返回的 `count`
- **0 个命名账号**:直接使用默认账号(后续命令不加 `--account`)。
- **1 个命名账号**:告知用户"将使用账号 X",直接加 `--account <名称>` 执行。
- **多个命名账号**:向用户展示列表,询问选择哪个,再用 `--account <选择的名称>` 执行所有后续命令。
账号选定后,本次操作全程固定该账号,**不重复询问**
---
## 输入判断
按优先级判断:
... ...
... ... @@ -40,6 +40,23 @@ metadata:
---
## 账号选择(前置步骤)
每次 skill 触发后,先运行:
```bash
python scripts/cli.py list-accounts
```
根据返回的 `count`
- **0 个命名账号**:直接使用默认账号(后续命令不加 `--account`)。
- **1 个命名账号**:告知用户"将使用账号 X",直接加 `--account <名称>` 执行。
- **多个命名账号**:向用户展示列表,询问选择哪个,再用 `--account <选择的名称>` 执行所有后续命令。
账号选定后,本次操作全程固定该账号,**不重复询问**
---
## 输入判断
按优先级判断:
... ...
... ... @@ -45,6 +45,23 @@ metadata:
---
## 账号选择(前置步骤)
每次 skill 触发后,先运行:
```bash
python scripts/cli.py list-accounts
```
根据返回的 `count`
- **0 个命名账号**:直接使用默认账号(后续命令不加 `--account`)。
- **1 个命名账号**:告知用户"将使用账号 X 发布",直接加 `--account <名称>` 执行。
- **多个命名账号**:向用户展示列表,**明确询问发布到哪个账号**,用 `--account <选择的名称>` 执行所有后续命令。
账号选定后,本次发布全程固定该账号,**不重复询问**
---
## 输入判断
按优先级判断:
... ...
"""account_manager 单元测试。"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
# 把 scripts/ 加入路径,使 account_manager 可导入
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
import account_manager
@pytest.fixture(autouse=True)
def tmp_config(tmp_path, monkeypatch):
"""将配置目录重定向到临时目录。"""
monkeypatch.setattr(account_manager, "_CONFIG_DIR", tmp_path / ".xhs")
monkeypatch.setattr(
account_manager, "_ACCOUNTS_FILE", tmp_path / ".xhs" / "accounts.json"
)
def test_add_account_assigns_port():
"""首个命名账号应分配端口 9223。"""
account_manager.add_account("work", "工作号")
port = account_manager.get_account_port("work")
assert port == 9223
def test_second_account_gets_next_port():
"""第二个账号应分配端口 9224。"""
account_manager.add_account("work")
account_manager.add_account("personal")
assert account_manager.get_account_port("personal") == 9224
def test_get_profile_dir_public():
"""get_profile_dir 应返回正确路径。"""
account_manager.add_account("work")
profile = account_manager.get_profile_dir("work")
assert "work" in profile
assert "chrome-profile" in profile
def test_get_account_port_unknown_raises():
"""不存在的账号应抛出 ValueError。"""
with pytest.raises(ValueError, match="不存在"):
account_manager.get_account_port("ghost")
def test_list_accounts_includes_port():
"""list_accounts 返回结果中应包含 port 字段。"""
account_manager.add_account("work", "工作")
accounts = account_manager.list_accounts()
assert len(accounts) == 1
assert accounts[0]["port"] == 9223
... ...