"""
Run a user-supplied Python script as a workflow task via subprocess.

Mirrors the subprocess pattern used by `python-bridge/engine_bridge.py`:
the script runs in its own Python interpreter, gets workflow variables on
stdin as JSON, and is expected to write its result to stdout.

The script contract (documented in the workflow editor tooltip):
- Workflow variables are delivered as a JSON object on stdin.
- The script may write logging text to stderr — that text is streamed to
  the execution log so the user sees progress.
- If the script writes a JSON object to stdout, it becomes the task
  result and is available via `last_task_result` and any
  `store_result_as` variable. Non-JSON stdout is wrapped as
  ``{"stdout": "<text>"}`` instead of failing.
- Non-zero exit code is treated as a workflow error (with stderr in the
  message) unless the step's `continue_on_error` flag is set.
"""

from __future__ import annotations

import json
import os
import shlex
import shutil
import subprocess
import sys
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional


WORKFLOW_VARIABLES_ENV_VAR = "EMDS_WORKFLOW_VARIABLES_JSON"

# A Python-script step must never hang QGIS forever. When the step leaves the
# Timeout field empty (or sets a non-positive value) we fall back to this
# bounded default so a stuck/slow script surfaces an error instead of freezing
# the whole application with no recovery. (BUG 3)
DEFAULT_SCRIPT_TIMEOUT_SECONDS = 600.0

# The runner executes on the QGIS UI thread. We wait for the child in short
# slices so the caller can pump the event loop (keeping the busy dialog alive)
# and so a timeout can actually fire.
_WAIT_TICK_SECONDS = 0.2

# How long to wait for a reader thread to finish before giving up on it. A
# thread that is still alive means the child left a pipe write-handle open; we
# must NOT then close that pipe (it would deadlock — see _teardown_streams).
_READER_JOIN_TIMEOUT_SECONDS = 2.0

# After a timeout kill, how long to wait for the child to actually die.
_KILL_WAIT_SECONDS = 5.0


class PythonScriptRunError(RuntimeError):
    """Raised when the script process fails or its output cannot be read."""


@dataclass
class PythonScriptRunResult:
    """Outcome of a single script execution."""

    exit_code: int
    stdout: str
    stderr: str
    parsed_stdout: Any = None
    duration_seconds: float = 0.0
    command: List[str] = field(default_factory=list)

    def to_task_result(self) -> Dict[str, Any]:
        """Shape the result as a workflow task-runner return value."""
        if isinstance(self.parsed_stdout, dict):
            return dict(self.parsed_stdout)
        return {
            "engine_type": "PythonScript",
            "exit_code": self.exit_code,
            "stdout": self.stdout,
            "stderr": self.stderr,
            "parsed_stdout": self.parsed_stdout,
        }


def _executable_stem(path: str) -> str:
    """Lower-cased basename of *path* without an ``.exe`` suffix."""
    name = os.path.basename(str(path or "")).lower()
    if name.endswith(".exe"):
        name = name[:-4]
    return name


def _is_qgis_gui_executable(path: str) -> bool:
    """True when *path* is the QGIS GUI launcher, not a Python interpreter.

    In QGIS 4 on Windows ``sys.executable`` is ``qgis-bin.exe`` (also
    ``qgis.exe``, ``qgis-ltr-bin.exe``, ``qgis-bin-g7.exe`` …). Handing any of
    these to the subprocess launches a SECOND QGIS that tries to open the
    ``.py`` file as a data source, replacing the running session. (BUG 1)
    """
    return _executable_stem(path).startswith("qgis")


def _looks_like_python(path: str) -> bool:
    """True when *path*'s basename looks like a Python interpreter."""
    stem = _executable_stem(path)
    return stem.startswith("python") or stem in {"py", "pypy", "pypy3"}


def _default_python_executable() -> str:
    """Best-effort path to a REAL Python interpreter for the default case.

    ``sys.executable`` is only usable when it is genuinely a Python interpreter;
    under QGIS on Windows it is the GUI launcher, so we look elsewhere: the
    recorded base interpreter (Python 3.11+), the interpreter that OSGeo4W lays
    down at ``sys.prefix`` (``…\\apps\\Python312\\python.exe``), and finally
    ``PATH``. The QGIS GUI executable is never returned.
    """
    candidates: List[str] = []

    exe = sys.executable or ""
    if _looks_like_python(exe):
        candidates.append(exe)

    # Python 3.11+ records the real base interpreter here even when the running
    # process was launched via a stub/launcher (QGIS embeds Python this way).
    base_exe = getattr(sys, "_base_executable", "") or ""
    if _looks_like_python(base_exe):
        candidates.append(base_exe)

    # OSGeo4W ships the interpreter alongside the prefix
    # (sys.prefix == …\\apps\\Python312, which contains python.exe).
    prefix = sys.prefix or ""
    if prefix:
        if os.name == "nt":
            candidates.append(os.path.join(prefix, "python.exe"))
            candidates.append(os.path.join(prefix, "python3.exe"))
        else:
            candidates.append(os.path.join(prefix, "bin", "python3"))
            candidates.append(os.path.join(prefix, "bin", "python"))

    # PATH lookups as a final fallback.
    for name in ("python3", "python"):
        found = shutil.which(name)
        if found:
            candidates.append(found)

    for candidate in candidates:
        if candidate and not _is_qgis_gui_executable(candidate) and os.path.isfile(candidate):
            return candidate
    return ""


def _resolve_python_executable(configured: str) -> str:
    """Pick the interpreter, resolving a real Python for the default case.

    Never returns the QGIS GUI executable: a blank configuration resolves an
    actual interpreter, and an explicit configuration that points at the GUI
    exe is rejected with a clear error rather than silently launching a second
    QGIS. (BUG 1)
    """
    text = (configured or "").strip()
    if text:
        candidate = Path(text).expanduser()
        resolved = str(candidate) if candidate.exists() else text
        if _is_qgis_gui_executable(resolved):
            raise PythonScriptRunError(
                f"The configured Python interpreter '{resolved}' is the QGIS GUI "
                "executable, not a Python interpreter. Set the step's 'Python "
                "Interpreter' to a real python.exe "
                "(e.g. C:\\OSGeo4W\\apps\\Python312\\python.exe)."
            )
        return resolved

    resolved = _default_python_executable()
    if not resolved:
        raise PythonScriptRunError(
            "Could not locate a Python interpreter for the Python Script step. "
            "Set the step's 'Python Interpreter' to a python.exe path "
            "(e.g. C:\\OSGeo4W\\apps\\Python312\\python.exe)."
        )
    return resolved


def _interpreter_matches_runtime(interpreter: str) -> bool:
    """True when *interpreter* is the same Python that QGIS is running.

    Only then is it safe to pass QGIS's ``PYTHONHOME``/``PYTHONPATH`` through to
    the child. A foreign/standalone interpreter that inherits them hangs at
    startup trying to load OSGeo4W's stdlib. (BUG 2)
    """
    try:
        interp_dir = os.path.normcase(os.path.dirname(os.path.realpath(interpreter)))
    except Exception:
        return False
    if not interp_dir:
        return False

    reference_dirs: List[str] = []
    for ref in (sys.executable or "", getattr(sys, "_base_executable", "") or ""):
        if _looks_like_python(ref):
            try:
                reference_dirs.append(os.path.dirname(os.path.realpath(ref)))
            except Exception:
                pass
    for prefix in (sys.prefix or "", sys.exec_prefix or ""):
        if prefix:
            try:
                reference_dirs.append(os.path.realpath(prefix))
                if os.name != "nt":
                    reference_dirs.append(os.path.join(os.path.realpath(prefix), "bin"))
            except Exception:
                pass

    return any(interp_dir == os.path.normcase(d) for d in reference_dirs if d)


def _build_child_env(
    interpreter: str,
    extra_env: Any,
    variables_payload: str,
) -> Dict[str, str]:
    """Assemble the subprocess environment.

    For a foreign interpreter the inherited QGIS ``PYTHON*`` locators are
    stripped so it uses its own stdlib instead of hanging on OSGeo4W's. (BUG 2)
    Caller-supplied ``extra_env`` still wins, so a user can re-add a variable if
    they genuinely need it.
    """
    env = dict(os.environ)
    if not _interpreter_matches_runtime(interpreter):
        for var in ("PYTHONHOME", "PYTHONPATH", "PYTHONSTARTUP"):
            env.pop(var, None)
    env.update(_coerce_env(extra_env))
    env["PYTHONIOENCODING"] = env.get("PYTHONIOENCODING", "utf-8")
    env["PYTHONUNBUFFERED"] = "1"
    env[WORKFLOW_VARIABLES_ENV_VAR] = variables_payload
    return env


def _split_arguments(arguments: Any) -> List[str]:
    """Normalize the args field — accept list or shell-style string."""
    if arguments is None:
        return []
    if isinstance(arguments, list):
        return [str(arg) for arg in arguments]
    text = str(arguments).strip()
    if not text:
        return []
    try:
        return shlex.split(text, posix=os.name != "nt")
    except ValueError:
        return [text]


def _coerce_env(extra_env: Any) -> Dict[str, str]:
    if not isinstance(extra_env, dict):
        return {}
    return {str(name): str(value) for name, value in extra_env.items() if name}


def _stream_stderr(process: subprocess.Popen, log_callback: Optional[Callable[[str], None]]) -> str:
    """Read stderr line-by-line, forward to log_callback, return joined text."""
    captured: List[str] = []
    if process.stderr is None:
        return ""
    for raw_line in iter(process.stderr.readline, ""):
        if raw_line == "":
            break
        line = raw_line.rstrip("\r\n")
        captured.append(line)
        if callable(log_callback):
            try:
                log_callback(line)
            except Exception:
                pass
    return "\n".join(captured)


def _drain_text_stream(
    stream,
    *,
    log_callback: Optional[Callable[[str], None]] = None,
) -> tuple[List[str], threading.Thread]:
    """Drain a process text stream in the background.

    The Python-script runner needs live stderr logging, but it also needs to
    drain stdout at the same time so scripts cannot deadlock when stdout fills
    the pipe. This helper keeps the process wait/timeout path independent from
    pipe reads.
    """
    chunks: List[str] = []

    def _reader() -> None:
        if stream is None:
            return
        try:
            if callable(log_callback):
                for raw_line in iter(stream.readline, ""):
                    if raw_line == "":
                        break
                    chunks.append(raw_line)
                    line = raw_line.rstrip("\r\n")
                    try:
                        log_callback(line)
                    except Exception:
                        pass
            else:
                while True:
                    chunk = stream.read(4096)
                    if not chunk:
                        break
                    chunks.append(chunk)
        except Exception:
            # The main thread owns user-facing error reporting. Stream reader
            # failures after process teardown are not actionable on their own.
            pass

    thread = threading.Thread(target=_reader, daemon=True)
    thread.start()
    return chunks, thread


def _wait_for_process(
    process: subprocess.Popen,
    timeout_seconds: Optional[float],
    poll_callback: Optional[Callable[[], None]],
) -> None:
    """Wait for *process* to exit, pumping ``poll_callback`` between slices.

    The runner executes on the QGIS UI thread, so we wait in short slices
    instead of one blocking call: between slices ``poll_callback`` lets the
    caller keep the busy dialog painted, and a non-``None`` ``timeout_seconds``
    is honoured so a hung child cannot freeze QGIS forever. Raises
    ``subprocess.TimeoutExpired`` when the deadline passes. (BUG 3)
    """
    import time

    deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds
    while True:
        slice_timeout = _WAIT_TICK_SECONDS
        if deadline is not None:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise subprocess.TimeoutExpired(process.args, timeout_seconds)
            slice_timeout = min(slice_timeout, remaining)
        try:
            process.wait(timeout=slice_timeout)
            return
        except subprocess.TimeoutExpired:
            if callable(poll_callback):
                try:
                    poll_callback()
                except Exception:
                    pass


def _safe_close(stream) -> None:
    try:
        if stream is not None:
            stream.close()
    except Exception:
        pass


def _teardown_streams(
    process: subprocess.Popen,
    stdout_thread: Optional[threading.Thread],
    stderr_thread: Optional[threading.Thread],
) -> None:
    """Join the reader threads and close the pipes without deadlocking. (BUG 3)

    A pipe must never be ``close()``d while its reader thread is still blocked
    inside a read on it: ``close()`` would block forever acquiring the
    buffered-IO lock the reader holds, freezing the calling (QGIS UI) thread —
    the exact hang observed after a script exits while a leaked child handle
    keeps a pipe write-end open. So we only close a stream once its reader has
    actually finished; if the reader is still stuck we leave the stream for the
    daemon thread / GC rather than deadlock.
    """
    # stdin has no reader thread, so closing it can never deadlock.
    _safe_close(process.stdin)

    for thread, stream in (
        (stdout_thread, process.stdout),
        (stderr_thread, process.stderr),
    ):
        if thread is not None:
            thread.join(timeout=_READER_JOIN_TIMEOUT_SECONDS)
            if thread.is_alive():
                # Reader still blocked (child left a pipe write-handle open).
                # Closing now would deadlock; leave it to be reclaimed by GC.
                continue
        _safe_close(stream)


def run_python_script(
    *,
    script_path: str,
    arguments: Any = None,
    working_directory: str = "",
    extra_env: Optional[Dict[str, str]] = None,
    python_executable: str = "",
    timeout_seconds: Optional[float] = None,
    workflow_variables: Optional[Dict[str, Any]] = None,
    log_callback: Optional[Callable[[str], None]] = None,
    poll_callback: Optional[Callable[[], None]] = None,
) -> PythonScriptRunResult:
    """Run *script_path* in an isolated Python subprocess and return its result.

    Pass ``log_callback`` to receive each stderr line in real time (the
    executor wires this to the workflow execution log). Pass ``poll_callback``
    to be called periodically while waiting for the child so the UI thread can
    stay responsive (the executor wires this to ``QApplication.processEvents``).
    """
    import time

    script = (script_path or "").strip()
    if not script:
        raise PythonScriptRunError("No Python script path was provided.")

    script_resolved = Path(script).expanduser()
    if not script_resolved.is_file():
        raise PythonScriptRunError(f"Python script not found: {script_resolved}")

    interpreter = _resolve_python_executable(python_executable)
    args = _split_arguments(arguments)
    command = [interpreter, str(script_resolved), *args]

    cwd_text = (working_directory or "").strip()
    cwd: Optional[str] = None
    if cwd_text:
        cwd_path = Path(cwd_text).expanduser()
        if not cwd_path.is_dir():
            raise PythonScriptRunError(f"Working directory does not exist: {cwd_path}")
        cwd = str(cwd_path)

    variables_payload = json.dumps(workflow_variables or {}, default=str)
    env = _build_child_env(interpreter, extra_env, variables_payload)

    creationflags = 0
    if os.name == "nt":
        creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)

    started_at = time.monotonic()
    try:
        process = subprocess.Popen(
            command,
            cwd=cwd,
            env=env,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            encoding="utf-8",
            errors="replace",
            creationflags=creationflags,
        )
    except OSError as exc:
        raise PythonScriptRunError(
            f"Failed to launch Python interpreter '{interpreter}': {exc}"
        ) from exc

    stdout_chunks: List[str] = []
    stderr_chunks: List[str] = []
    stdout_thread: Optional[threading.Thread] = None
    stderr_thread: Optional[threading.Thread] = None
    timed_out = False
    try:
        stdout_chunks, stdout_thread = _drain_text_stream(process.stdout)
        stderr_chunks, stderr_thread = _drain_text_stream(
            process.stderr,
            log_callback=log_callback,
        )

        try:
            if process.stdin is not None:
                process.stdin.write(variables_payload)
                process.stdin.close()
        except (BrokenPipeError, OSError):
            pass

        try:
            _wait_for_process(process, timeout_seconds, poll_callback)
        except subprocess.TimeoutExpired:
            timed_out = True
            process.kill()
            # Killing closes the child's pipe write-ends, so the reader threads
            # unblock and _teardown_streams below can close safely.
            try:
                process.wait(timeout=_KILL_WAIT_SECONDS)
            except Exception:
                pass
    finally:
        _teardown_streams(process, stdout_thread, stderr_thread)

    duration = time.monotonic() - started_at
    stdout_text = "".join(stdout_chunks)
    stderr_text = "".join(stderr_chunks).rstrip("\r\n")

    if timed_out:
        message = f"Python script timed out after {timeout_seconds} seconds: {script_resolved}"
        stderr_tail = stderr_text.strip().splitlines()
        if stderr_tail:
            message += f": {stderr_tail[-1]}"
        raise PythonScriptRunError(message)

    parsed_stdout: Any = None
    stripped_stdout = (stdout_text or "").strip()
    if stripped_stdout:
        try:
            parsed_stdout = json.loads(stripped_stdout)
        except json.JSONDecodeError:
            parsed_stdout = None

    if process.returncode != 0:
        first_stderr_line = stderr_text.strip().splitlines()[-1] if stderr_text.strip() else ""
        message = (
            f"Python script '{script_resolved.name}' exited with code "
            f"{process.returncode}"
        )
        if first_stderr_line:
            message += f": {first_stderr_line}"
        raise PythonScriptRunError(message)

    return PythonScriptRunResult(
        exit_code=process.returncode,
        stdout=stdout_text or "",
        stderr=stderr_text or "",
        parsed_stdout=parsed_stdout,
        duration_seconds=duration,
        command=list(command),
    )


def run_python_script_from_step_config(
    config: Dict[str, Any],
    *,
    workflow_variables: Optional[Dict[str, Any]] = None,
    log_callback: Optional[Callable[[str], None]] = None,
    poll_callback: Optional[Callable[[], None]] = None,
) -> Dict[str, Any]:
    """Run the Python script described by a workflow-definition step config.

    Returns the dict that the workflow task_runner should hand back to the
    executor (becomes the task's result and the source of any
    `store_result_as` variable).
    """
    config = dict(config or {})
    timeout_value = config.get("timeout_seconds")
    timeout: Optional[float]
    try:
        timeout = float(timeout_value) if timeout_value not in (None, "") else None
    except (TypeError, ValueError):
        timeout = None
    if timeout is None or timeout <= 0:
        # An empty/invalid/non-positive Timeout must not mean "wait forever":
        # fall back to a bounded default so a stuck script surfaces an error
        # instead of freezing QGIS with no recovery. (BUG 3)
        timeout = DEFAULT_SCRIPT_TIMEOUT_SECONDS

    result = run_python_script(
        script_path=str(config.get("script_path", "") or ""),
        arguments=config.get("arguments"),
        working_directory=str(config.get("working_directory", "") or ""),
        extra_env=config.get("environment"),
        python_executable=str(config.get("python_executable", "") or ""),
        timeout_seconds=timeout,
        workflow_variables=workflow_variables,
        log_callback=log_callback,
        poll_callback=poll_callback,
    )
    return result.to_task_result()
