wurangyu
38e99638c5
- NW 7.40 E2E-verified compatibility matrix in SKILL.md - DDIC fixes: include namespace, tabletype exists check, DDIC delete lock Accept - New commands: clone, enhancement, history, unit-test - 665 tests, 96% coverage
129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""clone 命令 — 跨系统克隆对象（下载源系统 → 上传目标系统）。"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
|
|
from sapcli.client import ADTClient
|
|
from sapcli.commands.crud import _sync_single
|
|
from sapcli.config import load_config
|
|
from sapcli.exceptions import ConfigError, InvalidNameError, ObjectNotFoundError, SapCliError
|
|
from sapcli.types import get_type_config, parse_object_name
|
|
|
|
logger = logging.getLogger("sapcli.commands.clone")
|
|
|
|
|
|
def _make_profile_client(config_path: str | None, profile: str, verify_ssl: bool = False) -> ADTClient:
|
|
"""按 profile 加载配置并建立已登录的 ADT 客户端。"""
|
|
sap_cfg, _ = load_config(config_path, profile=profile)
|
|
client = ADTClient(
|
|
sap_cfg.host, sap_cfg.client, sap_cfg.user, sap_cfg.password,
|
|
verify_ssl=verify_ssl,
|
|
)
|
|
client.login()
|
|
return client
|
|
|
|
|
|
def cmd_clone(args: argparse.Namespace, client: ADTClient) -> None:
|
|
"""从一个系统下载对象，上传到另一个系统。
|
|
|
|
注意：``client`` 参数在本命令中被忽略；源/目标客户端分别按
|
|
``--from`` / ``--to`` profile 构建。
|
|
"""
|
|
name: str = args.name
|
|
obj_type: str = args.type
|
|
from_profile: str = args.from_profile
|
|
to_profile: str = args.to_profile
|
|
corr_nr: str | None = getattr(args, "corr_nr", None)
|
|
config_path: str | None = getattr(args, "config", None)
|
|
verify_ssl: bool = getattr(args, "verify_ssl", False)
|
|
|
|
parsed = parse_object_name(name, obj_type)
|
|
type_label = get_type_config(obj_type).label
|
|
|
|
if parsed.src_uri is None:
|
|
print(f"\n ✗ {type_label}没有源代码，不支持 clone 操作")
|
|
raise InvalidNameError(f"{type_label}没有源代码，不支持 clone 操作")
|
|
|
|
print("=" * 60)
|
|
print(" 跨系统克隆对象")
|
|
print("=" * 60)
|
|
print(f" 对象类型: {type_label}")
|
|
print(f" 对象名称: {parsed.display_name}")
|
|
print(f" 源系统: {from_profile}")
|
|
print(f" 目标系统: {to_profile}")
|
|
if corr_nr:
|
|
print(f" 目标传输请求: {corr_nr}")
|
|
|
|
# ── Step 1: 从源系统下载源码 ──
|
|
print(f"\n → 连接源系统 ({from_profile})...")
|
|
try:
|
|
src_client = _make_profile_client(config_path, from_profile, verify_ssl)
|
|
except Exception as e:
|
|
print(f" ✗ 连接源系统失败: {e}")
|
|
raise ConfigError(f"连接源系统失败: {e}") from e
|
|
print(f" ✓ 已登录源系统")
|
|
|
|
try:
|
|
print(f"\n → 检查源对象是否存在...")
|
|
if not src_client.object_exists(parsed.obj_uri):
|
|
print(f" ✗ 源对象不存在: {parsed.display_name}")
|
|
raise ObjectNotFoundError(parsed.display_name, obj_type)
|
|
print(" ✓ 源对象存在")
|
|
|
|
print(f" → 正在从源系统下载源代码...")
|
|
source = src_client.get_source(parsed.src_uri)
|
|
source = source.replace("\r\n", "\n").replace("\r", "\n")
|
|
print(f" ✓ 下载成功 ({len(source)} 字符)")
|
|
finally:
|
|
src_client.session.close()
|
|
|
|
# 写入临时文件供 _sync_single 复用
|
|
tmp_fd, tmp_path = tempfile.mkstemp(suffix=".abap", prefix=f"clone_{parsed.file_base}_")
|
|
try:
|
|
with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
|
|
f.write(source)
|
|
|
|
# ── Step 2: 连接目标系统 ──
|
|
print(f"\n → 连接目标系统 ({to_profile})...")
|
|
try:
|
|
tgt_client = _make_profile_client(config_path, to_profile, verify_ssl)
|
|
except Exception as e:
|
|
print(f" ✗ 连接目标系统失败: {e}")
|
|
raise ConfigError(f"连接目标系统失败: {e}") from e
|
|
print(f" ✓ 已登录目标系统")
|
|
|
|
try:
|
|
print(f"\n → 检查目标对象是否存在...")
|
|
exists = tgt_client.object_exists(parsed.obj_uri)
|
|
if exists:
|
|
print(" ✓ 目标对象已存在（将覆盖）")
|
|
else:
|
|
print(" ℹ 目标对象不存在（将创建并同步）")
|
|
|
|
# ── Step 3: 同步到目标（不存在则自动创建 + 上传 + 激活）──
|
|
print(f"\n → 同步到目标系统...")
|
|
success, error_msg, actual_corr = _sync_single(
|
|
name, obj_type, tmp_path, tgt_client, corr_nr=corr_nr,
|
|
)
|
|
finally:
|
|
tgt_client.session.close()
|
|
finally:
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
|
|
if success:
|
|
print(f"\n {'=' * 60}")
|
|
print(f" ✓ 克隆完成: {parsed.display_name}")
|
|
print(f" {from_profile} → {to_profile}")
|
|
if actual_corr:
|
|
print(f" 目标传输请求: {actual_corr}")
|
|
print(f" {'=' * 60}")
|
|
return
|
|
|
|
raise SapCliError(error_msg or f"克隆失败: {parsed.display_name}")
|