diff --git a/scripts/service/account_service.py b/scripts/service/account_service.py index 5f92817..5f35eff 100644 --- a/scripts/service/account_service.py +++ b/scripts/service/account_service.py @@ -63,9 +63,11 @@ 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, _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 ( @@ -190,48 +192,26 @@ def _remove_profile_dir(path: str) -> None: def _resolve_or_fail(input_name: str, hint: str = "") -> Optional[str]: - """Resolve platform key or print ERROR:INVALID_PLATFORM and return None.""" - raw = (input_name or "").strip() - if not raw: - _say("ERROR:INVALID_PLATFORM 平台名不能为空。") - return None - - # 先尝试 platforms.py 硬编码（兼容历史） - key = resolve_platform_key(raw) - if key: - return key - - # 再查 DB（支持动态注册的平台） + """Resolve platform key or emit structured JSON error and return None.""" + init_db() + conn = get_conn() try: - init_db() - conn = get_conn() - try: - row = conn.execute( - "SELECT platform_key FROM platforms WHERE platform_key = ? AND enabled = 1", - (raw,), - ).fetchone() - if row: - return row[0] - rows = conn.execute( - "SELECT platform_key, aliases_json FROM platforms WHERE enabled = 1" - ).fetchall() - raw_lower = raw.lower() - for r in rows: - try: - aliases = json.loads(r[1] or "[]") - if raw_lower in [str(a).lower() for a in aliases]: - return r[0] - except Exception: - continue - finally: - conn.close() - except Exception: - pass + result = resolve_platform_key_dynamic(input_name, conn=conn) + finally: + conn.close() - _say(f"ERROR:INVALID_PLATFORM 无法识别的平台「{input_name}」。") - if hint: + 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) - else: + elif hint: + _say(hint) + elif result.error_code not in ("PLATFORM_NOT_REGISTERED", "INVALID_PLATFORM_KEY", "PLATFORM_DISABLED"): _say("支持：" + _platform_list_cn_for_help()) return None @@ -270,7 +250,10 @@ 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( @@ -292,9 +275,15 @@ def cmd_platform_ensure( 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 '{"web": true, "api_key": false, "rpa": true}' + capabilities_json = capabilities if capabilities else DEFAULT_CAPABILITIES_JSON init_db() conn = get_conn() @@ -353,8 +342,14 @@ def cmd_account_add_web( 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: @@ -500,7 +495,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}" @@ -1373,7 +1373,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() diff --git a/scripts/service/browser_service.py b/scripts/service/browser_service.py index f205ca3..1f39b4a 100644 --- a/scripts/service/browser_service.py +++ b/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, @@ -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 - 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", diff --git a/scripts/util/platforms.py b/scripts/util/platforms.py index 89eabc7..92023f5 100644 --- a/scripts/util/platforms.py +++ b/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 ]" + ), + ) + + +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) diff --git a/tests/test_custom_platforms.py b/tests/test_custom_platforms.py new file mode 100644 index 0000000..ebf48a2 --- /dev/null +++ b/tests/test_custom_platforms.py @@ -0,0 +1,246 @@ +# -*- 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 ensure" 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" diff --git a/tests/test_platform_dynamic.py b/tests/test_platform_dynamic.py new file mode 100644 index 0000000..4a2308a --- /dev/null +++ b/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"