from __future__ import annotations

import base64
import json
import os
from pathlib import Path
import re
import subprocess
import time
import threading
from copy import deepcopy

from .errors import MiniError, ObservationUnavailable, require


DEFAULT_CLI = Path(r"C:\Nexon\MabinogiMobile\MabinogiMobile_CLI.exe")
READS = {
    "status", "capabilities", "get_items", "get_inventory", "get_activity", "get_quests",
    "get_currencies", "get_gatherable_items", "get_alterable_items",
    "get_altering_works", "get_craftable_items", "get_my_info",
    "get_instruments", "get_music_scores",
}
_READ_LOCK = threading.RLock()
_QUEUED_READS_LOCK = threading.Lock()
_QUEUED_READS = {}
QUICK_ACTIONS = {"execute_gathering", "complete_altering_work"}
CHAT_ACTIONS = {"write_chat"}
MUSIC_ACTIONS = {"change_instrument", "play_music_score"}


def _argv(path: Path, command: str, body=None):
    require(command in READS | QUICK_ACTIONS | CHAT_ACTIONS | MUSIC_ACTIONS | {"stop_action"}, "허용되지 않은 미니 커넥터 명령")
    args = [str(path), command]
    if body is not None:
        # Official name-filter reads take raw UTF-8 text, not a JSON-quoted
        # string (same ABI as main game_connector.argv).
        raw = body if isinstance(body, str) else json.dumps(body, ensure_ascii=False, separators=(",", ":"))
        args.append("base64:" + base64.b64encode(raw.encode("utf-8")).decode("ascii"))
    return args


def _decode(code, stdout, *, allow_error=False):
    try:
        value = json.loads(stdout.lstrip("\ufeff"))
    except (TypeError, ValueError) as exc:
        raise MiniError("커넥터 응답을 해석하지 못했습니다.") from exc
    if not isinstance(value, (dict, list)):
        raise MiniError("커넥터 응답 형식이 변경되었습니다.")
    if isinstance(value, dict) and "status" in value:
        status = value["status"]
        body = value.get("body", {})
        if isinstance(body, str):
            try:
                body = json.loads(body)
            except ValueError:
                body = {"message": body}
        if status != "accepted":
            if allow_error and isinstance(body, dict):
                return body | {"status": status}
            message = body.get("message") if isinstance(body, dict) else str(body)
            raise MiniError(f"커넥터가 명령을 실행하지 않았습니다: {message or status}")
        value = body
    if code:
        if allow_error and isinstance(value, dict):
            return value
        raise MiniError(f"커넥터 실행 실패: {value}")
    if isinstance(value, dict) and value.get("error"):
        if allow_error:
            return value
        raise MiniError(f"커넥터 오류: {value}")
    return value


class Connector:
    """Official CLI boundary.

    Paid calls are not exposed through the automation service. ``quick_gather``
    is deliberately named and only used by the user-click Quick tab.
    """

    def __init__(self, path: str | Path = DEFAULT_CLI):
        self.path = Path(path)
        self.commands = {}
        self.last_read = {}
        self._read_binding = threading.local()

    def bind_observation(self, check, status=None):
        """Bind only the current automation worker, never UI/background reads."""
        self._read_binding.check = check
        self._read_binding.status = status or (lambda _: None)
        return self

    def _read_check(self):
        return getattr(getattr(self, "_read_binding", None), "check", None)

    def bind_input_guard(self, guard):
        """The owning worker's fresh input permission; never applied to reads."""
        self._read_binding.input_guard = guard
        return self

    def _input_guard(self):
        guard = getattr(self._read_binding, "input_guard", None)
        if callable(guard):
            guard()

    def _read_pause(self, seconds):
        check = self._read_check()
        if not callable(check):
            time.sleep(seconds)
            return
        until = time.monotonic() + max(0, seconds)
        while True:
            check()
            remaining = until - time.monotonic()
            if remaining <= 0:
                return
            time.sleep(min(.1, remaining))

    def _run_observation(self, args, timeout):
        """An in-flight read is cancellable; writes never use this function."""
        check = self._read_check()
        check()
        process = subprocess.Popen(args, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
            stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="strict",
            creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0)
        deadline = time.monotonic() + timeout
        try:
            while True:
                check()
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise subprocess.TimeoutExpired(args, timeout)
                try:
                    stdout, stderr = process.communicate(timeout=min(.2, remaining))
                    check()
                    return subprocess.CompletedProcess(args, process.returncode, stdout, stderr)
                except subprocess.TimeoutExpired:
                    continue
        finally:
            if process.poll() is None:
                process.terminate()
                try:
                    process.wait(timeout=1)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait(timeout=1)
            for stream in (process.stdout, process.stderr):
                if stream is not None:
                    stream.close()

    def call(self, command, body=None, timeout=25, *, allow_error=False):
        # Background tracking must not overlap other short read requests. Do not
        # hold this lock around long-running Quick actions or their completion.
        if command in READS:
            return self._queued_read(command, body, timeout, allow_error=allow_error)
        return self._call(command, body, timeout, allow_error=allow_error)

    def _queued_read(self, command, body, timeout, *, allow_error):
        """Coalesce only identical reads requested BEFORE a CLI launch.

        An already-started or completed read is never reused. In particular an
        after-inventory identity read cannot consume a before-inventory result.
        There is no TTL response cache, persistent protocol or write replay.
        """
        args = _argv(self.path, command, body)
        key = (os.path.normcase(os.path.abspath(args[0])), tuple(args[1:]), timeout, allow_error)
        check = self._read_check()
        while True:
            if callable(check):
                check()
            with _QUEUED_READS_LOCK:
                flight = _QUEUED_READS.get(key)
                owner = flight is None or flight["started"]
                if owner:
                    flight = {"started": False, "event": threading.Event(), "waiters": 0}
                    _QUEUED_READS[key] = flight
                else:
                    flight["waiters"] += 1
            if not owner:
                while not flight["event"].wait(.1):
                    if callable(check):
                        check()
                if callable(check):
                    check()
                if "value" in flight:
                    return deepcopy(flight["value"])
                # Another worker's cancellation/failure is not this caller's
                # cancellation, nor evidence for a game state. Request anew.
                continue
            acquired = False
            try:
                while not _READ_LOCK.acquire(timeout=.1):
                    if callable(check):
                        check()
                acquired = True
                if callable(check):
                    check()
                with _QUEUED_READS_LOCK:
                    flight["started"] = True
                value = self._call(command, body, timeout, allow_error=allow_error)
                if flight["waiters"]:
                    flight["value"] = deepcopy(value)
                return value
            finally:
                if acquired:
                    _READ_LOCK.release()
                with _QUEUED_READS_LOCK:
                    if _QUEUED_READS.get(key) is flight:
                        del _QUEUED_READS[key]
                flight["event"].set()

    def _call(self, command, body=None, timeout=25, *, allow_error=False):
        require(self.path.is_file(), "공식 MabinogiMobile_CLI.exe를 찾지 못했습니다.")
        if command not in READS:
            self._input_guard()
        try:
            if command in READS and callable(self._read_check()):
                result = self._run_observation(_argv(self.path, command, body), timeout)
            else:
                result = subprocess.run(
                    _argv(self.path, command, body),
                    stdin=subprocess.DEVNULL,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    encoding="utf-8",
                    errors="strict",
                    timeout=timeout,
                    creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
                    check=False,
                )
        except subprocess.TimeoutExpired as exc:
            error = ObservationUnavailable if command in READS else MiniError
            raise error("커넥터 응답 시간이 초과되었습니다. 행동 명령은 자동 재전송하지 않습니다.") from exc
        except (OSError, UnicodeError) as exc:
            error = ObservationUnavailable if command in READS else MiniError
            raise error(f"커넥터 실행/응답 읽기 실패: {exc}") from exc
        # Chat must not misreport success when an accepted-looking body is
        # accompanied by a failed CLI exit. Keep other command semantics intact.
        try:
            return _decode(result.returncode, result.stdout,
                           allow_error=allow_error and not (command == "write_chat" and result.returncode))
        except MiniError as exc:
            if command in READS:
                raise ObservationUnavailable(str(exc)) from exc
            raise

    @staticmethod
    def _validate_observation(command, value):
        """Response structure only; no exact quantities or action approvals."""
        valid = True
        if command == "status":
            valid = isinstance(value, dict) and value.get("pipe") == "connected"
        elif command == "capabilities":
            valid = (isinstance(value, dict) and isinstance(value.get("commands"), list)
                     and all(isinstance(row, dict) and isinstance(row.get("Command"), str)
                             for row in value["commands"]))
        elif command == "get_activity":
            valid = (isinstance(value, dict) and all(isinstance(value[key], dict)
                     for key in ("Dungeon", "combatState", "autoPlay", "autoTravel", "dialogue", "Mode")
                     if key in value))
        elif command in ("get_items", "get_gatherable_items"):
            rows = value.get("items") if isinstance(value, dict) else value
            valid = isinstance(rows, list)
            if valid and command == "get_items":
                # Every stock consumer first selects by identity and location.
                # Dropping an incomplete row here would turn unknown stock into
                # a false shortage. Do not invent counts, locks or locations.
                valid = all(isinstance(row, dict) and all(
                    isinstance(row.get(key), str) and row[key].strip()
                    for key in ("DisplayName", "Location")) for row in rows)
        elif command == "get_altering_works":
            valid = isinstance(value, dict) and isinstance(value.get("works"), list)
        elif command == "get_craftable_items":
            valid = isinstance(value, dict) and isinstance(value.get("items"), list)
        elif command in ("get_currencies", "get_instruments", "get_music_scores"):
            valid = isinstance(value, list)
        if not valid:
            raise ObservationUnavailable("커넥터 조회 응답을 아직 확인하지 못했습니다: " + command)

    def read(self, command, body=None, *, recovery_seconds=180):
        require(command in READS, "조회 전용 명령만 재확인할 수 있습니다.")
        interval = 3.0 if command in ("get_items", "get_inventory") else 1.0 if command == "get_activity" else 0.0
        wait = interval - (time.monotonic() - self.last_read.get(command, -1e9))
        if wait > 0:
            self._read_pause(wait)
        started = time.monotonic()
        delay = 2.0
        announced = False
        while True:
            check = self._read_check()
            bound = callable(check)
            if bound:
                check()
            try:
                value = self.call(command, body)
                if bound:
                    self._validate_observation(command, value)
                self.last_read[command] = time.monotonic()
                return value
            except MiniError as exc:
                if bound and not isinstance(exc, ObservationUnavailable):
                    raise  # Closed client/cancel/configuration is not a transient read.
                if not bound and time.monotonic() - started >= recovery_seconds:
                    raise
                if bound and not announced:
                    self._read_binding.status("커넥터 조회 재확인 · " + command + " · 입력 없이 같은 작업 유지")
                    announced = True
                self._read_pause(delay)
                delay = min(15.0, delay * 1.7)

    def connect(self):
        status = self.read("status", recovery_seconds=0)
        require(isinstance(status, dict) and status.get("pipe") == "connected", "게임 AI 커넥터가 연결되지 않았습니다.")
        catalog = self.read("capabilities", recovery_seconds=0)
        rows = catalog.get("commands") if isinstance(catalog, dict) else None
        require(isinstance(rows, list), "게임 명령 목록을 읽지 못했습니다.")
        self.commands = {row["Command"]: row for row in rows if isinstance(row, dict) and isinstance(row.get("Command"), str)}
        return sorted(self.commands)

    def require_command(self, command):
        if not self.commands:
            self.connect()
        require(command in self.commands, f"현재 게임에서 지원하지 않는 명령: {command}")
        return self.commands[command]

    def inventory(self):
        self.require_command("get_items")
        value = self.read("get_items")
        rows = value.get("items") if isinstance(value, dict) else value
        require(isinstance(rows, list), "가방 재고 응답 형식이 변경되었습니다.")
        return [row for row in rows if isinstance(row, dict)]

    def item_quantity(self, display_name):
        """Exact bag-only quantity; storage and locked status are not inferred."""
        require(isinstance(display_name, str) and display_name.strip(), "수량 확인 품목명이 필요합니다.")
        announced = False
        while True:
            check = self._read_check()
            if callable(check):
                check()
            matches = [row for row in self.inventory()
                       if row.get("DisplayName") == display_name and row.get("Location") == "inventory"]
            categories = [row.get("Category") for row in matches]
            counts = [row.get("Count") for row in matches]
            # A list-shaped response can still contain an unfinished target
            # row. Never replace that uncertainty with zero or restart a task.
            ambiguous = bool(categories and any(value != categories[0] for value in categories[1:]))
            if not ambiguous and all(type(value) is int and value >= 0 for value in counts):
                return sum(counts)
            message = ("동일 이름의 서로 다른 품목 분류가 있습니다: " if ambiguous else
                       "수량 확인 품목의 재고 형식이 변경되었습니다: ") + display_name
            if not callable(check):
                raise MiniError(message)  # Explicit UI/background queries remain bounded.
            if not announced:
                self._read_binding.status("품목 수량 재조회 · 확인된 진행 유지 · " + message)
                announced = True
            self._read_pause(2.0)

    def gatherable_items(self):
        self.require_command("get_gatherable_items")
        value = self.read("get_gatherable_items")
        rows = value.get("items") if isinstance(value, dict) else value
        require(isinstance(rows, list), "채집 가능 목록 형식이 변경되었습니다.")
        names = []
        for row in rows:
            name = row.get("DisplayName") if isinstance(row, dict) else row
            if isinstance(name, str) and name.strip():
                names.append(name.strip())
        return list(dict.fromkeys(names))

    def altering_works(self):
        self.require_command("get_altering_works")
        value = self.read("get_altering_works")
        require(isinstance(value, dict) and isinstance(value.get("works"), list), "가공 작업 목록 형식이 변경되었습니다.")
        return value

    def _music_rows(self, command, name_key, name_filter=None):
        require(name_filter is None or isinstance(name_filter, str), "악기·악보 검색어 형식을 확인해 주세요.")
        self.require_command(command)
        rows = self.read(command, name_filter, recovery_seconds=0)
        require(isinstance(rows, list) and all(
            isinstance(row, dict) and isinstance(row.get(name_key), str) and row[name_key].strip()
            for row in rows), "악기·악보 목록 응답 형식이 변경되었습니다.")
        # Keep duplicate names and official fields: the CLI exposes no row ID.
        return rows

    def instruments(self, name_filter=None):
        return self._music_rows("get_instruments", "Name", name_filter)

    def music_scores(self, name_filter=None):
        return self._music_rows("get_music_scores", "DisplayTitle", name_filter)

    def music_performance(self):
        """Official McpActivity.Performance; absent data is never an ended song."""
        self.require_command("get_activity")
        value = self.read("get_activity", recovery_seconds=0)
        performance = value.get("Performance") if isinstance(value, dict) else None
        if not isinstance(performance, dict) or type(performance.get("IsPlaying")) is not bool:
            raise ObservationUnavailable("현재 악기연주 상태를 아직 확인하지 못했습니다.")
        return performance

    def music_action(self, kind, name, *, before_send=None):
        """One explicit name-based request; fresh ownership check, no replay."""
        require(kind in ("instrument", "score"), "악기 교체 또는 악보 연주를 선택해 주세요.")
        require(isinstance(name, str) and name.strip(), "악기·악보의 정확한 이름을 선택해 주세요.")
        command = "change_instrument" if kind == "instrument" else "play_music_score"
        self.require_command(command)
        rows = self.instruments() if kind == "instrument" else self.music_scores()
        key = "Name" if kind == "instrument" else "DisplayTitle"
        require(any(row[key] == name for row in rows), "현재 보유 목록에 같은 이름이 없습니다. 목록을 새로 조회해 주세요.")
        if before_send:
            before_send()
        # Do not use read/recovery for an action, even after timeout or rejection.
        return self.call(command, {"name" if kind == "instrument" else "title": name}, timeout=25)

    def _wing_balance(self):
        self.require_command("get_currencies")
        rows = self.read("get_currencies")
        require(isinstance(rows, list), "재화 목록 형식이 변경되었습니다.")
        matches = [r for r in rows if isinstance(r, dict) and r.get("DisplayName") == "정령의 날개"]
        require(len(matches) == 1 and isinstance(matches[0].get("Amount"), int), "정령의 날개 수량을 확인하지 못했습니다.")
        return matches[0]["Amount"]

    def _dispatch_quick(self, command, body):
        """Launch once; Quick actions do not interpret or act on CLI outcomes.

        DEVNULL avoids pipes filling during a long action. The daemon only reaps
        the child process: no parsing, callbacks, polling the game or retrying.
        A successful launch is not proof of game acceptance or completion.
        """
        require(command in QUICK_ACTIONS, "퀵버튼 행동 명령만 전달할 수 있습니다.")
        require(self.path.is_file(), "공식 MabinogiMobile_CLI.exe를 찾지 못했습니다.")
        self._input_guard()
        try:
            process = subprocess.Popen(
                _argv(self.path, command, body),
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
            )
        except OSError as exc:
            raise MiniError(f"CLI를 실행하지 못해 퀵명령을 전달하지 않았습니다: {exc}") from exc
        threading.Thread(target=process.wait, name="mini-quick-process-reaper", daemon=True).start()

    def quick_gather(self, display_name, *, before_send=None):
        """One explicit Quick-button press, one paid request, never retried."""
        spec = self.require_command("execute_gathering")
        note = str(spec.get("Note", ""))
        cost = re.findall(r"Running this command consumes (\d+) 정령의 날개\.", note)
        require(cost == ["5"], "채집 명령의 날개 비용 명세가 변경되어 전송하지 않았습니다.")
        require(self._wing_balance() >= 5, "정령의 날개가 5개보다 적어 전송하지 않았습니다.")
        if before_send:
            before_send()
        self._dispatch_quick("execute_gathering", {"displayName": display_name})

    def stop_action(self):
        self.require_command("stop_action")
        # The only retryable result is the explicit no-effect invalid_state.
        # Preserve its structured code for the caller's state-gated retry.
        return self.call("stop_action", timeout=30, allow_error=True)

    def write_chat(self, text, *, confirmed=False, before_send=None):
        """One explicitly confirmed ordinary message, no retry or channel command.

        Official capabilities: raw string, max50 characters, requiresConfirm.
        The caller must retain uncertain outcomes; rate_limited is not replayed.
        This is intentionally separate from paid fire-and-forget Quick actions.
        """
        from .chat_messages import validate_chat_text
        require(confirmed is True, "멘트 내용을 확인한 뒤 전송해 주세요.")
        text = validate_chat_text(text)
        self.require_command("write_chat")
        if before_send:
            before_send()
        return self.call("write_chat", text, timeout=15, allow_error=True)

    def quick_collect_facility(self, group_names, *, before_send=None):
        """Collect all completed rows at one facility using a completed representative."""
        self.require_command("complete_altering_work")
        works = self.altering_works()["works"]
        matches = [r for r in works if isinstance(r, dict) and r.get("DisplayName") in group_names]
        facilities = {r.get("FacilityName") for r in matches if r.get("FacilityName")}
        require(len(facilities) <= 1, "같은 분야의 가공 시설이 여러 개로 응답되었습니다.")
        require(facilities, "해당 가공 시설에 등록된 작업이 없습니다.")
        facility = next(iter(facilities))
        completed = [r for r in works if r.get("FacilityName") == facility and r.get("IsCompleted") is True]
        require(completed, "해당 시설에 받을 완료품이 없습니다.")
        if before_send:
            before_send()
        self._dispatch_quick("complete_altering_work", {"displayName": completed[0]["DisplayName"]})
