Compare commits

...

9 Commits

Author SHA1 Message Date
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
0f54204a61 fix: account open fire-and-forget; open-browser direct+sync
All checks were successful
技能自动化发布 / release (push) Successful in 8s
2026-07-31 11:43:11 +08:00
f48e8aa75d feat: row actions use interaction direct/confirm
All checks were successful
技能自动化发布 / release (push) Successful in 10s
2026-07-31 11:15:05 +08:00
f5bab01421 feat: 数据管理行内打开文件位置与打开浏览器
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-07-31 10:02:00 +08:00
cbbec7cdfc fix: add init-db CLI for host install/upgrade metadata seed
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-07-26 16:03:27 +08:00
5f0d9e45f8 feat: add Chinese display metadata for data management UI
All checks were successful
技能自动化发布 / release (push) Successful in 8s
2026-07-26 15:46:18 +08:00
16 changed files with 1287 additions and 134 deletions

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.11
version: 2.0.20
author: 深圳匠厂科技有限公司
metadata:
openclaw:
@@ -27,19 +27,21 @@ allowed-tools:
## 何时使用本技能
1. **登记网页/RPA 账号**:`account add-web` — 平台 + 登录标识 + 自动生成 profile_dir;可选 `--auth-strategy` / `--session-persistent` / `--device-fingerprint`。
2. **登记凭据(API Key / Token / 加密密码)**:`account add-secret` — `none` / `env` / `windows_credential` / `local_encrypted`。
3. **查看/列出账号**:`account get/list` — JSON 输出;list 可按 `--login-id` / `--label` 筛选。
4. **用人话定位账号**:`account resolve --login-id <手机号/用户名>` 或 `--label` / `--query`(不必先查数字 id)。
5. **为 RPA 挑选账号**:`account pick-web` — 按平台/环境/角色,也可 `--login-id` / `--label`;业务首启推荐 `account ensure-web`(或 `pick-web --ensure`):无账号则自动 `add-web` 再返回。
6. **获取 API Key(凭据表)**:`credential pick` — 默认隐藏 secret,`--reveal` 时才返回。
7. **浏览器打开查看**:`account open <id>`(或兼容别名 `open <id>`)— 仅查看,不做站点侧登录判定。
8. **管理 lease**:`lease release/list/cleanup` — 防止 RPA 并发。
9. **初始化加密**:`account init-master-key` — 生成 Fernet master.key。
10. **加密备份与迁移**:`account export-credentials --plaintext` / `account import-credentials` — 导出明文 JSON、在新环境重新加密写入。
11. **按 lease 取明文密码**:`account get-credential <id> --reveal --lease-token <token>` — 结合 pick-web 返回的 lease 解密 `local_encrypted` 等路径。
12. **业务 skill 同进程登录**:`from service.rpa_helpers import ensure_logged_in` — 详见 [`references/RPA_HELPERS.md`](references/RPA_HELPERS.md)。
13. **有则用、无则建(网页账号)**:`account ensure-web --platform <p>` — 登记账号 ≠ 已登录;登录仍由业务技能 / `ensure_logged_in` 处理。
1. **安装/升级后初始化库与数据管理元数据**:`init-db`(宿主会自动调用;也可手工执行)
2. **登记网页/RPA 账号**:`account add-web` — 平台 + 登录标识 + 自动生成 profile_dir;可选 `--auth-strategy` / `--session-persistent` / `--device-fingerprint`。数据管理「新增」已绑定该能力(`replaces: create`),勿用宿主直写库新建账号。
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>`)— **本进程**直接拉起系统 Chrome/Edge(`--user-data-dir`),与「打开文件位置」同模式;不做登录判定。数据管理行内按钮均为 `interaction: direct` + `sync`。
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` 处理。
## CLI 调用

View File

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

138
assets/actions.json Normal file
View File

@@ -0,0 +1,138 @@
{
"schemaVersion": 1,
"skill": "account-manager",
"actions": [
{
"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": "平台",
"description": "关联平台信息表;宿主按列 lookup 渲染下拉"
},
"account_label": {
"type": "string",
"title": "账号名称"
},
"login_id": {
"type": "string",
"title": "登录账号"
},
"login_id_type": {
"type": "string",
"title": "登录标识类型",
"default": "unknown"
},
"tenant_id": {
"type": "string",
"title": "租户编号"
},
"environment": {
"type": "string",
"title": "环境",
"default": "production"
},
"role": {
"type": "string",
"title": "角色",
"default": "default"
},
"provider_code": {
"type": "string",
"title": "供应商代码"
},
"url": {
"type": "string",
"title": "入口链接"
}
}
},
"entrypoint": {
"type": "cli",
"command": "account",
"args": [
"add-web",
"--platform", "{{platform_key}}",
"--label", "{{account_label}}",
"--login-id", "{{login_id}}",
"--login-id-type", "{{login_id_type}}",
"--tenant-id", "{{tenant_id}}",
"--environment", "{{environment}}",
"--role", "{{role}}",
"--provider-code", "{{provider_code}}",
"--url", "{{url}}"
]
}
},
{
"id": "reveal-profile",
"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",
"description": "accounts 表主键",
"readOnly": true
}
}
},
"entrypoint": {
"type": "cli",
"command": "account",
"args": ["reveal-profile", "{{accountId}}"]
}
},
{
"id": "open-browser",
"label": "打开浏览器",
"description": "用该账号持久化 Profile 直接打开系统 Chrome/Edge,仅供肉眼核对(不做登录判定)。与打开文件位置同为一点即开。",
"placements": ["row"],
"executionProfile": "sync",
"interaction": "direct",
"bind": {
"tables": ["accounts"],
"inputMapping": { "accountId": "$row.$pk" }
},
"inputSchema": {
"type": "object",
"required": ["accountId"],
"properties": {
"accountId": {
"type": "number",
"title": "账号 ID",
"description": "accounts 表主键",
"readOnly": true
}
}
},
"entrypoint": {
"type": "cli",
"command": "account",
"args": ["open", "{{accountId}}"]
}
}
]
}

View File

@@ -17,6 +17,12 @@
## 子命令速查(按业务流程)
### 0. 运行时初始化(宿主标准)
| CLI | 说明 |
|---|---|
| `init-db` | 幂等初始化本地 SQLite、schema 迁移与数据管理中文元数据(`_jiangchang_*`)。成功输出 JSON:`{"ok":true,"skill":"account-manager","db_path":"..."}`。**匠厂安装/升级技能时会调用本命令**;业务命令内部也会触发同等 `init_db()`。 |
### A. 平台管理
| CLI | 说明 |
@@ -48,7 +54,8 @@
| `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(`--user-data-dir=profile`;无额外 python 子进程) |
| `account reveal-profile <id>` | 在系统文件管理器中打开该账号 `profile_dir` |
### D. 加密备份与迁移(v2)
@@ -69,6 +76,7 @@
| `lease list [--active]` | 列出租约 |
| `lease cleanup` | 清理过期租约 |
| `account open <id>` / `open <id>` | 浏览器查看 |
| `account reveal-profile <id>` | 打开 Profile 目录 |
### 兼容入口(旧版别名)

View File

@@ -87,6 +87,7 @@
|---|---|
| `ERROR:需要 playwright Python 包` | 未安装 `playwright` 包或本机无 Chrome/Edge |
| `ERROR:PROFILE_DIR_MISSING` | 账号记录中 `profile_dir` 为空 |
| `ERROR:REVEAL_PROFILE_FAILED` | `account reveal-profile` 无法打开目录(系统调用失败) |
---

View File

@@ -4,7 +4,7 @@ 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
account add-web|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
@@ -38,6 +38,7 @@ from service.account_service import (
cmd_delete_by_platform,
cmd_delete_by_platform_phone,
cmd_get,
cmd_init_db,
cmd_lease_cleanup,
cmd_lease_list,
cmd_lease_release,
@@ -48,7 +49,7 @@ from service.account_service import (
cmd_platform_get,
cmd_platform_list,
)
from service.browser_service import cmd_open
from service.browser_service import cmd_open, cmd_reveal_profile
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
from util.logging_config import get_skill_logger, setup_skill_logging
from util.platforms import _platform_list_cn_for_help, resolve_platform_key
@@ -60,6 +61,9 @@ from util.runtime_paths import _maybe_print_paths_debug_cli
_USAGE = """account-manager v2 — Credential & Browser Profile Manager
运行时 / 数据管理:
init-db # 幂等初始化本地库与数据管理中文元数据(宿主安装/升级会调用)
平台管理:
platform list [--domain logistics|llm|content]
platform get <platform>
@@ -100,6 +104,7 @@ _USAGE = """account-manager v2 — Credential & Browser Profile Manager
account mark-error <id> --code <code> --message <msg>
account mark-used <id>
account open <id>
account reveal-profile <id>
account init-master-key [--print|--from-env|--rotate]
account export-credentials --plaintext
account import-credentials [--replace-all] < backup.json
@@ -114,8 +119,9 @@ _USAGE = """account-manager v2 — Credential & Browser Profile Manager
lease list [--active]
lease cleanup
浏览器:
浏览器 / Profile:
account open <id>
account reveal-profile <id> # 在文件管理器中打开 profile_dir
open <id> # 兼容别名
兼容入口(旧版):
@@ -217,6 +223,10 @@ def main() -> None:
# Determine if first arg is a platform key (for legacy shorthand)
plat_from_first = resolve_platform_key(cmd)
# ---- init-db(宿主 install/upgrade 标准入口)----
if cmd == "init-db":
sys.exit(cmd_init_db())
# ---- platform subcommand ----
if cmd == "platform":
if len(av) < 3:
@@ -314,6 +324,11 @@ def main() -> None:
print("ERROR:CLI_BAD_ARGS account open 需要 <id>。")
sys.exit(1)
cmd_open(av[3])
elif sub == "reveal-profile":
if len(av) < 4:
print("ERROR:CLI_BAD_ARGS account reveal-profile 需要 <id>。")
sys.exit(1)
cmd_reveal_profile(av[3])
elif sub == "init-master-key":
cmd_account_init_master_key(av)
elif sub == "export-credentials":

View File

@@ -14,6 +14,7 @@ import sqlite3
import time
from typing import Optional
from db.display_metadata import init_display_metadata
from db.schema import SCHEMA_VERSION, ensure_schema
from util.constants import SKILL_SLUG
from util.logging_config import get_skill_logger
@@ -85,6 +86,7 @@ def init_db() -> None:
ensure_schema(conn)
cur = conn.cursor()
init_display_metadata(cur)
cur.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
conn.commit()

View File

@@ -0,0 +1,460 @@
"""匠厂数据管理展示元数据(_jiangchang_tables / _jiangchang_columns)。
SQLite 无原生表/字段 COMMENT;中文展示名称通过这两张元数据表提供给宿主「数据管理」页面。
字段在界面上的顺序由 CREATE TABLE / PRAGMA table_info(cid) 决定,**不**使用 display_order 排序。
"""
from __future__ import annotations
import json
from typing import Iterable
INTERNAL_METADATA_TABLES = frozenset({"_jiangchang_tables", "_jiangchang_columns"})
LEGACY_METADATA_TABLES = frozenset({"_schema_meta"})
METADATA_TABLES = INTERNAL_METADATA_TABLES | LEGACY_METADATA_TABLES
PLATFORMS_TABLE = "platforms"
ACCOUNTS_TABLE = "accounts"
CREDENTIALS_TABLE = "credentials"
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,
)
USER_BUSINESS_TABLES = (
PLATFORMS_TABLE,
ACCOUNTS_TABLE,
CREDENTIALS_TABLE,
ACCOUNT_LEASES_TABLE,
)
_JIANGCHANG_TABLES_DDL = """
CREATE TABLE IF NOT EXISTS _jiangchang_tables (
table_name TEXT PRIMARY KEY,
display_name TEXT NOT NULL,
description TEXT,
sort_order INTEGER,
visible INTEGER NOT NULL DEFAULT 1,
readonly INTEGER NOT NULL DEFAULT 0
)
"""
_JIANGCHANG_COLUMNS_DDL = """
CREATE TABLE IF NOT EXISTS _jiangchang_columns (
table_name TEXT NOT NULL,
column_name TEXT NOT NULL,
display_name TEXT NOT NULL,
description TEXT,
display_order INTEGER,
visible INTEGER NOT NULL DEFAULT 1,
searchable INTEGER NOT NULL DEFAULT 1,
editable INTEGER NOT NULL DEFAULT 1,
display_type TEXT,
options_json TEXT,
PRIMARY KEY (table_name, column_name)
)
"""
def _col(
column_name: str,
display_name: str,
*,
description: str | None = None,
visible: int = 1,
searchable: int = 1,
editable: int = 1,
display_type: str | None = None,
options_json: str | None = None,
) -> dict[str, object]:
return {
"column_name": column_name,
"display_name": display_name,
"description": description,
"visible": visible,
"searchable": searchable,
"editable": editable,
"display_type": display_type,
"options_json": options_json,
}
def _ts(column_name: str, display_name: str, description: str) -> dict[str, object]:
return _col(
column_name,
display_name,
description=description,
searchable=0,
editable=0,
display_type=DATETIME_UNIX_SECONDS,
)
PLATFORMS_TABLE_METADATA = {
"table_name": PLATFORMS_TABLE,
"display_name": "平台信息",
"description": "已注册平台(物流、LLM、内容等)及默认登录策略",
"sort_order": 1,
"visible": 1,
"readonly": 0,
}
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("provider_code", "供应商代码", description="供应商或平台代码,可空"),
_col("default_url", "默认链接", description="平台默认入口 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"),
_ts("created_at", "创建时间", "记录创建时间(Unix 秒级时间戳)"),
_ts("updated_at", "更新时间", "记录最后更新时间(Unix 秒级时间戳)"),
)
ACCOUNTS_TABLE_METADATA = {
"table_name": ACCOUNTS_TABLE,
"display_name": "账号信息",
"description": "网页/RPA 账号身份与浏览器 Profile 路径",
"sort_order": 2,
"visible": 1,
"readonly": 0,
}
ACCOUNTS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
_col("id", "编号", description="账号唯一编号", editable=0),
_col(
"platform_key",
"平台",
description="所属平台键(来自平台信息表)",
display_type="lookup",
options_json=PLATFORM_KEY_LOOKUP_JSON,
),
_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("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 等"),
_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 策略专用,可空"),
_ts("created_at", "创建时间", "记录创建时间(Unix 秒级时间戳)"),
_ts("updated_at", "更新时间", "记录最后更新时间(Unix 秒级时间戳)"),
)
CREDENTIALS_TABLE_METADATA = {
"table_name": CREDENTIALS_TABLE,
"display_name": "凭证信息",
"description": "API Key、Token、加密密码等凭据元数据(不含明文)",
"sort_order": 3,
"visible": 1,
"readonly": 0,
}
CREDENTIALS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
_col("id", "编号", description="凭据唯一编号", editable=0),
_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),
_col("secret_ref", "密钥引用", description="环境变量名或凭据管理器 target", editable=0, searchable=0),
_col("secret_mask", "密钥掩码", description="打码后的 secret 展示", editable=0, searchable=0),
_col(
"secret_ciphertext",
"密钥密文",
description="local_encrypted 时的 Fernet 密文,不可手工编辑",
editable=0,
searchable=0,
),
_ts("expires_at", "过期时间", "凭据过期时间,可空"),
_col("status", "状态", description="active / disabled / expired 等"),
_ts("last_used_at", "最后使用时间", "最近一次使用时间"),
_col("extra_json", "扩展信息", description="扩展 JSON 配置", editable=0),
_ts("created_at", "创建时间", "记录创建时间(Unix 秒级时间戳)"),
_ts("updated_at", "更新时间", "记录最后更新时间(Unix 秒级时间戳)"),
)
ACCOUNT_LEASES_TABLE_METADATA = {
"table_name": ACCOUNT_LEASES_TABLE,
"display_name": "账号租约",
"description": "RPA 并发租约,防止多任务同时使用同一浏览器 Profile",
"sort_order": 4,
"visible": 1,
"readonly": 1,
}
ACCOUNT_LEASES_COLUMN_METADATA: tuple[dict[str, object], ...] = (
_col("id", "编号", description="租约唯一编号", editable=0),
_col("account_id", "账号编号", description="关联的 accounts.id", editable=0),
_col("lease_token", "租约令牌", description="用于 release 的随机 token", editable=0, searchable=1),
_col("holder", "持有者", description="持有租约的业务技能或任务标识", editable=0),
_col("purpose", "用途", description="租约用途说明", editable=0),
_ts("expires_at", "过期时间", "租约过期时间"),
_ts("heartbeat_at", "心跳时间", "最近一次心跳时间"),
_ts("released_at", "释放时间", "主动释放时间,可空"),
_col("status", "状态", description="active / released / expired", editable=0),
_ts("created_at", "创建时间", "记录创建时间(Unix 秒级时间戳)"),
_ts("updated_at", "更新时间", "记录最后更新时间(Unix 秒级时间戳)"),
)
_TABLE_SEEDS: tuple[tuple[dict[str, object], tuple[dict[str, object], ...]], ...] = (
(PLATFORMS_TABLE_METADATA, PLATFORMS_COLUMN_METADATA),
(ACCOUNTS_TABLE_METADATA, ACCOUNTS_COLUMN_METADATA),
(CREDENTIALS_TABLE_METADATA, CREDENTIALS_COLUMN_METADATA),
(ACCOUNT_LEASES_TABLE_METADATA, ACCOUNT_LEASES_COLUMN_METADATA),
)
def init_display_metadata_tables(cursor) -> None:
cursor.execute(_JIANGCHANG_TABLES_DDL)
cursor.execute(_JIANGCHANG_COLUMNS_DDL)
def upsert_table_metadata(
cursor,
*,
table_name: str,
display_name: str,
description: str | None = None,
sort_order: int | None = None,
visible: int = 1,
readonly: int = 0,
) -> None:
cursor.execute(
"""
INSERT INTO _jiangchang_tables (
table_name, display_name, description, sort_order, visible, readonly
)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(table_name) DO UPDATE SET
display_name = excluded.display_name,
description = excluded.description,
sort_order = excluded.sort_order,
visible = excluded.visible,
readonly = excluded.readonly
""",
(table_name, display_name, description, sort_order, visible, readonly),
)
def upsert_column_metadata(
cursor,
*,
table_name: str,
column_name: str,
display_name: str,
description: str | None = None,
visible: int = 1,
searchable: int = 1,
editable: int = 1,
display_type: str | None = None,
options_json: str | None = None,
) -> None:
cursor.execute(
"""
INSERT INTO _jiangchang_columns (
table_name, column_name, display_name, description,
display_order, visible, searchable, editable, display_type, options_json
)
VALUES (?, ?, ?, ?, NULL, ?, ?, ?, ?, ?)
ON CONFLICT(table_name, column_name) DO UPDATE SET
display_name = excluded.display_name,
description = excluded.description,
display_order = NULL,
visible = excluded.visible,
searchable = excluded.searchable,
editable = excluded.editable,
display_type = excluded.display_type,
options_json = excluded.options_json
""",
(
table_name,
column_name,
display_name,
description,
visible,
searchable,
editable,
display_type,
options_json,
),
)
def seed_column_metadata_batch(
cursor,
*,
table_name: str,
columns: Iterable[dict[str, object]],
valid_column_names: Iterable[str] | None = None,
) -> None:
for column in columns:
upsert_column_metadata(
cursor,
table_name=table_name,
column_name=str(column["column_name"]),
display_name=str(column["display_name"]),
description=str(column["description"]) if column.get("description") else None,
visible=int(column.get("visible", 1)),
searchable=int(column.get("searchable", 1)),
editable=int(column.get("editable", 1)),
display_type=str(column["display_type"]) if column.get("display_type") else None,
options_json=str(column["options_json"]) if column.get("options_json") else None,
)
if valid_column_names is not None:
prune_stale_column_metadata(cursor, table_name, valid_column_names)
def prune_stale_column_metadata(cursor, table_name: str, valid_column_names: Iterable[str]) -> None:
valid = set(valid_column_names)
cursor.execute(
"SELECT column_name FROM _jiangchang_columns WHERE table_name = ?",
(table_name,),
)
for (column_name,) in cursor.fetchall():
if str(column_name) not in valid:
cursor.execute(
"DELETE FROM _jiangchang_columns WHERE table_name = ? AND column_name = ?",
(table_name, column_name),
)
def get_table_readonly(cursor, table_name: str) -> int | None:
cursor.execute(
"SELECT readonly FROM _jiangchang_tables WHERE table_name = ?",
(table_name,),
)
row = cursor.fetchone()
if not row:
return None
return int(row[0] or 0)
def get_column_editable(cursor, table_name: str, column_name: str) -> int | None:
cursor.execute(
"""
SELECT editable FROM _jiangchang_columns
WHERE table_name = ? AND column_name = ?
""",
(table_name, column_name),
)
row = cursor.fetchone()
if not row:
return None
return int(row[0] or 0)
def list_editable_business_columns(cursor, table_name: str) -> list[str]:
physical = list_physical_column_names(cursor, table_name)
cursor.execute(
"""
SELECT column_name, editable
FROM _jiangchang_columns
WHERE table_name = ?
""",
(table_name,),
)
editable_set = {
str(column_name)
for column_name, editable in cursor.fetchall()
if editable is not None and int(editable) == 1
}
return [name for name in physical if name in editable_set]
def table_allows_manual_create(cursor, table_name: str) -> bool:
readonly = get_table_readonly(cursor, table_name)
if readonly is None or int(readonly) != 0:
return False
return bool(list_editable_business_columns(cursor, table_name))
def seed_business_table_display_metadata(cursor, table_meta: dict[str, object], columns: tuple[dict[str, object], ...]) -> None:
table_name = str(table_meta["table_name"])
upsert_table_metadata(
cursor,
table_name=table_name,
display_name=str(table_meta["display_name"]),
description=str(table_meta["description"]) if table_meta.get("description") else None,
sort_order=int(table_meta["sort_order"]) if table_meta.get("sort_order") is not None else None,
visible=int(table_meta.get("visible", 1)),
readonly=int(table_meta.get("readonly", 0)),
)
seed_column_metadata_batch(
cursor,
table_name=table_name,
columns=columns,
valid_column_names=[str(column["column_name"]) for column in columns],
)
def init_display_metadata(cursor) -> None:
init_display_metadata_tables(cursor)
for table_meta, columns in _TABLE_SEEDS:
seed_business_table_display_metadata(cursor, table_meta, columns)
def iter_user_tables(cursor) -> Iterable[str]:
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
)
for (name,) in cursor.fetchall():
if name in METADATA_TABLES:
continue
yield str(name)
def _quote_identifier(name: str) -> str:
return '"' + name.replace('"', '""') + '"'
def list_physical_column_names(cursor, table_name: str) -> list[str]:
cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})")
rows = cursor.fetchall()
rows.sort(key=lambda row: int(row[0]))
return [str(row[1]) for row in rows]

View File

@@ -39,7 +39,12 @@ from service.legacy_commands import (
cmd_list,
)
from service.lease_commands import cmd_lease_cleanup, cmd_lease_list, cmd_lease_release
from service.platform_commands import cmd_platform_ensure, cmd_platform_get, cmd_platform_list
from service.platform_commands import (
cmd_init_db,
cmd_platform_ensure,
cmd_platform_get,
cmd_platform_list,
)
__all__ = [
"PlatformCallerMeta",
@@ -60,6 +65,7 @@ __all__ = [
"cmd_delete_by_platform",
"cmd_delete_by_platform_phone",
"cmd_get",
"cmd_init_db",
"cmd_lease_cleanup",
"cmd_lease_list",
"cmd_lease_release",

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,12 @@ 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 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:
@@ -114,10 +106,63 @@ def resolve_account_browser_url(account: dict, conn=None) -> str:
return (PLATFORMS.get(pk, {}).get("default_url") or "").strip()
def cmd_reveal_profile(account_id):
"""
在系统文件管理器中打开该账号的 profile_dir。
不写数据库、不启动浏览器。
"""
log = get_skill_logger()
init_db()
log.info("reveal_profile_start account_id=%r", account_id)
target = get_account_by_id(account_id)
if not target:
log.warning("reveal_profile_error account_id=%r not_found", account_id)
print("ERROR:ACCOUNT_NOT_FOUND")
sys.exit(1)
profile_dir = (target.get("profile_dir") or "").strip()
if not profile_dir:
log.warning("reveal_profile_error account_id=%s profile_dir_missing", account_id)
print("ERROR:PROFILE_DIR_MISSING 账号中 profile_dir 为空。")
sys.exit(1)
profile_abs = os.path.abspath(profile_dir)
os.makedirs(profile_abs, exist_ok=True)
try:
if sys.platform == "win32":
# explorer 即使成功也常返回非 0;用 startfile 打开目录更稳妥
os.startfile(profile_abs) # type: ignore[attr-defined]
elif sys.platform == "darwin":
subprocess.run(["open", profile_abs], check=False)
else:
subprocess.run(["xdg-open", profile_abs], check=False)
except OSError as exc:
log.error("reveal_profile_error account_id=%s error=%s", account_id, str(exc))
print(f"ERROR:REVEAL_PROFILE_FAILED {exc}")
sys.exit(1)
log.info("reveal_profile_ok account_id=%s profile_dir=%s", account_id, profile_abs)
print(
json.dumps(
{
"success": True,
"data": {
"account_id": int(target["id"]) if target.get("id") is not None else account_id,
"profile_dir": profile_abs,
},
},
ensure_ascii=False,
)
)
def cmd_open(account_id):
"""
打开该账号的持久化浏览器,仅用于肉眼核对。
不写数据库、不判定是否已登录。
与 reveal-profile 同模式:本进程直接拉起系统 Chrome/Edge 后立即返回。
不写数据库、不判定是否已登录、不起 Playwright / 额外 python 子进程。
"""
log = get_skill_logger()
init_db()
@@ -127,27 +172,25 @@ def cmd_open(account_id):
if not target:
log.warning("browser_open_error account_id=%r not_found", account_id)
print("ERROR:ACCOUNT_NOT_FOUND")
return
sys.exit(1)
channel = resolve_chromium_channel()
if not channel:
_print_browser_install_hint()
return
try:
import playwright # noqa: F401
except ImportError:
print(
"ERROR:需要 playwright Python 包。"
)
return
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 为空。")
return
os.makedirs(profile_dir, exist_ok=True)
sys.exit(1)
profile_abs = os.path.abspath(profile_dir)
os.makedirs(profile_abs, exist_ok=True)
conn = get_conn()
try:
@@ -155,37 +198,64 @@ def cmd_open(account_id):
finally:
conn.close()
# 与 profile_shortcut 一致:独立 user-data-dir 保存登录态
cmd = [browser_exe, f"--user-data-dir={profile_abs}"]
if url:
cmd.append(url)
log.info(
"browser_open_context profile_dir=%s url=%s channel=%s platform=%s",
profile_dir,
"browser_open_launch exe=%s profile_dir=%s url=%s platform=%s",
browser_exe,
profile_abs,
url,
channel,
target.get("platform_key"),
)
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
print(f"正在打开 [{target['account_label']}] 的 {browser_name}(仅查看,不写入数据库)")
print(f"地址:{url}")
print("本命令不做登录检测。")
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:
subprocess.run(
[sys.executable, "-m", "service.open_child_runner", cfg_path],
cwd=_SCRIPTS_ROOT,
env=subprocess_env_with_trace(),
# 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_ok account_id=%s browser=%s pid=%s",
account_id,
browser_name,
child.pid,
)
print(
json.dumps(
{
"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,
"browser_exe": browser_exe,
"profile_dir": profile_abs,
"url": url,
"pid": child.pid,
},
},
ensure_ascii=False,
)
log.info("browser_open_close account_id=%s channel=%s", account_id, channel)
except Exception as e:
log.error("browser_open_error account_id=%s error=%s", account_id, str(e))
raise
finally:
try:
os.unlink(cfg_path)
except OSError:
pass
)

View File

@@ -1,46 +0,0 @@
"""Playwright 仅打开浏览器子进程:由 browser_service.cmd_open 启动,argv[1] 为 JSON 配置路径。"""
import json
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():
with open(sys.argv[1], encoding="utf-8") as f:
c = json.load(f)
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

@@ -6,8 +6,38 @@ from typing import Optional
from db.accounts_repo import get_platform_from_db, list_platforms_from_db, upsert_platform
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.runtime_paths import get_db_path
def cmd_init_db() -> int:
"""幂等创建/迁移本地 SQLite 与数据管理元数据(宿主 install/upgrade 可调用)。"""
log = get_skill_logger()
log.info("cli_start subcommand=init_db")
try:
init_db()
except Exception as exc: # noqa: BLE001 — CLI 边界,统一成可机读失败
log.exception("init_db_failure")
print(
json.dumps(
{"ok": False, "skill": SKILL_SLUG, "error": str(exc)},
ensure_ascii=False,
)
)
return 1
print(
json.dumps(
{
"ok": True,
"skill": SKILL_SLUG,
"db_path": get_db_path(),
},
ensure_ascii=False,
)
)
return 0
def cmd_platform_list(domain: Optional[str] = None) -> None:

View File

@@ -2,6 +2,9 @@
"""
Browser launch helper checks (no real browser started).
"""
import json
from unittest import mock
import pytest
@@ -14,24 +17,68 @@ class TestPlaywrightLaunchHelpers:
args, _ignore = persistent_context_launch_parts()
assert "--start-maximized" in args
def test_open_child_runner_launch_keywords_documented(self):
"""launch_persistent_context kwargs contract (static read of source)."""
import os
path = os.path.join(
os.path.dirname(__file__),
"..",
"scripts",
"service",
"open_child_runner.py",
class TestCmdOpenDirectBrowser:
"""account open = same-process Chrome/Edge launch (like reveal-profile)."""
def test_cmd_open_launches_browser_exe(self, clean_db, tmp_path, capsys, monkeypatch):
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",
account_label="open-test",
login_id="1",
login_id_type="phone",
tenant_id=None,
environment="production",
role="default",
provider_code="sohu",
profile_dir=str(profile),
url="https://example.com",
extra_json=None,
now=1_700_000_000,
auth_strategy="password_auto",
)
path = os.path.abspath(path)
with open(path, encoding="utf-8") as f:
src = f.read()
assert "launch_persistent_context" in src
assert "headless=False" in src
assert "no_viewport=True" in src
assert '"channel"' in src or "'channel'" in src
clean_db.commit()
fake_exe = str(tmp_path / "chrome.exe")
monkeypatch.setattr(
browser_service,
"resolve_chromium_executable",
lambda: fake_exe,
)
class FakeProc:
pid = 5252
spawned = {}
def fake_popen(cmd, **kwargs):
spawned["cmd"] = list(cmd)
spawned["kwargs"] = kwargs
return FakeProc()
monkeypatch.setattr(browser_service.subprocess, "Popen", fake_popen)
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"]["pid"] == 5252
assert payload["data"]["browser_exe"] == fake_exe
assert spawned["cmd"][0] == fake_exe
assert spawned["cmd"][1].startswith("--user-data-dir=")
assert spawned["cmd"][2] == "https://example.com"
# must NOT spawn python / open_child_runner
joined = " ".join(spawned["cmd"])
assert "open_child_runner" not in joined
assert "python" not in joined.lower()
class TestResolveChromiumChannel:

View File

@@ -0,0 +1,301 @@
# -*- coding: utf-8 -*-
"""数据管理展示元数据:幂等初始化、中文名称、敏感字段权限。"""
from __future__ import annotations
import re
import pytest
from db.display_metadata import (
ACCOUNT_LEASES_TABLE,
ACCOUNTS_TABLE,
CREDENTIALS_TABLE,
PLATFORMS_TABLE,
USER_BUSINESS_TABLES,
get_table_readonly,
init_display_metadata_tables,
list_physical_column_names,
table_allows_manual_create,
upsert_column_metadata,
upsert_table_metadata,
)
def _has_cjk(text: str) -> bool:
return bool(re.search(r"[\u4e00-\u9fff]", text or ""))
@pytest.fixture
def metadata_db(clean_db):
"""clean_db already ran init_db(); return connection for metadata assertions."""
yield clean_db
def test_init_db_metadata_idempotent(metadata_db):
from db.connection import get_conn, init_db
init_db()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM _jiangchang_tables")
count_after_second = int(cur.fetchone()[0])
assert count_after_second == len(USER_BUSINESS_TABLES)
finally:
conn.close()
@pytest.mark.parametrize(
"table_name,expected_display",
[
(PLATFORMS_TABLE, "平台信息"),
(ACCOUNTS_TABLE, "账号信息"),
(CREDENTIALS_TABLE, "凭证信息"),
(ACCOUNT_LEASES_TABLE, "账号租约"),
],
)
def test_table_chinese_display_names(metadata_db, table_name, expected_display):
cur = metadata_db.cursor()
cur.execute(
"SELECT display_name FROM _jiangchang_tables WHERE table_name = ?",
(table_name,),
)
row = cur.fetchone()
assert row is not None
assert row[0] == expected_display
def test_all_business_columns_have_chinese_display_names(metadata_db):
cur = metadata_db.cursor()
for table_name in USER_BUSINESS_TABLES:
physical = set(list_physical_column_names(cur, table_name))
cur.execute(
"""
SELECT column_name, display_name
FROM _jiangchang_columns
WHERE table_name = ?
""",
(table_name,),
)
meta_rows = cur.fetchall()
assert len(meta_rows) == len(physical), f"{table_name} 元数据字段数与物理表不一致"
for column_name, display_name in meta_rows:
assert column_name in physical
assert _has_cjk(str(display_name)), f"{table_name}.{column_name} 缺少中文 display_name"
def test_sensitive_credential_fields_not_editable(metadata_db):
cur = metadata_db.cursor()
sensitive = ("secret_ref", "secret_mask", "secret_ciphertext", "secret_storage")
for column_name in sensitive:
cur.execute(
"""
SELECT editable, searchable
FROM _jiangchang_columns
WHERE table_name = ? AND column_name = ?
""",
(CREDENTIALS_TABLE, column_name),
)
editable, searchable = cur.fetchone()
assert int(editable) == 0, column_name
if column_name in ("secret_ref", "secret_ciphertext"):
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(
"""
SELECT editable FROM _jiangchang_columns
WHERE table_name = ? AND column_name = 'lease_token'
""",
(ACCOUNT_LEASES_TABLE,),
)
assert int(cur.fetchone()[0]) == 0
def test_account_leases_readonly(metadata_db):
cur = metadata_db.cursor()
assert int(get_table_readonly(cur, ACCOUNT_LEASES_TABLE)) == 1
assert table_allows_manual_create(cur, ACCOUNT_LEASES_TABLE) is False
def test_platforms_key_not_editable(metadata_db):
cur = metadata_db.cursor()
cur.execute(
"""
SELECT editable FROM _jiangchang_columns
WHERE table_name = ? AND column_name = 'platform_key'
""",
(PLATFORMS_TABLE,),
)
assert int(cur.fetchone()[0]) == 0
def test_profile_dir_not_editable(metadata_db):
cur = metadata_db.cursor()
cur.execute(
"""
SELECT editable FROM _jiangchang_columns
WHERE table_name = ? AND column_name = 'profile_dir'
""",
(ACCOUNTS_TABLE,),
)
assert int(cur.fetchone()[0]) == 0
def test_timestamp_columns_use_datetime_display_type(metadata_db):
cur = metadata_db.cursor()
for table_name in USER_BUSINESS_TABLES:
cur.execute(
"""
SELECT column_name, display_type, editable
FROM _jiangchang_columns
WHERE table_name = ? AND column_name IN ('created_at', 'updated_at')
""",
(table_name,),
)
for column_name, display_type, editable in cur.fetchall():
assert display_type == "datetime_unix_seconds", f"{table_name}.{column_name}"
assert int(editable) == 0, f"{table_name}.{column_name}"
def test_metadata_tables_hidden_from_user_tables(metadata_db):
from db.display_metadata import iter_user_tables
tables = list(iter_user_tables(metadata_db.cursor()))
assert "_jiangchang_tables" not in tables
assert "_jiangchang_columns" not in tables
assert "_schema_meta" not in tables
for name in USER_BUSINESS_TABLES:
assert name in tables
def test_reinit_clears_stale_column_metadata(metadata_db):
from db.connection import get_conn, init_db
conn = get_conn()
try:
conn.execute(
"""
INSERT INTO _jiangchang_columns
(table_name, column_name, display_name, visible, searchable, editable)
VALUES ('accounts', 'stale_field', '陈旧字段', 1, 1, 0)
"""
)
conn.commit()
finally:
conn.close()
init_db()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT COUNT(*) FROM _jiangchang_columns
WHERE table_name = 'accounts' AND column_name = 'stale_field'
"""
)
assert int(cur.fetchone()[0]) == 0
finally:
conn.close()
def test_missing_column_metadata_detected_in_memory_db():
import sqlite3
from db.display_metadata import init_display_metadata, list_physical_column_names
conn = sqlite3.connect(":memory:")
try:
conn.executescript(
"""
CREATE TABLE platforms (
platform_key TEXT PRIMARY KEY,
display_name TEXT NOT NULL
);
"""
)
init_display_metadata_tables(conn.cursor())
upsert_table_metadata(
conn.cursor(),
table_name="platforms",
display_name="平台信息",
)
upsert_column_metadata(
conn.cursor(),
table_name="platforms",
column_name="platform_key",
display_name="平台键",
editable=0,
)
conn.commit()
physical = list_physical_column_names(conn.cursor(), "platforms")
assert "display_name" in physical
cur = conn.cursor()
cur.execute(
"SELECT COUNT(*) FROM _jiangchang_columns WHERE table_name = 'platforms'"
)
assert int(cur.fetchone()[0]) < len(physical)
finally:
conn.close()
# Full seed should cover all physical columns
conn = sqlite3.connect(":memory:")
try:
from db.schema import ensure_schema
ensure_schema(conn)
init_display_metadata(conn.cursor())
conn.commit()
for table_name in USER_BUSINESS_TABLES:
physical = set(list_physical_column_names(conn.cursor(), table_name))
cur = conn.cursor()
cur.execute(
"SELECT column_name FROM _jiangchang_columns WHERE table_name = ?",
(table_name,),
)
meta_cols = {row[0] for row in cur.fetchall()}
assert meta_cols == physical, table_name
finally:
conn.close()

View File

@@ -0,0 +1,75 @@
# -*- coding: utf-8 -*-
"""account reveal-profile:打开 profile 目录(不启动浏览器)。"""
from __future__ import annotations
import json
import os
from unittest import mock
import pytest
def _insert_account(conn, *, profile_dir: str) -> int:
from db.accounts_repo import insert_account
from util.platforms import seed_platforms
seed_platforms(conn)
aid = insert_account(
conn,
platform_key="sohu",
account_label="reveal-test",
login_id="13800000000",
login_id_type="phone",
tenant_id=None,
environment="production",
role="default",
provider_code="sohu",
profile_dir=profile_dir,
url="https://mp.sohu.com",
extra_json=None,
now=1_700_000_000,
auth_strategy="password_auto",
session_persistent=1,
device_fingerprint=None,
)
conn.commit()
return aid
def test_reveal_profile_opens_dir(clean_db, tmp_path, capsys):
from service.browser_service import cmd_reveal_profile
profile = tmp_path / "profiles" / "sohu" / "reveal-test"
aid = _insert_account(clean_db, profile_dir=str(profile))
with mock.patch("service.browser_service.os.startfile") as startfile:
with mock.patch("service.browser_service.sys.platform", "win32"):
cmd_reveal_profile(str(aid))
startfile.assert_called_once()
called_path = startfile.call_args[0][0]
assert os.path.isdir(called_path)
assert os.path.abspath(str(profile)) == called_path
out = capsys.readouterr().out.strip()
payload = json.loads(out.splitlines()[-1])
assert payload["success"] is True
assert payload["data"]["account_id"] == aid
assert payload["data"]["profile_dir"] == called_path
def test_reveal_profile_missing_account(clean_db):
from service.browser_service import cmd_reveal_profile
with pytest.raises(SystemExit) as ei:
cmd_reveal_profile("999999")
assert ei.value.code == 1
def test_reveal_profile_empty_dir(clean_db):
from service.browser_service import cmd_reveal_profile
aid = _insert_account(clean_db, profile_dir="")
with pytest.raises(SystemExit) as ei:
cmd_reveal_profile(str(aid))
assert ei.value.code == 1

View File

@@ -23,6 +23,7 @@ def test_usage_includes_all_account_subcommands():
assert r.returncode == 1
usage = r.stdout
for cmd in (
"init-db",
"init-master-key",
"export-credentials",
"import-credentials",
@@ -31,7 +32,49 @@ def test_usage_includes_all_account_subcommands():
"get-credential",
"pick-web",
"resolve",
"reveal-profile",
):
assert cmd in usage, f"USAGE 缺少 {cmd}"
assert "--ensure" in usage
assert "--login-id" in usage
def test_init_db_cli_succeeds(fake_env):
"""宿主调用的 init-db 必须成功并写入展示元数据。"""
import json
env = {
**os.environ,
"PYTHONPATH": _scripts,
"JIANGCHANG_DATA_ROOT": fake_env["data_root"],
"JIANGCHANG_USER_ID": fake_env["user_id"],
}
r = subprocess.run(
[sys.executable, "main.py", "init-db"],
capture_output=True,
text=True,
env=env,
cwd=_scripts,
)
assert r.returncode == 0, r.stdout + r.stderr
payload = json.loads(r.stdout.strip().splitlines()[-1])
assert payload.get("ok") is True
assert payload.get("skill") == "account-manager"
assert payload.get("db_path")
from db.connection import get_conn
conn = get_conn()
try:
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM _jiangchang_columns")
assert int(cur.fetchone()[0]) > 0
cur.execute(
"""
SELECT display_name FROM _jiangchang_columns
WHERE table_name = 'accounts' AND column_name = 'role'
"""
)
assert cur.fetchone()[0] == "角色"
finally:
conn.close()