"""目录扫描模块。 扫描项目目录,识别 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..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..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), )