"""PTYSession - 伪终端会话，支持交互式命令

基于 asyncio + os.openpty() 实现伪终端，支持交互式命令和自动应答。
不依赖 pexpect，仅使用标准库。
"""

from __future__ import annotations

import asyncio
import fcntl
import logging
import os
import struct
import termios
import time
from dataclasses import dataclass

logger = logging.getLogger(__name__)

# 自动应答规则：(prompt_pattern, response)
# WARNING: auto_respond is disabled by default for safety.
# Enable it only when you explicitly want automatic yes/confirm responses.
_AUTO_RESPOND_RULES: list[tuple[str, str]] = [
    (r"\[y/N\]\s*$", "y"),
    (r"\[Y/n\]\s*$", "y"),
    (r"\[yes/no\]\s*$", "yes"),
    (r"\(yes/no\)\s*$", "yes"),
    (r"\(yes/no/\[fingerprint\]\)\s*$", "yes"),
    (r"continue\?\s*$", "y"),
    (r"are you sure\?\s*$", "y"),
    (r"password:\s*$", ""),  # 密码提示不自动应答，需要人工介入
    (r"passphrase:\s*$", ""),
]


@dataclass
class PTYOutput:
    """PTY 输出结果

    Attributes:
        output: 输出文本
        exit_code: 退出码（-1 表示超时或未结束）
        timed_out: 是否超时
    """

    output: str
    exit_code: int = -1
    timed_out: bool = False


class PTYSession:
    """伪终端会话 - 支持交互式命令

    使用 os.openpty() 创建伪终端对，通过 asyncio 异步读写。
    支持自动检测提示并应答（如 yes/no 确认）。

    Usage:
        pty = PTYSession()
        await pty.start()
        output = await pty.run_command("ssh-keygen", timeout=10)
        await pty.close()
    """

    def __init__(
        self,
        auto_respond: bool = False,
        custom_rules: list[tuple[str, str]] | None = None,
        default_timeout: float = 30.0,
        buffer_size: int = 4096,
    ):
        """初始化 PTY 会话

        Args:
            auto_respond: 是否自动应答已知提示
            custom_rules: 自定义应答规则列表 [(prompt_pattern, response)]
            default_timeout: 默认超时时间（秒）
            buffer_size: 读取缓冲区大小
        """
        self._auto_respond = auto_respond
        self._respond_rules = list(_AUTO_RESPOND_RULES)
        if custom_rules:
            self._respond_rules.extend(custom_rules)
        self._default_timeout = default_timeout
        self._buffer_size = buffer_size

        self._master_fd: int | None = None
        self._slave_fd: int | None = None
        self._process: asyncio.subprocess.Process | None = None
        self._running = False
        self._output_buffer = ""

    @property
    def is_running(self) -> bool:
        """PTY 会话是否已启动（伪终端已创建）"""
        return self._running

    async def start(self) -> None:
        """启动 PTY 会话（创建伪终端对）

        在执行命令前调用，创建 master/slave 文件描述符。
        """
        if self._running:
            return

        self._master_fd, self._slave_fd = os.openpty()

        # 设置终端大小
        try:
            winsize = struct.pack("HHHH", 24, 80, 0, 0)
            fcntl.ioctl(self._slave_fd, termios.TIOCSWINSZ, winsize)
        except Exception:
            pass

        # 设置 master fd 为非阻塞
        flags = fcntl.fcntl(self._master_fd, fcntl.F_GETFL)
        fcntl.fcntl(self._master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)

        self._running = True
        logger.debug("PTY session started: master=%d slave=%d", self._master_fd, self._slave_fd)

    async def run_command(
        self,
        command: str,
        timeout: float | None = None,
        cwd: str | None = None,
        env: dict[str, str] | None = None,
    ) -> PTYOutput:
        """在 PTY 中运行命令，等待完成

        Args:
            command: 要执行的命令
            timeout: 超时时间（秒）
            cwd: 工作目录
            env: 环境变量

        Returns:
            PTYOutput 输出结果
        """
        if not self._running:
            await self.start()

        timeout = timeout or self._default_timeout
        self._output_buffer = ""

        # 构建环境
        cmd_env = dict(os.environ)
        if env:
            cmd_env.update(env)

        # 启动子进程，使用 slave 作为 stdin/stdout/stderr
        self._process = await asyncio.create_subprocess_shell(
            command,
            stdin=self._slave_fd,
            stdout=self._slave_fd,
            stderr=self._slave_fd,
            cwd=cwd,
            env=cmd_env,
            start_new_session=True,
        )

        # 关闭 slave 端（子进程已继承）
        os.close(self._slave_fd)
        self._slave_fd = None

        # 异步读取输出
        try:
            exit_code = await self._read_until_exit(timeout)
        except asyncio.TimeoutError:
            if self._process and self._process.returncode is None:
                self._process.kill()
                await self._process.wait()
            self._output_buffer += "\n[PTY 命令执行超时]"
            return PTYOutput(
                output=self._output_buffer,
                exit_code=-1,
                timed_out=True,
            )

        return PTYOutput(
            output=self._output_buffer,
            exit_code=exit_code,
            timed_out=False,
        )

    async def send(self, line: str) -> None:
        """向 PTY 发送一行输入

        Args:
            line: 要发送的文本（自动追加换行符）
        """
        if self._master_fd is None:
            return

        data = (line + "\n").encode("utf-8")
        try:
            os.write(self._master_fd, data)
        except OSError as e:
            logger.warning("PTY write failed: %s", e)

    async def read_output(self, timeout: float = 1.0) -> str:
        """读取当前可用的 PTY 输出

        Args:
            timeout: 读取超时（秒）

        Returns:
            读取到的输出文本
        """
        if self._master_fd is None:
            return ""

        output = ""
        deadline = time.monotonic() + timeout

        while time.monotonic() < deadline:
            try:
                chunk = os.read(self._master_fd, self._buffer_size)
                if chunk:
                    text = chunk.decode("utf-8", errors="replace")
                    output += text
                    self._output_buffer += text

                    # 自动应答
                    if self._auto_respond:
                        await self._try_auto_respond(text)
            except (BlockingIOError, OSError):
                # 没有数据可读
                await asyncio.sleep(0.05)
                continue

            if output:
                # 读取到数据后短暂等待看是否还有更多
                await asyncio.sleep(0.05)

        return output

    async def close(self) -> None:
        """关闭 PTY 会话，清理资源"""
        self._running = False

        if self._process is not None and self._process.returncode is None:
            try:
                self._process.terminate()
                await asyncio.wait_for(self._process.wait(), timeout=5.0)
            except (asyncio.TimeoutError, ProcessLookupError):
                try:
                    self._process.kill()
                except ProcessLookupError:
                    pass

        if self._master_fd is not None:
            try:
                os.close(self._master_fd)
            except OSError:
                pass
            self._master_fd = None

        if self._slave_fd is not None:
            try:
                os.close(self._slave_fd)
            except OSError:
                pass
            self._slave_fd = None

        self._process = None
        logger.debug("PTY session closed")

    async def _read_until_exit(self, timeout: float) -> int:
        """持续读取输出直到进程退出

        Args:
            timeout: 超时时间（秒）

        Returns:
            进程退出码
        """
        deadline = time.monotonic() + timeout

        while True:
            # 检查超时
            if time.monotonic() > deadline:
                raise asyncio.TimeoutError()

            # 检查进程是否已退出
            if self._process.returncode is not None:
                # 进程已退出，再读一次剩余输出
                await self._drain_remaining_output()
                return self._process.returncode

            # 读取输出
            try:
                chunk = os.read(self._master_fd, self._buffer_size)
                if chunk:
                    text = chunk.decode("utf-8", errors="replace")
                    self._output_buffer += text

                    # 自动应答
                    if self._auto_respond:
                        await self._try_auto_respond(text)
            except (BlockingIOError, OSError):
                pass

            # 检查进程状态
            if self._process.returncode is None:
                try:
                    await asyncio.wait_for(
                        self._process.wait(), timeout=0.05
                    )
                except asyncio.TimeoutError:
                    pass
            else:
                await self._drain_remaining_output()
                return self._process.returncode

            await asyncio.sleep(0.02)

    async def _drain_remaining_output(self) -> None:
        """排空剩余输出"""
        for _ in range(10):  # 最多尝试 10 次
            try:
                chunk = os.read(self._master_fd, self._buffer_size)
                if chunk:
                    text = chunk.decode("utf-8", errors="replace")
                    self._output_buffer += text
                else:
                    break
            except (BlockingIOError, OSError):
                break
            await asyncio.sleep(0.01)

    async def _try_auto_respond(self, recent_output: str) -> None:
        """检测提示并自动应答

        Args:
            recent_output: 最近的输出文本
        """
        import re

        for pattern, response in self._respond_rules:
            if not response:
                # 空响应规则（如密码提示）跳过
                continue
            if re.search(pattern, recent_output, re.IGNORECASE | re.MULTILINE):
                logger.debug("Auto-responding to prompt '%s' with '%s'", pattern, response)
                await self.send(response)
                break
