diff --git a/scripts/cli/app.py b/scripts/cli/app.py index 57f1e2e..7b1028b 100644 --- a/scripts/cli/app.py +++ b/scripts/cli/app.py @@ -141,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 # ========================================================================= @@ -416,6 +458,7 @@ def _cmd_add_web(argv: list[str]) -> None: if not plat: print("ERROR:CLI_BAD_ARGS account add-web 需要 --platform 。") return + platform_meta = _parse_platform_caller_meta(argv) cmd_account_add_web( platform_input=plat, login_id=login_id, @@ -431,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, ) @@ -503,6 +547,7 @@ def _cmd_pick_web_new(argv: list[str]) -> None: if not plat: print("ERROR:CLI_BAD_ARGS account pick-web 需要 --platform 。") return + platform_meta = _parse_platform_caller_meta(argv) cmd_account_pick_web( platform_input=plat, environment=env, @@ -512,6 +557,7 @@ def _cmd_pick_web_new(argv: list[str]) -> None: ttl_sec=ttl, holder=holder, purpose=purpose, + platform_meta=platform_meta, ) diff --git a/scripts/db/accounts_repo.py b/scripts/db/accounts_repo.py index a4d0170..0c7d72a 100644 --- a/scripts/db/accounts_repo.py +++ b/scripts/db/accounts_repo.py @@ -806,6 +806,115 @@ def upsert_platform( 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 # ============================================================================ diff --git a/scripts/service/account_service.py b/scripts/service/account_service.py index 5f35eff..3561123 100644 --- a/scripts/service/account_service.py +++ b/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 ( @@ -43,6 +44,7 @@ from db.accounts_repo import ( update_account_session_status, update_credential_last_used, upsert_platform, + ensure_platform_from_caller_meta, ) from db.auth_strategy import ( ALL_AUTH_STRATEGIES, @@ -64,6 +66,7 @@ from util.constants import SKILL_SLUG from util.logging_config import get_skill_logger from util.platforms import ( DEFAULT_CAPABILITIES_JSON, + PlatformResolveResult, _platform_list_cn_for_help, get_platform_spec, is_valid_platform_key_format, @@ -191,6 +194,108 @@ 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 emit structured JSON error and return None.""" init_db() @@ -203,15 +308,8 @@ def _resolve_or_fail(input_name: str, hint: str = "") -> Optional[str]: if result.ok: return result.platform_key - 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 and result.error_code == "PLATFORM_NOT_REGISTERED": - _say(hint) - elif hint: - _say(hint) - elif result.error_code not in ("PLATFORM_NOT_REGISTERED", "INVALID_PLATFORM_KEY", "PLATFORM_DISABLED"): + _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 @@ -331,13 +429,14 @@ 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 @@ -1047,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 diff --git a/tests/test_custom_platforms.py b/tests/test_custom_platforms.py index ebf48a2..5d91b77 100644 --- a/tests/test_custom_platforms.py +++ b/tests/test_custom_platforms.py @@ -161,7 +161,7 @@ class TestCustomPlatformAccounts: payload = json.loads(out) assert payload["success"] is False assert payload["error"]["code"] == "ERROR:PLATFORM_NOT_REGISTERED" - assert "platform ensure" in payload["error"]["message"] + 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 @@ -244,3 +244,155 @@ class TestPlatformKeyFormatValidation: 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