132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
# -*- 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.constants import SKILL_SLUG
|
||
from util.logging_config import get_skill_logger
|
||
from util.platforms import DEFAULT_CAPABILITIES_JSON, is_valid_platform_key_format, seed_platforms
|
||
from util.runtime_paths import get_db_path
|
||
|
||
|
||
def cmd_init_db() -> int:
|
||
"""幂等创建/迁移本地 SQLite 与数据管理元数据（宿主 install/upgrade 可调用）。"""
|
||
log = get_skill_logger()
|
||
log.info("cli_start subcommand=init_db")
|
||
try:
|
||
init_db()
|
||
except Exception as exc: # noqa: BLE001 — CLI 边界，统一成可机读失败
|
||
log.exception("init_db_failure")
|
||
print(
|
||
json.dumps(
|
||
{"ok": False, "skill": SKILL_SLUG, "error": str(exc)},
|
||
ensure_ascii=False,
|
||
)
|
||
)
|
||
return 1
|
||
print(
|
||
json.dumps(
|
||
{
|
||
"ok": True,
|
||
"skill": SKILL_SLUG,
|
||
"db_path": get_db_path(),
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
)
|
||
return 0
|
||
|
||
|
||
def cmd_platform_list(domain: Optional[str] = None) -> None:
|
||
"""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
|