Compare commits

...

15 Commits

Author SHA1 Message Date
7d0ec94210 feat: take over host delete so profile dirs are removed
All checks were successful
技能自动化发布 / release (push) Successful in 7s
2026-10-01 09:56:55 +08:00
1ba7a0f4a1 feat: open browser with stable CDP port for scrape attach
All checks were successful
技能自动化发布 / release (push) Successful in 11s
2026-09-16 15:40:50 +08:00
71b20e6e4a fix: use --restore-last-session instead of unsigned Preferences restore_on_startup
All checks were successful
技能自动化发布 / release (push) Successful in 5s
2026-09-16 14:38:03 +08:00
7709292f1c fix: recognize modern Chrome Sessions so open restores login
All checks were successful
技能自动化发布 / release (push) Successful in 10s
2026-09-16 14:13:38 +08:00
b93b2ea36e fix: reuse browser, harden session restore without URL clash (v2.0.27)
Some checks failed
技能自动化发布 / release (push) Has been cancelled
2026-09-15 11:46:08 +08:00
44eafff757 fix: open-browser enable session restore (v2.0.26)
All checks were successful
技能自动化发布 / release (push) Successful in 8s
2026-09-15 11:00:16 +08:00
3ce0cc40a1 Release v2.0.25 force-release lease row action
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-09-12 16:18:36 +08:00
3f1cde1f99 fix: validate platform keys at the gate and stop auto-register
All checks were successful
技能自动化发布 / release (push) Successful in 6s
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-22 15:40:30 +08:00
9a20e7fc69 feat: simplify account create/edit and add row copy
All checks were successful
技能自动化发布 / release (push) Successful in 8s
Data-management add/edit now use short forms with enums; copy clones a platform account into a new empty profile.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-22 14:48:46 +08:00
097dde0ebb fix: open account browser maximized without Chrome first-run
All checks were successful
技能自动化发布 / release (push) Successful in 5s
Human open now uses start-maximized and no-first-run flags, and profile shortcuts share the same args plus the platform URL.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-22 14:11:05 +08:00
718aeefd44 fix: unify row open labels and silent success
All checks were successful
技能自动化发布 / release (push) Successful in 5s
Rename 打开文件位置 to 打开目录, and print empty SAS message so the host skips 执行成功 after opening Explorer or the browser.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-22 13:15:26 +08:00
123195fcd9 fix: align add-web form fields to column names for lookup
All checks were successful
技能自动化发布 / release (push) Successful in 7s
2026-07-31 15:14:31 +08:00
ba991a2462 feat: bind accounts create to add-web via replaces
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-07-31 14:41:36 +08:00
e9455cbb86 feat: data-management lookup for platform/account fields
All checks were successful
技能自动化发布 / release (push) Successful in 7s
2026-07-31 13:37:07 +08:00
87877010ac fix: account open launches Chrome/Edge directly like reveal-profile
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-07-31 11:58:18 +08:00
38 changed files with 2620 additions and 414 deletions

37
CHANGELOG.md Normal file
View File

@@ -0,0 +1,37 @@
# Changelog
本文件对应技能市场 **「更新日志」** Tab。
发布 CI 会按当前 tag 版本提取匹配的 `##` 小节。
## 2.0.31
- 数据管理「删除」改为技能接管(`replaces: delete`):删账号行和凭证的同时删除硬盘 Profile 目录,避免再建同名账号带回旧登录态
## 2.0.30
- 「打开浏览器」为本账号 Profile 附加稳定本机调试口(与科力普等采集技能约定同一端口算法),便于采集挂接已打开的原生窗口;仍使用 `--restore-last-session` 恢复登录态
## 2.0.29
- 「打开浏览器」有可恢复会话时改用 `--restore-last-session` 强制恢复标签与 Session Cookie;不再裸写受 Secure Preferences HMAC 保护的 `restore_on_startup`(该写法会被 Chrome 否掉并落到新标签页)
## 2.0.28
- 「打开浏览器」识别现代 Chrome `Default/Sessions/Session_*`(及 `Tabs_*`),有可恢复会话时不再强制带平台首页 URL,避免冲掉 Session Cookie 登录态
## 2.0.27
- 「打开浏览器」:Profile 已占用时复用窗口;加固 Preferences/Local State「继续上次会话」;有历史会话时冷启动不带 URL,避免冲掉登录态
## 2.0.26
- 「打开浏览器」启动前写入 Profile「继续上次会话」,减轻独立 Profile 下 Session Cookie 关窗后丢失(科力普等站点更明显)
## 2.0.25
- 「账号租约」表新增行右键「强制释放租约」:仅状态为 active 时显示,确认后立刻释放,无需租约令牌
- 同步提供 CLI:`lease force-release --id <租约编号>`
## 2.0.24
- (历史版本,见 git tag)

View File

@@ -1,7 +1,7 @@
---
name: 账号与凭据管理
description: "通用账号 & 凭据 & 登录流程管理器。管理网页/RPA 账号、API Key/Token;支持 9 种登录策略(账密自动、扫码、2FA、SSO 等);提供加密本地存储(Fernet);提供 rpa_helpers 库供业务 skill 同进程调用 ensure_logged_in 完成多平台自动登录。涵盖物流、LLM、内容、开发、API 五类平台。"
version: 2.0.16
version: 2.0.31
author: 深圳匠厂科技有限公司
metadata:
openclaw:
@@ -28,20 +28,24 @@ allowed-tools:
## 何时使用本技能
1. **安装/升级后初始化库与数据管理元数据**:`init-db`(宿主会自动调用;也可手工执行)
2. **登记网页/RPA 账号**:`account add-web` — 平台 + 登录标识 + 自动生成 profile_dir;可选 `--auth-strategy` / `--session-persistent` / `--device-fingerprint`。
3. **登记凭据(API Key / Token / 加密密码)**:`account add-secret` — `none` / `env` / `windows_credential` / `local_encrypted`。
4. **查看/列出账号**:`account get/list` — JSON 输出;list 可按 `--login-id` / `--label` 筛选。
5. **用人话定位账号**:`account resolve --login-id <手机号/用户名>` 或 `--label` / `--query`(不必先查数字 id)。
6. **为 RPA 挑选账号**:`account pick-web` — 按平台/环境/角色,也可 `--login-id` / `--label`;业务首启推荐 `account ensure-web`(或 `pick-web --ensure`):无账号则自动 `add-web` 再返回。
7. **获取 API Key(凭据表)**:`credential pick` — 默认隐藏 secret,`--reveal` 时才返回。
8. **浏览器打开查看**:`account open <id>`(或兼容别名 `open <id>`)— 拉起 Profile 后立即返回,不做站点侧登录判定。数据管理「账号信息」行内 **打开浏览器** / **打开文件位置** 均为 `interaction: direct` + `sync`(见 `assets/actions.json`)。
9. **打开 Profile 目录**:`account reveal-profile <id>` — 在系统文件管理器中打开该账号 `profile_dir`。
10. **管理 lease**:`lease release/list/cleanup` — 防止 RPA 并发。
11. **初始化加密**:`account init-master-key` — 生成 Fernet master.key。
12. **加密备份与迁移**:`account export-credentials --plaintext` / `account import-credentials` — 导出明文 JSON、在新环境重新加密写入。
13. **按 lease 取明文密码**:`account get-credential <id> --reveal --lease-token <token>` — 结合 pick-web 返回的 lease 解密 `local_encrypted` 等路径。
14. **业务 skill 同进程登录**:`from service.rpa_helpers import ensure_logged_in` — 详见 [`references/RPA_HELPERS.md`](references/RPA_HELPERS.md)。
15. **有则用、无则建(网页账号)**:`account ensure-web --platform <p>` — 登记账号 ≠ 已登录;登录仍由业务技能 / `ensure_logged_in` 处理。
2. **登记平台**:`platform ensure --key <小写键> --display-name <展示名>`。键与展示名必须分开;`add-web` / `ensure-web` / `pick-web` 只引用已注册平台,**不会**根据 `--platform-display-name` 自动建平台。数据管理「平台信息 → 新增」走本命令。
3. **登记网页/RPA 账号**:`account add-web` — 平台 + 账号名称 + 可选登录账号即可;环境/角色/登录策略跟平台默认。数据管理「新增」已绑定该能力(`replaces: create`),勿用宿主直写库新建账号。
4. **复制账号**:`account copy <id>` — 同平台新空 Profile,名称自动加「副本」,登录账号留空。数据管理行内「复制」一点即走。
5. **编辑账号**:`account update --id <id>` — 只改名称、登录账号、标识类型、入口链接、状态。数据管理「编辑」已绑定(`replaces: update`),不再摊开整张表。
6. **删除账号**:`delete id <id>` — 删除账号行、关联凭证,以及硬盘上的 `profile_dir`。数据管理「删除」已绑定(`replaces: delete`),勿用宿主直写库,否则会留下旧 Profile / 登录态。
7. **登记凭据(API Key / Token / 加密密码)**:`account add-secret` — `none` / `env` / `windows_credential` / `local_encrypted`。
8. **查看/列出账号**:`account get/list` — JSON 输出;list 可按 `--login-id` / `--label` 筛选。
9. **用人话定位账号**:`account resolve --login-id <手机号/用户名>` 或 `--label` / `--query`(不必先查数字 id)。
10. **为 RPA 挑选账号**:`account pick-web` — 按平台/环境/角色,也可 `--login-id` / `--label`;业务首启推荐 `account ensure-web`(或 `pick-web --ensure`):无账号则自动 `add-web` 再返回。
11. **获取 API Key(凭据表)**:`credential pick` — 默认隐藏 secret,`--reveal` 时才返回。
12. **打开浏览器**:`account open <id>`(或兼容别名 `open <id>`)— **本进程**直接拉起系统 Chrome/Edge(独立 `--user-data-dir`),最大化并跳过 Chrome 首次欢迎页,直接打开平台链接;与「打开目录」同为一点即走,不做登录判定。数据管理「打开目录 / 打开浏览器」成功不弹提示。
13. **打开目录**:`account reveal-profile <id>` — 在系统文件管理器中打开该账号 `profile_dir`。
14. **管理 lease**:`lease release/force-release/list/cleanup` — 防止 RPA 并发;数据管理「账号租约」行右键可强制释放。
15. **初始化加密**:`account init-master-key` — 生成 Fernet master.key。
16. **加密备份与迁移**:`account export-credentials --plaintext` / `account import-credentials` — 导出明文 JSON、在新环境重新加密写入。
17. **按 lease 取明文密码**:`account get-credential <id> --reveal --lease-token <token>` — 结合 pick-web 返回的 lease 解密 `local_encrypted` 等路径。
18. **业务 skill 同进程登录**:`from service.rpa_helpers import ensure_logged_in` — 详见 [`references/RPA_HELPERS.md`](references/RPA_HELPERS.md)。
19. **有则用、无则建(网页账号)**:`account ensure-web --platform <p>` — 登记账号 ≠ 已登录;登录仍由业务技能 / `ensure_logged_in` 处理。
## CLI 调用

View File

@@ -1,6 +1,6 @@
# assets
- `actions.json`:匠厂宿主 Skill Action 清单(数据管理「账号信息」行内:打开文件位置 / 打开浏览器)。
- `actions.json`:匠厂宿主 Skill Action 清单(数据管理「账号信息」:新增/编辑/删除接管,行内打开目录 / 打开浏览器 / 复制)。
- `examples/`:CLI 成功输出形状示例(虚构路径与数据)。
- `schemas/`:`get` / `list-json` / `pick-web` 单条对象的轻量 JSON Schema(`additionalProperties: true` 允许扩展字段)。

View File

@@ -2,9 +2,231 @@
"schemaVersion": 1,
"skill": "account-manager",
"actions": [
{
"id": "platform-ensure",
"label": "新增平台",
"description": "登记平台:小写英文键 + 展示名必须分开。不会把中文名当成键。",
"replaces": "create",
"placements": [],
"executionProfile": "sync",
"interaction": "form",
"bind": {
"tables": ["platforms"]
},
"inputSchema": {
"type": "object",
"required": ["platform_key", "display_name"],
"properties": {
"platform_key": {
"type": "string",
"title": "平台键",
"description": "小写英文/数字/下划线或连字符,例如 douyin、deepseek。不要填中文。"
},
"display_name": {
"type": "string",
"title": "展示名称",
"description": "给人看的名称,可以是中文,例如 抖音。"
},
"default_url": {
"type": "string",
"title": "默认链接"
}
}
},
"steps": [{ "when": "afterSuccess", "do": "reloadTable" }],
"entrypoint": {
"type": "cli",
"command": "platform",
"args": [
"ensure",
"--key", "{{platform_key}}",
"--display-name", "{{display_name}}",
"--url", "{{default_url}}"
]
}
},
{
"id": "account-add-web",
"label": "新增账号",
"description": "登记网页/RPA 账号并自动创建浏览器 Profile 目录(走 account add-web,非宿主直写库)。",
"replaces": "create",
"placements": [],
"executionProfile": "sync",
"interaction": "form",
"bind": {
"tables": ["accounts"]
},
"inputSchema": {
"type": "object",
"required": ["platform_key", "account_label"],
"properties": {
"platform_key": {
"type": "string",
"title": "平台",
"display_type": "lookup",
"lookup": {
"ref_table": "platforms",
"value_column": "platform_key",
"label_column": "display_name",
"filter_column": "enabled",
"filter_value": 1
}
},
"account_label": {
"type": "string",
"title": "账号名称"
},
"login_id": {
"type": "string",
"title": "登录账号"
}
}
},
"steps": [{ "when": "afterSuccess", "do": "reloadTable" }],
"entrypoint": {
"type": "cli",
"command": "account",
"args": [
"add-web",
"--platform", "{{platform_key}}",
"--label", "{{account_label}}",
"--login-id", "{{login_id}}"
]
}
},
{
"id": "account-update",
"label": "编辑账号",
"description": "只改名称、登录账号、标识类型、入口链接和启用状态。",
"replaces": "update",
"placements": [],
"executionProfile": "sync",
"interaction": "form",
"bind": {
"tables": ["accounts"],
"inputMapping": {
"accountId": "$row.$pk",
"account_label": "$row.account_label",
"login_id": "$row.login_id",
"login_id_type": "$row.login_id_type",
"url": "$row.url",
"status": "$row.status"
}
},
"inputSchema": {
"type": "object",
"required": ["accountId", "account_label", "status"],
"properties": {
"accountId": {
"type": "number",
"title": "编号",
"readOnly": true
},
"account_label": {
"type": "string",
"title": "账号名称"
},
"login_id": {
"type": "string",
"title": "登录账号"
},
"login_id_type": {
"type": "string",
"title": "登录标识类型",
"enum": ["unknown", "phone", "email", "username", "customer_code", "api_account"]
},
"url": {
"type": "string",
"title": "入口链接",
"display_type": "url"
},
"status": {
"type": "string",
"title": "状态",
"enum": ["active", "disabled", "needs_review"]
}
}
},
"steps": [{ "when": "afterSuccess", "do": "reloadTable" }],
"entrypoint": {
"type": "cli",
"command": "account",
"args": [
"update",
"--id", "{{accountId}}",
"--label", "{{account_label}}",
"--login-id", "{{login_id}}",
"--login-id-type", "{{login_id_type}}",
"--url", "{{url}}",
"--status", "{{status}}"
]
}
},
{
"id": "account-delete",
"label": "删除账号",
"description": "删除账号记录、关联凭证,以及硬盘上的浏览器 Profile 目录。再建同名账号会得到空 Profile,不会带回旧登录态。",
"replaces": "delete",
"placements": [],
"executionProfile": "sync",
"interaction": "confirm",
"confirmation": {
"message": "将删除这条账号、关联凭证,以及硬盘上的浏览器 Profile 目录。登录态不可恢复。是否继续?"
},
"bind": {
"tables": ["accounts"],
"inputMapping": { "accountId": "$row.$pk" }
},
"inputSchema": {
"type": "object",
"required": ["accountId"],
"properties": {
"accountId": {
"type": "number",
"title": "账号 ID",
"readOnly": true
}
}
},
"steps": [{ "when": "afterSuccess", "do": "reloadTable" }],
"entrypoint": {
"type": "cli",
"command": "delete",
"args": ["id", "{{accountId}}"]
}
},
{
"id": "account-copy",
"label": "复制",
"description": "复制为同平台新账号:新空 Profile,登录账号留空,名称自动加「副本」。",
"placements": ["row"],
"executionProfile": "sync",
"interaction": "direct",
"bind": {
"tables": ["accounts"],
"inputMapping": { "accountId": "$row.$pk" }
},
"inputSchema": {
"type": "object",
"required": ["accountId"],
"properties": {
"accountId": {
"type": "number",
"title": "账号 ID",
"readOnly": true
}
}
},
"steps": [{ "when": "afterSuccess", "do": "reloadTable" }],
"entrypoint": {
"type": "cli",
"command": "account",
"args": ["copy", "{{accountId}}"]
}
},
{
"id": "reveal-profile",
"label": "打开文件位置",
"label": "打开目录",
"description": "在系统文件管理器中打开该账号的浏览器 Profile 目录。",
"placements": ["row"],
"executionProfile": "sync",
@@ -34,7 +256,7 @@
{
"id": "open-browser",
"label": "打开浏览器",
"description": "用该账号持久化 Profile 打开 Chrome/Edge,仅供肉眼核对(不做登录判定)。一点即开,不等待关闭。",
"description": "用该账号持久化 Profile 直接打开系统 Chrome/Edge,仅供肉眼核对(不做登录判定)。与打开目录同为一点即开。",
"placements": ["row"],
"executionProfile": "sync",
"interaction": "direct",
@@ -59,6 +281,44 @@
"command": "account",
"args": ["open", "{{accountId}}"]
}
},
{
"id": "lease-force-release",
"label": "强制释放租约",
"description": "强制释放这条活跃租约,让账号立刻可被其它任务占用。仅状态为 active 时可用。",
"placements": ["row"],
"executionProfile": "sync",
"interaction": "confirm",
"confirmation": {
"message": "强制释放后,其它任务可立刻占用该账号。若原任务仍在跑,可能冲突。是否继续?"
},
"bind": {
"tables": ["account_leases"],
"inputMapping": {
"leaseId": "$row.$pk"
},
"visibleWhen": {
"status": "active"
}
},
"inputSchema": {
"type": "object",
"required": ["leaseId"],
"properties": {
"leaseId": {
"type": "number",
"title": "租约编号",
"description": "当前行编号,一般无需修改",
"readOnly": true
}
}
},
"steps": [{ "when": "afterSuccess", "do": "reloadTable" }],
"entrypoint": {
"type": "cli",
"command": "lease",
"args": ["force-release", "--id", "{{leaseId}}"]
}
}
]
}

View File

@@ -29,6 +29,7 @@
|---|---|
| `platform list [--domain logistics\|llm\|content]` | 列出平台注册(`--domain` 仅接受这三种;`dev`/`sim` 等平台请省略 `--domain` 查全量或在 JSON 里筛选) |
| `platform get <platform>` | 获取数据库中的平台详情 |
| `platform ensure --key <key> --display-name <name> [--url …]` | **登记/更新平台**。`--key` 必须已是小写英文键(如 `douyin`);`--display-name` **必填**(可中文)。不会从 `add-web` / `pick-web` 自动建平台。 |
### B. 加密初始化(v2)
@@ -43,7 +44,9 @@
| CLI | 说明 |
|---|---|
| `account add-web --platform <p> [--login-id <id>] [--login-id-type <type>] [...]` | 创建网页/RPA 账号;可选 `--auth-strategy` / `--session-persistent 0\|1` / `--device-fingerprint <fp>` |
| `account add-web --platform <p> [--login-id <id>] [--login-id-type <type>] [...]` | 创建网页/RPA 账号;省略 `--login-id-type` 时按登录账号推断;可选 `--auth-strategy` / `--session-persistent 0\|1` / `--device-fingerprint <fp>` |
| `account copy <id>` | 复制为同平台新账号(新空 Profile,名称加「副本」,不拷登录账号/凭证/设备指纹) |
| `account update --id <id> [--label] [--login-id] [--login-id-type] [--url] [--status]` | 更新数据管理编辑表单上的短字段 |
| `account add-secret --platform <p> --credential-type <t> --secret-storage env\|windows_credential\|none\|local_encrypted` | 创建凭据;`local_encrypted` **必须** `--secret-stdin` |
| `account get <id> [--with-credentials]` | 输出账号 JSON(若带 `--with-credentials` 则附加关联凭据摘要) |
| `account resolve [--id <id>] [--platform <p>] [--login-id <手机号/用户名>] [--label <备注>] [--query <…>]` | **用人话定位一条账号**(成功输出同 `get`);多条匹配返回 `ERROR:AMBIGUOUS_ACCOUNT` |
@@ -54,7 +57,7 @@
| `account mark-session <id> --session-status <status>` | 标记会话状态 |
| `account mark-error <id> --code <code> --message <msg>` | 标记错误 |
| `account mark-used <id>` | 标记已使用 |
| `account open <id>` | 打开持久化浏览器 profile(仅查看;**拉起后立即返回**,不等待关闭) |
| `account open <id>` | 本进程直接打开系统 Chrome/Edge(独立 profile;最大化;跳过首次欢迎页;打开平台 URL) |
| `account reveal-profile <id>` | 在系统文件管理器中打开该账号 `profile_dir` |
### D. 加密备份与迁移(v2)
@@ -73,10 +76,11 @@
| `credential pick --platform <p> [--type <type>] [--environment <env>] [--role <role>] [--reveal]` | v1 兼容:挑选凭据 |
| `credential resolve --id <credential_id> [--reveal]` | v1 兼容:按凭据 id 解析 |
| `lease release <lease_token>` | 释放租约 |
| `lease force-release --id <租约编号>` | 按行编号强制释放活跃租约(数据管理右键) |
| `lease list [--active]` | 列出租约 |
| `lease cleanup` | 清理过期租约 |
| `account open <id>` / `open <id>` | 浏览器查看 |
| `account reveal-profile <id>` | 打开 Profile 目录 |
| `account open <id>` / `open <id>` | 打开浏览器 |
| `account reveal-profile <id>` | 打开目录 |
### 兼容入口(旧版别名)
@@ -87,7 +91,7 @@
| `pick-web <platform>` | `account pick-web --platform <platform>`(无 lease 参数) |
| `list [platform\|all]` | 人类可读列表 |
| `add <platform> <phone>` | 映射到 `account add-web`(`login_id_type=phone`),不传 v2 扩展参数 |
| `delete id\|platform <...>` | 删除账号 |
| `delete id\|platform <...>` | 删除账号(含硬盘 `profile_dir`;数据管理「删除」走 `delete id`) |
---
@@ -109,6 +113,19 @@ python main.py account add-web \
---
## account copy / update(数据管理)
```bash
python main.py account copy 3
python main.py account update --id 4 --label "抖音 备用" --login-id 13900001111 --status active
```
- `copy`:复制平台、环境、角色、入口链接、登录策略;**新建空 Profile**;登录账号留空;名称自动「副本」/「副本 2」。不复制凭证、设备指纹、会话状态。
- `update`:只改名称、登录账号、标识类型、入口链接、状态。空入口链接回退为平台默认 URL。
- 数据管理「新增」只填平台 / 账号名称 / 登录账号;「编辑」走 `replaces: update` 短表单;「删除」走 `replaces: delete`(`delete id <id>`,含硬盘 Profile);行内「复制」为 `interaction: direct`。
---
## account add-secret(`local_encrypted` 示例)
```bash
@@ -208,7 +225,7 @@ ACCOUNT_MANAGER_MASTER_KEY="<base64-44-chars>" python main.py account init-maste
## 通用说明
- **平台参数**:可为 `scripts/util/platforms.py` 中的英文键、`display_name`、任一 alias(不区分大小写)。详见 [PLATFORMS.md](PLATFORMS.md)。
- **平台参数**:已注册的英文键、或能唯一对应到一个键的 `display_name`/alias(不区分大小写)。冲突别名(如「公众号」同时对应 `wechat` / `wechat_mp`)会返回 `AMBIGUOUS_PLATFORM`,请改传明确键。未注册的键**不会**自动插入 `platforms` 表。详见 [PLATFORMS.md](PLATFORMS.md)。
- **数据与路径**:库文件、profile、`master.key` 布局见 [RUNTIME.md](RUNTIME.md)。
- **机器可读集成**:见 [INTEGRATION.md](INTEGRATION.md)。
- **错误码**:见 [ERRORS.md](ERRORS.md)。

View File

@@ -46,7 +46,13 @@
| 前缀 | 含义 |
|---|---|
| `ERROR:INVALID_PLATFORM` | 平台名无法识别 |
| `ERROR:INVALID_PLATFORM` | 平台名为空,或中文/人话名称无法对应到已注册平台 |
| `ERROR:INVALID_PLATFORM_KEY` | 键格式非法(大写、空格、中文当键等);展示名不能当键 |
| `ERROR:PLATFORM_NOT_REGISTERED` | 键格式合法但尚未 `platform ensure`;**已拒绝写入**,不会自动建平台 |
| `ERROR:AMBIGUOUS_PLATFORM` | 展示名/别名对应多个键(如「公众号」→ `wechat` 与 `wechat_mp`),请改传明确键 |
| `ERROR:PLATFORM_ALIAS_CONFLICT` | `platform ensure` 的展示名/别名已被其它平台占用 |
| `ERROR:INVALID_PLATFORM_DISPLAY_NAME` | `platform ensure` 未提供 `--display-name` |
| `ERROR:PLATFORM_DISABLED` | 平台已禁用 |
| `ERROR:NO_ACCOUNTS_FOUND` | 列表无结果 |
| `ERROR:ACCOUNT_NOT_FOUND` | 指定 id 不存在 |
| `ERROR:ACCOUNT_DISABLED` | 账号状态为 disabled |
@@ -61,7 +67,7 @@
| 前缀 | 含义 |
|---|---|
| `ERROR:LEASE_CONFLICT` | 同账号已有 active 未过期 lease |
| `ERROR:LEASE_NOT_FOUND` | 未找到活跃租约 |
| `ERROR:LEASE_NOT_FOUND` | 未找到活跃租约(含 force-release 时行不存在或已非 active) |
| `ERROR:LEASE_EXPIRED` | 租约已过期 |
---

View File

@@ -4,6 +4,8 @@
业务 skill **不能自己散落存储账号、密码、API Key、Token**。应通过 account-manager **CLI**(进程隔离)或 **CLI + `service.rpa_helpers` 同进程库**(浏览器自动化)组合获取。
新平台必须先 `platform ensure --key <小写键> --display-name <展示名>` 再引用。`add-web` / `ensure-web` / `pick-web` **不会**根据 `--platform-display-name` 自动插入 `platforms` 行。
## 两种集成方式
| 方式 | 适用场景 | 调用形态 |
@@ -26,6 +28,7 @@
| 目的 | 命令 | 成功时 stdout(摘要) |
|---|---|---|
| 枚举平台 | `platform list [--domain ...]` | `{"success": true, "data": {"platforms": [...]}}` |
| 登记平台 | `platform ensure --key <小写键> --display-name <展示名>` | `{"ok": true, "platform_key": "...", "action": "created\|updated"}` |
| 平台详情 | `platform get <p>` | `{"success": true, "data": {...}}` |
| 取单条账号 JSON | `account get <id>` | **单行 JSON 对象**(无固定 `success` 字段) |
| 批量账号 JSON | `account list ...` | **单行 JSON 数组** |
@@ -37,6 +40,8 @@
| 导入加密凭据 | `account import-credentials [--replace-all]` | **通常为空**(计数在 stderr) |
| 初始化 master key | `account init-master-key [--print\|--from-env\|--rotate]` | **单行 JSON** |
| 新增网页账号 | `account add-web ...` | `{"success": true, "data": {...}}` |
| 复制网页账号 | `account copy <id>` | `{"success": true, "message": "已复制为「…」", "data": {...}}` |
| 编辑网页账号 | `account update --id <id> ...` | `{"success": true, "message": "已保存", "data": {...}}` |
| 新增凭据 | `account add-secret ...` | `{"success": true, "data": {...}}` |
| 获取 API Key | `credential pick ... [--reveal]` | `{"success": true, ...}` |
| 兼容入口 | `get <id>` / `list-json <p>` / `pick-web <p>` | 同上对应新式命令 |

View File

@@ -2,7 +2,12 @@
**权威来源**:`scripts/util/platforms.py` 中的 `PLATFORMS`(当前 **31** 条)。若本文与代码不一致,**以代码为准**。
解析函数:`resolve_platform_key(name)` — 支持英文键、`display_name`、任一 alias(大小写不敏感)。
解析:
- **引用**(`add-web` / `ensure-web` / `pick-web` / `platform get`):`resolve_platform_key_dynamic` — 已有键精确匹配;已有键可大小写折叠(`DeepSeek` → `deepseek`);展示名/别名仅在**唯一**对应一个键时通过。未注册键**拒绝写入**,不会自动 `INSERT`。
- **登记**(仅 `platform ensure`):键必须符合 `^[a-z0-9][a-z0-9_-]{1,63}$` 且**本身已是小写**(`HaoKan` 非法);`--display-name` 必填;别名不得抢其它平台已占用的展示名/别名。
`resolve_platform_key(name)` 仍是内置表的静态别名映射(无 DB、冲突别名后写覆盖),新代码请走动态解析。
---
@@ -99,3 +104,5 @@ CLI `platform list --domain` **仅接受** `logistics | llm | content`(见 `cm
## 平台注册到数据库
`seed_platforms(conn)` 在业务流程中随 `init_db()` 触发,将 `PLATFORMS` upsert 进 `platforms` 表(仅覆盖定义字段;**不会**自动清空用户自定义的 `default_auth_strategy`,除非迁移逻辑另行更新——详见 [`SCHEMA.md`](SCHEMA.md))。
业务 skill 要加新平台:先 `platform ensure --key <小写键> --display-name <展示名> [--url …]`,再 `add-web` / `ensure-web`。不要把中文名或 `HaoKan` 这种大小写混写当成键;也不要指望 `--platform-display-name` 自动建行。

View File

@@ -10,6 +10,7 @@ description: "通用账号 & 凭据 & 登录流程管理器。管理网页/RPA
## 它能帮你做什么
- **登记自定义平台** — `platform ensure --key douyin --display-name 抖音`;键必须是小写英文,展示名可以是中文。其它技能不能靠 add-web 自动建平台。
- **添加网页/RPA 账号** — 选平台,给出登录标识(邮箱/手机/用户名),自动创建浏览器 profile 目录。
- **添加 API Key/Token / 加密密码** — `none` / `env` / `windows_credential` / `local_encrypted`(Fernet 密文入库)。
- **查看/列出账号** — 按平台、环境、角色等多维度筛选。
@@ -58,4 +59,4 @@ description: "通用账号 & 凭据 & 登录流程管理器。管理网页/RPA
---
**作者**:深圳匠厂科技有限公司 · **版本**:2.0.0
**作者**:深圳匠厂科技有限公司 · **版本**:2.0.27

View File

@@ -4,10 +4,10 @@ CLI entry point: argv dispatch for account-manager v2.
Subcommand tree:
platform list/get/ensure
account add-web|ensure-web|add-secret|get|get-credential|list|resolve|pick-web|mark-session|mark-error|mark-used|open|reveal-profile
account add-web|copy|update|ensure-web|add-secret|get|get-credential|list|resolve|pick-web|mark-session|mark-error|mark-used|open|reveal-profile
init-master-key|export-credentials|import-credentials
credential pick|resolve
lease release|list|cleanup
lease release|force-release|list|cleanup
(legacy) add|list|get|list-json|pick-web|open|delete
Machine commands output single-line JSON where appropriate.
@@ -23,7 +23,9 @@ from service.account_crypto_commands import (
from service.account_service import (
cmd_account_add_secret,
cmd_account_add_web,
cmd_account_copy,
cmd_account_ensure_web,
cmd_account_update,
cmd_account_get,
cmd_account_get_credential,
cmd_account_list,
@@ -40,6 +42,7 @@ from service.account_service import (
cmd_get,
cmd_init_db,
cmd_lease_cleanup,
cmd_lease_force_release,
cmd_lease_list,
cmd_lease_release,
cmd_list,
@@ -71,11 +74,14 @@ _USAGE = """account-manager v2 — Credential & Browser Profile Manager
[--url <url>] [--auth-strategy <s>]
账号管理:
account add-web --platform <p> --login-id <id> --login-id-type <type>
account add-web --platform <p> [--login-id <id>] [--login-id-type <type>]
[--label <label>] [--tenant-id <t>] [--environment <env>]
[--role <role>] [--provider-code <code>] [--url <url>]
[--extra-json <json>] [--profile-dir <path>]
[--auth-strategy <s>] [--session-persistent 0|1] [--device-fingerprint <fp>]
account copy <id>
account update --id <id> [--label <label>] [--login-id <id>]
[--login-id-type <type>] [--url <url>] [--status active|disabled|needs_review]
account add-secret --platform <p> --credential-type <type>
--secret-storage env|windows_credential|none|local_encrypted
[--secret-ref <env_var>] [--secret-stdin]
@@ -116,6 +122,7 @@ _USAGE = """account-manager v2 — Credential & Browser Profile Manager
租约管理:
lease release <lease_token>
lease force-release --id <租约编号>
lease list [--active]
lease cleanup
@@ -256,7 +263,7 @@ def main() -> None:
args = p.parse_args(av[3:])
sys.exit(cmd_platform_ensure(
key=args.key,
display_name=args.display_name or args.key,
display_name=args.display_name,
domain=args.domain,
url=args.url,
auth_strategy=args.auth_strategy,
@@ -276,6 +283,13 @@ def main() -> None:
sub = av[2]
if sub == "add-web":
_cmd_add_web(av)
elif sub == "copy":
if len(av) < 4:
print("ERROR:CLI_BAD_ARGS account copy 需要 <id>。")
sys.exit(1)
cmd_account_copy(av[3])
elif sub == "update":
_cmd_account_update(av)
elif sub == "ensure-web":
_cmd_ensure_web(av)
elif sub == "add-secret":
@@ -357,7 +371,7 @@ def main() -> None:
# ---- lease subcommand ----
elif cmd == "lease":
if len(av) < 3:
print("ERROR:CLI_BAD_ARGS lease 需要子命令:release | list | cleanup")
print("ERROR:CLI_BAD_ARGS lease 需要子命令:release | force-release | list | cleanup")
sys.exit(1)
sub = av[2]
if sub == "release":
@@ -365,13 +379,25 @@ def main() -> None:
print("ERROR:CLI_BAD_ARGS lease release 需要 <lease_token>。")
sys.exit(1)
cmd_lease_release(av[3])
elif sub == "force-release":
lease_id = _get_opt(av, "--id")
if lease_id is None and len(av) >= 4 and not str(av[3]).startswith("-"):
lease_id = av[3]
if lease_id is None:
print("ERROR:CLI_BAD_ARGS lease force-release 需要 --id <租约编号>。")
sys.exit(1)
try:
cmd_lease_force_release(int(lease_id))
except ValueError:
print("ERROR:CLI_BAD_ARGS 租约编号必须是数字。")
sys.exit(1)
elif sub == "list":
active = _has_flag(av, "--active")
cmd_lease_list(active)
elif sub == "cleanup":
cmd_lease_cleanup()
else:
print("ERROR:CLI_BAD_ARGS lease 子命令必须是 release、list 或 cleanup。")
print("ERROR:CLI_BAD_ARGS lease 子命令必须是 release、force-release、list 或 cleanup。")
sys.exit(1)
# ---- Legacy aliases ----
@@ -467,7 +493,7 @@ def main() -> None:
def _cmd_add_web(argv: list[str]) -> None:
plat = _get_opt(argv, "--platform", "")
login_id = _get_opt(argv, "--login-id")
login_id_type = _get_opt(argv, "--login-id-type", "unknown")
login_id_type = _get_opt(argv, "--login-id-type")
label = _get_opt(argv, "--label")
tenant_id = _get_opt(argv, "--tenant-id")
environment = _get_opt(argv, "--environment", "production")
@@ -511,6 +537,34 @@ def _cmd_add_web(argv: list[str]) -> None:
)
def _cmd_account_update(argv: list[str]) -> None:
from db.accounts_repo import _UNSET
aid = _get_opt(argv, "--id")
if not aid and len(argv) >= 4 and not str(argv[3]).startswith("-"):
aid = argv[3]
label = _get_opt(argv, "--label")
login_id_type = _get_opt(argv, "--login-id-type")
status = _get_opt(argv, "--status")
login_id = _UNSET
if "--login-id" in argv:
login_id = _get_opt(argv, "--login-id") or ""
url = _UNSET
if "--url" in argv:
url = _get_opt(argv, "--url") or ""
if not aid:
print("ERROR:CLI_BAD_ARGS account update 需要 --id <id>。")
return
cmd_account_update(
aid,
label=label,
login_id=login_id,
login_id_type=login_id_type,
url=url,
status=status,
)
def _cmd_get_credential(argv: list[str]) -> None:
aid_raw = argv[3]
try:
@@ -619,7 +673,7 @@ def _cmd_pick_web_new(argv: list[str]) -> None:
def _cmd_ensure_web(argv: list[str]) -> None:
plat = _get_opt(argv, "--platform", "")
login_id = _get_opt(argv, "--login-id")
login_id_type = _get_opt(argv, "--login-id-type", "unknown")
login_id_type = _get_opt(argv, "--login-id-type")
label = _get_opt(argv, "--label")
tenant_id = _get_opt(argv, "--tenant-id")
environment = _get_opt(argv, "--environment")
@@ -651,7 +705,7 @@ def _cmd_ensure_web(argv: list[str]) -> None:
cmd_account_ensure_web(
platform_input=plat,
login_id=login_id,
login_id_type=login_id_type or "unknown",
login_id_type=login_id_type,
label=label,
tenant_id=tenant_id,
environment=environment,

View File

@@ -471,6 +471,58 @@ def update_account(
)
def list_account_labels(conn) -> list[str]:
cur = conn.cursor()
cur.execute("SELECT account_label FROM accounts WHERE account_label IS NOT NULL")
return [str(row[0]) for row in cur.fetchall() if row[0]]
def update_account_user_fields(
conn,
account_id: int,
*,
account_label: Optional[str] = None,
login_id: Any = _UNSET,
login_id_type: Optional[str] = None,
url: Any = _UNSET,
status: Optional[str] = None,
) -> None:
"""Patch fields the data-management edit form is allowed to change."""
sets: list[str] = []
params: list[Any] = []
if account_label is not None:
label = str(account_label).strip()
if not label:
raise ValueError("account_label must not be empty")
sets.append("account_label = ?")
params.append(label)
if login_id is not _UNSET:
text = str(login_id).strip() if login_id is not None else ""
sets.append("login_id = ?")
params.append(text or None)
if login_id_type is not None:
typed = str(login_id_type).strip() or "unknown"
sets.append("login_id_type = ?")
params.append(typed)
if url is not _UNSET:
text = str(url).strip() if url is not None else ""
sets.append("url = ?")
params.append(text)
if status is not None:
sets.append("status = ?")
params.append(status)
if not sets:
return
sets.append("updated_at = ?")
params.append(_now_unix())
params.append(int(account_id))
cur = conn.cursor()
cur.execute(
f"UPDATE accounts SET {', '.join(sets)} WHERE id = ?",
tuple(params),
)
def account_exists_with_login_id(conn, platform_key: str, login_id: str, environment: str, role: str) -> bool:
cur = conn.cursor()
cur.execute(
@@ -743,6 +795,59 @@ def release_lease(lease_token: str) -> bool:
conn.close()
def release_lease_by_id(lease_id: int) -> Optional[dict]:
"""Force-release an active lease by row id. Returns released row summary or None."""
log = get_skill_logger()
log.info("lease_force_release_attempt lease_id=%s", lease_id)
conn = get_conn()
try:
cur = conn.cursor()
now = _now_unix()
cur.execute(
"SELECT id, account_id, lease_token, holder, purpose, status "
"FROM account_leases WHERE id = ?",
(int(lease_id),),
)
row = cur.fetchone()
if not row:
log.warning("lease_force_release_not_found lease_id=%s", lease_id)
return None
status = str(row[5] or "")
if status != "active":
log.warning(
"lease_force_release_not_active lease_id=%s status=%s",
lease_id,
status,
)
return None
cur.execute(
"UPDATE account_leases SET status = 'released', released_at = ?, updated_at = ? "
"WHERE id = ? AND status = 'active'",
(now, now, int(lease_id)),
)
conn.commit()
if cur.rowcount == 0:
return None
token = str(row[2] or "")
log.info(
"lease_force_release_success lease_id=%s account_id=%s token_prefix=%s",
lease_id,
row[1],
token[:8] if token else "",
)
return {
"id": int(row[0]),
"account_id": int(row[1]),
"lease_token": token,
"holder": row[3] or "",
"purpose": row[4] or "",
"status": "released",
"released_at": _unix_to_iso_local(now),
}
finally:
conn.close()
def list_active_leases() -> list[dict]:
"""List all active (non-expired, non-released) leases."""
conn = get_conn()

View File

@@ -6,6 +6,7 @@ SQLite 无原生表/字段 COMMENT;中文展示名称通过这两张元数据
from __future__ import annotations
import json
from typing import Iterable
INTERNAL_METADATA_TABLES = frozenset({"_jiangchang_tables", "_jiangchang_columns"})
@@ -20,6 +21,103 @@ ACCOUNT_LEASES_TABLE = "account_leases"
DATETIME_UNIX_SECONDS = "datetime_unix_seconds"
STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"})
# Data-management lookup: host resolves options_json object into select options.
PLATFORM_KEY_LOOKUP_JSON = json.dumps(
{
"ref_table": "platforms",
"value_column": "platform_key",
"label_column": "display_name",
"filter_column": "enabled",
"filter_value": 1,
},
ensure_ascii=False,
)
ACCOUNT_ID_LOOKUP_JSON = json.dumps(
{
"ref_table": "accounts",
"value_column": "id",
"label_column": "account_label",
},
ensure_ascii=False,
)
def _enum_options(pairs: tuple[tuple[str, str], ...]) -> str:
return json.dumps(
[{"value": value, "label": label} for value, label in pairs],
ensure_ascii=False,
)
LOGIN_ID_TYPE_OPTIONS_JSON = _enum_options(
(
("unknown", "未指定"),
("phone", "手机号"),
("email", "邮箱"),
("username", "用户名"),
("customer_code", "客户代码"),
("api_account", "API 账号"),
)
)
ENVIRONMENT_OPTIONS_JSON = _enum_options(
(
("production", "生产"),
("simulator", "仿真"),
("staging", "预发"),
("test", "测试"),
)
)
ROLE_OPTIONS_JSON = _enum_options(
(
("default", "默认"),
("booking", "订舱"),
("tracking", "跟踪"),
("api", "API"),
("admin", "管理"),
("sales", "销售"),
)
)
ACCOUNT_STATUS_OPTIONS_JSON = _enum_options(
(
("active", "启用"),
("disabled", "停用"),
("needs_review", "待复核"),
)
)
SESSION_STATUS_OPTIONS_JSON = _enum_options(
(
("unknown", "未知"),
("likely_valid", "可能有效"),
("needs_login", "需要登录"),
("expired", "已过期"),
)
)
AUTH_STRATEGY_OPTIONS_JSON = _enum_options(
(
("password_auto", "账密自动"),
("password_with_captcha", "账密+验证码"),
("password_plus_2fa", "账密+2FA"),
("qr_code_manual", "扫码"),
("per_session_manual", "每次手动"),
("device_bound", "设备绑定"),
("client_certificate", "客户端证书"),
("api_token", "API Token"),
("sso_redirect", "SSO 跳转"),
)
)
YES_NO_OPTIONS_JSON = _enum_options((("1", "是"), ("0", "否")))
ENABLED_OPTIONS_JSON = _enum_options((("1", "启用"), ("0", "停用")))
DOMAIN_OPTIONS_JSON = _enum_options(
(
("logistics", "物流"),
("llm", "大模型"),
("content", "内容"),
("dev", "开发"),
("sim", "仿真"),
("generic", "通用"),
)
)
USER_BUSINESS_TABLES = (
PLATFORMS_TABLE,
ACCOUNTS_TABLE,
@@ -101,13 +199,28 @@ PLATFORMS_TABLE_METADATA = {
PLATFORMS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
_col("platform_key", "平台键", description="平台唯一标识(英文 kebab-case)", editable=0),
_col("display_name", "展示名称", description="平台在界面上的显示名称"),
_col("domain", "业务域", description="logistics / llm / content / dev / sim 等"),
_col(
"domain",
"业务域",
description="物流 / 大模型 / 内容等",
options_json=DOMAIN_OPTIONS_JSON,
),
_col("provider_code", "供应商代码", description="供应商或平台代码,可空"),
_col("default_url", "默认链接", description="平台默认入口 URL"),
_col("default_url", "默认链接", description="平台默认入口 URL", display_type="url"),
_col("aliases_json", "别名列表", description="平台别名 JSON 数组", editable=0),
_col("capabilities_json", "能力声明", description='能力 JSON,如 {"web":true,"api_key":false}', editable=0),
_col("enabled", "是否启用", description="1=启用,0=停用"),
_col("default_auth_strategy", "默认登录策略", description="新建账号时的默认 auth_strategy"),
_col(
"enabled",
"是否启用",
description="启用或停用该平台",
options_json=ENABLED_OPTIONS_JSON,
),
_col(
"default_auth_strategy",
"默认登录策略",
description="新建账号时的默认登录方式",
options_json=AUTH_STRATEGY_OPTIONS_JSON,
),
_ts("created_at", "创建时间", "记录创建时间(Unix 秒级时间戳)"),
_ts("updated_at", "更新时间", "记录最后更新时间(Unix 秒级时间戳)"),
)
@@ -123,26 +236,73 @@ ACCOUNTS_TABLE_METADATA = {
ACCOUNTS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
_col("id", "编号", description="账号唯一编号", editable=0),
_col("platform_key", "平台", description="所属平台键"),
_col(
"platform_key",
"平台",
description="所属平台键(来自平台信息表)",
display_type="lookup",
options_json=PLATFORM_KEY_LOOKUP_JSON,
editable=0,
),
_col("account_label", "账号名称", description="人可读的账号名称"),
_col("login_id", "登录账号", description="手机号、邮箱、用户名等登录标识"),
_col("login_id_type", "登录标识类型", description="phone / email / username / unknown 等"),
_col("tenant_id", "租户编号", description="租户或客户标识,可空"),
_col("environment", "环境", description="production / simulator / staging 等"),
_col("role", "角色", description="booking / tracking / api / default 等"),
_col("provider_code", "供应商代码", description="冗余供应商码,便于查询"),
_col(
"login_id_type",
"登录标识类型",
description="未指定 / 手机号 / 邮箱 / 用户名等",
options_json=LOGIN_ID_TYPE_OPTIONS_JSON,
),
_col("tenant_id", "租户编号", description="租户或客户标识,可空", editable=0),
_col(
"environment",
"环境",
description="生产 / 仿真 / 预发 / 测试",
options_json=ENVIRONMENT_OPTIONS_JSON,
editable=0,
),
_col(
"role",
"角色",
description="默认 / 订舱 / 跟踪 / API 等",
options_json=ROLE_OPTIONS_JSON,
editable=0,
),
_col("provider_code", "供应商代码", description="冗余供应商码,便于查询", editable=0),
_col("profile_dir", "浏览器配置目录", description="Playwright 持久化 Profile 路径", editable=0),
_col("url", "入口链接", description="账号默认入口 URL"),
_col("status", "状态", description="active / disabled / needs_review 等"),
_col("session_status", "会话状态", description="unknown / likely_valid / needs_login 等"),
_col("url", "入口链接", description="账号默认入口 URL", display_type="url"),
_col(
"status",
"状态",
description="启用 / 停用 / 待复核",
options_json=ACCOUNT_STATUS_OPTIONS_JSON,
),
_col(
"session_status",
"会话状态",
description="运行时登录态,由技能回写",
options_json=SESSION_STATUS_OPTIONS_JSON,
editable=0,
),
_ts("last_used_at", "最后使用时间", "最近一次被业务技能使用的时间"),
_ts("last_login_check_at", "最后登录检查时间", "最近一次登录状态检查时间"),
_col("last_error_code", "最近错误码", description="最近一次错误代码", editable=0),
_col("last_error_message", "最近错误信息", description="最近一次错误说明", editable=0),
_col("extra_json", "扩展信息", description="扩展 JSON 配置", editable=0),
_col("auth_strategy", "登录方式", description="九种登录策略之一"),
_col("session_persistent", "持久会话", description="是否需要持久化浏览器会话(0/1)"),
_col("device_fingerprint", "设备指纹", description="device_bound 策略专用,可空"),
_col(
"auth_strategy",
"登录方式",
description="跟随平台默认,一般无需手改",
options_json=AUTH_STRATEGY_OPTIONS_JSON,
editable=0,
),
_col(
"session_persistent",
"持久会话",
description="是否需要持久化浏览器会话",
options_json=YES_NO_OPTIONS_JSON,
editable=0,
),
_col("device_fingerprint", "设备指纹", description="设备绑定策略专用,可空", editable=0),
_ts("created_at", "创建时间", "记录创建时间(Unix 秒级时间戳)"),
_ts("updated_at", "更新时间", "记录最后更新时间(Unix 秒级时间戳)"),
)
@@ -158,8 +318,20 @@ CREDENTIALS_TABLE_METADATA = {
CREDENTIALS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
_col("id", "编号", description="凭据唯一编号", editable=0),
_col("account_id", "账号编号", description="关联的 accounts.id,可空"),
_col("platform_key", "平台", description="所属平台键"),
_col(
"account_id",
"账号编号",
description="关联的 accounts.id,可空",
display_type="lookup",
options_json=ACCOUNT_ID_LOOKUP_JSON,
),
_col(
"platform_key",
"平台",
description="所属平台键(来自平台信息表)",
display_type="lookup",
options_json=PLATFORM_KEY_LOOKUP_JSON,
),
_col("credential_label", "凭据名称", description="人可读的凭据名称"),
_col("credential_type", "凭据类型", description="api_key / password / bearer_token 等"),
_col("secret_storage", "密钥存储方式", description="none / env / windows_credential / local_encrypted", editable=0),

View File

@@ -10,10 +10,8 @@ from typing import Optional
from db.accounts_repo import (
insert_account,
resolve_auth_strategy_for_platform,
ensure_platform_from_caller_meta,
)
from db.auth_strategy import (
ALL_AUTH_STRATEGIES,
AUTH_STRATEGY_CLIENT_CERTIFICATE,
AUTH_STRATEGY_DEVICE_BOUND,
AUTH_STRATEGY_PASSWORD_PLUS_2FA,
@@ -22,14 +20,10 @@ from db.auth_strategy import (
AUTH_STRATEGY_SSO_REDIRECT,
)
from db.connection import get_conn, init_db
from util.logging_config import get_skill_logger
from util.platforms import (
DEFAULT_CAPABILITIES_JSON,
PlatformResolveResult,
_platform_list_cn_for_help,
is_valid_platform_key_format,
resolve_platform_key_dynamic,
seed_platforms,
)
@@ -137,7 +131,7 @@ def _remove_profile_dir(path: str) -> None:
@dataclass
class PlatformCallerMeta:
"""调用方为未注册自定义平台提供的自动注册元数据(pick-web / add-web)。"""
"""调用方可附带的平台展示信息。不再用于自动插入 platforms 行。"""
display_name: Optional[str] = None
url: Optional[str] = None
@@ -145,8 +139,17 @@ class PlatformCallerMeta:
auth_strategy: Optional[str] = None
aliases: list[str] = field(default_factory=list)
def is_sufficient_for_auto_register(self) -> bool:
return bool((self.display_name or "").strip() and (self.url or "").strip())
_RESOLVE_CODES_WITHOUT_PLATFORM_LIST = frozenset(
{
"PLATFORM_NOT_REGISTERED",
"INVALID_PLATFORM_KEY",
"PLATFORM_DISABLED",
"AMBIGUOUS_PLATFORM",
"PLATFORM_ALIAS_CONFLICT",
"INVALID_PLATFORM_DISPLAY_NAME",
}
)
def _emit_platform_resolve_error(result: PlatformResolveResult, hint: str = "") -> None:
@@ -164,77 +167,9 @@ def _resolve_or_auto_register(
*,
hint: str = "",
) -> Optional[str]:
"""解析平台;未注册时可在元数据充足时自动 ensure(幂等、仅填空字段)。"""
init_db()
conn = get_conn()
try:
result = resolve_platform_key_dynamic(platform_input, conn=conn)
if result.ok:
return result.platform_key
if result.error_code in ("PLATFORM_DISABLED", "INVALID_PLATFORM_KEY", "INVALID_PLATFORM"):
_emit_platform_resolve_error(result, hint=hint)
return None
if result.error_code != "PLATFORM_NOT_REGISTERED":
_emit_platform_resolve_error(result, hint=hint)
if result.error_code not in (
"PLATFORM_NOT_REGISTERED",
"INVALID_PLATFORM_KEY",
"PLATFORM_DISABLED",
):
_say("支持:" + _platform_list_cn_for_help())
return None
raw_key = (platform_input or "").strip().lower()
if not is_valid_platform_key_format(raw_key):
_emit_platform_resolve_error(
PlatformResolveResult(
error_code="INVALID_PLATFORM_KEY",
message=result.message or f"平台 key 格式非法:「{platform_input}」。",
),
hint=hint,
)
return None
if not meta or not meta.is_sufficient_for_auto_register():
_emit_platform_resolve_error(
PlatformResolveResult(
error_code="PLATFORM_NOT_REGISTERED",
message=(
f"平台 {raw_key} 尚未注册。"
"请传入 --platform-display-name 与 --platform-url,"
"或先执行 account platform ensure。"
),
),
hint=hint,
)
return None
if meta.auth_strategy and meta.auth_strategy not in ALL_AUTH_STRATEGIES:
_say(_err_json(
"ERR_AUTH_STRATEGY_NOT_SUPPORTED",
f"不支持的 --platform-auth-strategy:{meta.auth_strategy}",
))
return None
seed_platforms(conn)
action = ensure_platform_from_caller_meta(
conn,
raw_key,
display_name=(meta.display_name or "").strip(),
domain=(meta.domain or "generic").strip() or "generic",
default_url=(meta.url or "").strip(),
aliases=meta.aliases,
default_auth_strategy=meta.auth_strategy,
capabilities_json=DEFAULT_CAPABILITIES_JSON,
)
get_skill_logger().info(
"platform_auto_registered key=%s action=%s", raw_key, action
)
return raw_key
finally:
conn.close()
"""解析已注册平台;未注册则拒绝。meta 仅兼容旧调用方,不再自动建平台。"""
del meta
return _resolve_or_fail(platform_input, hint=hint)
def _resolve_or_fail(input_name: str, hint: str = "") -> Optional[str]:
@@ -250,6 +185,6 @@ def _resolve_or_fail(input_name: str, hint: str = "") -> Optional[str]:
return result.platform_key
_emit_platform_resolve_error(result, hint=hint)
if result.error_code not in ("PLATFORM_NOT_REGISTERED", "INVALID_PLATFORM_KEY", "PLATFORM_DISABLED"):
if result.error_code not in _RESOLVE_CODES_WITHOUT_PLATFORM_LIST:
_say("支持:" + _platform_list_cn_for_help())
return None

View File

@@ -36,6 +36,7 @@ from service.secret_store import (
make_windows_credential_target,
write_secret,
)
from util.account_identity import resolve_login_id_type
from util.logging_config import get_skill_logger
from util.platforms import get_platform_spec, seed_platforms
from util.profile_shortcut import create_profile_browser_shortcut
@@ -63,6 +64,7 @@ def create_web_account_record(
) -> Optional[dict]:
"""Create a web/RPA account. On error prints JSON and returns None; on success returns payload (no stdout)."""
log = get_skill_logger()
login_id_type = resolve_login_id_type(login_id_type, login_id)
log.info("account_add_web_attempt platform=%s login_id_type=%s env=%s role=%s",
platform_input, login_id_type, environment, role)
@@ -162,7 +164,9 @@ def create_web_account_record(
finally:
conn.close()
shortcut_path = create_profile_browser_shortcut(resolved_profile_dir, safe_label, log)
shortcut_path = create_profile_browser_shortcut(
resolved_profile_dir, safe_label, log, url=resolved_url,
)
log.info("account_add_web_success id=%s platform=%s profile_dir=%s", new_id, key, resolved_profile_dir)
ok_payload = {

View File

@@ -15,7 +15,7 @@ def cmd_account_ensure_web(
platform_input: str,
*,
login_id: Optional[str] = None,
login_id_type: str = "unknown",
login_id_type: Optional[str] = None,
label: Optional[str] = None,
tenant_id: Optional[str] = None,
environment: Optional[str] = None,

View File

@@ -0,0 +1,233 @@
# -*- coding: utf-8 -*-
"""Account copy / update CLI commands (data-management 复制 / 编辑)."""
from __future__ import annotations
import json
from typing import Any, Optional
from db.accounts_repo import (
_UNSET,
get_account_by_id,
list_account_labels,
update_account_user_fields,
)
from db.connection import get_conn, init_db
from service._svc_common import _err_json, _say
from service.account_add_commands import create_web_account_record
from util.account_identity import (
KNOWN_ACCOUNT_STATUSES,
next_copy_label,
resolve_login_id_type,
)
from util.logging_config import get_skill_logger
from util.platforms import get_platform_spec, seed_platforms
from util.profile_shortcut import create_profile_browser_shortcut
def _extra_json_str(account: dict) -> Optional[str]:
extra = account.get("extra_json")
if extra is None or extra == "":
return None
if isinstance(extra, (dict, list)):
return json.dumps(extra, ensure_ascii=False)
text = str(extra).strip()
return text or None
def _ok_result(data: dict, message: str) -> str:
return json.dumps(
{
"type": "result",
"schema_version": 1,
"status": "success",
"message": message,
"success": True,
"data": data,
},
ensure_ascii=False,
)
def cmd_account_copy(account_id) -> None:
"""Clone a web account: same platform/strategy, new empty Profile, empty login_id."""
log = get_skill_logger()
log.info("account_copy_attempt id=%r", account_id)
try:
aid = int(account_id)
except (TypeError, ValueError):
_say(_err_json("ERROR:CLI_BAD_ARGS", "account copy 需要整数 id。"))
return
init_db()
source = get_account_by_id(aid)
if not source:
_say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={aid} 不存在。"))
return
conn = get_conn()
try:
labels = list_account_labels(conn)
finally:
conn.close()
new_label = next_copy_label(str(source.get("account_label") or ""), labels)
tenant = str(source.get("tenant_id") or "").strip() or None
provider = str(source.get("provider_code") or "").strip() or None
url = str(source.get("url") or "").strip() or None
extra = _extra_json_str(source)
session_persistent = source.get("session_persistent")
try:
session_persistent_i = int(session_persistent) if session_persistent is not None else None
except (TypeError, ValueError):
session_persistent_i = None
ok_payload = create_web_account_record(
str(source.get("platform_key") or ""),
None,
str(source.get("login_id_type") or "unknown"),
new_label,
tenant,
str(source.get("environment") or "production") or "production",
str(source.get("role") or "default") or "default",
provider,
url,
extra,
None,
auth_strategy=source.get("auth_strategy"),
session_persistent=session_persistent_i,
device_fingerprint=None,
platform_key=str(source.get("platform_key") or "") or None,
)
if not ok_payload:
return
ok_payload["copied_from_id"] = aid
log.info(
"account_copy_success from=%s to=%s label=%s",
aid,
ok_payload.get("id"),
new_label,
)
_say(_ok_result(ok_payload, f"已复制为「{new_label}」"))
def cmd_account_update(
account_id,
*,
label: Optional[str] = None,
login_id: Any = _UNSET,
login_id_type: Optional[str] = None,
url: Any = _UNSET,
status: Optional[str] = None,
) -> None:
"""Update the short set of user-facing account fields."""
log = get_skill_logger()
log.info("account_update_attempt id=%r", account_id)
try:
aid = int(account_id)
except (TypeError, ValueError):
_say(_err_json("ERROR:CLI_BAD_ARGS", "account update 需要整数 --id。"))
return
init_db()
current = get_account_by_id(aid)
if not current:
_say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={aid} 不存在。"))
return
next_label = None
if label is not None:
next_label = str(label).strip()
if not next_label:
_say(_err_json("ERROR:CLI_BAD_ARGS", "账号名称不能为空。"))
return
next_status = None
if status is not None:
next_status = str(status).strip()
if next_status not in KNOWN_ACCOUNT_STATUSES:
_say(_err_json(
"ERROR:CLI_BAD_ARGS",
f"不支持的状态:{status}。支持:active / disabled / needs_review。",
))
return
resolved_login_id = _UNSET
if login_id is not _UNSET:
resolved_login_id = str(login_id).strip() if login_id is not None else ""
next_type = None
if login_id_type is not None or resolved_login_id is not _UNSET:
lid_for_infer = (
resolved_login_id
if resolved_login_id is not _UNSET
else current.get("login_id")
)
explicit = login_id_type if login_id_type is not None else current.get("login_id_type")
next_type = resolve_login_id_type(str(explicit or ""), lid_for_infer)
resolved_url = _UNSET
if url is not _UNSET:
text = str(url).strip() if url is not None else ""
if text:
resolved_url = text
else:
conn = get_conn()
try:
seed_platforms(conn)
spec = get_platform_spec(str(current.get("platform_key") or ""), conn=conn)
finally:
conn.close()
resolved_url = str(spec.get("default_url") or "")
conn = get_conn()
try:
update_account_user_fields(
conn,
aid,
account_label=next_label,
login_id=resolved_login_id,
login_id_type=next_type,
url=resolved_url,
status=next_status,
)
conn.commit()
except ValueError as exc:
conn.rollback()
_say(_err_json("ERROR:CLI_BAD_ARGS", str(exc)))
return
except Exception:
conn.rollback()
log.exception("account_update_failure")
_say(_err_json("ERROR:DB_ERROR", "Database update failed."))
return
finally:
conn.close()
updated = get_account_by_id(aid)
if not updated:
_say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={aid} 不存在。"))
return
profile_dir = str(updated.get("profile_dir") or "").strip()
if profile_dir:
create_profile_browser_shortcut(
profile_dir,
str(updated.get("account_label") or ""),
log,
url=str(updated.get("url") or "") or None,
)
log.info("account_update_success id=%s", aid)
_say(_ok_result(
{
"id": updated["id"],
"platform_key": updated.get("platform_key"),
"account_label": updated.get("account_label"),
"login_id": updated.get("login_id") or "",
"login_id_type": updated.get("login_id_type"),
"url": updated.get("url") or "",
"status": updated.get("status"),
},
"已保存",
))

View File

@@ -16,6 +16,7 @@ PyArmor trial per-file line limits. This module re-exports the public API.
from service._svc_common import PlatformCallerMeta
from service.account_add_commands import cmd_account_add_secret, cmd_account_add_web
from service.account_ensure_commands import cmd_account_ensure_web
from service.account_mutate_commands import cmd_account_copy, cmd_account_update
from service.account_mark_commands import (
cmd_account_mark_error,
cmd_account_mark_session,
@@ -38,7 +39,12 @@ from service.legacy_commands import (
cmd_delete_by_platform_phone,
cmd_list,
)
from service.lease_commands import cmd_lease_cleanup, cmd_lease_list, cmd_lease_release
from service.lease_commands import (
cmd_lease_cleanup,
cmd_lease_force_release,
cmd_lease_list,
cmd_lease_release,
)
from service.platform_commands import (
cmd_init_db,
cmd_platform_ensure,
@@ -50,7 +56,9 @@ __all__ = [
"PlatformCallerMeta",
"cmd_account_add_secret",
"cmd_account_add_web",
"cmd_account_copy",
"cmd_account_ensure_web",
"cmd_account_update",
"cmd_account_get",
"cmd_account_get_credential",
"cmd_account_list",
@@ -67,6 +75,7 @@ __all__ = [
"cmd_get",
"cmd_init_db",
"cmd_lease_cleanup",
"cmd_lease_force_release",
"cmd_lease_list",
"cmd_lease_release",
"cmd_list",

View File

@@ -2,7 +2,7 @@
"""
Chromium/Edge browser profile opener.
Uses system Chrome/Edge channel, Playwright launch_persistent_context.
Uses system Chrome/Edge executable with --user-data-dir (same profile as RPA shortcuts).
Does NOT download Playwright Chromium.
Does NOT check login status.
Does NOT write secret to SQLite.
@@ -11,20 +11,16 @@ import json
import os
import subprocess
import sys
import tempfile
from db.connection import get_conn, init_db
from db.accounts_repo import get_account_by_id, get_platform_from_db
from util.logging_config import (
get_skill_logger,
subprocess_env_with_trace,
from service.chrome_profile_prefs import (
ensure_session_restore,
profile_has_restorable_session,
)
from util.logging_config import get_skill_logger
from util.platforms import PLATFORMS
# 子进程入口脚本与本模块同目录(独立解释器执行,非 import)
_SERVICE_DIR = os.path.dirname(os.path.abspath(__file__))
_SCRIPTS_ROOT = os.path.dirname(_SERVICE_DIR)
def _win_find_exe(candidates):
for p in candidates:
@@ -83,6 +79,64 @@ def resolve_chromium_channel():
return "chrome"
# 给人核对/手工登录用:跳过 Chrome 首次欢迎页并最大化。不要带 Playwright 的 --enable-automation。
HUMAN_CHROME_FLAGS = (
"--start-maximized",
"--no-first-run",
"--no-default-browser-check",
"--disable-search-engine-choice-screen",
)
# 强制恢复上次会话(含 Session Cookie)。不依赖 Preferences.restore_on_startup
#(该项受 Secure Preferences HMAC 保护,裸写会被 Chrome 否掉)。
RESTORE_LAST_SESSION_FLAG = "--restore-last-session"
# 与 scrape-product-colipu 等业务技能约定同一算法,便于 CDP 挂接已打开的账号窗口。
_CDP_PORT_BASE = 9330
_CDP_PORT_SPAN = 60
def cdp_port_for_profile(profile_dir: str) -> int:
"""按 profile 绝对路径派生稳定本机调试端口(9330–9389)。"""
import hashlib
key = os.path.abspath(profile_dir).replace("/", "\\").lower().encode("utf-8")
digest = hashlib.sha1(key).hexdigest()
return _CDP_PORT_BASE + (int(digest[:8], 16) % _CDP_PORT_SPAN)
def human_chrome_launch_args(profile_dir, url=None, restore_session=False, *, enable_cdp=True):
"""chrome.exe 之后的参数(Popen 列表形式,路径不额外加引号)。"""
profile_abs = os.path.abspath(profile_dir)
args = [f"--user-data-dir={profile_abs}", *HUMAN_CHROME_FLAGS]
if restore_session:
args.append(RESTORE_LAST_SESSION_FLAG)
if enable_cdp:
port = cdp_port_for_profile(profile_abs)
args.append(f"--remote-debugging-port={port}")
args.append("--remote-debugging-address=127.0.0.1")
cleaned = (url or "").strip()
if cleaned:
args.append(cleaned)
return args
def human_chrome_shortcut_arguments(profile_dir, url=None, restore_session=False, *, enable_cdp=True):
"""Windows .lnk Arguments 字符串(user-data-dir 带引号)。"""
profile_abs = os.path.abspath(profile_dir)
parts = [f'--user-data-dir="{profile_abs}"', *HUMAN_CHROME_FLAGS]
if restore_session:
parts.append(RESTORE_LAST_SESSION_FLAG)
if enable_cdp:
port = cdp_port_for_profile(profile_abs)
parts.append(f"--remote-debugging-port={port}")
parts.append("--remote-debugging-address=127.0.0.1")
cleaned = (url or "").strip()
if cleaned:
parts.append(cleaned)
return " ".join(parts)
def _print_browser_install_hint():
print("❌ 未检测到 Google Chrome 或 Microsoft Edge,请先安装:")
print(" • Chrome: https://www.google.com/chrome/")
@@ -155,6 +209,10 @@ def cmd_reveal_profile(account_id):
print(
json.dumps(
{
"type": "result",
"schema_version": 1,
"status": "success",
"message": "",
"success": True,
"data": {
"account_id": int(target["id"]) if target.get("id") is not None else account_id,
@@ -166,33 +224,56 @@ def cmd_reveal_profile(account_id):
)
def _spawn_browser_child(cmd: list[str], *, cwd: str, env: dict) -> subprocess.Popen:
"""Detach Playwright child so `account open` can return immediately."""
popen_kwargs: dict = {
"cwd": cwd,
"env": env,
"stdin": subprocess.DEVNULL,
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"close_fds": True,
}
if sys.platform == "win32":
# Keep browser window; detach from parent console / Skill Action wait.
flags = subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS
create_no_window = getattr(subprocess, "CREATE_NO_WINDOW", 0)
if create_no_window:
flags |= create_no_window
popen_kwargs["creationflags"] = flags
else:
popen_kwargs["start_new_session"] = True
return subprocess.Popen(cmd, **popen_kwargs)
def _list_profile_browser_pids(profile_dir: str):
"""找出已占用本 Profile 的 chrome/msedge PID(与采集技能同逻辑)。"""
import json as _json
profile_abs = os.path.abspath(profile_dir)
needle = os.path.normcase(profile_abs)
if sys.platform != "win32":
return []
ps = (
"Get-CimInstance Win32_Process -Filter "
"\"name='chrome.exe' or name='msedge.exe'\" |"
" Select-Object ProcessId,CommandLine | ConvertTo-Json -Compress"
)
try:
completed = subprocess.run(
["powershell", "-NoProfile", "-Command", ps],
capture_output=True,
text=True,
timeout=20,
check=False,
)
except Exception:
return []
raw = (completed.stdout or "").strip()
if not raw:
return []
try:
payload = _json.loads(raw)
except Exception:
return []
rows = payload if isinstance(payload, list) else [payload]
pids = []
for row in rows:
if not isinstance(row, dict):
continue
cmdline = str(row.get("CommandLine") or "")
if needle not in os.path.normcase(cmdline):
continue
try:
pids.append(int(row.get("ProcessId")))
except (TypeError, ValueError):
continue
return sorted(set(pids))
def cmd_open(account_id):
"""
打开该账号的持久化浏览器,仅用于肉眼核对。
不写数据库、不判定是否已登录。
拉起浏览器子进程后立即返回(不等待窗口关闭),便于 Skill Action sync/direct。
打开该账号的持久化浏览器,仅用于肉眼核对 / 手工登录 / 抓 DOM。
与 reveal-profile 同模式:本进程直接拉起系统 Chrome/Edge 后立即返回。
最大化、跳过 Chrome 首次欢迎页;不写数据库、不判定是否已登录、不起 Playwright。
"""
log = get_skill_logger()
init_db()
@@ -204,23 +285,60 @@ def cmd_open(account_id):
print("ERROR:ACCOUNT_NOT_FOUND")
sys.exit(1)
channel = resolve_chromium_channel()
if not channel:
_print_browser_install_hint()
sys.exit(1)
try:
import playwright # noqa: F401
except ImportError:
print("ERROR:需要 playwright Python 包。")
sys.exit(1)
browser_exe = resolve_chromium_executable()
if not browser_exe:
channel = resolve_chromium_channel()
if sys.platform == "win32" or not channel:
_print_browser_install_hint()
sys.exit(1)
# 非 Windows:退回 PATH 上的常见命令名
browser_exe = "google-chrome" if channel == "chrome" else "microsoft-edge"
profile_dir = (target.get("profile_dir") or "").strip()
if not profile_dir:
log.warning("browser_open_error account_id=%s profile_dir_missing", account_id)
print("ERROR:PROFILE_DIR_MISSING 账号中 profile_dir 为空。")
sys.exit(1)
os.makedirs(profile_dir, exist_ok=True)
profile_abs = os.path.abspath(profile_dir)
os.makedirs(profile_abs, exist_ok=True)
existing = _list_profile_browser_pids(profile_abs)
if existing:
log.info(
"browser_open_reuse account_id=%s profile_dir=%s pids=%s",
account_id,
profile_abs,
existing,
)
print(
json.dumps(
{
"type": "result",
"schema_version": 1,
"status": "success",
"message": "已复用打开中的浏览器窗口",
"success": True,
"data": {
"account_id": int(target["id"]) if target.get("id") is not None else account_id,
"account_label": target.get("account_label"),
"browser": "Google Chrome",
"browser_exe": browser_exe,
"profile_dir": profile_abs,
"url": resolve_account_browser_url(target),
"pid": existing[0],
"reused": True,
"pids": existing,
},
},
ensure_ascii=False,
)
)
return
# 清理 exit_type 等(不写 MAC 保护的 restore_on_startup);有会话时靠 --restore-last-session
prefs_result = ensure_session_restore(profile_abs)
log.info("browser_open_prefs %s", prefs_result)
conn = get_conn()
try:
@@ -228,56 +346,81 @@ def cmd_open(account_id):
finally:
conn.close()
# 有可恢复会话:不带 URL + --restore-last-session(恢复标签与 Session Cookie)。
# 首次/空 Profile:仍打开平台首页便于登录。
has_session = profile_has_restorable_session(profile_abs)
launch_url = None if has_session else url
cdp_port = cdp_port_for_profile(profile_abs)
cmd = [
browser_exe,
*human_chrome_launch_args(
profile_abs, launch_url, restore_session=has_session, enable_cdp=True
),
]
log.info(
"browser_open_context profile_dir=%s url=%s channel=%s platform=%s",
profile_dir,
url,
channel,
"browser_open_launch exe=%s profile_dir=%s launch_url=%s has_session=%s "
"restore_last_session=%s cdp_port=%s platform=%s",
browser_exe,
profile_abs,
launch_url or "(restore)",
has_session,
has_session,
cdp_port,
target.get("platform_key"),
)
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
cfg = {"channel": channel, "profile_dir": profile_dir, "url": url}
with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False, encoding="utf-8"
) as jf:
json.dump(cfg, jf, ensure_ascii=False)
cfg_path = jf.name
try:
child = _spawn_browser_child(
[sys.executable, "-m", "service.open_child_runner", cfg_path],
cwd=_SCRIPTS_ROOT,
env=subprocess_env_with_trace(),
)
except Exception as e:
try:
os.unlink(cfg_path)
except OSError:
pass
log.error("browser_open_error account_id=%s error=%s", account_id, str(e))
print(f"ERROR:BROWSER_OPEN_FAILED {e}")
# GUI 浏览器:不挂控制台黑框;不等待关闭(与 reveal 一致立刻返回)
popen_kwargs: dict = {
"stdin": subprocess.DEVNULL,
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"close_fds": True,
}
if os.path.isabs(browser_exe):
popen_kwargs["cwd"] = os.path.dirname(browser_exe)
if sys.platform == "win32":
popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
else:
popen_kwargs["start_new_session"] = True
child = subprocess.Popen(cmd, **popen_kwargs)
except OSError as exc:
log.error("browser_open_error account_id=%s error=%s", account_id, str(exc))
print(f"ERROR:BROWSER_OPEN_FAILED {exc}")
sys.exit(1)
browser_name = (
"Microsoft Edge"
if os.path.basename(browser_exe).lower() == "msedge.exe"
else "Google Chrome"
)
log.info(
"browser_open_spawned account_id=%s channel=%s pid=%s",
"browser_open_ok account_id=%s browser=%s pid=%s",
account_id,
channel,
browser_name,
child.pid,
)
print(
json.dumps(
{
"type": "result",
"schema_version": 1,
"status": "success",
"message": "",
"success": True,
"data": {
"account_id": int(target["id"]) if target.get("id") is not None else account_id,
"account_label": target.get("account_label"),
"browser": browser_name,
"channel": channel,
"profile_dir": os.path.abspath(profile_dir),
"browser_exe": browser_exe,
"profile_dir": profile_abs,
"url": url,
"detached": True,
"pid": child.pid,
"has_session": has_session,
"restore_last_session": has_session,
"cdp_port": cdp_port,
"session_restore_on_startup": prefs_result.get("restore_after"),
},
},
ensure_ascii=False,

View File

@@ -0,0 +1,163 @@
"""Chrome/Edge 独立 Profile:启动前清理退出标记,便于会话文件可被恢复。
注意:``session.restore_on_startup`` 受 Secure Preferences HMAC 保护,
裸写 Preferences 会被 Chrome 校验否掉并回落到新标签页。真正强制恢复会话
应在启动参数使用 ``--restore-last-session``(见 browser_service)。
"""
from __future__ import annotations
import json
import os
from typing import Any, Dict
from util.logging_config import get_skill_logger
# Chromium ≤ roughly mid-era: session files lived under Default/
_LEGACY_SESSION_FILES = (
"Current Session",
"Last Session",
"Current Tabs",
"Last Tabs",
)
def profile_has_restorable_session(profile_dir: str) -> bool:
"""True when the profile has on-disk session data Chrome can restore.
Modern Chrome/Edge write ``Default/Sessions/Session_*`` (and ``Tabs_*``).
Older builds used ``Default/Current Session`` etc. Either form counts.
"""
default_dir = os.path.join(os.path.abspath(profile_dir), "Default")
if any(
os.path.isfile(os.path.join(default_dir, name))
for name in _LEGACY_SESSION_FILES
):
return True
sessions_dir = os.path.join(default_dir, "Sessions")
if not os.path.isdir(sessions_dir):
return False
try:
names = os.listdir(sessions_dir)
except OSError:
return False
return any(
name.startswith("Session_") or name.startswith("Tabs_")
for name in names
if os.path.isfile(os.path.join(sessions_dir, name))
)
def preferences_path(profile_dir: str) -> str:
return os.path.join(os.path.abspath(profile_dir), "Default", "Preferences")
def local_state_path(profile_dir: str) -> str:
return os.path.join(os.path.abspath(profile_dir), "Local State")
def _patch_preferences_dict(data: Dict[str, Any]) -> Dict[str, Any]:
"""只改不受 Secure Preferences MAC 保护的退出标记;不动 restore_on_startup。"""
restore_before = int((data.get("session") or {}).get("restore_on_startup") or 0)
exit_before = str((data.get("profile") or {}).get("exit_type") or "")
profile = data.setdefault("profile", {})
if str(profile.get("exit_type") or "").lower() == "crashed":
profile["exit_type"] = "Normal"
profile["exited_cleanly"] = True
browser = data.setdefault("browser", {})
browser["has_seen_welcome_page"] = True
browser["check_default_browser"] = False
return {"restore_before": restore_before, "exit_type_before": exit_before}
def _patch_local_state(profile_dir: str, log) -> Dict[str, Any]:
path = local_state_path(profile_dir)
if not os.path.isfile(path):
return {"ok": True, "skipped": True, "path": path}
try:
with open(path, "r", encoding="utf-8") as fh:
data = json.load(fh)
except Exception as exc:
log.warning("chrome_local_state_read_failed path=%s err=%s", path, exc)
return {"ok": False, "path": path, "error": str(exc)}
profile = data.setdefault("profile", {})
info = profile.setdefault("info_cache", {}).setdefault("Default", {})
info["active_time"] = float(info.get("active_time") or 0.0) or 1.0
exited = profile.get("exited_cleanly", True)
profile["exited_cleanly"] = True
try:
with open(path, "w", encoding="utf-8") as fh:
json.dump(data, fh, ensure_ascii=False, separators=(",", ":"))
except Exception as exc:
log.warning("chrome_local_state_write_failed path=%s err=%s", path, exc)
return {"ok": False, "path": path, "error": str(exc), "exited_cleanly_before": exited}
log.info("chrome_local_state_patched path=%s exited_cleanly %s->True", path, exited)
return {"ok": True, "path": path, "exited_cleanly_before": exited}
def ensure_session_restore(profile_dir: str) -> Dict[str, Any]:
"""启动前清理 exit_type / exited_cleanly(不受 MAC 保护);不写 restore_on_startup。"""
log = get_skill_logger()
profile_abs = os.path.abspath(profile_dir)
prefs = preferences_path(profile_abs)
os.makedirs(os.path.dirname(prefs), exist_ok=True)
data: Dict[str, Any] = {}
existed = os.path.isfile(prefs)
if existed:
try:
with open(prefs, "r", encoding="utf-8") as fh:
data = json.load(fh)
except Exception:
log.exception("chrome_prefs_read_failed path=%s", prefs)
data = {}
meta = _patch_preferences_dict(data)
before = int(meta["restore_before"])
exit_before = str(meta["exit_type_before"] or "")
try:
with open(prefs, "w", encoding="utf-8") as fh:
json.dump(data, fh, ensure_ascii=False, separators=(",", ":"))
except Exception as exc:
log.warning("chrome_prefs_write_failed path=%s err=%s", prefs, exc)
return {
"ok": False,
"path": prefs,
"existed": existed,
"restore_before": before,
"restore_after": before,
"exit_type_before": exit_before,
"error": str(exc),
}
after = int((data.get("session") or {}).get("restore_on_startup") or 0)
exit_after = str((data.get("profile") or {}).get("exit_type") or "")
local = _patch_local_state(profile_abs, log)
log.info(
"chrome_session_restore_prepared profile_dir=%s prefs=%s existed=%s "
"restore_on_startup_unchanged=%s exit_type %s->%s local_state=%s",
profile_abs,
prefs,
existed,
after,
exit_before or "-",
exit_after or "-",
local,
)
return {
"ok": True,
"path": prefs,
"existed": existed,
"restore_before": before,
"restore_after": after,
"exit_type_before": exit_before,
"exit_type_after": exit_after,
"local_state": local,
"prefs_restore_on_startup_touched": False,
}

View File

@@ -1,6 +1,11 @@
# -*- coding: utf-8 -*-
"""Lease management CLI commands."""
from db.accounts_repo import cleanup_expired_leases, list_active_leases, release_lease
from db.accounts_repo import (
cleanup_expired_leases,
list_active_leases,
release_lease,
release_lease_by_id,
)
from service._svc_common import _err_json, _ok_json, _say
from util.logging_config import get_skill_logger
@@ -16,6 +21,41 @@ def cmd_lease_release(lease_token: str) -> None:
_say(_err_json("ERROR:LEASE_NOT_FOUND", "未找到活跃的租约,或租约已释放/过期。"))
def cmd_lease_force_release(lease_id: int) -> None:
"""lease force-release --id <lease_row_id> — data-management row action."""
log = get_skill_logger()
log.info("cli_start subcommand=lease_force_release lease_id=%s", lease_id)
try:
lid = int(lease_id)
except (TypeError, ValueError):
_say(_err_json("ERROR:CLI_BAD_ARGS", "租约编号必须是数字。"))
return
if lid <= 0:
_say(_err_json("ERROR:CLI_BAD_ARGS", "租约编号必须是正整数。"))
return
released = release_lease_by_id(lid)
if not released:
_say(
_err_json(
"ERROR:LEASE_NOT_FOUND",
"未找到可释放的活跃租约(可能已释放、已过期清理,或不存在)。",
)
)
return
_say(
_ok_json(
{
"id": released["id"],
"account_id": released["account_id"],
"holder": released.get("holder") or "",
"purpose": released.get("purpose") or "",
"status": "released",
"message": "已强制释放租约,该账号可被其它任务占用。",
}
)
)
def cmd_lease_list(active_only: bool = False) -> None:
"""lease list [--active]"""
log = get_skill_logger()

View File

@@ -1,56 +0,0 @@
"""Playwright 仅打开浏览器子进程:由 browser_service.cmd_open 启动,argv[1] 为 JSON 配置路径。
父进程拉起后即返回;本子进程自行维护浏览器生命周期,读完配置后删除临时 JSON。
"""
import json
import os
import sys
from playwright.sync_api import sync_playwright
from util.playwright_stealth import (
STEALTH_INIT_SCRIPT,
persistent_context_launch_parts,
stealth_enabled,
)
def main():
cfg_path = sys.argv[1]
with open(cfg_path, encoding="utf-8") as f:
c = json.load(f)
try:
os.unlink(cfg_path)
except OSError:
pass
with sync_playwright() as p:
args, ignore_automation = persistent_context_launch_parts()
lc_kw = dict(
user_data_dir=c["profile_dir"],
headless=False,
channel=c["channel"],
no_viewport=True,
args=args,
)
if ignore_automation is not None:
lc_kw["ignore_default_args"] = ignore_automation
ctx = p.chromium.launch_persistent_context(**lc_kw)
if stealth_enabled():
ctx.add_init_script(STEALTH_INIT_SCRIPT)
try:
page = ctx.pages[0] if ctx.pages else ctx.new_page()
page.goto(c["url"], wait_until="domcontentloaded", timeout=60000)
ctx.wait_for_event("close", timeout=86400000)
except Exception as e:
print(e, file=sys.stderr)
raise
finally:
try:
ctx.close()
except Exception:
pass
if __name__ == "__main__":
main()

View File

@@ -8,7 +8,7 @@ from db.connection import get_conn, init_db
from service._svc_common import _err_json, _ok_json, _resolve_or_fail, _say
from util.constants import SKILL_SLUG
from util.logging_config import get_skill_logger
from util.platforms import DEFAULT_CAPABILITIES_JSON, is_valid_platform_key_format, seed_platforms
from util.platforms import DEFAULT_CAPABILITIES_JSON, seed_platforms
from util.runtime_paths import get_db_path
@@ -76,6 +76,20 @@ def cmd_platform_get(input_key: str) -> None:
))
def _aliases_for_ensure(aliases: Optional[str], key: str, display_name: str) -> tuple[list[str], str]:
from util.platforms import _parse_aliases_json
claimed: list[str] = []
extra = _parse_aliases_json(aliases) if aliases else []
for item in (display_name, key, *extra):
text = (item or "").strip()
if text and text not in claimed:
claimed.append(text)
if aliases:
return claimed, aliases
return claimed, json.dumps(claimed, ensure_ascii=False)
def cmd_platform_ensure(
key: str,
display_name: str,
@@ -89,25 +103,33 @@ def cmd_platform_ensure(
幂等地注册或更新平台到 DB。
输出单行 JSON:{"ok": true, "platform_key": "xxx", "action": "created|updated"}
"""
from util.platform_validate import validate_platform_registration
log = get_skill_logger()
key = (key or "").strip()
display_name = (display_name or key).strip()
display_name = (display_name or "").strip()
if not key:
_say('ERROR:INVALID_ARGS platform ensure 需要 --key 参数')
return 1
if not is_valid_platform_key_format(key):
_say(_err_json(
"ERROR:INVALID_PLATFORM_KEY",
f"平台 key 格式非法:「{key}」。建议使用小写字母、数字、下划线或连字符。",
))
return 1
aliases_json = aliases if aliases else json.dumps([key, display_name], ensure_ascii=False)
capabilities_json = capabilities if capabilities else DEFAULT_CAPABILITIES_JSON
init_db()
conn = get_conn()
try:
alias_list, aliases_json = _aliases_for_ensure(aliases, key, display_name)
check = validate_platform_registration(
key,
display_name,
aliases=alias_list,
conn=conn,
)
if not check.ok:
_say(_err_json(
f"ERROR:{check.error_code}",
check.message or "平台登记校验失败。",
))
return 1
capabilities_json = capabilities if capabilities else DEFAULT_CAPABILITIES_JSON
existed = bool(
conn.execute(
"SELECT 1 FROM platforms WHERE platform_key = ?", (key,)

View File

@@ -0,0 +1,53 @@
# -*- coding: utf-8 -*-
"""账号名称副本编号、登录标识类型推断。"""
from __future__ import annotations
import re
from typing import Iterable, Optional
_COPY_SUFFIX_RE = re.compile(r"^(.*?)(?: 副本(?: (\d+))?)?$")
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
_CN_MOBILE_RE = re.compile(r"^1\d{10}$")
_PHONE_RE = re.compile(r"^\+?\d{8,15}$")
_PHONE_STRIP_RE = re.compile(r"[\s\-()]")
KNOWN_LOGIN_ID_TYPES = frozenset(
{"email", "username", "phone", "customer_code", "api_account", "unknown"}
)
KNOWN_ACCOUNT_STATUSES = frozenset({"active", "disabled", "needs_review"})
def infer_login_id_type(login_id: Optional[str]) -> str:
"""从登录标识猜类型;空则 unknown。"""
text = (login_id or "").strip()
if not text:
return "unknown"
if _EMAIL_RE.match(text):
return "email"
compact = _PHONE_STRIP_RE.sub("", text)
if _CN_MOBILE_RE.match(compact) or _PHONE_RE.match(compact):
return "phone"
return "username"
def resolve_login_id_type(explicit: Optional[str], login_id: Optional[str]) -> str:
"""有明确类型(且不是 unknown)时用明确值,否则按 login_id 推断。"""
typed = (explicit or "").strip()
if typed and typed != "unknown":
return typed
return infer_login_id_type(login_id)
def next_copy_label(source_label: str, existing_labels: Iterable[str]) -> str:
"""在已有名称中生成「原名 副本」/「原名 副本 2」…,避免叠「副本 副本」。"""
raw = (source_label or "").strip() or "账号"
existing = {item.strip().lower() for item in existing_labels if (item or "").strip()}
match = _COPY_SUFFIX_RE.match(raw)
stem = (match.group(1).strip() if match else raw) or raw
n = 1
while n <= 9999:
candidate = f"{stem} 副本" if n == 1 else f"{stem} 副本 {n}"
if candidate.lower() not in existing:
return candidate
n += 1
return f"{stem} 副本 {n}"

View File

@@ -0,0 +1,264 @@
# -*- coding: utf-8 -*-
"""平台键 / 展示名校验:引用已有平台 vs 登记新平台。"""
from __future__ import annotations
import re
from typing import Any, Iterable, Optional
from util.platforms import (
PLATFORMS,
PlatformResolveResult,
_parse_aliases_json,
is_valid_platform_key_format,
)
_ASCII_TOKEN = re.compile(r"[A-Za-z0-9]")
def msg_empty() -> str:
return "平台名不能为空。请传平台键(如 douyin、deepseek),不要传空字符串。"
def msg_invalid_key(raw: str) -> str:
return (
f"平台键不合法:「{raw}」。\n"
"平台键必须是小写英文/数字/下划线或连字符,例如 douyin、deepseek。\n"
"「抖音」「DeepSeek」是展示名,不能当键。请改成已有键,或先登记:\n"
"platform ensure --key <键> --display-name <中文名>\n"
"查看已有键:platform list"
)
def msg_unrecognized(raw: str) -> str:
return (
f"无法识别平台:「{raw}」。\n"
"请传平台键(如 douyin),不要自创中文键。\n"
"先执行 platform list 核对;若是新平台,用 "
"platform ensure --key <键> --display-name <中文名> 登记后再调 add-web / ensure-web。"
)
def msg_not_registered(key: str) -> str:
return (
f"平台 `{key}` 尚未注册,已拒绝写入,以免产生不规范平台。\n"
f"若确认是新平台:platform ensure --key {key} --display-name <中文名> [--url …]\n"
"若其实是已有站:请改用 platform list 里的平台键。"
)
def msg_ambiguous(raw: str, keys: Iterable[str]) -> str:
listed = "、".join(f"`{k}`" for k in keys)
return f"平台「{raw}」对应多个键({listed}),请改传明确的平台键。"
def msg_alias_conflict(alias: str, owner: str, key: str) -> str:
return (
f"别名「{alias}」已被平台 `{owner}` 占用,不能再登记到 `{key}`。"
"请换展示名/别名,或复用已有平台键。"
)
def msg_display_name_required() -> str:
return (
"登记平台必须同时提供 --display-name(给人看的名称,可以是中文)。\n"
"平台键与展示名要分开,例如 --key douyin --display-name 抖音。"
)
def msg_disabled(key: str) -> str:
return f"平台 {key} 已禁用。"
def _looks_like_attempted_key(raw: str) -> bool:
return bool(_ASCII_TOKEN.search(raw or "")) and not is_valid_platform_key_format(raw)
def _lookup_stored_key(conn, key_lower: str) -> Optional[tuple[str, int]]:
"""Return (platform_key, enabled) for an exact key, builtin or DB."""
needle = (key_lower or "").strip().lower()
if not needle:
return None
if conn is not None:
row = conn.execute(
"SELECT platform_key, enabled FROM platforms WHERE lower(platform_key) = ?",
(needle,),
).fetchone()
if row:
return str(row[0]), int(row[1] or 0)
if needle in PLATFORMS:
return needle, 1
return None
def _iter_enabled_name_map(conn) -> list[tuple[str, set[str]]]:
"""Each enabled platform with its display_name + aliases (not the key)."""
out: list[tuple[str, set[str]]] = []
seen: set[str] = set()
if conn is not None:
rows = conn.execute(
"SELECT platform_key, enabled, display_name, aliases_json FROM platforms"
).fetchall()
for platform_key, enabled, display_name, aliases_json in rows:
key = str(platform_key)
seen.add(key)
if not enabled:
continue
names: set[str] = set()
if display_name:
names.add(str(display_name).strip())
for alias in _parse_aliases_json(aliases_json):
names.add(alias)
out.append((key, {n for n in names if n}))
for key, spec in PLATFORMS.items():
if key in seen:
continue
names = {str(spec.get("display_name") or "").strip()}
for alias in spec.get("aliases") or []:
names.add(str(alias).strip())
out.append((key, {n for n in names if n}))
return out
def _alias_matches(raw: str, conn) -> list[str]:
needle = (raw or "").strip()
if not needle:
return []
lower = needle.lower()
matches: list[str] = []
for key, names in _iter_enabled_name_map(conn):
lowered = {n.lower() for n in names}
if needle in names or lower in lowered:
if key not in matches:
matches.append(key)
return matches
def _keys_claiming_label(label: str, conn) -> list[str]:
"""Keys that already use this string as key, display_name, or alias."""
text = (label or "").strip()
if not text:
return []
owners: list[str] = []
stored = _lookup_stored_key(conn, text.lower())
if stored and stored[0] not in owners:
owners.append(stored[0])
for key in _alias_matches(text, conn):
if key not in owners:
owners.append(key)
return owners
def resolve_platform_ref(input_name: str, conn=None) -> PlatformResolveResult:
"""引用已有平台:已有键优先,其次唯一展示名/别名。不创建平台行。"""
raw = (input_name or "").strip()
if not raw:
return PlatformResolveResult(error_code="INVALID_PLATFORM", message=msg_empty())
close_conn = False
if conn is None:
from db.connection import get_conn, init_db
init_db()
conn = get_conn()
close_conn = True
try:
if is_valid_platform_key_format(raw):
stored = _lookup_stored_key(conn, raw)
if stored:
key, enabled = stored
if not enabled:
return PlatformResolveResult(
error_code="PLATFORM_DISABLED",
message=msg_disabled(key),
)
return PlatformResolveResult(platform_key=key)
folded = raw.lower()
if is_valid_platform_key_format(folded):
stored = _lookup_stored_key(conn, folded)
if stored:
key, enabled = stored
if not enabled:
return PlatformResolveResult(
error_code="PLATFORM_DISABLED",
message=msg_disabled(key),
)
return PlatformResolveResult(platform_key=key)
matches = _alias_matches(raw, conn)
if len(matches) == 1:
return PlatformResolveResult(platform_key=matches[0])
if len(matches) > 1:
return PlatformResolveResult(
error_code="AMBIGUOUS_PLATFORM",
message=msg_ambiguous(raw, matches),
)
if is_valid_platform_key_format(raw):
return PlatformResolveResult(
error_code="PLATFORM_NOT_REGISTERED",
message=msg_not_registered(raw),
)
if _looks_like_attempted_key(raw) or is_valid_platform_key_format(folded):
return PlatformResolveResult(
error_code="INVALID_PLATFORM_KEY",
message=msg_invalid_key(raw),
)
return PlatformResolveResult(
error_code="INVALID_PLATFORM",
message=msg_unrecognized(raw),
)
finally:
if close_conn:
conn.close()
def validate_platform_registration(
key: str,
display_name: str,
aliases: Optional[Iterable[Any]] = None,
conn=None,
) -> PlatformResolveResult:
"""登记/更新平台:键必须已是规范小写;展示名必填;别名不得抢别人的。"""
raw_key = (key or "").strip()
if not raw_key:
return PlatformResolveResult(error_code="INVALID_PLATFORM", message=msg_empty())
if not is_valid_platform_key_format(raw_key):
return PlatformResolveResult(
error_code="INVALID_PLATFORM_KEY",
message=msg_invalid_key(raw_key),
)
name = (display_name or "").strip()
if not name:
return PlatformResolveResult(
error_code="INVALID_PLATFORM_DISPLAY_NAME",
message=msg_display_name_required(),
)
close_conn = False
if conn is None:
from db.connection import get_conn, init_db
init_db()
conn = get_conn()
close_conn = True
try:
claimed = [name]
if aliases:
for item in aliases:
text = str(item).strip()
if text:
claimed.append(text)
for label in claimed:
if label.lower() == raw_key:
continue
owners = [k for k in _keys_claiming_label(label, conn) if k != raw_key]
if owners:
return PlatformResolveResult(
error_code="PLATFORM_ALIAS_CONFLICT",
message=msg_alias_conflict(label, owners[0], raw_key),
)
return PlatformResolveResult(platform_key=raw_key)
finally:
if close_conn:
conn.close()

View File

@@ -337,8 +337,8 @@ def resolve_platform_key(name: str) -> Optional[str]:
def is_valid_platform_key_format(name: str) -> bool:
"""平台 key 格式:^[a-z0-9][a-z0-9_-]{1,63}$"""
s = (name or "").strip().lower()
"""平台 key 格式:^[a-z0-9][a-z0-9_-]{1,63}$(不大小写折叠;登记必须本身已是小写)。"""
s = (name or "").strip()
if not s:
return False
return bool(PLATFORM_KEY_PATTERN.match(s))
@@ -367,87 +367,11 @@ class PlatformResolveResult:
return bool(self.platform_key) and not self.error_code
def _resolve_platform_from_db(conn, raw: str) -> PlatformResolveResult:
"""在 platforms 表中解析:精确 key(enabled=1)与 aliases_json。"""
raw_stripped = (raw or "").strip()
raw_lower = raw_stripped.lower()
if is_valid_platform_key_format(raw_stripped):
row = conn.execute(
"SELECT platform_key, enabled FROM platforms WHERE platform_key = ?",
(raw_lower,),
).fetchone()
if row:
if not row[1]:
return PlatformResolveResult(
error_code="PLATFORM_DISABLED",
message=f"平台 {row[0]} 已禁用。",
)
return PlatformResolveResult(platform_key=row[0])
rows = conn.execute(
"SELECT platform_key, enabled, aliases_json FROM platforms"
).fetchall()
for platform_key, enabled, aliases_json in rows:
if not enabled:
continue
names = {platform_key.lower()}
for alias in _parse_aliases_json(aliases_json):
names.add(alias.lower())
names.add(alias)
if raw_lower in names or raw_stripped in names:
return PlatformResolveResult(platform_key=platform_key)
return PlatformResolveResult()
def resolve_platform_key_dynamic(input_name: str, conn=None) -> PlatformResolveResult:
"""
统一平台解析:内置 PLATFORMS → DB platforms 表 → 错误分类。
"""
raw = (input_name or "").strip()
if not raw:
return PlatformResolveResult(
error_code="INVALID_PLATFORM",
message="平台名不能为空。",
)
"""引用已有平台(键 / 唯一展示名或别名)。实现见 util.platform_validate。"""
from util.platform_validate import resolve_platform_ref
builtin = resolve_platform_key(raw)
if builtin:
return PlatformResolveResult(platform_key=builtin)
close_conn = False
if conn is None:
from db.connection import get_conn, init_db
init_db()
conn = get_conn()
close_conn = True
try:
db_result = _resolve_platform_from_db(conn, raw)
if db_result.ok or db_result.error_code:
return db_result
finally:
if close_conn:
conn.close()
if not is_valid_platform_key_format(raw):
return PlatformResolveResult(
error_code="INVALID_PLATFORM_KEY",
message=(
f"平台 key 格式非法:「{raw}」。"
"建议使用小写字母、数字、下划线或连字符,例如 jiangchang。"
),
)
key_hint = raw.lower()
return PlatformResolveResult(
error_code="PLATFORM_NOT_REGISTERED",
message=(
f"平台 {key_hint} 尚未注册,请先执行 "
f"account platform ensure --key {key_hint} --display-name <名称> [--url <url>]"
),
)
return resolve_platform_ref(input_name, conn)
def get_platform_spec(platform_key: str, conn=None) -> dict[str, Any]:

View File

@@ -9,7 +9,10 @@ import subprocess
import sys
from typing import Optional
from service.browser_service import resolve_chromium_executable
from service.browser_service import (
human_chrome_shortcut_arguments,
resolve_chromium_executable,
)
_WIN_LNK_FORBIDDEN = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
@@ -28,9 +31,10 @@ def create_profile_browser_shortcut(
profile_dir: str,
account_label: str,
log: Optional[logging.Logger] = None,
url: Optional[str] = None,
) -> Optional[str]:
"""
在 profile_dir 内创建 {account_label}.lnk,目标为系统 Chrome/Edge + --user-data-dir。
在 profile_dir 内创建 {account_label}.lnk,与「打开浏览器」同一套人用参数。
非 Windows 或未安装 Chrome/Edge 时静默跳过,返回 None。
成功返回 .lnk 绝对路径。
@@ -52,7 +56,7 @@ def create_profile_browser_shortcut(
lnk_name = f"{_lnk_safe_filename(account_label)}.lnk"
lnk_path = os.path.join(profile_abs, lnk_name)
arguments = f'--user-data-dir="{profile_abs}"'
arguments = human_chrome_shortcut_arguments(profile_abs, url)
working_dir = os.path.dirname(browser_exe)
ps = (

View File

@@ -0,0 +1,153 @@
# -*- coding: utf-8 -*-
"""account copy / update 与数据管理短表单语义。"""
from __future__ import annotations
import json
import os
from db.accounts_repo import find_accounts, get_account_by_id
def _add(clean_db, **kwargs):
from service.account_service import cmd_account_add_web
defaults = dict(
platform_input="douyin",
login_id="13800138000",
login_id_type="phone",
label="抖音 默认",
tenant_id="t1",
environment="production",
role="default",
provider_code=None,
url="https://www.douyin.com",
extra_json_str='{"hook":"keep"}',
profile_dir_override=None,
auth_strategy="qr_code_manual",
session_persistent=1,
device_fingerprint=None,
)
defaults.update(kwargs)
cmd_account_add_web(**defaults)
class TestAccountCopy:
def test_copy_clones_platform_not_login_or_profile(self, clean_db, capsys):
from service.account_service import cmd_account_copy
_add(clean_db)
source = find_accounts(platform_key="douyin")[0]
cmd_account_copy(source["id"])
out = capsys.readouterr().out.strip()
payload = json.loads(out.splitlines()[-1])
assert payload["success"] is True
assert payload["type"] == "result"
assert "已复制为" in payload["message"]
copy_id = payload["data"]["id"]
assert copy_id != source["id"]
copied = get_account_by_id(copy_id)
assert copied["platform_key"] == "douyin"
assert copied["account_label"] == "抖音 默认 副本"
assert copied["login_id"] == ""
assert copied["tenant_id"] == "t1"
assert copied["environment"] == "production"
assert copied["role"] == "default"
assert copied["url"] == "https://www.douyin.com"
assert copied["auth_strategy"] == "qr_code_manual"
assert copied["session_persistent"] == 1
assert copied["device_fingerprint"] is None
assert copied["status"] == "active"
assert copied["session_status"] == "unknown"
assert (copied.get("extra_json") or {}).get("hook") == "keep"
assert copied["profile_dir"]
assert copied["profile_dir"] != source["profile_dir"]
assert os.path.isdir(copied["profile_dir"])
def test_copy_second_increments_label(self, clean_db, capsys):
from service.account_service import cmd_account_copy
_add(clean_db)
source = find_accounts(platform_key="douyin")[0]
cmd_account_copy(source["id"])
capsys.readouterr()
cmd_account_copy(source["id"])
payload = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
assert payload["data"]["account_label"] == "抖音 默认 副本 2"
def test_copy_missing_id(self, clean_db, capsys):
from service.account_service import cmd_account_copy
cmd_account_copy(99999)
payload = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
assert payload["success"] is False
assert payload["error"]["code"] == "ERROR:ACCOUNT_NOT_FOUND"
class TestAccountUpdate:
def test_update_short_fields(self, clean_db, capsys):
from service.account_service import cmd_account_update
_add(clean_db)
source = find_accounts(platform_key="douyin")[0]
cmd_account_update(
source["id"],
label="抖音 备用",
login_id="backup@example.com",
login_id_type="unknown",
url="https://www.douyin.com/user",
status="disabled",
)
payload = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
assert payload["success"] is True
assert payload["message"] == "已保存"
updated = get_account_by_id(source["id"])
assert updated["account_label"] == "抖音 备用"
assert updated["login_id"] == "backup@example.com"
assert updated["login_id_type"] == "email"
assert updated["url"] == "https://www.douyin.com/user"
assert updated["status"] == "disabled"
assert updated["auth_strategy"] == "qr_code_manual"
assert updated["profile_dir"] == source["profile_dir"]
def test_update_empty_login_clears(self, clean_db, capsys):
from service.account_service import cmd_account_update
_add(clean_db)
source = find_accounts(platform_key="douyin")[0]
cmd_account_update(source["id"], login_id="", login_id_type="unknown")
capsys.readouterr()
updated = get_account_by_id(source["id"])
assert updated["login_id"] == ""
assert updated["login_id_type"] == "unknown"
assert updated["account_label"] == "抖音 默认"
def test_update_rejects_bad_status(self, clean_db, capsys):
from service.account_service import cmd_account_update
_add(clean_db)
source = find_accounts(platform_key="douyin")[0]
cmd_account_update(source["id"], status="nope")
payload = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
assert payload["success"] is False
class TestAddWebInfersLoginType:
def test_email_without_type(self, clean_db, capsys):
from service.account_service import cmd_account_add_web
cmd_account_add_web(
platform_input="deepseek",
login_id="a@b.com",
login_id_type=None,
label="ds",
tenant_id=None,
environment="production",
role="default",
provider_code=None,
url=None,
extra_json_str=None,
profile_dir_override=None,
)
capsys.readouterr()
acc = find_accounts(platform_key="deepseek")[0]
assert acc["login_id_type"] == "email"

View File

@@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
"""delete id 会删库记录和硬盘 Profile,避免假删除残留登录态。"""
from __future__ import annotations
import os
from db.accounts_repo import find_accounts, get_account_by_id
def _add_web(**kwargs):
from service.account_service import cmd_account_add_web
defaults = dict(
platform_input="maersk",
login_id="demo@maersk.com",
login_id_type="email",
label="Maersk delete test",
tenant_id="tenant-demo",
environment="simulator",
role="booking",
provider_code=None,
url=None,
extra_json_str=None,
profile_dir_override=None,
)
defaults.update(kwargs)
cmd_account_add_web(**defaults)
def test_delete_by_id_removes_row_and_profile(clean_db, fake_env):
from service.account_service import cmd_delete_by_id
_add_web()
acc = find_accounts(platform_key="maersk")[0]
profile_dir = acc["profile_dir"]
marker = os.path.join(profile_dir, "Default", "Cookies")
os.makedirs(os.path.dirname(marker), exist_ok=True)
with open(marker, "w", encoding="utf-8") as handle:
handle.write("session")
assert os.path.isfile(marker)
cmd_delete_by_id(acc["id"])
assert get_account_by_id(acc["id"]) is None
assert find_accounts(platform_key="maersk") == []
assert not os.path.exists(profile_dir)
def test_delete_by_id_missing_account(clean_db, capsys):
from service.account_service import cmd_delete_by_id
cmd_delete_by_id(99999)
out = capsys.readouterr().out
assert "ERROR:ACCOUNT_NOT_FOUND" in out

View File

@@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
from util.account_identity import infer_login_id_type, next_copy_label, resolve_login_id_type
def test_infer_login_id_type():
assert infer_login_id_type(None) == "unknown"
assert infer_login_id_type("") == "unknown"
assert infer_login_id_type("demo@maersk.com") == "email"
assert infer_login_id_type("13800138000") == "phone"
assert infer_login_id_type("+8613800138000") == "phone"
assert infer_login_id_type("creator_01") == "username"
def test_resolve_login_id_type_keeps_explicit():
assert resolve_login_id_type("phone", "demo@x.com") == "phone"
assert resolve_login_id_type("unknown", "demo@x.com") == "email"
assert resolve_login_id_type(None, "13800138000") == "phone"
def test_next_copy_label_increments():
assert next_copy_label("抖音 默认", []) == "抖音 默认 副本"
assert next_copy_label("抖音 默认", ["抖音 默认 副本"]) == "抖音 默认 副本 2"
assert next_copy_label("抖音 默认 副本", ["抖音 默认 副本"]) == "抖音 默认 副本 2"
assert next_copy_label(
"抖音 默认",
["抖音 默认 副本", "抖音 默认 副本 2"],
) == "抖音 默认 副本 3"

View File

@@ -2,6 +2,9 @@
"""
Browser launch helper checks (no real browser started).
"""
import json
from unittest import mock
import pytest
@@ -14,34 +17,77 @@ class TestPlaywrightLaunchHelpers:
args, _ignore = persistent_context_launch_parts()
assert "--start-maximized" in args
def test_open_child_runner_cleans_config_after_read(self):
import os
path = os.path.join(
os.path.dirname(__file__),
"..",
"scripts",
"service",
"open_child_runner.py",
class TestEnsureSessionRestoreDoesNotTouchProtectedPref:
def test_leaves_restore_on_startup_unchanged(self, tmp_path, fake_env):
from pathlib import Path
from service.chrome_profile_prefs import ensure_session_restore, preferences_path
from util.logging_config import setup_skill_logging
setup_skill_logging("account-manager", "openclaw.skill.account_manager_test")
profile = tmp_path / "prof"
prefs = Path(preferences_path(str(profile)))
prefs.parent.mkdir(parents=True, exist_ok=True)
prefs.write_text(
json.dumps(
{
"session": {"restore_on_startup": 5},
"profile": {"exit_type": "Crashed", "exited_cleanly": False},
}
),
encoding="utf-8",
)
path = os.path.abspath(path)
with open(path, encoding="utf-8") as f:
src = f.read()
assert "os.unlink(cfg_path)" in src
assert "wait_for_event" in src
assert "launch_persistent_context" in src
assert "headless=False" in src
assert "no_viewport=True" in src
result = ensure_session_restore(str(profile))
assert result["ok"] is True
assert result["prefs_restore_on_startup_touched"] is False
data = json.loads(prefs.read_text(encoding="utf-8"))
assert data["session"]["restore_on_startup"] == 5
assert data["profile"]["exit_type"] == "Normal"
assert data["profile"]["exited_cleanly"] is True
class TestCmdOpenDetached:
def test_cmd_open_spawns_and_returns(self, clean_db, tmp_path, capsys, monkeypatch):
class TestProfileHasRestorableSession:
def test_empty_profile_false(self, tmp_path):
from service.chrome_profile_prefs import profile_has_restorable_session
assert profile_has_restorable_session(str(tmp_path / "prof")) is False
def test_legacy_current_session(self, tmp_path):
from service.chrome_profile_prefs import profile_has_restorable_session
default = tmp_path / "prof" / "Default"
default.mkdir(parents=True)
(default / "Current Session").write_bytes(b"x")
assert profile_has_restorable_session(str(tmp_path / "prof")) is True
def test_modern_sessions_dir(self, tmp_path):
from service.chrome_profile_prefs import profile_has_restorable_session
sessions = tmp_path / "prof" / "Default" / "Sessions"
sessions.mkdir(parents=True)
(sessions / "Session_13434010951759956").write_bytes(b"sess")
assert profile_has_restorable_session(str(tmp_path / "prof")) is True
def test_modern_tabs_only(self, tmp_path):
from service.chrome_profile_prefs import profile_has_restorable_session
sessions = tmp_path / "prof" / "Default" / "Sessions"
sessions.mkdir(parents=True)
(sessions / "Tabs_13434010952144072").write_bytes(b"tabs")
assert profile_has_restorable_session(str(tmp_path / "prof")) is True
class TestCmdOpenDirectBrowser:
"""account open = same-process Chrome/Edge launch (like reveal-profile)."""
def _seed_account(self, clean_db, tmp_path, profile):
from db.accounts_repo import insert_account
from service import browser_service
from util.platforms import seed_platforms
seed_platforms(clean_db)
profile = tmp_path / "prof"
aid = insert_account(
clean_db,
platform_key="sohu",
@@ -59,31 +105,133 @@ class TestCmdOpenDetached:
auth_strategy="password_auto",
)
clean_db.commit()
return aid
monkeypatch.setattr(browser_service, "resolve_chromium_channel", lambda: "chrome")
def _patch_launch(self, monkeypatch, browser_service, tmp_path):
fake_exe = str(tmp_path / "chrome.exe")
monkeypatch.setattr(
browser_service,
"resolve_chromium_executable",
lambda: fake_exe,
)
class FakeProc:
pid = 4242
pid = 5252
spawned = {}
def fake_spawn(cmd, *, cwd, env):
spawned["cmd"] = cmd
spawned["cwd"] = cwd
def fake_popen(cmd, **kwargs):
spawned["cmd"] = list(cmd)
spawned["kwargs"] = kwargs
return FakeProc()
monkeypatch.setattr(browser_service, "_spawn_browser_child", fake_spawn)
monkeypatch.setattr(browser_service.subprocess, "Popen", fake_popen)
return fake_exe, spawned
def test_cmd_open_launches_browser_exe(self, clean_db, tmp_path, capsys, monkeypatch):
from service import browser_service
profile = tmp_path / "prof"
aid = self._seed_account(clean_db, tmp_path, profile)
fake_exe, spawned = self._patch_launch(monkeypatch, browser_service, tmp_path)
browser_service.cmd_open(str(aid))
out = capsys.readouterr().out.strip()
import json
payload = json.loads(out.splitlines()[-1])
assert payload["success"] is True
assert payload["data"]["detached"] is True
assert payload["data"]["pid"] == 4242
assert "-m" in spawned["cmd"]
assert "service.open_child_runner" in spawned["cmd"]
assert payload["message"] == ""
assert payload["type"] == "result"
assert payload["status"] == "success"
assert payload["data"]["pid"] == 5252
assert payload["data"]["browser_exe"] == fake_exe
assert spawned["cmd"][0] == fake_exe
assert spawned["cmd"][1].startswith("--user-data-dir=")
assert "--start-maximized" in spawned["cmd"]
assert "--no-first-run" in spawned["cmd"]
assert "--no-default-browser-check" in spawned["cmd"]
assert "--disable-search-engine-choice-screen" in spawned["cmd"]
assert spawned["cmd"][-1] == "https://example.com"
assert "--enable-automation" not in spawned["cmd"]
# must NOT spawn python / open_child_runner
joined = " ".join(spawned["cmd"])
assert "open_child_runner" not in joined
assert "python" not in joined.lower()
def test_cmd_open_skips_url_when_modern_sessions_exist(
self, clean_db, tmp_path, capsys, monkeypatch
):
"""Modern Chrome Sessions/ must restore via --restore-last-session, no homepage URL."""
from service import browser_service
profile = tmp_path / "prof"
sessions = profile / "Default" / "Sessions"
sessions.mkdir(parents=True)
(sessions / "Session_13434010951759956").write_bytes(b"sess")
aid = self._seed_account(clean_db, tmp_path, profile)
_fake_exe, spawned = self._patch_launch(monkeypatch, browser_service, tmp_path)
browser_service.cmd_open(str(aid))
out = capsys.readouterr().out.strip()
payload = json.loads(out.splitlines()[-1])
assert payload["success"] is True
assert payload["data"]["has_session"] is True
assert payload["data"]["restore_last_session"] is True
assert "https://example.com" not in spawned["cmd"]
assert "--restore-last-session" in spawned["cmd"]
assert spawned["cmd"][-1] == "--restore-last-session" or spawned["cmd"][-1].startswith("--")
class TestHumanChromeLaunchArgs:
def test_flags_then_url(self, tmp_path):
from service.browser_service import HUMAN_CHROME_FLAGS, human_chrome_launch_args
profile = str(tmp_path / "prof")
args = human_chrome_launch_args(profile, "https://chat.deepseek.com")
assert args[0].startswith("--user-data-dir=")
for flag in HUMAN_CHROME_FLAGS:
assert flag in args
assert "--restore-last-session" not in args
assert any(a.startswith("--remote-debugging-port=") for a in args)
assert args[-1] == "https://chat.deepseek.com"
assert "--enable-automation" not in args
def test_restore_session_flag_without_url(self, tmp_path):
from service.browser_service import human_chrome_launch_args
args = human_chrome_launch_args(str(tmp_path / "prof"), None, restore_session=True)
assert "--restore-last-session" in args
assert any(a.startswith("--remote-debugging-port=") for a in args)
assert not any(a.startswith("http") for a in args)
def test_shortcut_arguments_quote_profile_and_include_url(self, tmp_path):
from service.browser_service import human_chrome_shortcut_arguments
profile = tmp_path / "my prof"
profile.mkdir()
text = human_chrome_shortcut_arguments(str(profile), "https://example.com")
assert '--user-data-dir="' in text
assert "--start-maximized" in text
assert "--no-first-run" in text
assert "--restore-last-session" not in text
assert "--remote-debugging-port=" in text
assert text.endswith("https://example.com")
def test_shortcut_restore_session(self, tmp_path):
from service.browser_service import human_chrome_shortcut_arguments
text = human_chrome_shortcut_arguments(str(tmp_path / "p"), None, restore_session=True)
assert "--restore-last-session" in text
assert "--remote-debugging-port=" in text
assert "http" not in text
def test_cdp_port_stable_for_profile(self, tmp_path):
from service.browser_service import cdp_port_for_profile
p = str(tmp_path / "prof")
assert cdp_port_for_profile(p) == cdp_port_for_profile(p)
assert 9330 <= cdp_port_for_profile(p) < 9390
class TestResolveChromiumChannel:

View File

@@ -56,7 +56,7 @@ class TestDynamicPlatformResolution:
result = resolve_platform_key_dynamic("江厂 空格", conn=clean_db)
assert not result.ok
assert result.error_code == "INVALID_PLATFORM_KEY"
assert result.error_code == "INVALID_PLATFORM"
def test_empty_platform(self, clean_db):
from util.platforms import resolve_platform_key_dynamic
@@ -97,6 +97,43 @@ class TestPlatformEnsure:
assert payload["platform_key"] == "jiangchang"
assert payload["action"] in ("created", "updated")
def test_ensure_requires_display_name(self, clean_db, capsys):
from service.account_service import cmd_platform_ensure
rc = cmd_platform_ensure(key="hao_kan", display_name="")
assert rc == 1
payload = json.loads(capsys.readouterr().out.strip())
assert payload["error"]["code"] == "ERROR:INVALID_PLATFORM_DISPLAY_NAME"
count = clean_db.execute(
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'hao_kan'"
).fetchone()[0]
assert count == 0
def test_ensure_rejects_mixed_case_key(self, clean_db, capsys):
from service.account_service import cmd_platform_ensure
rc = cmd_platform_ensure(key="HaoKan", display_name="好看")
assert rc == 1
payload = json.loads(capsys.readouterr().out.strip())
assert payload["error"]["code"] == "ERROR:INVALID_PLATFORM_KEY"
count = clean_db.execute(
"SELECT COUNT(*) FROM platforms WHERE lower(platform_key) = 'haokan'"
).fetchone()[0]
assert count == 0
def test_ensure_rejects_alias_owned_by_another_platform(self, clean_db, capsys):
from service.account_service import cmd_platform_ensure
rc = cmd_platform_ensure(key="hao_kan", display_name="抖音")
assert rc == 1
payload = json.loads(capsys.readouterr().out.strip())
assert payload["error"]["code"] == "ERROR:PLATFORM_ALIAS_CONFLICT"
assert "douyin" in payload["error"]["message"]
count = clean_db.execute(
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'hao_kan'"
).fetchone()[0]
assert count == 0
class TestCustomPlatformAccounts:
def test_add_web_after_ensure(self, clean_db):
@@ -161,7 +198,8 @@ class TestCustomPlatformAccounts:
payload = json.loads(out)
assert payload["success"] is False
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
assert "--platform-display-name" in payload["error"]["message"]
assert "platform ensure" in payload["error"]["message"]
assert "--platform-display-name" not in payload["error"]["message"]
def test_add_web_unregistered_returns_structured_error(self, clean_db, capsys):
from service.account_service import cmd_account_add_web
@@ -255,7 +293,7 @@ JIANGCHANG_META_KWARGS = dict(
)
class TestAutoRegisterFromCallerMeta:
class TestCallerMetaDoesNotAutoRegister:
def _meta(self):
from service.account_service import PlatformCallerMeta
@@ -293,9 +331,10 @@ class TestAutoRegisterFromCallerMeta:
payload = json.loads(out)
assert payload["success"] is False
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
assert "--platform-display-name" in payload["error"]["message"]
assert "platform ensure" in payload["error"]["message"]
assert "--platform-display-name" not in payload["error"]["message"]
def test_pick_web_unregistered_with_meta_auto_registers(self, clean_db, capsys):
def test_pick_web_unregistered_with_meta_does_not_insert(self, clean_db, capsys):
from service.account_service import cmd_account_pick_web, PlatformCallerMeta
from util.platforms import resolve_platform_key_dynamic
@@ -310,28 +349,33 @@ class TestAutoRegisterFromCallerMeta:
out = capsys.readouterr().out.strip()
payload = json.loads(out)
assert payload["success"] is False
assert payload["error"]["code"] == "ERROR:NO_ACCOUNT"
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
assert "已拒绝写入" in payload["error"]["message"]
result = resolve_platform_key_dynamic("auto_reg_plat", conn=clean_db)
assert result.ok
assert result.platform_key == "auto_reg_plat"
assert not result.ok
assert result.error_code == "PLATFORM_NOT_REGISTERED"
count = clean_db.execute(
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'auto_reg_plat'"
).fetchone()[0]
assert count == 0
def test_add_web_unregistered_with_meta_auto_registers(self, clean_db):
def test_add_web_unregistered_with_meta_does_not_insert(self, clean_db, capsys):
from service.account_service import cmd_account_add_web, PlatformCallerMeta
from db.accounts_repo import find_accounts
meta = PlatformCallerMeta(
display_name="匠厂后台",
display_name="未登记测试",
url=JIANGCHANG_URL,
domain="jc2009.com",
domain="example.com",
auth_strategy="qr_code_manual",
aliases=["jiangchang"],
aliases=["unreg_meta_plat"],
)
cmd_account_add_web(
platform_input="jiangchang",
platform_input="unreg_meta_plat",
login_id="manual",
login_id_type="manual",
label="匠厂后台",
label="未登记",
tenant_id=None,
environment="production",
role="admin",
@@ -341,9 +385,15 @@ class TestAutoRegisterFromCallerMeta:
profile_dir_override=None,
platform_meta=meta,
)
accs = find_accounts(platform_key="jiangchang")
assert len(accs) == 1
assert accs[0]["url"] == JIANGCHANG_URL
out = capsys.readouterr().out.strip()
payload = json.loads(out.splitlines()[0])
assert payload["success"] is False
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
assert find_accounts(platform_key="unreg_meta_plat") == []
count = clean_db.execute(
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'unreg_meta_plat'"
).fetchone()[0]
assert count == 0
def test_invalid_platform_key_no_register(self, clean_db, capsys):
from service.account_service import cmd_account_pick_web, PlatformCallerMeta
@@ -382,7 +432,7 @@ class TestAutoRegisterFromCallerMeta:
).fetchone()[0]
assert enabled == 0
def test_auto_register_idempotent(self, clean_db):
def test_caller_meta_does_not_insert_idempotent(self, clean_db, capsys):
from service.account_service import cmd_account_pick_web, PlatformCallerMeta
meta = PlatformCallerMeta(
@@ -395,4 +445,7 @@ class TestAutoRegisterFromCallerMeta:
count = clean_db.execute(
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'idem_plat'"
).fetchone()[0]
assert count == 1
assert count == 0
out = capsys.readouterr().out.strip().splitlines()[-1]
payload = json.loads(out)
assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"

View File

@@ -102,6 +102,44 @@ def test_sensitive_credential_fields_not_editable(metadata_db):
assert int(searchable) == 0, column_name
def test_account_and_credential_platform_lookups(metadata_db):
import json
cur = metadata_db.cursor()
for table_name, column_name in (
(ACCOUNTS_TABLE, "platform_key"),
(CREDENTIALS_TABLE, "platform_key"),
):
cur.execute(
"""
SELECT display_type, options_json
FROM _jiangchang_columns
WHERE table_name = ? AND column_name = ?
""",
(table_name, column_name),
)
display_type, options_json = cur.fetchone()
assert display_type == "lookup", table_name
payload = json.loads(options_json)
assert payload["ref_table"] == "platforms"
assert payload["value_column"] == "platform_key"
assert payload["label_column"] == "display_name"
cur.execute(
"""
SELECT display_type, options_json
FROM _jiangchang_columns
WHERE table_name = ? AND column_name = 'account_id'
""",
(CREDENTIALS_TABLE,),
)
display_type, options_json = cur.fetchone()
assert display_type == "lookup"
payload = json.loads(options_json)
assert payload["ref_table"] == "accounts"
assert payload["value_column"] == "id"
def test_lease_token_not_editable(metadata_db):
cur = metadata_db.cursor()
cur.execute(
@@ -144,6 +182,62 @@ def test_profile_dir_not_editable(metadata_db):
assert int(cur.fetchone()[0]) == 0
def test_account_edit_form_columns_and_enums(metadata_db):
import json
cur = metadata_db.cursor()
expected_editable = {
"account_label",
"login_id",
"login_id_type",
"url",
"status",
}
readonly_runtime = {
"platform_key",
"tenant_id",
"environment",
"role",
"provider_code",
"session_status",
"auth_strategy",
"session_persistent",
"device_fingerprint",
}
for column_name in expected_editable:
cur.execute(
"""
SELECT editable FROM _jiangchang_columns
WHERE table_name = ? AND column_name = ?
""",
(ACCOUNTS_TABLE, column_name),
)
assert int(cur.fetchone()[0]) == 1, column_name
for column_name in readonly_runtime:
cur.execute(
"""
SELECT editable FROM _jiangchang_columns
WHERE table_name = ? AND column_name = ?
""",
(ACCOUNTS_TABLE, column_name),
)
assert int(cur.fetchone()[0]) == 0, column_name
cur.execute(
"""
SELECT options_json FROM _jiangchang_columns
WHERE table_name = ? AND column_name = 'status'
""",
(ACCOUNTS_TABLE,),
)
options = json.loads(cur.fetchone()[0])
values = {item["value"] for item in options}
labels = {item["label"] for item in options}
assert values == {"active", "disabled", "needs_review"}
assert "启用" in labels
assert "停用" in labels
def test_timestamp_columns_use_datetime_display_type(metadata_db):
cur = metadata_db.cursor()
for table_name in USER_BUSINESS_TABLES:

View File

@@ -144,6 +144,67 @@ class TestLeaseRelease:
assert result["success"] is False
assert result["error"]["code"] == "ERROR:LEASE_NOT_FOUND"
def test_force_release_by_id(self, clean_db):
"""force-release by lease row id frees the account for pick-web."""
from db.accounts_repo import acquire_lease, find_account_ids, release_lease_by_id
from service.account_service import cmd_account_add_web, cmd_lease_force_release
import io, sys
cmd_account_add_web(
"maersk", "force-release@maersk.com", "email", "FR", "t1", "production", "booking",
None, None, None, None,
)
ids = find_account_ids("maersk")
acc_id = ids[0]
lease = acquire_lease(
account_id=acc_id,
lease_token="f1e2d3c4b5a697887766554433221100",
holder="stuck-holder",
purpose="download",
ttl_sec=900,
)
assert find_account_ids("maersk") == []
released = release_lease_by_id(int(lease["id"]))
assert released is not None
assert released["status"] == "released"
assert acc_id in find_account_ids("maersk")
# Re-acquire and exercise CLI command output shape
lease2 = acquire_lease(
account_id=acc_id,
lease_token="a0b1c2d3e4f5061728394a5b6c7d8e9f",
holder="stuck-2",
purpose="download",
ttl_sec=900,
)
buf = io.StringIO()
old = sys.stdout
sys.stdout = buf
try:
cmd_lease_force_release(int(lease2["id"]))
finally:
sys.stdout = old
line = buf.getvalue().strip().splitlines()[-1]
result = json.loads(line)
assert result.get("success") is True
assert (result.get("data") or {}).get("status") == "released"
def test_force_release_inactive_fails(self, clean_db):
from service.account_service import cmd_lease_force_release
import io, sys
captured = io.StringIO()
sys.stdout = captured
try:
cmd_lease_force_release(999999)
finally:
sys.stdout = sys.__stdout__
result = json.loads(captured.getvalue())
assert result["success"] is False
assert result["error"]["code"] == "ERROR:LEASE_NOT_FOUND"
class TestLeaseCleanup:
"""lease cleanup / expired lease handling."""

View File

@@ -74,3 +74,60 @@ class TestResolvePlatformKeyDynamicParsing:
clean_db.commit()
result = resolve_platform_key_dynamic("bad_alias", conn=clean_db)
assert result.platform_key == "bad_alias"
class TestPlatformKeyFormatAndAmbiguity:
def test_format_rejects_mixed_case(self):
from util.platforms import is_valid_platform_key_format
assert is_valid_platform_key_format("haokan") is True
assert is_valid_platform_key_format("HaoKan") is False
assert is_valid_platform_key_format("hao_kan") is True
def test_deepseek_display_resolves_to_key(self, clean_db):
from util.platforms import resolve_platform_key_dynamic
result = resolve_platform_key_dynamic("DeepSeek", conn=clean_db)
assert result.ok
assert result.platform_key == "deepseek"
def test_douyin_alias_still_unique(self, clean_db):
from util.platforms import resolve_platform_key_dynamic
result = resolve_platform_key_dynamic("抖音", conn=clean_db)
assert result.platform_key == "douyin"
def test_wechat_key_wins(self, clean_db):
from util.platforms import resolve_platform_key_dynamic
result = resolve_platform_key_dynamic("wechat", conn=clean_db)
assert result.ok
assert result.platform_key == "wechat"
def test_gongzhonghao_is_ambiguous(self, clean_db):
from util.platforms import resolve_platform_key_dynamic
result = resolve_platform_key_dynamic("公众号", conn=clean_db)
assert not result.ok
assert result.error_code == "AMBIGUOUS_PLATFORM"
assert "wechat" in (result.message or "")
assert "wechat_mp" in (result.message or "")
def test_unregistered_lookalike_key(self, clean_db):
from util.platforms import resolve_platform_key_dynamic
result = resolve_platform_key_dynamic("HaoKan", conn=clean_db)
assert result.error_code == "INVALID_PLATFORM_KEY"
assert "platform ensure" in (result.message or "")
def test_unregistered_valid_key_not_inserted(self, clean_db):
from util.platforms import resolve_platform_key_dynamic
result = resolve_platform_key_dynamic("hao_kan", conn=clean_db)
assert result.error_code == "PLATFORM_NOT_REGISTERED"
assert "已拒绝写入" in (result.message or "")
count = clean_db.execute(
"SELECT COUNT(*) FROM platforms WHERE platform_key = 'hao_kan'"
).fetchone()[0]
assert count == 0

View File

@@ -31,6 +31,33 @@ class TestCreateProfileBrowserShortcut:
out = create_profile_browser_shortcut(str(tmp_path), "demo")
assert out is None
def test_windows_lnk_uses_human_chrome_args(self, tmp_path, monkeypatch) -> None:
captured = {}
def fake_run(argv, **kwargs):
captured["argv"] = list(argv)
class Result:
returncode = 0
return Result()
monkeypatch.setattr("util.profile_shortcut.sys.platform", "win32")
monkeypatch.setattr(
"util.profile_shortcut.resolve_chromium_executable",
lambda: r"C:\Program Files\Google\Chrome\Application\chrome.exe",
)
monkeypatch.setattr("util.profile_shortcut.subprocess.run", fake_run)
monkeypatch.setattr("util.profile_shortcut.os.path.isfile", lambda path: True)
out = create_profile_browser_shortcut(
str(tmp_path), "demo", url="https://chat.deepseek.com",
)
assert out is not None
ps = captured["argv"][-1]
assert "--start-maximized" in ps
assert "--no-first-run" in ps
assert "https://chat.deepseek.com" in ps
assert "--enable-automation" not in ps
@pytest.mark.skipif(sys.platform != "win32", reason="Windows .lnk only")
def test_creates_lnk_on_windows(self, tmp_path, monkeypatch) -> None:
from service.browser_service import resolve_chromium_executable

View File

@@ -54,6 +54,9 @@ def test_reveal_profile_opens_dir(clean_db, tmp_path, capsys):
out = capsys.readouterr().out.strip()
payload = json.loads(out.splitlines()[-1])
assert payload["success"] is True
assert payload["message"] == ""
assert payload["type"] == "result"
assert payload["status"] == "success"
assert payload["data"]["account_id"] == aid
assert payload["data"]["profile_dir"] == called_path

View File

@@ -0,0 +1,121 @@
# -*- coding: utf-8 -*-
"""数据管理行内打开类:label 对齐模板,不与宿主 CRUD 重复。"""
from __future__ import annotations
import json
import os
import unittest
def _skill_root() -> str:
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def _load_actions() -> list[dict]:
path = os.path.join(_skill_root(), "assets", "actions.json")
with open(path, encoding="utf-8") as f:
manifest = json.load(f)
return list(manifest.get("actions") or [])
def _by_id(actions: list[dict], action_id: str) -> dict:
for item in actions:
if item.get("id") == action_id:
return item
raise AssertionError(f"missing action {action_id}")
class TestRowOpenActions(unittest.TestCase):
def test_reveal_and_open_use_template_labels(self) -> None:
actions = _load_actions()
reveal = _by_id(actions, "reveal-profile")
open_browser = _by_id(actions, "open-browser")
self.assertEqual(reveal.get("label"), "打开目录")
self.assertEqual(open_browser.get("label"), "打开浏览器")
self.assertEqual(reveal.get("placements"), ["row"])
self.assertEqual(open_browser.get("placements"), ["row"])
self.assertEqual(reveal.get("interaction"), "direct")
self.assertEqual(open_browser.get("interaction"), "direct")
self.assertFalse(reveal.get("replaces"))
self.assertFalse(open_browser.get("replaces"))
def test_platform_create_takes_over_host_add(self) -> None:
create = _by_id(_load_actions(), "platform-ensure")
self.assertEqual(create.get("replaces"), "create")
self.assertEqual(create.get("placements") or [], [])
self.assertEqual(create.get("label"), "新增平台")
self.assertEqual((create.get("bind") or {}).get("tables"), ["platforms"])
props = (create.get("inputSchema") or {}).get("properties") or {}
self.assertEqual(set(props), {"platform_key", "display_name", "default_url"})
args = (create.get("entrypoint") or {}).get("args") or []
self.assertIn("--key", args)
self.assertIn("--display-name", args)
def test_create_takes_over_host_add_without_row_duplicate(self) -> None:
create = _by_id(_load_actions(), "account-add-web")
self.assertEqual(create.get("replaces"), "create")
self.assertEqual(create.get("placements") or [], [])
self.assertEqual(create.get("label"), "新增账号")
props = (create.get("inputSchema") or {}).get("properties") or {}
self.assertEqual(set(props), {"platform_key", "account_label", "login_id"})
def test_update_takes_over_host_edit(self) -> None:
update = _by_id(_load_actions(), "account-update")
self.assertEqual(update.get("replaces"), "update")
self.assertEqual(update.get("placements") or [], [])
self.assertEqual(update.get("label"), "编辑账号")
props = (update.get("inputSchema") or {}).get("properties") or {}
self.assertEqual(
set(props),
{"accountId", "account_label", "login_id", "login_id_type", "url", "status"},
)
self.assertIn("enum", props["login_id_type"])
self.assertIn("enum", props["status"])
def test_delete_takes_over_host_delete(self) -> None:
delete = _by_id(_load_actions(), "account-delete")
self.assertEqual(delete.get("replaces"), "delete")
self.assertEqual(delete.get("placements") or [], [])
self.assertEqual(delete.get("interaction"), "confirm")
self.assertEqual(delete.get("label"), "删除账号")
bind = delete.get("bind") or {}
self.assertEqual(bind.get("tables"), ["accounts"])
self.assertEqual(bind.get("inputMapping"), {"accountId": "$row.$pk"})
self.assertEqual(
(delete.get("entrypoint") or {}).get("command"),
"delete",
)
self.assertEqual(
(delete.get("entrypoint") or {}).get("args"),
["id", "{{accountId}}"],
)
steps = delete.get("steps") or []
self.assertTrue(any(s.get("do") == "reloadTable" for s in steps))
def test_copy_is_direct_row_action(self) -> None:
copy = _by_id(_load_actions(), "account-copy")
self.assertEqual(copy.get("label"), "复制")
self.assertEqual(copy.get("placements"), ["row"])
self.assertEqual(copy.get("interaction"), "direct")
self.assertFalse(copy.get("replaces"))
def test_lease_force_release_row_action(self) -> None:
action = _by_id(_load_actions(), "lease-force-release")
self.assertEqual(action.get("label"), "强制释放租约")
self.assertEqual(action.get("placements"), ["row"])
self.assertEqual(action.get("interaction"), "confirm")
self.assertEqual(action.get("executionProfile"), "sync")
bind = action.get("bind") or {}
self.assertEqual(bind.get("tables"), ["account_leases"])
self.assertEqual(bind.get("inputMapping"), {"leaseId": "$row.$pk"})
self.assertEqual(bind.get("visibleWhen"), {"status": "active"})
self.assertEqual(
(action.get("entrypoint") or {}).get("args"),
["force-release", "--id", "{{leaseId}}"],
)
steps = action.get("steps") or []
self.assertTrue(any(s.get("do") == "reloadTable" for s in steps))
if __name__ == "__main__":
unittest.main()

View File

@@ -27,6 +27,8 @@ def test_usage_includes_all_account_subcommands():
"init-master-key",
"export-credentials",
"import-credentials",
"copy",
"update",
"add-web",
"ensure-web",
"get-credential",