client-commons/skill-template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: 沉淀 account-manager 多账号与 Profile 工业标准
All checks were successful
技能自动化发布 / release (push) Successful in 7s
Details

Browse Source
This commit is contained in:
chendelian
2026-07-24 16:50:57 +08:00
parent 0ea924958f
commit c15e23e0d4
20 changed files with 604 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
examples/real_browser_rpa/scripts/service/account_client.py
View File

@@ -159,6 +159,8 @@ def _normalize_error_code(raw: str) -> str:
def _validate_pick_payload(payload: dict[str, Any]) -> dict[str, Any]:
if not isinstance(payload, dict):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "ensure-web 返回格式异常。")
if payload.get("success") is False:
err = payload.get("error") if isinstance(payload.get("error"), dict) else {}
code = _normalize_error_code(str(err.get("code") or ""))
@@ -174,10 +176,10 @@ def _validate_pick_payload(payload: dict[str, Any]) -> dict[str, Any]:
raise AccountManagerError("NO_ACCOUNT", message or "没有可用账号。")
raise AccountManagerError(code or "ENSURE_WEB_FAILED", message or "ensure-web 失败。")
if not isinstance(payload, dict):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "ensure-web 返回格式异常。")
if not payload.get("profile_dir"):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "ensure-web 返回的账号缺少 profile_dir。")
if not payload.get("lease_token"):
raise AccountManagerError("LEASE_REQUIRED", "account-manager 未返回 lease_token，禁止打开浏览器。")
return payload
@@ -226,7 +228,44 @@ def _ensure_web_with_lease(
return _validate_pick_payload(payload)
def _pick_by_id(platform: str, account_id: int) -> Dict[str, Any]:
def _pick_existing_with_lease(
platform: str,
*,
login_id: Optional[str] = None,
label: Optional[str] = None,
) -> Dict[str, Any]:
argv = [
"account",
"pick-web",
"--platform",
platform,
"--lease",
"--holder",
LEASE_HOLDER,
"--ttl-sec",
LEASE_TTL_SEC,
"--purpose",
"rpa",
]
if login_id:
argv.extend(["--login-id", login_id])
elif label:
argv.extend(["--label", label])
else:
raise AccountManagerError("AMBIGUOUS_ACCOUNT", "指定账号缺少可用于带租约选择的登录标识或名称。")
proc = _run_argv(argv)
out = proc.stdout or ""
if proc.returncode != 0 and not out.strip():
raise AccountManagerError(
"PICK_WEB_FAILED",
(proc.stderr or "").strip() or "pick-web 子进程失败",
)
payload = _parse_last_json(out)
return _validate_pick_payload(payload)
def _pick_by_id_with_lease(platform: str, account_id: int) -> Dict[str, Any]:
proc = _run_argv(["account", "get", str(account_id)])
out = proc.stdout or ""
if proc.returncode != 0 or out.startswith("ERROR:"):
@@ -239,6 +278,8 @@ def _pick_by_id(platform: str, account_id: int) -> Dict[str, Any]:
else:
raise AccountManagerError("ACCOUNT_NOT_FOUND", "account get 返回非对象 JSON。")
if data.get("success") is False:
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"指定账号 {account_id} 不存在或获取失败。")
platform_key = str(data.get("platform_key") or "").lower()
if platform_key != platform.lower():
raise AccountManagerError(
@@ -248,9 +289,26 @@ def _pick_by_id(platform: str, account_id: int) -> Dict[str, Any]:
profile_dir = str(data.get("profile_dir") or "").strip()
if not profile_dir:
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"账号 {account_id} 缺少 profile_dir。")
data["profile_dir"] = profile_dir
data["lease_token"] = ""
return data
if str(data.get("status") or "active").lower() != "active":
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"账号 {account_id} 当前不是 active 状态。")
try:
selected = _pick_existing_with_lease(
platform,
login_id=(str(data.get("login_id") or "").strip() or None),
label=(str(data.get("account_label") or "").strip() or None),
)
except AccountManagerError as exc:
if exc.code in ("NO_ACCOUNT", "ACCOUNT_NOT_FOUND"):
raise AccountManagerError("LEASE_CONFLICT", LEASE_BUSY_MESSAGE) from exc
raise
if str(selected.get("id")) != str(account_id):
release_lease(str(selected.get("lease_token") or ""))
raise AccountManagerError(
"AMBIGUOUS_ACCOUNT",
f"无法唯一取得指定账号 {account_id} 的租约，请检查重复的登录标识或账号名称。",
)
return selected
def pick_web_account(
@@ -271,7 +329,7 @@ def pick_web_account(
login_id = aid_raw
aid_raw = ""
else:
return _pick_by_id(platform_key, int(aid_raw))
return _pick_by_id_with_lease(platform_key, int(aid_raw))
lid = (login_id or "").strip() or None
if lid and _is_placeholder_config(lid):

63
examples/real_browser_rpa/tests/test_account_client.py Normal file
View File

@@ -0,0 +1,63 @@
from __future__ import annotations
import json
import subprocess
import unittest
from unittest.mock import patch
from service.account_client import AccountManagerError, pick_web_account
def _completed(payload: dict) -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess(
args=["python", "account-manager"],
returncode=0,
stdout=json.dumps(payload, ensure_ascii=False),
stderr="",
)
class TestExplicitAccountLease(unittest.TestCase):
@patch("service.account_client._run_argv")
def test_explicit_account_id_is_reselected_with_lease(self, run_argv) -> None:
account = {
"id": 42,
"platform_key": "target-platform",
"account_label": "主账号",
"login_id": "user@example.com",
"profile_dir": "D:/data/profile-42",
"status": "active",
}
leased = {**account, "lease_token": "lease-42"}
run_argv.side_effect = [_completed(account), _completed(leased)]
result = pick_web_account("target-platform", account_id="42")
self.assertEqual(result["id"], 42)
self.assertEqual(result["lease_token"], "lease-42")
self.assertEqual(run_argv.call_args_list[0].args[0], ["account", "get", "42"])
pick_args = run_argv.call_args_list[1].args[0]
self.assertIn("pick-web", pick_args)
self.assertIn("--lease", pick_args)
self.assertIn("user@example.com", pick_args)
@patch("service.account_client._run_argv")
def test_missing_lease_token_is_rejected(self, run_argv) -> None:
account = {
"id": 42,
"platform_key": "target-platform",
"account_label": "主账号",
"login_id": "user@example.com",
"profile_dir": "D:/data/profile-42",
"status": "active",
}
run_argv.side_effect = [_completed(account), _completed(account)]
with self.assertRaises(AccountManagerError) as raised:
pick_web_account("target-platform", account_id="42")
self.assertEqual(raised.exception.code, "LEASE_REQUIRED")
if __name__ == "__main__":
unittest.main()

72
examples/simulator_browser_rpa/scripts/service/account_client.py
View File

@@ -159,6 +159,8 @@ def _normalize_error_code(raw: str) -> str:
def _validate_pick_payload(payload: dict[str, Any]) -> dict[str, Any]:
if not isinstance(payload, dict):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "ensure-web 返回格式异常。")
if payload.get("success") is False:
err = payload.get("error") if isinstance(payload.get("error"), dict) else {}
code = _normalize_error_code(str(err.get("code") or ""))
@@ -174,10 +176,10 @@ def _validate_pick_payload(payload: dict[str, Any]) -> dict[str, Any]:
raise AccountManagerError("NO_ACCOUNT", message or "没有可用账号。")
raise AccountManagerError(code or "ENSURE_WEB_FAILED", message or "ensure-web 失败。")
if not isinstance(payload, dict):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "ensure-web 返回格式异常。")
if not payload.get("profile_dir"):
raise AccountManagerError("ACCOUNT_NOT_FOUND", "ensure-web 返回的账号缺少 profile_dir。")
if not payload.get("lease_token"):
raise AccountManagerError("LEASE_REQUIRED", "account-manager 未返回 lease_token，禁止打开浏览器。")
return payload
@@ -226,7 +228,44 @@ def _ensure_web_with_lease(
return _validate_pick_payload(payload)
def _pick_by_id(platform: str, account_id: int) -> Dict[str, Any]:
def _pick_existing_with_lease(
platform: str,
*,
login_id: Optional[str] = None,
label: Optional[str] = None,
) -> Dict[str, Any]:
argv = [
"account",
"pick-web",
"--platform",
platform,
"--lease",
"--holder",
LEASE_HOLDER,
"--ttl-sec",
LEASE_TTL_SEC,
"--purpose",
"rpa",
]
if login_id:
argv.extend(["--login-id", login_id])
elif label:
argv.extend(["--label", label])
else:
raise AccountManagerError("AMBIGUOUS_ACCOUNT", "指定账号缺少可用于带租约选择的登录标识或名称。")
proc = _run_argv(argv)
out = proc.stdout or ""
if proc.returncode != 0 and not out.strip():
raise AccountManagerError(
"PICK_WEB_FAILED",
(proc.stderr or "").strip() or "pick-web 子进程失败",
)
payload = _parse_last_json(out)
return _validate_pick_payload(payload)
def _pick_by_id_with_lease(platform: str, account_id: int) -> Dict[str, Any]:
proc = _run_argv(["account", "get", str(account_id)])
out = proc.stdout or ""
if proc.returncode != 0 or out.startswith("ERROR:"):
@@ -239,6 +278,8 @@ def _pick_by_id(platform: str, account_id: int) -> Dict[str, Any]:
else:
raise AccountManagerError("ACCOUNT_NOT_FOUND", "account get 返回非对象 JSON。")
if data.get("success") is False:
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"指定账号 {account_id} 不存在或获取失败。")
platform_key = str(data.get("platform_key") or "").lower()
if platform_key != platform.lower():
raise AccountManagerError(
@@ -248,9 +289,26 @@ def _pick_by_id(platform: str, account_id: int) -> Dict[str, Any]:
profile_dir = str(data.get("profile_dir") or "").strip()
if not profile_dir:
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"账号 {account_id} 缺少 profile_dir。")
data["profile_dir"] = profile_dir
data["lease_token"] = ""
return data
if str(data.get("status") or "active").lower() != "active":
raise AccountManagerError("ACCOUNT_NOT_FOUND", f"账号 {account_id} 当前不是 active 状态。")
try:
selected = _pick_existing_with_lease(
platform,
login_id=(str(data.get("login_id") or "").strip() or None),
label=(str(data.get("account_label") or "").strip() or None),
)
except AccountManagerError as exc:
if exc.code in ("NO_ACCOUNT", "ACCOUNT_NOT_FOUND"):
raise AccountManagerError("LEASE_CONFLICT", LEASE_BUSY_MESSAGE) from exc
raise
if str(selected.get("id")) != str(account_id):
release_lease(str(selected.get("lease_token") or ""))
raise AccountManagerError(
"AMBIGUOUS_ACCOUNT",
f"无法唯一取得指定账号 {account_id} 的租约，请检查重复的登录标识或账号名称。",
)
return selected
def pick_web_account(
@@ -271,7 +329,7 @@ def pick_web_account(
login_id = aid_raw
aid_raw = ""
else:
return _pick_by_id(platform_key, int(aid_raw))
return _pick_by_id_with_lease(platform_key, int(aid_raw))
lid = (login_id or "").strip() or None
if lid and _is_placeholder_config(lid):

63
examples/simulator_browser_rpa/tests/test_account_client.py Normal file
View File

@@ -0,0 +1,63 @@
from __future__ import annotations
import json
import subprocess
import unittest
from unittest.mock import patch
from service.account_client import AccountManagerError, pick_web_account
def _completed(payload: dict) -> subprocess.CompletedProcess[str]:
return subprocess.CompletedProcess(
args=["python", "account-manager"],
returncode=0,
stdout=json.dumps(payload, ensure_ascii=False),
stderr="",
)
class TestExplicitAccountLease(unittest.TestCase):
@patch("service.account_client._run_argv")
def test_explicit_account_id_is_reselected_with_lease(self, run_argv) -> None:
account = {
"id": 42,
"platform_key": "target-platform",
"account_label": "主账号",
"login_id": "user@example.com",
"profile_dir": "D:/data/profile-42",
"status": "active",
}
leased = {**account, "lease_token": "lease-42"}
run_argv.side_effect = [_completed(account), _completed(leased)]
result = pick_web_account("target-platform", account_id="42")
self.assertEqual(result["id"], 42)
self.assertEqual(result["lease_token"], "lease-42")
self.assertEqual(run_argv.call_args_list[0].args[0], ["account", "get", "42"])
pick_args = run_argv.call_args_list[1].args[0]
self.assertIn("pick-web", pick_args)
self.assertIn("--lease", pick_args)
self.assertIn("user@example.com", pick_args)
@patch("service.account_client._run_argv")
def test_missing_lease_token_is_rejected(self, run_argv) -> None:
account = {
"id": 42,
"platform_key": "target-platform",
"account_label": "主账号",
"login_id": "user@example.com",
"profile_dir": "D:/data/profile-42",
"status": "active",
}
run_argv.side_effect = [_completed(account), _completed(account)]
with self.assertRaises(AccountManagerError) as raised:
pick_web_account("target-platform", account_id="42")
self.assertEqual(raised.exception.code, "LEASE_REQUIRED")
if __name__ == "__main__":
unittest.main()