user_code_parser.py

"""Parser for extracting user-added code from generated files."""

import ast

from simple_logger.logger import get_logger

LOGGER = get_logger(name=__name__)


def parse_user_code_from_file(file_path: str) -> tuple[str, str]:
    """
    Extract user-added code and imports from a generated file.

    Args:
        file_path: Path to the generated file

    Returns:
        Tuple of (user_code, user_imports)

    Raises:
        FileNotFoundError: If the file does not exist
        PermissionError: If the file cannot be accessed due to permissions
        UnicodeDecodeError: If the file contains invalid encoding
        Exception: For other unexpected errors during file reading
    """
    try:
        with open(file_path, encoding="utf-8") as fd:
            data = fd.read()
    except FileNotFoundError:
        raise FileNotFoundError(f"File not found: {file_path}. Please ensure the file exists.")
    except PermissionError:
        raise PermissionError(f"Permission denied when accessing file: {file_path}. Check file permissions.")
    except UnicodeDecodeError as e:
        raise UnicodeDecodeError(
            e.encoding,
            e.object,
            e.start,
            e.end,
            f"Failed to decode file {file_path} with UTF-8 encoding. The file may contain invalid characters.",
        )
    except Exception as e:
        raise Exception(f"Unexpected error reading file {file_path}: {type(e).__name__}: {e!s}")

    end_of_generated_code_line = "    # End of generated code"
    user_code: str = ""
    user_imports_list: list[str] = []  # Collect imports in a list instead of concatenating strings

    if end_of_generated_code_line in data:
        _end_of_generated_code_index = data.index(end_of_generated_code_line)
        user_code = data[_end_of_generated_code_index + len(end_of_generated_code_line) :]

        try:
            tree = ast.parse(data)
            imports = [imp for imp in tree.body if isinstance(imp, ast.Import) or isinstance(imp, ast.ImportFrom)]
            splited_data = data.splitlines()
            source_lines_count = len(splited_data)

            # Standard imports that are always generated by the template
            template_imports = {
                "from typing import Any",
                "from ocp_resources.resource import NamespacedResource",
                "from ocp_resources.resource import Resource",
                "from ocp_resources.resource import NamespacedResource, MissingRequiredArgumentError",
                "from ocp_resources.resource import Resource, MissingRequiredArgumentError",
            }

            for _import in imports:
                # Validate line numbers to prevent IndexError
                start_line = _import.lineno - 1  # Convert to 0-based index
                end_line = (_import.end_lineno - 1) if _import.end_lineno else start_line

                # Ensure line numbers are within bounds
                if start_line < 0 or start_line >= source_lines_count:
                    continue
                if end_line < 0 or end_line >= source_lines_count:
                    end_line = start_line

                import_lines = []

                # Extract lines for multi-line imports with proper boundary checks
                if start_line != end_line:
                    for line_idx in range(start_line, min(end_line + 1, source_lines_count)):
                        if 0 <= line_idx < source_lines_count:
                            import_lines.append(splited_data[line_idx])
                else:
                    # Single line import
                    if 0 <= start_line < source_lines_count:
                        import_lines.append(splited_data[start_line])

                import_str = "\n".join(import_lines).strip()

                # Only include imports that are not in the template
                if import_str and import_str not in template_imports:
                    user_imports_list.append(import_str)
        except SyntaxError as e:
            LOGGER.error(f"SyntaxError in file {file_path}: {e}")
            raise  # Re-raise the error so it can be caught by tests

    # Join all collected imports at once
    user_imports = "\n".join(user_imports_list) + "\n" if user_imports_list else ""
    return user_code, user_imports