# Copyright (c) Streamlit Inc. (2018-2022) Snowflake Inc. (2022-2026)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import os
from typing import TYPE_CHECKING, TypeAlias

from streamlit import cli_util, config, env_util

if TYPE_CHECKING:
    from collections.abc import Callable

    from streamlit.watcher.event_based_path_watcher import EventBasedPathWatcher
    from streamlit.watcher.polling_path_watcher import PollingPathWatcher


# local_sources_watcher.py caches the return value of
# get_default_path_watcher_class(), so it needs to differentiate between the
# cases where it:
#   1. has yet to call get_default_path_watcher_class()
#   2. has called get_default_path_watcher_class(), which returned that no
#      path watcher should be installed.
# This forces us to define this stub class since the cached value equaling
# None corresponds to case 1 above.
class NoOpPathWatcher:
    def __init__(
        self,
        _path_str: str,
        _on_changed: Callable[[str], None],
        *,  # keyword-only arguments:
        glob_pattern: str | None = None,
        allow_nonexistent: bool = False,
    ) -> None:
        pass


# EventBasedPathWatcher will be a stub and have no functional
# implementation if its import failed (due to missing watchdog module),
# so we can't reference it directly in this type.
PathWatcherType: TypeAlias = (
    type["EventBasedPathWatcher"] | type["PollingPathWatcher"] | type[NoOpPathWatcher]
)


def _is_watchdog_available() -> bool:
    """Check if the watchdog module is installed."""
    try:
        import watchdog  # noqa: F401

        return True
    except ImportError:
        return False


def report_watchdog_availability() -> None:
    if (
        config.get_option("server.fileWatcherType") not in {"poll", "none"}
        and not _is_watchdog_available()
    ):
        msg = "\n  $ xcode-select --install" if env_util.IS_DARWIN else ""

        cli_util.print_to_cli(
            "  For better performance, install the Watchdog module:",
            fg="blue",
            bold=True,
        )
        cli_util.print_to_cli(
            f"""{msg}
  $ pip install watchdog
            """
        )


def _watch_path(
    path: str,
    on_path_changed: Callable[[str], None],
    watcher_type: str | None = None,
    *,  # keyword-only arguments:
    glob_pattern: str | None = None,
    allow_nonexistent: bool = False,
) -> bool:
    """Create a PathWatcher for the given path if we have a viable
    PathWatcher class.

    Parameters
    ----------
    path
        Path to watch.
    on_path_changed
        Function that's called when the path changes.
    watcher_type
        Optional watcher_type string. If None, it will default to the
        'server.fileWatcherType` config option.
    glob_pattern
        Optional glob pattern to use when watching a directory. If set, only
        files matching the pattern will be counted as being created/deleted
        within the watched directory.
    allow_nonexistent
        If True, allow the file or directory at the given path to be
        nonexistent.

    Returns
    -------
    bool
        True if the path is being watched, or False if we have no
        PathWatcher class.
    """
    if watcher_type is None:
        watcher_type = config.get_option("server.fileWatcherType")

    watcher_class = get_path_watcher_class(watcher_type)
    if watcher_class is NoOpPathWatcher:
        return False

    watcher_class(
        path,
        on_path_changed,
        glob_pattern=glob_pattern,
        allow_nonexistent=allow_nonexistent,
    )
    return True


def watch_file(
    path: str,
    on_file_changed: Callable[[str], None],
    watcher_type: str | None = None,
    *,  # keyword-only arguments:
    allow_nonexistent: bool = False,
) -> bool:
    """Watch a file for changes.

    The callback is invoked when the file's content changes (detected via MD5).
    If allow_nonexistent is True, the watcher will also detect when the file
    is created.

    Parameters
    ----------
    path
        Path to the file to watch.
    on_file_changed
        Callback invoked with the file path when changes are detected.
    watcher_type
        Optional watcher type ('watchdog', 'poll', 'auto', or 'none').
    allow_nonexistent
        If True, watch for file creation even if the file doesn't exist yet.
        Note: The file's parent directory must exist for watching to work.
        If the parent directory doesn't exist, the watcher silently skips
        watching (the file can't be created without its parent directory).

    Returns
    -------
    bool
        True if the watcher was successfully created.
    """
    return _watch_path(
        path, on_file_changed, watcher_type, allow_nonexistent=allow_nonexistent
    )


def watch_dir(
    path: str,
    on_dir_changed: Callable[[str], None],
    watcher_type: str | None = None,
    *,  # keyword-only arguments:
    glob_pattern: str | None = None,
    allow_nonexistent: bool = False,
) -> bool:
    """Watch a directory for file changes.

    The callback is invoked for any file activity within the directory,
    including file creation, deletion, and content modifications. The callback
    receives the path of the actual changed file (not the directory path).

    Note: The glob_pattern parameter only affects the initial state detection
    (which files are counted when determining if the directory changed). It does
    NOT filter which file events trigger the callback - all file events in the
    directory will invoke the callback regardless of glob_pattern.

    Parameters
    ----------
    path
        Path to the directory to watch.
    on_dir_changed
        Callback invoked with the changed file path when changes are detected.
    watcher_type
        Optional watcher type ('watchdog', 'poll', 'auto', or 'none').
    glob_pattern
        Glob pattern for initial state detection (e.g., "*.py"). Does not
        filter runtime events.
    allow_nonexistent
        If True, watch for directory creation even if it doesn't exist yet.

    Returns
    -------
    bool
        True if the watcher was successfully created.
    """
    # Add a trailing slash to the path to ensure
    # that its interpreted as a directory.
    path = os.path.join(path, "")

    return _watch_path(
        path,
        on_dir_changed,
        watcher_type,
        glob_pattern=glob_pattern,
        allow_nonexistent=allow_nonexistent,
    )


def get_default_path_watcher_class() -> PathWatcherType:
    """Return the class to use for path changes notifications, based on the
    server.fileWatcherType config option.
    """
    return get_path_watcher_class(config.get_option("server.fileWatcherType"))


def get_path_watcher_class(watcher_type: str) -> PathWatcherType:
    """Return the PathWatcher class that corresponds to the given watcher_type
    string. Acceptable values are 'auto', 'watchdog', 'poll' and 'none'.
    """
    if watcher_type in {"watchdog", "auto"} and _is_watchdog_available():
        # Lazy-import this module to prevent unnecessary imports of the watchdog package.
        from streamlit.watcher.event_based_path_watcher import EventBasedPathWatcher

        return EventBasedPathWatcher
    if watcher_type in {"auto", "poll"}:
        from streamlit.watcher.polling_path_watcher import PollingPathWatcher

        return PollingPathWatcher
    return NoOpPathWatcher