import asyncio
import concurrent.futures
import logging
import os
import signal
import traceback
from pathlib import Path
from typing import Optional

from flet.controls.types import AppView, WebRenderer
from flet.controls.update_behavior import UpdateBehavior
from flet.messaging.session import Session
from flet.utils import (
    get_bool_env_var,
    get_current_script_dir,
    get_free_tcp_port,
    is_embedded,
    is_linux_server,
    is_pyodide,
    open_in_browser,
)
from flet.utils.deprecated import deprecated
from flet.utils.pip import (
    ensure_flet_desktop_package_installed,
    ensure_flet_web_package_installed,
)

logger = logging.getLogger("flet")


@deprecated("Use run() instead.", version="0.70.0", show_parentheses=True)
def app(*args, **kwargs):
    return run(*args, **kwargs)


@deprecated("Use run() instead.", version="0.70.0", show_parentheses=True)
async def app_async(*args, **kwargs):
    return await run_async(*args, **kwargs)


def run(
    target,
    name="",
    host=None,
    port=0,
    view: Optional[AppView] = AppView.FLET_APP,
    assets_dir="assets",
    upload_dir=None,
    web_renderer: WebRenderer = WebRenderer.AUTO,
    use_color_emoji=False,
    route_url_strategy="path",
    export_asgi_app=False,
):
    if is_pyodide():
        __run_pyodide(target)
        return

    if export_asgi_app:
        ensure_flet_web_package_installed()
        from flet_web.fastapi.serve_fastapi_web_app import get_fastapi_web_app

        return get_fastapi_web_app(
            session_handler=target,
            page_name=__get_page_name(name),
            assets_dir=__get_assets_dir_path(assets_dir, relative_to_cwd=True),
            upload_dir=__get_upload_dir_path(upload_dir, relative_to_cwd=True),
            web_renderer=web_renderer,
            use_color_emoji=use_color_emoji,
            route_url_strategy=route_url_strategy,
        )

    return asyncio.run(
        run_async(
            target=target,
            name=name,
            host=host,
            port=port,
            view=view,
            assets_dir=assets_dir,
            upload_dir=upload_dir,
            web_renderer=web_renderer,
            use_color_emoji=use_color_emoji,
            route_url_strategy=route_url_strategy,
        )
    )


async def run_async(
    target,
    name="",
    host=None,
    port=0,
    view: Optional[AppView] = AppView.FLET_APP,
    assets_dir="assets",
    upload_dir=None,
    web_renderer: WebRenderer = WebRenderer.AUTO,
    use_color_emoji=False,
    route_url_strategy="path",
):
    if is_pyodide():
        __run_pyodide(target)
        return

    if isinstance(view, str):
        view = AppView(view)

    if isinstance(web_renderer, str):
        web_renderer = WebRenderer(web_renderer)

    force_web_server = get_bool_env_var("FLET_FORCE_WEB_SERVER") or is_linux_server()
    if force_web_server:
        view = AppView.WEB_BROWSER

    env_port = os.getenv("FLET_SERVER_PORT")
    if env_port is not None and env_port:
        port = int(env_port)

    if port == 0 and force_web_server:
        port = 8000

    env_host = os.getenv("FLET_SERVER_IP")
    if env_host is not None and env_host:
        host = env_host

    assets_dir = __get_assets_dir_path(assets_dir)
    upload_dir = __get_upload_dir_path(upload_dir)
    page_name = __get_page_name(name)

    is_socket_server = (
        is_embedded() or view == AppView.FLET_APP or view == AppView.FLET_APP_HIDDEN
    ) and not force_web_server

    url_prefix = os.getenv("FLET_DISPLAY_URL_PREFIX")

    def on_app_startup(page_url):
        if url_prefix is not None:
            print(url_prefix, page_url)
        else:
            logger.info(f"App URL: {page_url}")

        if view == AppView.WEB_BROWSER and url_prefix is None and not force_web_server:
            open_in_browser(page_url)

    loop = asyncio.get_running_loop()

    terminate = asyncio.Event()

    def exit_gracefully(signum, frame):
        logger.debug("Gracefully terminating Flet app...")
        loop.call_soon_threadsafe(terminate.set)
        signal.signal(signal.SIGINT, signal.SIG_DFL)
        signal.signal(signal.SIGTERM, signal.SIG_DFL)

    signal.signal(signal.SIGINT, exit_gracefully)
    signal.signal(signal.SIGTERM, exit_gracefully)

    conn = (
        await __run_socket_server(
            port=port,
            session_handler=target,
            blocking=is_embedded(),
        )
        if is_socket_server
        else await __run_web_server(
            session_handler=target,
            host=host,
            port=port,
            page_name=page_name,
            assets_dir=assets_dir,
            upload_dir=upload_dir,
            web_renderer=web_renderer,
            use_color_emoji=use_color_emoji,
            route_url_strategy=route_url_strategy,
            blocking=(view == AppView.WEB_BROWSER or view is None or force_web_server),
            on_startup=on_app_startup,
        )
    )

    logger.info("Flet app has started...")

    try:
        if (
            (
                view == AppView.FLET_APP
                or view == AppView.FLET_APP_HIDDEN
                or view == AppView.FLET_APP_WEB
            )
            and not force_web_server
            and not is_embedded()
            and url_prefix is None
        ):
            ensure_flet_desktop_package_installed()
            from flet_desktop import close_flet_view, open_flet_view_async

            on_app_startup(conn.page_url)

            fvp, pid_file = await open_flet_view_async(
                conn.page_url,
                assets_dir if view != AppView.FLET_APP_WEB else None,
                view == AppView.FLET_APP_HIDDEN,
            )
            try:
                await fvp.wait()
            except:
                pass

            close_flet_view(pid_file)

        elif url_prefix and is_socket_server:
            on_app_startup(conn.page_url)

            try:
                await terminate.wait()
            except KeyboardInterrupt:
                pass

    finally:
        await conn.close()


async def __run_socket_server(port=0, session_handler=None, blocking=False):
    from flet.messaging.flet_socket_server import FletSocketServer

    uds_path = os.getenv("FLET_SERVER_UDS_PATH")

    executor = concurrent.futures.ThreadPoolExecutor()

    async def on_session_created(session: Session):
        logger.info("App session started")
        try:
            assert session_handler is not None
            UpdateBehavior.reset()
            if asyncio.iscoroutinefunction(session_handler):
                await session_handler(session.page)
            else:
                # run synchronously
                session_handler(session.page)

            if UpdateBehavior.auto_update_enabled():
                session.auto_update(session.page)

        except Exception as e:
            print(
                f"Unhandled error processing page session {session.id}:",
                traceback.format_exc(),
            )
            session.error(f"The application encountered an error: {e}")

    conn = FletSocketServer(
        loop=asyncio.get_running_loop(),
        port=port,
        uds_path=uds_path,
        on_session_created=on_session_created,
        blocking=blocking,
        executor=executor,
    )
    await conn.start()
    return conn


async def __run_web_server(
    session_handler,
    host,
    port,
    page_name,
    assets_dir,
    upload_dir,
    web_renderer: Optional[WebRenderer],
    use_color_emoji,
    route_url_strategy,
    blocking,
    on_startup,
):
    ensure_flet_web_package_installed()
    from flet_web.fastapi.serve_fastapi_web_app import serve_fastapi_web_app

    url_host = "127.0.0.1" if host in [None, "", "*"] else host

    if port == 0:
        port = get_free_tcp_port()

    logger.info(f"Starting Flet web server on port {port}...")

    log_level = logging.getLogger("flet").getEffectiveLevel()
    if log_level == logging.CRITICAL or log_level == logging.NOTSET:
        log_level = logging.FATAL

    return await serve_fastapi_web_app(
        session_handler,
        host=host,
        url_host=url_host,
        port=port,
        page_name=page_name,
        assets_dir=assets_dir,
        upload_dir=upload_dir,
        web_renderer=web_renderer,
        use_color_emoji=use_color_emoji,
        route_url_strategy=route_url_strategy,
        blocking=blocking,
        on_startup=on_startup,
        log_level=logging.getLevelName(log_level).lower(),
    )


def __run_pyodide(target):
    from flet.messaging.pyodide_connection import PyodideConnection

    async def on_session_created(session: Session):
        logger.info("App session started")
        try:
            assert target is not None
            UpdateBehavior.reset()
            if asyncio.iscoroutinefunction(target):
                await target(session.page)
            else:
                target(session.page)

            if UpdateBehavior.auto_update_enabled():
                session.auto_update(session.page)
        except Exception as e:
            print(
                f"Unhandled error processing page session {session.id}:",
                traceback.format_exc(),
            )
            session.error(f"There was an error while processing your request: {e}")

    conn = PyodideConnection(
        on_session_created=on_session_created,
    )


def __get_page_name(name: str):
    env_page_name = os.getenv("FLET_WEB_APP_PATH")
    return env_page_name if not name and env_page_name else name


def __get_assets_dir_path(assets_dir: Optional[str], relative_to_cwd=False):
    if assets_dir:
        if not Path(assets_dir).is_absolute():
            if "_MEI" in __file__:
                # support for "onefile" PyInstaller
                assets_dir = str(
                    Path(__file__).parent.parent.joinpath(assets_dir).resolve()
                )
            else:
                assets_dir = str(
                    Path(os.getcwd() if relative_to_cwd else get_current_script_dir())
                    .joinpath(assets_dir)
                    .resolve()
                )
        logger.info(f"Assets path configured: {assets_dir}")

    env_assets_dir = os.getenv("FLET_ASSETS_DIR")
    if env_assets_dir:
        assets_dir = env_assets_dir
    return assets_dir


def __get_upload_dir_path(upload_dir: Optional[str], relative_to_cwd=False):
    if upload_dir:
        if not Path(upload_dir).is_absolute():
            upload_dir = str(
                Path(os.getcwd() if relative_to_cwd else get_current_script_dir())
                .joinpath(upload_dir)
                .resolve()
            )
    return upload_dir