learningcircuit--local-deep-research/.pre-commit-hooks/fix-exception-logging.py

#!/usr/bin/env python3
"""
Auto-fixer for exception variable leaks and exc_info in production logs.

Handles two patterns:
1. Exception variables in log messages (f"...{e}", "...%s", e)
   → Removes the exception variable from the message
2. exc_info=True on warning/error/critical logs
   → Removes the exc_info keyword argument

Note: this hook is part of the reason we do NOT enforce ``raise X from e``
universally (see ADR-0003). Preserving exception chains via ``from e``
would re-expose the details this hook strips from logs.

Run standalone:
    python .pre-commit-hooks/fix-exception-logging.py src/

Or via the pre-commit hook:
    python .pre-commit-hooks/check-sensitive-logging.py --fix <files>
"""

import ast
import re
import sys
from pathlib import Path
from typing import List, Optional, Set, Tuple


# ---------------------------------------------------------------------------
# AST-based violation finder
# ---------------------------------------------------------------------------


class _ViolationFinder(ast.NodeVisitor):
    """Find exception-var and exc_info violations with precise locations."""

    def __init__(self) -> None:
        self.exc_var_violations: List[Tuple[int, int, str, Set[str]]] = []
        # (start_line, end_line, level, except_var_names)
        self.exc_info_violations: List[Tuple[int, int]] = []
        # (keyword_line, keyword_end_line)
        self._except_var_stack: List[Optional[str]] = []

    def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
        self._except_var_stack.append(node.name)
        self.generic_visit(node)
        self._except_var_stack.pop()

    def visit_Call(self, node: ast.Call) -> None:
        if self._is_logger_call(node):
            level = self._get_level(node)
            if level and level not in {"exception", "debug"}:
                self._check_exc_var(node, level)
            if level in {"warning", "error", "critical"}:
                self._check_exc_info(node)
        self.generic_visit(node)

    @staticmethod
    def _is_logger_call(node: ast.Call) -> bool:
        if isinstance(node.func, ast.Attribute):
            if node.func.attr in {
                "debug",
                "info",
                "warning",
                "error",
                "critical",
                "exception",
            }:
                if isinstance(node.func.value, ast.Name):
                    return node.func.value.id == "logger"
                if isinstance(node.func.value, ast.Attribute):
                    return node.func.value.attr == "logger"
        return False

    @staticmethod
    def _get_level(node: ast.Call) -> Optional[str]:
        if isinstance(node.func, ast.Attribute):
            return node.func.attr
        return None

    def _check_exc_var(self, node: ast.Call, level: str) -> None:
        except_vars = {v for v in self._except_var_stack if v is not None}
        if not except_vars:
            return
        for arg in node.args:
            if self._references(arg, except_vars):
                self.exc_var_violations.append(
                    (
                        node.lineno,
                        node.end_lineno or node.lineno,
                        level,
                        except_vars,
                    )
                )
                return

    def _check_exc_info(self, node: ast.Call) -> None:
        for kw in node.keywords:
            if (
                kw.arg == "exc_info"
                and isinstance(kw.value, ast.Constant)
                and kw.value.value is True
            ):
                self.exc_info_violations.append(
                    (kw.lineno, kw.end_lineno or kw.lineno)
                )

    def _references(self, expr: ast.AST, names: Set[str]) -> bool:
        if isinstance(expr, ast.Name):
            return expr.id in names
        if isinstance(expr, ast.JoinedStr):
            return any(
                isinstance(v, ast.FormattedValue)
                and self._references(v.value, names)
                for v in expr.values
            )
        if isinstance(expr, ast.Call):
            return any(self._references(a, names) for a in expr.args)
        if isinstance(expr, ast.BinOp):
            return self._references(expr.left, names) or self._references(
                expr.right, names
            )
        if isinstance(expr, ast.Tuple):
            return any(self._references(el, names) for el in expr.elts)
        return False


# ---------------------------------------------------------------------------
# Text-based fixers
# ---------------------------------------------------------------------------

# Patterns for exception variable references in f-strings
# Matches: {e}, {e!s}, {e!r}, {exc}, {err}, {error}, {ex}
_FSTR_EXC_RE = re.compile(
    r"""
    [,;:\s\-–—]*           # leading separator/whitespace before {e}
    \{                      # opening brace
    (?:e|ex|exc|err|error)  # common exception variable names
    (?:![sra])?             # optional conversion (!s, !r, !a)
    (?::[^}]*)?             # optional format spec
    \}                      # closing brace
    [,;:\s\-–—.]*           # trailing separator/punctuation after {e}
    """,
    re.VERBOSE,
)

# Matches trailing , e) or , exc) etc. as positional arg in logger call
_PCTARG_EXC_RE = re.compile(r",\s*(?:e|ex|exc|err|error)\s*(?=\))")

# Matches %s (or %r, %d) placeholders that correspond to the exception arg
_PCT_PLACEHOLDER_RE = re.compile(r"[,;:\s\-–—]*%[srd][,;:\s\-–—.]*")


def _fix_exc_var_in_line(line: str, except_vars: Set[str]) -> str:
    """Remove exception variable references from a single line."""
    original = line

    # Build dynamic regex for the specific variable names in this scope
    var_names = "|".join(re.escape(v) for v in except_vars)

    # Fix f-string interpolation: {e}, {exc}, etc.
    fstr_re = re.compile(
        r"[,;:\s\-–—]*"
        r"\{"
        rf"(?:{var_names})"
        r"(?:![sra])?"
        r"(?::[^}]*)?"
        r"\}"
        r"[,;:\s\-–—.]*",
    )
    line = fstr_re.sub("", line)

    # If we removed the only content of an f-string, convert to regular string
    # f"" → ""
    line = re.sub(r'\bf("")', r"\1", line)
    line = re.sub(r"\bf('')", r"\1", line)

    # Fix %-style positional arg: , e) → )
    pctarg_re = re.compile(rf",\s*(?:{var_names})\s*(?=\))")
    line = pctarg_re.sub("", line)

    # If we removed a positional arg, also clean up the %s in the format string
    if line != original and "%s" in line:
        # Only remove trailing %s (the one that matched the exception var)
        # This is conservative — only handles the simple single-%s case
        line = re.sub(r"[,;:\s\-–—]*%s[,;:\s\-–—.]*(?=\")", "", line, count=1)

    return line


def _fix_exc_info_in_line(line: str) -> str:
    """Remove exc_info=True from a line."""
    # Handle: , exc_info=True) or , exc_info=True,  or (exc_info=True)
    line = re.sub(r",\s*exc_info=True", "", line)
    line = re.sub(r"exc_info=True,\s*", "", line)
    line = re.sub(r"exc_info=True", "", line)
    return line


def _remove_unused_as_clause(lines: List[str], except_vars: Set[str]) -> None:
    """Remove 'as e' from except clauses when e is no longer used."""
    for i, line in enumerate(lines):
        for var in except_vars:
            # Match: except SomeError as e:  or  except Exception as e:
            pattern = re.compile(
                rf"(\s*except\s+\S+(?:\s*,\s*\S+)*)\s+as\s+{re.escape(var)}\s*:"
            )
            m = pattern.match(line)
            if m:
                # Check if the variable is still used anywhere in nearby lines
                # (simple heuristic: check next 20 lines in the except block)
                still_used = False
                for j in range(i + 1, min(i + 20, len(lines))):
                    if re.search(rf"\b{re.escape(var)}\b", lines[j]):
                        still_used = True
                        break
                    # Stop at next except/def/class/return at same or lesser indent
                    if re.match(r"\S", lines[j]):
                        break
                if not still_used:
                    lines[i] = pattern.sub(r"\1:", line)


# ---------------------------------------------------------------------------
# File-level fix
# ---------------------------------------------------------------------------


def fix_file(filepath: str) -> Tuple[bool, int]:
    """Fix a file in-place. Returns (changed, fix_count)."""
    path = Path(filepath)
    try:
        source = path.read_text(encoding="utf-8")
    except Exception:
        return False, 0

    try:
        tree = ast.parse(source, filename=filepath)
    except SyntaxError:
        return False, 0

    finder = _ViolationFinder()
    finder.visit(tree)

    if not finder.exc_var_violations and not finder.exc_info_violations:
        return False, 0

    lines = source.splitlines(keepends=True)
    fix_count = 0

    # Collect all except variable names used across violations in this file
    all_except_vars: Set[str] = set()

    # Fix exception variable violations
    for start_line, end_line, _level, except_vars in finder.exc_var_violations:
        all_except_vars |= except_vars
        for i in range(start_line - 1, min(end_line, len(lines))):
            fixed = _fix_exc_var_in_line(lines[i], except_vars)
            if fixed != lines[i]:
                lines[i] = fixed
                fix_count += 1

    # Fix exc_info violations
    for start_line, end_line in finder.exc_info_violations:
        for i in range(start_line - 1, min(end_line, len(lines))):
            fixed = _fix_exc_info_in_line(lines[i])
            if fixed != lines[i]:
                lines[i] = fixed
                fix_count += 1

    # Clean up unused 'as e' clauses
    if all_except_vars:
        _remove_unused_as_clause(lines, all_except_vars)

    if fix_count > 0:
        path.write_text("".join(lines), encoding="utf-8")
        return True, fix_count

    return False, 0


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main() -> int:
    paths: List[str] = []
    for arg in sys.argv[1:]:
        p = Path(arg)
        if p.is_dir():
            paths.extend(str(f) for f in p.rglob("*.py"))
        elif p.suffix == ".py":
            paths.append(str(p))

    if not paths:
        print("Usage: fix-exception-logging.py <file_or_dir> ...")
        return 1

    total_files = 0
    total_fixes = 0
    for filepath in sorted(paths):
        changed, fixes = fix_file(filepath)
        if changed:
            total_files += 1
            total_fixes += fixes
            print(f"  Fixed {filepath} ({fixes} changes)")

    if total_files:
        print(f"\nFixed {total_fixes} issues in {total_files} files")
    else:
        print("No issues to fix")
    return 0


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