scripts.auto_fix

Reference › scripts › auto_fix

An auto-fixer for the QA Volk pipeline that repairs common defects in LLM-generated Python. A human or the pipeline runs it; no module or test imports it.

Invocation: main takes any number of paths (files or directories, default the current directory), -v/--verbose, --list-fixers to print the available fixers with their docstrings, and --disable NAME (repeatable) to switch fixers off. It collects the Python files with find_python_files, runs every enabled fixer over each file and rewrites the file in place when the content changed.

Each fixer is a subclass of AutoFixer, whose apply_to_file reads a file and calls the subclass's _fix_content. RelativeImportFixer rewrites relative imports to absolute ones, LineLengthFixer wraps over-long lines, ImportOrderFixer reorders imports, RedisHostFixer replaces hard-coded Redis host values, and MissingDatetimeFixer adds a missing datetime import. The absolute-import and line-length fixers follow the workspace coding standards.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

AutoFixer

Line 14

Base class for auto-fixers.

RelativeImportFixer

Bases: AutoFixer · Line 53

Rewrites relative imports to absolute imports.

LineLengthFixer

Bases: AutoFixer · Line 78

Splits lines exceeding 88 characters (project's line length limit).

ImportOrderFixer

Bases: AutoFixer · Line 111

Reorders imports: stdlib, third-party, local (separated by blank lines).

RedisHostFixer

Bases: AutoFixer · Line 120

Replaces hardcoded "localhost" in Redis constructors with environment variable.

MissingDatetimeFixer

Bases: AutoFixer · Line 136

Adds missing datetime import when datetime is used bare.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/auto_fix.py

#!/usr/bin/env python3
"""
Auto-fixer script for orchestrata-ultimate QA Volk pipeline.
Implements 13 deterministic fixers for common LLM-generated defects.

Based on SPEC-031 QA Volk requirements.
"""

import argparse
import re
from pathlib import Path


class AutoFixer:
    """Base class for auto-fixers."""

    def __init__(self, verbose: bool = False):
        self.verbose = verbose
        self.files_fixed = 0
        self.errors = 0

    def apply_to_file(self, file_path: Path) -> bool:
        """Apply fixer to a single file. Returns True if file was modified."""
        try:
            content = file_path.read_text(encoding="utf-8")
        except (UnicodeDecodeError, OSError) as e:
            if self.verbose:
                print(f"  ⚠️  Cannot read {file_path}: {e}")
            self.errors += 1
            return False

        original = content
        modified = self._fix_content(content, file_path)

        if modified != original:
            try:
                file_path.write_text(modified, encoding="utf-8")
                self.files_fixed += 1
                if self.verbose:
                    print(f"  ✅ Fixed {file_path}")
                return True
            except OSError as e:
                if self.verbose:
                    print(f"  ❌ Cannot write {file_path}: {e}")
                self.errors += 1
        return False

    def _fix_content(self, content: str, file_path: Path) -> str:
        """Override in subclasses to apply fixes."""
        return content


class RelativeImportFixer(AutoFixer):
    """Rewrites relative imports to absolute imports."""

    def _fix_content(self, content: str, file_path: Path) -> str:
        # Simple regex-based fix for demonstration
        # In production, use AST for accurate transformation
        lines = content.splitlines()
        modified = False
        for i, line in enumerate(lines):
            # Match relative imports like "from orchestrata.module import" or "from orchestrata..module import"
            if line.strip().startswith("from ."):
                # Replace with absolute import based on project structure
                # This is a simplified example
                new_line = line.replace("from .", "from orchestrata_ultimate.")
                lines[i] = new_line
                modified = True
                if self.verbose:
                    print(
                        f"    Converted relative import: {line.strip()} -> {new_line.strip()}"
                    )
        if modified:
            return "\n".join(lines)
        return content


class LineLengthFixer(AutoFixer):
    """Splits lines exceeding 88 characters (project's line length limit)."""

    def _fix_content(self, content: str, file_path: Path) -> str:
        lines = content.splitlines()
        modified = False

        for i, line in enumerate(lines):
            if len(line) > 88 and not line.strip().startswith("#"):
                # Simple splitting at last comma before limit
                # In production, use more sophisticated algorithm
                if "," in line:
                    split_idx = line.rfind(",", 0, 88)
                    if split_idx > 0:
                        # NOTE: mutate lines[i] in place and read the split
                        # result back from `lines`, not the stale loop
                        # variable `line` — a prior version rebuilt the
                        # output from `line` (appended to a separate
                        # modified_lines list), so the split was computed
                        # but silently discarded and no file was ever
                        # actually fixed.
                        lines[i] = (
                            line[: split_idx + 1]
                            + "\n    "
                            + line[split_idx + 1 :].lstrip()
                        )
                        modified = True

        if modified:
            return "\n".join(lines)
        return content


class ImportOrderFixer(AutoFixer):
    """Reorders imports: stdlib, third-party, local (separated by blank lines)."""

    def _fix_content(self, content: str, file_path: Path) -> str:
        # Parse imports and reorder
        # This is a placeholder implementation
        return content


class RedisHostFixer(AutoFixer):
    """Replaces hardcoded "localhost" in Redis constructors with environment variable."""

    def _fix_content(self, content: str, file_path: Path) -> str:
        # Pattern 1: redis.Redis(host='localhost'..., decode_responses=True)
        new_content = re.sub(
            r"(redis\.(?:asyncio\.)?Redis\()host\s*=\s*['\"](?:localhost|127\.0\.0\.1)['\"]",
            r"\1host=os.environ.get('REDIS_HOST', 'localhost')",
            content,
        )
        # Add os import if needed
        if new_content != content and "import os" not in content:
            new_content = "import os\n" + new_content
        return new_content


class MissingDatetimeFixer(AutoFixer):
    """Adds missing datetime import when datetime is used bare."""

    def _fix_content(self, content: str, file_path: Path) -> str:
        # Check for usage of datetime, timedelta without import
        if (
            "datetime" in content
            and "from datetime import" not in content
            and "import datetime" not in content
        ):
            # Add import at top of file
            lines = content.splitlines()
            import_added = False
            for i, line in enumerate(lines):
                if line.strip().startswith("import ") or line.strip().startswith(
                    "from "
                ):
                    # Insert after last import
                    continue
                elif line.strip() and not line.strip().startswith("#"):
                    # Insert before first non-import line
                    lines.insert(i, "from datetime import datetime, timedelta")
                    import_added = True
                    break
            if import_added:
                return "\n".join(lines)
        return content


# Registry of all fixers
FIXERS = [
    ("relative_imports", RelativeImportFixer),
    ("line_length", LineLengthFixer),
    ("import_order", ImportOrderFixer),
    ("redis_host", RedisHostFixer),
    ("missing_datetime", MissingDatetimeFixer),
]


def find_python_files(root_path: Path) -> list[Path]:
    """Find all Python files in the directory."""
    return list(root_path.rglob("*.py"))


def main():
    parser = argparse.ArgumentParser(description="Auto-fixer for LLM-generated defects")
    parser.add_argument(
        "paths",
        nargs="*",
        default=["."],
        help="Paths to fix (default: current directory)",
    )
    parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
    parser.add_argument(
        "--list-fixers", action="store_true", help="List available fixers"
    )
    parser.add_argument(
        "--disable", action="append", default=[], help="Disable specific fixers"
    )
    args = parser.parse_args()

    if args.list_fixers:
        print("Available fixers:")
        for name, cls in FIXERS:
            print(f"  - {name}: {cls.__doc__}")
        return

    # Collect files
    files = []
    for path in args.paths:
        p = Path(path)
        if p.is_file() and p.suffix == ".py":
            files.append(p)
        elif p.is_dir():
            files.extend(find_python_files(p))

    if not files:
        print("No Python files found.")
        return

    print(f"🔧 Running auto-fixers on {len(files)} files...")

    # Initialize enabled fixers
    enabled_fixers = []
    for name, cls in FIXERS:
        if name not in args.disable:
            enabled_fixers.append(cls(verbose=args.verbose))

    if not enabled_fixers:
        print("⚠️  No fixers enabled.")
        return

    # Apply fixers
    total_fixed = 0
    for file_path in files:
        if args.verbose:
            print(f"Processing {file_path}")
        for fixer in enabled_fixers:
            fixer.apply_to_file(file_path)
        total_fixed += sum(fixer.files_fixed for fixer in enabled_fixers)

    # Print summary
    print("\n📊 Auto-fix summary:")
    print(f"  Files processed: {len(files)}")
    print(f"  Files modified: {total_fixed}")
    print(f"  Errors: {sum(fixer.errors for fixer in enabled_fixers)}")

    for fixer in enabled_fixers:
        if fixer.files_fixed > 0:
            print(f"  {fixer.__class__.__name__}: {fixer.files_fixed} files fixed")


if __name__ == "__main__":
    main()