oc_chat
Loading...
Searching...
No Matches
import_graph.py
Go to the documentation of this file.
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# tools/import_graph.py
4
5"""
6AST import-graph extractor: closures, reverse edges, blind spots.
7
8Purpose
9-------
10Answer "what does this set of modules actually drag in?" with a number
11somebody else can reproduce. Built for plan 256, which had to re-derive
12the `oc_orchestrator` extraction closure and found that the V0.1
13analysis (2026-07-08) was computed by a script left in a session
14scratchpad, so none of its numbers could be re-run. This tool is the
15permanent fix: it is committed, it takes its seed set and its roots as
16arguments rather than hardcoding one question, and it is deterministic,
17so a reviewer can re-run it and diff the receipt.
18
19It parses with :mod:`ast` and NEVER imports the code it analyses. A
20module that raises at import time, or that does not even parse, is
21analysed (or listed as a blind spot) rather than crashing the run.
22
23Responsibilities
24----------------
25- :class:`Module` / :class:`Edge` / :class:`BlindSpot` / :class:`Graph` —
26 the data model: one node per file, one edge per resolved import, one
27 blind spot per import this tool provably cannot resolve.
28- :func:`build_graph` — discover modules under the roots and extract
29 every edge, resolving relative imports and classifying each target as
30 standard library, third party, same package or sibling package.
31- :func:`closure` — the transitive closure of same-package edges from a
32 seed set, honouring `--cut` edge suppressions.
33- :func:`parse_layer` — several named seed sets measured against one
34 parsed tree, so a multi-layer question costs one receipt rather than
35 one whole-graph copy per layer.
36- :func:`reverse_edges` — who imports each in-tree module.
37- :func:`receipt` / :func:`render_table` — the JSON receipt and the
38 human table.
39- :func:`main` — the CLI.
40
41Diagnostics
42-----------
43Domain: OC_CHAT.MAIN
44Levels:
45 L2 - lifecycle events (graph built, closure computed)
46 L3 - semantic details (per-module edge counts)
47 L4 - deep tracing (per-edge)
48
49Contracts
50---------
51- The code under measurement is never imported, only parsed.
52- Determinism is a requirement, not a nicety: two runs over an unchanged
53 tree produce byte-identical stdout and byte-identical JSON. The
54 receipt therefore carries no clock and no absolute path -- roots
55 appear as their directory NAME and module paths are root-relative
56 POSIX, so a receipt is machine-independent. The tree the numbers
57 belong to is pinned by the caller through ``--label``.
58- A lazy (function-body) import is a dependency and is counted as one;
59 it is flagged ``deferred`` so a reader can see which couplings are
60 cheap to cut. A ``TYPE_CHECKING`` import is likewise counted and
61 flagged through ``scope``.
62- Known limitation, stated rather than hidden: only the
63 ``importlib.import_module`` call shape is recognised as a string
64 import. ``__import__``, ``importlib.util.spec_from_file_location`` and
65 hand-rolled loaders are invisible to this tool. Grep for them
66 separately before trusting a closure to be complete.
67- Console output is ASCII only.
68
69Dependencies
70------------
71- oc_diagnostics (structured debug logging via _dbg)
72"""
73
74from __future__ import annotations
75
76import argparse
77import ast
78import fnmatch
79import json
80import sys
81from dataclasses import dataclass
82from pathlib import Path
83from typing import Iterable, Sequence
84
85from oc_diagnostics import _dbg
86
87_dbg("OC_CHAT.MAIN", "module", "import graph extractor loaded", 4)
88
89#: Receipt schema; bump when a field changes meaning.
90SCHEMA_VERSION = "1.0"
91
92#: Edge target classifications.
93STDLIB = "stdlib"
94THIRD_PARTY = "third_party"
95SAME_PACKAGE = "same_package"
96SIBLING = "sibling_package"
97
98#: Edge kinds. ``module`` and ``lazy`` are import statements at module
99#: level and inside a function body; ``string`` is an
100#: ``importlib.import_module`` call with a literal argument.
101KIND_MODULE = "module"
102KIND_LAZY = "lazy"
103KIND_STRING = "string"
104
105#: Edge scopes.
106SCOPE_RUNTIME = "runtime"
107SCOPE_TYPE_CHECKING = "type_checking"
108
109#: Blind-spot reasons.
110REASON_NON_LITERAL = "import_module called with a non-literal argument"
111REASON_RELATIVE = "relative import escapes the scanned roots"
112REASON_UNPARSABLE = "file could not be parsed"
113REASON_SHADOWED = "module name already claimed by an earlier root"
114
115#: Directory names never walked.
116_SKIP_DIRS = frozenset({
117 "__pycache__", ".git", ".venv", ".mypy_cache", ".pytest_cache",
118 ".ruff_cache", "node_modules", ".formatter_archive",
119})
120
121
122class ImportGraphError(RuntimeError):
123 """A named refusal: an unknown seed, an unreadable root, a bad cut."""
124
125
126@dataclass(frozen=True)
127class Module:
128 """One parsed source file."""
129
130 name: str
131 path: str
132 lines: int
133 is_package: bool
134
135
136@dataclass(frozen=True)
137class Edge:
138 """One resolved import, from an in-tree module to some target."""
139
140 source: str
141 target: str
142 line: int
143 kind: str
144 scope: str
145 deferred: bool
146 classification: str
147
148
149@dataclass(frozen=True)
151 """An import this tool provably cannot resolve, named not dropped."""
152
153 source: str
154 line: int
155 reason: str
156 snippet: str
157
158
159@dataclass(frozen=True)
160class Graph:
161 """The whole measured tree."""
162
163 modules: dict[str, Module]
164 edges: tuple[Edge, ...]
165 blind_spots: tuple[BlindSpot, ...]
166
167
168@dataclass(frozen=True)
169class Cut:
170 """A suppressed edge class, as ``source_glob->target_glob``."""
171
172 source: str
173 target: str
174
175 def matches(self, edge: Edge) -> bool:
176 """True when *edge* is suppressed by this cut."""
177 return (
178 fnmatch.fnmatchcase(edge.source, self.source)
179 and fnmatch.fnmatchcase(edge.target, self.target)
180 )
181
182 def spec(self) -> str:
183 """The round-trippable text form."""
184 return f"{self.source}->{self.target}"
185
186
187@dataclass(frozen=True)
189 """The transitive same-package reach of one seed set."""
190
191 name: str
192 seeds: tuple[str, ...]
193 members: tuple[str, ...]
194 package_inits: tuple[str, ...]
195 cut_edges: tuple[Edge, ...]
196 external: dict[str, tuple[str, ...]]
197 lines: int
198
199 @property
200 def size(self) -> int:
201 """Modules in the closure, seeds included."""
202 return len(self.members)
203
204 @property
205 def size_excluding_inits(self) -> int:
206 """Modules in the closure that are not package markers."""
207 return len(self.members) - len(self.package_inits)
208
209
210# ---------------------------------------------------------------------
211# Discovery
212# ---------------------------------------------------------------------
213
214def _module_name(root: Path, path: Path) -> tuple[str, bool]:
215 """Dotted name for *path* under *root*, and whether it is a package."""
216 parts = list(path.relative_to(root).parts)
217 if parts[-1] == "__init__.py":
218 return ".".join(parts[:-1]), True
219 parts[-1] = parts[-1][: -len(".py")]
220 return ".".join(parts), False
221
222
223def _excluded(relative: str, excludes: Sequence[str]) -> bool:
224 """True when the root-relative POSIX path matches an exclude glob."""
225 return any(fnmatch.fnmatchcase(relative, pattern) for pattern in excludes)
226
227
229 roots: Sequence[Path],
230 *,
231 excludes: Sequence[str] = (),
232) -> tuple[dict[str, Module], dict[str, Path], list[BlindSpot]]:
233 """Find every ``*.py`` under *roots*, first root wins on a clash."""
234 func = "discover"
235 modules: dict[str, Module] = {}
236 files: dict[str, Path] = {}
237 shadowed: list[BlindSpot] = []
238 for root in roots:
239 if not root.is_dir():
240 raise ImportGraphError(f"root is not a directory: {root.name}")
241 for path in sorted(root.rglob("*.py")):
242 if any(part in _SKIP_DIRS for part in path.relative_to(root).parts):
243 continue
244 relative = path.relative_to(root).as_posix()
245 if _excluded(relative, excludes):
246 continue
247 name, is_package = _module_name(root, path)
248 if not name:
249 continue
250 if name in modules:
251 shadowed.append(BlindSpot(
252 source=name, line=0, reason=REASON_SHADOWED,
253 snippet=relative,
254 ))
255 continue
256 text = _read(path)
257 modules[name] = Module(
258 name=name, path=relative, lines=_count_lines(text),
259 is_package=is_package,
260 )
261 files[name] = path
262 _dbg("OC_CHAT.MAIN", func, f"{len(modules)} module(s) discovered", 2)
263 return modules, files, shadowed
264
265
266def _read(path: Path) -> str:
267 """Read *path* as UTF-8, tolerating a stray decoding error."""
268 return path.read_text(encoding="utf-8", errors="replace")
269
270
271def _count_lines(text: str) -> int:
272 """Physical line count, counting a final unterminated line."""
273 if not text:
274 return 0
275 return text.count("\n") + (0 if text.endswith("\n") else 1)
276
277
278# ---------------------------------------------------------------------
279# Classification
280# ---------------------------------------------------------------------
281
282def classify(target: str, *, package: str, siblings: Sequence[str]) -> str:
283 """Classify an import target by its top-level name."""
284 top = target.split(".", 1)[0]
285 if top == package:
286 return SAME_PACKAGE
287 if top in siblings:
288 return SIBLING
289 if top in sys.stdlib_module_names:
290 return STDLIB
291 return THIRD_PARTY
292
293
294# ---------------------------------------------------------------------
295# Extraction
296# ---------------------------------------------------------------------
297
299 module: str, is_package: bool, level: int, tail: str | None,
300) -> str | None:
301 """Absolute name for a relative import, or None when it escapes."""
302 parts = module.split(".") if module else []
303 base = parts if is_package else parts[:-1]
304 climb = level - 1
305 if climb > len(base):
306 return None
307 if climb:
308 base = base[: len(base) - climb]
309 if not base:
310 return None
311 return ".".join([*base, tail] if tail else base)
312
313
314def _importlib_aliases(tree: ast.AST) -> tuple[set[str], set[str]]:
315 """Names bound to ``importlib`` and to ``import_module``."""
316 modules: set[str] = set()
317 functions: set[str] = set()
318 for node in ast.walk(tree):
319 if isinstance(node, ast.Import):
320 for alias in node.names:
321 if alias.name == "importlib":
322 modules.add(alias.asname or "importlib")
323 elif isinstance(node, ast.ImportFrom) and node.module == "importlib":
324 for alias in node.names:
325 if alias.name == "import_module":
326 functions.add(alias.asname or "import_module")
327 return modules, functions
328
329
330def _is_type_checking(test: ast.expr) -> bool:
331 """True for ``if TYPE_CHECKING:`` in either spelling."""
332 if isinstance(test, ast.Name):
333 return test.id == "TYPE_CHECKING"
334 if isinstance(test, ast.Attribute):
335 return test.attr == "TYPE_CHECKING"
336 return False
337
338
339class _Extractor(ast.NodeVisitor):
340 """Walk one module, recording edges and blind spots.
341
342 Tracks two pieces of context the plain visitor does not carry: are we
343 inside a function body (the import is deferred), and are we inside an
344 ``if TYPE_CHECKING:`` block (the import is typing-only).
345 """
346
348 self,
349 *,
350 module: Module,
351 index: Iterable[str],
352 package: str,
353 siblings: Sequence[str],
354 importlib_names: tuple[set[str], set[str]],
355 source_lines: Sequence[str],
356 ) -> None:
357 self.module = module
358 self.index = frozenset(index)
359 self.package = package
360 self.siblings = tuple(siblings)
361 self.importlib_modules, self.import_module_names = importlib_names
362 self.source_lines = source_lines
363 self.edges: list[Edge] = []
364 self.blind_spots: list[BlindSpot] = []
365 self._deferred = False
366 self._type_checking = False
367
368 # -- context -----------------------------------------------------
369
370 def _visit_function(self, node: ast.AST) -> None:
371 """A function body: everything inside it is deferred."""
372 previous = self._deferred
373 self._deferred = True
374 self.generic_visit(node)
375 self._deferred = previous
376
377 visit_FunctionDef = _visit_function
378 visit_AsyncFunctionDef = _visit_function
379 visit_Lambda = _visit_function
380
381 def visit_If(self, node: ast.If) -> None:
382 """``if TYPE_CHECKING:`` marks its body, never its else branch."""
383 if _is_type_checking(node.test):
384 previous = self._type_checking
385 self._type_checking = True
386 for statement in node.body:
387 self.visit(statement)
388 self._type_checking = previous
389 for statement in node.orelse:
390 self.visit(statement)
391 return
392 self.generic_visit(node)
393
394 # -- emission ----------------------------------------------------
395
396 def _scope(self) -> str:
397 """The scope tag for an edge found at the current position."""
398 return SCOPE_TYPE_CHECKING if self._type_checking else SCOPE_RUNTIME
399
400 def _emit(self, target: str, line: int, kind: str, seen: set[str]) -> None:
401 """Record *target* plus every in-tree ancestor package of it."""
402 names: list[str] = []
403 parts = target.split(".")
404 for depth in range(1, len(parts) + 1):
405 candidate = ".".join(parts[:depth])
406 if candidate in self.index:
407 names.append(candidate)
408 if target not in self.index:
409 names.append(target)
410 for name in names:
411 if name in seen or name == self.module.name:
412 continue
413 seen.add(name)
414 self.edges.append(Edge(
415 source=self.module.name, target=name, line=line, kind=kind,
416 scope=self._scope(), deferred=self._deferred,
417 classification=classify(
418 name, package=self.package, siblings=self.siblings,
419 ),
420 ))
421
422 def _kind(self) -> str:
423 """Import-statement kind at the current position."""
424 return KIND_LAZY if self._deferred else KIND_MODULE
425
426 def _snippet(self, line: int) -> str:
427 """The source line, trimmed, for a blind-spot record."""
428 if 1 <= line <= len(self.source_lines):
429 return self.source_lines[line - 1].strip()[:200]
430 return ""
431
432 # -- imports -----------------------------------------------------
433
434 def visit_Import(self, node: ast.Import) -> None:
435 """``import a.b.c`` imports every package on the way down."""
436 seen: set[str] = set()
437 for alias in node.names:
438 self._emit(alias.name, node.lineno, self._kind(), seen)
439
440 def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
441 """``from X import y`` reaches X and, when it exists, ``X.y``."""
442 seen: set[str] = set()
443 if node.level:
444 base = _resolve_relative(
445 self.module.name, self.module.is_package, node.level,
446 node.module,
447 )
448 if base is None:
449 self.blind_spots.append(BlindSpot(
450 source=self.module.name, line=node.lineno,
451 reason=REASON_RELATIVE,
452 snippet=self._snippet(node.lineno),
453 ))
454 return
455 elif node.module:
456 base = node.module
457 else:
458 return
459 self._emit(base, node.lineno, self._kind(), seen)
460 for alias in node.names:
461 if alias.name == "*":
462 continue
463 candidate = f"{base}.{alias.name}"
464 if candidate in self.index:
465 self._emit(candidate, node.lineno, self._kind(), seen)
466
467 # -- string imports ----------------------------------------------
468
469 def _is_import_module_call(self, node: ast.Call) -> bool:
470 """True for the ``importlib.import_module`` call shapes we know."""
471 target = node.func
472 if isinstance(target, ast.Name):
473 return target.id in self.import_module_names
474 if isinstance(target, ast.Attribute) and target.attr == "import_module":
475 base = target.value
476 return isinstance(base, ast.Name) and (
477 base.id in self.importlib_modules or base.id == "importlib"
478 )
479 return False
480
481 def visit_Call(self, node: ast.Call) -> None:
482 """A string import is an edge; a computed one is a blind spot."""
483 if self._is_import_module_call(node):
484 first = node.args[0] if node.args else None
485 if isinstance(first, ast.Constant) and isinstance(first.value, str):
486 self._emit(first.value, node.lineno, KIND_STRING, set())
487 else:
488 self.blind_spots.append(BlindSpot(
489 source=self.module.name, line=node.lineno,
490 reason=REASON_NON_LITERAL,
491 snippet=self._snippet(node.lineno),
492 ))
493 self.generic_visit(node)
494
495
497 roots: Sequence[Path | str],
498 *,
499 package: str,
500 siblings: Sequence[str] = (),
501 excludes: Sequence[str] = (),
502) -> Graph:
503 """Discover every module under *roots* and extract its import edges."""
504 func = "build_graph"
505 resolved = [Path(root) for root in roots]
506 modules, files, blind_spots = discover(resolved, excludes=excludes)
507 edges: list[Edge] = []
508 for name in sorted(modules):
509 module = modules[name]
510 text = _read(files[name])
511 try:
512 tree = ast.parse(text, filename=module.path)
513 except SyntaxError as exc:
514 blind_spots.append(BlindSpot(
515 source=name, line=exc.lineno or 0, reason=REASON_UNPARSABLE,
516 snippet=str(exc.msg)[:200],
517 ))
518 continue
519 extractor = _Extractor(
520 module=module, index=modules, package=package,
521 siblings=siblings, importlib_names=_importlib_aliases(tree),
522 source_lines=text.splitlines(),
523 )
524 extractor.visit(tree)
525 edges.extend(extractor.edges)
526 blind_spots.extend(extractor.blind_spots)
527 _dbg("OC_CHAT.MAIN", func, f"{name}: {len(extractor.edges)} edge(s)", 4)
528 ordered = tuple(sorted(
529 edges,
530 key=lambda edge: (edge.source, edge.line, edge.target, edge.kind),
531 ))
532 spots = tuple(sorted(
533 blind_spots, key=lambda spot: (spot.source, spot.line, spot.reason),
534 ))
535 _dbg(
536 "OC_CHAT.MAIN", func,
537 f"{len(ordered)} edge(s), {len(spots)} blind spot(s)", 2,
538 )
539 return Graph(modules=modules, edges=ordered, blind_spots=spots)
540
541
542# ---------------------------------------------------------------------
543# Closure and reverse edges
544# ---------------------------------------------------------------------
545
546def parse_layer(spec: str) -> tuple[str, tuple[str, ...]]:
547 """Parse ``name=mod1,mod2`` into a named seed set."""
548 if "=" not in spec:
549 raise ImportGraphError(f"layer needs the name=modules form: {spec}")
550 name, joined = spec.split("=", 1)
551 seeds = tuple(part.strip() for part in joined.split(",") if part.strip())
552 if not name.strip() or not seeds:
553 raise ImportGraphError(f"layer has an empty side: {spec}")
554 return name.strip(), seeds
555
556
557def parse_cut(spec: str) -> Cut:
558 """Parse ``source_glob->target_glob`` into a :class:`Cut`."""
559 if "->" not in spec:
560 raise ImportGraphError(f"cut needs the source->target form: {spec}")
561 source, target = spec.split("->", 1)
562 if not source.strip() or not target.strip():
563 raise ImportGraphError(f"cut has an empty side: {spec}")
564 return Cut(source=source.strip(), target=target.strip())
565
566
568 graph: Graph,
569 seeds: Sequence[str],
570 *,
571 name: str = "closure",
572 cuts: Sequence[Cut] = (),
573) -> Closure:
574 """Transitive same-package reach of *seeds*, honouring *cuts*."""
575 func = "closure"
576 unknown = [seed for seed in seeds if seed not in graph.modules]
577 if unknown:
578 raise ImportGraphError(
579 "seed not found in the scanned tree: " + ", ".join(sorted(unknown)),
580 )
581 outgoing: dict[str, list[Edge]] = {}
582 for edge in graph.edges:
583 outgoing.setdefault(edge.source, []).append(edge)
584
585 members: set[str] = set(seeds)
586 cut_edges: list[Edge] = []
587 pending = list(seeds)
588 while pending:
589 current = pending.pop()
590 for edge in outgoing.get(current, ()):
591 if any(cut.matches(edge) for cut in cuts):
592 cut_edges.append(edge)
593 continue
594 if edge.classification != SAME_PACKAGE:
595 continue
596 if edge.target not in graph.modules or edge.target in members:
597 continue
598 members.add(edge.target)
599 pending.append(edge.target)
600
601 ordered = tuple(sorted(members))
602 external: dict[str, set[str]] = {
603 STDLIB: set(), THIRD_PARTY: set(), SIBLING: set(),
604 }
605 for edge in graph.edges:
606 if edge.source not in members:
607 continue
608 if any(cut.matches(edge) for cut in cuts):
609 continue
610 if edge.classification in external:
611 external[edge.classification].add(edge.target)
612 _dbg("OC_CHAT.MAIN", func, f"{name}: closure of {len(ordered)}", 2)
613 return Closure(
614 name=name,
615 seeds=tuple(sorted(seeds)),
616 members=ordered,
617 package_inits=tuple(
618 member for member in ordered if graph.modules[member].is_package
619 ),
620 cut_edges=tuple(sorted(
621 set(cut_edges),
622 key=lambda edge: (edge.source, edge.line, edge.target),
623 )),
624 external={key: tuple(sorted(value)) for key, value in external.items()},
625 lines=sum(graph.modules[member].lines for member in ordered),
626 )
627
628
629def reverse_edges(graph: Graph) -> dict[str, tuple[str, ...]]:
630 """For every in-tree module, the sorted set of modules importing it."""
631 importers: dict[str, set[str]] = {}
632 for edge in graph.edges:
633 if edge.target in graph.modules:
634 importers.setdefault(edge.target, set()).add(edge.source)
635 return {
636 target: tuple(sorted(sources))
637 for target, sources in sorted(importers.items())
638 }
639
640
641# ---------------------------------------------------------------------
642# Output
643# ---------------------------------------------------------------------
644
645def _tally(values: Iterable[str]) -> dict[str, int]:
646 """Sorted count-by-value, for the receipt's summary blocks."""
647 counts: dict[str, int] = {}
648 for value in values:
649 counts[value] = counts.get(value, 0) + 1
650 return {key: counts[key] for key in sorted(counts)}
651
652
653def _closure_payload(result: Closure) -> dict[str, object]:
654 """One closure, as receipt JSON."""
655 return {
656 "name": result.name,
657 "seeds": list(result.seeds),
658 "size": result.size,
659 "size_excluding_inits": result.size_excluding_inits,
660 "lines": result.lines,
661 "members": list(result.members),
662 "package_inits": list(result.package_inits),
663 "external": {key: list(value) for key, value in result.external.items()},
664 "cut_edges": [_edge_payload(edge) for edge in result.cut_edges],
665 }
666
667
668def _edge_payload(edge: Edge) -> dict[str, object]:
669 """One edge, as receipt JSON."""
670 return {
671 "source": edge.source, "target": edge.target, "line": edge.line,
672 "kind": edge.kind, "scope": edge.scope, "deferred": edge.deferred,
673 "classification": edge.classification,
674 }
675
676
678 graph: Graph,
679 closures: Sequence[Closure],
680 *,
681 label: str,
682 params: dict[str, object],
683) -> dict[str, object]:
684 """Build the JSON receipt: no clock, no absolute path, fully sorted."""
685 func = "receipt"
686 _dbg("OC_CHAT.MAIN", func, f"receipt for {len(closures)} closure(s)", 3)
687 return {
688 "schema_version": SCHEMA_VERSION,
689 "label": label,
690 "params": params,
691 "tree": {
692 "modules": len(graph.modules),
693 "lines": sum(m.lines for m in graph.modules.values()),
694 "edges": len(graph.edges),
695 "edges_by_kind": _tally(e.kind for e in graph.edges),
696 "edges_by_scope": _tally(e.scope for e in graph.edges),
697 "edges_by_class": _tally(e.classification for e in graph.edges),
698 "deferred_edges": sum(1 for e in graph.edges if e.deferred),
699 "blind_spots": len(graph.blind_spots),
700 },
701 "blind_spots": [
702 {
703 "source": spot.source, "line": spot.line,
704 "reason": spot.reason, "snippet": spot.snippet,
705 }
706 for spot in graph.blind_spots
707 ],
708 "closures": [_closure_payload(result) for result in closures],
709 "modules": {
710 name: {
711 "path": module.path, "lines": module.lines,
712 "is_package": module.is_package,
713 }
714 for name, module in sorted(graph.modules.items())
715 },
716 "edges": [_edge_payload(edge) for edge in graph.edges],
717 "reverse_edges": {
718 target: list(sources)
719 for target, sources in reverse_edges(graph).items()
720 },
721 }
722
723
724def _closure_block(graph: Graph, result: Closure) -> list[str]:
725 """The human table for one closure."""
726 lines = [
727 "",
728 f"== closure: {result.name} ==",
729 f"seeds ({len(result.seeds)}): {', '.join(result.seeds)}",
730 f"size: {result.size} module(s), "
731 f"{result.size_excluding_inits} excluding package markers, "
732 f"{result.lines} line(s)",
733 "",
734 " lines module",
735 ]
736 for member in result.members:
737 marker = " (pkg)" if graph.modules[member].is_package else ""
738 lines.append(f" {graph.modules[member].lines:5d} {member}{marker}")
739 for key, title in (
740 (SIBLING, "sibling"), (THIRD_PARTY, "third party"), (STDLIB, "stdlib"),
741 ):
742 values = result.external.get(key, ())
743 lines.append(f" {title}: {', '.join(values) if values else 'none'}")
744 if result.cut_edges:
745 lines.append(f" cut edges ({len(result.cut_edges)}):")
746 for edge in result.cut_edges:
747 lines.append(
748 f" {edge.source}:{edge.line} -> {edge.target} "
749 f"[{edge.kind}]",
750 )
751 return lines
752
753
755 graph: Graph,
756 closures: Sequence[Closure],
757 *,
758 label: str,
759 roots: Sequence[str],
760) -> str:
761 """The human report. ASCII only, and stable across runs."""
762 lines = [
763 "== import graph ==",
764 f"label: {label}",
765 f"roots: {', '.join(roots)}",
766 f"modules: {len(graph.modules)} "
767 f"lines: {sum(m.lines for m in graph.modules.values())} "
768 f"edges: {len(graph.edges)}",
769 f"by kind: {_tally(e.kind for e in graph.edges)}",
770 f"by class: {_tally(e.classification for e in graph.edges)}",
771 f"deferred: {sum(1 for e in graph.edges if e.deferred)} "
772 f"type-checking: "
773 f"{sum(1 for e in graph.edges if e.scope == SCOPE_TYPE_CHECKING)}",
774 f"blind spots: {len(graph.blind_spots)}",
775 ]
776 for spot in graph.blind_spots:
777 lines.append(f" {spot.source}:{spot.line} {spot.reason}")
778 if spot.snippet:
779 lines.append(f" {spot.snippet}")
780 for result in closures:
781 lines.extend(_closure_block(graph, result))
782 return "\n".join(lines) + "\n"
783
784
785# ---------------------------------------------------------------------
786# CLI
787# ---------------------------------------------------------------------
788
789def build_parser() -> argparse.ArgumentParser:
790 """The CLI surface; every question is an argument, none is baked in."""
791 parser = argparse.ArgumentParser(
792 prog="import_graph",
793 description="AST import-graph closures with reproducible receipts.",
794 )
795 parser.add_argument(
796 "--root", action="append", required=True, metavar="DIR",
797 help="directory on the import path; repeatable, first wins on clash",
798 )
799 parser.add_argument(
800 "--package", required=True,
801 help="top-level package whose edges the closure follows",
802 )
803 parser.add_argument(
804 "--sibling", action="append", default=[], metavar="NAME",
805 help="top-level name to classify as a sibling package; repeatable",
806 )
807 parser.add_argument(
808 "--seed", action="append", default=[], metavar="MODULE",
809 help="closure seed; repeatable",
810 )
811 parser.add_argument(
812 "--closure-name", default="closure",
813 help="name recorded for this closure in the receipt",
814 )
815 parser.add_argument(
816 "--layer", action="append", default=[], metavar="NAME=MODULES",
817 help="a named, comma-separated seed set; repeatable, so one run "
818 "and one receipt can cover several layers of the same tree",
819 )
820 parser.add_argument(
821 "--cut", action="append", default=[], metavar="SRC->DST",
822 help="suppress matching edges; both sides accept glob wildcards. "
823 "Cuts are global to the run: a layer with no matching edge is "
824 "simply unaffected",
825 )
826 parser.add_argument(
827 "--exclude", action="append", default=[], metavar="GLOB",
828 help="skip root-relative POSIX paths matching this glob; repeatable",
829 )
830 parser.add_argument(
831 "--json", metavar="PATH", help="write the JSON receipt here",
832 )
833 parser.add_argument(
834 "--label", default="",
835 help="what tree these numbers belong to; recorded verbatim",
836 )
837 return parser
838
839
840def main(argv: Sequence[str] | None = None) -> int:
841 """Run one measurement. 0 on success, 1 on a named refusal."""
842 func = "main"
843 args = build_parser().parse_args(list(argv) if argv is not None else None)
844 try:
845 cuts = [parse_cut(spec) for spec in args.cut]
846 layers: list[tuple[str, Sequence[str]]] = []
847 if args.seed:
848 layers.append((args.closure_name, args.seed))
849 layers.extend(parse_layer(spec) for spec in args.layer)
850 graph = build_graph(
851 [Path(root) for root in args.root], package=args.package,
852 siblings=args.sibling, excludes=args.exclude,
853 )
854 closures: list[Closure] = []
855 for name, seeds in layers:
856 closures.append(closure(graph, seeds, name=name))
857 if cuts:
858 closures.append(closure(
859 graph, seeds, name=f"{name}+cuts", cuts=cuts,
860 ))
861 except ImportGraphError as exc:
862 print(f"import_graph: {exc}", file=sys.stderr)
863 return 1
864
865 root_names = [Path(root).resolve().name for root in args.root]
866 print(render_table(graph, closures, label=args.label, roots=root_names),
867 end="")
868 if args.json:
869 params: dict[str, object] = {
870 "roots": root_names,
871 "package": args.package,
872 "siblings": sorted(args.sibling),
873 "excludes": sorted(args.exclude),
874 "seeds": sorted(args.seed),
875 "cuts": [cut.spec() for cut in cuts],
876 "closure_name": args.closure_name,
877 "layers": sorted(args.layer),
878 }
879 document = receipt(graph, closures, label=args.label, params=params)
880 text = json.dumps(document, indent=2, sort_keys=True) + "\n"
881 Path(args.json).write_text(text, encoding="utf-8", newline="\n")
882 _dbg("OC_CHAT.MAIN", func, f"receipt written to {args.json}", 2)
883 return 0
884
885
886if __name__ == "__main__":
887 raise SystemExit(main())
str _snippet(self, int line)
None visit_Call(self, ast.Call node)
bool _is_import_module_call(self, ast.Call node)
None visit_ImportFrom(self, ast.ImportFrom node)
None __init__(self, *, Module module, Iterable[str] index, str package, Sequence[str] siblings, tuple[set[str], set[str]] importlib_names, Sequence[str] source_lines)
None _visit_function(self, ast.AST node)
None visit_Import(self, ast.Import node)
None _emit(self, str target, int line, str kind, set[str] seen)
None visit_If(self, ast.If node)
int size_excluding_inits(self)
bool matches(self, Edge edge)
bool _is_type_checking(ast.expr test)
Closure closure(Graph graph, Sequence[str] seeds, *, str name="closure", Sequence[Cut] cuts=())
int _count_lines(str text)
tuple[set[str], set[str]] _importlib_aliases(ast.AST tree)
str|None _resolve_relative(str module, bool is_package, int level, str|None tail)
str _read(Path path)
list[str] _closure_block(Graph graph, Closure result)
bool _excluded(str relative, Sequence[str] excludes)
int main(Sequence[str]|None argv=None)
dict[str, tuple[str,...]] reverse_edges(Graph graph)
argparse.ArgumentParser build_parser()
Graph build_graph(Sequence[Path|str] roots, *, str package, Sequence[str] siblings=(), Sequence[str] excludes=())
str classify(str target, *, str package, Sequence[str] siblings)
Cut parse_cut(str spec)
tuple[str, tuple[str,...]] parse_layer(str spec)
str render_table(Graph graph, Sequence[Closure] closures, *, str label, Sequence[str] roots)
dict[str, int] _tally(Iterable[str] values)
dict[str, object] _edge_payload(Edge edge)
dict[str, object] receipt(Graph graph, Sequence[Closure] closures, *, str label, dict[str, object] params)
tuple[str, bool] _module_name(Path root, Path path)
tuple[dict[str, Module], dict[str, Path], list[BlindSpot]] discover(Sequence[Path] roots, *, Sequence[str] excludes=())
dict[str, object] _closure_payload(Closure result)