7.7 KiB
rpa_helpers 库 — 业务 skill 同进程登录
scripts/service/rpa_helpers/ 是 account-manager v2 提供给业务 skill 的 同进程 Python 库。业务代码调用 ensure_logged_in,库按账号记录的 auth_strategy 分发到对应登录流程(底层基于 Playwright)。
与 CLI 的关系
| 路径 | 职责 |
|---|---|
| CLI subprocess | 获取结构化账号 JSON、租约、get-credential --reveal 明文密码;进程隔离、易于跨语言 |
| rpa_helpers | 在已有 BrowserContext/Page 上执行登录编排与人类化输入 |
推荐组合:CLI → pick / lease → get-credential → launch_browser_with_profile + ensure_logged_in。
集成方式
Import 路径
import sys
AM_ROOT = r"D:\OpenClaw\client-commons\account-manager"
sys.path.insert(0, f"{AM_ROOT}/scripts")
from service.rpa_helpers import (
AuthStrategyError,
LoginResult,
close_browser,
ensure_logged_in,
launch_browser_with_profile,
)
务必:sys.path 追加的是 …/account-manager/scripts,因此模块命名空间为 service.rpa_helpers(不要写成 from scripts.service.rpa_helpers …)。
运行时依赖
- Python 3.10+
- 已安装
playwrightPython 包(宿主 runtime 提供) - 本机已安装 Chrome 或 Edge(启动时使用 channel 模式,优先 Chrome,失败回落 Edge——见
service.browser_service.resolve_chromium_channel)
文档不提供 pip install … 指令;版本由匠厂宿主统一管理。
公开 API 一览
下列符号由 service/rpa_helpers/__init__.py 导出(共 19 个名字):
| 分类 | 符号 |
|---|---|
| dispatcher + 策略函数 | ensure_logged_in,以及 ensure_logged_in_password_auto、ensure_logged_in_password_with_captcha、ensure_logged_in_password_plus_2fa、ensure_logged_in_qr_code_manual、ensure_logged_in_per_session_manual、ensure_logged_in_device_bound、ensure_logged_in_client_certificate、ensure_logged_in_api_token、ensure_logged_in_sso_redirect |
| 结果 / 异常 | LoginResult,AuthStrategyError |
| 浏览器 | launch_browser_with_profile,close_browser |
| 人类化输入 | pause,human_type,human_click,human_select,human_scroll_to |
browser 模块
launch_browser_with_profile(profile_dir, *, channel=None, headless=False, extra_args=None, locale="zh-CN") -> tuple[BrowserContext, Page]
- 创建持久化上下文(
user_data_dir=profile_dir),注入可选 stealth 脚本。 channel默认为宿主检测结果(Chrome 优先)。- 必须在
finally中调用close_browser(context),否则 playwright driver 可能泄漏。
close_browser(ctx) -> None
关闭上下文并停止内部的 Playwright session manager。
human 模块(人类化操作)
| 函数 | 行为 |
|---|---|
pause(min_ms=200, max_ms=800) |
随机等待 |
human_type(locator, text, *, per_char_ms_range=(50, 200), pre_pause=True) |
滚动进入视图 → 停顿 → click → 逐字符 type |
human_click(locator) |
滚动 → 停顿 → click |
human_select(locator, value_or_label) |
滚动 → 停顿 → select_option(依次尝试 value / label) |
human_scroll_to(locator) |
平滑滚动到元素 |
locator 均为 Playwright Locator。
auth_flows 模块
ensure_logged_in(page, account, *, credentials=None, **kwargs) -> LoginResult
- 读取
account["auth_strategy"]。为空字符串 → 抛出AuthStrategyError(ERR_AUTH_STRATEGY_NOT_SUPPORTED)。 - 若策略需要密码则要求
credentials["plaintext"];缺失 →ERR_CREDENTIAL_MISSING。 - 分派到对应
ensure_logged_in_*。 - 未知策略 →
AuthStrategyError(ERR_AUTH_STRATEGY_NOT_SUPPORTED)。
附加 kwargs 会透传给具体策略(如 captcha_value、wait_human_sec)。
LoginResult
| 字段 | 类型 | 说明 |
|---|---|---|
ok |
bool |
流程是否达到判定条件 |
needs_human |
bool |
是否需要人工协助(扫码 / OTP / SSO 等) |
message |
str |
描述信息(便于日志) |
AuthStrategyError
属性:code: str,message: str。错误码与 CLI / ERRORS.md 对齐。
九个 ensure_logged_in_* 入口
业务代码通常只调用 ensure_logged_in。如需针对性 reuse,可直接 import auth_flows 中的同名函数;签名见源码 scripts/service/rpa_helpers/auth_flows.py。
已登录检测(_is_already_logged_in)
核心逻辑:
- 读取当前
page.url;空白页 → False。 - 仅在账号
extra_json(可为 dict 或 JSON 字符串)中出现以下 hook 之一才可能返回 True:logged_in_url_pattern:正则,匹配当前 URL。logged_in_selector:CSS/XPath selector,locator.first.is_visible(timeout=500)。
- 若无任一 hook → 默认 False → 除非已有持久化 session,否则会完整跑登录流程。
接入建议
- 首次上线:
auth_strategy=qr_code_manual+ 无 hook → 用户扫码登录。 - 登录成功后记录典型 URL 片段或 DOM selector。
- 写入账号
extra_json,例如:
{
"logged_in_url_pattern": r"/user/",
"logged_in_selector": "div.avatar-wrapper"
}
- 下次启动 →
_is_already_logged_in命中 →needs_human可能降低。
per_session_manual 的差异
该策略 开场不会调用 _is_already_logged_in,总是导航到业务 URL 并等待人工完成登录(适合强制会话)。
设备绑定策略(device_bound)
如账号设置了 device_fingerprint,库会使用 _current_machine_fingerprint()(hostname + MAC 派生)比对;不一致抛出 ERR_DEVICE_FINGERPRINT_MISMATCH。
客户端证书策略(client_certificate)
直接抛出 ERR_AUTH_STRATEGY_NOT_AUTOMATABLE;请在宿主机手动装载证书后使用其它路径衔接。
完整示例(CLI + rpa_helpers)
import json
import subprocess
import sys
AM = r"D:\OpenClaw\client-commons\account-manager"
PY = sys.executable
MAIN = f"{AM}/scripts/main.py"
sys.path.insert(0, f"{AM}/scripts")
from service.rpa_helpers import launch_browser_with_profile, close_browser, ensure_logged_in
def am_json(argv: list[str]) -> dict:
proc = subprocess.run([PY, MAIN, *argv], capture_output=True, text=True, check=False)
line = (proc.stdout or "").splitlines()[0]
return json.loads(line)
acc = am_json(
["account", "pick-web", "--platform", "icbc_sim", "--lease", "--holder", "monitor-rates"]
)
cred = am_json(
[
"account",
"get-credential",
str(acc["id"]),
"--reveal",
"--lease-token",
acc["lease_token"],
]
)
plaintext = cred.get("plaintext")
if plaintext in (None, ""):
raise RuntimeError("no plaintext returned; check secret_storage / lease")
ctx, page = launch_browser_with_profile(acc["profile_dir"])
try:
result = ensure_logged_in(page, acc, credentials={"plaintext": plaintext})
if not result.ok:
raise RuntimeError(result.message)
# …业务代码…
finally:
close_browser(ctx)
subprocess.run([PY, MAIN, "lease", "release", acc["lease_token"]], check=False)
测试建议
- account-manager 仓库内置 107 个 pytest(截至 v2.0.0 文档修订时),覆盖 CLI / schema / rpa_helpers mock 路径。
- 业务 skill:
- 单元测试:patch
ensure_logged_in→ 返回LoginResult(ok=True, …)。 - 端到端:在 Windows + 真浏览器环境标记
@pytest.mark.integration,禁止在 CI 泄露明文密码。
- 单元测试:patch
更多 CLI 细节参见 CLI.md、INTEGRATION.md。