# -*- 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"), }, "已保存", ))