guanlan-src/app_qualityGate/app_qualityGate_guanlan/audits/module_boundary_audit.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""模块边界审计（用户令 2026-09-22：按算法/数据接入管理/前端/后端等模块重构，组件化、高内聚低耦合、可插拔）。

## 它查什么（每条都由 `configs/modules.yaml` 驱动）

| 规则 | 内容 | 退出码 |
|---|---|---|
| R1 结构 | 每个登记模块目录含 `<pkg>/__init__.py`、`<pkg>/api.py`、`common/`（模块内通用组件）、`README.MD` | 5 |
| R2 边界 | 模块之间**只许经对方 `api.py`**：`app_x/**/*.py` 里 import 另一个 `app_y` 的实现包（`app_y.app_y_guanlan.*`，非 `api`）即违规 | 6 |
| R3 依赖方向 | 只许 import `allow` 里列出的模块；反向依赖/成环即违规 | 7 |
| R4 公共层纯净 | `app_common/**` 不得 import 任何其它业务模块 | 7 |
| R5 接口可用 | 每个 `api.py` 的 `_TARGETS`（含带注解写法）转发目标必须真实存在 | 8 |

另外**统计迁移进度**（不算失败）：`src/**`、`scripts/**` 里还有多少件尚未迁进模块目录（P1–P7 的度量）。

## 用法
    python scripts/module_boundary_audit.py            # 审计（rc=0 通过）
    python scripts/module_boundary_audit.py --brief    # 只打结论与进度
退出码: 0 通过 · 5 结构缺失 · 6 边界违规 · 7 依赖违规 · 8 接口目标缺失
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]

import ast
import pathlib
import re
import sys

ROOT = _install_root(__file__)
sys.path.insert(0, str(ROOT))

from src import paths as P                                                  # noqa: E402


def load_registry() -> dict:
    import yaml
    return yaml.safe_load(P.config('modules.yaml').read_text(encoding='utf-8'))


def module_pkg(name: str) -> str:
    return f'{name}_guanlan'


def _imports_of(path: pathlib.Path) -> list[str]:
    """→ 该文件里所有 import 的模块名（`import a.b` 与 `from a.b import c` 都算 a.b）。"""
    try:
        tree = ast.parse(path.read_text(encoding='utf-8'))
    except SyntaxError:
        return []
    out = []
    for n in ast.walk(tree):
        if isinstance(n, ast.Import):
            out += [a.name for a in n.names]
        elif isinstance(n, ast.ImportFrom) and n.module:
            out.append(n.module)
    return out


def audit(verbose: bool = True):
    reg = load_registry()
    mods = {m['name']: m for m in reg['modules']}
    problems: list[tuple[str, str]] = []          # (规则, 说明)
    notes: list[str] = []

    # R1 结构（P9 新布局：实现包直接挂模块根；common/ 作模块内通用组件目录；README 大小写都认）
    for name, m in mods.items():
        base = ROOT / name
        readme = next((base / r for r in ('README.MD', 'README.md') if (base / r).is_file()), None)
        for rel in (f'common/__init__.py', f'{module_pkg(name)}/__init__.py', f'{module_pkg(name)}/api.py'):
            if not (base / rel).is_file():
                problems.append(('R1', f'{name}/ 缺 {rel}'))
        if readme is None:
            problems.append(('R1', f'{name}/ 缺 README.MD（或 README.md）'))
    # R5 接口目标存在（P9：api 在模块根下；★同时认 `_TARGETS: dict[...] = {…}` 这种带注解的写法）
    for name in mods:
        api = ROOT / name / module_pkg(name) / 'api.py'
        if not api.is_file():
            continue
        try:
            tree = ast.parse(api.read_text(encoding='utf-8'))
        except SyntaxError as e:
            problems.append(('R1', f'{name}/api.py 语法错误: {e}'))
            continue
        tg = {}
        for n in tree.body:
            if isinstance(n, ast.AnnAssign) and getattr(n.target, 'id', '') == '_TARGETS':
                val = n.value
            elif isinstance(n, ast.Assign) and getattr(n.targets[0], 'id', '') == '_TARGETS':
                val = n.value
            else:
                continue
            try:
                tg = ast.literal_eval(val)
            except Exception:  # noqa: BLE001
                tg = {}
        for pub, target in tg.items():
            rel = pathlib.Path(*target.split('.'))
            if not ((ROOT / rel).with_suffix('.py').is_file() or (ROOT / rel / '__init__.py').is_file()):
                problems.append(('R5', f'{name}.api.{pub} → {target} 的转发目标不存在'))

    # R2/R3/R4 边界与依赖
    for name, m in mods.items():
        allow = set(m.get('allow') or [])
        for f in (ROOT / name).rglob('*.py'):
            for imp in _imports_of(f):
                parts = imp.split('.')
                if parts[0] not in mods or parts[0] == name:
                    continue
                other = parts[0]
                # 只许 app_<other>.app_<other>_guanlan.api（P9 布局；旧的 …common.… 别名写法同样放行）
                parts_n = [p for i, p in enumerate(parts) if not (i == 1 and p == 'common')]
                ok_face = (len(parts_n) >= 3 and parts_n[1] == module_pkg(other) and parts_n[2] == 'api')
                if not ok_face:
                    problems.append(('R2', f'{f.relative_to(ROOT).as_posix()} 直接 import {imp}'
                                           f'（应走 {other}.{module_pkg(other)}.api）'))
                if other not in allow:
                    problems.append(('R3', f'{f.relative_to(ROOT).as_posix()} import {other}'
                                           f'，但 {name} 的 allow 未包含它'))
                if name == 'app_common':
                    problems.append(('R4', f'{f.relative_to(ROOT).as_posix()} 公共层依赖了业务模块 {other}'))

    # R6 入口与布局解耦: 安装器必须能在当前布局下解析出包版本
    _ver_cands = ['app_common/app_common_guanlan/version.py', 'src/version.py']
    _found = None
    for _c in _ver_cands:
        _f = ROOT / _c
        if _f.is_file() and re.search(r"^VERSION\s*=\s*'([^']+)'", _f.read_text(encoding='utf-8'), re.M):
            _found = (_c, re.search(r"^VERSION\s*=\s*'([^']+)'", _f.read_text(encoding='utf-8'), re.M).group(1))
            break
    if not _found:
        problems.append(('R6', f'当前布局下没有任何候选版本文件含 `VERSION = \'…\'` 字面量（候选: {_ver_cands}）'
                               ' ⇒ install.ps1/install.sh 会读不到包版本'))
    else:
        _ps1 = (ROOT / 'install.ps1').read_text(encoding='utf-8') if (ROOT / 'install.ps1').is_file() else ''
        _sh = (ROOT / 'install.sh').read_text(encoding='utf-8') if (ROOT / 'install.sh').is_file() else ''
        if _ps1 and 'app_common' not in _ps1:
            problems.append(('R6', 'install.ps1 未包含新布局的候选路径（app_common/...）⇒ 平台件搬家后会读不到版本'))
        if _sh and 'app_common' not in _sh:
            problems.append(('R6', 'install.sh 未包含新布局的候选路径（app_common/...）⇒ 平台件搬家后会读不到版本'))

    # R7 入口读 JSON 必须指定 UTF-8（PS 5.1 默认 ANSI）
    _ps1f = ROOT / 'install.ps1'
    if _ps1f.is_file():
        for _i, _ln in enumerate(_ps1f.read_text(encoding='utf-8').splitlines(), 1):
            if _ln.strip().startswith('#'):
                continue          # 注释里提到 Get-Content/.json 不算
            if 'Get-Content' in _ln and '.json' in _ln and '-Encoding' not in _ln:
                problems.append(('R7', f'install.ps1:{_i} 读 JSON 未指定 -Encoding UTF8 ⇒ PS 5.1 会按 ANSI 解码中文'))

    # R8 CLI 兼容壳必须同面转发实现的公开名（否则 importlib 调用其函数会 AttributeError）
    for _shim in sorted((ROOT / 'scripts').glob('*.py')):
        _txt = _shim.read_text(encoding='utf-8')
        if '兼容转发（P' not in _txt:
            continue
        if 'globals().setdefault' not in _txt and 'sys.modules[__name__]' not in _txt:
            problems.append(('R8', f'scripts/{_shim.name} 是兼容壳但未转发实现公开名 '
                                   '⇒ 用 importlib 调其函数的调用方会 AttributeError'))

    # 迁移进度（不算失败）
    legacy = [p for p in list((ROOT / 'src').rglob('*.py')) + list((ROOT / 'scripts').rglob('*.py'))
              if p.is_file()]
    notes.append(f'尚未迁入模块目录的既有实现件: src/** {len(list((ROOT / "src").rglob("*.py")))} 件 · '
                 f'scripts/** {len(list((ROOT / "scripts").rglob("*.py")))} 件（迁移进度按 docs/重构方案_模块化_v0.1.md 的 P1–P9 推进）')

    rc = 0
    if any(p[0] in ('R1',) for p in problems):
        rc = 5
    if any(p[0] == 'R2' for p in problems):
        rc = rc or 6
    if any(p[0] in ('R3', 'R4', 'R6', 'R7', 'R8') for p in problems):
        rc = rc or 7
    if any(p[0] == 'R5' for p in problems):
        rc = rc or 8
    if verbose:
        print('模块边界审计 · 登记模块 %d 个（%s）' % (len(mods), '、'.join(mods)))
        for n in notes:
            print('  [i]', n)
        if problems:
            for rule, msg in problems:
                print('  [X] %s %s' % (rule, msg))
        else:
            print('  [OK] 结构齐全 · 模块间只经 api 调用 · 依赖方向合规 · 公共层纯净 · 接口目标全在位')
    return rc, problems, notes, mods


def main() -> int:
    brief = '--brief' in sys.argv
    rc, problems, notes, mods = audit(verbose=not brief)
    if brief:
        print('[OK] module_boundary_audit rc=0（登记 %d 模块）' % len(mods) if rc == 0
              else '[X] module_boundary_audit rc=%d（%d 条违规）' % (rc, len(problems)))
    return rc


if __name__ == '__main__':
    raise SystemExit(main())