reflex_copy/reflex/compiler/compiler.py

"""Compiler for the reflex apps."""

from __future__ import annotations

from collections.abc import Iterable, Sequence
from datetime import datetime
from inspect import getmodule
from pathlib import Path
from typing import TYPE_CHECKING

from reflex import constants
from reflex.compiler import templates, utils
from reflex.components.base.fragment import Fragment
from reflex.components.component import (
    BaseComponent,
    Component,
    ComponentStyle,
    CustomComponent,
    StatefulComponent,
)
from reflex.config import environment, get_config
from reflex.constants.compiler import PageNames
from reflex.state import BaseState
from reflex.style import SYSTEM_COLOR_MODE
from reflex.utils import console, path_ops
from reflex.utils.exceptions import ReflexError
from reflex.utils.exec import is_prod_mode
from reflex.utils.imports import ImportVar
from reflex.utils.prerequisites import get_web_dir
from reflex.vars.base import LiteralVar, Var


def _compile_document_root(root: Component) -> str:
    """Compile the document root.

    Args:
        root: The document root to compile.

    Returns:
        The compiled document root.
    """
    return templates.DOCUMENT_ROOT.render(
        imports=utils.compile_imports(root._get_all_imports()),
        document=root.render(),
    )


def _normalize_library_name(lib: str) -> str:
    """Normalize the library name.

    Args:
        lib: The library name to normalize.

    Returns:
        The normalized library name.
    """
    if lib == "react":
        return "React"
    return lib.replace("@", "").replace("/", "_").replace("-", "_")


def _compile_app(app_root: Component) -> str:
    """Compile the app template component.

    Args:
        app_root: The app root to compile.

    Returns:
        The compiled app.
    """
    from reflex.components.dynamic import bundled_libraries

    window_libraries = [
        (_normalize_library_name(name), name) for name in bundled_libraries
    ] + [
        ("utils_context", f"$/{constants.Dirs.UTILS}/context"),
        ("utils_state", f"$/{constants.Dirs.UTILS}/state"),
    ]

    return templates.APP_ROOT.render(
        imports=utils.compile_imports(app_root._get_all_imports()),
        custom_codes=app_root._get_all_custom_code(),
        hooks=app_root._get_all_hooks(),
        window_libraries=window_libraries,
        render=app_root.render(),
        dynamic_imports=app_root._get_all_dynamic_imports(),
    )


def _compile_theme(theme: str) -> str:
    """Compile the theme.

    Args:
        theme: The theme to compile.

    Returns:
        The compiled theme.
    """
    return templates.THEME.render(theme=theme)


def _compile_contexts(state: type[BaseState] | None, theme: Component | None) -> str:
    """Compile the initial state and contexts.

    Args:
        state: The app state.
        theme: The top-level app theme.

    Returns:
        The compiled context file.
    """
    appearance = getattr(theme, "appearance", None)
    if appearance is None or str(LiteralVar.create(appearance)) == '"inherit"':
        appearance = LiteralVar.create(SYSTEM_COLOR_MODE)

    last_compiled_time = str(datetime.now())
    return (
        templates.CONTEXT.render(
            initial_state=utils.compile_state(state),
            state_name=state.get_name(),
            client_storage=utils.compile_client_storage(state),
            is_dev_mode=not is_prod_mode(),
            last_compiled_time=last_compiled_time,
            default_color_mode=appearance,
        )
        if state
        else templates.CONTEXT.render(
            is_dev_mode=not is_prod_mode(),
            default_color_mode=appearance,
            last_compiled_time=last_compiled_time,
        )
    )


def _compile_page(
    component: BaseComponent,
    state: type[BaseState] | None,
) -> str:
    """Compile the component given the app state.

    Args:
        component: The component to compile.
        state: The app state.

    Returns:
        The compiled component.
    """
    imports = component._get_all_imports()
    imports = utils.compile_imports(imports)

    # Compile the code to render the component.
    kwargs = {"state_name": state.get_name()} if state is not None else {}

    return templates.PAGE.render(
        imports=imports,
        dynamic_imports=component._get_all_dynamic_imports(),
        custom_codes=component._get_all_custom_code(),
        hooks=component._get_all_hooks(),
        render=component.render(),
        **kwargs,
    )


def compile_root_stylesheet(stylesheets: list[str]) -> tuple[str, str]:
    """Compile the root stylesheet.

    Args:
        stylesheets: The stylesheets to include in the root stylesheet.

    Returns:
        The path and code of the compiled root stylesheet.
    """
    output_path = utils.get_root_stylesheet_path()

    code = _compile_root_stylesheet(stylesheets)

    return output_path, code


def _validate_stylesheet(stylesheet_full_path: Path, assets_app_path: Path) -> None:
    """Validate the stylesheet.

    Args:
        stylesheet_full_path: The stylesheet to validate.
        assets_app_path: The path to the assets directory.

    Raises:
        ValueError: If the stylesheet is not supported.
        FileNotFoundError: If the stylesheet is not found.
    """
    suffix = stylesheet_full_path.suffix[1:] if stylesheet_full_path.suffix else ""
    if suffix not in constants.Reflex.STYLESHEETS_SUPPORTED:
        raise ValueError(f"Stylesheet file {stylesheet_full_path} is not supported.")
    if not stylesheet_full_path.absolute().is_relative_to(assets_app_path.absolute()):
        raise FileNotFoundError(
            f"Cannot include stylesheets from outside the assets directory: {stylesheet_full_path}"
        )
    if not stylesheet_full_path.name:
        raise ValueError(
            f"Stylesheet file name cannot be empty: {stylesheet_full_path}"
        )
    if (
        len(
            stylesheet_full_path.absolute()
            .relative_to(assets_app_path.absolute())
            .parts
        )
        == 1
        and stylesheet_full_path.stem == PageNames.STYLESHEET_ROOT
    ):
        raise ValueError(
            f"Stylesheet file name cannot be '{PageNames.STYLESHEET_ROOT}': {stylesheet_full_path}"
        )


def _compile_root_stylesheet(stylesheets: list[str]) -> str:
    """Compile the root stylesheet.

    Args:
        stylesheets: The stylesheets to include in the root stylesheet.

    Returns:
        The compiled root stylesheet.

    Raises:
        FileNotFoundError: If a specified stylesheet in assets directory does not exist.
    """
    # Add stylesheets from plugins.
    sheets = [
        sheet
        for plugin in get_config().plugins
        for sheet in plugin.get_stylesheet_paths()
    ] + ["@radix-ui/themes/styles.css"]

    failed_to_import_sass = False
    assets_app_path = Path.cwd() / constants.Dirs.APP_ASSETS

    stylesheets_files: list[Path] = []
    stylesheets_urls = []

    for stylesheet in stylesheets:
        if not utils.is_valid_url(stylesheet):
            # check if stylesheet provided exists.
            stylesheet_full_path = assets_app_path / stylesheet.strip("/")

            if not stylesheet_full_path.exists():
                raise FileNotFoundError(
                    f"The stylesheet file {stylesheet_full_path} does not exist."
                )

            if stylesheet_full_path.is_dir():
                all_files = (
                    file
                    for ext in constants.Reflex.STYLESHEETS_SUPPORTED
                    for file in stylesheet_full_path.rglob("*." + ext)
                )
                for file in all_files:
                    if file.is_dir():
                        continue
                    # Validate the stylesheet.
                    _validate_stylesheet(file, assets_app_path)
                    stylesheets_files.append(file)

            else:
                # Validate the stylesheet.
                _validate_stylesheet(stylesheet_full_path, assets_app_path)
                stylesheets_files.append(stylesheet_full_path)
        else:
            stylesheets_urls.append(stylesheet)

    sheets.extend(dict.fromkeys(stylesheets_urls))

    for stylesheet in stylesheets_files:
        target_path = stylesheet.relative_to(assets_app_path).with_suffix(".css")
        target = get_web_dir() / constants.Dirs.STYLES / target_path

        target.parent.mkdir(parents=True, exist_ok=True)

        if stylesheet.suffix == ".css":
            path_ops.cp(src=stylesheet, dest=target, overwrite=True)
        else:
            try:
                from sass import compile as sass_compile

                target.write_text(
                    data=sass_compile(
                        filename=str(stylesheet),
                        output_style="compressed",
                    ),
                    encoding="utf8",
                )
            except ImportError:
                failed_to_import_sass = True

        str_target_path = "./" + str(target_path)
        sheets.append(str_target_path) if str_target_path not in sheets else None

    if failed_to_import_sass:
        console.error(
            'The `libsass` package is required to compile sass/scss stylesheet files. Run `pip install "libsass>=0.23.0"`.'
        )

    return templates.STYLE.render(stylesheets=sheets)


def _compile_component(component: Component | StatefulComponent) -> str:
    """Compile a single component.

    Args:
        component: The component to compile.

    Returns:
        The compiled component.
    """
    return templates.COMPONENT.render(component=component)


def _compile_components(
    components: set[CustomComponent],
) -> tuple[str, dict[str, list[ImportVar]]]:
    """Compile the components.

    Args:
        components: The components to compile.

    Returns:
        The compiled components.
    """
    imports = {
        "react": [ImportVar(tag="memo")],
        f"$/{constants.Dirs.STATE_PATH}": [ImportVar(tag="E"), ImportVar(tag="isTrue")],
    }
    component_renders = []

    # Compile each component.
    for component in components:
        component_render, component_imports = utils.compile_custom_component(component)
        component_renders.append(component_render)
        imports = utils.merge_imports(imports, component_imports)

    dynamic_imports = {
        comp_import: None
        for comp_render in component_renders
        if "dynamic_imports" in comp_render
        for comp_import in comp_render["dynamic_imports"]
    }

    custom_codes = {
        comp_custom_code: None
        for comp_render in component_renders
        for comp_custom_code in comp_render.get("custom_code", [])
    }

    # Compile the components page.
    return (
        templates.COMPONENTS.render(
            imports=utils.compile_imports(imports),
            components=component_renders,
            dynamic_imports=dynamic_imports,
            custom_codes=custom_codes,
        ),
        imports,
    )


def _compile_stateful_components(
    page_components: list[BaseComponent],
) -> str:
    """Walk the page components and extract shared stateful components.

    Any StatefulComponent that is shared by more than one page will be rendered
    to a separate module and marked rendered_as_shared so subsequent
    renderings will import the component from the shared module instead of
    directly including the code for it.

    Args:
        page_components: The Components or StatefulComponents to compile.

    Returns:
        The rendered stateful components code.
    """
    all_import_dicts = []
    rendered_components = {}

    def get_shared_components_recursive(component: BaseComponent):
        """Get the shared components for a component and its children.

        A shared component is a StatefulComponent that appears in 2 or more
        pages and is a candidate for writing to a common file and importing
        into each page where it is used.

        Args:
            component: The component to collect shared StatefulComponents for.
        """
        for child in component.children:
            # Depth-first traversal.
            get_shared_components_recursive(child)

        # When the component is referenced by more than one page, render it
        # to be included in the STATEFUL_COMPONENTS module.
        # Skip this step in dev mode, thereby avoiding potential hot reload errors for larger apps
        if (
            isinstance(component, StatefulComponent)
            and component.references > 1
            and is_prod_mode()
        ):
            # Reset this flag to render the actual component.
            component.rendered_as_shared = False

            # Include dynamic imports in the shared component.
            if dynamic_imports := component._get_all_dynamic_imports():
                rendered_components.update(dict.fromkeys(dynamic_imports))

            # Include custom code in the shared component.
            rendered_components.update(
                dict.fromkeys(component._get_all_custom_code()),
            )

            # Include all imports in the shared component.
            all_import_dicts.append(component._get_all_imports())

            # Indicate that this component now imports from the shared file.
            component.rendered_as_shared = True

    for page_component in page_components:
        get_shared_components_recursive(page_component)

    # Don't import from the file that we're about to create.
    all_imports = utils.merge_imports(*all_import_dicts)
    all_imports.pop(
        f"$/{constants.Dirs.UTILS}/{constants.PageNames.STATEFUL_COMPONENTS}", None
    )

    return templates.STATEFUL_COMPONENTS.render(
        imports=utils.compile_imports(all_imports),
        memoized_code="\n".join(rendered_components),
    )


def compile_document_root(
    head_components: list[Component],
    html_lang: str | None = None,
    html_custom_attrs: dict[str, Var | str] | None = None,
) -> tuple[str, str]:
    """Compile the document root.

    Args:
        head_components: The components to include in the head.
        html_lang: The language of the document, will be added to the html root element.
        html_custom_attrs: custom attributes added to the html root element.

    Returns:
        The path and code of the compiled document root.
    """
    # Get the path for the output file.
    output_path = utils.get_page_path(constants.PageNames.DOCUMENT_ROOT)

    # Create the document root.
    document_root = utils.create_document_root(
        head_components, html_lang=html_lang, html_custom_attrs=html_custom_attrs
    )

    # Compile the document root.
    code = _compile_document_root(document_root)
    return output_path, code


def compile_app(app_root: Component) -> tuple[str, str]:
    """Compile the app root.

    Args:
        app_root: The app root component to compile.

    Returns:
        The path and code of the compiled app wrapper.
    """
    # Get the path for the output file.
    output_path = utils.get_page_path(constants.PageNames.APP_ROOT)

    # Compile the document root.
    code = _compile_app(app_root)
    return output_path, code


def compile_theme(style: ComponentStyle) -> tuple[str, str]:
    """Compile the theme.

    Args:
        style: The style to compile.

    Returns:
        The path and code of the compiled theme.
    """
    output_path = utils.get_theme_path()

    # Create the theme.
    theme = utils.create_theme(style)

    # Compile the theme.
    code = _compile_theme(str(LiteralVar.create(theme)))
    return output_path, code


def compile_contexts(
    state: type[BaseState] | None,
    theme: Component | None,
) -> tuple[str, str]:
    """Compile the initial state / context.

    Args:
        state: The app state.
        theme: The top-level app theme.

    Returns:
        The path and code of the compiled context.
    """
    # Get the path for the output file.
    output_path = utils.get_context_path()

    return output_path, _compile_contexts(state, theme)


def compile_page(
    path: str, component: BaseComponent, state: type[BaseState] | None
) -> tuple[str, str]:
    """Compile a single page.

    Args:
        path: The path to compile the page to.
        component: The component to compile.
        state: The app state.

    Returns:
        The path and code of the compiled page.
    """
    # Get the path for the output file.
    output_path = utils.get_page_path(path)

    # Add the style to the component.
    code = _compile_page(component, state)
    return output_path, code


def compile_components(
    components: set[CustomComponent],
) -> tuple[str, str, dict[str, list[ImportVar]]]:
    """Compile the custom components.

    Args:
        components: The custom components to compile.

    Returns:
        The path and code of the compiled components.
    """
    # Get the path for the output file.
    output_path = utils.get_components_path()

    # Compile the components.
    code, imports = _compile_components(components)
    return output_path, code, imports


def compile_stateful_components(
    pages: Iterable[Component],
) -> tuple[str, str, list[BaseComponent]]:
    """Separately compile components that depend on State vars.

    StatefulComponents are compiled as their own component functions with their own
    useContext declarations, which allows page components to be stateless and avoid
    re-rendering along with parts of the page that actually depend on state.

    Args:
        pages: The pages to extract stateful components from.

    Returns:
        The path and code of the compiled stateful components.
    """
    output_path = utils.get_stateful_components_path()

    # Compile the stateful components.
    page_components = [StatefulComponent.compile_from(page) or page for page in pages]
    code = _compile_stateful_components(page_components)
    return output_path, code, page_components


def purge_web_pages_dir():
    """Empty out .web/pages directory."""
    if not is_prod_mode() and environment.REFLEX_PERSIST_WEB_DIR.get():
        # Skip purging the web directory in dev mode if REFLEX_PERSIST_WEB_DIR is set.
        return

    # Empty out the web pages directory.
    utils.empty_dir(get_web_dir() / constants.Dirs.PAGES, keep_files=["_app.js"])


if TYPE_CHECKING:
    from reflex.app import ComponentCallable, UnevaluatedPage


def _into_component_once(
    component: Component
    | ComponentCallable
    | tuple[Component, ...]
    | str
    | Var
    | int
    | float,
) -> Component | None:
    """Convert a component to a Component.

    Args:
        component: The component to convert.

    Returns:
        The converted component.
    """
    if isinstance(component, Component):
        return component
    if isinstance(component, (Var, int, float, str)):
        return Fragment.create(component)
    if isinstance(component, Sequence):
        return Fragment.create(*component)
    return None


def readable_name_from_component(
    component: Component | ComponentCallable,
) -> str | None:
    """Get the readable name of a component.

    Args:
        component: The component to get the name of.

    Returns:
        The readable name of the component.
    """
    if isinstance(component, Component):
        return type(component).__name__
    if isinstance(component, (Var, int, float, str)):
        return str(component)
    if isinstance(component, Sequence):
        return ", ".join(str(c) for c in component)
    if callable(component):
        module_name = getattr(component, "__module__", None)
        if module_name is not None:
            module = getmodule(component)
            if module is not None:
                module_name = module.__name__
        if module_name is not None:
            return f"{module_name}.{component.__name__}"
        return component.__name__
    return None


def into_component(component: Component | ComponentCallable) -> Component:
    """Convert a component to a Component.

    Args:
        component: The component to convert.

    Returns:
        The converted component.

    Raises:
        TypeError: If the component is not a Component.

    # noqa: DAR401
    """
    if (converted := _into_component_once(component)) is not None:
        return converted
    try:
        if (
            callable(component)
            and (converted := _into_component_once(component())) is not None
        ):
            return converted
    except KeyError as e:
        if isinstance(e, ReflexError):
            raise
        key = e.args[0] if e.args else None
        if key is not None and isinstance(key, Var):
            raise TypeError(
                "Cannot access a primitive map with a Var. Consider calling rx.Var.create() on the map."
            ).with_traceback(e.__traceback__) from None
        raise
    except TypeError as e:
        if isinstance(e, ReflexError):
            raise
        message = e.args[0] if e.args else None
        if message and isinstance(message, str):
            if message.endswith("has no len()") and (
                "ArrayCastedVar" in message
                or "ObjectCastedVar" in message
                or "StringCastedVar" in message
            ):
                raise TypeError(
                    "Cannot pass a Var to a built-in function. Consider using .length() for accessing the length of an iterable Var."
                ).with_traceback(e.__traceback__) from None
            if message.endswith(
                "indices must be integers or slices, not NumberCastedVar"
            ) or message.endswith(
                "indices must be integers or slices, not BooleanCastedVar"
            ):
                raise TypeError(
                    "Cannot index into a primitive sequence with a Var. Consider calling rx.Var.create() on the sequence."
                ).with_traceback(e.__traceback__) from None
        if "CastedVar" in str(e):
            raise TypeError(
                "Cannot pass a Var to a built-in function. Consider moving the operation to the backend, using existing Var operations, or defining a custom Var operation."
            ).with_traceback(e.__traceback__) from None
        raise

    raise TypeError(f"Expected a Component, got {type(component)}")


def compile_unevaluated_page(
    route: str,
    page: UnevaluatedPage,
    state: type[BaseState] | None = None,
    style: ComponentStyle | None = None,
    theme: Component | None = None,
) -> tuple[Component, bool]:
    """Compiles an uncompiled page into a component and adds meta information.

    Args:
        route: The route of the page.
        page: The uncompiled page object.
        state: The state of the app.
        style: The style of the page.
        theme: The theme of the page.

    Returns:
        The compiled component and whether state should be enabled.
    """
    # Generate the component if it is a callable.
    component = into_component(page.component)

    component._add_style_recursive(style or {}, theme)

    enable_state = False
    # Ensure state is enabled if this page uses state.
    if state is None:
        if page.on_load or component._has_stateful_event_triggers():
            enable_state = True
        else:
            for var in component._get_vars(include_children=True):
                var_data = var._get_all_var_data()
                if not var_data:
                    continue
                if not var_data.state:
                    continue
                enable_state = True
                break

    from reflex.app import OverlayFragment
    from reflex.utils.format import make_default_page_title

    component = OverlayFragment.create(component)

    meta_args = {
        "title": (
            page.title
            if page.title is not None
            else make_default_page_title(get_config().app_name, route)
        ),
        "image": page.image,
        "meta": page.meta,
    }

    if page.description is not None:
        meta_args["description"] = page.description

    # Add meta information to the component.
    utils.add_meta(
        component,
        **meta_args,
    )

    return component, enable_state


class ExecutorSafeFunctions:
    """Helper class to allow parallelisation of parts of the compilation process.

    This class (and its class attributes) are available at global scope.

    In a multiprocessing context (like when using a ProcessPoolExecutor), the content of this
    global class is logically replicated to any FORKED process.

    How it works:
    * Before the child process is forked, ensure that we stash any input data required by any future
      function call in the child process.
    * After the child process is forked, the child process will have a copy of the global class, which
      includes the previously stashed input data.
    * Any task submitted to the child process simply needs a way to communicate which input data the
      requested function call requires.

    Why do we need this? Passing input data directly to child process often not possible because the input data is not picklable.
    The mechanic described here removes the need to pickle the input data at all.

    Limitations:
    * This can never support returning unpicklable OUTPUT data.
    * Any object mutations done by the child process will not propagate back to the parent process (fork goes one way!).

    """

    COMPONENTS: dict[str, BaseComponent] = {}
    UNCOMPILED_PAGES: dict[str, UnevaluatedPage] = {}
    STATE: type[BaseState] | None = None

    @classmethod
    def compile_page(cls, route: str) -> tuple[str, str]:
        """Compile a page.

        Args:
            route: The route of the page to compile.

        Returns:
            The path and code of the compiled page.
        """
        return compile_page(route, cls.COMPONENTS[route], cls.STATE)

    @classmethod
    def compile_unevaluated_page(
        cls,
        route: str,
        style: ComponentStyle,
        theme: Component | None,
    ) -> tuple[str, Component, tuple[str, str]]:
        """Compile an unevaluated page.

        Args:
            route: The route of the page to compile.
            style: The style of the page.
            theme: The theme of the page.

        Returns:
            The route, compiled component, and compiled page.
        """
        component, enable_state = compile_unevaluated_page(
            route, cls.UNCOMPILED_PAGES[route], cls.STATE, style, theme
        )
        return route, component, compile_page(route, component, cls.STATE)

    @classmethod
    def compile_theme(cls, style: ComponentStyle | None) -> tuple[str, str]:
        """Compile the theme.

        Args:
            style: The style to compile.

        Returns:
            The path and code of the compiled theme.

        Raises:
            ValueError: If the style is not set.
        """
        if style is None:
            raise ValueError("STYLE should be set")
        return compile_theme(style)