sap-cli-skill/assets/sapcli/sorter.py

"""依赖排序模块。

使用 Kahn 算法对清单对象进行拓扑排序，
按 depends_on 显式依赖和 TYPE_PRIORITY 类型优先级决定执行顺序。
"""
from __future__ import annotations

import logging
from collections import deque

from sapcli.exceptions import CyclicDependencyError
from sapcli.manifest import ManifestEntry

logger = logging.getLogger("sapcli.sorter")

# 类型默认优先级（数值越小越先执行）
TYPE_PRIORITY: dict[str, int] = {
    "domain": 10,
    "dataelement": 20,
    "structure": 25,
    "table": 30,
    "tabletype": 40,
    "view": 42,
    "lockobject": 44,
    "searchhelp": 46,
    "messageclass": 48,
    "interface": 50,
    "class": 60,
    "functiongroup": 70,
    "function": 75,
    "include": 78,
    "cdsview": 79,
    "report": 80,
    # DCL/DDLX 依赖 CDS View，排在 cdsview(79)/report(80) 之后
    "dcl": 81,
    "ddlX": 82,
}

# 未注册类型的默认优先级
_DEFAULT_PRIORITY = 90


def topological_sort(objects: list[ManifestEntry]) -> list[ManifestEntry]:
    """对清单对象进行拓扑排序。

    1. 按 depends_on 建有向边（被依赖 → 当前对象）
    2. 无入边的节点按 TYPE_PRIORITY 排序作为 tie-breaker
    3. 检测循环依赖 → 抛出 CyclicDependencyError

    Returns:
        排序后的对象列表（依赖在前，被依赖在后...不，应该是依赖在前，被依赖的对象先执行）。
    """
    if not objects:
        return []

    name_map: dict[str, ManifestEntry] = {obj.name: obj for obj in objects}
    names = set(name_map.keys())

    # ── 建图：计算入度 ──
    in_degree: dict[str, int] = {name: 0 for name in names}
    # adj[A] = [B, C, ...] 表示 A 完成后可以释放 B、C 的入度
    adj: dict[str, list[str]] = {name: [] for name in names}

    for obj in objects:
        for dep in obj.depends_on:
            if dep not in names:
                # 依赖不在当前列表中，忽略（外部依赖）
                logger.debug("忽略外部依赖: %s → %s", obj.name, dep)
                continue
            # dep → obj（dep 先执行，obj 后执行）
            adj[dep].append(obj.name)
            in_degree[obj.name] += 1

    # ── Kahn 算法 ──
    # 初始化：入度为 0 的节点
    ready: list[str] = [
        name for name, deg in in_degree.items() if deg == 0
    ]
    # 按 TYPE_PRIORITY 排序
    ready.sort(key=lambda n: _priority(n_map=name_map, n=n))

    queue: deque[str] = deque(ready)
    sorted_names: list[str] = []

    while queue:
        # 取出优先级最高（数值最小）的节点
        # 由于 deque 不支持按优先级取，先排序再取
        # 实际实现：每次从 ready 列表中取第一个
        current = queue.popleft()
        sorted_names.append(current)

        # 释放后续节点
        newly_ready: list[str] = []
        for neighbor in adj[current]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                newly_ready.append(neighbor)

        # 新就绪的节点按优先级排序后加入队列
        newly_ready.sort(key=lambda n: _priority(n_map=name_map, n=n))
        queue.extend(newly_ready)

    # ── 循环检测 ──
    if len(sorted_names) != len(names):
        remaining = names - set(sorted_names)
        # 尝试找出循环链
        cycle = _detect_cycle(remaining, adj, in_degree)
        raise CyclicDependencyError(cycle)

    result = [name_map[n] for n in sorted_names]
    logger.info("拓扑排序完成: %d 个对象", len(result))
    return result


def _priority(*, n_map: dict[str, ManifestEntry], n: str) -> int:
    """获取节点优先级数值。"""
    obj_type = n_map[n].type if n in n_map else ""
    return TYPE_PRIORITY.get(obj_type, _DEFAULT_PRIORITY)


def _detect_cycle(
    remaining: set[str],
    adj: dict[str, list[str]],
    in_degree: dict[str, int],
) -> list[str]:
    """尝试在剩余节点中检测循环，返回循环中的节点名列表。"""
    # 沿着入度 > 0 的边走，迟早会回到已访问的节点
    if not remaining:
        return []

    visited: set[str] = set()
    path: list[str] = []
    start = next(iter(remaining))

    current = start
    for _ in range(len(remaining) + 1):
        if current in visited:
            # 找到循环起点
            idx = path.index(current)
            return path[idx:]
        visited.add(current)
        path.append(current)

        # 找下一个仍在 remaining 中的后继
        found_next = False
        for neighbor in adj.get(current, []):
            if neighbor in remaining:
                current = neighbor
                found_next = True
                break
        if not found_next:
            break

    # 无法确定精确循环链，返回所有剩余节点
    return list(remaining)