from __future__ import annotations

import ctypes
import json
import os
import shutil
import signal
import threading
import time
import uuid
from pathlib import Path
from typing import Any
import subprocess
import sys

from .config import RuntimeOptions, build_runtime_options, validate_config
from .types import BuyTaskRecord

_TASKS: dict[str, BuyTaskRecord] = {}
_TASKS_LOCK = threading.Lock()


def _collect_payment_fields_from_event(event: Any) -> dict[str, Any]:
    result: dict[str, Any] = {}
    state = getattr(event, "state", None)
    for key in ("payment_qr_url", "order_id", "order_detail_url", "payment_code_url"):
        value = getattr(state, key, None) if state is not None else None
        if value not in (None, ""):
            result[key] = value

    message = getattr(event, "message", None)
    if isinstance(message, str):
        markers = {
            "PAYMENT_QR_URL=": "payment_qr_url",
            "ORDER_ID=": "order_id",
            "ORDER_DETAIL_URL=": "order_detail_url",
            "PAYMENT_CODE_URL=": "payment_code_url",
        }
        for prefix, field in markers.items():
            if message.startswith(prefix):
                result[field] = message.split("=", 1)[1]
                break
    return result


def _package_root() -> Path:
    return Path(__file__).resolve().parents[1]


def _managed_runs_root(root: str | Path | None = None) -> Path:
    target = Path(root) if root is not None else _package_root() / "btb_runs"
    target.mkdir(parents=True, exist_ok=True)
    return target


def _dump_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = path.with_suffix(path.suffix + ".tmp")
    with open(tmp_path, "w", encoding="utf-8") as handle:
        json.dump(payload, handle, ensure_ascii=False, indent=2)
    os.replace(tmp_path, path)


def _load_json(path: Path) -> dict[str, Any]:
    with open(path, "r", encoding="utf-8") as handle:
        return json.load(handle)


def _read_text_tail(path: Path, *, max_lines: int = 20) -> str:
    if not path.exists():
        return ""
    with open(path, "r", encoding="utf-8", errors="replace") as handle:
        lines = handle.readlines()
    return "".join(lines[-max_lines:]).strip()


def _heartbeat_timeout_seconds(status: dict[str, Any]) -> float:
    value = status.get("heartbeat_timeout_seconds")
    if isinstance(value, (int, float)) and value > 0:
        return float(value)
    return 30.0


def _update_managed_status(
    run_dir: Path,
    **fields: Any,
) -> dict[str, Any]:
    status_path = run_dir / "status.json"
    status = _load_json(status_path)
    status.update(fields)
    status["updated_at"] = time.time()
    _dump_json(status_path, status)
    return status


def _pid_is_running(pid: int | None) -> bool:
    if not isinstance(pid, int) or pid <= 0:
        return False
    if os.name == "nt":
        process_query_limited_information = 0x1000
        synchronize = 0x00100000
        still_active = 259

        handle = ctypes.windll.kernel32.OpenProcess(
            process_query_limited_information | synchronize,
            False,
            pid,
        )
        if not handle:
            return False
        try:
            exit_code = ctypes.c_ulong()
            if not ctypes.windll.kernel32.GetExitCodeProcess(
                handle,
                ctypes.byref(exit_code),
            ):
                return False
            return exit_code.value == still_active
        finally:
            ctypes.windll.kernel32.CloseHandle(handle)
    try:
        os.kill(pid, 0)
    except (OSError, SystemError):
        return False
    return True


def _terminate_pid(pid: int) -> None:
    if os.name == "nt":
        subprocess.run(
            ["taskkill", "/PID", str(pid), "/T", "/F"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        return
    os.kill(pid, signal.SIGTERM)


def _build_managed_failure_result(
    run_dir: Path,
    run_id: str,
    *,
    error: str,
    status: dict[str, Any],
) -> dict[str, Any]:
    stdout_tail = _read_text_tail(run_dir / "stdout.log")
    stderr_tail = _read_text_tail(run_dir / "stderr.log")
    result = {
        "ok": False,
        "run_id": run_id,
        "status": "failed",
        "error": error,
        "payment_qr_url": status.get("payment_qr_url"),
        "order_id": status.get("order_id"),
        "order_detail_url": status.get("order_detail_url"),
        "payment_code_url": status.get("payment_code_url"),
        "logs_path": status.get("logs_path"),
        "last_message": status.get("last_message"),
    }
    if stdout_tail:
        result["stdout_tail"] = stdout_tail
    if stderr_tail:
        result["stderr_tail"] = stderr_tail
    return result


def _mark_managed_run_failed(
    run_dir: Path,
    *,
    run_id: str,
    error: str,
) -> dict[str, Any]:
    status = _update_managed_status(
        run_dir,
        status="failed",
        finished_at=time.time(),
        error=error,
        last_message=error,
    )
    result = _build_managed_failure_result(run_dir, run_id, error=error, status=status)
    _dump_json(run_dir / "result.json", result)
    return status


def _reconcile_managed_run(run_dir: Path, status: dict[str, Any]) -> dict[str, Any]:
    current_status = status.get("status")
    if current_status in {
        "succeeded",
        "completed",
        "duplicate_order",
        "failed",
        "cancelled",
    }:
        return status

    pid = status.get("pid")
    if _pid_is_running(pid):
        updated_at = status.get("updated_at")
        heartbeat_timeout = _heartbeat_timeout_seconds(status)
        if (
            isinstance(updated_at, (int, float))
            and time.time() - float(updated_at) > heartbeat_timeout
        ):
            return _mark_managed_run_failed(
                run_dir,
                run_id=status["run_id"],
                error="managed runner heartbeat timed out after {0:.1f}s".format(
                    heartbeat_timeout
                ),
            )
        return status

    result_path = run_dir / "result.json"
    if result_path.exists():
        result = _load_json(result_path)
        terminal_status = result.get("status")
        if terminal_status in {
            "succeeded",
            "completed",
            "duplicate_order",
            "failed",
            "cancelled",
        }:
            fields: dict[str, Any] = {
                "status": terminal_status,
                "finished_at": status.get("finished_at") or time.time(),
            }
            if result.get("payment_qr_url"):
                fields["payment_qr_url"] = result["payment_qr_url"]
            if result.get("order_id") not in (None, ""):
                fields["order_id"] = result["order_id"]
            if result.get("order_detail_url"):
                fields["order_detail_url"] = result["order_detail_url"]
            if result.get("payment_code_url"):
                fields["payment_code_url"] = result["payment_code_url"]
            if result.get("last_message"):
                fields["last_message"] = result["last_message"]
            if terminal_status == "failed" and result.get("error"):
                fields["error"] = result["error"]
            return _update_managed_status(run_dir, **fields)

    error = "managed runner exited before writing final status"
    stderr_tail = _read_text_tail(run_dir / "stderr.log")
    stdout_tail = _read_text_tail(run_dir / "stdout.log")
    if stderr_tail:
        error = "{0}; stderr: {1}".format(error, stderr_tail)
    elif stdout_tail:
        error = "{0}; stdout: {1}".format(error, stdout_tail)
    return _mark_managed_run_failed(run_dir, run_id=status["run_id"], error=error)


def _append_log(task_id: str, message: str) -> None:
    with _TASKS_LOCK:
        record = _TASKS[task_id]
        record.logs.append(message)
        if len(record.logs) > 200:
            record.logs = record.logs[-200:]


def _update_task(task_id: str, **fields: Any) -> None:
    with _TASKS_LOCK:
        record = _TASKS[task_id]
        for key, value in fields.items():
            setattr(record, key, value)


def _run_buy_task(
    task_id: str,
    config: dict[str, Any],
    runtime_options: RuntimeOptions,
) -> None:
    from app_cmd.config.BuyConfig import BuyConfig
    from task.buy import Buy

    buy_job = Buy(
        config=BuyConfig.from_runtime_options(
            json.dumps(config, ensure_ascii=False),
            runtime_options,
        ),
    )
    _update_task(task_id, status="running", started_at=time.time())
    succeeded = False
    try:
        for event in buy_job.stream():
            message = event.message
            if message is None:
                continue
            _append_log(task_id, message)
            if "抢票成功" in message:
                succeeded = True
            payment_fields = _collect_payment_fields_from_event(event)
            if payment_fields:
                _update_task(task_id, **payment_fields)

        _update_task(
            task_id,
            status="succeeded" if succeeded else "completed",
            finished_at=time.time(),
        )
    except Exception as exc:
        _append_log(task_id, "task exception: {0!r}".format(exc))
        _update_task(
            task_id,
            status="failed",
            finished_at=time.time(),
            error=repr(exc),
        )


def start_buy(
    config_or_path: str | Path | dict[str, Any],
    *,
    runtime_options: dict[str, Any] | RuntimeOptions | None = None,
) -> dict[str, Any]:
    validation = validate_config(config_or_path)
    if not validation.ok:
        return {"ok": False, "validation": validation.to_dict(), "task": None}

    assert validation.normalized_config is not None
    runtime = build_runtime_options()
    runtime = runtime.merged_with(runtime_options)

    task_id = uuid.uuid4().hex
    record = BuyTaskRecord(
        task_id=task_id,
        status="pending",
        detail=validation.normalized_config.get("detail", "unknown-task"),
        created_at=time.time(),
    )
    with _TASKS_LOCK:
        _TASKS[task_id] = record

    thread = threading.Thread(
        target=_run_buy_task,
        args=(task_id, validation.normalized_config, runtime),
        daemon=True,
        name="biliTickerBuy-task-{0}".format(task_id[:8]),
    )
    thread.start()

    return {
        "ok": True,
        "validation": validation.to_dict(),
        "task": record.to_dict(),
    }


def task_status(task_id: str) -> dict[str, Any]:
    with _TASKS_LOCK:
        record = _TASKS.get(task_id)
        if record is None:
            return {"ok": False, "error": "task not found", "task_id": task_id}
        return {"ok": True, "task": record.to_dict()}


def run_buy_sync(
    config_or_path: str | Path | dict[str, Any],
    *,
    runtime_options: dict[str, Any] | RuntimeOptions | None = None,
) -> dict[str, Any]:
    from app_cmd.config.BuyConfig import BuyConfig
    from task.buy import Buy

    validation = validate_config(config_or_path)
    if not validation.ok:
        return {
            "ok": False,
            "validation": validation.to_dict(),
            "logs": [],
            "payment_qr_url": None,
        }

    assert validation.normalized_config is not None
    runtime = build_runtime_options()
    runtime = runtime.merged_with(runtime_options)

    buy_job = Buy(
        config=BuyConfig.from_runtime_options(
            json.dumps(validation.normalized_config, ensure_ascii=False),
            runtime,
        ),
    )

    logs: list[str] = []
    payment_qr_url: str | None = None
    order_id: int | str | None = None
    order_detail_url: str | None = None
    payment_code_url: str | None = None
    succeeded = False
    for event in buy_job.stream():
        message = event.message
        if message is None:
            continue
        logs.append(message)
        if "抢票成功" in message:
            succeeded = True
        payment_fields = _collect_payment_fields_from_event(event)
        if "payment_qr_url" in payment_fields:
            payment_qr_url = payment_fields["payment_qr_url"]
        if "order_id" in payment_fields:
            order_id = payment_fields["order_id"]
        if "order_detail_url" in payment_fields:
            order_detail_url = payment_fields["order_detail_url"]
        if "payment_code_url" in payment_fields:
            payment_code_url = payment_fields["payment_code_url"]

    return {
        "ok": True,
        "validation": validation.to_dict(),
        "status": "succeeded" if succeeded else "completed",
        "logs": logs,
        "payment_qr_url": payment_qr_url,
        "order_id": order_id,
        "order_detail_url": order_detail_url,
        "payment_code_url": payment_code_url,
    }


def start_managed_buy(
    config_or_path: str | Path | dict[str, Any],
    *,
    runtime_options: dict[str, Any] | RuntimeOptions | None = None,
    run_id: str | None = None,
    runs_root: str | Path | None = None,
) -> dict[str, Any]:
    validation = validate_config(config_or_path)
    if not validation.ok:
        return {"ok": False, "validation": validation.to_dict(), "run": None}

    assert validation.normalized_config is not None
    runtime = build_runtime_options(show_qrcode=False)
    runtime = runtime.merged_with(runtime_options)

    managed_root = _managed_runs_root(runs_root)
    assigned_run_id = run_id or uuid.uuid4().hex
    run_dir = managed_root / assigned_run_id
    if run_dir.exists():
        return {
            "ok": False,
            "validation": validation.to_dict(),
            "run": None,
            "error": "run_id already exists",
            "run_id": assigned_run_id,
        }
    run_dir.mkdir(parents=True, exist_ok=False)

    run_metadata = {
        "run_id": assigned_run_id,
        "created_at": time.time(),
    }
    _dump_json(run_dir / "run.json", run_metadata)
    _dump_json(run_dir / "config.json", validation.normalized_config)
    _dump_json(run_dir / "runtime.json", runtime.to_dict())

    status = {
        "ok": True,
        "run_id": assigned_run_id,
        "status": "pending",
        "detail": validation.normalized_config.get("detail", assigned_run_id),
        "pid": None,
        "created_at": run_metadata["created_at"],
        "started_at": None,
        "updated_at": run_metadata["created_at"],
        "finished_at": None,
        "payment_qr_url": None,
        "order_id": None,
        "order_detail_url": None,
        "payment_code_url": None,
        "error": None,
        "last_message": None,
        "heartbeat_timeout_seconds": max(float(runtime.interval) / 1000.0 * 20.0, 30.0),
        "logs_path": str(run_dir / "events.log"),
        "result_path": str(run_dir / "result.json"),
        "config_path": str(run_dir / "config.json"),
        "runtime_path": str(run_dir / "runtime.json"),
    }
    _dump_json(run_dir / "status.json", status)

    runner_path = _package_root() / "interface" / "managed_runner.py"
    command = [sys.executable, str(runner_path), str(run_dir)]
    env = os.environ.copy()
    env.setdefault("BTB_APP_LOG_NAME", "app-{0}.log".format(assigned_run_id))

    creationflags = 0
    startupinfo = None
    if os.name == "nt":
        creationflags = (
            getattr(subprocess, "DETACHED_PROCESS", 0)
            | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
            | getattr(subprocess, "CREATE_NO_WINDOW", 0)
        )
        startupinfo = subprocess.STARTUPINFO()
        startupinfo.dwFlags |= getattr(subprocess, "STARTF_USESHOWWINDOW", 0)
        startupinfo.wShowWindow = getattr(subprocess, "SW_HIDE", 0)

    stdout_handle = open(run_dir / "stdout.log", "a", encoding="utf-8")
    stderr_handle = open(run_dir / "stderr.log", "a", encoding="utf-8")
    try:
        process = subprocess.Popen(
            command,
            cwd=str(_package_root()),
            stdout=stdout_handle,
            stderr=stderr_handle,
            env=env,
            creationflags=creationflags,
            startupinfo=startupinfo,
        )
    finally:
        stdout_handle.close()
        stderr_handle.close()

    status["pid"] = process.pid
    status["updated_at"] = time.time()
    _dump_json(run_dir / "status.json", status)

    time.sleep(0.2)
    returncode = process.poll()
    if returncode is not None:
        failed_status = _mark_managed_run_failed(
            run_dir,
            run_id=assigned_run_id,
            error="managed runner exited immediately with code {0}".format(returncode),
        )
        return {
            "ok": False,
            "validation": validation.to_dict(),
            "run": failed_status,
            "error": failed_status.get("error"),
        }

    return {
        "ok": True,
        "validation": validation.to_dict(),
        "run": {
            "run_id": assigned_run_id,
            "run_dir": str(run_dir),
            "status_path": str(run_dir / "status.json"),
            "result_path": str(run_dir / "result.json"),
            "logs_path": str(run_dir / "events.log"),
            "pid": process.pid,
        },
    }


def managed_task_status(
    run_id: str,
    *,
    runs_root: str | Path | None = None,
) -> dict[str, Any]:
    run_dir = _managed_runs_root(runs_root) / run_id
    status_path = run_dir / "status.json"
    if not status_path.exists():
        return {"ok": False, "error": "managed run not found", "run_id": run_id}
    status = _reconcile_managed_run(run_dir, _load_json(status_path))
    result_path = run_dir / "result.json"
    if result_path.exists():
        status["result"] = _load_json(result_path)
    return {"ok": True, "run": status}


def cancel_managed_buy(
    run_id: str,
    *,
    runs_root: str | Path | None = None,
) -> dict[str, Any]:
    run_dir = _managed_runs_root(runs_root) / run_id
    status_path = run_dir / "status.json"
    if not status_path.exists():
        return {"ok": False, "error": "managed run not found", "run_id": run_id}

    status = _load_json(status_path)
    current_status = status.get("status")
    if current_status in {
        "succeeded",
        "completed",
        "duplicate_order",
        "failed",
        "cancelled",
    }:
        return {
            "ok": True,
            "run": status,
            "cancelled": current_status == "cancelled",
            "message": "run already finished",
        }

    pid = status.get("pid")
    process_was_running = _pid_is_running(pid)
    if process_was_running:
        assert pid is not None
        _terminate_pid(pid)

    updated_status = _update_managed_status(
        run_dir,
        status="cancelled",
        finished_at=time.time(),
        error=None,
        last_message="cancelled by API",
    )

    result_path = run_dir / "result.json"
    _dump_json(
        result_path,
        {
            "ok": False,
            "run_id": run_id,
            "status": "cancelled",
            "payment_qr_url": updated_status.get("payment_qr_url"),
            "order_id": updated_status.get("order_id"),
            "order_detail_url": updated_status.get("order_detail_url"),
            "payment_code_url": updated_status.get("payment_code_url"),
            "logs_path": updated_status.get("logs_path"),
            "last_message": "cancelled by API",
        },
    )
    return {
        "ok": True,
        "run": updated_status,
        "cancelled": True,
        "process_was_running": process_was_running,
    }


def delete_managed_buy(
    run_id: str,
    *,
    runs_root: str | Path | None = None,
    force: bool = False,
) -> dict[str, Any]:
    managed_root = _managed_runs_root(runs_root)
    run_dir = managed_root / run_id
    if not run_dir.exists():
        return {"ok": False, "error": "managed run not found", "run_id": run_id}

    try:
        run_dir.relative_to(managed_root)
    except ValueError:
        return {"ok": False, "error": "run dir escapes managed root", "run_id": run_id}

    status_path = run_dir / "status.json"
    status = _load_json(status_path) if status_path.exists() else {}
    pid = status.get("pid")
    process_is_running = _pid_is_running(pid)

    if process_is_running and not force:
        return {
            "ok": False,
            "error": "managed run is still running; cancel it first or pass force=True",
            "run_id": run_id,
            "run": status,
        }

    cancelled = None
    if process_is_running and force:
        cancelled = cancel_managed_buy(run_id, runs_root=runs_root)

    shutil.rmtree(run_dir)
    return {
        "ok": True,
        "run_id": run_id,
        "deleted": True,
        "force": force,
        "cancelled": cancelled,
    }