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

"""目录扫描模块。

扫描项目目录，识别 ABAP 开发对象。支持两种布局：

1. **三层布局（推荐，sap-cli-skill 铁律 5）**：
   `src/<开发包>/<对象类型>/<文件名>.abap`
   例：`src/TMP/class/zcl_demo.abap`
   本地开发包固定写 `TMP`（对应 SAP 端 `$TMP` 本地包，目录名去掉 `$`）。

2. **扁平布局（旧项目，向后兼容）**：
   `<对象类型>/<文件名>.abap`（项目根目录下，对象类型目录用复数）
   例：`classes/zcl_demo.abap`

对象类型目录名**单复数均接受、大小写不敏感**（`class/` 与 `classes/` 等价）。

函数模块（function 类型）两种命名均支持：
- 子目录式：`function/<函数组>/<函数模块>.abap`
- ADT 式：`function/<函数组>.fugr.<函数模块>.abap`（abapGit 惯例）
对象名统一为 `函数组/函数模块`（大写）。
"""
from __future__ import annotations

import logging
import os
from dataclasses import dataclass

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

# 对象类型目录名（复数，历史命名）→ 对象类型
DIRECTORY_TYPE_MAP: dict[str, str] = {
    "reports": "report",
    "classes": "class",
    "interfaces": "interface",
    "functions": "function",
    "domains": "domain",
    "dataelements": "dataelement",
    "tables": "table",
    "structures": "structure",
    "tabletypes": "tabletype",
    "includes": "include",
    "cdsviews": "cdsview",
    "messageclasses": "messageclass",
    "views": "view",
    "searchhelps": "searchhelp",
    "lockobjects": "lockobject",
    "dcl": "dcl",
    "ddlx": "ddlX",
}

# 单数形式目录名（sap-cli-skill 结构规范使用单数）→ 对象类型
_SINGULAR_TYPE_MAP: dict[str, str] = {
    "report": "report",
    "class": "class",
    "interface": "interface",
    "function": "function",
    "domain": "domain",
    "dataelement": "dataelement",
    "table": "table",
    "structure": "structure",
    "tabletype": "tabletype",
    "include": "include",
    "cdsview": "cdsview",
    "messageclass": "messageclass",
    "view": "view",
    "searchhelp": "searchhelp",
    "lockobject": "lockobject",
    "dcl": "dcl",
    "ddlx": "ddlX",
}

# 查找用总表：小写目录名 → 对象类型
TYPE_DIR_LOOKUP: dict[str, str] = {**DIRECTORY_TYPE_MAP, **_SINGULAR_TYPE_MAP}

FUNCTIONS_DIR = "functions"

# 源码根目录（三层布局的第 1 层父目录）
SRC_DIR = "src"

# ADT 式函数模块文件名分隔符：<fugr>.fugr.<fm>.abap
FUNC_ADT_SEP = ".fugr."


@dataclass
class ScannedObject:
    """扫描到的对象。"""

    name: str   # 大写对象名，function 类型含 / (GROUP/FUNC)
    type: str   # 对象类型
    file: str   # 相对于项目根目录的路径


def scan_project(project_path: str) -> list[ScannedObject]:
    """扫描项目目录，返回识别到的所有对象列表。

    先扫项目根目录下的对象类型目录（旧布局），
    再扫 src/<开发包>/ 下的对象类型目录（三层布局）。
    同名同类型对象只保留第一次扫到的。
    """
    results: list[ScannedObject] = []
    seen: set[tuple[str, str]] = set()

    # 扫描范围：(目录绝对路径, 该目录相对项目根的路径前缀)
    scopes: list[tuple[str, str]] = [(project_path, "")]

    src_path = os.path.join(project_path, SRC_DIR)
    if os.path.isdir(src_path):
        for pkg_name in sorted(os.listdir(src_path)):
            if pkg_name.startswith("."):
                continue
            pkg_path = os.path.join(src_path, pkg_name)
            if os.path.isdir(pkg_path):
                scopes.append((pkg_path, f"{SRC_DIR}/{pkg_name}"))

    for root_path, rel_prefix in scopes:
        try:
            entries = sorted(os.listdir(root_path))
        except OSError as exc:  # 权限/竞态
            logger.debug("目录不可读，跳过: %s (%s)", root_path, exc)
            continue

        for dir_name in entries:
            if dir_name.startswith("."):
                continue
            obj_type = TYPE_DIR_LOOKUP.get(dir_name.lower())
            if obj_type is None:
                continue
            dir_path = os.path.join(root_path, dir_name)
            if not os.path.isdir(dir_path):
                continue

            if obj_type == "function":
                _scan_functions(dir_path, dir_name, rel_prefix, results, seen)
            else:
                _scan_flat_directory(dir_path, dir_name, obj_type, rel_prefix, results, seen)

    logger.info("扫描完成: %d 个对象", len(results))
    return results


def _rel(rel_prefix: str, *parts: str) -> str:
    """拼相对项目根的路径（统一用 / 分隔，去掉空段）。"""
    segs = [rel_prefix, *parts]
    return "/".join(s for s in segs if s)


def _add(
    results: list[ScannedObject],
    seen: set[tuple[str, str]],
    name: str,
    obj_type: str,
    rel_path: str,
) -> None:
    key = (name, obj_type)
    if key in seen:
        logger.debug("重复对象，跳过: %s (%s) ← %s", name, obj_type, rel_path)
        return
    seen.add(key)
    results.append(ScannedObject(name=name, type=obj_type, file=rel_path))
    logger.debug("扫描到: %s (%s) → %s", name, obj_type, rel_path)


def _scan_flat_directory(
    dir_path: str,
    dir_name: str,
    obj_type: str,
    rel_prefix: str,
    results: list[ScannedObject],
    seen: set[tuple[str, str]],
) -> None:
    """扫描普通目录（一对一：文件名 = 对象名）。"""
    for filename in sorted(os.listdir(dir_path)):
        if not filename.endswith(".abap"):
            continue
        if filename.startswith("."):
            continue
        if not os.path.isfile(os.path.join(dir_path, filename)):
            continue

        obj_name = filename[:-5].upper()  # 去掉 .abap，转大写
        _add(results, seen, obj_name, obj_type, _rel(rel_prefix, dir_name, filename))


def _scan_functions(
    dir_path: str,
    dir_name: str,
    rel_prefix: str,
    results: list[ScannedObject],
    seen: set[tuple[str, str]],
) -> None:
    """扫描函数模块目录。

    支持两种命名：
    - 子目录式: <函数组>/<函数模块>.abap
    - ADT 式:   <函数组>.fugr.<函数模块>.abap
    对象名: <函数组大写>/<函数模块大写>
    """
    for entry in sorted(os.listdir(dir_path)):
        if entry.startswith("."):
            continue
        entry_path = os.path.join(dir_path, entry)

        if os.path.isdir(entry_path):
            # 子目录式：子目录名 = 函数组名
            group_name = entry.upper()
            for filename in sorted(os.listdir(entry_path)):
                if not filename.endswith(".abap") or filename.startswith("."):
                    continue
                if not os.path.isfile(os.path.join(entry_path, filename)):
                    continue
                func_name = filename[:-5].upper()
                _add(
                    results, seen,
                    f"{group_name}/{func_name}", "function",
                    _rel(rel_prefix, dir_name, entry, filename),
                )
            continue

        if not entry.endswith(".abap"):
            continue

        # ADT 式：<fugr>.fugr.<fm>.abap
        base = entry[:-5]
        low = base.lower()
        if FUNC_ADT_SEP not in low:
            logger.debug("函数模块文件名不符合约定，跳过: %s", entry)
            continue
        idx = low.index(FUNC_ADT_SEP)
        group_name = base[:idx].upper()
        func_name = base[idx + len(FUNC_ADT_SEP):].upper()
        if not group_name or not func_name:
            logger.debug("函数模块文件名解析失败，跳过: %s", entry)
            continue
        _add(
            results, seen,
            f"{group_name}/{func_name}", "function",
            _rel(rel_prefix, dir_name, entry),
        )