"""Live user sessions on hosts (Linux loginctl / Windows qwinsta) via SSH or WinRM."""

from __future__ import annotations

import json
import re
from dataclasses import dataclass
from datetime import datetime, timezone

from app.models import Event, Host
from app.services.event_actor_user import extract_event_actor_user
from app.services.linux_admin_settings import LinuxAdminConfig
from app.services.ssh_connect import (
    HostNotLinuxError as SshHostNotLinuxError,
    HostTargetMissingError as SshHostTargetMissingError,
    SshCommandResult,
    is_linux_host,
    iter_ssh_targets,
    run_ssh_command,
)
from app.services.win_admin_settings import WinAdminConfig
from app.services.winrm_connect import (
    HostNotWindowsError,
    HostTargetMissingError,
    WinRmCmdResult,
    is_windows_host,
    run_winrm_logoff,
    run_winrm_on_host_targets,
    run_winrm_qwinsta,
)

SESSION_EVENT_TYPES_LINUX = frozenset(
    {
        "session.logind.new",
        "ssh.login.success",
        "privilege.sudo.command",
    }
)
SESSION_EVENT_TYPES_WINDOWS = frozenset({"rdp.login.success"})

SESSION_TERMINATED_AT_KEY = "session_terminated_at"
SESSION_TERMINATED_BY_KEY = "session_terminated_by"

_LOGIND_SESSION_ID_RE = re.compile(r"^\d+$|^c\d+$", re.IGNORECASE)
_LOGIND_USER_RE = re.compile(r"^[a-z_][a-z0-9._-]*$", re.IGNORECASE)
_LOGIND_STATES = frozenset(
    {
        "active",
        "online",
        "closing",
        "opening",
        "degraded",
        "lingering",
        "preparing",
        "unknown",
    }
)
_LOGIND_TTY_RE = re.compile(r"^(pts|tty)/", re.IGNORECASE)
_NO_SESSION_MARKERS = ("no session", "unknown session", "does not exist")


@dataclass(frozen=True)
class HostSessionRow:
    session_id: str
    user: str
    tty: str | None = None
    state: str | None = None
    source_ip: str | None = None
    session_name: str | None = None


def _details_dict(event: Event) -> dict:
    raw = event.details
    return raw if isinstance(raw, dict) else {}


def _event_login_user(event: Event) -> str:
    actor = extract_event_actor_user(event.type, event.details)
    if actor:
        return actor.strip()
    return str(_details_dict(event).get("user") or "").strip()


def event_session_terminated(event: Event) -> bool:
    details = _details_dict(event)
    if details.get("session_terminated") is True:
        return True
    at = details.get(SESSION_TERMINATED_AT_KEY)
    return at is not None and str(at).strip() != ""


def mark_event_session_terminated(event: Event, *, by_username: str | None = None) -> None:
    details = dict(_details_dict(event))
    details[SESSION_TERMINATED_AT_KEY] = datetime.now(timezone.utc).isoformat()
    if by_username and by_username.strip():
        details[SESSION_TERMINATED_BY_KEY] = by_username.strip()
    event.details = details


def event_session_id(event: Event) -> str | None:
    details = _details_dict(event)
    for key in ("session_id", "sid"):
        val = details.get(key)
        if val is not None and str(val).strip():
            return str(val).strip()
    return None


def event_supports_session_terminate(event: Event) -> bool:
    if event.type in SESSION_EVENT_TYPES_LINUX:
        return is_linux_host_event(event)
    if event.type in SESSION_EVENT_TYPES_WINDOWS:
        return is_windows_host_event(event)
    return False


def is_linux_host_event(event: Event) -> bool:
    host = event.host
    return host is not None and is_linux_host(host)


def is_windows_host_event(event: Event) -> bool:
    host = event.host
    return host is not None and is_windows_host(host)


def _normalize_logind_tty(value: object) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    if not text or text == "-":
        return None
    return text


def _logind_row_priority(row: HostSessionRow) -> tuple[int, int]:
    """Prefer interactive TTY sessions over ephemeral scope/SSH-exec rows (tty «-»)."""
    has_tty = 1 if row.tty and _LOGIND_TTY_RE.match(row.tty) else 0
    try:
        sid_num = int(row.session_id)
    except ValueError:
        sid_num = 0
    return (has_tty, sid_num)


def filter_logind_session_rows(rows: list[HostSessionRow]) -> list[HostSessionRow]:
    """Drop no-TTY duplicates when the same user already has a pts/tty session."""
    if len(rows) < 2:
        return rows

    by_user: dict[str, list[HostSessionRow]] = {}
    for row in rows:
        key = row.user.casefold()
        by_user.setdefault(key, []).append(row)

    out: list[HostSessionRow] = []
    for group in by_user.values():
        if len(group) == 1:
            out.append(group[0])
            continue
        with_tty = [r for r in group if r.tty and _LOGIND_TTY_RE.match(r.tty)]
        if with_tty:
            out.append(max(with_tty, key=_logind_row_priority))
            continue
        out.append(max(group, key=_logind_row_priority))
    out.sort(key=lambda r: (r.user.casefold(), -_logind_row_priority(r)[1]))
    return out


def parse_loginctl_sessions_json(stdout: str) -> list[HostSessionRow]:
    text = stdout.strip()
    if not text:
        return []
    try:
        payload = json.loads(text)
    except json.JSONDecodeError:
        return []
    if not isinstance(payload, list):
        return []

    rows: list[HostSessionRow] = []
    for item in payload:
        if not isinstance(item, dict):
            continue
        sid = str(item.get("session") or "").strip()
        user = str(item.get("user") or "").strip()
        if not sid or not user:
            continue
        if not _LOGIND_SESSION_ID_RE.match(sid):
            continue
        if not _LOGIND_USER_RE.match(user):
            continue
        state = str(item.get("state") or "").strip().lower() or None
        if state and state not in _LOGIND_STATES:
            continue
        rows.append(
            HostSessionRow(
                session_id=sid,
                user=user,
                tty=_normalize_logind_tty(item.get("tty")),
                state=state,
            )
        )
    return filter_logind_session_rows(rows)


def _looks_like_loginctl_row(parts: list[str]) -> bool:
    if len(parts) < 6:
        return False
    sid, uid, user, state = parts[0], parts[1], parts[2], parts[5].lower()
    if not _LOGIND_SESSION_ID_RE.match(sid):
        return False
    if not uid.isdigit():
        return False
    if not _LOGIND_USER_RE.match(user):
        return False
    return state in _LOGIND_STATES


def parse_loginctl_sessions(stdout: str) -> list[HostSessionRow]:
    rows: list[HostSessionRow] = []
    for line in stdout.splitlines():
        text = line.strip()
        if not text or text.startswith("SESSION"):
            continue
        parts = text.split()
        if not _looks_like_loginctl_row(parts):
            continue
        sid, user = parts[0], parts[2]
        tty = parts[4] if len(parts) > 4 and parts[4] != "-" else None
        state = parts[5] if len(parts) > 5 else None
        rows.append(
            HostSessionRow(
                session_id=sid,
                user=user,
                tty=_normalize_logind_tty(tty),
                state=state,
            )
        )
    return filter_logind_session_rows(rows)


def _shell_quote(value: str) -> str:
    return "'" + value.replace("'", "'\"'\"'") + "'"


def _linux_session_exists_message(stderr: str, stdout: str) -> bool:
    blob = f"{stderr}\n{stdout}".casefold()
    return any(marker in blob for marker in _NO_SESSION_MARKERS)


def _linux_show_session_user(host: Host, cfg: LinuxAdminConfig, session_id: str) -> str | None:
    sid = session_id.strip()
    if not sid:
        return None
    remote_cmd = f"loginctl show-session {_shell_quote(sid)} -p Name --value"
    targets = iter_ssh_targets(host)
    for target in targets:
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=remote_cmd,
            connect_timeout_sec=15,
            command_timeout_sec=30,
            need_root=True,
            login_shell=False,
        )
        if not result.ok:
            continue
        user = result.stdout.strip().splitlines()[-1].strip() if result.stdout.strip() else ""
        if user and _LOGIND_USER_RE.match(user):
            return user
    return None


def list_linux_sessions(host: Host, cfg: LinuxAdminConfig) -> tuple[list[HostSessionRow], SshCommandResult]:
    if not is_linux_host(host):
        raise SshHostNotLinuxError("Host is not Linux")
    targets = iter_ssh_targets(host)
    json_cmd = "loginctl list-sessions --output=json --no-legend"
    text_cmd = "loginctl list-sessions --no-legend --no-pager"
    last: SshCommandResult | None = None
    for target in targets:
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=json_cmd,
            connect_timeout_sec=15,
            command_timeout_sec=45,
            need_root=True,
            login_shell=False,
        )
        last = result
        if result.ok and result.stdout.strip():
            rows = parse_loginctl_sessions_json(result.stdout)
            if rows:
                return rows, result
            rows = parse_loginctl_sessions(result.stdout)
            if rows:
                return rows, result
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=text_cmd,
            connect_timeout_sec=15,
            command_timeout_sec=45,
            need_root=True,
            login_shell=False,
        )
        last = result
        if result.ok and result.stdout.strip():
            return parse_loginctl_sessions(result.stdout), result
    assert last is not None
    return [], last


def terminate_linux_session(
    host: Host,
    cfg: LinuxAdminConfig,
    session_id: str,
) -> SshCommandResult:
    sid = session_id.strip()
    if not sid:
        return SshCommandResult(ok=False, message="session_id is required", target="")
    if not is_linux_host(host):
        raise SshHostNotLinuxError("Host is not Linux")
    targets = iter_ssh_targets(host)
    remote_cmd = f"loginctl terminate-session {_shell_quote(sid)}"
    last: SshCommandResult | None = None
    session_user = _linux_show_session_user(host, cfg, sid)
    for target in targets:
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=remote_cmd,
            connect_timeout_sec=15,
            command_timeout_sec=45,
            need_root=True,
            login_shell=False,
        )
        last = result
        if result.ok:
            return result
    if session_user:
        return _terminate_linux_user_sessions(host, cfg, session_user)
    if last is not None and _linux_session_exists_message(last.stderr, last.stdout):
        return SshCommandResult(
            ok=False,
            message=(
                f"Сессия {sid} не найдена (устарела). Обновите список и повторите "
                "или завершите все сессии пользователя."
            ),
            target=last.target,
            stdout=last.stdout,
            stderr=last.stderr,
            exit_code=last.exit_code,
        )
    assert last is not None
    return last


def parse_qwinsta_sessions(stdout: str, *, filter_user: str | None = None) -> list[HostSessionRow]:
    rows: list[HostSessionRow] = []
    norm_filter = (filter_user or "").strip().lower()

    def norm_user(value: str) -> str:
        return value.replace("B26\\", "").replace("b26\\", "").lower()

    for line in stdout.splitlines():
        text = line.strip()
        if not text or re.match(r"^SESSION", text, re.I) or text.startswith("---"):
            continue
        parts = text.split()
        if len(parts) < 4:
            continue
        session_name = parts[0].lstrip(">")
        user_name = parts[1]
        try:
            sid = int(parts[2])
        except ValueError:
            continue
        state = " ".join(parts[3:])
        if norm_filter and norm_filter not in norm_user(user_name):
            continue
        rows.append(
            HostSessionRow(
                session_id=str(sid),
                user=user_name,
                session_name=session_name,
                state=state,
            )
        )

    if rows or not norm_filter:
        return rows

    return parse_qwinsta_sessions(stdout, filter_user=None)


def list_windows_sessions(host: Host, cfg: WinAdminConfig) -> tuple[list[HostSessionRow], WinRmCmdResult | None]:
    if not is_windows_host(host):
        raise HostNotWindowsError("Host is not Windows")

    def action(target: str) -> WinRmCmdResult:
        return run_winrm_qwinsta(target=target, user=cfg.user, password=cfg.password)

    result, _attempts = run_winrm_on_host_targets(host, user=cfg.user, password=cfg.password, action=action)
    if result is None or not result.ok:
        return [], result
    return parse_qwinsta_sessions(result.stdout), result


def terminate_windows_session(
    host: Host,
    cfg: WinAdminConfig,
    session_id: str,
) -> WinRmCmdResult | None:
    try:
        sid = int(session_id.strip())
    except ValueError:
        return WinRmCmdResult(ok=False, message="Invalid Windows session_id", target="")

    def action(target: str) -> WinRmCmdResult:
        return run_winrm_logoff(
            target=target,
            user=cfg.user,
            password=cfg.password,
            session_id=sid,
        )

    result, _attempts = run_winrm_on_host_targets(host, user=cfg.user, password=cfg.password, action=action)
    return result


def terminate_session_for_event(
    event: Event,
    *,
    linux_cfg: LinuxAdminConfig,
    win_cfg: WinAdminConfig,
    session_id: str | None = None,
) -> SshCommandResult | WinRmCmdResult:
    host = event.host
    if host is None:
        raise ValueError("Event has no host")

    sid = (session_id or event_session_id(event) or "").strip()
    if is_linux_host(host):
        if not sid:
            user = _event_login_user(event)
            if user:
                return _terminate_linux_user_sessions(host, linux_cfg, user)
            raise ValueError("session_id is required for this event")
        return terminate_linux_session(host, linux_cfg, sid)

    if is_windows_host(host):
        if not sid:
            user = _event_login_user(event)
            sessions, qwinsta = list_windows_sessions(host, win_cfg)
            if not qwinsta or not qwinsta.ok:
                raise ValueError(qwinsta.message if qwinsta else "qwinsta failed")
            matched = [s for s in sessions if user.lower() in s.user.lower()] if user else sessions
            if len(matched) == 1:
                sid = matched[0].session_id
            elif not matched:
                raise ValueError("No matching Windows session for user")
            else:
                raise ValueError("Multiple sessions; specify session_id")
        result = terminate_windows_session(host, win_cfg, sid)
        if result is None:
            raise ValueError("WinRM logoff failed")
        return result

    raise ValueError("Unsupported host OS for session terminate")


def _terminate_linux_user_sessions(host: Host, cfg: LinuxAdminConfig, user: str) -> SshCommandResult:
    safe_user = _shell_quote(user.strip())
    remote_cmd = f"loginctl terminate-user {safe_user}"
    targets = iter_ssh_targets(host)
    last: SshCommandResult | None = None
    for target in targets:
        result = run_ssh_command(
            target=target,
            user=cfg.user,
            password=cfg.password,
            remote_cmd=remote_cmd,
            connect_timeout_sec=15,
            command_timeout_sec=45,
            need_root=True,
            login_shell=False,
        )
        last = result
        if result.ok:
            return result
    assert last is not None
    return last