account-manager/scripts/service/_svc_common.py

# -*- 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,
)
from db.auth_strategy import (
    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.platforms import (
    PlatformResolveResult,
    _platform_list_cn_for_help,
    resolve_platform_key_dynamic,
)


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:
    """调用方可附带的平台展示信息。不再用于自动插入 platforms 行。"""

    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)


_RESOLVE_CODES_WITHOUT_PLATFORM_LIST = frozenset(
    {
        "PLATFORM_NOT_REGISTERED",
        "INVALID_PLATFORM_KEY",
        "PLATFORM_DISABLED",
        "AMBIGUOUS_PLATFORM",
        "PLATFORM_ALIAS_CONFLICT",
        "INVALID_PLATFORM_DISPLAY_NAME",
    }
)


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]:
    """解析已注册平台；未注册则拒绝。meta 仅兼容旧调用方，不再自动建平台。"""
    del meta
    return _resolve_or_fail(platform_input, hint=hint)


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 _RESOLVE_CODES_WITHOUT_PLATFORM_LIST:
        _say("支持：" + _platform_list_cn_for_help())
    return None