# Copyright (c) Streamlit Inc. (2018-2022) Snowflake Inc. (2022-2025)
#
# 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

import streamlit.watcher
from streamlit import cli_util, config, env_util
from streamlit.watcher.polling_path_watcher import PollingPathWatcher

if TYPE_CHECKING:
    from collections.abc import Callable


# 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["streamlit.watcher.event_based_path_watcher.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,
) -> bool:
    return _watch_path(path, on_file_changed, watcher_type)


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:
    # 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"}:
        return PollingPathWatcher
    return NoOpPathWatcher