feat: add requirements.txt with cryptography for host venv install

Co-authored-by: Cursor <cursoragent@cursor.com>

SKILL.md

@@ -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.0
+version: 2.0.7
 author: 深圳匠厂科技有限公司
 metadata:
   openclaw:

release.ps1

@@ -1,3 +1,44 @@
+# Vendored from jiangchang-platform-kit `tools/release.ps1`.
+# Source: https://git.jc2009.com/client-jiangchang/jiangchang-platform-kit/-/blob/main/tools/release.ps1
+# The PyPI wheel does not ship this script; sync periodically when upstream workflow changes.
+
+<#
+.SYNOPSIS
+ One-command release script for skill repos.
+
+.DESCRIPTION
+ - Optional auto-commit
+ - Push current branch
+ - Auto-increment semantic tag (vX.Y.Z)
+ - Create & push tag
+ - Fail fast on unsafe states
+
+.EXAMPLES
+ # Safe mode (recommended): requires clean working tree
+ .\release.ps1
+
+ # Auto commit tracked/untracked changes then release
+ .\release.ps1 -AutoCommit -CommitMessage "chore: update skill config"
+
+ # Dry run (show what would happen)
+ .\release.ps1 -DryRun
+
+ # Custom tag prefix
+ .\release.ps1 -Prefix "v" -Message "正式发布"
+
+.NOTES
+ Requires: git, PowerShell 5+
+
+ 加密与 ZIP 内容由 CI 工作流 reusable-release-skill.yaml 的「Encrypt Source Code」步骤执行。CI 会：
+ - 加密 scripts/（递归 PyArmor -r，输出到包内 scripts/，与源码目录树一致）
+ - 复制 SKILL.md
+ - 复制 references/（排除 REQUIREMENTS.md）
+ - 复制 assets/
+ - 复制 tests/
+ - 复制 evals/（除 scripts 外均为明文；若目录不存在则跳过；复制时排除常见缓存与运行产物）
+ 本脚本在打 tag 前会做一次 scripts/ 结构自检，避免子目录未提交却仍发布。
+#>
+
 [CmdletBinding()]
 param(
  [string]$Prefix = "v",
@@ -11,13 +52,218 @@ param(
 Set-StrictMode -Version Latest
 $ErrorActionPreference = "Stop"
 
-$scriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
-$sharedScript = Join-Path $scriptDir "..\jiangchang-platform-kit\tools\release.ps1"
-$sharedScript = [System.IO.Path]::GetFullPath($sharedScript)
-
-if (-not (Test-Path $sharedScript)) {
-  throw "Shared release script not found: $sharedScript"
+function Invoke-Git {
+ param([Parameter(Mandatory = $true)][string]$Args)
+ Write-Host ">> git $Args" -ForegroundColor DarkGray
+ # 将 git 的 stderr 合并进 stdout，避免 PowerShell 在 $ErrorActionPreference=Stop 下
+ # 把 "remote: ..." / "warning: LF -> CRLF" 等非错误输出误判为异常。
+ & cmd /c "git $Args 2>&1"
+ if ($LASTEXITCODE -ne 0) {
+ throw "git command failed: git $Args"
+ }
 }
 
-& $sharedScript @PSBoundParameters
-exit $LASTEXITCODE
+function Get-GitOutput {
+ param([Parameter(Mandatory = $true)][string]$Args)
+ $output = & cmd /c "git $Args" 2>$null
+ if ($LASTEXITCODE -ne 0) {
+ throw "git command failed: git $Args"
+ }
+ return @($output)
+}
+
+function Test-Repo {
+ & git rev-parse --is-inside-work-tree *> $null
+ return ($LASTEXITCODE -eq 0)
+}
+
+function Get-CurrentBranch {
+ $b = (Get-GitOutput "branch --show-current" | Select-Object -First 1).Trim()
+ return $b
+}
+
+function Get-StatusPorcelain {
+ $lines = @(Get-GitOutput "status --porcelain")
+ return $lines
+}
+
+function Parse-SemVerTag {
+ param(
+ [string]$Tag,
+ [string]$TagPrefix
+ )
+ $escaped = [regex]::Escape($TagPrefix)
+ $m = [regex]::Match($Tag, "^${escaped}(\d+)\.(\d+)\.(\d+)$")
+ if (-not $m.Success) { return $null }
+
+ return [pscustomobject]@{
+ Raw = $Tag
+ Major = [int]$m.Groups[1].Value
+ Minor = [int]$m.Groups[2].Value
+ Patch = [int]$m.Groups[3].Value
+ }
+}
+
+function Get-NextTag {
+ param([string]$TagPrefix)
+
+ $tags = Get-GitOutput "tag --list"
+ $parsed = @()
+
+ foreach ($t in $tags) {
+ $t = $t.Trim()
+ if (-not $t) { continue }
+ $obj = Parse-SemVerTag -Tag $t -TagPrefix $TagPrefix
+ if ($null -ne $obj) { $parsed += $obj }
+ }
+
+ if ($parsed.Count -eq 0) {
+ return "${TagPrefix}1.0.1"
+ }
+
+ $latest = $parsed | Sort-Object Major, Minor, Patch | Select-Object -Last 1
+ return "$TagPrefix$($latest.Major).$($latest.Minor).$([int]$latest.Patch + 1)"
+}
+
+function Assert-SkillReleasePackagingSources {
+ param([Parameter(Mandatory = $true)][string]$SkillRoot)
+
+ $scriptsDir = Join-Path $SkillRoot "scripts"
+ $mainPy = Join-Path $scriptsDir "main.py"
+ if (-not (Test-Path -LiteralPath $mainPy)) {
+ return
+ }
+
+ $text = Get-Content -LiteralPath $mainPy -Raw -Encoding UTF8 -ErrorAction SilentlyContinue
+ if ([string]::IsNullOrWhiteSpace($text)) {
+ return
+ }
+
+ $need = @()
+ if ($text -match '(?m)^\s*from\s+cli\.') { $need += 'cli' }
+ if ($text -match '(?m)^\s*import\s+cli\b') { $need += 'cli' }
+ if ($text -match '(?m)^\s*from\s+service\.') { $need += 'service' }
+ if ($text -match '(?m)^\s*import\s+service\b') { $need += 'service' }
+ if ($text -match '(?m)^\s*from\s+db\.') { $need += 'db' }
+ if ($text -match '(?m)^\s*import\s+db\b') { $need += 'db' }
+ if ($text -match '(?m)^\s*from\s+util\.') { $need += 'util' }
+ if ($text -match '(?m)^\s*import\s+util\b') { $need += 'util' }
+
+ foreach ($p in ($need | Select-Object -Unique)) {
+ $folder = Join-Path $scriptsDir $p
+ if (-not (Test-Path -LiteralPath $folder)) {
+ throw "Release check failed: scripts/main.py imports from '$p' but folder is missing: $folder"
+ }
+ }
+
+ $pyFiles = @(Get-ChildItem -LiteralPath $scriptsDir -Filter *.py -Recurse -File -ErrorAction SilentlyContinue)
+ Write-Host "Packaging check: $($pyFiles.Count) Python file(s) under scripts/ (CI will obfuscate all recursively)." -ForegroundColor DarkGray
+}
+
+function Ensure-CleanOrAutoCommit {
+ param(
+ [switch]$DoAutoCommit,
+ [switch]$NeedClean,
+ [switch]$IsDryRun,
+ [string]$Msg
+ )
+
+ $status = @(Get-StatusPorcelain)
+ if ($status.Length -eq 0) { return }
+
+ if ($NeedClean) {
+ Write-Host "Working tree is not clean and -RequireClean is enabled." -ForegroundColor Yellow
+ & git status --short
+ throw "Abort: dirty working tree."
+ }
+
+ # 默认一键发布：有改动就自动提交；也可用 -AutoCommit 显式开启
+ $commitMsg = $Msg
+ if ([string]::IsNullOrWhiteSpace($commitMsg)) {
+ $commitMsg = "chore: auto release commit ($(Get-Date -Format 'yyyy-MM-dd HH:mm:ss'))"
+ }
+
+ if (-not $DoAutoCommit) {
+ Write-Host "Detected uncommitted changes, auto-committing before release..." -ForegroundColor Yellow
+ }
+
+ if ($IsDryRun) {
+ Write-Host "[DryRun] Would run: git add -A" -ForegroundColor Yellow
+ Write-Host "[DryRun] Would run: git commit -m `"$commitMsg`"" -ForegroundColor Yellow
+ return
+ }
+
+ Invoke-Git "add -A"
+ Invoke-Git "commit -m `"$commitMsg`""
+}
+
+try {
+ Write-Host "=== Release Script Start ===" -ForegroundColor Cyan
+
+ if (-not (Test-Repo)) {
+ throw "Current directory is not a git repository."
+ }
+
+ $branch = Get-CurrentBranch
+ if ([string]::IsNullOrWhiteSpace($branch)) {
+ throw "Unable to determine current branch."
+ }
+
+ if ($branch -notin @("main", "master")) {
+ throw "Current branch is '$branch'. Release is only allowed from main/master."
+ }
+
+ Invoke-Git "fetch --tags --prune origin"
+
+ Ensure-CleanOrAutoCommit -DoAutoCommit:$AutoCommit -NeedClean:$RequireClean -IsDryRun:$DryRun -Msg $CommitMessage
+
+ $skillRoot = (Get-GitOutput "rev-parse --show-toplevel" | Select-Object -First 1).Trim()
+ if (Test-Path -LiteralPath (Join-Path $skillRoot "SKILL.md")) {
+ Assert-SkillReleasePackagingSources -SkillRoot $skillRoot
+ }
+
+ # 用 cmd /c + 2>&1 将 stderr 合并进 stdout，避免 PowerShell 在 $ErrorActionPreference=Stop
+ # 下把首次发布时 "fatal: ambiguous argument '@{u}'" 之类的 stderr 误判为异常。
+ $upstream = & cmd /c 'git rev-parse --abbrev-ref --symbolic-full-name "@{u}" 2>&1'
+ $hasUpstream = ($LASTEXITCODE -eq 0)
+
+ if ($DryRun) {
+ if ($hasUpstream) {
+ Write-Host "[DryRun] Would run: git push" -ForegroundColor Yellow
+ } else {
+ Write-Host "[DryRun] Would run: git push -u origin $branch" -ForegroundColor Yellow
+ }
+ } else {
+ if ($hasUpstream) {
+ Invoke-Git "push"
+ } else {
+ Invoke-Git "push -u origin $branch"
+ }
+ }
+
+ $nextTag = Get-NextTag -TagPrefix $Prefix
+ Write-Host "Next tag: $nextTag" -ForegroundColor Green
+
+ $existing = @(Get-GitOutput "tag --list `"$nextTag`"")
+ if ($existing.Length -gt 0) {
+ throw "Tag already exists: $nextTag"
+ }
+
+ if ($DryRun) {
+ Write-Host "[DryRun] Would run: git tag -a $nextTag -m `"$Message`"" -ForegroundColor Yellow
+ Write-Host "[DryRun] Would run: git push origin $nextTag" -ForegroundColor Yellow
+ Write-Host "=== DryRun Complete ===" -ForegroundColor Cyan
+ exit 0
+ }
+
+ Invoke-Git "tag -a $nextTag -m `"$Message`""
+ Invoke-Git "push origin $nextTag"
+
+ Write-Host "Release success: $nextTag" -ForegroundColor Green
+ Write-Host "=== Release Script Done ===" -ForegroundColor Cyan
+ exit 0
+}
+catch {
+ Write-Host "Release failed: $($_.Exception.Message)" -ForegroundColor Red
+ exit 1
+}

requirements-platform-kit.txt

@@ -0,0 +1,7 @@
+# Optional: install shared SDK from Gitea Package Registry (matches vendored scripts/jiangchang_skill_core at release time).
+#
+#   pip install -r requirements-platform-kit.txt ^
+#     --index-url https://git.jc2009.com/api/packages/client-jiangchang/pypi/simple/ ^
+#     --extra-index-url https://pypi.org/simple
+#
+jiangchang-platform-kit>=0.1.0

requirements.txt

@@ -0,0 +1,3 @@
+# 公共依赖由宿主共享 runtime 提供；这里只声明技能特有 Python 依赖。
+# Fernet 加密（local_encrypted 凭据、master key、CLI 启动 import）依赖此包。
+cryptography>=42.0.0,<45

scripts/cli/app.py

@@ -3,7 +3,7 @@
 CLI entry point: argv dispatch for account-manager v2.
 
 Subcommand tree:
-  platform  list/get
+  platform  list/get/ensure
   account   add-web|add-secret|get|get-credential|list|pick-web|mark-session|mark-error|mark-used|open
             init-master-key|export-credentials|import-credentials
   credential pick|resolve
@@ -42,6 +42,7 @@ from service.account_service import (
     cmd_list,
     cmd_list_json,
     cmd_pick_web,
+    cmd_platform_ensure,
     cmd_platform_get,
     cmd_platform_list,
 )
@@ -60,6 +61,8 @@ _USAGE = """account-manager v2 — Credential & Browser Profile Manager
 平台管理:
   platform list [--domain logistics|llm|content]
   platform get <platform>
+  platform ensure --key <key> --display-name <name> [--domain <d>]
+                  [--url <url>] [--auth-strategy <s>]
 
 账号管理:
   account add-web --platform <p> --login-id <id> --login-id-type <type>
@@ -138,6 +141,48 @@ def _get_int_opt(argv: list[str], flag: str, default: int) -> int:
     return default
 
 
+def _get_all_opts(argv: list[str], flag: str) -> list[str]:
+    """Collect all values for a repeatable flag (e.g. --platform-alias)."""
+    vals: list[str] = []
+    i = 0
+    while i < len(argv):
+        if argv[i] == flag and i + 1 < len(argv):
+            vals.append(argv[i + 1])
+            i += 2
+        else:
+            i += 1
+    return vals
+
+
+def _parse_platform_aliases(argv: list[str]) -> list[str]:
+    aliases: list[str] = []
+    for raw in _get_all_opts(argv, "--platform-alias"):
+        for part in (raw or "").split(","):
+            s = part.strip()
+            if s and s not in aliases:
+                aliases.append(s)
+    return aliases
+
+
+def _parse_platform_caller_meta(argv: list[str]):
+    from service.account_service import PlatformCallerMeta
+
+    display_name = _get_opt(argv, "--platform-display-name")
+    url = _get_opt(argv, "--platform-url")
+    domain = _get_opt(argv, "--platform-domain")
+    auth_strategy = _get_opt(argv, "--platform-auth-strategy")
+    aliases = _parse_platform_aliases(argv)
+    if not any([display_name, url, domain, auth_strategy, aliases]):
+        return None
+    return PlatformCallerMeta(
+        display_name=display_name,
+        url=url,
+        domain=domain,
+        auth_strategy=auth_strategy,
+        aliases=aliases,
+    )
+
+
 # =========================================================================
 # Main dispatch
 # =========================================================================
@@ -161,7 +206,7 @@ def main() -> None:
     # ---- platform subcommand ----
     if cmd == "platform":
         if len(av) < 3:
-            print("ERROR:CLI_BAD_ARGS platform 需要子命令：list | get")
+            print("ERROR:CLI_BAD_ARGS platform 需要子命令：list | get | ensure")
             sys.exit(1)
         sub = av[2]
         if sub == "list":
@@ -172,8 +217,30 @@ def main() -> None:
                 print("ERROR:CLI_BAD_ARGS platform get 需要 <platform>。")
                 sys.exit(1)
             cmd_platform_get(av[3])
+        elif sub == "ensure":
+            import argparse as _ap
+            p = _ap.ArgumentParser(prog="platform ensure")
+            p.add_argument("--key", required=True)
+            p.add_argument("--display-name", default="")
+            p.add_argument("--domain", default="generic")
+            p.add_argument("--url", default="")
+            p.add_argument("--auth-strategy", default=None)
+            p.add_argument("--aliases", default=None,
+                           help="JSON 数组字符串，如 '[\"1688\",\"阿里巴巴1688\"]'")
+            p.add_argument("--capabilities", default=None,
+                           help='JSON 对象，如 \'{"web":true,"api_key":false,"rpa":true}\'')
+            args = p.parse_args(av[3:])
+            sys.exit(cmd_platform_ensure(
+                key=args.key,
+                display_name=args.display_name or args.key,
+                domain=args.domain,
+                url=args.url,
+                auth_strategy=args.auth_strategy,
+                aliases=args.aliases,
+                capabilities=args.capabilities,
+            ))
         else:
-            print("ERROR:CLI_BAD_ARGS platform 子命令必须是 list 或 get。")
+            print("ERROR:CLI_BAD_ARGS platform 子命令必须是 list、get 或 ensure。")
             sys.exit(1)
 
     # ---- account subcommand ----
@@ -391,6 +458,7 @@ def _cmd_add_web(argv: list[str]) -> None:
     if not plat:
         print("ERROR:CLI_BAD_ARGS account add-web 需要 --platform <platform>。")
         return
+    platform_meta = _parse_platform_caller_meta(argv)
     cmd_account_add_web(
         platform_input=plat,
         login_id=login_id,
@@ -406,6 +474,7 @@ def _cmd_add_web(argv: list[str]) -> None:
         auth_strategy=auth_strategy,
         session_persistent=session_persistent,
         device_fingerprint=device_fingerprint,
+        platform_meta=platform_meta,
     )
 
 
@@ -478,6 +547,7 @@ def _cmd_pick_web_new(argv: list[str]) -> None:
     if not plat:
         print("ERROR:CLI_BAD_ARGS account pick-web 需要 --platform <platform>。")
         return
+    platform_meta = _parse_platform_caller_meta(argv)
     cmd_account_pick_web(
         platform_input=plat,
         environment=env,
@@ -487,6 +557,7 @@ def _cmd_pick_web_new(argv: list[str]) -> None:
         ttl_sec=ttl,
         holder=holder,
         purpose=purpose,
+        platform_meta=platform_meta,
     )
 
 

scripts/db/accounts_repo.py

@@ -763,6 +763,158 @@ def get_active_lease_by_token(conn, lease_token: str) -> Optional[dict[str, Any]
     }
 
 
+def upsert_platform(
+    conn,
+    key: str,
+    display_name: str,
+    domain: str = "generic",
+    provider_code: str | None = None,
+    default_url: str = "",
+    aliases_json: str = "[]",
+    capabilities_json: str = '{"web": true, "api_key": false, "rpa": true}',
+    default_auth_strategy: str | None = None,
+) -> None:
+    """幂等地注册或更新平台到 DB。"""
+    now = int(time.time())
+    if provider_code is None:
+        provider_code = key
+    conn.execute(
+        """
+        INSERT INTO platforms (
+            platform_key, display_name, domain, provider_code,
+            default_url, aliases_json, capabilities_json,
+            enabled, default_auth_strategy, created_at, updated_at
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)
+        ON CONFLICT(platform_key) DO UPDATE SET
+            display_name          = excluded.display_name,
+            domain                = excluded.domain,
+            provider_code         = excluded.provider_code,
+            default_url           = excluded.default_url,
+            aliases_json          = excluded.aliases_json,
+            capabilities_json     = excluded.capabilities_json,
+            enabled               = 1,
+            default_auth_strategy = COALESCE(excluded.default_auth_strategy,
+                                             platforms.default_auth_strategy),
+            updated_at            = excluded.updated_at
+        """,
+        (
+            key, display_name, domain, provider_code,
+            default_url, aliases_json, capabilities_json,
+            default_auth_strategy, now, now,
+        ),
+    )
+    conn.commit()
+
+
+def ensure_platform_from_caller_meta(
+    conn,
+    key: str,
+    *,
+    display_name: str,
+    domain: str = "generic",
+    default_url: str = "",
+    aliases: list[str] | None = None,
+    default_auth_strategy: str | None = None,
+    capabilities_json: str = '{"web": true, "api_key": false, "rpa": true}',
+) -> str:
+    """
+    调用方元数据自动注册平台：不存在则创建；已存在则仅填充空字段，不覆盖非空自定义值。
+    返回 created | updated | unchanged。
+    """
+    now = int(time.time())
+    provider_code = key
+    alias_list: list[str] = []
+    for item in [key, display_name, *(aliases or [])]:
+        s = str(item).strip()
+        if s and s not in alias_list:
+            alias_list.append(s)
+    aliases_json = json.dumps(alias_list, ensure_ascii=False)
+
+    row = conn.execute(
+        """
+        SELECT display_name, domain, provider_code, default_url, aliases_json,
+               capabilities_json, default_auth_strategy, enabled
+        FROM platforms WHERE platform_key = ?
+        """,
+        (key,),
+    ).fetchone()
+
+    if row is None:
+        upsert_platform(
+            conn,
+            key=key,
+            display_name=display_name,
+            domain=domain,
+            provider_code=provider_code,
+            default_url=default_url,
+            aliases_json=aliases_json,
+            capabilities_json=capabilities_json,
+            default_auth_strategy=default_auth_strategy,
+        )
+        return "created"
+
+    cur_display, cur_domain, cur_provider, cur_url, cur_aliases_json, cur_caps, cur_auth, enabled = row
+    if not enabled:
+        return "unchanged"
+
+    merged_display = (cur_display or "").strip() or display_name
+    merged_domain = (cur_domain or "").strip() or domain
+    merged_provider = (cur_provider or "").strip() or provider_code
+    merged_url = (cur_url or "").strip() or (default_url or "").strip()
+    merged_auth = cur_auth or default_auth_strategy
+
+    existing_aliases = _safe_json_loads(cur_aliases_json, [])
+    if not isinstance(existing_aliases, list):
+        existing_aliases = []
+    merged_aliases: list[str] = []
+    for item in list(existing_aliases) + alias_list:
+        s = str(item).strip()
+        if s and s not in merged_aliases:
+            merged_aliases.append(s)
+    merged_aliases_json = json.dumps(merged_aliases, ensure_ascii=False)
+
+    merged_caps = cur_caps or capabilities_json
+
+    unchanged = (
+        merged_display == (cur_display or "")
+        and merged_domain == (cur_domain or "")
+        and merged_provider == (cur_provider or "")
+        and merged_url == (cur_url or "")
+        and merged_aliases_json == (cur_aliases_json or "[]")
+        and merged_auth == cur_auth
+    )
+    if unchanged:
+        return "unchanged"
+
+    conn.execute(
+        """
+        UPDATE platforms SET
+            display_name = ?,
+            domain = ?,
+            provider_code = ?,
+            default_url = ?,
+            aliases_json = ?,
+            capabilities_json = ?,
+            default_auth_strategy = ?,
+            updated_at = ?
+        WHERE platform_key = ?
+        """,
+        (
+            merged_display,
+            merged_domain,
+            merged_provider,
+            merged_url,
+            merged_aliases_json,
+            merged_caps,
+            merged_auth,
+            now,
+            key,
+        ),
+    )
+    conn.commit()
+    return "updated"
+
+
 # ============================================================================
 # Internal helpers
 # ============================================================================

scripts/service/_svc_common.py

@@ -0,0 +1,255 @@
+# -*- coding: utf-8 -*-
+"""Shared helpers for account service command modules."""
+import json
+import os
+import shutil
+import sys
+from dataclasses import dataclass, field
+from typing import Optional
+
+from db.accounts_repo import (
+    insert_account,
+    resolve_auth_strategy_for_platform,
+    ensure_platform_from_caller_meta,
+)
+from db.auth_strategy import (
+    ALL_AUTH_STRATEGIES,
+    AUTH_STRATEGY_CLIENT_CERTIFICATE,
+    AUTH_STRATEGY_DEVICE_BOUND,
+    AUTH_STRATEGY_PASSWORD_PLUS_2FA,
+    AUTH_STRATEGY_PER_SESSION_MANUAL,
+    AUTH_STRATEGY_QR_CODE_MANUAL,
+    AUTH_STRATEGY_SSO_REDIRECT,
+)
+from db.connection import get_conn, init_db
+from util.logging_config import get_skill_logger
+from util.platforms import (
+    DEFAULT_CAPABILITIES_JSON,
+    PlatformResolveResult,
+    _platform_list_cn_for_help,
+    is_valid_platform_key_format,
+    resolve_platform_key_dynamic,
+    seed_platforms,
+)
+
+
+def _map_secret_read_error(exc: BaseException) -> tuple[str, str]:
+    """Map secret read/write exceptions to ERROR:* codes (no secret text in code)."""
+    msg = str(exc)
+    if "SECRET_ENV_MISSING" in msg:
+        return "ERROR:SECRET_ENV_MISSING", msg
+    if "WINDOWS_CREDENTIAL_UNAVAILABLE" in msg:
+        return "ERROR:WINDOWS_CREDENTIAL_UNAVAILABLE", msg
+    return "ERROR:SECRET_READ_FAILED", msg
+
+
+def _credential_ok_line(payload: dict) -> str:
+    """Single-line JSON for credential commands (machine integration)."""
+    return json.dumps({"success": True, **payload}, ensure_ascii=False)
+
+
+def _derive_session_persistent(auth_strategy: str) -> int:
+    """Derive default session_persistent from auth_strategy when CLI omits it."""
+    if auth_strategy in (
+        AUTH_STRATEGY_QR_CODE_MANUAL,
+        AUTH_STRATEGY_PASSWORD_PLUS_2FA,
+        AUTH_STRATEGY_DEVICE_BOUND,
+        AUTH_STRATEGY_SSO_REDIRECT,
+    ):
+        return 1
+    if auth_strategy in (AUTH_STRATEGY_PER_SESSION_MANUAL, AUTH_STRATEGY_CLIENT_CERTIFICATE):
+        return 0
+    return 1
+
+
+def _insert_secret_holder_account(
+    conn,
+    *,
+    platform_key: str,
+    account_label: str,
+    environment: str,
+    role: str,
+    provider_code: str,
+    url: Optional[str],
+    now: int,
+) -> int:
+    """Create a minimal accounts row so credentials can join on environment/role."""
+    return insert_account(
+        conn=conn,
+        platform_key=platform_key,
+        account_label=account_label,
+        login_id=None,
+        login_id_type="api_account",
+        tenant_id=None,
+        environment=environment,
+        role=role,
+        provider_code=provider_code,
+        profile_dir=None,
+        url=url,
+        extra_json="{}",
+        now=now,
+        auth_strategy=resolve_auth_strategy_for_platform(conn, platform_key),
+    )
+
+
+def _err_json(code: str, message: str) -> str:
+    """Return a JSON error line."""
+    return json.dumps(
+        {"success": False, "error": {"code": code, "message": message}},
+        ensure_ascii=False,
+    )
+
+
+def _ok_json(data) -> str:
+    """Return a JSON success line."""
+    return json.dumps(
+        {"success": True, "data": data},
+        ensure_ascii=False,
+    )
+
+
+def _say(msg: str) -> None:
+    """Print to stdout."""
+    print(msg)
+
+
+def _say_err(msg: str) -> None:
+    """Print to stderr."""
+    print(msg, file=sys.stderr)
+
+
+def _fail_human(code: str, hint_lines: list[str]) -> None:
+    """Print human-readable error block."""
+    _say(f"ERROR:{code}")
+    for line in hint_lines:
+        _say(line)
+
+
+def _remove_profile_dir(path: str) -> None:
+    p = (path or "").strip()
+    if not p or not os.path.isdir(p):
+        return
+    try:
+        shutil.rmtree(p)
+    except OSError as e:
+        _say_err(f"WARNING: 未能删除用户数据目录（可手工删）：{p}\n   {e}")
+
+
+@dataclass
+class PlatformCallerMeta:
+    """调用方为未注册自定义平台提供的自动注册元数据（pick-web / add-web）。"""
+
+    display_name: Optional[str] = None
+    url: Optional[str] = None
+    domain: Optional[str] = None
+    auth_strategy: Optional[str] = None
+    aliases: list[str] = field(default_factory=list)
+
+    def is_sufficient_for_auto_register(self) -> bool:
+        return bool((self.display_name or "").strip() and (self.url or "").strip())
+
+
+def _emit_platform_resolve_error(result: PlatformResolveResult, hint: str = "") -> None:
+    code = result.error_code or "INVALID_PLATFORM"
+    if not code.startswith("ERROR:"):
+        code = f"ERROR:{code}"
+    _say(_err_json(code, result.message or "无法识别平台。"))
+    if hint:
+        _say(hint)
+
+
+def _resolve_or_auto_register(
+    platform_input: str,
+    meta: Optional[PlatformCallerMeta] = None,
+    *,
+    hint: str = "",
+) -> Optional[str]:
+    """解析平台；未注册时可在元数据充足时自动 ensure（幂等、仅填空字段）。"""
+    init_db()
+    conn = get_conn()
+    try:
+        result = resolve_platform_key_dynamic(platform_input, conn=conn)
+        if result.ok:
+            return result.platform_key
+
+        if result.error_code in ("PLATFORM_DISABLED", "INVALID_PLATFORM_KEY", "INVALID_PLATFORM"):
+            _emit_platform_resolve_error(result, hint=hint)
+            return None
+
+        if result.error_code != "PLATFORM_NOT_REGISTERED":
+            _emit_platform_resolve_error(result, hint=hint)
+            if result.error_code not in (
+                "PLATFORM_NOT_REGISTERED",
+                "INVALID_PLATFORM_KEY",
+                "PLATFORM_DISABLED",
+            ):
+                _say("支持：" + _platform_list_cn_for_help())
+            return None
+
+        raw_key = (platform_input or "").strip().lower()
+        if not is_valid_platform_key_format(raw_key):
+            _emit_platform_resolve_error(
+                PlatformResolveResult(
+                    error_code="INVALID_PLATFORM_KEY",
+                    message=result.message or f"平台 key 格式非法：「{platform_input}」。",
+                ),
+                hint=hint,
+            )
+            return None
+
+        if not meta or not meta.is_sufficient_for_auto_register():
+            _emit_platform_resolve_error(
+                PlatformResolveResult(
+                    error_code="PLATFORM_NOT_REGISTERED",
+                    message=(
+                        f"平台 {raw_key} 尚未注册。"
+                        "请传入 --platform-display-name 与 --platform-url，"
+                        "或先执行 account platform ensure。"
+                    ),
+                ),
+                hint=hint,
+            )
+            return None
+
+        if meta.auth_strategy and meta.auth_strategy not in ALL_AUTH_STRATEGIES:
+            _say(_err_json(
+                "ERR_AUTH_STRATEGY_NOT_SUPPORTED",
+                f"不支持的 --platform-auth-strategy：{meta.auth_strategy}",
+            ))
+            return None
+
+        seed_platforms(conn)
+        action = ensure_platform_from_caller_meta(
+            conn,
+            raw_key,
+            display_name=(meta.display_name or "").strip(),
+            domain=(meta.domain or "generic").strip() or "generic",
+            default_url=(meta.url or "").strip(),
+            aliases=meta.aliases,
+            default_auth_strategy=meta.auth_strategy,
+            capabilities_json=DEFAULT_CAPABILITIES_JSON,
+        )
+        get_skill_logger().info(
+            "platform_auto_registered key=%s action=%s", raw_key, action
+        )
+        return raw_key
+    finally:
+        conn.close()
+
+
+def _resolve_or_fail(input_name: str, hint: str = "") -> Optional[str]:
+    """Resolve platform key or emit structured JSON error and return None."""
+    init_db()
+    conn = get_conn()
+    try:
+        result = resolve_platform_key_dynamic(input_name, conn=conn)
+    finally:
+        conn.close()
+
+    if result.ok:
+        return result.platform_key
+
+    _emit_platform_resolve_error(result, hint=hint)
+    if result.error_code not in ("PLATFORM_NOT_REGISTERED", "INVALID_PLATFORM_KEY", "PLATFORM_DISABLED"):
+        _say("支持：" + _platform_list_cn_for_help())
+    return None

scripts/service/account_add_commands.py

@@ -0,0 +1,545 @@
+# -*- coding: utf-8 -*-
+"""Account creation CLI commands (add-web, add-secret)."""
+import json
+import os
+import sys
+import time
+from typing import Optional
+
+from db.accounts_repo import (
+    get_account_by_id,
+    insert_account,
+    insert_credential,
+    resolve_auth_strategy_for_platform,
+)
+from db.auth_strategy import (
+    ALL_AUTH_STRATEGIES,
+    AUTH_STRATEGY_DEVICE_BOUND,
+)
+from db.connection import get_conn, init_db
+from db.credentials_repo import insert_credential_encrypted
+from service._svc_common import (
+    PlatformCallerMeta,
+    _derive_session_persistent,
+    _err_json,
+    _insert_secret_holder_account,
+    _ok_json,
+    _resolve_or_auto_register,
+    _resolve_or_fail,
+    _say,
+    _say_err,
+)
+from service.crypto import CredentialDecryptError
+from service.master_key import MasterKeyMissingError, is_master_key_available
+from service.secret_store import (
+    mask_secret,
+    make_windows_credential_target,
+    write_secret,
+)
+from util.logging_config import get_skill_logger
+from util.platforms import get_platform_spec, seed_platforms
+from util.profile_shortcut import create_profile_browser_shortcut
+from util.runtime_paths import get_default_profile_dir_for_web
+
+
+def cmd_account_add_web(
+    platform_input: str,
+    login_id: Optional[str],
+    login_id_type: str,
+    label: Optional[str],
+    tenant_id: Optional[str],
+    environment: str,
+    role: str,
+    provider_code: Optional[str],
+    url: Optional[str],
+    extra_json_str: Optional[str],
+    profile_dir_override: Optional[str],
+    *,
+    auth_strategy: Optional[str] = None,
+    session_persistent: Optional[int] = None,
+    device_fingerprint: Optional[str] = None,
+    platform_meta: Optional[PlatformCallerMeta] = None,
+) -> None:
+    """account add-web — create a web/RPA account record."""
+    log = get_skill_logger()
+    log.info("account_add_web_attempt platform=%s login_id_type=%s env=%s role=%s",
+             platform_input, login_id_type, environment, role)
+
+    key = _resolve_or_auto_register(platform_input, platform_meta)
+    if not key:
+        return
+
+    init_db()
+    now = int(time.time())
+    conn = get_conn()
+    try:
+        seed_platforms(conn)
+        platform_spec = get_platform_spec(key, conn=conn)
+    finally:
+        conn.close()
+
+    safe_label = (label or "").strip() or f"{platform_spec.get('display_name', key)} {environment} {role}"
+    extra = {}
+    if extra_json_str:
+        try:
+            extra = json.loads(extra_json_str)
+        except json.JSONDecodeError:
+            _say(_err_json("ERROR:CLI_BAD_ARGS", "extra_json 不是合法的 JSON 字符串。"))
+            return
+
+    if profile_dir_override:
+        resolved_profile_dir = os.path.abspath(profile_dir_override)
+    else:
+        resolved_profile_dir = get_default_profile_dir_for_web(
+            platform_key=key,
+            label=safe_label,
+            login_id=login_id,
+            domain=platform_spec.get("domain", "generic"),
+        )
+    os.makedirs(resolved_profile_dir, exist_ok=True)
+
+    resolved_url = (url or "").strip() or platform_spec.get("default_url", "")
+    resolved_provider = provider_code or platform_spec.get("provider_code") or key
+    resolved_extra = json.dumps(extra, ensure_ascii=False)
+
+    conn = get_conn()
+    try:
+        seed_platforms(conn)
+        if auth_strategy is None:
+            resolved_auth = resolve_auth_strategy_for_platform(conn, key)
+        else:
+            if auth_strategy not in ALL_AUTH_STRATEGIES:
+                _say(_err_json(
+                    "ERR_AUTH_STRATEGY_NOT_SUPPORTED",
+                    f"不支持的 auth_strategy：{auth_strategy}",
+                ))
+                return
+            resolved_auth = auth_strategy
+
+        if session_persistent is None:
+            sp = _derive_session_persistent(resolved_auth)
+        else:
+            if session_persistent not in (0, 1):
+                _say(_err_json(
+                    "ERR_INVALID_SESSION_PERSISTENT",
+                    "session_persistent 必须是 0 或 1。",
+                ))
+                return
+            sp = session_persistent
+
+        if device_fingerprint is not None and resolved_auth != AUTH_STRATEGY_DEVICE_BOUND:
+            _say(_err_json(
+                "ERR_DEVICE_FINGERPRINT_NOT_APPLICABLE",
+                "device_fingerprint 仅在 auth_strategy=device_bound 时可设置。",
+            ))
+            return
+
+        new_id = insert_account(
+            conn=conn,
+            platform_key=key,
+            account_label=safe_label,
+            login_id=login_id or None,
+            login_id_type=login_id_type,
+            tenant_id=tenant_id or None,
+            environment=environment,
+            role=role,
+            provider_code=resolved_provider,
+            profile_dir=resolved_profile_dir,
+            url=resolved_url,
+            extra_json=resolved_extra,
+            now=now,
+            auth_strategy=resolved_auth,
+            session_persistent=sp,
+            device_fingerprint=device_fingerprint,
+        )
+        conn.commit()
+    except Exception:
+        conn.rollback()
+        log.exception("account_add_web_failure")
+        _say(_err_json("ERROR:DB_ERROR", "Database insert failed."))
+        return
+    finally:
+        conn.close()
+
+    shortcut_path = create_profile_browser_shortcut(resolved_profile_dir, safe_label, log)
+
+    log.info("account_add_web_success id=%s platform=%s profile_dir=%s", new_id, key, resolved_profile_dir)
+    ok_payload = {
+        "id": new_id,
+        "platform_key": key,
+        "account_label": safe_label,
+        "profile_dir": resolved_profile_dir,
+        "url": resolved_url,
+        "environment": environment,
+        "role": role,
+        "tenant_id": tenant_id or "",
+        "provider_code": resolved_provider,
+        "status": "active",
+        "session_status": "unknown",
+        "auth_strategy": resolved_auth,
+        "session_persistent": sp,
+        "device_fingerprint": device_fingerprint,
+    }
+    if shortcut_path:
+        ok_payload["profile_shortcut"] = shortcut_path
+    _say(_ok_json(ok_payload))
+
+
+def cmd_account_add_secret(
+    platform_input: str,
+    credential_type: str,
+    label: Optional[str],
+    secret_storage: str,
+    secret_stdin: bool,
+    secret_ref: Optional[str],
+    account_id: Optional[int],
+    environment: Optional[str],
+    role: Optional[str],
+    extra_json_str: Optional[str],
+) -> None:
+    """account add-secret — create a credential record with masked secret."""
+    log = get_skill_logger()
+    log.info("account_add_secret_attempt platform=%s type=%s storage=%s",
+             platform_input, credential_type, secret_storage)
+
+    key = _resolve_or_fail(platform_input)
+    if not key:
+        return
+
+    storage_n = secret_storage.strip().lower()
+    if storage_n not in ("none", "env", "windows_credential", "local_encrypted"):
+        _say(_err_json("ERROR:SECRET_STORAGE_UNSUPPORTED",
+                       f"不支持的存储方式：{secret_storage}。支持：none / env / windows_credential / local_encrypted"))
+        return
+
+    if storage_n == "none" and credential_type not in ("browser_profile", "note"):
+        _say(_err_json("ERROR:SECRET_STORAGE_UNSUPPORTED",
+                       "storage=none 仅允许 credential_type=browser_profile 或 note。"))
+        return
+    if storage_n == "env" and not secret_ref:
+        _say(_err_json("ERROR:SECRET_REF_MISSING",
+                       "env 存储时必须提供 --secret-ref 环境变量名。"))
+        return
+
+    init_db()
+    conn = get_conn()
+    try:
+        seed_platforms(conn)
+        platform_spec = get_platform_spec(key, conn=conn)
+    finally:
+        conn.close()
+    now = int(time.time())
+    safe_label = (label or "").strip() or f"{platform_spec.get('display_name', key)} {credential_type}"
+
+    if storage_n == "local_encrypted":
+        if not is_master_key_available():
+            _say(_err_json(
+                "ERR_MASTER_KEY_MISSING",
+                "master key 未初始化。先跑 `account init-master-key`。",
+            ))
+            return
+        if not secret_stdin:
+            _say(_err_json(
+                "ERR_LOCAL_ENCRYPTED_REQUIRES_STDIN",
+                "local_encrypted 后端必须 --secret-stdin 输入密文，避免命令历史泄漏。",
+            ))
+            return
+        if credential_type not in (
+            "password",
+            "api_key",
+            "api_token",
+            "bearer_token",
+            "refresh_token",
+            "oauth_client",
+        ):
+            _say(_err_json(
+                "ERR_LOCAL_ENCRYPTED_TYPE_NOT_ALLOWED",
+                f"local_encrypted 不接受 credential_type={credential_type}（仅 secret 类）。",
+            ))
+            return
+        if secret_ref:
+            _say_err("[add-secret] WARNING: secret_ref ignored for local_encrypted (secret comes from stdin).")
+
+        extra: dict = {}
+        if extra_json_str:
+            try:
+                extra = json.loads(extra_json_str)
+            except json.JSONDecodeError:
+                extra = {}
+        resolved_env = environment or "production"
+        resolved_role = role or "api"
+        extra["environment"] = resolved_env
+        extra["role"] = resolved_role
+        resolved_extra = json.dumps(extra, ensure_ascii=False)
+        resolved_url = (platform_spec.get("default_url") or "").strip() or None
+        resolved_provider = platform_spec.get("provider_code") or key
+
+        plaintext = sys.stdin.readline().rstrip("\r\n")
+        if not plaintext:
+            _say(_err_json("ERR_LOCAL_ENCRYPTED_EMPTY_STDIN", "stdin 没读到密码内容。"))
+            return
+        secret_mask_out = mask_secret(plaintext)
+
+        conn = get_conn()
+        cred_id: Optional[int] = None
+        holder_id: Optional[int] = None
+        try:
+            seed_platforms(conn)
+            holder_id = account_id
+            if holder_id is not None:
+                acc = get_account_by_id(holder_id)
+                if not acc:
+                    _say(_err_json(
+                        "ERROR:ACCOUNT_NOT_FOUND",
+                        f"account_id={holder_id} 不存在。",
+                    ))
+                    return
+            if holder_id is None:
+                holder_id = _insert_secret_holder_account(
+                    conn,
+                    platform_key=key,
+                    account_label=safe_label,
+                    environment=resolved_env,
+                    role=resolved_role,
+                    provider_code=resolved_provider,
+                    url=resolved_url,
+                    now=now,
+                )
+            cred_id = insert_credential_encrypted(
+                conn,
+                account_id=holder_id,
+                platform_key=key,
+                credential_label=safe_label,
+                credential_type=credential_type,
+                plaintext=plaintext,
+                expires_at=None,
+                extra_json=resolved_extra,
+                now=now,
+            )
+            conn.commit()
+        except MasterKeyMissingError as e:
+            conn.rollback()
+            _say(_err_json("ERR_MASTER_KEY_MISSING", str(e)))
+            return
+        except CredentialDecryptError as e:
+            conn.rollback()
+            _say(_err_json("ERR_CREDENTIAL_DECRYPT_FAILED", str(e)))
+            return
+        except Exception:
+            conn.rollback()
+            log.exception("account_add_secret_failure")
+            _say(_err_json("ERROR:DB_ERROR", "Database insert failed."))
+            return
+        finally:
+            conn.close()
+
+        log.info(
+            "account_add_secret_encrypted_success cred_id=%s account_id=%s",
+            cred_id,
+            holder_id,
+        )
+        _say(_ok_json({
+            "credential_id": cred_id,
+            "account_id": holder_id,
+            "platform_key": key,
+            "credential_type": credential_type,
+            "secret_storage": "local_encrypted",
+            "secret_mask": secret_mask_out,
+        }))
+        return
+
+    if storage_n == "env":
+        secret_ref_value = secret_ref.strip()
+        secret_mask_value = mask_secret(f"env:{secret_ref_value}")
+        extra = {}
+        if extra_json_str:
+            try:
+                extra = json.loads(extra_json_str)
+            except json.JSONDecodeError:
+                extra = {}
+        resolved_env = environment or "production"
+        resolved_role = role or "api"
+        extra["environment"] = resolved_env
+        extra["role"] = resolved_role
+        extra_json_final = json.dumps(extra, ensure_ascii=False)
+
+        resolved_url = (platform_spec.get("default_url") or "").strip() or None
+        resolved_provider = platform_spec.get("provider_code") or key
+
+        conn = get_conn()
+        try:
+            seed_platforms(conn)
+            holder_id = account_id
+            if holder_id is None:
+                holder_id = _insert_secret_holder_account(
+                    conn,
+                    platform_key=key,
+                    account_label=safe_label,
+                    environment=resolved_env,
+                    role=resolved_role,
+                    provider_code=resolved_provider,
+                    url=resolved_url,
+                    now=now,
+                )
+            cred_id = insert_credential(
+                conn=conn,
+                account_id=holder_id,
+                platform_key=key,
+                credential_label=safe_label,
+                credential_type=credential_type,
+                secret_storage=storage_n,
+                secret_ref=secret_ref_value,
+                secret_mask=secret_mask_value,
+                expires_at=None,
+                extra_json=extra_json_final,
+                now=now,
+            )
+            conn.commit()
+        except Exception:
+            conn.rollback()
+            log.exception("account_add_secret_failure")
+            _say(_err_json("ERROR:DB_ERROR", "Database insert failed."))
+            return
+        finally:
+            conn.close()
+
+        log.info("account_add_secret_success id=%s platform=%s storage=env ref=%s",
+                 cred_id, key, secret_ref_value)
+        _say(_ok_json({
+            "account_id": holder_id,
+            "credential_id": cred_id,
+            "platform_key": key,
+            "credential_label": safe_label,
+            "credential_type": credential_type,
+            "secret_storage": storage_n,
+            "secret_ref": secret_ref_value,
+            "secret_mask": secret_mask_value,
+            "warning": "env 存储方式：请确保环境变量在运行时可用。",
+        }))
+        return
+
+    if storage_n == "windows_credential":
+        if not secret_stdin:
+            _say(_err_json("ERROR:CLI_BAD_ARGS",
+                           "windows_credential 需要 --secret-stdin 从 stdin 读取 secret。"))
+            return
+        secret_value = sys.stdin.readline().strip()
+        if not secret_value:
+            _say(_err_json("ERROR:CLI_BAD_ARGS", "stdin 中未读到 secret 值。"))
+            return
+
+        secret_mask_value = mask_secret(secret_value)
+        cred_label_for_target = safe_label
+        target = make_windows_credential_target(key, credential_type, cred_label_for_target)
+
+        try:
+            write_secret("windows_credential", target, secret_value)
+        except (OSError, ValueError) as e:
+            log.error("secret_write_failed storage=windows_credential error=%s", str(e))
+            _say(_err_json("ERROR:SECRET_WRITE_FAILED", f"写 Windows Credential Manager 失败：{e}"))
+            return
+
+        extra = {}
+        if extra_json_str:
+            try:
+                extra = json.loads(extra_json_str)
+            except json.JSONDecodeError:
+                extra = {}
+        resolved_env = environment or "production"
+        resolved_role = role or "api"
+        extra["environment"] = resolved_env
+        extra["role"] = resolved_role
+        extra_json_final = json.dumps(extra, ensure_ascii=False)
+
+        resolved_url = (platform_spec.get("default_url") or "").strip() or None
+        resolved_provider = platform_spec.get("provider_code") or key
+
+        conn = get_conn()
+        try:
+            seed_platforms(conn)
+            holder_id = account_id
+            if holder_id is None:
+                holder_id = _insert_secret_holder_account(
+                    conn,
+                    platform_key=key,
+                    account_label=safe_label,
+                    environment=resolved_env,
+                    role=resolved_role,
+                    provider_code=resolved_provider,
+                    url=resolved_url,
+                    now=now,
+                )
+            cred_id = insert_credential(
+                conn=conn,
+                account_id=holder_id,
+                platform_key=key,
+                credential_label=safe_label,
+                credential_type=credential_type,
+                secret_storage=storage_n,
+                secret_ref=target,
+                secret_mask=secret_mask_value,
+                expires_at=None,
+                extra_json=extra_json_final,
+                now=now,
+            )
+            conn.commit()
+        except Exception:
+            conn.rollback()
+            log.exception("account_add_secret_failure")
+            _say(_err_json("ERROR:DB_ERROR", "Database insert failed."))
+            return
+        finally:
+            conn.close()
+
+        log.info("account_add_secret_success id=%s platform=%s storage=windows_credential",
+                 cred_id, key)
+        secret_value = ""
+
+        _say(_ok_json({
+            "account_id": holder_id,
+            "credential_id": cred_id,
+            "platform_key": key,
+            "credential_label": safe_label,
+            "credential_type": credential_type,
+            "secret_storage": storage_n,
+            "secret_ref": target,
+            "secret_mask": secret_mask_value,
+        }))
+        return
+
+    conn = get_conn()
+    try:
+        seed_platforms(conn)
+        cred_id = insert_credential(
+            conn=conn,
+            account_id=account_id,
+            platform_key=key,
+            credential_label=safe_label,
+            credential_type=credential_type,
+            secret_storage=storage_n,
+            secret_ref=None,
+            secret_mask="",
+            expires_at=None,
+            extra_json=extra_json_str or "{}",
+            now=now,
+        )
+        conn.commit()
+    except Exception:
+        conn.rollback()
+        log.exception("account_add_secret_failure")
+        _say(_err_json("ERROR:DB_ERROR", "Database insert failed."))
+        return
+    finally:
+        conn.close()
+
+    log.info("account_add_secret_success id=%s platform=%s storage=none", cred_id, key)
+    _say(_ok_json({
+        "account_id": account_id,
+        "credential_id": cred_id,
+        "platform_key": key,
+        "credential_label": safe_label,
+        "credential_type": credential_type,
+        "secret_storage": storage_n,
+        "secret_ref": "",
+        "secret_mask": "",
+    }))

scripts/service/account_mark_commands.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+"""Account status mark CLI commands."""
+from db.accounts_repo import (
+    get_account_by_id,
+    update_account_error,
+    update_account_last_used,
+    update_account_session_status,
+)
+from service._svc_common import _err_json, _ok_json, _say
+from util.logging_config import get_skill_logger
+
+
+def cmd_account_mark_session(account_id: int, session_status: str) -> None:
+    """account mark-session <id> --session-status <status>"""
+    log = get_skill_logger()
+    log.info("account_mark_session id=%s status=%s", account_id, session_status)
+    valid_statuses = ("unknown", "likely_valid", "needs_login", "expired")
+    if session_status not in valid_statuses:
+        _say(_err_json("ERROR:CLI_BAD_ARGS",
+                       f"session_status 必须为 {valid_statuses} 之一。"))
+        return
+    acc = get_account_by_id(account_id)
+    if not acc:
+        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={account_id} 不存在。"))
+        return
+    update_account_session_status(account_id, session_status)
+    _say(_ok_json({"id": account_id, "session_status": session_status}))
+
+
+def cmd_account_mark_error(account_id: int, error_code: str, error_message: str) -> None:
+    """account mark-error <id> --code <code> --message <message>"""
+    log = get_skill_logger()
+    log.info("account_mark_error id=%s code=%s", account_id, error_code)
+    acc = get_account_by_id(account_id)
+    if not acc:
+        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={account_id} 不存在。"))
+        return
+    update_account_error(account_id, error_code, error_message)
+    _say(_ok_json({"id": account_id, "error_code": error_code}))
+
+
+def cmd_account_mark_used(account_id: int) -> None:
+    """account mark-used <id>"""
+    log = get_skill_logger()
+    log.info("account_mark_used id=%s", account_id)
+    acc = get_account_by_id(account_id)
+    if not acc:
+        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={account_id} 不存在。"))
+        return
+    update_account_last_used(account_id)
+    _say(_ok_json({"id": account_id, "marked": "used"}))

scripts/service/account_query_commands.py

@@ -0,0 +1,325 @@
+# -*- coding: utf-8 -*-
+"""Account query and pick CLI commands."""
+import json
+import uuid
+from typing import Optional
+
+from db.accounts_repo import (
+    acquire_lease,
+    find_account_ids,
+    find_accounts,
+    find_credentials,
+    get_account_by_id,
+    get_active_lease_by_token,
+    update_account_last_used,
+)
+from db.connection import get_conn, init_db
+from db.credentials_repo import (
+    CredentialNotEncryptedError,
+    list_credentials_by_account,
+    read_credential_plaintext,
+)
+from service._svc_common import (
+    PlatformCallerMeta,
+    _err_json,
+    _resolve_or_auto_register,
+    _resolve_or_fail,
+    _say,
+    _say_err,
+)
+from service.crypto import CredentialDecryptError
+from service.master_key import MasterKeyMissingError
+from util.logging_config import get_skill_logger
+from util.platforms import seed_platforms
+from util.runtime_paths import _runtime_paths_debug_text
+
+
+def cmd_account_get(account_id, include_credentials: bool = False) -> None:
+    """Get a single account by id. Returns JSON."""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=account_get id=%r", account_id)
+    init_db()
+    acc = get_account_by_id(account_id)
+    if not acc:
+        log.warning("account_get_not_found id=%r", account_id)
+        _say(f"ERROR:ACCOUNT_NOT_FOUND 账号 id={account_id} 不存在。")
+        _say_err(_runtime_paths_debug_text())
+        return
+    if include_credentials:
+        creds = find_credentials(
+            platform_key=acc["platform_key"],
+            limit=20,
+        )
+        linked_creds = [c for c in creds if c["account_id"] == acc["id"]]
+        if not linked_creds:
+            linked_creds = [c for c in creds if c["account_id"] is None]
+        acc["credentials"] = [
+            {
+                "id": c["id"],
+                "credential_label": c["credential_label"],
+                "credential_type": c["credential_type"],
+                "secret_storage": c["secret_storage"],
+                "secret_ref": c["secret_ref"],
+                "secret_mask": c["secret_mask"],
+                "status": c["status"],
+            }
+            for c in linked_creds
+        ]
+    _say(json.dumps(acc, ensure_ascii=False))
+
+
+def cmd_account_get_credential(
+    account_id: int,
+    *,
+    reveal: bool = False,
+    lease_token: Optional[str] = None,
+) -> None:
+    """account get-credential — metadata or plaintext when lease proves access."""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=account_get_credential id=%s reveal=%s", account_id, reveal)
+
+    init_db()
+    acc = get_account_by_id(account_id)
+    if not acc:
+        log.warning("account_get_credential_not_found id=%s", account_id)
+        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={account_id} 不存在。"))
+        return
+
+    conn = get_conn()
+    try:
+        creds_all = list_credentials_by_account(conn, account_id)
+        creds = [c for c in creds_all if (c.get("status") or "").strip().lower() == "active"]
+        if not creds:
+            _say(_err_json(
+                "ERR_CREDENTIAL_MISSING",
+                f"账号 id={account_id} 没有活跃凭据。",
+            ))
+            return
+        cred = creds[0]
+
+        base_out = {
+            "success": True,
+            "account_id": account_id,
+            "credential_id": cred["id"],
+            "credential_type": cred["credential_type"],
+            "secret_storage": cred["secret_storage"],
+            "secret_mask": cred["secret_mask"],
+        }
+
+        if not reveal:
+            _say(json.dumps(base_out, ensure_ascii=False))
+            return
+
+        tok = (lease_token or "").strip()
+        if not tok:
+            log.warning("account_get_credential_reveal_missing_token id=%s", account_id)
+            _say(_err_json(
+                "ERR_LEASE_INVALID",
+                "reveal 必须带 --lease-token。",
+            ))
+            return
+
+        lease = get_active_lease_by_token(conn, tok)
+        lease_prefix = tok[:8]
+        if lease is None:
+            log.warning(
+                "account_get_credential_reveal_bad_lease id=%s lease_prefix=%s",
+                account_id,
+                lease_prefix,
+            )
+            _say(_err_json(
+                "ERR_LEASE_INVALID",
+                "lease token 无效或已过期",
+            ))
+            return
+
+        if int(lease["account_id"]) != int(account_id):
+            log.warning(
+                "account_get_credential_reveal_lease_mismatch id=%s lease_account=%s lease_prefix=%s",
+                account_id,
+                lease["account_id"],
+                lease_prefix,
+            )
+            _say(_err_json(
+                "ERR_LEASE_INVALID",
+                "lease token 不属于该 account",
+            ))
+            return
+
+        try:
+            pt = read_credential_plaintext(conn, int(cred["id"]))
+        except MasterKeyMissingError as e:
+            log.warning("account_get_credential_master_missing id=%s", account_id)
+            _say(_err_json("ERR_MASTER_KEY_MISSING", str(e)))
+            return
+        except CredentialDecryptError as e:
+            log.warning(
+                "account_get_credential_decrypt_failed id=%s cred_id=%s lease_prefix=%s",
+                account_id,
+                cred["id"],
+                lease_prefix,
+            )
+            _say(_err_json("ERR_CREDENTIAL_DECRYPT_FAILED", str(e)))
+            return
+        except CredentialNotEncryptedError:
+            out = {**base_out, "plaintext": None, "note": "credential storage=none, no plaintext to reveal"}
+            log.info(
+                "account_get_credential_reveal_note_plain id=%s cred_id=%s lease_prefix=%s",
+                account_id,
+                cred["id"],
+                lease_prefix,
+            )
+            _say(json.dumps(out, ensure_ascii=False))
+            return
+
+        log.info(
+            "account_get_credential_reveal_success account_id=%s cred_id=%s lease_prefix=%s",
+            account_id,
+            cred["id"],
+            lease_prefix,
+        )
+        _say(json.dumps({**base_out, "plaintext": pt}, ensure_ascii=False))
+    finally:
+        conn.close()
+
+
+def cmd_get(account_id) -> None:
+    """Legacy: get <id> — same as account get <id>."""
+    cmd_account_get(account_id, include_credentials=False)
+
+
+def cmd_account_list(
+    platform_input: Optional[str] = None,
+    environment: Optional[str] = None,
+    tenant_id: Optional[str] = None,
+    role: Optional[str] = None,
+    limit: int = 200,
+) -> None:
+    """account list — output JSON array of accounts matching filters."""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=account_list platform=%s env=%s tenant=%s role=%s",
+             platform_input, environment, tenant_id, role)
+    init_db()
+    key = None
+    if platform_input and platform_input not in ("all", "全部", ""):
+        key = _resolve_or_fail(platform_input)
+        if not key:
+            return
+
+    accounts = find_accounts(
+        platform_key=key,
+        environment=environment,
+        tenant_id=tenant_id,
+        role=role,
+        limit=limit,
+    )
+    _say(json.dumps(accounts, ensure_ascii=False))
+
+
+def cmd_list_json(platform_input: str, limit: int = 200) -> None:
+    """Legacy: list-json <platform|all> [--limit N]"""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=list_json filter=%r limit=%s", platform_input, limit)
+    init_db()
+    raw = (platform_input or "all").strip()
+    if not raw or raw.lower() == "all" or raw == "全部":
+        key = None
+    else:
+        key = _resolve_or_fail(raw)
+        if not key:
+            return
+    lim = max(1, min(int(limit), 500))
+    accounts = find_accounts(platform_key=key, limit=lim)
+    _say(json.dumps(accounts, ensure_ascii=False))
+
+
+def cmd_account_pick_web(
+    platform_input: str,
+    environment: Optional[str] = None,
+    tenant_id: Optional[str] = None,
+    role: Optional[str] = None,
+    do_lease: bool = False,
+    ttl_sec: int = 900,
+    holder: Optional[str] = None,
+    purpose: Optional[str] = None,
+    platform_meta: Optional[PlatformCallerMeta] = None,
+) -> None:
+    """Pick a web/RPA account for use. Optionally acquire a lease."""
+    log = get_skill_logger()
+    log.info("account_pick_web_attempt platform=%s env=%s tenant=%s role=%s lease=%s",
+             platform_input, environment, tenant_id, role, do_lease)
+
+    key = _resolve_or_auto_register(platform_input, platform_meta)
+    if not key:
+        return
+
+    init_db()
+    conn = get_conn()
+    try:
+        seed_platforms(conn)
+    finally:
+        conn.close()
+
+    candidate_ids = find_account_ids(
+        platform_key=key,
+        environment=environment,
+        tenant_id=tenant_id,
+        role=role,
+    )
+    if not candidate_ids:
+        _say(_err_json("ERROR:NO_ACCOUNT",
+                       "没有符合条件的可用账号（status=active 且未被其他 lease 占用）。"))
+        return
+
+    picked_id = candidate_ids[0]
+    acc = get_account_by_id(picked_id)
+    if not acc:
+        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", "选中的账号突然不存在了。"))
+        return
+
+    update_account_last_used(picked_id)
+
+    result = {
+        "id": acc["id"],
+        "platform_key": acc["platform_key"],
+        "account_label": acc["account_label"],
+        "login_id": acc["login_id"],
+        "profile_dir": acc["profile_dir"],
+        "url": acc["url"],
+        "tenant_id": acc["tenant_id"],
+        "environment": acc["environment"],
+        "role": acc["role"],
+        "provider_code": acc["provider_code"],
+        "session_status": acc["session_status"],
+        "auth_strategy": acc["auth_strategy"],
+        "session_persistent": acc["session_persistent"],
+        "device_fingerprint": acc.get("device_fingerprint"),
+    }
+
+    lease_info = None
+    if do_lease:
+        lease_token = uuid.uuid4().hex[:32]
+        resolved_holder = holder or "unknown"
+        try:
+            lease_info = acquire_lease(
+                account_id=picked_id,
+                lease_token=lease_token,
+                holder=resolved_holder,
+                purpose=purpose,
+                ttl_sec=ttl_sec,
+            )
+            result["lease_token"] = lease_info["lease_token"]
+            result["lease_expires_at"] = lease_info["expires_at"]
+        except ValueError as e:
+            log.warning("account_pick_web_lease_conflict id=%s", picked_id)
+            _say(_err_json("ERROR:LEASE_CONFLICT", str(e)))
+            return
+
+    log.info("account_pick_web_success id=%s token_prefix=%s",
+             picked_id, lease_info["lease_token"][:8] if lease_info else "none")
+    _say(json.dumps(result, ensure_ascii=False))
+
+
+def cmd_pick_web(platform_input: str) -> None:
+    """Legacy: pick-web <platform> — same as account pick-web without lease."""
+    cmd_account_pick_web(platform_input, do_lease=False)

scripts/service/browser_service.py

@@ -13,8 +13,8 @@ import subprocess
 import sys
 import tempfile
 
-from db.connection import init_db
-from db.accounts_repo import get_account_by_id
+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,
@@ -33,13 +33,12 @@ def _win_find_exe(candidates):
     return None
 
 
-def resolve_chromium_channel():
+def resolve_chromium_executable():
     """
-    Playwright channel: 'chrome' | 'msedge' | None
-    Windows 优先 Chrome，其次 Edge；其它系统默认 chrome（由 Playwright 解析 PATH）。
+    Windows 上返回 chrome.exe 或 msedge.exe 的绝对路径；未找到或非 Windows 返回 None。
     """
     if sys.platform != "win32":
-        return "chrome"
+        return None
 
     pf = os.environ.get("ProgramFiles", r"C:\Program Files")
     pfx86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)")
@@ -53,7 +52,7 @@ def resolve_chromium_channel():
         ]
     )
     if chrome:
-        return "chrome"
+        return chrome
 
     edge = _win_find_exe(
         [
@@ -63,17 +62,58 @@ def resolve_chromium_channel():
         ]
     )
     if edge:
-        return "msedge"
+        return edge
 
     return None
 
 
+def resolve_chromium_channel():
+    """
+    Playwright channel: 'chrome' | 'msedge' | None
+    Windows 优先 Chrome，其次 Edge；其它系统默认 chrome（由 Playwright 解析 PATH）。
+    """
+    if sys.platform != "win32":
+        return "chrome"
+
+    exe = resolve_chromium_executable()
+    if not exe:
+        return None
+    if os.path.basename(exe).lower() == "msedge.exe":
+        return "msedge"
+    return "chrome"
+
+
 def _print_browser_install_hint():
     print("❌ 未检测到 Google Chrome 或 Microsoft Edge，请先安装：")
     print("   • Chrome: https://www.google.com/chrome/")
     print("   • Edge:   https://www.microsoft.com/zh-cn/edge")
 
 
+def resolve_account_browser_url(account: dict, conn=None) -> str:
+    """account.url 优先，其次 platforms 表 default_url，最后内置 PLATFORMS。"""
+    url = (account.get("url") or "").strip()
+    if url:
+        return url
+    pk = (account.get("platform_key") or "").strip()
+    if not pk:
+        return ""
+
+    close_conn = False
+    if conn is None:
+        init_db()
+        conn = get_conn()
+        close_conn = True
+    try:
+        plat = get_platform_from_db(conn, pk)
+        if plat and (plat.get("default_url") or "").strip():
+            return plat["default_url"].strip()
+    finally:
+        if close_conn:
+            conn.close()
+
+    return (PLATFORMS.get(pk, {}).get("default_url") or "").strip()
+
+
 def cmd_open(account_id):
     """
     打开该账号的持久化浏览器，仅用于肉眼核对。
@@ -109,9 +149,11 @@ def cmd_open(account_id):
         return
     os.makedirs(profile_dir, exist_ok=True)
 
-    # Use account url, fall back to platform default
-    platform_spec = PLATFORMS.get(target["platform_key"], {})
-    url = (target.get("url") or "").strip() or platform_spec.get("default_url", "")
+    conn = get_conn()
+    try:
+        url = resolve_account_browser_url(target, conn=conn)
+    finally:
+        conn.close()
 
     log.info(
         "browser_open_context profile_dir=%s url=%s channel=%s platform=%s",

scripts/service/credential_commands.py

@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+"""Credential pick and resolve CLI commands."""
+from typing import Optional
+
+from db.accounts_repo import find_credentials, get_credential_by_id, update_credential_last_used
+from db.connection import init_db
+from service._svc_common import (
+    _credential_ok_line,
+    _err_json,
+    _map_secret_read_error,
+    _resolve_or_fail,
+    _say,
+)
+from service.secret_store import read_secret
+from util.logging_config import get_skill_logger
+
+
+def cmd_credential_pick(
+    platform_input: str,
+    credential_type: Optional[str] = None,
+    environment: Optional[str] = None,
+    role: Optional[str] = None,
+    reveal: bool = False,
+) -> None:
+    """Pick a credential for a platform, optionally with reveal."""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=credential_pick platform=%s type=%s env=%s role=%s reveal=%s",
+             platform_input, credential_type, environment, role, reveal)
+
+    key = _resolve_or_fail(platform_input)
+    if not key:
+        return
+
+    init_db()
+    creds = find_credentials(
+        platform_key=key,
+        credential_type=credential_type,
+        environment=environment,
+        role=role,
+        status="active",
+        limit=1,
+    )
+    if not creds:
+        _say(_err_json("ERROR:CREDENTIAL_NOT_FOUND",
+                       "没有符合条件的可用凭据。"))
+        return
+
+    cred = creds[0]
+    update_credential_last_used(cred["id"])
+
+    result = {
+        "id": cred["id"],
+        "platform_key": cred["platform_key"],
+        "credential_label": cred["credential_label"],
+        "credential_type": cred["credential_type"],
+        "secret_storage": cred["secret_storage"],
+        "secret_ref": cred["secret_ref"],
+        "secret_mask": cred["secret_mask"],
+        "status": cred["status"],
+    }
+
+    if reveal:
+        try:
+            secret_value = read_secret(cred["secret_storage"], cred["secret_ref"])
+            result["secret"] = secret_value
+            log.info(
+                "credential_pick_reveal_success id=%s storage=%s",
+                cred["id"],
+                cred["secret_storage"],
+            )
+        except (ValueError, OSError) as e:
+            code, msg = _map_secret_read_error(e)
+            log.error("credential_pick_reveal_failure id=%s error=%s", cred["id"], msg)
+            _say(_err_json(code, msg))
+            return
+    else:
+        log.info("credential_pick_success id=%s (masked)", cred["id"])
+
+    _say(_credential_ok_line(result))
+
+
+def cmd_credential_resolve(credential_id: int, reveal: bool = False) -> None:
+    """Resolve a credential by id."""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=credential_resolve id=%s reveal=%s", credential_id, reveal)
+
+    init_db()
+    cred = get_credential_by_id(credential_id)
+    if not cred:
+        _say(_err_json("ERROR:CREDENTIAL_NOT_FOUND", f"凭据 id={credential_id} 不存在。"))
+        return
+    if cred["status"] != "active":
+        _say(_err_json("ERROR:CREDENTIAL_DISABLED", f"凭据 id={credential_id} 状态为 {cred['status']}，不可用。"))
+        return
+
+    result = {
+        "id": cred["id"],
+        "account_id": cred["account_id"],
+        "platform_key": cred["platform_key"],
+        "credential_label": cred["credential_label"],
+        "credential_type": cred["credential_type"],
+        "secret_storage": cred["secret_storage"],
+        "secret_ref": cred["secret_ref"],
+        "secret_mask": cred["secret_mask"],
+        "status": cred["status"],
+    }
+
+    if reveal:
+        try:
+            secret_value = read_secret(cred["secret_storage"], cred["secret_ref"])
+            result["secret"] = secret_value
+            log.info("credential_resolve_reveal_success id=%s", credential_id)
+        except (ValueError, OSError) as e:
+            code, msg = _map_secret_read_error(e)
+            log.error("credential_resolve_reveal_failure id=%s error=%s", credential_id, msg)
+            _say(_err_json(code, msg))
+            return
+    else:
+        log.info("credential_resolve_success id=%s (masked)", credential_id)
+
+    _say(_credential_ok_line(result))

scripts/service/lease_commands.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+"""Lease management CLI commands."""
+from db.accounts_repo import cleanup_expired_leases, list_active_leases, release_lease
+from service._svc_common import _err_json, _ok_json, _say
+from util.logging_config import get_skill_logger
+
+
+def cmd_lease_release(lease_token: str) -> None:
+    """lease release <token>"""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=lease_release token_prefix=%s", lease_token[:8])
+    ok = release_lease(lease_token)
+    if ok:
+        _say(_ok_json({"lease_token": lease_token, "status": "released"}))
+    else:
+        _say(_err_json("ERROR:LEASE_NOT_FOUND", "未找到活跃的租约，或租约已释放/过期。"))
+
+
+def cmd_lease_list(active_only: bool = False) -> None:
+    """lease list [--active]"""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=lease_list active_only=%s", active_only)
+    leases = list_active_leases()
+    if active_only:
+        pass
+    _say(_ok_json({"leases": leases}))
+
+
+def cmd_lease_cleanup() -> None:
+    """lease cleanup — mark expired leases as expired."""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=lease_cleanup")
+    count = cleanup_expired_leases()
+    _say(_ok_json({"cleaned": count}))

scripts/service/legacy_commands.py

@@ -0,0 +1,128 @@
+# -*- coding: utf-8 -*-
+"""Legacy delete and human-readable list CLI commands."""
+from db.accounts_repo import get_account_by_id, find_accounts
+from db.connection import get_conn, init_db
+from service._svc_common import (
+    _remove_profile_dir,
+    _resolve_or_fail,
+    _say,
+    _say_err,
+)
+from util.logging_config import get_skill_logger
+from util.platforms import get_platform_spec
+from util.runtime_paths import _runtime_paths_debug_text
+
+
+def cmd_delete_by_id(account_id) -> None:
+    """delete id <id> — legacy"""
+    log = get_skill_logger()
+    log.info("delete_by_id id=%r", account_id)
+    acc = get_account_by_id(account_id)
+    if not acc:
+        _say(f"ERROR:ACCOUNT_NOT_FOUND 账号 id={account_id} 不存在。")
+        _say_err(_runtime_paths_debug_text())
+        return
+    profile_dir = acc.get("profile_dir", "")
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("DELETE FROM accounts WHERE id = ?", (int(account_id),))
+        cur.execute("DELETE FROM credentials WHERE account_id = ?", (int(account_id),))
+        conn.commit()
+    finally:
+        conn.close()
+    _remove_profile_dir(profile_dir)
+    log.info("delete_by_id_done id=%s", account_id)
+    _say(f"✅ 已删除账号 ID {account_id}。")
+
+
+def cmd_delete_by_platform(platform_input: str) -> None:
+    """delete platform <platform> — legacy"""
+    log = get_skill_logger()
+    log.info("delete_by_platform platform=%r", platform_input)
+    key = _resolve_or_fail(platform_input)
+    if not key:
+        return
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT id, profile_dir FROM accounts WHERE platform_key = ?", (key,))
+        rows = cur.fetchall()
+        if not rows:
+            _say("ERROR:NO_ACCOUNTS_FOUND 该平台下没有账号记录。")
+            return
+        for rid, pdir in rows:
+            _remove_profile_dir(pdir)
+        cur.execute("DELETE FROM accounts WHERE platform_key = ?", (key,))
+        cur.execute(
+            "DELETE FROM credentials WHERE platform_key = ? AND account_id IS NOT NULL",
+            (key,),
+        )
+        conn.commit()
+        log.info("delete_by_platform_done platform=%s count=%s", key, len(rows))
+        display = get_platform_spec(key, conn=conn).get("display_name", key)
+        _say(f"✅ 已删除 {display} 下共 {len(rows)} 条账号。")
+    finally:
+        conn.close()
+
+
+def cmd_delete_by_platform_phone(platform_input: str, phone: str) -> None:
+    """Legacy: delete by platform+phone — now uses login_id matching"""
+    log = get_skill_logger()
+    log.info("delete_by_platform_phone platform=%r", platform_input)
+    key = _resolve_or_fail(platform_input)
+    if not key:
+        return
+    phone_norm = "".join(c for c in phone if c.isdigit())
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute(
+            "SELECT id, profile_dir FROM accounts WHERE platform_key = ? AND login_id = ?",
+            (key, phone_norm),
+        )
+        row = cur.fetchone()
+        if not row:
+            _say("ERROR:ACCOUNT_NOT_FOUND 该平台下无此手机号。")
+            return
+        rid, pdir = row
+        _remove_profile_dir(pdir)
+        cur.execute("DELETE FROM accounts WHERE id = ?", (rid,))
+        cur.execute("DELETE FROM credentials WHERE account_id = ?", (rid,))
+        conn.commit()
+        log.info("delete_by_platform_phone_done id=%s", rid)
+        _say(f"✅ 已删除账号 ID {rid}。")
+    finally:
+        conn.close()
+
+
+def cmd_list(platform="all", limit: int = 10) -> None:
+    """Legacy: human-readable list output."""
+    log = get_skill_logger()
+    log.info("list filter=%r", platform)
+    init_db()
+    accounts = find_accounts(
+        platform_key=None if (not platform or platform in ("all", "全部")) else platform,
+        limit=limit,
+    )
+    if not accounts:
+        _say("ERROR:NO_ACCOUNTS_FOUND")
+        _say_err(_runtime_paths_debug_text())
+        return
+    log.info("list_ok rows=%s", len(accounts))
+    for i, acc in enumerate(accounts):
+        _say(f"id：{acc['id']}")
+        _say(f"label：{acc['account_label']}")
+        _say(f"platform：{acc['platform_key']}")
+        _say(f"login_id：{acc['login_id']}")
+        _say(f"env：{acc['environment']}  role：{acc['role']}")
+        _say(f"status：{acc['status']}  session：{acc['session_status']}")
+        _say(f"profile_dir：{acc['profile_dir']}")
+        _say(f"url：{acc['url']}")
+        _say(f"created_at：{acc['created_at']}")
+        _say(f"updated_at：{acc['updated_at']}")
+        if acc.get("last_error_code"):
+            _say(f"last_error：{acc['last_error_code']} {acc.get('last_error_message', '')}")
+        if i < len(accounts) - 1:
+            _say("_______________________________________")
+            _say("")

scripts/service/platform_commands.py

@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+"""Platform management CLI commands."""
+import json
+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.logging_config import get_skill_logger
+from util.platforms import DEFAULT_CAPABILITIES_JSON, is_valid_platform_key_format, seed_platforms
+
+
+def cmd_platform_list(domain: Optional[str] = None) -> None:
+    """platform list [--domain logistics|llm|content]"""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=platform_list domain=%s", domain)
+    init_db()
+    conn = get_conn()
+    try:
+        seed_platforms(conn)
+        platforms = list_platforms_from_db(conn, domain=domain)
+    finally:
+        conn.close()
+    _say(_ok_json({"platforms": platforms}))
+
+
+def cmd_platform_get(input_key: str) -> None:
+    """platform get <platform>"""
+    log = get_skill_logger()
+    log.info("cli_start subcommand=platform_get key=%s", input_key)
+    key = _resolve_or_fail(input_key)
+    if not key:
+        return
+    init_db()
+    conn = get_conn()
+    try:
+        plat = get_platform_from_db(conn, key)
+    finally:
+        conn.close()
+    if plat:
+        _say(_ok_json(plat))
+    else:
+        _say(_err_json(
+            "ERROR:PLATFORM_NOT_REGISTERED",
+            f"Platform '{key}' not found in DB.",
+        ))
+
+
+def cmd_platform_ensure(
+    key: str,
+    display_name: str,
+    domain: str = "generic",
+    url: str = "",
+    auth_strategy: Optional[str] = None,
+    aliases: Optional[str] = None,
+    capabilities: Optional[str] = None,
+) -> int:
+    """
+    幂等地注册或更新平台到 DB。
+    输出单行 JSON：{"ok": true, "platform_key": "xxx", "action": "created|updated"}
+    """
+    log = get_skill_logger()
+    key = (key or "").strip()
+    display_name = (display_name or key).strip()
+    if not key:
+        _say('ERROR:INVALID_ARGS platform ensure 需要 --key 参数')
+        return 1
+    if not is_valid_platform_key_format(key):
+        _say(_err_json(
+            "ERROR:INVALID_PLATFORM_KEY",
+            f"平台 key 格式非法：「{key}」。建议使用小写字母、数字、下划线或连字符。",
+        ))
+        return 1
+
+    aliases_json = aliases if aliases else json.dumps([key, display_name], ensure_ascii=False)
+    capabilities_json = capabilities if capabilities else DEFAULT_CAPABILITIES_JSON
+
+    init_db()
+    conn = get_conn()
+    try:
+        existed = bool(
+            conn.execute(
+                "SELECT 1 FROM platforms WHERE platform_key = ?", (key,)
+            ).fetchone()
+        )
+        upsert_platform(
+            conn,
+            key=key,
+            display_name=display_name,
+            domain=domain,
+            default_url=url,
+            aliases_json=aliases_json,
+            capabilities_json=capabilities_json,
+            default_auth_strategy=auth_strategy,
+        )
+    finally:
+        conn.close()
+    action = "updated" if existed else "created"
+    log.info("platform_ensure key=%s action=%s", key, action)
+    _say(json.dumps({"ok": True, "platform_key": key, "action": action}, ensure_ascii=False))
+    return 0

scripts/util/platforms.py

@@ -6,11 +6,17 @@ Central authoritative registry for all platforms (logistics, LLM, content, etc.)
 Import and use PLATFORMS dict, resolve_platform_key(), and seed_platforms().
 """
 import json
+import re
 import time
+from dataclasses import dataclass
 from typing import Any, Optional
 
 from util.logging_config import get_skill_logger
 
+PLATFORM_KEY_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]{1,63}$")
+
+DEFAULT_CAPABILITIES_JSON = '{"web": true, "api_key": false, "rpa": true}'
+
 # ---------------------------------------------------------------------------
 # Platform specification fields:
 #   display_name       – Human-readable name (e.g. "Maersk")
@@ -309,7 +315,7 @@ _PLATFORM_ALIAS_MAP = _build_alias_map()
 
 def resolve_platform_key(name: str) -> Optional[str]:
     """
-    Resolve user input to a canonical platform key.
+    Resolve user input to a canonical platform key (built-in PLATFORMS only).
     Supports: key, display_name, any alias. Case-insensitive.
     Returns None if unrecognized.
     """
@@ -330,8 +336,154 @@ def resolve_platform_key(name: str) -> Optional[str]:
     return None
 
 
+def is_valid_platform_key_format(name: str) -> bool:
+    """平台 key 格式：^[a-z0-9][a-z0-9_-]{1,63}$"""
+    s = (name or "").strip().lower()
+    if not s:
+        return False
+    return bool(PLATFORM_KEY_PATTERN.match(s))
+
+
+def _parse_aliases_json(val: Any) -> list[str]:
+    if val is None:
+        return []
+    try:
+        data = json.loads(val) if isinstance(val, str) else val
+        if isinstance(data, list):
+            return [str(a).strip() for a in data if str(a).strip()]
+    except (json.JSONDecodeError, TypeError, ValueError):
+        pass
+    return []
+
+
+@dataclass(frozen=True)
+class PlatformResolveResult:
+    platform_key: Optional[str] = None
+    error_code: Optional[str] = None
+    message: Optional[str] = None
+
+    @property
+    def ok(self) -> bool:
+        return bool(self.platform_key) and not self.error_code
+
+
+def _resolve_platform_from_db(conn, raw: str) -> PlatformResolveResult:
+    """在 platforms 表中解析：精确 key（enabled=1）与 aliases_json。"""
+    raw_stripped = (raw or "").strip()
+    raw_lower = raw_stripped.lower()
+
+    if is_valid_platform_key_format(raw_stripped):
+        row = conn.execute(
+            "SELECT platform_key, enabled FROM platforms WHERE platform_key = ?",
+            (raw_lower,),
+        ).fetchone()
+        if row:
+            if not row[1]:
+                return PlatformResolveResult(
+                    error_code="PLATFORM_DISABLED",
+                    message=f"平台 {row[0]} 已禁用。",
+                )
+            return PlatformResolveResult(platform_key=row[0])
+
+    rows = conn.execute(
+        "SELECT platform_key, enabled, aliases_json FROM platforms"
+    ).fetchall()
+    for platform_key, enabled, aliases_json in rows:
+        if not enabled:
+            continue
+        names = {platform_key.lower()}
+        for alias in _parse_aliases_json(aliases_json):
+            names.add(alias.lower())
+            names.add(alias)
+        if raw_lower in names or raw_stripped in names:
+            return PlatformResolveResult(platform_key=platform_key)
+
+    return PlatformResolveResult()
+
+
+def resolve_platform_key_dynamic(input_name: str, conn=None) -> PlatformResolveResult:
+    """
+    统一平台解析：内置 PLATFORMS → DB platforms 表 → 错误分类。
+    """
+    raw = (input_name or "").strip()
+    if not raw:
+        return PlatformResolveResult(
+            error_code="INVALID_PLATFORM",
+            message="平台名不能为空。",
+        )
+
+    builtin = resolve_platform_key(raw)
+    if builtin:
+        return PlatformResolveResult(platform_key=builtin)
+
+    close_conn = False
+    if conn is None:
+        from db.connection import get_conn, init_db
+
+        init_db()
+        conn = get_conn()
+        close_conn = True
+    try:
+        db_result = _resolve_platform_from_db(conn, raw)
+        if db_result.ok or db_result.error_code:
+            return db_result
+    finally:
+        if close_conn:
+            conn.close()
+
+    if not is_valid_platform_key_format(raw):
+        return PlatformResolveResult(
+            error_code="INVALID_PLATFORM_KEY",
+            message=(
+                f"平台 key 格式非法：「{raw}」。"
+                "建议使用小写字母、数字、下划线或连字符，例如 jiangchang。"
+            ),
+        )
+
+    key_hint = raw.lower()
+    return PlatformResolveResult(
+        error_code="PLATFORM_NOT_REGISTERED",
+        message=(
+            f"平台 {key_hint} 尚未注册，请先执行 "
+            f"account platform ensure --key {key_hint} --display-name <名称> [--url <url>]"
+        ),
+    )
+
+
+def get_platform_spec(platform_key: str, conn=None) -> dict[str, Any]:
+    """内置 PLATFORMS 优先结构；自定义平台从 DB platforms 表读取。"""
+    spec = PLATFORMS.get(platform_key)
+    if spec:
+        return dict(spec)
+
+    close_conn = False
+    if conn is None:
+        from db.connection import get_conn, init_db
+
+        init_db()
+        conn = get_conn()
+        close_conn = True
+    try:
+        from db.accounts_repo import get_platform_from_db
+
+        plat = get_platform_from_db(conn, platform_key)
+        if plat:
+            return {
+                "display_name": plat["display_name"],
+                "domain": plat.get("domain") or "generic",
+                "provider_code": plat.get("provider_code") or platform_key,
+                "default_url": plat.get("default_url") or "",
+                "aliases": plat.get("aliases") or [],
+                "capabilities": plat.get("capabilities") or {},
+            }
+    finally:
+        if close_conn:
+            conn.close()
+    return {}
+
+
 def get_platform(key: str) -> Optional[dict[str, Any]]:
-    """Return the full platform spec dict for a validated key."""
+    """Return the full platform spec dict for a validated built-in key."""
     return PLATFORMS.get(key)
 
 

scripts/util/profile_shortcut.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+"""在 Profile 目录内创建浏览器快捷方式（Windows .lnk）。"""
+from __future__ import annotations
+
+import logging
+import os
+import re
+import subprocess
+import sys
+from typing import Optional
+
+from service.browser_service import resolve_chromium_executable
+
+_WIN_LNK_FORBIDDEN = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
+
+
+def _lnk_safe_filename(name: str, fallback: str = "browser") -> str:
+    t = _WIN_LNK_FORBIDDEN.sub("_", (name or "").strip())
+    t = t.rstrip(" .")
+    return t or fallback
+
+
+def _ps_single_quote(value: str) -> str:
+    return "'" + value.replace("'", "''") + "'"
+
+
+def create_profile_browser_shortcut(
+    profile_dir: str,
+    account_label: str,
+    log: Optional[logging.Logger] = None,
+) -> Optional[str]:
+    """
+    在 profile_dir 内创建 {account_label}.lnk，目标为系统 Chrome/Edge + --user-data-dir。
+
+    非 Windows 或未安装 Chrome/Edge 时静默跳过，返回 None。
+    成功返回 .lnk 绝对路径。
+    """
+    if log is None:
+        log = logging.getLogger(__name__)
+
+    if sys.platform != "win32":
+        log.debug("profile_shortcut_skip platform=%s", sys.platform)
+        return None
+
+    browser_exe = resolve_chromium_executable()
+    if not browser_exe:
+        log.warning("profile_shortcut_skip reason=no_chromium")
+        return None
+
+    profile_abs = os.path.abspath(profile_dir)
+    os.makedirs(profile_abs, exist_ok=True)
+
+    lnk_name = f"{_lnk_safe_filename(account_label)}.lnk"
+    lnk_path = os.path.join(profile_abs, lnk_name)
+    arguments = f'--user-data-dir="{profile_abs}"'
+    working_dir = os.path.dirname(browser_exe)
+
+    ps = (
+        "$WshShell = New-Object -ComObject WScript.Shell; "
+        f"$s = $WshShell.CreateShortcut({_ps_single_quote(lnk_path)}); "
+        f"$s.TargetPath = {_ps_single_quote(browser_exe)}; "
+        f"$s.Arguments = {_ps_single_quote(arguments)}; "
+        f"$s.IconLocation = {_ps_single_quote(browser_exe + ',0')}; "
+        f"$s.WorkingDirectory = {_ps_single_quote(working_dir)}; "
+        "$s.Save()"
+    )
+
+    try:
+        subprocess.run(
+            ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+    except (OSError, subprocess.CalledProcessError) as exc:
+        log.warning("profile_shortcut_failed profile_dir=%s error=%s", profile_abs, exc)
+        return None
+
+    if not os.path.isfile(lnk_path):
+        log.warning("profile_shortcut_missing_after_create path=%s", lnk_path)
+        return None
+
+    log.info("profile_shortcut_created path=%s browser=%s", lnk_path, browser_exe)
+    return lnk_path

tests/test_custom_platforms.py

@@ -0,0 +1,398 @@
+# -*- coding: utf-8 -*-
+"""自定义平台：动态解析、platform ensure、add-web / pick-web / browser URL。"""
+from __future__ import annotations
+
+import json
+import os
+
+import pytest
+
+
+JIANGCHANG_URL = "https://jc2009.com/admin/index.html"
+
+
+def _ensure_jiangchang(clean_db) -> None:
+    from service.account_service import cmd_platform_ensure
+
+    rc = cmd_platform_ensure(
+        key="jiangchang",
+        display_name="匠厂后台",
+        domain="generic",
+        url=JIANGCHANG_URL,
+        auth_strategy=None,
+        aliases=json.dumps(["jc", "匠厂后台"], ensure_ascii=False),
+        capabilities=None,
+    )
+    assert rc == 0
+
+
+class TestBuiltinPlatformsUnchanged:
+    def test_douyin_alias(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        result = resolve_platform_key_dynamic("抖音", conn=clean_db)
+        assert result.ok
+        assert result.platform_key == "douyin"
+
+    def test_toutiao_key(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        result = resolve_platform_key_dynamic("toutiao", conn=clean_db)
+        assert result.ok
+        assert result.platform_key == "toutiao"
+
+
+class TestDynamicPlatformResolution:
+    def test_unregistered_valid_key(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        result = resolve_platform_key_dynamic("unknown_custom", conn=clean_db)
+        assert not result.ok
+        assert result.error_code == "PLATFORM_NOT_REGISTERED"
+        assert "platform ensure" in (result.message or "")
+
+    def test_invalid_platform_key_format(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        result = resolve_platform_key_dynamic("江厂 空格", conn=clean_db)
+        assert not result.ok
+        assert result.error_code == "INVALID_PLATFORM_KEY"
+
+    def test_empty_platform(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        result = resolve_platform_key_dynamic("", conn=clean_db)
+        assert result.error_code == "INVALID_PLATFORM"
+
+    def test_registered_jiangchang(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        _ensure_jiangchang(clean_db)
+        result = resolve_platform_key_dynamic("jiangchang", conn=clean_db)
+        assert result.ok
+        assert result.platform_key == "jiangchang"
+
+    def test_registered_alias_jc(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        _ensure_jiangchang(clean_db)
+        result = resolve_platform_key_dynamic("jc", conn=clean_db)
+        assert result.ok
+        assert result.platform_key == "jiangchang"
+
+
+class TestPlatformEnsure:
+    def test_ensure_returns_json(self, clean_db, capsys):
+        from service.account_service import cmd_platform_ensure
+
+        rc = cmd_platform_ensure(
+            key="jiangchang",
+            display_name="匠厂后台",
+            url=JIANGCHANG_URL,
+        )
+        assert rc == 0
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["ok"] is True
+        assert payload["platform_key"] == "jiangchang"
+        assert payload["action"] in ("created", "updated")
+
+
+class TestCustomPlatformAccounts:
+    def test_add_web_after_ensure(self, clean_db):
+        from service.account_service import cmd_account_add_web
+        from db.accounts_repo import find_accounts
+
+        _ensure_jiangchang(clean_db)
+        cmd_account_add_web(
+            platform_input="jiangchang",
+            login_id="manual",
+            login_id_type="manual",
+            label="匠厂后台 手工登录",
+            tenant_id=None,
+            environment="production",
+            role="admin",
+            provider_code=None,
+            url=None,
+            extra_json_str=None,
+            profile_dir_override=None,
+        )
+        accs = find_accounts(platform_key="jiangchang")
+        assert len(accs) == 1
+        assert accs[0]["platform_key"] == "jiangchang"
+        assert accs[0]["url"] == JIANGCHANG_URL
+        assert accs[0]["profile_dir"]
+
+    def test_pick_web_after_ensure(self, clean_db, capsys):
+        from service.account_service import cmd_account_add_web, cmd_account_pick_web
+
+        _ensure_jiangchang(clean_db)
+        cmd_account_add_web(
+            platform_input="jiangchang",
+            login_id="manual",
+            login_id_type="manual",
+            label="匠厂后台",
+            tenant_id=None,
+            environment="production",
+            role="admin",
+            provider_code=None,
+            url=None,
+            extra_json_str=None,
+            profile_dir_override=None,
+        )
+        capsys.readouterr()
+        cmd_account_pick_web(
+            platform_input="jiangchang",
+            do_lease=True,
+            holder="test-holder",
+            ttl_sec=300,
+        )
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["platform_key"] == "jiangchang"
+        assert payload["profile_dir"]
+        assert payload.get("lease_token")
+
+    def test_pick_web_unregistered_returns_structured_error(self, clean_db, capsys):
+        from service.account_service import cmd_account_pick_web
+
+        cmd_account_pick_web(platform_input="unknown_custom")
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["success"] is False
+        assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
+        assert "--platform-display-name" in payload["error"]["message"]
+
+    def test_add_web_unregistered_returns_structured_error(self, clean_db, capsys):
+        from service.account_service import cmd_account_add_web
+
+        cmd_account_add_web(
+            platform_input="unknown_custom",
+            login_id="manual",
+            login_id_type="manual",
+            label="x",
+            tenant_id=None,
+            environment="production",
+            role="admin",
+            provider_code=None,
+            url=None,
+            extra_json_str=None,
+            profile_dir_override=None,
+        )
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["success"] is False
+        assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
+
+
+class TestBrowserServiceUrlResolution:
+    def test_custom_platform_default_url_from_db(self, clean_db):
+        from db.accounts_repo import insert_account, upsert_platform
+        from service.browser_service import resolve_account_browser_url
+        import time
+
+        now = int(time.time())
+        upsert_platform(
+            clean_db,
+            key="jiangchang",
+            display_name="匠厂后台",
+            default_url=JIANGCHANG_URL,
+        )
+        aid = insert_account(
+            conn=clean_db,
+            platform_key="jiangchang",
+            account_label="test",
+            login_id=None,
+            login_id_type="manual",
+            tenant_id=None,
+            environment="production",
+            role="admin",
+            provider_code="jiangchang",
+            profile_dir="/tmp/test-profile",
+            url="",
+            extra_json="{}",
+            now=now,
+            auth_strategy="qr_code_manual",
+        )
+        clean_db.commit()
+        account = {
+            "id": aid,
+            "platform_key": "jiangchang",
+            "url": "",
+        }
+        assert resolve_account_browser_url(account, conn=clean_db) == JIANGCHANG_URL
+
+    def test_account_url_overrides_platform_default(self, clean_db):
+        from service.browser_service import resolve_account_browser_url
+
+        account = {
+            "platform_key": "jiangchang",
+            "url": "https://custom.example.com/admin",
+        }
+        assert (
+            resolve_account_browser_url(account, conn=clean_db)
+            == "https://custom.example.com/admin"
+        )
+
+
+class TestPlatformKeyFormatValidation:
+    def test_ensure_rejects_invalid_key(self, clean_db, capsys):
+        from service.account_service import cmd_platform_ensure
+
+        rc = cmd_platform_ensure(key="江厂 空格", display_name="bad")
+        assert rc == 1
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["error"]["code"] == "ERROR:INVALID_PLATFORM_KEY"
+
+
+JIANGCHANG_META_KWARGS = dict(
+    display_name="匠厂后台",
+    url=JIANGCHANG_URL,
+    domain="jc2009.com",
+    auth_strategy="qr_code_manual",
+    aliases=["jiangchang", "匠厂", "匠厂后台", "jc2009"],
+)
+
+
+class TestAutoRegisterFromCallerMeta:
+    def _meta(self):
+        from service.account_service import PlatformCallerMeta
+
+        return PlatformCallerMeta(**JIANGCHANG_META_KWARGS)
+
+    def test_pick_web_existing_platform_unchanged(self, clean_db, capsys):
+        from service.account_service import cmd_account_add_web, cmd_account_pick_web
+
+        _ensure_jiangchang(clean_db)
+        cmd_account_add_web(
+            platform_input="jiangchang",
+            login_id="manual",
+            login_id_type="manual",
+            label="匠厂后台",
+            tenant_id=None,
+            environment="production",
+            role="admin",
+            provider_code=None,
+            url=None,
+            extra_json_str=None,
+            profile_dir_override=None,
+        )
+        capsys.readouterr()
+        cmd_account_pick_web(platform_input="jiangchang", do_lease=True, holder="h1")
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["platform_key"] == "jiangchang"
+        assert payload.get("profile_dir")
+
+    def test_pick_web_unregistered_without_meta(self, clean_db, capsys):
+        from service.account_service import cmd_account_pick_web
+
+        cmd_account_pick_web(platform_input="auto_reg_plat")
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["success"] is False
+        assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED"
+        assert "--platform-display-name" in payload["error"]["message"]
+
+    def test_pick_web_unregistered_with_meta_auto_registers(self, clean_db, capsys):
+        from service.account_service import cmd_account_pick_web, PlatformCallerMeta
+        from util.platforms import resolve_platform_key_dynamic
+
+        meta = PlatformCallerMeta(
+            display_name="自动注册测试",
+            url="https://example.com/admin",
+            domain="example.com",
+            auth_strategy="qr_code_manual",
+            aliases=["auto_reg_plat"],
+        )
+        cmd_account_pick_web(platform_input="auto_reg_plat", platform_meta=meta)
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["success"] is False
+        assert payload["error"]["code"] == "ERROR:NO_ACCOUNT"
+
+        result = resolve_platform_key_dynamic("auto_reg_plat", conn=clean_db)
+        assert result.ok
+        assert result.platform_key == "auto_reg_plat"
+
+    def test_add_web_unregistered_with_meta_auto_registers(self, clean_db):
+        from service.account_service import cmd_account_add_web, PlatformCallerMeta
+        from db.accounts_repo import find_accounts
+
+        meta = PlatformCallerMeta(
+            display_name="匠厂后台",
+            url=JIANGCHANG_URL,
+            domain="jc2009.com",
+            auth_strategy="qr_code_manual",
+            aliases=["jiangchang"],
+        )
+        cmd_account_add_web(
+            platform_input="jiangchang",
+            login_id="manual",
+            login_id_type="manual",
+            label="匠厂后台",
+            tenant_id=None,
+            environment="production",
+            role="admin",
+            provider_code=None,
+            url=None,
+            extra_json_str=None,
+            profile_dir_override=None,
+            platform_meta=meta,
+        )
+        accs = find_accounts(platform_key="jiangchang")
+        assert len(accs) == 1
+        assert accs[0]["url"] == JIANGCHANG_URL
+
+    def test_invalid_platform_key_no_register(self, clean_db, capsys):
+        from service.account_service import cmd_account_pick_web, PlatformCallerMeta
+
+        meta = PlatformCallerMeta(display_name="坏", url="https://x.com")
+        cmd_account_pick_web(platform_input="bad key!", platform_meta=meta)
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["error"]["code"] == "ERROR:INVALID_PLATFORM_KEY"
+        row = clean_db.execute(
+            "SELECT COUNT(*) FROM platforms WHERE platform_key LIKE '%bad%'"
+        ).fetchone()[0]
+        assert row == 0
+
+    def test_disabled_platform_not_re_enabled(self, clean_db, capsys):
+        from db.accounts_repo import upsert_platform
+        from service.account_service import cmd_account_pick_web, PlatformCallerMeta
+
+        upsert_platform(
+            clean_db,
+            key="jiangchang",
+            display_name="匠厂后台",
+            default_url=JIANGCHANG_URL,
+        )
+        clean_db.execute("UPDATE platforms SET enabled = 0 WHERE platform_key = 'jiangchang'")
+        clean_db.commit()
+
+        meta = PlatformCallerMeta(display_name="匠厂后台", url=JIANGCHANG_URL)
+        cmd_account_pick_web(platform_input="jiangchang", platform_meta=meta)
+        out = capsys.readouterr().out.strip()
+        payload = json.loads(out)
+        assert payload["error"]["code"] == "ERROR:PLATFORM_DISABLED"
+
+        enabled = clean_db.execute(
+            "SELECT enabled FROM platforms WHERE platform_key = 'jiangchang'"
+        ).fetchone()[0]
+        assert enabled == 0
+
+    def test_auto_register_idempotent(self, clean_db):
+        from service.account_service import cmd_account_pick_web, PlatformCallerMeta
+
+        meta = PlatformCallerMeta(
+            display_name="幂等测试",
+            url="https://example.com/admin",
+            aliases=["idem_plat"],
+        )
+        cmd_account_pick_web(platform_input="idem_plat", platform_meta=meta)
+        cmd_account_pick_web(platform_input="idem_plat", platform_meta=meta)
+        count = clean_db.execute(
+            "SELECT COUNT(*) FROM platforms WHERE platform_key = 'idem_plat'"
+        ).fetchone()[0]
+        assert count == 1

tests/test_platform_dynamic.py

@@ -0,0 +1,76 @@
+# -*- coding: utf-8 -*-
+"""resolve_platform_key_dynamic 与 JSON 解析边界。"""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+
+class TestResolvePlatformKeyDynamicParsing:
+    def test_builtin_still_works_without_db_row(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+
+        result = resolve_platform_key_dynamic("马士基", conn=clean_db)
+        assert result.platform_key == "maersk"
+
+    def test_disabled_platform_returns_disabled(self, clean_db):
+        from db.accounts_repo import upsert_platform
+        from util.platforms import resolve_platform_key_dynamic
+
+        upsert_platform(
+            clean_db,
+            key="custom_off",
+            display_name="Off",
+            default_url="",
+        )
+        clean_db.execute(
+            "UPDATE platforms SET enabled = 0 WHERE platform_key = ?",
+            ("custom_off",),
+        )
+        clean_db.commit()
+        result = resolve_platform_key_dynamic("custom_off", conn=clean_db)
+        assert result.error_code == "PLATFORM_DISABLED"
+
+    def test_get_platform_spec_from_db(self, clean_db):
+        from db.accounts_repo import upsert_platform
+        from util.platforms import get_platform_spec
+
+        upsert_platform(
+            clean_db,
+            key="jiangchang",
+            display_name="匠厂后台",
+            default_url="https://jc2009.com/admin/index.html",
+        )
+        spec = get_platform_spec("jiangchang", conn=clean_db)
+        assert spec["display_name"] == "匠厂后台"
+        assert spec["default_url"] == "https://jc2009.com/admin/index.html"
+
+    def test_malformed_aliases_json_does_not_crash(self, clean_db):
+        from util.platforms import resolve_platform_key_dynamic
+        import time
+
+        now = int(time.time())
+        clean_db.execute(
+            """
+            INSERT INTO platforms (
+                platform_key, display_name, domain, provider_code,
+                default_url, aliases_json, capabilities_json,
+                enabled, default_auth_strategy, created_at, updated_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, 1, NULL, ?, ?)
+            """,
+            (
+                "bad_alias",
+                "Bad",
+                "generic",
+                "bad_alias",
+                "",
+                "not-json",
+                "{}",
+                now,
+                now,
+            ),
+        )
+        clean_db.commit()
+        result = resolve_platform_key_dynamic("bad_alias", conn=clean_db)
+        assert result.platform_key == "bad_alias"

tests/test_profile_shortcut.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+"""Profile 目录浏览器快捷方式测试。"""
+import os
+import sys
+
+import pytest
+
+from util.profile_shortcut import _lnk_safe_filename, create_profile_browser_shortcut
+
+
+class TestLnkSafeFilename:
+    def test_strips_forbidden_chars(self) -> None:
+        assert _lnk_safe_filename('小红书/测试:账号') == "小红书_测试_账号"
+
+    def test_empty_uses_fallback(self) -> None:
+        assert _lnk_safe_filename("   ") == "browser"
+
+
+class TestCreateProfileBrowserShortcut:
+    def test_skips_non_windows(self, tmp_path, monkeypatch) -> None:
+        monkeypatch.setattr("util.profile_shortcut.sys.platform", "linux")
+        out = create_profile_browser_shortcut(str(tmp_path), "demo")
+        assert out is None
+
+    def test_skips_when_no_browser(self, tmp_path, monkeypatch) -> None:
+        monkeypatch.setattr("util.profile_shortcut.sys.platform", "win32")
+        monkeypatch.setattr(
+            "util.profile_shortcut.resolve_chromium_executable",
+            lambda: None,
+        )
+        out = create_profile_browser_shortcut(str(tmp_path), "demo")
+        assert out is None
+
+    @pytest.mark.skipif(sys.platform != "win32", reason="Windows .lnk only")
+    def test_creates_lnk_on_windows(self, tmp_path, monkeypatch) -> None:
+        from service.browser_service import resolve_chromium_executable
+
+        browser_exe = resolve_chromium_executable()
+        if not browser_exe:
+            pytest.skip("no Chrome/Edge installed")
+
+        label = "ShortcutTest"
+        out = create_profile_browser_shortcut(str(tmp_path), label)
+        assert out is not None
+        assert os.path.isfile(out)
+        assert out.endswith(f"{label}.lnk")
+
+    @pytest.mark.skipif(sys.platform != "win32", reason="Windows .lnk only")
+    def test_add_web_creates_shortcut(self, clean_db, fake_env, monkeypatch) -> None:
+        from service.browser_service import resolve_chromium_executable
+        from service.account_service import cmd_account_add_web
+        from db.accounts_repo import find_accounts
+
+        if not resolve_chromium_executable():
+            pytest.skip("no Chrome/Edge installed")
+
+        label = "Maersk shortcut test"
+        cmd_account_add_web(
+            platform_input="maersk",
+            login_id="shortcut@test.com",
+            login_id_type="email",
+            label=label,
+            tenant_id="tenant-test",
+            environment="production",
+            role="booking",
+            provider_code=None,
+            url=None,
+            extra_json_str=None,
+            profile_dir_override=None,
+        )
+        accs = find_accounts(platform_key="maersk")
+        assert len(accs) == 1
+        profile_dir = accs[0]["profile_dir"]
+        lnk_path = os.path.join(profile_dir, f"{label}.lnk")
+        assert os.path.isfile(lnk_path), f"expected shortcut at {lnk_path}"