typing_audit.py
python
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago
| 1 | #!/usr/bin/env python3 |
| 2 | """Typing audit — zero-tolerance type-safety enforcement for mission-critical code. |
| 3 | |
| 4 | Every banned pattern maps to a future Rust port liability: if Python cannot |
| 5 | name a type, ``rustc`` cannot either. The ratchet keeps the rule enforced |
| 6 | continuously so violations never accumulate. |
| 7 | |
| 8 | Patterns checked |
| 9 | ---------------- |
| 10 | *Any-as-type* — ``dict[str, Any]``, ``list[Any]``, ``type[Any]``, |
| 11 | ``Any | X``, ``X | Any``, ``Mapping[str, Any]``, etc. |
| 12 | |
| 13 | *object-as-type* — same severity as Any; erases all structural information. |
| 14 | |
| 15 | *cast()* — all usage banned; it conceals a broken callee return type. |
| 16 | |
| 17 | *# type: ignore* — every suppressed error is an unaudited assumption. |
| 18 | |
| 19 | *Bare collections* — ``list``, ``dict``, ``set``, ``tuple`` without ``[T]``. |
| 20 | |
| 21 | *Optional[X]* and *Union[X, Y]* — use ``X | None`` and ``X | Y`` (PEP 604). |
| 22 | |
| 23 | *Legacy typing imports* — ``List``, ``Dict``, ``Set``, ``Tuple``. |
| 24 | |
| 25 | *Bare Callable / Callable returning Any* — must carry a full signature. |
| 26 | |
| 27 | *Untyped varargs* — ``*args: Any``, ``**kwargs: Any``, and unannotated |
| 28 | ``*args`` / ``**kwargs`` (annotation absent entirely). |
| 29 | |
| 30 | *Untyped function definitions* — missing return or parameter annotation. |
| 31 | |
| 32 | *Unconstrained TypeVar* — ``TypeVar(...)`` with no ``bound=`` and no |
| 33 | constraint arguments; behaves identically to ``Any`` in practice. |
| 34 | |
| 35 | *Naked dict at boundary* — ``dict[str, X]`` as a parameter or return type |
| 36 | is banned at function/method boundaries. Every dict with known keys must |
| 37 | be a ``TypedDict``; every dict with dynamic keys must justify its key space. |
| 38 | The only valid ``dict[str, ...]`` at a boundary is an explicitly named |
| 39 | ``TypedDict`` subclass. This rule exists because ``rustc`` cannot infer |
| 40 | struct fields from a ``HashMap<String, X>`` — named fields must be declared. |
| 41 | Pattern ``boundary_dict`` fires on ``: dict[str,`` and ``-> dict[str,``. |
| 42 | |
| 43 | Usage:: |
| 44 | |
| 45 | python tools/typing_audit.py # musehub/ + tests/ |
| 46 | python tools/typing_audit.py --dirs musehub/ tests/ |
| 47 | python tools/typing_audit.py --dirs musehub/ --max-any 0 --max-untyped 0 |
| 48 | python tools/typing_audit.py --json artifacts/typing_audit.json |
| 49 | """ |
| 50 | |
| 51 | from __future__ import annotations |
| 52 | |
| 53 | import argparse |
| 54 | import ast |
| 55 | import io |
| 56 | import json |
| 57 | import operator |
| 58 | import re |
| 59 | import sys |
| 60 | import tokenize |
| 61 | from collections import defaultdict |
| 62 | from pathlib import Path |
| 63 | from typing import TypedDict |
| 64 | |
| 65 | # --------------------------------------------------------------------------- |
| 66 | # Type aliases — avoid dict[str, X] at function/class-field boundaries. |
| 67 | # --------------------------------------------------------------------------- |
| 68 | |
| 69 | type PatternCounts = dict[str, int] |
| 70 | type PatternLines = dict[str, list[int]] |
| 71 | type PatternMap = dict[str, re.Pattern[str]] |
| 72 | type PerFileViolations = dict[str, PatternCounts] |
| 73 | |
| 74 | # --------------------------------------------------------------------------- |
| 75 | # Data shapes — TypedDicts replace every dict[str, Any] in the old script. |
| 76 | # All shapes mirror the Rust struct that will eventually own them. |
| 77 | # --------------------------------------------------------------------------- |
| 78 | |
| 79 | |
| 80 | class UntypedDef(TypedDict): |
| 81 | """A function or method that is missing a required type annotation. |
| 82 | |
| 83 | ``issue`` is one of: |
| 84 | |
| 85 | - ``"missing_return_type"`` — no return annotation. |
| 86 | - ``"missing_param_type"`` — a non-self/cls parameter lacks annotation. |
| 87 | - ``"untyped_args"`` — ``*args`` is annotated as ``Any`` or has |
| 88 | no annotation at all. |
| 89 | - ``"untyped_kwargs"`` — ``**kwargs`` is annotated as ``Any`` or has |
| 90 | no annotation at all. |
| 91 | - ``"unconstrained_typevar"``— a ``TypeVar`` with no ``bound=`` and no |
| 92 | positional constraints. |
| 93 | """ |
| 94 | |
| 95 | file: str |
| 96 | line: int |
| 97 | name: str |
| 98 | issue: str |
| 99 | |
| 100 | |
| 101 | class FileResult(TypedDict): |
| 102 | """Typing-violation summary for a single Python source file.""" |
| 103 | |
| 104 | file: str |
| 105 | imports_any: bool |
| 106 | patterns: PatternCounts |
| 107 | pattern_lines: PatternLines |
| 108 | type_ignore_variants: PatternCounts |
| 109 | untyped_defs: list[UntypedDef] |
| 110 | |
| 111 | |
| 112 | class Offender(TypedDict): |
| 113 | """A file with at least one typing violation, ranked by total count.""" |
| 114 | |
| 115 | file: str |
| 116 | total: int |
| 117 | patterns: PatternCounts |
| 118 | |
| 119 | |
| 120 | class ReportSummary(TypedDict): |
| 121 | """High-level aggregate counts for the entire scan.""" |
| 122 | |
| 123 | total_files_scanned: int |
| 124 | files_importing_any: int |
| 125 | total_any_patterns: int |
| 126 | untyped_defs: int |
| 127 | |
| 128 | |
| 129 | class Report(TypedDict): |
| 130 | """Full typing-audit report produced by :func:`generate_report`.""" |
| 131 | |
| 132 | summary: ReportSummary |
| 133 | pattern_totals: PatternCounts |
| 134 | type_ignore_variants: PatternCounts |
| 135 | top_offenders: list[Offender] |
| 136 | per_file: PerFileViolations |
| 137 | untyped_defs: list[UntypedDef] |
| 138 | |
| 139 | |
| 140 | # --------------------------------------------------------------------------- |
| 141 | # String-literal masking |
| 142 | # --------------------------------------------------------------------------- |
| 143 | |
| 144 | |
| 145 | def _mask_string_literals(source: str) -> str: |
| 146 | """Replace string-literal content with spaces, preserving newlines. |
| 147 | |
| 148 | Pattern matching runs on the masked source so that raw regex strings, |
| 149 | docstrings, and string constants never produce false positives. All |
| 150 | newlines are preserved so that line numbers stay accurate. |
| 151 | |
| 152 | Tokenisation errors (e.g. incomplete source snippets) are silently |
| 153 | ignored — the original source is returned unchanged so the caller still |
| 154 | produces *some* output rather than silently dropping the file. |
| 155 | |
| 156 | Args: |
| 157 | source: Full UTF-8 source text of a Python file. |
| 158 | |
| 159 | Returns: |
| 160 | A copy of *source* with the content of every string token replaced |
| 161 | by space characters (newlines within multi-line strings preserved). |
| 162 | """ |
| 163 | chars = list(source) |
| 164 | lines = source.splitlines(keepends=True) |
| 165 | |
| 166 | # Pre-compute cumulative line offsets for O(1) (row, col) → offset. |
| 167 | offsets: list[int] = [0] |
| 168 | for ln in lines: |
| 169 | offsets.append(offsets[-1] + len(ln)) |
| 170 | |
| 171 | def _abs(row: int, col: int) -> int: |
| 172 | return offsets[row - 1] + col |
| 173 | |
| 174 | # Token types that contain string literal content — including f-string |
| 175 | # middle segments which are FSTRING_MIDDLE (not STRING) in Python 3.12+. |
| 176 | _FSTRING_MIDDLE = getattr(tokenize, "FSTRING_MIDDLE", None) |
| 177 | _STRING_TYPES = {tokenize.STRING} |
| 178 | if _FSTRING_MIDDLE is not None: |
| 179 | _STRING_TYPES.add(_FSTRING_MIDDLE) |
| 180 | |
| 181 | try: |
| 182 | gen = tokenize.generate_tokens(io.StringIO(source).readline) |
| 183 | for tok_type, _tok_str, (srow, scol), (erow, ecol), _ in gen: |
| 184 | if tok_type not in _STRING_TYPES: |
| 185 | continue |
| 186 | start = _abs(srow, scol) |
| 187 | end = _abs(erow, ecol) |
| 188 | for i in range(start, end): |
| 189 | if chars[i] not in {"\n", "\r"}: |
| 190 | chars[i] = " " |
| 191 | except tokenize.TokenError: |
| 192 | pass |
| 193 | |
| 194 | return "".join(chars) |
| 195 | |
| 196 | |
| 197 | # --------------------------------------------------------------------------- |
| 198 | # Pattern registry |
| 199 | # --------------------------------------------------------------------------- |
| 200 | |
| 201 | #: All patterns that count toward the violation total. |
| 202 | #: Keys are stable identifiers used in JSON output and tests. |
| 203 | #: |
| 204 | #: NOTE: do NOT use re.IGNORECASE — Python type annotations are case-sensitive. |
| 205 | #: ``List`` and ``list`` are distinct identifiers; matching ``list[any]`` |
| 206 | #: (where ``any`` is the built-in function) would be a false positive. |
| 207 | _PATTERNS: PatternMap = { |
| 208 | # Any-as-type ───────────────────────────────────────────────────────── |
| 209 | "dict_str_any": re.compile(r"\bdict\[str,\s*Any\]|\bDict\[str,\s*Any\]"), |
| 210 | "list_any": re.compile(r"\blist\[Any\]|\bList\[Any\]"), |
| 211 | "type_any": re.compile(r"\btype\[Any\]"), |
| 212 | "any_in_union": re.compile(r"\bAny\s*\||\|\s*Any\b"), |
| 213 | "return_any": re.compile(r"->\s*Any\b"), |
| 214 | "param_any": re.compile(r":\s*Any\b"), |
| 215 | "mapping_any": re.compile(r"\bMapping\[str,\s*Any\]"), |
| 216 | "optional_any": re.compile(r"\bOptional\[Any\]"), |
| 217 | "sequence_any": re.compile(r"\bSequence\[Any\]|\bIterable\[Any\]"), |
| 218 | "tuple_any": re.compile(r"\btuple\[[^\n]*Any[^\n]*\]|\bTuple\[[^\n]*Any[^\n]*\]"), |
| 219 | # object-as-type ────────────────────────────────────────────────────── |
| 220 | "param_object": re.compile(r":\s*object\b"), |
| 221 | "return_object": re.compile(r"->\s*object\b"), |
| 222 | # Handles one level of nesting, e.g. dict[str, list[object]]. |
| 223 | "collection_object": re.compile( |
| 224 | r"\b(?:dict|list|set|tuple|Sequence|Mapping)" |
| 225 | r"\[[^\n\[\]]*(?:\[[^\n\[\]]*\][^\n\[\]]*)*\bobject\b" |
| 226 | ), |
| 227 | # cast() — banned ───────────────────────────────────────────────────── |
| 228 | "cast_usage": re.compile(r"\bcast\("), |
| 229 | # type: ignore — suppresses real errors ─────────────────────────────── |
| 230 | "type_ignore": re.compile(r"#\s*type:\s*ignore"), |
| 231 | # Bare collections (no type parameters) ─────────────────────────────── |
| 232 | # Negative lookaheads exclude parameterised forms and prose. |
| 233 | "bare_list": re.compile(r"(?::\s*|->\s*)list\b(?!\[|\(|\s+[a-z])"), |
| 234 | "bare_dict": re.compile(r"(?::\s*|->\s*)dict\b(?!\[|\(|\s+[a-z])"), |
| 235 | "bare_set": re.compile(r"(?::\s*|->\s*)set\b(?!\[|\(|\s+[a-z])"), |
| 236 | "bare_tuple": re.compile(r"(?::\s*|->\s*)tuple\b(?!\[|\(|\s+[a-z])"), |
| 237 | # Optional[X] — use X | None (PEP 604) ──────────────────────────────── |
| 238 | "optional_usage": re.compile(r"\bOptional\[(?!Any\b)"), |
| 239 | # Union[X, Y] — use X | Y (PEP 604) ────────────────────────────────── |
| 240 | "union_usage": re.compile(r"\bUnion\["), |
| 241 | # Legacy typing imports (use lowercase builtins) ────────────────────── |
| 242 | "legacy_List": re.compile(r"\bList\["), |
| 243 | "legacy_Dict": re.compile(r"\bDict\["), |
| 244 | "legacy_Set": re.compile(r"\bSet\["), |
| 245 | "legacy_Tuple": re.compile(r"\bTuple\["), |
| 246 | # Callable — must carry full signature ──────────────────────────────── |
| 247 | "bare_callable": re.compile(r"(?::\s*|->\s*)Callable\b(?!\[)"), |
| 248 | "callable_any": re.compile(r"\bCallable\[[^\n]*,\s*Any\s*\]"), |
| 249 | # Untyped varargs — *args: Any / **kwargs: Any ──────────────────────── |
| 250 | # Unannotated *args/**kwargs are caught by the AST walker instead. |
| 251 | "varargs_any": re.compile(r"\*{1,2}\w+:\s*Any\b"), |
| 252 | # Naked dict at boundary — dict[str, X] as param/return type is banned. |
| 253 | # Every structured boundary must use a TypedDict (or dataclass/enum). |
| 254 | # Matches ": dict[str," and "-> dict[str," — the two annotation positions. |
| 255 | "boundary_dict": re.compile(r"(?::\s*|->\s*)dict\[str\s*,"), |
| 256 | } |
| 257 | |
| 258 | # Category groupings for the human-readable report, in display order. |
| 259 | _CATEGORY_ORDER: list[tuple[str, list[str]]] = [ |
| 260 | ("Any-as-type", [ |
| 261 | "dict_str_any", "list_any", "type_any", "any_in_union", |
| 262 | "return_any", "param_any", |
| 263 | "mapping_any", "optional_any", "sequence_any", "tuple_any", |
| 264 | ]), |
| 265 | ("object-as-type", ["param_object", "return_object", "collection_object"]), |
| 266 | ("cast() usage", ["cast_usage"]), |
| 267 | ("type: ignore", ["type_ignore"]), |
| 268 | ("Bare collections", ["bare_list", "bare_dict", "bare_set", "bare_tuple"]), |
| 269 | ("Optional (use X | None)", ["optional_usage"]), |
| 270 | ("Union (use X | Y)", ["union_usage"]), |
| 271 | ("Legacy typing imports", ["legacy_List", "legacy_Dict", "legacy_Set", "legacy_Tuple"]), |
| 272 | ("Callable (must carry full signature)", ["bare_callable", "callable_any"]), |
| 273 | ("Untyped varargs", ["varargs_any"]), |
| 274 | ("Naked dict at boundary (use TypedDict)", ["boundary_dict"]), |
| 275 | ] |
| 276 | |
| 277 | # Directories that are never source code and must be skipped during scanning. |
| 278 | _SKIP_DIRS: frozenset[str] = frozenset({ |
| 279 | "venv", ".venv", "env", ".env", |
| 280 | "__pycache__", |
| 281 | ".git", ".muse", ".mypy_cache", ".ruff_cache", ".pytest_cache", ".tox", |
| 282 | "dist", "build", "site-packages", "__pypackages__", |
| 283 | "node_modules", |
| 284 | }) |
| 285 | |
| 286 | |
| 287 | # --------------------------------------------------------------------------- |
| 288 | # Pattern helpers |
| 289 | # --------------------------------------------------------------------------- |
| 290 | |
| 291 | |
| 292 | def _count_pattern_in_line(line: str, pattern: re.Pattern[str]) -> int: |
| 293 | """Return the number of non-overlapping matches of *pattern* in *line*.""" |
| 294 | return len(pattern.findall(line)) |
| 295 | |
| 296 | |
| 297 | def _imports_any(source: str) -> bool: |
| 298 | """Return ``True`` if the source file imports ``Any`` from ``typing`` |
| 299 | or ``typing_extensions``. |
| 300 | |
| 301 | Excludes commented-out import lines (lines where ``from`` is preceded only |
| 302 | by ``#`` and optional whitespace). |
| 303 | """ |
| 304 | return bool(re.search( |
| 305 | r"^[ \t]*from\s+typing(?:_extensions)?\s+import\s+.*\bAny\b", |
| 306 | source, |
| 307 | re.MULTILINE, |
| 308 | )) |
| 309 | |
| 310 | |
| 311 | def _classify_type_ignore(line: str) -> str: |
| 312 | """Classify the style of a ``# type: ignore`` comment. |
| 313 | |
| 314 | Returns ``"type_ignore[code]"`` for code-specific ignores, or |
| 315 | ``"type_ignore[blanket]"`` for bare ``# type: ignore``. |
| 316 | |
| 317 | Args: |
| 318 | line: A single source line that contains ``# type: ignore``. |
| 319 | |
| 320 | Returns: |
| 321 | A string label for the variant. |
| 322 | """ |
| 323 | m = re.search(r"#\s*type:\s*ignore\[([^\]]+)\]", line) |
| 324 | if m: |
| 325 | return f"type_ignore[{m.group(1)}]" |
| 326 | return "type_ignore[blanket]" |
| 327 | |
| 328 | |
| 329 | # --------------------------------------------------------------------------- |
| 330 | # AST-based detection |
| 331 | # --------------------------------------------------------------------------- |
| 332 | |
| 333 | |
| 334 | def _is_any_annotation(node: ast.expr | None) -> bool: |
| 335 | """Return ``True`` if *node* is the bare ``Any`` name.""" |
| 336 | return isinstance(node, ast.Name) and node.id == "Any" |
| 337 | |
| 338 | |
| 339 | def _find_untyped_defs(source: str, filepath: str) -> list[UntypedDef]: |
| 340 | """Walk the AST and collect every function with a missing annotation. |
| 341 | |
| 342 | Checks: |
| 343 | |
| 344 | - Missing return type (``node.returns is None``). |
| 345 | - Missing parameter annotation (excluding ``self`` and ``cls``). |
| 346 | - ``*args`` annotated as ``Any`` **or** with no annotation at all. |
| 347 | - ``**kwargs`` annotated as ``Any`` **or** with no annotation at all. |
| 348 | - ``TypeVar(...)`` assignments with no ``bound=`` and no constraints. |
| 349 | |
| 350 | Line numbers for parameter violations use the argument's own line number |
| 351 | (``arg.lineno``) rather than the function definition line, so the report |
| 352 | points directly at the problematic parameter. |
| 353 | |
| 354 | Skips files that cannot be parsed. |
| 355 | |
| 356 | Args: |
| 357 | source: Full source text of the file. |
| 358 | filepath: Path string used in the returned records. |
| 359 | |
| 360 | Returns: |
| 361 | A list of :class:`UntypedDef` records, one per violation found. |
| 362 | """ |
| 363 | results: list[UntypedDef] = [] |
| 364 | try: |
| 365 | tree = ast.parse(source) |
| 366 | except SyntaxError: |
| 367 | return results |
| 368 | |
| 369 | for node in ast.walk(tree): |
| 370 | if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 371 | continue |
| 372 | |
| 373 | if node.returns is None: |
| 374 | results.append(UntypedDef( |
| 375 | file=filepath, |
| 376 | line=node.lineno, |
| 377 | name=node.name, |
| 378 | issue="missing_return_type", |
| 379 | )) |
| 380 | |
| 381 | all_args = ( |
| 382 | node.args.args |
| 383 | + node.args.posonlyargs |
| 384 | + node.args.kwonlyargs |
| 385 | ) |
| 386 | for arg in all_args: |
| 387 | if arg.arg in {"self", "cls"}: |
| 388 | continue |
| 389 | if arg.annotation is None: |
| 390 | results.append(UntypedDef( |
| 391 | file=filepath, |
| 392 | line=arg.lineno, |
| 393 | name=f"{node.name}.{arg.arg}", |
| 394 | issue="missing_param_type", |
| 395 | )) |
| 396 | |
| 397 | vararg = node.args.vararg |
| 398 | if vararg is not None: |
| 399 | if vararg.annotation is None or _is_any_annotation(vararg.annotation): |
| 400 | results.append(UntypedDef( |
| 401 | file=filepath, |
| 402 | line=vararg.lineno, |
| 403 | name=f"{node.name}.*{vararg.arg}", |
| 404 | issue="untyped_args", |
| 405 | )) |
| 406 | |
| 407 | kwarg = node.args.kwarg |
| 408 | if kwarg is not None: |
| 409 | if kwarg.annotation is None or _is_any_annotation(kwarg.annotation): |
| 410 | results.append(UntypedDef( |
| 411 | file=filepath, |
| 412 | line=kwarg.lineno, |
| 413 | name=f"{node.name}.**{kwarg.arg}", |
| 414 | issue="untyped_kwargs", |
| 415 | )) |
| 416 | |
| 417 | # TypeVar without constraints or bound — behaves identically to Any. |
| 418 | results.extend(_find_unconstrained_typevars(tree, filepath)) |
| 419 | |
| 420 | return results |
| 421 | |
| 422 | |
| 423 | def _find_unconstrained_typevars(tree: ast.Module, filepath: str) -> list[UntypedDef]: |
| 424 | """Return a record for every ``TypeVar(...)`` with no bound or constraints. |
| 425 | |
| 426 | A bare ``T = TypeVar("T")`` is semantically equivalent to ``T: Any``. |
| 427 | The Rust port requires every generic to carry an explicit trait bound. |
| 428 | |
| 429 | Args: |
| 430 | tree: Parsed AST of the file. |
| 431 | filepath: Path string used in the returned records. |
| 432 | |
| 433 | Returns: |
| 434 | A list of :class:`UntypedDef` records for unconstrained ``TypeVar`` |
| 435 | definitions. |
| 436 | """ |
| 437 | results: list[UntypedDef] = [] |
| 438 | for node in ast.walk(tree): |
| 439 | # Match: T = TypeVar("T") or T = TypeVar("T", bound=...) |
| 440 | if not isinstance(node, ast.Assign): |
| 441 | continue |
| 442 | value = node.value |
| 443 | if not isinstance(value, ast.Call): |
| 444 | continue |
| 445 | func = value.func |
| 446 | if not (isinstance(func, ast.Name) and func.id == "TypeVar"): |
| 447 | continue |
| 448 | # A TypeVar is constrained when it has: |
| 449 | # - positional args beyond the name (constraint types), OR |
| 450 | # - a keyword arg named "bound" |
| 451 | extra_args = value.args[1:] # args[0] is the name string |
| 452 | kw_names = {kw.arg for kw in value.keywords} |
| 453 | if extra_args or "bound" in kw_names: |
| 454 | continue # constrained — OK |
| 455 | # Unconstrained TypeVar. |
| 456 | target_name = ( |
| 457 | node.targets[0].id |
| 458 | if isinstance(node.targets[0], ast.Name) |
| 459 | else "<TypeVar>" |
| 460 | ) |
| 461 | results.append(UntypedDef( |
| 462 | file=filepath, |
| 463 | line=node.lineno, |
| 464 | name=target_name, |
| 465 | issue="unconstrained_typevar", |
| 466 | )) |
| 467 | return results |
| 468 | |
| 469 | |
| 470 | # --------------------------------------------------------------------------- |
| 471 | # File and directory scanner |
| 472 | # --------------------------------------------------------------------------- |
| 473 | |
| 474 | |
| 475 | def scan_file(filepath: Path) -> FileResult | None: |
| 476 | """Scan a single Python file and return its violation summary. |
| 477 | |
| 478 | String literals are masked before pattern matching so that raw regex |
| 479 | strings and docstring prose never produce false positives. The |
| 480 | ``# type: ignore`` check runs on the *original* source because those |
| 481 | comments are not string literals. |
| 482 | |
| 483 | Returns ``None`` when the file cannot be read (I/O or encoding error). |
| 484 | |
| 485 | Args: |
| 486 | filepath: Absolute or relative path to the Python file. |
| 487 | |
| 488 | Returns: |
| 489 | A :class:`FileResult` on success, ``None`` on I/O failure. |
| 490 | """ |
| 491 | try: |
| 492 | source = filepath.read_text(encoding="utf-8") |
| 493 | except (OSError, UnicodeDecodeError): |
| 494 | return None |
| 495 | |
| 496 | masked = _mask_string_literals(source) |
| 497 | |
| 498 | original_lines = source.splitlines() |
| 499 | masked_lines = masked.splitlines() |
| 500 | |
| 501 | patterns: defaultdict[str, int] = defaultdict(int) |
| 502 | pattern_lines: defaultdict[str, list[int]] = defaultdict(list) |
| 503 | type_ignore_variants: defaultdict[str, int] = defaultdict(int) |
| 504 | |
| 505 | for lineno, (orig_line, masked_line) in enumerate( |
| 506 | zip(original_lines, masked_lines), 1 |
| 507 | ): |
| 508 | stripped = masked_line.strip() |
| 509 | if not stripped or stripped.startswith("#"): |
| 510 | continue |
| 511 | |
| 512 | for name, pattern in _PATTERNS.items(): |
| 513 | # All patterns run on the masked line — string literals are blanked |
| 514 | # so raw regex strings and docstring prose never trigger false |
| 515 | # positives. Comments are NOT masked (they are not string tokens) |
| 516 | # so "# type: ignore" on real code lines is still detected. |
| 517 | count = _count_pattern_in_line(masked_line, pattern) |
| 518 | if count > 0: |
| 519 | patterns[name] += count |
| 520 | pattern_lines[name].append(lineno) |
| 521 | |
| 522 | if name == "type_ignore": |
| 523 | # Classify against the original line so we can distinguish |
| 524 | # blanket ignores from code-specific ones. |
| 525 | variant = _classify_type_ignore(orig_line) |
| 526 | type_ignore_variants[variant] += 1 |
| 527 | |
| 528 | return FileResult( |
| 529 | file=str(filepath), |
| 530 | imports_any=_imports_any(source), |
| 531 | patterns=dict(patterns), |
| 532 | pattern_lines=dict(pattern_lines), |
| 533 | type_ignore_variants=dict(type_ignore_variants), |
| 534 | untyped_defs=_find_untyped_defs(source, str(filepath)), |
| 535 | ) |
| 536 | |
| 537 | |
| 538 | def scan_directory(directory: Path) -> list[FileResult]: |
| 539 | """Recursively scan all Python files in *directory*. |
| 540 | |
| 541 | Skips virtual environments, caches, build artefacts, and VCS/tool |
| 542 | metadata directories (see ``_SKIP_DIRS``). |
| 543 | |
| 544 | Args: |
| 545 | directory: Root of the directory tree to scan. |
| 546 | |
| 547 | Returns: |
| 548 | A list of :class:`FileResult` objects, one per successfully scanned file. |
| 549 | """ |
| 550 | results: list[FileResult] = [] |
| 551 | for py_file in sorted(directory.rglob("*.py")): |
| 552 | if any(part in _SKIP_DIRS for part in py_file.parts): |
| 553 | continue |
| 554 | file_result = scan_file(py_file) |
| 555 | if file_result is not None: |
| 556 | results.append(file_result) |
| 557 | return results |
| 558 | |
| 559 | |
| 560 | # --------------------------------------------------------------------------- |
| 561 | # Report generation |
| 562 | # --------------------------------------------------------------------------- |
| 563 | |
| 564 | |
| 565 | def _offender_sort_key(entry: Offender) -> int: |
| 566 | """Return the sort key for an :class:`Offender` (total violation count).""" |
| 567 | return entry["total"] |
| 568 | |
| 569 | |
| 570 | def generate_report(results: list[FileResult]) -> Report: |
| 571 | """Aggregate per-file scan results into a :class:`Report`. |
| 572 | |
| 573 | Args: |
| 574 | results: List of :class:`FileResult` objects from :func:`scan_file` |
| 575 | or :func:`scan_directory`. |
| 576 | |
| 577 | Returns: |
| 578 | A :class:`Report` ready for human display or JSON serialisation. |
| 579 | """ |
| 580 | totals: defaultdict[str, int] = defaultdict(int) |
| 581 | files_with_any_import = 0 |
| 582 | per_file: PerFileViolations = {} |
| 583 | top_offenders: list[Offender] = [] |
| 584 | all_type_ignore_variants: defaultdict[str, int] = defaultdict(int) |
| 585 | all_untyped_defs: list[UntypedDef] = [] |
| 586 | |
| 587 | for r in results: |
| 588 | filepath = r["file"] |
| 589 | if r["imports_any"]: |
| 590 | files_with_any_import += 1 |
| 591 | |
| 592 | file_total = 0 |
| 593 | file_patterns: PatternCounts = {} |
| 594 | for pattern, count in r["patterns"].items(): |
| 595 | totals[pattern] += count |
| 596 | file_patterns[pattern] = count |
| 597 | file_total += count |
| 598 | |
| 599 | if file_total > 0: |
| 600 | per_file[filepath] = file_patterns |
| 601 | top_offenders.append(Offender( |
| 602 | file=filepath, |
| 603 | total=file_total, |
| 604 | patterns=file_patterns, |
| 605 | )) |
| 606 | |
| 607 | for variant, count in r["type_ignore_variants"].items(): |
| 608 | all_type_ignore_variants[variant] += count |
| 609 | |
| 610 | all_untyped_defs.extend(r["untyped_defs"]) |
| 611 | |
| 612 | top_offenders.sort(key=_offender_sort_key, reverse=True) |
| 613 | |
| 614 | return Report( |
| 615 | summary=ReportSummary( |
| 616 | total_files_scanned=len(results), |
| 617 | files_importing_any=files_with_any_import, |
| 618 | total_any_patterns=sum(totals.values()), |
| 619 | untyped_defs=len(all_untyped_defs), |
| 620 | ), |
| 621 | pattern_totals=dict(totals), |
| 622 | type_ignore_variants=dict(all_type_ignore_variants), |
| 623 | # Store all offenders in JSON; display is capped separately in the |
| 624 | # human-readable printer. |
| 625 | top_offenders=top_offenders, |
| 626 | per_file=per_file, |
| 627 | # Store the full list — callers that need all records can use --json. |
| 628 | untyped_defs=all_untyped_defs, |
| 629 | ) |
| 630 | |
| 631 | |
| 632 | # --------------------------------------------------------------------------- |
| 633 | # Human-readable report printer |
| 634 | # --------------------------------------------------------------------------- |
| 635 | |
| 636 | |
| 637 | def print_human_summary(report: Report, top_n: int = 15) -> None: |
| 638 | """Print a formatted, human-readable summary of *report* to stdout. |
| 639 | |
| 640 | Args: |
| 641 | report: A :class:`Report` produced by :func:`generate_report`. |
| 642 | top_n: How many offenders to display in the top-offenders list. |
| 643 | """ |
| 644 | s = report["summary"] |
| 645 | totals = report["pattern_totals"] |
| 646 | |
| 647 | print("\n" + "=" * 70) |
| 648 | print(" TYPING AUDIT — Violation Report") |
| 649 | print("=" * 70) |
| 650 | print(f" Files scanned: {s['total_files_scanned']}") |
| 651 | print(f" Files importing Any: {s['files_importing_any']}") |
| 652 | print(f" Total violations: {s['total_any_patterns']}") |
| 653 | print(f" Untyped defs: {s['untyped_defs']}") |
| 654 | print() |
| 655 | |
| 656 | has_violations = False |
| 657 | for category, pattern_names in _CATEGORY_ORDER: |
| 658 | category_total = sum(totals.get(p, 0) for p in pattern_names) |
| 659 | if category_total == 0: |
| 660 | continue |
| 661 | has_violations = True |
| 662 | print(f" {category}:") |
| 663 | for p in pattern_names: |
| 664 | count = totals.get(p, 0) |
| 665 | if count > 0: |
| 666 | print(f" {p:38s} {count:5d}") |
| 667 | print() |
| 668 | |
| 669 | if not has_violations: |
| 670 | print(" Pattern breakdown: (none)") |
| 671 | print() |
| 672 | |
| 673 | if report["type_ignore_variants"]: |
| 674 | print(" # type: ignore variants:") |
| 675 | for variant, count in sorted( |
| 676 | report["type_ignore_variants"].items(), |
| 677 | key=operator.itemgetter(1), |
| 678 | reverse=True, |
| 679 | ): |
| 680 | print(f" {variant:44s} {count:5d}") |
| 681 | print() |
| 682 | |
| 683 | print(f" Top {top_n} offenders:") |
| 684 | for entry in report["top_offenders"][:top_n]: |
| 685 | print(f" {entry['total']:4d} {entry['file']}") |
| 686 | print("=" * 70 + "\n") |
| 687 | |
| 688 | |
| 689 | # --------------------------------------------------------------------------- |
| 690 | # CLI |
| 691 | # --------------------------------------------------------------------------- |
| 692 | |
| 693 | |
| 694 | def main() -> None: |
| 695 | """Entry point: parse CLI flags, run the scan, and enforce the ratchet. |
| 696 | |
| 697 | Scans the specified directories (or individual files), prints a human |
| 698 | summary, optionally writes a JSON report, and exits non-zero when either |
| 699 | the pattern violation count exceeds ``--max-any`` or the untyped-def |
| 700 | count exceeds ``--max-untyped``. |
| 701 | """ |
| 702 | parser = argparse.ArgumentParser( |
| 703 | description=( |
| 704 | "Audit typing violations: Any, object, cast, bare collections, " |
| 705 | "Optional/Union (legacy), Callable without signature, untyped " |
| 706 | "varargs, type: ignore, untyped defs, unconstrained TypeVars." |
| 707 | ), |
| 708 | ) |
| 709 | parser.add_argument( |
| 710 | "--dirs", |
| 711 | nargs="+", |
| 712 | default=["musehub/", "tests/"], |
| 713 | help="Directories or individual .py files to scan. Default: musehub/ tests/", |
| 714 | ) |
| 715 | parser.add_argument( |
| 716 | "--json", |
| 717 | type=str, |
| 718 | metavar="PATH", |
| 719 | help="Write the JSON report to PATH.", |
| 720 | ) |
| 721 | parser.add_argument( |
| 722 | "--max-any", |
| 723 | type=int, |
| 724 | default=None, |
| 725 | metavar="N", |
| 726 | help="Exit non-zero if total pattern violations exceed N (ratchet mode).", |
| 727 | ) |
| 728 | parser.add_argument( |
| 729 | "--max-untyped", |
| 730 | type=int, |
| 731 | default=None, |
| 732 | metavar="N", |
| 733 | help="Exit non-zero if total untyped-def count exceeds N (ratchet mode).", |
| 734 | ) |
| 735 | parser.add_argument( |
| 736 | "--top-n", |
| 737 | type=int, |
| 738 | default=15, |
| 739 | metavar="N", |
| 740 | help="Number of offenders to display in the human summary. Default: 15.", |
| 741 | ) |
| 742 | args = parser.parse_args() |
| 743 | |
| 744 | all_results: list[FileResult] = [] |
| 745 | for d in args.dirs: |
| 746 | p = Path(d) |
| 747 | if p.is_file() and p.suffix == ".py": |
| 748 | result = scan_file(p) |
| 749 | if result is not None: |
| 750 | all_results.append(result) |
| 751 | elif p.is_dir(): |
| 752 | all_results.extend(scan_directory(p)) |
| 753 | else: |
| 754 | print(f"WARNING: {d} does not exist, skipping", file=sys.stderr) |
| 755 | |
| 756 | report = generate_report(all_results) |
| 757 | print_human_summary(report, top_n=args.top_n) |
| 758 | |
| 759 | if args.json: |
| 760 | out = Path(args.json) |
| 761 | out.parent.mkdir(parents=True, exist_ok=True) |
| 762 | out.write_text(json.dumps(report, indent=2), encoding="utf-8") |
| 763 | print(f" JSON report written to {args.json}") |
| 764 | |
| 765 | failed = False |
| 766 | |
| 767 | if args.max_any is not None: |
| 768 | total = report["summary"]["total_any_patterns"] |
| 769 | if total > args.max_any: |
| 770 | print( |
| 771 | f"\n❌ RATCHET FAILED (patterns): {total} violations exceed " |
| 772 | f"threshold of {args.max_any}", |
| 773 | file=sys.stderr, |
| 774 | ) |
| 775 | failed = True |
| 776 | else: |
| 777 | print( |
| 778 | f"\n✅ RATCHET OK (patterns): {total} violations within " |
| 779 | f"threshold of {args.max_any}", |
| 780 | ) |
| 781 | |
| 782 | if args.max_untyped is not None: |
| 783 | untyped = report["summary"]["untyped_defs"] |
| 784 | if untyped > args.max_untyped: |
| 785 | print( |
| 786 | f"\n❌ RATCHET FAILED (untyped defs): {untyped} exceed " |
| 787 | f"threshold of {args.max_untyped}", |
| 788 | file=sys.stderr, |
| 789 | ) |
| 790 | failed = True |
| 791 | else: |
| 792 | print( |
| 793 | f"\n✅ RATCHET OK (untyped defs): {untyped} within " |
| 794 | f"threshold of {args.max_untyped}", |
| 795 | ) |
| 796 | |
| 797 | if failed: |
| 798 | sys.exit(1) |
| 799 | |
| 800 | |
| 801 | if __name__ == "__main__": |
| 802 | main() |
File History
1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago