account-manager/scripts/service/account_mutate_commands.py

# -*- coding: utf-8 -*-
"""Account copy / update CLI commands (data-management 复制 / 编辑)."""
from __future__ import annotations

import json
from typing import Any, Optional

from db.accounts_repo import (
    _UNSET,
    get_account_by_id,
    list_account_labels,
    update_account_user_fields,
)
from db.connection import get_conn, init_db
from service._svc_common import _err_json, _say
from service.account_add_commands import create_web_account_record
from util.account_identity import (
    KNOWN_ACCOUNT_STATUSES,
    next_copy_label,
    resolve_login_id_type,
)
from util.logging_config import get_skill_logger
from util.platforms import get_platform_spec, seed_platforms
from util.profile_shortcut import create_profile_browser_shortcut


def _extra_json_str(account: dict) -> Optional[str]:
    extra = account.get("extra_json")
    if extra is None or extra == "":
        return None
    if isinstance(extra, (dict, list)):
        return json.dumps(extra, ensure_ascii=False)
    text = str(extra).strip()
    return text or None


def _ok_result(data: dict, message: str) -> str:
    return json.dumps(
        {
            "type": "result",
            "schema_version": 1,
            "status": "success",
            "message": message,
            "success": True,
            "data": data,
        },
        ensure_ascii=False,
    )


def cmd_account_copy(account_id) -> None:
    """Clone a web account: same platform/strategy, new empty Profile, empty login_id."""
    log = get_skill_logger()
    log.info("account_copy_attempt id=%r", account_id)
    try:
        aid = int(account_id)
    except (TypeError, ValueError):
        _say(_err_json("ERROR:CLI_BAD_ARGS", "account copy 需要整数 id。"))
        return

    init_db()
    source = get_account_by_id(aid)
    if not source:
        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={aid} 不存在。"))
        return

    conn = get_conn()
    try:
        labels = list_account_labels(conn)
    finally:
        conn.close()

    new_label = next_copy_label(str(source.get("account_label") or ""), labels)
    tenant = str(source.get("tenant_id") or "").strip() or None
    provider = str(source.get("provider_code") or "").strip() or None
    url = str(source.get("url") or "").strip() or None
    extra = _extra_json_str(source)
    session_persistent = source.get("session_persistent")
    try:
        session_persistent_i = int(session_persistent) if session_persistent is not None else None
    except (TypeError, ValueError):
        session_persistent_i = None

    ok_payload = create_web_account_record(
        str(source.get("platform_key") or ""),
        None,
        str(source.get("login_id_type") or "unknown"),
        new_label,
        tenant,
        str(source.get("environment") or "production") or "production",
        str(source.get("role") or "default") or "default",
        provider,
        url,
        extra,
        None,
        auth_strategy=source.get("auth_strategy"),
        session_persistent=session_persistent_i,
        device_fingerprint=None,
        platform_key=str(source.get("platform_key") or "") or None,
    )
    if not ok_payload:
        return

    ok_payload["copied_from_id"] = aid
    log.info(
        "account_copy_success from=%s to=%s label=%s",
        aid,
        ok_payload.get("id"),
        new_label,
    )
    _say(_ok_result(ok_payload, f"已复制为「{new_label}」"))


def cmd_account_update(
    account_id,
    *,
    label: Optional[str] = None,
    login_id: Any = _UNSET,
    login_id_type: Optional[str] = None,
    url: Any = _UNSET,
    status: Optional[str] = None,
) -> None:
    """Update the short set of user-facing account fields."""
    log = get_skill_logger()
    log.info("account_update_attempt id=%r", account_id)
    try:
        aid = int(account_id)
    except (TypeError, ValueError):
        _say(_err_json("ERROR:CLI_BAD_ARGS", "account update 需要整数 --id。"))
        return

    init_db()
    current = get_account_by_id(aid)
    if not current:
        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={aid} 不存在。"))
        return

    next_label = None
    if label is not None:
        next_label = str(label).strip()
        if not next_label:
            _say(_err_json("ERROR:CLI_BAD_ARGS", "账号名称不能为空。"))
            return

    next_status = None
    if status is not None:
        next_status = str(status).strip()
        if next_status not in KNOWN_ACCOUNT_STATUSES:
            _say(_err_json(
                "ERROR:CLI_BAD_ARGS",
                f"不支持的状态：{status}。支持：active / disabled / needs_review。",
            ))
            return

    resolved_login_id = _UNSET
    if login_id is not _UNSET:
        resolved_login_id = str(login_id).strip() if login_id is not None else ""

    next_type = None
    if login_id_type is not None or resolved_login_id is not _UNSET:
        lid_for_infer = (
            resolved_login_id
            if resolved_login_id is not _UNSET
            else current.get("login_id")
        )
        explicit = login_id_type if login_id_type is not None else current.get("login_id_type")
        next_type = resolve_login_id_type(str(explicit or ""), lid_for_infer)

    resolved_url = _UNSET
    if url is not _UNSET:
        text = str(url).strip() if url is not None else ""
        if text:
            resolved_url = text
        else:
            conn = get_conn()
            try:
                seed_platforms(conn)
                spec = get_platform_spec(str(current.get("platform_key") or ""), conn=conn)
            finally:
                conn.close()
            resolved_url = str(spec.get("default_url") or "")

    conn = get_conn()
    try:
        update_account_user_fields(
            conn,
            aid,
            account_label=next_label,
            login_id=resolved_login_id,
            login_id_type=next_type,
            url=resolved_url,
            status=next_status,
        )
        conn.commit()
    except ValueError as exc:
        conn.rollback()
        _say(_err_json("ERROR:CLI_BAD_ARGS", str(exc)))
        return
    except Exception:
        conn.rollback()
        log.exception("account_update_failure")
        _say(_err_json("ERROR:DB_ERROR", "Database update failed."))
        return
    finally:
        conn.close()

    updated = get_account_by_id(aid)
    if not updated:
        _say(_err_json("ERROR:ACCOUNT_NOT_FOUND", f"账号 id={aid} 不存在。"))
        return

    profile_dir = str(updated.get("profile_dir") or "").strip()
    if profile_dir:
        create_profile_browser_shortcut(
            profile_dir,
            str(updated.get("account_label") or ""),
            log,
            url=str(updated.get("url") or "") or None,
        )

    log.info("account_update_success id=%s", aid)
    _say(_ok_result(
        {
            "id": updated["id"],
            "platform_key": updated.get("platform_key"),
            "account_label": updated.get("account_label"),
            "login_id": updated.get("login_id") or "",
            "login_id_type": updated.get("login_id_type"),
            "url": updated.get("url") or "",
            "status": updated.get("status"),
        },
        "已保存",
    ))