ocp/tools/inventory.py

"""Symbol inventory: what the app imports from OCP, and what this wheel provides.

Two jobs, one source of truth:

  --emit    parse the app's sources for every `from OCP.<mod> import <name>`
            plus the attributes reached on those names, and write inventory.json.
            That file is the coverage work-queue for the remaining increments
            and, re-run later, the drift detector for newly used symbols.

  --check   import the installed OCP and report what the inventory asks for but
            the wheel does not provide, grouped by module.

Static analysis only reaches names reached through an imported symbol
(`TopExp.MapShapes_s`), not methods called on instances (`shape.IsSame(...)`) —
those are covered by running the app's own suite in the parity venv, where a
gap is a loud AttributeError.  So a green --check is necessary, not sufficient.

    python tools/inventory.py --emit --app ../app/backend
    python tools/inventory.py --check
"""

from __future__ import annotations

import argparse
import ast
import collections
import json
import pathlib
import sys

HERE = pathlib.Path(__file__).resolve().parent
DEFAULT_INVENTORY = HERE.parent / "inventory.json"
DEFAULT_APP = HERE.parent.parent / "app" / "backend"


def scan(app_root: pathlib.Path) -> dict:
    """Collect {module: {symbol: [attributes reached on it]}} from the app."""
    modules: dict[str, dict[str, set[str]]] = collections.defaultdict(
        lambda: collections.defaultdict(set)
    )
    files = 0

    for path in sorted(app_root.rglob("*.py")):
        if "__pycache__" in path.parts:
            continue
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"))
        except (SyntaxError, UnicodeDecodeError):
            continue

        # local alias -> (module, original name)
        local: dict[str, tuple[str, str]] = {}
        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom) and node.module:
                if node.module == "OCP" or node.module.startswith("OCP."):
                    mod = node.module[4:] or "OCP"
                    for alias in node.names:
                        local[alias.asname or alias.name] = (mod, alias.name)
                        modules[mod][alias.name]  # noqa: B018 — create the entry

        if local:
            files += 1

        for node in ast.walk(tree):
            if (
                isinstance(node, ast.Attribute)
                and isinstance(node.value, ast.Name)
                and node.value.id in local
            ):
                mod, name = local[node.value.id]
                modules[mod][name].add(node.attr)

    return {
        "app_root": str(app_root),
        "files_importing_ocp": files,
        "modules": {
            mod: {name: sorted(attrs) for name, attrs in sorted(syms.items())}
            for mod, syms in sorted(modules.items())
        },
    }


def emit(app_root: pathlib.Path, out: pathlib.Path) -> int:
    data = scan(app_root)
    out.write_text(json.dumps(data, indent=2) + "\n")
    n_mod = len(data["modules"])
    n_sym = sum(len(v) for v in data["modules"].values())
    print(f"{n_sym} symbols across {n_mod} modules "
          f"({data['files_importing_ocp']} files) -> {out}")
    return 0


def check(inventory: pathlib.Path) -> int:
    import importlib

    data = json.loads(inventory.read_text())
    missing: dict[str, list[str]] = {}
    present = 0

    for mod, symbols in data["modules"].items():
        try:
            m = importlib.import_module(f"OCP.{mod}")
        except ImportError:
            missing[mod] = [f"<module missing> ({len(symbols)} symbols)"]
            continue

        for name, attrs in symbols.items():
            obj = getattr(m, name, None)
            if obj is None:
                missing.setdefault(mod, []).append(name)
                continue
            present += 1
            for attr in attrs:
                if not hasattr(obj, attr):
                    missing.setdefault(mod, []).append(f"{name}.{attr}")

    total = sum(len(v) for v in data["modules"].values())
    print(f"{present}/{total} imported symbols available")
    if missing:
        print(f"\nmissing, by module ({len(missing)} modules):")
        for mod in sorted(missing):
            print(f"  {mod}: {', '.join(sorted(missing[mod]))}")
        return 1
    print("complete")
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--emit", action="store_true", help="rewrite the inventory")
    ap.add_argument("--check", action="store_true",
                    help="compare the installed OCP against the inventory")
    ap.add_argument("--app", type=pathlib.Path, default=DEFAULT_APP)
    ap.add_argument("--inventory", type=pathlib.Path, default=DEFAULT_INVENTORY)
    args = ap.parse_args()

    if args.emit:
        if not args.app.is_dir():
            print(f"app sources not found: {args.app}", file=sys.stderr)
            return 2
        return emit(args.app, args.inventory)
    if args.check:
        if not args.inventory.exists():
            print(f"no inventory at {args.inventory}; run --emit first",
                  file=sys.stderr)
            return 2
        return check(args.inventory)

    ap.print_help()
    return 2


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