sap-cli-skill/assets/sapcli/commands/cds.py

"""CDS View 命令：cds (download / sync / create)。"""
from __future__ import annotations

import argparse
import logging
import os

from sapcli.client import ADTClient

logger = logging.getLogger("sapcli.commands.cds")


def cmd_cds(args: argparse.Namespace, client: ADTClient) -> None:
    """CDS View 操作。"""
    action = getattr(args, "cds_action", None)

    if action == "download":
        _cds_download(args, client)
    elif action == "sync":
        _cds_sync(args, client)
    elif action == "create":
        _cds_create(args, client)
    else:
        print("  用法: sap-cli cds [download|sync|create]")
        print("    download  — 下载 CDS View DDL 源码")
        print("    sync      — 同步本地 DDL 到 SAP")
        print("    create    — 创建新的 CDS View")


def _cds_download(args: argparse.Namespace, client: ADTClient) -> None:
    """下载 CDS View DDL 源码。"""
    name: str = args.name
    save_path: str = getattr(args, "path", ".")

    print("=" * 60)
    print("  SAP CDS View DDL 源码下载")
    print("=" * 60)
    print(f"  CDS 名称: {name}")
    print(f"  保存路径: {save_path}")

    print(f"\n  → 正在下载 DDL 源码...")
    try:
        source = client.get_cds_source(name)
    except Exception as e:
        print(f"  ✗ 下载失败: {e}")
        return

    source = source.replace("\r\n", "\n").replace("\r", "\n")
    line_count = len(source.splitlines())
    print(f"  ✓ DDL 源码下载成功! {len(source)} 字符, {line_count} 行")

    if not os.path.isdir(save_path):
        os.makedirs(save_path, exist_ok=True)

    filename = f"{name.lower()}.ddl"
    filepath = os.path.join(save_path, filename)

    with open(filepath, "w", encoding="utf-8") as f:
        f.write(source)
    print(f"\n  ✓ 文件已保存: {filepath}")


def _cds_sync(args: argparse.Namespace, client: ADTClient) -> None:
    """同步本地 DDL 到 SAP。"""
    name: str = args.name
    ddl_path: str = args.path

    print("=" * 60)
    print("  SAP CDS View DDL 同步")
    print("=" * 60)
    print(f"  CDS 名称: {name}")
    print(f"  DDL 文件: {ddl_path}")

    if not os.path.isfile(ddl_path):
        print(f"\n  ✗ DDL 文件不存在: {ddl_path}")
        return

    with open(ddl_path, "r", encoding="utf-8") as f:
        ddl_source = f.read()

    obj_uri = f"/sap/bc/adt/ddic/ddlsources/{name.lower()}"
    src_uri = f"/sap/bc/adt/ddic/ddlsources/{name.lower()}/source/main"

    # 检查对象是否存在
    print(f"\n  → 检查 CDS View 是否存在...")
    exists = client.object_exists(obj_uri)

    if not exists:
        print(f"  ℹ CDS View 不存在，需要先创建")
        try:
            client.create_cds(name, name, ddl_source)
            print(f"  ✓ CDS View 创建并同步成功!")
        except Exception as e:
            print(f"  ✗ 创建失败: {e}")
        return

    print(f"  ✓ CDS View 存在")

    # 锁定 → 写入 → 解锁 → 激活
    print(f"\n  → 正在同步 DDL 源码...")
    try:
        lock_handle, corr_nr = client.lock(obj_uri)
        try:
            client.set_source(src_uri, ddl_source, lock_handle, corr_nr)
            print(f"  ✓ DDL 源码写入成功")
        finally:
            client.unlock(obj_uri, lock_handle)
    except Exception as e:
        print(f"  ✗ 写入失败: {e}")
        return

    # 激活
    print(f"\n  → 正在激活...")
    try:
        success, messages = client.activate(name.upper(), obj_uri)
        if success:
            print(f"  ✓ 激活成功")
        else:
            errors = [m for m in messages if m["type"] == "E"]
            print(f"  ✗ 激活失败: {len(errors)} 个错误")
            for e in errors:
                print(f"     [错误] 行 {e['line']}: {e['text']}")
    except Exception as e:
        print(f"  ⚠ 激活失败: {e}")

    print(f"\n  ✓ CDS View 同步完成!")


def _cds_create(args: argparse.Namespace, client: ADTClient) -> None:
    """创建新的 CDS View。"""
    name: str = args.name
    description: str = getattr(args, "description", None) or name
    ddl_path: str | None = getattr(args, "ddl_path", None)

    print("=" * 60)
    print("  SAP 创建 CDS View")
    print("=" * 60)
    print(f"  CDS 名称: {name}")
    print(f"  描述:     {description}")

    # 读取 DDL 源码
    ddl_source = ""
    if ddl_path and os.path.isfile(ddl_path):
        with open(ddl_path, "r", encoding="utf-8") as f:
            ddl_source = f.read()
        print(f"  DDL 文件: {ddl_path}")
    else:
        # 生成默认模板
        ddl_source = _default_cds_template(name, description)
        print(f"  DDL: 使用默认模板")

    print(f"\n  → 正在创建 CDS View...")
    try:
        obj_uri, src_uri = client.create_cds(name, description, ddl_source)
    except Exception as e:
        print(f"  ✗ 创建失败: {e}")
        return

    print(f"  ✓ CDS View 创建成功!")
    print(f"  ✓ URI: {obj_uri}")

    # 保存本地文件
    save_dir = getattr(args, "path", ".")
    if save_dir:
        if not os.path.isdir(save_dir):
            os.makedirs(save_dir, exist_ok=True)
        filepath = os.path.join(save_dir, f"{name.lower()}.ddl")
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(ddl_source)
        print(f"  ✓ DDL 已保存: {filepath}")


def _default_cds_template(name: str, description: str) -> str:
    """生成默认 CDS View DDL 模板。"""
    view_name = name[:16].upper()
    return (
        f"@AbapCatalog.sqlViewName: \'{view_name}\'\n"
        f"@EndUserText.label: \'{description}\'\n"
        f"define view {name.lower()}\n"
        f"  as select from sflight\n"
        f"  {{\n"
        f"    carrid,\n"
        f"    connid,\n"
        f"    fldate\n"
        f"  }}\n"
    )