scripts/import_integrity_audit.py

#!/usr/bin/env python3
"""Fleet Inspector — Static Import Integrity Audit.



Runs two checks across the repository and writes a Markdown report:



  1. ``compileall`` — byte-compile every ``*.py`` to surface syntax errors.

  2. Import-graph scan — parse each module with ``ast``, collect top-level

     imports, and flag local modules that no other file references

     (potential orphans from previous Termux/Laptop/Agent sessions).



Outputs ``data/audit/import_integrity.md`` (path overridable via CLI).



Usage:

    python scripts/import_integrity_audit.py [--root DIR] [--out FILE]

"""

from __future__ import annotations

import argparse
import ast
import compileall
import io
import os
import sys
from contextlib import redirect_stdout
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable

# Directories we never want to walk.
SKIP_DIRS = {
    ".git",
    ".github",
    "__pycache__",
    ".venv",
    "venv",
    "env",
    "node_modules",
    "data",
    "secrets",
    "docs",
}


def iter_python_files(root: Path) -> Iterable[Path]:
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for name in filenames:
            if name.endswith(".py"):
                yield Path(dirpath) / name


def run_compileall(root: Path) -> tuple[bool, str]:
    """Byte-compile every .py under root. Returns (ok, captured_output)."""
    buf = io.StringIO()
    with redirect_stdout(buf):
        ok = compileall.compile_dir(
            str(root),
            quiet=1,
            rx=__import__("re").compile(r"/(\.git|__pycache__|node_modules|\.venv|venv|env)/"),
        )
    return bool(ok), buf.getvalue()


def collect_imports(py_files: list[Path], root: Path) -> tuple[dict[str, set[str]], list[tuple[Path, str]]]:
    """Return (module -> set of imported top-level names, list of parse errors)."""
    imports: dict[str, set[str]] = {}
    errors: list[tuple[Path, str]] = []
    for path in py_files:
        try:
            tree = ast.parse(path.read_text(encoding="utf-8", errors="replace"))
        except SyntaxError as exc:
            errors.append((path, f"{exc.__class__.__name__}: {exc}"))
            continue
        names: set[str] = set()
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    names.add(alias.name.split(".")[0])
            elif isinstance(node, ast.ImportFrom):
                if node.level == 0 and node.module:
                    names.add(node.module.split(".")[0])
        rel = path.relative_to(root).as_posix()
        imports[rel] = names
    return imports, errors


def discover_local_modules(root: Path) -> dict[str, Path]:
    """Map top-level local module name -> path."""
    modules: dict[str, Path] = {}
    for entry in root.iterdir():
        if entry.name in SKIP_DIRS or entry.name.startswith("."):
            continue
        if entry.is_file() and entry.suffix == ".py":
            modules[entry.stem] = entry
        elif entry.is_dir() and (entry / "__init__.py").exists():
            modules[entry.name] = entry
    return modules


def find_orphans(local_modules: dict[str, Path], imports: dict[str, set[str]]) -> list[str]:
    """Local modules never imported by anything else in the tree."""
    referenced: set[str] = set()
    for names in imports.values():
        referenced.update(names)

    entrypoint_hints = {"app", "ignite_oppo_mesh", "ignite_tia", "tia_worker",
                        "create_hf_space", "diagnostic_audit", "backend_loader",
                        "module_resolver"}

    def is_test_module(name: str) -> bool:
        return name.startswith("test_")

    orphans: list[str] = []
    for name in sorted(local_modules):
        if name in referenced:
            continue
        if name in entrypoint_hints or is_test_module(name):
            continue
        orphans.append(name)
    return orphans


def render_report(

    root: Path,

    compile_ok: bool,

    compile_log: str,

    parse_errors: list[tuple[Path, str]],

    local_modules: dict[str, Path],

    imports: dict[str, set[str]],

    orphans: list[str],

) -> str:
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    py_count = len(imports) + len(parse_errors)
    lines = [
        "# Import Integrity Audit",
        "",
        f"_Generated: {now}_",
        f"_Root: `{root}`_",
        "",
        "## 1. compileall",
        "",
        f"- Status: **{'PASS' if compile_ok else 'FAIL'}**",
        f"- Python files scanned: **{py_count}**",
        "",
    ]
    if compile_log.strip():
        lines += ["<details><summary>compileall output</summary>", "", "```", compile_log.strip(), "```", "", "</details>", ""]
    else:
        lines += ["No compileall diagnostics.", ""]

    lines += ["## 2. Parse errors", ""]
    if parse_errors:
        for path, msg in parse_errors:
            lines.append(f"- `{path.relative_to(root).as_posix()}` — {msg}")
    else:
        lines.append("None.")
    lines.append("")

    lines += [
        "## 3. Local top-level modules",
        "",
        f"Discovered **{len(local_modules)}** top-level local modules at the repo root.",
        "",
    ]
    for name in sorted(local_modules):
        rel = local_modules[name].relative_to(root).as_posix()
        lines.append(f"- `{name}` → `{rel}`")
    lines.append("")

    lines += [
        "## 4. Orphan candidates",
        "",
        "Modules with no inbound imports from other tracked Python files.",
        "Entry points (`app`, `ignite_*`, `tia_worker`, `create_hf_space`,",
        "`diagnostic_audit`, `backend_loader`, `module_resolver`) and `test_*`",
        "files are excluded from this list.",
        "",
    ]
    if orphans:
        for name in orphans:
            rel = local_modules[name].relative_to(root).as_posix()
            lines.append(f"- `{name}` (`{rel}`) — review or remove if no longer used")
    else:
        lines.append("No orphan candidates detected.")
    lines.append("")

    lines += ["## 5. Summary", ""]
    status = "PASS" if compile_ok and not parse_errors else "ATTENTION REQUIRED"
    lines += [
        f"- Overall status: **{status}**",
        f"- compileall: {'PASS' if compile_ok else 'FAIL'}",
        f"- Parse errors: {len(parse_errors)}",
        f"- Orphan candidates: {len(orphans)}",
        "",
    ]
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", default=".", help="Repository root")
    parser.add_argument("--out", default="data/audit/import_integrity.md", help="Report output path")
    args = parser.parse_args()

    root = Path(args.root).resolve()
    out = (root / args.out).resolve() if not Path(args.out).is_absolute() else Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)

    py_files = list(iter_python_files(root))
    compile_ok, compile_log = run_compileall(root)
    imports, parse_errors = collect_imports(py_files, root)
    local_modules = discover_local_modules(root)
    orphans = find_orphans(local_modules, imports)

    report = render_report(root, compile_ok, compile_log, parse_errors,
                           local_modules, imports, orphans)
    out.write_text(report, encoding="utf-8")
    print(f"Wrote {out}")
    return 0 if compile_ok and not parse_errors else 1


if __name__ == "__main__":
    sys.exit(main())