Spaces:
Running
Running
| #!/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()) | |