# SPDX-FileCopyrightText: 2025 XLeitstelle Planen und Bauen <xleitstelle@gv.hamburg.de>
# SPDX-FileContributor: Tobias Kraft <tobias.kraft@gv.hamburg.de>
# SPDX-FileContributor: Anton Jacobsson <anton.jacobsson@init.de>
#
# SPDX-License-Identifier: EUPL-1.2

"""XMAS-App server and dependency management.

`ServerManager` handles dependency installation and everything needed
to get the embedded XMAS-App web server running inside QGIS:

1. **Dependency installation** — checks whether the required pip packages
   (listed in `app-requirements.txt`) are available and, if not, installs
   them into the QGIS user-site directory via a background `QgsTask`.
2. **Server lifecycle** — starts the XMAS-App in a thread, monitors it, and
   exposes `stop()` for clean shutdown on plugin unload.
3. **Health checking** — polls the webapp HTTP endpoint until it responds,
   then emits a signal so the dock widget can load the web view.

`pyqtwebengine_requirement()` and `verify_dependency_installation()` are also
used by `xmas_plugin.__init__` to ensure `QWebEngineView` is available before
the main plugin class is instantiated.
"""

import importlib
import importlib.metadata as md
import importlib.util
import logging
import os
import platform
import socket
import sys
import threading
from pathlib import Path
from urllib.parse import urlparse

from packaging.requirements import Requirement
from packaging.version import Version
from qgis.core import (
    QgsApplication,
    QgsDataSourceUri,
    QgsNetworkAccessManager,
    QgsTask,
)
from qgis.gui import QgisInterface
from qgis.PyQt import QtCore
from qgis.PyQt.QtCore import QObject, QProcess, QSize, QTimer, QUrl
from qgis.PyQt.QtGui import QFontDatabase
from qgis.PyQt.QtNetwork import QNetworkProxy, QNetworkProxyFactory, QNetworkProxyQuery
from qgis.PyQt.QtWidgets import (
    QApplication,
    QDialog,
    QDialogButtonBox,
    QHBoxLayout,
    QLabel,
    QMessageBox,
    QPlainTextEdit,
    QProgressBar,
    QPushButton,
    QVBoxLayout,
    QWidget,
)

from xmas_plugin import settings_manager
from xmas_plugin.settings_manager import get_appschema, load_setting
from xmas_plugin.util.db import get_db_schema, get_db_uri
from xmas_plugin.util.helpers import (
    PIP_TARGET_PATH,
    get_main_window,
    get_plugin_root,
)
from xmas_plugin.util.metadata import DEPENDENCIES, PLUGIN_DIR_NAME, PLUGIN_NAME

logger = logging.getLogger(PLUGIN_DIR_NAME)


def python_interpreter_for_qgis() -> str | None:
    """Locate the python interpreter used by QGIS."""
    try:
        # non-Linux
        path_py = os.environ["PYTHONHOME"]
    except Exception:
        # Linux
        path_py = sys.executable

    # convert to Path for easier processing
    path_py = Path(path_py)

    # pre-defined paths for python executable
    dict_pybin = {
        "Darwin": path_py / "bin" / "python3",
        "Windows": path_py / ("../../bin/pythonw.exe"),
        "Linux": path_py,
    }

    # python executable
    path_pybin = dict_pybin[platform.system()]

    if path_pybin.exists():
        return str(path_pybin)
    else:
        return None


def _effective_parent(parent=None):
    return parent or QApplication.activeWindow()


def _get_proxy_server() -> str | None:
    nam = QgsNetworkAccessManager.instance()
    if not nam:
        return
    proxy = (
        QNetworkProxyFactory.systemProxyForQuery(
            QNetworkProxyQuery(QUrl("https://files.pythonhosted.org/"))
        )[0]
        if nam.useSystemProxy()
        else nam.proxy()
    )
    url = QUrl()
    match proxy.type():
        case QNetworkProxy.HttpProxy | QNetworkProxy.HttpCachingProxy:
            url.setScheme("http")
        case QNetworkProxy.Socks5Proxy:
            url.setScheme("socks5")
        case _:
            return
    url.setHost(proxy.hostName())
    url.setPort(proxy.port())
    if user := proxy.user():
        url.setUserName(user)
    if pw := proxy.password():
        url.setPassword(pw)
    return url.toString()


def _pip_install_interactive(
    deps: list[str], *, parent=None, label: str = "Abhängigkeit", target: str = ""
) -> bool:
    parent = _effective_parent(parent)
    py = python_interpreter_for_qgis()
    if not py:
        QMessageBox.critical(
            parent,
            PLUGIN_NAME,
            f"Der Pfad des QGIS-Python-Interpreters konnte nicht ermittelt werden.\n\nInstallation von {label} wird abgebrochen.",
        )
        return False

    args = [
        "-m",
        "pip",
        "install",
    ]
    args += deps

    # configure pip environment
    pip_env = {"PIP_NO_DEPS": "1", "PIP_IGNORE_INSTALLED": "1", "PIP_NO_INPUT": "1"}
    if target:
        pip_env["PIP_TARGET"] = target
    if proxy := _get_proxy_server():
        pip_env["PIP_PROXY"] = proxy
    os.environ.update(pip_env)

    # Pre-confirm
    box = QMessageBox(
        QMessageBox.Question,
        PLUGIN_NAME,
        f"{label} wurde nicht gefunden.\n\nMöchten Sie die Installation jetzt starten?",
        QMessageBox.Yes | QMessageBox.No,
        parent,
    )
    box.setDefaultButton(QMessageBox.Yes)

    # make it a true top-level, always-on-top tool window
    flags = QtCore.Qt.Dialog | QtCore.Qt.WindowStaysOnTopHint | QtCore.Qt.Tool
    box.setWindowFlags(flags)
    box.setWindowModality(QtCore.Qt.ApplicationModal)

    ret = box.exec_()
    if ret != QMessageBox.Yes:
        return False

    # Cancel-able dialog
    dlg = PipInstallDialog(f"{label} wird installiert", parent=parent)
    dlg.raise_()
    dlg.activateWindow()
    dlg.start(py, args)
    dlg.exec_()
    ok = dlg.ok()

    return ok


def verify_nice_gui_app_installation(parent=None) -> bool:
    """Verify that the webapp was installed correctly ."""
    if not verify_dependency_installation(
        DEPENDENCIES["optional"], parent=parent, label="Webapp", target=PIP_TARGET_PATH
    ):
        return False

    # ensure the module exists afterwards (fast guard)
    if not importlib.util.find_spec("xmas_app.main"):
        logger.error("xmas_app.main not found after installation")
        return False
    return True


def pyqtwebengine_requirement() -> str:
    """Build a PyQtWebEngine requirement compatible with the installed PyQt5.

    Example:
        PyQt5.__version__ == "5.15.13" -> "PyQtWebEngine~=5.15"
    """
    import PyQt5  # TODO: Update when updating to PyQt6

    try:
        ver_str = getattr(PyQt5, "__version__", None)
        if not ver_str:
            from PyQt5.QtCore import QT_VERSION_STR

            ver_str = QT_VERSION_STR
    except Exception:
        return "PyQtWebEngine"

    v = Version(ver_str)
    base = f"{v.major}.{v.minor}"  # Go to closest major.minor release, e.g. "5.15"
    return f"PyQtWebEngine~={base}"


def verify_dependency_installation(
    reqs: list[str], *, parent=None, label: str = "Abhängigkeit", target: str = ""
) -> bool:
    """Ensure that the given Python requirements are installed (with correct versions).

    If not, trigger an interactive pip install via the QGIS Python interpreter.
    """
    missing: list[str] = []

    for req in reqs:
        requirement = Requirement(req)
        try:
            installed_version = md.version(requirement.name)
            if installed_version not in requirement.specifier:
                missing.append(str(requirement))
        except Exception:
            # Package not installed or version not resolvable
            missing.append(str(requirement))

    if not missing:
        return True

    # check if pip is even available
    try:
        pip_available = importlib.util.find_spec("pip.__main__") is not None
    except ModuleNotFoundError:
        pip_available = False
    if not pip_available:
        logger.error("pip not available, installation of %s impossible", label)
        QMessageBox.critical(
            parent,
            PLUGIN_NAME,
            f"Paketmanager 'pip' wurde nicht gefunden.\n\nInstallation von {label} kann nicht erfolgen.",
        )
        return False

    # app dependencies are read from a requirements.txt
    deps = (
        ["-r", f"{str(Path(get_plugin_root()) / 'app-requirements.txt')}"]
        if target
        else missing
    )

    if not _pip_install_interactive(deps, parent=parent, label=label, target=target):
        logger.error("installation of %s failed", label)
        return False

    # add target to sys.path after installation
    if target:
        sys.path.insert(0, PIP_TARGET_PATH)

    return True


def is_port_in_use(port: int, host: str) -> bool:
    """Return True if (host, port) is already bound by another process."""
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(1.0)  # just in case
    try:
        # If bind fails, it means the port is in use
        s.bind((host, port))
    except OSError:
        return True
    finally:
        s.close()
    return False


class ServerManager(QObject):
    """Manage the embedded XPlan web-server.

    Responsible for launching and stopping the server process, tracking its
    running state, and emitting `status_changed` whenever the server goes up
    or down.
    """

    def __init__(self, iface: QgisInterface) -> None:
        """Initialize the ServerManager.

        Args:
            iface: a pointer at the QgisInterface instance.
        """
        super().__init__()
        self.iface = iface
        self.server_url = settings_manager.get_normalized_url()
        if not self.server_url:
            parent = get_main_window()
            if parent is not None:
                QMessageBox.warning(
                    parent,
                    "Fehlende URL",
                    "Keine URL zum Webserver in den Plugin-Einstellungen gefunden. Bitte URL eingeben.",
                )
        self.server_host = urlparse(self.server_url).hostname
        self.server_port = urlparse(self.server_url).port
        logger.info(
            f"server_url: {self.server_url} host: {self.server_host} port: {self.server_port}"
        )
        self.server_status = False
        self.server = None
        self.thread = None
        logger.debug("ServerManager initialised")

    # Separate task that require main UI-thread
    def start_server_preflight_ui(self) -> None:
        """Runs on UI thread. May show dialogs."""
        if self.thread and self.thread.is_alive():
            logger.info("Webapp server already running.")
            return

        if not self.server_status:
            logger.info("No running server found..")
            self.iface.messageBar().pushInfo(
                PLUGIN_NAME, "Installationsstatus der Webapp wird überprüft..."
            )
            # Check if webapp was previously installed in vendor dir and add to sys.path
            if Path(PIP_TARGET_PATH).exists():
                sys.path.insert(0, PIP_TARGET_PATH)

            if not verify_nice_gui_app_installation(parent=get_main_window()):
                raise RuntimeError(
                    "Installation der Webapp abgebrochen oder fehlgeschlagen"
                )

            if is_port_in_use(self.server_port, self.server_host):
                raise RuntimeError(f"Port {self.server_port} ist bereits belegt")

    def _start_server_no_ui(self) -> None:
        """Task thread only. No Qt/UI calls."""
        if is_port_in_use(self.server_port, self.server_host):
            raise RuntimeError(f"Port {self.server_port} ist bereits belegt")

        try:
            creds = self._get_qgis_db_creds()
            logger.warning(
                "[DB EFFECTIVE] conn=%r service=%r host=%r port=%r db=%r auth=%r user=%r",
                load_setting("db_connection", ""),
                creds.get("PGSERVICE"),
                creds.get("PGHOST"),
                creds.get("PGPORT"),
                creds.get("PGDATABASE"),
            )
            logger.debug("DB credentials found")
        except ValueError as e:
            raise RuntimeError(
                "DB-Verbindungsparameter konnten nicht ermittelt werden"
            ) from e

        schema_type, schema_version = get_appschema()

        env_vars = {
            "appschema": schema_type,
            "appschema_version": schema_version,
            "app_mode": "prod",
            "app_port": str(self.server_port),
            **{k: str(v) for k, v in creds.items() if v not in (None, "")},
        }

        if (conn := load_setting("db_connection", "")) and (
            db_schema := get_db_schema(conn)
        ):
            env_vars["xmas_db_schema"] = db_schema

        OWNED_ENV_KEYS = (
            "PGSERVICE",
            "PGHOST",
            "PGPORT",
            "PGDATABASE",
            "PGUSER",
            "PGPASSWORD",
            "appschema",
            "appschema_version",
            "app_mode",
            "app_port",
            "xmas_db_schema",
        )
        for k in OWNED_ENV_KEYS:
            os.environ.pop(k, None)

        os.environ.update(env_vars)
        logger.warning(
            "[ENV NOW] PGSERVICE=%r PGHOST=%r PGPORT=%r PGDATABASE=%r appschema=%r/%r xmas_db_schema=%r",
            os.environ.get("PGSERVICE"),
            os.environ.get("PGHOST"),
            os.environ.get("PGPORT"),
            os.environ.get("PGDATABASE"),
            os.environ.get("appschema"),
            os.environ.get("appschema_version"),
            os.environ.get("xmas_db_schema"),
        )

        try:
            import uvicorn
            from xmas_app.main import create_app
        except ImportError as e:
            # no UI-message here (sub-thread), just raise with a clear message
            raise RuntimeError(f"Webapp konnte nicht initialisiert werden: {e}") from e

        config = uvicorn.Config(
            app=create_app,
            factory=True,
            host=self.server_host,
            port=self.server_port,
            reload=True,  # reload=True defaults to SelectorEventLoop which is compatible to async psycopg
            log_config=None,
        )
        self.server = uvicorn.Server(config)

        self.thread = threading.Thread(
            target=self._run_server, daemon=True, name="XMAS-Uvicorn-Thread"
        )
        self.thread.start()
        self.server_status = True

    def _run_server(self):
        logger.debug("_run_server entered")
        try:
            self.server.run()
        except Exception as e:
            logger.exception(f"Uvicorn server crashed: {e}")
        finally:
            logger.info("_run_server exited – cleaning handles")
            self.server_status = False
            self.thread = None  # clear stale handle

    def _start_server_task(self, task: QgsTask) -> bool:
        try:
            self._start_server_no_ui()
            return True
        except Exception:
            logger.exception("Start task crashed")
            raise

    def start_server_in_background(self) -> None:
        # prevent double-start if you want
        if getattr(self, "task", None) and self.task.status() in (
            QgsTask.Running,
            QgsTask.Queued,
        ):
            return

        self.iface.messageBar().pushSuccess(PLUGIN_NAME, "Webserver wird gestartet.")

        self.task = QgsTask.fromFunction(
            description="XMAS Webserver Start",
            function=self._start_server_task,
            on_finished=self._on_server_started,
        )
        QgsApplication.taskManager().addTask(self.task)

    def _on_server_started(self, exception: Exception | None, result: bool) -> None:
        if exception:
            logger.error("Start task failed: %r", exception, exc_info=True)
            self.iface.messageBar().pushCritical(PLUGIN_NAME, f"{exception!r}")
            self.server_status = False
            return

        if result:
            logger.info("Server start task finished successfully.")
            self.iface.messageBar().pushSuccess(PLUGIN_NAME, "Webserver gestartet.")

    def request_stop(self) -> None:
        if self.server:
            self.server.should_exit = True
            # optional hard stop flag if available
            if hasattr(self.server, "force_exit"):
                self.server.force_exit = False  # keep graceful by default

    def is_alive(self) -> bool:
        return bool(self.thread and self.thread.is_alive())

    def _get_qgis_db_creds(self) -> dict:
        """Get PG connection params.

        Retrieves the QGIS Postgres connection from the plugin settings,
        uses QgsDataSourceUri to extract individual params.

        Returns:
            A dict with PG connection params.
        """
        db_conn_name = load_setting("db_connection", "")
        if not db_conn_name:
            raise ValueError("No 'db_connection' found in QGIS plugin settings.")

        uri = QgsDataSourceUri(get_db_uri(db_conn_name))

        creds = {
            "PGUSER": uri.username(),
            "PGPASSWORD": uri.password(),
            "PGSERVICE": uri.service(),
            "PGHOST": uri.host(),
            "PGPORT": uri.port(),
            "PGDATABASE": uri.database(),
        }

        return creds


class PipInstallDialog(QDialog):
    """A dialog to install required dependencies via pip.

    Shows installation progress and installed packages to provide feedback to the user.
    """

    def __init__(self, title: str, parent=None):
        super().__init__(parent)
        self.setWindowTitle(title)
        self.setModal(True)
        # Console-like proportions
        self.resize(900, 520)  # wider than tall
        self.setMinimumSize(QSize(720, 420))

        self.proc = QProcess(self)
        self.proc.setProcessChannelMode(QProcess.MergedChannels)
        self.proc.readyReadStandardOutput.connect(self._on_ready)
        self.proc.finished.connect(self._on_finished)
        self.proc.errorOccurred.connect(self._on_error)

        self.label = QLabel("Installation läuft… (dies kann einige Minuten dauern)")
        self.out = QPlainTextEdit(readOnly=True)
        # monospaced, no wrapping → proper “console”
        self.out.setLineWrapMode(QPlainTextEdit.NoWrap)
        self.out.setWordWrapMode(0)
        self.out.setUndoRedoEnabled(False)
        fixed = QFontDatabase.systemFont(QFontDatabase.FixedFont)
        self.out.setFont(fixed)

        self.progress = QProgressBar()
        self.progress.setRange(0, 0)  # indeterminate

        self.buttons = QDialogButtonBox()
        self.btn_cancel = self.buttons.addButton(QDialogButtonBox.Cancel)
        self.btn_cancel.clicked.connect(self._on_cancel)

        # small actions row: copy log, clear
        actions = QWidget()
        ah = QHBoxLayout(actions)
        ah.setContentsMargins(0, 0, 0, 0)
        self.btn_copy = QPushButton("Ausgabe kopieren")
        self.btn_clear = QPushButton("Leeren")
        self.btn_copy.clicked.connect(lambda: self._copy_all())
        self.btn_clear.clicked.connect(lambda: self.out.setPlainText(""))
        ah.addWidget(self.btn_copy)
        ah.addWidget(self.btn_clear)
        ah.addStretch()

        lay = QVBoxLayout(self)
        lay.addWidget(self.label)
        lay.addWidget(self.out)
        lay.addWidget(actions)
        lay.addWidget(self.progress)
        lay.addWidget(self.buttons)

        self._result_ok = False

    # API
    def start(self, program: str, arguments: list[str]) -> None:
        self.out.appendPlainText(f"$ {program} " + " ".join(arguments))
        self.proc.start(program, arguments)
        self.show()

    def ok(self) -> bool:
        return self._result_ok

    # slots
    def _on_ready(self):
        data = bytes(self.proc.readAllStandardOutput()).decode(errors="replace")
        if data:
            self.out.appendPlainText(data.rstrip())

    def _on_finished(self, exitCode: int, exitStatus: QProcess.ExitStatus):
        self.progress.setRange(0, 1)
        self.progress.setValue(1)
        self._result_ok = exitStatus == QProcess.NormalExit and exitCode == 0

        if self._result_ok:
            self.label.setText("✅ Installation erfolgreich abgeschlossen.")
            self.progress.setStyleSheet(
                "QProgressBar::chunk { background-color: #4CAF50; }"
            )
            self.out.appendPlainText(
                "\n[Erfolg] Installation erfolgreich abgeschlossen.\n"
            )
        else:
            self.label.setText("❌ Installation fehlgeschlagen oder abgebrochen.")
            self.progress.setStyleSheet(
                "QProgressBar::chunk { background-color: #E53935; }"
            )
            self.out.appendPlainText(f"\n[Fehler] Beendet mit Code {exitCode}.\n")

        self.btn_cancel.setEnabled(True)
        self.btn_cancel.setText("Schließen")

    def _on_error(self, err: QProcess.ProcessError):
        self.out.appendPlainText(f"\n[Fehler] QProcess error: {err}")

    def _on_cancel(self):
        if self.proc.state() != QProcess.NotRunning:
            self.out.appendPlainText("\nAbbruch angefordert…")
            self.btn_cancel.setEnabled(False)
            self.proc.terminate()
            QTimer.singleShot(
                2500,
                lambda: (
                    self.proc.kill()
                    if self.proc.state() != QProcess.NotRunning
                    else None
                ),
            )
        else:
            self.reject()

    def _copy_all(self):
        self.out.selectAll()
        self.out.copy()
