chore: auto release commit (2026-06-19 11:07:44)

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/account_service.py

@@ -18,6 +18,7 @@ import sys
 import tempfile
 import time
 import uuid
+from dataclasses import dataclass, field
 from typing import Optional
 
 from db.accounts_repo import (
@@ -42,6 +43,8 @@ from db.accounts_repo import (
     update_account_last_used,
     update_account_session_status,
     update_credential_last_used,
+    upsert_platform,
+    ensure_platform_from_caller_meta,
 )
 from db.auth_strategy import (
     ALL_AUTH_STRATEGIES,
@@ -62,9 +65,12 @@ from db.connection import get_conn, init_db
 from util.constants import SKILL_SLUG
 from util.logging_config import get_skill_logger
 from util.platforms import (
-    PLATFORMS,
+    DEFAULT_CAPABILITIES_JSON,
+    PlatformResolveResult,
     _platform_list_cn_for_help,
-    resolve_platform_key,
+    get_platform_spec,
+    is_valid_platform_key_format,
+    resolve_platform_key_dynamic,
     seed_platforms,
 )
 from util.runtime_paths import (
@@ -188,17 +194,124 @@ def _remove_profile_dir(path: str) -> None:
         _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 print ERROR:INVALID_PLATFORM and return None."""
+    """Resolve platform key or emit structured JSON error and return None."""
-    key = resolve_platform_key((input_name or "").strip())
+    init_db()
-    if not key:
+    conn = get_conn()
-        _say(f"ERROR:INVALID_PLATFORM 无法识别的平台「{input_name}」。")
+    try:
-        if hint:
+        result = resolve_platform_key_dynamic(input_name, conn=conn)
-            _say(hint)
+    finally:
-        else:
+        conn.close()
-            _say("支持：" + _platform_list_cn_for_help())
+
-        return None
+    if result.ok:
-    return key
+        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
 
 
 # ============================================================================
@@ -235,7 +348,65 @@ def cmd_platform_get(input_key: str) -> None:
     if plat:
         _say(_ok_json(plat))
     else:
-        _say(_err_json("ERROR:INVALID_PLATFORM", f"Platform '{key}' not found in DB."))
+        _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
 
 
 # ============================================================================
@@ -258,19 +429,26 @@ def cmd_account_add_web(
     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_fail(platform_input)
+    key = _resolve_or_auto_register(platform_input, platform_meta)
     if not key:
         return
 
     init_db()
-    platform_spec = PLATFORMS.get(key, {})
     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:
@@ -416,7 +594,12 @@ def cmd_account_add_secret(
         return
 
     init_db()
-    platform_spec = PLATFORMS.get(key, {})
+    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}"
 
@@ -963,13 +1146,14 @@ def cmd_account_pick_web(
     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_fail(platform_input)
+    key = _resolve_or_auto_register(platform_input, platform_meta)
     if not key:
         return
 
@@ -1289,7 +1473,8 @@ def cmd_delete_by_platform(platform_input: str) -> None:
         )
         conn.commit()
         log.info("delete_by_platform_done platform=%s count=%s", key, len(rows))
-        _say(f"✅ 已删除 {PLATFORMS.get(key, {}).get('display_name', key)} 下共 {len(rows)} 条账号。")
+        display = get_platform_spec(key, conn=conn).get("display_name", key)
+        _say(f"✅ 已删除 {display} 下共 {len(rows)} 条账号。")
     finally:
         conn.close()
 

scripts/service/browser_service.py

@@ -13,8 +13,8 @@ import subprocess
 import sys
 import tempfile
 
-from db.connection import init_db
+from db.connection import get_conn, init_db
-from db.accounts_repo import get_account_by_id
+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,
@@ -74,6 +74,31 @@ def _print_browser_install_hint():
     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 +134,11 @@ def cmd_open(account_id):
         return
     os.makedirs(profile_dir, exist_ok=True)
 
-    # Use account url, fall back to platform default
+    conn = get_conn()
-    platform_spec = PLATFORMS.get(target["platform_key"], {})
+    try:
-    url = (target.get("url") or "").strip() or platform_spec.get("default_url", "")
+        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/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)
 
 

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"