console v0.4.0 [extra]

"""console.Module — интерактивный PTY-терминал (выделен из admin).

Веб: /ws/terminal (ui_ws, сырой PTY-прокси). Telegram: tg_session
(ConsoleManager — промпт-цикл, секреты, hostkey; admin делегирует).

Без конфига и секретов: авторизация — право console_session
(WS-хендшейк проверяет транспорт, TG-старт — admin через access).
"""
import logging

from core.base_module import BaseModule

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        self.ctx = ctx
        from . import tg_session as _tg
        self.tg = _tg.ConsoleManager()
        ctx.api.register("console", "manager", lambda: self.tg)
        ctx.api.register("console", "session_user_ids",
                         self.tg.active_user_ids)

    def start(self) -> None:
        # Все setup'ы выполнены: цепляемся к admin-меню и движкам
        # (живой console_sessions для шимов ai_chat/opencode).
        try:
            warnings = self.tg.attach(self.ctx)
        except Exception as e:  # noqa: BLE001
            logger.warning("console: attach: %s", e)
            return
        for w in warnings or []:
            logger.warning("console: %s", w)

    def health(self) -> dict:
        try:
            from . import ui_ws as _ws
            nws = len(getattr(_ws, "_sessions", {}) or {})
        except Exception:  # noqa: BLE001
            nws = -1
        try:
            ntg = len(self.tg.active_user_ids())
        except Exception:  # noqa: BLE001
            ntg = -1
        return {"ok": True, "module": self.name,
                "sessions_ws": nws, "sessions_tg": ntg}

"""console.tg_session — TG-консоль: интерактивный shell в личке Telegram.

Владелец сессий — модуль console (admin только делегирует: кнопка
console_session и свободный текст при активной сессии).

В отличие от веб-терминала (сырой PTY-прокси), Telegram — покадровый:
каждое сообщение = одна итерация промпт-ориентированного цикла, а не
одноразовая команда «cmd; echo marker». Состояния сессии:
  idle          — ждёт команду;
  await_secret  — интерактив попросил секрет (пароль/PIN/passphrase):
                  следующее сообщение уходит в stdin БЕЗ эха в чат и в лог;
  await_hostkey — ssh показал fingerprint нового хоста: ждём «да»/«нет».

Команды сессии (в любом состоянии): #quitconsole (мягко закрыть),
#killconsole (убить зависший процесс, напр. повисший ssh).
"""
from __future__ import annotations

import logging
import re
import threading
import time

logger = logging.getLogger("botmod_console.tg")

#: Лимит Telegram на сообщение; вывод режем с запасом.
TG_TEXT_LIMIT = 3500
#: Бюджет одного сообщения (сек): хватает на ssh-handshake и sudo.
ROUND_BUDGET_S = 120.0
#: Шаг ожидания: короткий, чтобы замечать тишину (см. QUIET_S).
EXPECT_STEP_S = 2.0
#: Тишина после вывода (сек): виток считается завершённым. Нужно для
#: remote-shell (ssh): шелл-промпта там никогда не будет — только вывод
#: и молчание. Пустой вывод тишиной не считаем (sleep 300 ждёт бюджета).
QUIET_S = 4.0
#: Простой без сообщений, после которого сессия прибивается ленивым sweep.
IDLE_TTL_S = 1800

_ANSI_RE = re.compile(
    r"\x1b\[[0-9;?]*[a-zA-Z]"
    r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)"
    r"|\x1b[()][0-9A-Z]"
    r"|\x1b[=>M78c]"
    r"|\x9b[0-9;?]*[a-zA-Z]"
)
#: password/passphrase/PIN/токен в конце буфера (с двоеточием или без —
#: у openssh «user@host's password:», у sudo «[sudo] password for u:»).
_SECRET_RE = re.compile(
    r"(?i)(?:password|passphrase|passwd|пароль|pin|token|секрет|otp|2fa|mfa)"
    r"[^:\n]{0,40}:\s*$"
    r"|\b(?:password|passphrase|пароль)\s*$"
)
_HOSTKEY_RE = re.compile(
    r"Are you sure you want to continue connecting", re.IGNORECASE)
_YES_RE = re.compile(r"^(да|д|yes|y|ага)\.?$", re.IGNORECASE)
_NO_RE = re.compile(r"^(нет|н|no|n|не)\.?$", re.IGNORECASE)


def strip_ansi(text: str) -> str:
    """Убрать escape-последовательности терминала (в TG им не место)."""
    if not text:
        return ""
    cleaned = _ANSI_RE.sub("", text)
    cleaned = cleaned.replace("\r\n", "\n").replace("\r", "\n")
    return re.sub(r"\n{3,}", "\n\n", cleaned)


def truncate(text: str, limit: int = TG_TEXT_LIMIT) -> str:
    """Обрезать под лимит TG с пометкой (полный вывод — в логе сессии)."""
    if len(text) <= limit:
        return text
    return (text[:limit].rstrip()
            + "\n…[обрезано, полный вывод в console_sessions.log]")


def _audit_log(user_id, text: str) -> None:
    """Команды/вывод сессии — в console_sessions.log (как раньше admin)."""
    try:
        from botmod_admin.engine import log_console as _log
    except Exception:  # noqa: BLE001
        logger.info("console tg user=%s: %s", user_id, (text or "")[:200])
        return
    try:
        _log(user_id, text)
    except Exception:  # noqa: BLE001
        logger.warning("console tg: log_console недоступен")


class LiveConsoleUsers:
    """Живое множество активных TG-сессий для `uid in admin_menu.console_sessions`.

    Подменяет пустые заглушки в шимах ai_chat/opencode, чтобы их движки
    не отвечали дублями, пока пользователь работает в консоли.
    """

    def __init__(self, manager: "ConsoleManager"):
        self._manager = manager

    def __contains__(self, uid) -> bool:
        try:
            return self._manager.is_active(uid)
        except Exception:  # noqa: BLE001
            return False

    def __len__(self) -> int:
        try:
            return len(self._manager.active_user_ids())
        except Exception:  # noqa: BLE001
            return 0

    def __iter__(self):
        try:
            return iter(self._manager.active_user_ids())
        except Exception:  # noqa: BLE001
            return iter(())


class _TgSession:
    __slots__ = ("user_id", "child", "prompt", "prompt_re", "state",
                 "stash", "lock", "last_active")

    def __init__(self, user_id, child, prompt: str):
        self.user_id = user_id
        self.child = child
        self.prompt = prompt
        # Промпт — литерал: в expect() идёт как re.escape, иначе
        # скобки [...] читаются как символьный класс и совпадения нет.
        self.prompt_re = re.compile(re.escape(prompt))
        self.state = "idle"
        self.stash = ""
        self.lock = threading.Lock()
        self.last_active = time.monotonic()


class ConsoleManager:
    """Сессии TG-консоли. Привязка к отправке — через attach(menu)."""

    def __init__(self):
        self._sessions: dict = {}
        self._guard = threading.Lock()
        self._menu = None
        # Бюджеты (сек) — атрибуты, чтобы тесты могли их уменьшать.
        self.cmd_budget_s = ROUND_BUDGET_S
        self.secret_budget_s = 60.0
        self.hostkey_budget_s = 90.0

    # -- привязка ------------------------------------------------------
    def attach(self, ctx) -> list:
        """Связать с admin-меню и движками (вызывается из console.start).

        Возвращает список предупреждений (для лога).
        """
        warnings = []
        try:
            menu = ctx.api.call("admin.menu")
        except Exception as e:  # noqa: BLE001
            warnings.append("admin.menu недоступен: %s" % e)
            menu = None
        if menu is not None:
            menu.console_delegate = self
            self._menu = menu
        else:
            warnings.append("без admin TG-консоль недоступна (WS работает)")
        live = LiveConsoleUsers(self)
        for owner, method in (("ai_chat", "engine"), ("opencode", "chat")):
            try:
                engine = ctx.api.call("%s.%s" % (owner, method))
            except Exception as e:  # noqa: BLE001
                warnings.append("%s.%s недоступен: %s" % (owner, method, e))
                continue
            try:
                admin_menu = getattr(engine, "admin_menu", None)
                if admin_menu is not None:
                    admin_menu.console_sessions = live
            except Exception as e:  # noqa: BLE001
                warnings.append("%s: шим не подменён: %s" % (owner, e))
        return warnings

    def _send(self, chat_id, text, context=None) -> None:
        if self._menu is not None:
            try:
                self._menu.send_msg(chat_id, text, context=context)
                return
            except Exception:  # noqa: BLE001
                logger.warning("console tg: send_msg не удался")
        logger.info("console tg -> %s: %s", chat_id, (text or "")[:200])

    # -- реестр ----------------------------------------------------------
    def active_user_ids(self) -> set:
        with self._guard:
            return set(self._sessions.keys())

    def is_active(self, user_id) -> bool:
        with self._guard:
            sess = self._sessions.get(user_id)
        return sess is not None and self._alive(sess)

    @staticmethod
    def _alive(sess: _TgSession) -> bool:
        try:
            return bool(sess.child.isalive())
        except Exception:  # noqa: BLE001
            return False

    def sweep(self, idle_s: float = IDLE_TTL_S) -> int:
        """Прибить простаивающие сессии. Возвращает число убитых."""
        now = time.monotonic()
        dead = []
        with self._guard:
            items = list(self._sessions.items())
        for uid, sess in items:
            if now - sess.last_active > idle_s or not self._alive(sess):
                dead.append(uid)
        for uid in dead:
            self._close(uid, silent=True)
        return len(dead)

    # -- жизненный цикл ----------------------------------------------------
    def start(self, chat_id, context, user_id) -> None:
        try:
            import pexpect as _pexpect
        except ImportError:
            self._send(chat_id, "Консоль недоступна: нет pexpect "
                                "(modules/console/requirements.txt).",
                       context=context)
            return
        self.sweep()
        with self._guard:
            if user_id in self._sessions:
                sess = self._sessions[user_id]
                if self._alive(sess):
                    self._send(chat_id, "Консоль уже запущена. "
                                        "Команды сессии: #quitconsole — закрыть, "
                                        "#killconsole — убить зависший процесс.",
                               context=context)
                    return
                self._sessions.pop(user_id, None)
        prompt = "[TGCONSOLE:%s]>" % user_id
        try:
            child = _pexpect.spawn(
                "/bin/bash", ["--noprofile", "--norc", "-i"],
                encoding="utf-8", echo=False, timeout=5,
                env={"TERM": "dumb", "PATH": "/usr/local/sbin:/usr/local/bin:"
                                              "/usr/sbin:/usr/bin:/sbin:/bin"})
            child.sendline("export PS1='%s'; unset PROMPT_COMMAND" % prompt)
            child.expect_exact(prompt, timeout=10)
        except Exception as e:  # noqa: BLE001
            logger.error("console tg: spawn: %s", e)
            self._send(chat_id, "Ошибка при запуске консоли.",
                       context=context)
            return
        with self._guard:
            self._sessions[user_id] = _TgSession(user_id, child, prompt)
        _audit_log(user_id, "СЕССИЯ ЗАПУЩЕНА (tg)")
        self._send(chat_id, "Консоль запущена. Введите команды.\n"
                            "#quitconsole — закрыть, #killconsole — убить "
                            "зависший процесс.",
                   context=context)

    def handle_text(self, chat_id, context, user_id, text) -> bool:
        """Очередное сообщение пользователя сессии. Всегда True (съедено)."""
        self.sweep()
        with self._guard:
            sess = self._sessions.get(user_id)
        if sess is None:
            self._send(chat_id, "Консольная сессия не активна.",
                       context=context)
            return True
        if not self._alive(sess):
            self._close(user_id, silent=True)
            self._send(chat_id, "Сессия завершена (shell вышел — например, "
                                "командой exit). Откройте заново кнопкой "
                                "💻 Консоль.",
                       context=context)
            return True
        cmd = (text or "").strip()
        if cmd in ("#quitconsole", "#killconsole"):
            self._close(user_id, kill=(cmd == "#killconsole"))
            _audit_log(user_id, "СЕССИЯ ЗАВЕРШЕНА (tg, %s)" % cmd)
            self._send(chat_id, "Консоль завершена.", context=context)
            return True
        with sess.lock:
            sess.last_active = time.monotonic()
            try:
                if sess.state == "await_secret":
                    self._feed_secret(sess, chat_id, context, text or "")
                elif sess.state == "await_hostkey":
                    self._feed_hostkey(sess, chat_id, context, cmd)
                else:
                    self._run_command(sess, chat_id, context, text or "")
                sess.last_active = time.monotonic()
            except Exception as e:  # noqa: BLE001
                logger.error("console tg run: %s", e)
                self._send(chat_id, "Ошибка выполнения команды в консоли.",
                           context=context)
        return True

    # -- выполнение ----------------------------------------------------------
    def _expect_step(self, child, sess, patterns, left: float) -> int:
        """Один шаг expect. Возвращает индекс (последний — шаг TIMEOUT)."""
        try:
            return child.expect(patterns, timeout=min(EXPECT_STEP_S, left))
        except Exception as e:  # noqa: BLE001
            logger.warning("console tg expect: %s", e)
            return len(patterns) - 2  # EOF-ветка: безопасное закрытие

    @staticmethod
    def _snapshot(child) -> str:
        """Весь вывод с прошлого совпадения (before при TIMEOUT — весь
        буфер, поэтому присваиваем, а не дописываем)."""
        before = getattr(child, "before", "") or ""
        if isinstance(before, bytes):
            return before.decode("utf-8", errors="replace")
        return str(before)

    def _run_command(self, sess: _TgSession, chat_id, context,
                     command: str) -> None:
        child = sess.child
        self._drain(child)
        child.sendline(command)
        patterns = [sess.prompt_re, _SECRET_RE, _HOSTKEY_RE,
                    _pexpect_EOF(), _pexpect_TIMEOUT()]
        output = ""
        seen_len = 0
        last_data = time.monotonic()
        deadline = time.monotonic() + self.cmd_budget_s
        while True:
            left = deadline - time.monotonic()
            if left <= 0:
                self._send(chat_id, "Вывод команды (превышен бюджет %d c, "
                                    "процесс продолжает работать):\n%s"
                           % (int(self.cmd_budget_s),
                              truncate("Вывод команды:\n"
                                       + self._clean(output, command))),
                           context=context)
                return
            index = self._expect_step(child, sess, patterns, left)
            if index == 4:
                # Шаг TIMEOUT: тихо. Вывод есть + тишина QUIET_S —
                # виток завершён (remote-shell: промпта не будет).
                output = self._snapshot(child)
                if len(output) > seen_len:
                    seen_len = len(output)
                    last_data = time.monotonic()
                if (output.strip()
                        and time.monotonic() - last_data >= QUIET_S):
                    self._finish_turn(sess, chat_id, context, command,
                                      output, quiet=True)
                    return
                continue
            if index == 0:
                self._finish_turn(sess, chat_id, context, command,
                                  self._snapshot(child))
                return
            if index == 1:
                # before (всё с прошлого совпадения) + after (сам промпт).
                output = (self._snapshot(child)
                          + self._to_text(child.after))
                if not self._secret_at_end(output):
                    continue  # ложное срабатывание в середине вывода
                sess.state = "await_secret"
                sess.stash = output
                self._send(chat_id, "🔑 Требуется секрет (пароль/PIN). "
                                    "Пришлите его следующим сообщением — "
                                    "оно будет удалено и не попадёт в лог. "
                                    "Отмена: #quitconsole.",
                           context=context)
                return
            if index == 2:
                output = (self._snapshot(child)
                          + self._to_text(child.after))
                sess.state = "await_hostkey"
                sess.stash = output
                self._send(chat_id, "🖥 Новый SSH-хост:\n%s\nОтветьте "
                                    "«да» (добавить и продолжить) или "
                                    "«нет»." % truncate(
                                        self._clean_fingerprint(output)),
                           context=context)
                return
            output = self._snapshot(child)
            self._close_session_obj(sess)
            _audit_log(sess.user_id, "$ %s\n[процесс завершён]\n%s"
                       % (command, self._clean(output, command)))
            self._send(chat_id, "Процесс завершён. Сессия закрыта — "
                                "откройте заново кнопкой 💻 Консоль.",
                       context=context)
            return

    def _finish_turn(self, sess: _TgSession, chat_id, context,
                     command: str, output: str, quiet: bool = False) -> None:
        """Успешный виток: промпт (quiet=False) или вывод+тишина (ssh)."""
        cleaned = self._clean(output, command)
        _audit_log(sess.user_id, "$ %s\n%s%s"
                   % (command, cleaned, " [по тишине]" if quiet else ""))
        if not cleaned:
            cleaned = "Команда выполнена, без вывода."
        self._send(chat_id, "Вывод команды:\n%s" % truncate(cleaned),
                   context=context)

    def _feed_secret(self, sess: _TgSession, chat_id, context,
                     secret: str) -> None:
        """Секрет уходит в stdin; в чат и лог — только факт ввода."""
        child = sess.child
        child.sendline(secret)
        patterns = [sess.prompt_re, _SECRET_RE, _pexpect_EOF(),
                    _pexpect_TIMEOUT()]
        seen_len = 0
        last_data = time.monotonic()
        deadline = time.monotonic() + self.secret_budget_s
        while True:
            left = deadline - time.monotonic()
            if left <= 0:
                self._send(chat_id, "Нет ответа после ввода секрета "
                                    "(процесс продолжает работать).",
                           context=context)
                return
            index = self._expect_step(child, sess, patterns, left)
            if index == 3:
                # Тихо: вывод + тишина QUIET_S = секрет принят
                # (remote-shell: промпта не будет).
                output = self._snapshot(child)
                if len(output) > seen_len:
                    seen_len = len(output)
                    last_data = time.monotonic()
                if (output.strip()
                        and time.monotonic() - last_data >= QUIET_S):
                    self._finish_secret(sess, chat_id, context, output)
                    return
                continue
            if index == 0:
                self._finish_secret(sess, chat_id, context,
                                    self._snapshot(child))
                return
            if index == 1:
                output = self._snapshot(child)
                _audit_log(sess.user_id, "[секрет введён, скрыт; "
                                         "запрошен снова]")
                self._send(chat_id, "Секрет не принят, попробуйте ещё раз "
                                    "(или #quitconsole).",
                           context=context)
                return
            output = self._snapshot(child)
            self._close_session_obj(sess)
            _audit_log(sess.user_id, "[секрет введён, скрыт; "
                                     "процесс завершён]")
            self._send(chat_id, "Процесс завершён. Сессия закрыта.",
                       context=context)
            sess.stash = ""
            return

    def _finish_secret(self, sess: _TgSession, chat_id, context,
                       output: str) -> None:
        """Секрет принят: промпт или вывод+тишина (remote-shell)."""
        sess.state = "idle"
        full = self._clean(sess.stash + output, "")
        _audit_log(sess.user_id, "[секрет введён, скрыт]\n%s" % full)
        if not full:
            full = "Готово."
        self._send(chat_id, "Вывод команды:\n%s" % truncate(full),
                   context=context)
        sess.stash = ""

    def _feed_hostkey(self, sess: _TgSession, chat_id, context,
                      answer: str) -> None:
        child = sess.child
        if _YES_RE.match(answer or ""):
            child.sendline("yes")
        elif _NO_RE.match(answer or ""):
            child.sendline("no")
        else:
            self._send(chat_id, "Ответьте «да» или «нет» (или #quitconsole).",
                       context=context)
            return
        patterns = [sess.prompt_re, _SECRET_RE, _HOSTKEY_RE,
                    _pexpect_EOF(), _pexpect_TIMEOUT()]
        seen_len = 0
        last_data = time.monotonic()
        deadline = time.monotonic() + self.hostkey_budget_s
        while True:
            left = deadline - time.monotonic()
            if left <= 0:
                sess.state = "idle"
                self._send(chat_id, "Нет ответа от ssh (процесс продолжает "
                                    "работать).",
                           context=context)
                return
            index = self._expect_step(child, sess, patterns, left)
            if index == 4:
                # Тихо: вывод + тишина = ssh ушёл дальше без промптов
                # (напр. вход по ключу в remote-shell).
                output = self._snapshot(child)
                if len(output) > seen_len:
                    seen_len = len(output)
                    last_data = time.monotonic()
                if (output.strip()
                        and time.monotonic() - last_data >= QUIET_S):
                    self._finish_hostkey(sess, chat_id, context, answer,
                                         output)
                    return
                continue
            if index == 0:
                self._finish_hostkey(sess, chat_id, context, answer,
                                     self._snapshot(child))
                return
            if index in (1, 2):
                output = (self._snapshot(child)
                          + self._to_text(child.after))
                sess.stash = output
                sess.state = ("await_secret" if index == 1
                              else "await_hostkey")
                self._send(chat_id, "🔑 Требуется секрет — пришлите "
                                    "следующим сообщением."
                           if index == 1 else
                           "🖥 Снова запрос hostkey — ответьте «да»/«нет».",
                           context=context)
                return
            self._close_session_obj(sess)
            self._send(chat_id, "Процесс завершён. Сессия закрыта.",
                       context=context)
            sess.stash = ""
            return

    def _finish_hostkey(self, sess: _TgSession, chat_id, context,
                        answer: str, output: str) -> None:
        sess.state = "idle"
        full = self._clean(sess.stash + output, "")
        _audit_log(sess.user_id, "[hostkey: %s]\n%s" % (answer, full))
        if not full:
            full = "Готово."
        self._send(chat_id, "Вывод команды:\n%s" % truncate(full),
                   context=context)
        sess.stash = ""

    # -- закрытие ------------------------------------------------------------
    def _close(self, user_id, kill: bool = False, silent: bool = False) -> None:
        with self._guard:
            sess = self._sessions.pop(user_id, None)
        if sess is not None:
            self._close_session_obj(sess, kill=kill)

    def _close_session_obj(self, sess: _TgSession,
                           kill: bool = False) -> None:
        with self._guard:
            if self._sessions.get(sess.user_id) is sess:
                self._sessions.pop(sess.user_id, None)
        child = sess.child
        try:
            import signal as _signal
            try:
                if kill:
                    child.kill(_signal.SIGKILL)
                else:
                    child.sendline("exit")
                    child.expect(_pexpect_EOF(), timeout=3)
            except Exception:  # noqa: BLE001
                try:
                    child.terminate(force=True)
                except Exception:  # noqa: BLE001
                    pass
            try:
                child.close(force=True)
            except Exception:  # noqa: BLE001
                pass
        except Exception:  # noqa: BLE001
            pass

    # -- утилиты ---------------------------------------------------------------
    @staticmethod
    def _to_text(chunk) -> str:
        if chunk is None:
            return ""
        if isinstance(chunk, bytes):
            return chunk.decode("utf-8", errors="replace")
        return str(chunk)

    @staticmethod
    def _drain(child) -> None:
        """Сбросить непрочитанный остаток pty перед новой командой."""
        try:
            import pexpect as _pexpect
            while True:
                child.read_nonblocking(size=4096, timeout=0)
        except Exception:  # noqa: BLE001
            pass

    def _clean(self, output: str, command: str) -> str:
        text = strip_ansi(output or "")
        lines = [ln for ln in text.split("\n")
                 if "[TGCONSOLE:" not in ln]
        text = "\n".join(lines).strip()
        cmd = (command or "").strip()
        if cmd and text.startswith(cmd):
            text = text[len(cmd):].strip()
        return text

    @staticmethod
    def _clean_fingerprint(output: str) -> str:
        text = strip_ansi(output or "").strip()
        lines = [ln for ln in text.split("\n") if ln.strip()]
        tail = "\n".join(lines[-8:])
        return tail or "(пустой вывод)"

    @staticmethod
    def _secret_at_end(output: str) -> bool:
        tail = strip_ansi(output or "")[-160:]
        return bool(_SECRET_RE.search(tail))


def _pexpect_EOF():
    import pexpect as _pexpect
    return _pexpect.EOF


def _pexpect_TIMEOUT():
    import pexpect as _pexpect
    return _pexpect.TIMEOUT

"""console.ui_tg — кнопка /start (сессии — tg_session.ConsoleManager).

Роутинг TG-текста при активной сессии — через admin (legacy-движок
делегирует в console_delegate).
"""

TG_COMMANDS = {}

TG_MENU = [
    {"id": "console", "title": "💻 Консоль", "callback": "console_session",
     "right": "console_session"},
]

"""console.ui_web — витрина сервиса Терминал для кабинета (WS — в ui_ws)."""

ROUTES = []

SERVICE = {"key": "console_session", "icon": "💻", "title": "Терминал", "right": "console_session",
           "phase": 4, "desc": "Интерактивная shell-сессия в браузере (/ws/terminal)"}

PANELS = []


def handle_api(ctx, config, method, req):
    from core.errors import UserError

    raise UserError("неизвестный метод: %s" % method)

"""console.ui_ws — /ws/terminal (интерактивный PTY-терминал в панели).

Порт TerminalWebSocket из старого messengers/web.py: xterm.js на фронте,
на бэке — pty + /bin/bash. Одна сессия на пользователя (новая закрывает
старую). Право console_session.
"""

WS_PATH = "/ws/terminal"

from botmod_transport_web.ws import make_base as _make_base

BaseSocket = _make_base()

_sessions = {}  # "web:<uid>" -> WSSocket
_lock = None


def _get_lock():
    global _lock
    if _lock is None:
        import threading as _th
        _lock = _th.Lock()
    return _lock


class WSSocket(BaseSocket):
    RIGHT = "console_session"

    def on_authed(self, sess):
        import os as _os
        import pty as _pty
        import tornado.ioloop

        self.ws_uid = sess["uid"]
        self.ws_loop = tornado.ioloop.IOLoop.current()
        self.ws_master = None
        self.ws_pid = None
        key = f"web:{self.ws_uid}"
        try:
            self.mod_ctx.api.call("audit.record", key, "console.open",
                                  "terminal", "")
        except Exception:  # noqa: BLE001
            pass
        with _get_lock():
            old = _sessions.get(key)
            if old and old is not self:
                try:
                    old.close()
                except Exception:  # noqa: BLE001
                    pass
            _sessions[key] = self
        try:
            master, slave = _pty.openpty()
            pid = _os.fork()
            if pid == 0:
                try:
                    _os.setsid()
                    _os.dup2(slave, 0)
                    _os.dup2(slave, 1)
                    _os.dup2(slave, 2)
                    if slave > 2:
                        _os.close(slave)
                    if master > 2:
                        _os.close(master)
                    _os.environ["TERM"] = "xterm-256color"
                    _os.execv("/bin/bash", ["/bin/bash", "-i"])
                except Exception:  # noqa: BLE001
                    _os._exit(1)
            _os.close(slave)
            self.ws_master = master
            self.ws_pid = pid
            self._stop = False
            import threading as _th
            _th.Thread(target=self._reader, daemon=True).start()
        except Exception as e:  # noqa: BLE001
            try:
                self.close(code=1011, reason=str(e)[:120])
            except Exception:  # noqa: BLE001
                pass

    def _reader(self):
        import logging as _logging
        import os as _os
        import select as _select

        logger = _logging.getLogger("botmod_console.ws")
        try:
            while not getattr(self, "_stop", False):
                r, _, _ = _select.select([self.ws_master], [], [], 0.5)
                if not r:
                    try:
                        wpid, _ = _os.waitpid(self.ws_pid, _os.WNOHANG)
                        if wpid == self.ws_pid:
                            self.ws_loop.add_callback(
                                self._send_data, "\r\n[процесс завершён]\r\n")
                            try:
                                self.close()
                            except Exception:  # noqa: BLE001
                                pass
                            break
                    except ChildProcessError:
                        break
                    continue
                try:
                    data = _os.read(self.ws_master, 4096)
                except OSError:
                    break
                if not data:
                    break
                try:
                    text = data.decode("utf-8", errors="replace")
                except Exception:  # noqa: BLE001
                    text = ""
                if text:
                    self.ws_loop.add_callback(self._send_data, text)
        except Exception as e:  # noqa: BLE001
            logger.debug("term reader web:%s: %s", getattr(self, "ws_uid", "?"), e)

    def _send_data(self, text):
        import json as _json
        try:
            self.write_message(_json.dumps({"data": text}, ensure_ascii=False))
        except Exception:  # noqa: BLE001
            pass

    def on_message(self, raw):
        import fcntl as _fcntl
        import json as _json
        import os as _os
        import signal as _signal
        import struct as _struct
        import termios as _termios

        if getattr(self, "ws_uid", None) is None or self.ws_master is None:
            return
        try:
            msg = _json.loads(raw)
        except Exception:  # noqa: BLE001
            try:
                _os.write(self.ws_master, raw.encode("utf-8", errors="replace"))
            except OSError:
                pass
            return
        if "data" in msg:
            try:
                _os.write(self.ws_master,
                          str(msg["data"]).encode("utf-8", errors="replace"))
            except OSError:
                pass
        if "cols" in msg and "rows" in msg:
            try:
                cols = int(msg["cols"])
                rows = int(msg["rows"])
                winsize = _struct.pack("HHHH", rows, cols, 0, 0)
                _fcntl.ioctl(self.ws_master, _termios.TIOCSWINSZ, winsize)
                try:
                    _os.kill(self.ws_pid, _signal.SIGWINCH)
                except ProcessLookupError:
                    pass
            except Exception:  # noqa: BLE001
                pass

    def on_close(self):
        import logging as _logging
        import os as _os
        import signal as _signal

        _logging.getLogger("botmod_console.ws").info(
            "WS term закрыт: web:%s", getattr(self, "ws_uid", "?"))
        key = f"web:{getattr(self, 'ws_uid', '?')}"
        with _get_lock():
            if _sessions.get(key) is self:
                _sessions.pop(key, None)
        self._stop = True
        pid = getattr(self, "ws_pid", None)
        master = getattr(self, "ws_master", None)
        if pid:
            try:
                _os.kill(pid, _signal.SIGHUP)
            except ProcessLookupError:
                pass
            try:
                _os.waitpid(pid, 0)
            except ChildProcessError:
                pass
        if master is not None:
            try:
                _os.close(master)
            except OSError:
                pass