Compare commits

...

6 Commits

Author SHA1 Message Date
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 1136 additions and 134 deletions

View File

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

View File

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

64
assets/actions.json Normal file
View File

@@ -0,0 +1,64 @@
{
"schemaVersion": 1,
"skill": "account-manager",
"actions": [
{
"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. 平台管理 ### A. 平台管理
| CLI | 说明 | | CLI | 说明 |
@@ -48,7 +54,8 @@
| `account mark-session <id> --session-status <status>` | 标记会话状态 | | `account mark-session <id> --session-status <status>` | 标记会话状态 |
| `account mark-error <id> --code <code> --message <msg>` | 标记错误 | | `account mark-error <id> --code <code> --message <msg>` | 标记错误 |
| `account mark-used <id>` | 标记已使用 | | `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) ### D. 加密备份与迁移(v2)
@@ -69,6 +76,7 @@
| `lease list [--active]` | 列出租约 | | `lease list [--active]` | 列出租约 |
| `lease cleanup` | 清理过期租约 | | `lease cleanup` | 清理过期租约 |
| `account open <id>` / `open <id>` | 浏览器查看 | | `account open <id>` / `open <id>` | 浏览器查看 |
| `account reveal-profile <id>` | 打开 Profile 目录 |
### 兼容入口(旧版别名) ### 兼容入口(旧版别名)

View File

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

View File

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

View File

@@ -0,0 +1,421 @@
"""匠厂数据管理展示元数据(_jiangchang_tables / _jiangchang_columns)。
SQLite 无原生表/字段 COMMENT;中文展示名称通过这两张元数据表提供给宿主「数据管理」页面。
字段在界面上的顺序由 CREATE TABLE / PRAGMA table_info(cid) 决定,**不**使用 display_order 排序。
"""
from __future__ import annotations
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"})
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="所属平台键"),
_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,可空"),
_col("platform_key", "平台", description="所属平台键"),
_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, 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_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__ = [ __all__ = [
"PlatformCallerMeta", "PlatformCallerMeta",
@@ -60,6 +65,7 @@ __all__ = [
"cmd_delete_by_platform", "cmd_delete_by_platform",
"cmd_delete_by_platform_phone", "cmd_delete_by_platform_phone",
"cmd_get", "cmd_get",
"cmd_init_db",
"cmd_lease_cleanup", "cmd_lease_cleanup",
"cmd_lease_list", "cmd_lease_list",
"cmd_lease_release", "cmd_lease_release",

View File

@@ -2,7 +2,7 @@
""" """
Chromium/Edge browser profile opener. 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 download Playwright Chromium.
Does NOT check login status. Does NOT check login status.
Does NOT write secret to SQLite. Does NOT write secret to SQLite.
@@ -11,20 +11,12 @@ import json
import os import os
import subprocess import subprocess
import sys import sys
import tempfile
from db.connection import get_conn, init_db from db.connection import get_conn, init_db
from db.accounts_repo import get_account_by_id, get_platform_from_db from db.accounts_repo import get_account_by_id, get_platform_from_db
from util.logging_config import ( from util.logging_config import get_skill_logger
get_skill_logger,
subprocess_env_with_trace,
)
from util.platforms import PLATFORMS 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): def _win_find_exe(candidates):
for p in 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() 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): def cmd_open(account_id):
""" """
打开该账号的持久化浏览器,仅用于肉眼核对。 打开该账号的持久化浏览器,仅用于肉眼核对。
不写数据库、不判定是否已登录。 与 reveal-profile 同模式:本进程直接拉起系统 Chrome/Edge 后立即返回。
不写数据库、不判定是否已登录、不起 Playwright / 额外 python 子进程。
""" """
log = get_skill_logger() log = get_skill_logger()
init_db() init_db()
@@ -127,27 +172,25 @@ def cmd_open(account_id):
if not target: if not target:
log.warning("browser_open_error account_id=%r not_found", account_id) log.warning("browser_open_error account_id=%r not_found", account_id)
print("ERROR:ACCOUNT_NOT_FOUND") print("ERROR:ACCOUNT_NOT_FOUND")
return sys.exit(1)
channel = resolve_chromium_channel() browser_exe = resolve_chromium_executable()
if not channel: if not browser_exe:
_print_browser_install_hint() channel = resolve_chromium_channel()
return if sys.platform == "win32" or not channel:
_print_browser_install_hint()
try: sys.exit(1)
import playwright # noqa: F401 # 非 Windows:退回 PATH 上的常见命令名
except ImportError: browser_exe = "google-chrome" if channel == "chrome" else "microsoft-edge"
print(
"ERROR:需要 playwright Python 包。"
)
return
profile_dir = (target.get("profile_dir") or "").strip() profile_dir = (target.get("profile_dir") or "").strip()
if not profile_dir: if not profile_dir:
log.warning("browser_open_error account_id=%s profile_dir_missing", account_id) log.warning("browser_open_error account_id=%s profile_dir_missing", account_id)
print("ERROR:PROFILE_DIR_MISSING 账号中 profile_dir 为空。") print("ERROR:PROFILE_DIR_MISSING 账号中 profile_dir 为空。")
return sys.exit(1)
os.makedirs(profile_dir, exist_ok=True)
profile_abs = os.path.abspath(profile_dir)
os.makedirs(profile_abs, exist_ok=True)
conn = get_conn() conn = get_conn()
try: try:
@@ -155,37 +198,64 @@ def cmd_open(account_id):
finally: finally:
conn.close() conn.close()
# 与 profile_shortcut 一致:独立 user-data-dir 保存登录态
cmd = [browser_exe, f"--user-data-dir={profile_abs}"]
if url:
cmd.append(url)
log.info( log.info(
"browser_open_context profile_dir=%s url=%s channel=%s platform=%s", "browser_open_launch exe=%s profile_dir=%s url=%s platform=%s",
profile_dir, browser_exe,
profile_abs,
url, url,
channel,
target.get("platform_key"), 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: try:
subprocess.run( # GUI 浏览器:不挂控制台黑框;不等待关闭(与 reveal 一致立刻返回)
[sys.executable, "-m", "service.open_child_runner", cfg_path], popen_kwargs: dict = {
cwd=_SCRIPTS_ROOT, "stdin": subprocess.DEVNULL,
env=subprocess_env_with_trace(), "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.accounts_repo import get_platform_from_db, list_platforms_from_db, upsert_platform
from db.connection import get_conn, init_db from db.connection import get_conn, init_db
from service._svc_common import _err_json, _ok_json, _resolve_or_fail, _say 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.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, 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: def cmd_platform_list(domain: Optional[str] = None) -> None:

View File

@@ -2,6 +2,9 @@
""" """
Browser launch helper checks (no real browser started). Browser launch helper checks (no real browser started).
""" """
import json
from unittest import mock
import pytest import pytest
@@ -14,24 +17,68 @@ class TestPlaywrightLaunchHelpers:
args, _ignore = persistent_context_launch_parts() args, _ignore = persistent_context_launch_parts()
assert "--start-maximized" in args 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( class TestCmdOpenDirectBrowser:
os.path.dirname(__file__), """account open = same-process Chrome/Edge launch (like reveal-profile)."""
"..",
"scripts", def test_cmd_open_launches_browser_exe(self, clean_db, tmp_path, capsys, monkeypatch):
"service", from db.accounts_repo import insert_account
"open_child_runner.py", 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) clean_db.commit()
with open(path, encoding="utf-8") as f:
src = f.read() fake_exe = str(tmp_path / "chrome.exe")
assert "launch_persistent_context" in src monkeypatch.setattr(
assert "headless=False" in src browser_service,
assert "no_viewport=True" in src "resolve_chromium_executable",
assert '"channel"' in src or "'channel'" in src 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: class TestResolveChromiumChannel:

View File

@@ -0,0 +1,263 @@
# -*- 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_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 assert r.returncode == 1
usage = r.stdout usage = r.stdout
for cmd in ( for cmd in (
"init-db",
"init-master-key", "init-master-key",
"export-credentials", "export-credentials",
"import-credentials", "import-credentials",
@@ -31,7 +32,49 @@ def test_usage_includes_all_account_subcommands():
"get-credential", "get-credential",
"pick-web", "pick-web",
"resolve", "resolve",
"reveal-profile",
): ):
assert cmd in usage, f"USAGE 缺少 {cmd}" assert cmd in usage, f"USAGE 缺少 {cmd}"
assert "--ensure" in usage assert "--ensure" in usage
assert "--login-id" 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()