Medios-Macina/plugins/scp/__init__.py

from __future__ import annotations

import fnmatch
import posixpath
import shlex
import stat
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import quote, unquote, urlparse

import paramiko
from scp import SCPClient

from PluginCore.base import Provider, SearchResult, parse_inline_query_arguments


def _coerce_bool(value: Any, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if value is None:
        return default
    text = str(value).strip().lower()
    if not text:
        return default
    if text in {"1", "true", "yes", "on"}:
        return True
    if text in {"0", "false", "no", "off"}:
        return False
    return default


def _coerce_int(value: Any, default: int) -> int:
    try:
        return int(value)
    except Exception:
        return default


def _format_epoch(raw_value: Any) -> str:
    try:
        stamp = int(raw_value)
    except Exception:
        return ""
    try:
        return datetime.fromtimestamp(stamp).strftime("%Y-%m-%d %H:%M")
    except Exception:
        return str(raw_value or "")


def _safe_filename(name: Any) -> str:
    raw = str(name or "").strip()
    if not raw:
        raw = "download"
    cleaned = "".join(ch if ch.isalnum() or ch in {"-", "_", ".", " "} else "_" for ch in raw)
    cleaned = cleaned.strip(" ._")
    return cleaned or "download"


def _unique_path(path: Path) -> Path:
    if not path.exists():
        return path
    stem = path.stem or "download"
    suffix = path.suffix
    counter = 1
    while True:
        candidate = path.with_name(f"{stem}_{counter}{suffix}")
        if not candidate.exists():
            return candidate
        counter += 1


class SCP(Provider):
    PLUGIN_NAME = "scp"
    URL = ("scp://", "sftp://")
    SUPPORTED_CMDLETS = frozenset({"add-file", "download-file", "search-file"})

    @property
    def label(self) -> str:
        return "SCP"

    @property
    def preserve_order(self) -> bool:
        return True

    @classmethod
    def config_schema(cls) -> List[Dict[str, Any]]:
        return [
            {
                "key": "host",
                "label": "Host",
                "default": "",
                "required": True,
                "placeholder": "ssh.example.com",
            },
            {
                "key": "port",
                "label": "Port",
                "type": "integer",
                "default": 22,
            },
            {
                "key": "username",
                "label": "Username",
                "default": "",
                "required": True,
                "placeholder": "deploy",
            },
            {
                "key": "password",
                "label": "Password",
                "type": "secret",
                "secret": True,
                "default": "",
            },
            {
                "key": "key_path",
                "label": "SSH Key Path",
                "type": "path",
                "default": "",
                "placeholder": "C:/Users/Admin/.ssh/id_ed25519",
            },
            {
                "key": "base_path",
                "label": "Base Path",
                "default": "/",
                "placeholder": "/srv/files",
            },
            {
                "key": "timeout",
                "label": "Timeout Seconds",
                "type": "integer",
                "default": 20,
            },
            {
                "key": "search_depth",
                "label": "Default Search Depth",
                "type": "integer",
                "default": 1,
            },
            {
                "key": "allow_agent",
                "label": "Use SSH Agent",
                "type": "boolean",
                "default": True,
            },
            {
                "key": "look_for_keys",
                "label": "Look For Default Keys",
                "type": "boolean",
                "default": True,
            },
        ]

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        super().__init__(config)
        _instance_name, conf = self.resolve_plugin_instance()
        defaults = self._settings_from_config(conf)
        self._host = str(defaults.get("host") or "").strip()
        self._port = int(defaults.get("port") or 22)
        self._username = str(defaults.get("username") or "").strip()
        self._password = str(defaults.get("password") or "").strip()
        self._key_path = str(defaults.get("key_path") or "").strip()
        self._timeout = max(1, int(defaults.get("timeout") or 20))
        self._search_depth = max(0, int(defaults.get("search_depth") or 1))
        self._allow_agent = bool(defaults.get("allow_agent"))
        self._look_for_keys = bool(defaults.get("look_for_keys"))
        self._base_path = self._normalize_remote_path(defaults.get("base_path") or "/", default="/")

    def _settings_from_config(self, conf: Optional[Dict[str, Any]], *, instance_name: Optional[str] = None) -> Dict[str, Any]:
        entry = dict(conf or {})
        return {
            "instance": str(instance_name or entry.get("_instance_name") or "").strip() or None,
            "host": str(entry.get("host") or "").strip(),
            "port": _coerce_int(entry.get("port"), 22),
            "username": str(entry.get("username") or entry.get("user") or "").strip(),
            "password": str(entry.get("password") or "").strip(),
            "key_path": str(entry.get("key_path") or entry.get("identity_file") or "").strip(),
            "timeout": max(1, _coerce_int(entry.get("timeout"), 20)),
            "search_depth": max(0, _coerce_int(entry.get("search_depth"), 1)),
            "allow_agent": _coerce_bool(entry.get("allow_agent"), True),
            "look_for_keys": _coerce_bool(entry.get("look_for_keys"), True),
            "base_path": self._normalize_remote_path(entry.get("base_path") or "/", default="/"),
        }

    def _resolve_settings(
        self,
        *,
        filters: Optional[Dict[str, Any]] = None,
        instance_name: Optional[str] = None,
        require_explicit: bool = False,
    ) -> Dict[str, Any]:
        requested = self.requested_instance_name(filters, instance=instance_name)
        resolved_name, conf = self.resolve_plugin_instance(
            requested,
            require_explicit=require_explicit or bool(requested),
        )
        settings = self._settings_from_config(conf, instance_name=resolved_name)
        if settings.get("instance") is None and requested:
            settings["instance"] = requested
        return settings

    def validate(self) -> bool:
        settings = self._resolve_settings()
        return bool(settings.get("host") and settings.get("username"))

    def config_helper_text(self) -> str:
        return "Test the SSH/SCP connection before searching. You can also generate an RSA key pair from here."

    def config_actions(self) -> List[Dict[str, Any]]:
        return [
            {
                "id": "test_connection",
                "label": "Test connection",
                "variant": "primary",
            },
            {
                "id": "generate_ssh_key",
                "label": "Generate SSH key",
                "variant": "default",
            },
        ]

    def run_config_action(self, action_id: str, **_kwargs: Any) -> Dict[str, Any]:
        normalized = str(action_id or "").strip().lower()
        if normalized == "test_connection":
            return self._run_test_connection()
        if normalized == "generate_ssh_key":
            return self._generate_ssh_keypair()
        return super().run_config_action(action_id, **_kwargs)

    def extract_query_arguments(self, query: str) -> Tuple[str, Dict[str, Any]]:
        text, inline = parse_inline_query_arguments(query)
        filters: Dict[str, Any] = {}

        instance_name = str(inline.get("instance") or inline.get("store") or "").strip()
        if instance_name:
            filters["instance"] = instance_name

        if inline.get("path"):
            filters["path"] = inline.get("path")
        if inline.get("depth"):
            filters["depth"] = max(0, _coerce_int(inline.get("depth"), self._search_depth))
        if inline.get("type"):
            filters["type"] = str(inline.get("type") or "").strip().lower()

        return text, filters

    def get_table_title(self, query: str, filters: Optional[Dict[str, Any]] = None) -> str:
        settings = self._resolve_settings(filters=filters)
        active_path = self._normalize_remote_path((filters or {}).get("path") or settings.get("base_path") or "/", default=str(settings.get("base_path") or "/"))
        instance_name = str(settings.get("instance") or "").strip()
        text = str(query or "").strip()
        if not text or text == "*":
            return f"SCP{f'[{instance_name}]' if instance_name else ''}: {active_path}"
        return f"SCP{f'[{instance_name}]' if instance_name else ''}: {text} @ {active_path}"

    def get_table_metadata(self, query: str, filters: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        settings = self._resolve_settings(filters=filters)
        return {
            "plugin": self.name,
            "instance": settings.get("instance"),
            "host": settings.get("host"),
            "path": self._normalize_remote_path((filters or {}).get("path") or settings.get("base_path") or "/", default=str(settings.get("base_path") or "/")),
            "query": str(query or "").strip(),
        }

    def search(
        self,
        query: str,
        limit: int = 50,
        filters: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> List[SearchResult]:
        _ = kwargs
        active_filters = dict(filters or {})
        settings = self._resolve_settings(filters=active_filters, require_explicit=True)
        if not settings.get("host") or not settings.get("username"):
            requested = self.requested_instance_name(active_filters)
            if requested:
                raise RuntimeError(f"SCP instance '{requested}' is unavailable")
            return []
        start_path = self._normalize_remote_path(active_filters.get("path") or settings.get("base_path") or "/", default=str(settings.get("base_path") or "/"))
        search_depth = max(0, _coerce_int(active_filters.get("depth"), int(settings.get("search_depth") or self._search_depth)))
        type_filter = str(active_filters.get("type") or "any").strip().lower()
        needle = str(query or "").strip()
        max_results = max(0, int(limit or 0))
        if max_results <= 0:
            return []

        ssh = self._connect_ssh(settings)
        sftp = None
        try:
            try:
                sftp = self._open_sftp(ssh)
            except Exception as exc:
                if not self._is_sftp_negotiation_error(exc):
                    raise
                return self._search_directory_via_ssh(
                    ssh,
                    start_path,
                    needle=needle,
                    limit=max_results,
                    search_depth=search_depth,
                    type_filter=type_filter,
                    settings=settings,
                )

            return self._search_directory(
                sftp,
                start_path,
                needle=needle,
                limit=max_results,
                search_depth=search_depth,
                type_filter=type_filter,
                settings=settings,
            )
        finally:
            self._close_client(sftp)
            self._close_client(ssh)

    def selector(
        self,
        selected_items: List[Any],
        *,
        ctx: Any,
        stage_is_last: bool = True,
        **_kwargs: Any,
    ) -> bool:
        if not stage_is_last:
            return False

        target_path = ""
        target_title = ""
        instance_name = ""
        for item in selected_items or []:
            metadata = self._item_metadata(item)
            if not metadata.get("is_dir"):
                continue
            settings = self._resolve_settings(instance_name=str(metadata.get("instance") or "").strip() or None, require_explicit=bool(metadata.get("instance")))
            target_path = self._normalize_remote_path(metadata.get("scp_path") or metadata.get("selection_path"), default=str(settings.get("base_path") or "/"))
            target_title = str(metadata.get("title") or metadata.get("name") or "").strip()
            instance_name = str(settings.get("instance") or metadata.get("instance") or "").strip()
            if target_path:
                break

        if not target_path:
            return False

        settings = self._resolve_settings(instance_name=instance_name or None, require_explicit=bool(instance_name))
        ssh = self._connect_ssh(settings)
        sftp = None
        try:
            try:
                sftp = self._open_sftp(ssh)
            except Exception as exc:
                if not self._is_sftp_negotiation_error(exc):
                    raise
                rows = self._search_directory_via_ssh(
                    ssh,
                    target_path,
                    needle="*",
                    limit=500,
                    search_depth=0,
                    type_filter="any",
                    settings=settings,
                )
            else:
                rows = self._search_directory(
                    sftp,
                    target_path,
                    needle="*",
                    limit=500,
                    search_depth=0,
                    type_filter="any",
                    settings=settings,
                )
        finally:
            self._close_client(sftp)
            self._close_client(ssh)

        try:
            from SYS.result_table import Table
            from SYS.rich_display import stdout_console
        except Exception:
            return True

        title = target_title or target_path
        table = Table(f"SCP{f'[{instance_name}]' if instance_name else ''}: {title}")._perseverance(True)
        table.set_table("scp")
        try:
            table.set_table_metadata({
                "plugin": "scp",
                "instance": instance_name or None,
                "host": settings.get("host"),
                "path": target_path,
                "view": "directory",
            })
        except Exception:
            pass
        source_args = ["-plugin", "scp"]
        if instance_name:
            source_args.extend(["-instance", instance_name])
        source_args.extend([f"path:{target_path}", "*"])
        table.set_source_command("search-file", source_args)

        payloads: List[Dict[str, Any]] = []
        for row in rows:
            table.add_result(row)
            payloads.append(row.to_dict())

        try:
            ctx.set_last_result_table(table, payloads, subject={"plugin": "scp", "instance": instance_name or None, "path": target_path})
            ctx.set_current_stage_table(table)
        except Exception:
            pass

        try:
            stdout_console().print()
            stdout_console().print(table)
        except Exception:
            pass

        return True

    def show_selection_details(
        self,
        selected_items: List[Any],
        *,
        ctx: Any,
        stage_is_last: bool = True,
        source_command: str = "",
        table_type: str = "",
        table_metadata: Optional[Dict[str, Any]] = None,
        **_kwargs: Any,
    ) -> bool:
        _ = table_type
        item, _payload, _meta = self.resolve_selection_detail_subject(
            selected_items,
            stage_is_last=stage_is_last,
            source_command=source_command,
            require_media_kind="file",
        )
        if item is None:
            return False

        metadata = self._item_metadata(item)
        if bool(metadata.get("is_dir")):
            return False

        title = str(metadata.get("title") or metadata.get("name") or metadata.get("path") or "").strip() or "SCP Item"
        instance_name = str(metadata.get("instance") or (table_metadata or {}).get("instance") or "").strip()
        scp_url = str(metadata.get("scp_url") or metadata.get("selection_url") or metadata.get("path") or "").strip()
        remote_path = str(metadata.get("scp_path") or "").strip()
        host = str(metadata.get("host") or "").strip()
        modified = str(metadata.get("modified") or "").strip()

        try:
            from SYS.detail_view_helpers import prepare_detail_metadata, render_selection_detail_view
        except Exception:
            return super().show_selection_details(
                selected_items,
                ctx=ctx,
                stage_is_last=stage_is_last,
                source_command=source_command,
                table_type=table_type,
                table_metadata=table_metadata,
            )

        detail_metadata = prepare_detail_metadata(
            item,
            title=title,
            store=instance_name or self.name,
            path=scp_url or remote_path or None,
            tags=metadata.get("tag") or metadata.get("tags"),
            extra_fields={
                "Plugin": self.name,
                "Host": host or None,
                "Instance": instance_name or None,
                "Remote Path": remote_path or None,
                "Directory": str(metadata.get("detail") or "").strip() or None,
                "Modified": modified or None,
                "Scp Url": scp_url or None,
            },
        )

        return render_selection_detail_view(
            ctx=ctx,
            item=item,
            title=f"SCP Item: {title}",
            metadata=detail_metadata,
            table_name=self.name,
            detail_order=["Title", "Instance", "Host", "Remote Path", "Directory", "Modified", "Path", "Ext", "SCP URL", "Plugin"],
            value_case="preserve",
        )

    def download(self, result: SearchResult, output_dir: Path) -> Optional[Path]:
        metadata = getattr(result, "full_metadata", None)
        if isinstance(metadata, dict) and metadata.get("is_dir"):
            return None
        target = str(getattr(result, "path", "") or "").strip()
        if not target:
            return None
        instance_name = str(metadata.get("instance") or "").strip() if isinstance(metadata, dict) else ""
        return self.download_url(target, output_dir, title=getattr(result, "title", None), instance=instance_name or None)

    def download_url(self, url: str, output_dir: Path, **kwargs: Any) -> Optional[Path]:
        parsed = kwargs.get("parsed") if isinstance(kwargs.get("parsed"), dict) else {}
        settings = self._connection_settings_for_url(
            url,
            instance_name=str(kwargs.get("instance") or parsed.get("instance") or "").strip() or None,
        )
        remote_path = settings["path"]
        if not remote_path or remote_path == "/":
            return None

        filename_hint = str(kwargs.get("title") or "").strip()
        parsed_name = posixpath.basename(remote_path.rstrip("/"))
        filename = _safe_filename(filename_hint or unquote(parsed_name) or "download")

        destination_dir = Path(output_dir)
        destination_dir.mkdir(parents=True, exist_ok=True)
        destination = _unique_path(destination_dir / filename)

        ssh = self._connect_ssh(settings)
        scp_client = None
        try:
            scp_client = self._open_scp(ssh)
            scp_client.get(remote_path, local_path=str(destination))
            return destination
        except Exception:
            try:
                destination.unlink(missing_ok=True)
            except Exception:
                pass
            return None
        finally:
            self._close_client(scp_client)
            self._close_client(ssh)

    def resolve_pipe_result_download(
        self,
        result: Any,
        pipe_obj: Any,
    ) -> Tuple[Optional[Path], Optional[str], Optional[Path]]:
        metadata = self._item_metadata(result, pipe_obj=pipe_obj)
        if metadata.get("is_dir"):
            return None, None, None

        download_url = str(
            metadata.get("selection_url")
            or metadata.get("scp_url")
            or metadata.get("path")
            or ""
        ).strip()
        if not download_url.startswith(("scp://", "sftp://")):
            return None, None, None

        temp_dir = Path(tempfile.mkdtemp(prefix="scp-add-file-"))
        downloaded = self.download_url(
            download_url,
            temp_dir,
            title=metadata.get("title"),
            instance=metadata.get("instance"),
        )
        if downloaded is None:
            try:
                temp_dir.rmdir()
            except Exception:
                pass
            return None, None, None

        try:
            if pipe_obj is not None:
                pipe_obj.is_temp = True
        except Exception:
            pass
        return downloaded, None, temp_dir

    def upload(self, file_path: str, **kwargs: Any) -> str:
        local_path = Path(str(file_path or "")).expanduser()
        if not local_path.exists() or not local_path.is_file():
            raise FileNotFoundError(f"File not found: {local_path}")

        pipe_obj = kwargs.get("pipe_obj")

        settings = self._resolve_settings(
            instance_name=str(kwargs.get("instance") or kwargs.get("store") or "").strip() or None,
            require_explicit=bool(kwargs.get("instance") or kwargs.get("store")),
        )
        if not settings.get("host") or not settings.get("username"):
            requested = str(kwargs.get("instance") or kwargs.get("store") or "").strip()
            if requested:
                raise RuntimeError(f"SCP instance '{requested}' is unavailable")
            raise RuntimeError("No configured SCP instance is available")

        remote_dir = self._normalize_remote_path(
            kwargs.get("remote_path") or kwargs.get("path") or settings.get("base_path") or "/",
            default=str(settings.get("base_path") or "/"),
        )
        remote_name = posixpath.basename(str(kwargs.get("remote_name") or local_path.name).replace("\\", "/")) or local_path.name
        remote_path = self._join_remote_path(remote_dir, remote_name)

        ssh = self._connect_ssh(settings)
        sftp = None
        scp_client = None
        try:
            try:
                sftp = self._open_sftp(ssh)
            except Exception as exc:
                if not self._is_sftp_negotiation_error(exc):
                    raise
                self._ensure_directory_via_ssh(ssh, remote_dir)
                if self._remote_filename_exists_via_ssh(ssh, remote_path):
                    try:
                        if pipe_obj is not None:
                            if not isinstance(getattr(pipe_obj, "extra", None), dict):
                                pipe_obj.extra = {}
                            pipe_obj.extra["upload_duplicate"] = True
                            pipe_obj.extra["upload_duplicate_rule"] = "filename"
                            pipe_obj.extra["upload_duplicate_target"] = remote_path
                    except Exception:
                        pass
                    return self._build_url(remote_path, settings=settings)
            else:
                self._ensure_directory(sftp, remote_dir, base_path=str(settings.get("base_path") or "/"))
                if self._remote_filename_exists(sftp, remote_path):
                    try:
                        if pipe_obj is not None:
                            if not isinstance(getattr(pipe_obj, "extra", None), dict):
                                pipe_obj.extra = {}
                            pipe_obj.extra["upload_duplicate"] = True
                            pipe_obj.extra["upload_duplicate_rule"] = "filename"
                            pipe_obj.extra["upload_duplicate_target"] = remote_path
                    except Exception:
                        pass
                    return self._build_url(remote_path, settings=settings)
            scp_client = self._open_scp(ssh)
            scp_client.put(str(local_path), remote_path=remote_path)
        finally:
            self._close_client(scp_client)
            self._close_client(sftp)
            self._close_client(ssh)

        return self._build_url(remote_path, settings=settings)

    def _remote_filename_exists(self, sftp: Any, remote_path: str) -> bool:
        try:
            sftp.stat(remote_path)
            return True
        except Exception:
            return False

    def _remote_filename_exists_via_ssh(self, ssh: Any, remote_path: str) -> bool:
        normalized = self._normalize_remote_path(remote_path, default=self._base_path)
        quoted_path = shlex.quote(normalized)
        status, _, _ = self._run_ssh_command(ssh, f"test -e {quoted_path}")
        return status == 0

    def _run_test_connection(self) -> Dict[str, Any]:
        settings = self._resolve_settings()
        if not settings.get("host"):
            return {"ok": False, "message": "Set 'host' before testing the SCP connection."}
        if not settings.get("username"):
            return {"ok": False, "message": "Set 'username' before testing the SCP connection."}

        ssh = None
        sftp = None
        try:
            ssh = self._connect_ssh(settings)
            base_path = str(settings.get("base_path") or "/")
            transport_detail = "SFTP available"
            try:
                sftp = self._open_sftp(ssh)
            except Exception as exc:
                if not self._is_sftp_negotiation_error(exc):
                    raise
                is_dir = self._path_exists_via_ssh(ssh, base_path)
                transport_detail = "SFTP unavailable; using SSH command fallback"
            else:
                try:
                    attrs = sftp.stat(base_path)
                    is_dir = stat.S_ISDIR(getattr(attrs, "st_mode", 0))
                except Exception:
                    is_dir = False
            detail = f" and confirmed {base_path}" if is_dir else ""
            key_path = str(settings.get("key_path") or "").strip()
            auth_mode = f"key {key_path}" if key_path else "password/agent auth"
            return {
                "ok": True,
                "message": f"Connected to SCP {settings.get('host')}:{settings.get('port')} as {settings.get('username')} via {auth_mode}. {transport_detail}{detail}.",
            }
        except Exception as exc:
            return {"ok": False, "message": f"SCP connection failed: {exc}"}
        finally:
            self._close_client(sftp)
            self._close_client(ssh)

    def _generate_ssh_keypair(self) -> Dict[str, Any]:
        settings = self._resolve_settings()
        key_path = str(settings.get("key_path") or "").strip()
        target = Path(key_path).expanduser() if key_path else (Path.home() / ".ssh" / "medeia_scp_rsa")
        try:
            target.parent.mkdir(parents=True, exist_ok=True)
        except Exception as exc:
            return {"ok": False, "message": f"Could not create key directory: {exc}"}

        public_path = target.with_name(target.name + ".pub")
        if target.exists() or public_path.exists():
            return {
                "ok": False,
                "message": f"SSH key already exists at {target}. Remove it or choose a different key_path first.",
            }

        try:
            key = paramiko.RSAKey.generate(bits=4096)
            key.write_private_key_file(str(target))
            comment = f"{settings.get('username') or 'medeia'}@{settings.get('host') or 'scp'}"
            public_path.write_text(f"{key.get_name()} {key.get_base64()} {comment}\n", encoding="utf-8")
            try:
                target.chmod(0o600)
            except Exception:
                pass
            return {
                "ok": True,
                "message": f"Generated SSH key pair at {target}. Save the config to persist key_path.",
                "config_updates": {"key_path": str(target)},
            }
        except Exception as exc:
            try:
                target.unlink(missing_ok=True)
            except Exception:
                pass
            try:
                public_path.unlink(missing_ok=True)
            except Exception:
                pass
            return {"ok": False, "message": f"SSH key generation failed: {exc}"}

    def _connect_ssh(self, overrides: Optional[Dict[str, Any]] = None) -> paramiko.SSHClient:
        settings = dict(overrides or {})
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(
            hostname=str(settings.get("host") or self._host),
            port=int(settings.get("port") or self._port),
            username=str(settings.get("username") or self._username),
            password=str(settings.get("password") or self._password) or None,
            key_filename=str(settings.get("key_path") or self._key_path) or None,
            timeout=self._timeout,
            allow_agent=self._allow_agent if "allow_agent" not in settings else bool(settings.get("allow_agent")),
            look_for_keys=self._look_for_keys if "look_for_keys" not in settings else bool(settings.get("look_for_keys")),
        )
        return client

    def _open_sftp(self, ssh: Any) -> Any:
        return ssh.open_sftp()

    def _open_scp(self, ssh: Any) -> Any:
        return SCPClient(ssh.get_transport())

    def _is_sftp_negotiation_error(self, exc: Exception) -> bool:
        text = str(exc or "").strip().lower()
        if isinstance(exc, EOFError):
            return True
        return any(
            marker in text
            for marker in (
                "eof during negotiation",
                "open failed",
                "channel closed",
                "administratively prohibited",
                "subsystem request failed",
            )
        )

    def _run_ssh_command(self, ssh: Any, command: str) -> Tuple[int, str, str]:
        stdin, stdout, stderr = ssh.exec_command(command, timeout=self._timeout)
        try:
            stdin.close()
        except Exception:
            pass
        output = stdout.read().decode("utf-8", errors="replace")
        error = stderr.read().decode("utf-8", errors="replace")
        status = 0
        try:
            status = int(stdout.channel.recv_exit_status())
        except Exception:
            status = 0
        return status, output, error

    def _path_exists_via_ssh(self, ssh: Any, remote_path: str) -> bool:
        normalized = self._normalize_remote_path(remote_path, default=self._base_path)
        quoted_path = shlex.quote(normalized)
        status, _, _ = self._run_ssh_command(ssh, f"test -d {quoted_path}")
        return status == 0

    def _ensure_directory_via_ssh(self, ssh: Any, remote_path: str) -> None:
        normalized = self._normalize_remote_path(remote_path, default=self._base_path)
        if normalized == "/":
            return
        quoted_path = shlex.quote(normalized)
        status, _, error = self._run_ssh_command(ssh, f"mkdir -p {quoted_path}")
        if status != 0:
            raise RuntimeError(error.strip() or f"mkdir -p failed for {normalized}")

    def _close_client(self, client: Any) -> None:
        if client is None:
            return
        try:
            client.close()
        except Exception:
            pass

    def _normalize_remote_path(self, value: Any, *, default: str) -> str:
        text = str(value or "").strip().replace("\\", "/")
        if not text:
            text = default
        elif text.startswith(("scp://", "sftp://")):
            try:
                text = unquote(urlparse(text).path or "/")
            except Exception:
                text = default
        elif not text.startswith("/"):
            text = posixpath.join(default, text)

        normalized = posixpath.normpath(text)
        normalized = "/" + normalized.lstrip("/")
        return normalized or "/"

    def _join_remote_path(self, parent: Any, child: Any) -> str:
        left = self._normalize_remote_path(parent, default=self._base_path)
        right = str(child or "").strip().replace("\\", "/")
        if not right:
            return left
        return self._normalize_remote_path(posixpath.join(left, right), default="/")

    def _build_url(
        self,
        remote_path: Any,
        *,
        settings: Optional[Dict[str, Any]] = None,
        host: Optional[str] = None,
        port: Optional[int] = None,
        scheme: str = "scp",
    ) -> str:
        resolved = dict(settings or {})
        path_text = self._normalize_remote_path(remote_path, default="/")
        host_text = str(host or resolved.get("host") or self._host).strip()
        port_value = int(port or resolved.get("port") or self._port)
        port_suffix = f":{port_value}" if port_value and port_value != 22 else ""
        return f"{scheme}://{host_text}{port_suffix}{quote(path_text, safe='/-._~!$&\'()*+,;=:@')}"

    def _connection_settings_for_url(self, url: str, *, instance_name: Optional[str] = None) -> Dict[str, Any]:
        settings = self._resolve_settings(instance_name=instance_name, require_explicit=bool(instance_name))
        parsed = urlparse(str(url or "").strip())
        scheme = (parsed.scheme or "scp").strip().lower()
        host = parsed.hostname or settings.get("host") or self._host
        port = parsed.port or settings.get("port") or self._port
        username = parsed.username or settings.get("username") or self._username
        password = parsed.password or settings.get("password") or self._password
        path_text = self._normalize_remote_path(unquote(parsed.path or "/"), default=str(settings.get("base_path") or "/"))
        return {
            "instance": settings.get("instance"),
            "scheme": scheme,
            "host": host,
            "port": port,
            "username": username,
            "password": password,
            "key_path": settings.get("key_path") or self._key_path,
            "allow_agent": settings.get("allow_agent", self._allow_agent),
            "look_for_keys": settings.get("look_for_keys", self._look_for_keys),
            "path": path_text,
            "timeout": settings.get("timeout", self._timeout),
            "base_path": settings.get("base_path", self._base_path),
        }

    def _search_directory(
        self,
        sftp: Any,
        start_path: str,
        *,
        needle: str,
        limit: int,
        search_depth: int,
        type_filter: str,
        settings: Dict[str, Any],
    ) -> List[SearchResult]:
        results: List[SearchResult] = []
        visited: set[str] = set()

        def walk(current_path: str, depth_left: int) -> None:
            normalized = self._normalize_remote_path(current_path, default=str(settings.get("base_path") or self._base_path))
            if normalized in visited or len(results) >= limit:
                return
            visited.add(normalized)

            for entry in self._list_directory(sftp, normalized):
                if len(results) >= limit:
                    return
                if self._matches_entry(entry, needle=needle, type_filter=type_filter):
                    results.append(self._build_result(entry, settings=settings))
                if entry.get("is_dir") and depth_left > 0:
                    walk(str(entry.get("scp_path") or normalized), depth_left - 1)

        walk(start_path, max(0, search_depth))
        return results

    def _search_directory_via_ssh(
        self,
        ssh: Any,
        start_path: str,
        *,
        needle: str,
        limit: int,
        search_depth: int,
        type_filter: str,
        settings: Dict[str, Any],
    ) -> List[SearchResult]:
        entries = self._list_directory_via_ssh(ssh, start_path, depth=search_depth)
        results: List[SearchResult] = []
        for entry in entries:
            if len(results) >= limit:
                break
            if self._matches_entry(entry, needle=needle, type_filter=type_filter):
                results.append(self._build_result(entry, settings=settings))
        return results

    def _matches_entry(self, entry: Dict[str, Any], *, needle: str, type_filter: str) -> bool:
        is_dir = bool(entry.get("is_dir"))
        if type_filter in {"dir", "dirs", "folder", "folders"} and not is_dir:
            return False
        if type_filter in {"file", "files"} and is_dir:
            return False

        text = str(needle or "").strip().lower()
        if not text or text in {"*", "all", "list"}:
            return True

        haystacks = [
            str(entry.get("name") or "").lower(),
            str(entry.get("scp_path") or "").lower(),
        ]
        for token in [part for part in text.split() if part]:
            if any(ch in token for ch in "*?[]"):
                if not any(fnmatch.fnmatch(haystack, token) for haystack in haystacks):
                    return False
            elif not any(token in haystack for haystack in haystacks):
                return False
        return True

    def _build_result(self, entry: Dict[str, Any], *, settings: Dict[str, Any]) -> SearchResult:
        scp_path = str(entry.get("scp_path") or "/")
        scp_url = self._build_url(scp_path, settings=settings)
        is_dir = bool(entry.get("is_dir"))
        size_value = entry.get("size")
        modified = str(entry.get("modified") or "")
        parent = posixpath.dirname(scp_path.rstrip("/")) or "/"
        instance_name = str(settings.get("instance") or "").strip()
        metadata = {
            "plugin": "scp",
            "instance": instance_name or None,
            "host": settings.get("host"),
            "scp_path": scp_path,
            "scp_url": scp_url,
            "selection_url": scp_url,
            "is_dir": is_dir,
            "name": str(entry.get("name") or "").strip(),
        }
        if size_value is not None:
            metadata["size"] = size_value
        if modified:
            metadata["modified"] = modified

        selection_args = ["-url", scp_url]
        selection_action = ["download-file", "-plugin", "scp"]
        if instance_name:
            selection_args = ["-instance", instance_name, *selection_args]
            selection_action.extend(["-instance", instance_name])
        selection_action.extend(["-url", scp_url])

        return SearchResult(
            table="scp",
            title=str(entry.get("name") or scp_path),
            path=scp_url,
            detail=parent,
            annotations=["folder" if is_dir else "file"],
            media_kind="folder" if is_dir else "file",
            size_bytes=int(size_value) if isinstance(size_value, int) else None,
            tag={"scp", "folder" if is_dir else "file"},
            columns=[
                ("Name", str(entry.get("name") or "")),
                ("Type", "dir" if is_dir else "file"),
                ("Directory", parent),
                ("Size", "" if size_value is None else str(size_value)),
                ("Modified", modified),
            ],
            selection_args=None if is_dir else selection_args,
            selection_action=None if is_dir else selection_action,
            full_metadata=metadata,
        )

    def _list_directory(self, sftp: Any, remote_path: str) -> List[Dict[str, Any]]:
        try:
            attrs = sftp.listdir_attr(remote_path)
        except Exception:
            return []

        entries: List[Dict[str, Any]] = []
        for attr in attrs:
            name_text = str(getattr(attr, "filename", "") or "").strip()
            if not name_text or name_text in {".", ".."}:
                continue
            mode = getattr(attr, "st_mode", 0)
            is_dir = stat.S_ISDIR(mode)
            size_value = getattr(attr, "st_size", None)
            try:
                size_int = int(size_value) if size_value is not None else None
            except Exception:
                size_int = None
            entries.append(
                {
                    "name": name_text,
                    "scp_path": self._join_remote_path(remote_path, name_text),
                    "is_dir": is_dir,
                    "size": size_int,
                    "modified": _format_epoch(getattr(attr, "st_mtime", None)),
                }
            )
        return entries

    def _list_directory_via_ssh(self, ssh: Any, remote_path: str, *, depth: int) -> List[Dict[str, Any]]:
        normalized = self._normalize_remote_path(remote_path, default=self._base_path)
        max_depth = max(1, int(depth) + 1)
        quoted_path = shlex.quote(normalized)
        command = (
            f"find {quoted_path} -mindepth 1 -maxdepth {max_depth} "
            f"\\( -type d -o -type f \\) -exec sh -c 'for path do "
            f"if [ -d \"$path\" ]; then kind=d; else kind=f; fi; "
            f"name=$(basename \"$path\"); "
            f"printf \"%s\\0%s\\0%s\\0\" \"$kind\" \"$path\" \"$name\"; "
            f"done' sh {{}} +"
        )
        status, output, error = self._run_ssh_command(ssh, command)
        if status != 0:
            error_text = error.strip().lower()
            if "no such file" in error_text or "cannot access" in error_text:
                return []
            raise RuntimeError(error.strip() or f"SSH listing failed for {normalized}")

        chunks = [part for part in output.split("\0") if part]
        entries: List[Dict[str, Any]] = []
        for index in range(0, len(chunks), 3):
            if index + 2 >= len(chunks):
                break
            kind = chunks[index]
            scp_path = self._normalize_remote_path(chunks[index + 1], default=normalized)
            name_text = str(chunks[index + 2] or "").strip()
            if not name_text or name_text in {".", ".."}:
                continue
            entries.append(
                {
                    "name": name_text,
                    "scp_path": scp_path,
                    "is_dir": kind == "d",
                    "size": None,
                    "modified": "",
                }
            )
        return entries

    def _ensure_directory(self, sftp: Any, remote_path: str, *, base_path: str) -> None:
        normalized = self._normalize_remote_path(remote_path, default=base_path)
        if normalized == "/":
            return
        partial = ""
        for segment in [part for part in normalized.split("/") if part]:
            partial = f"{partial}/{segment}"
            try:
                attrs = sftp.stat(partial)
                if stat.S_ISDIR(getattr(attrs, "st_mode", 0)):
                    continue
            except Exception:
                pass
            try:
                sftp.mkdir(partial)
            except Exception:
                try:
                    attrs = sftp.stat(partial)
                    if stat.S_ISDIR(getattr(attrs, "st_mode", 0)):
                        continue
                except Exception:
                    pass
                raise

    def _item_metadata(self, item: Any, *, pipe_obj: Any = None) -> Dict[str, Any]:
        metadata: Dict[str, Any] = {}
        for source in (item, pipe_obj):
            if isinstance(source, dict):
                for key in ("title", "path", "url"):
                    if source.get(key) is not None and key not in metadata:
                        metadata[key] = source.get(key)
                nested = source.get("full_metadata") or source.get("metadata")
                if isinstance(nested, dict):
                    metadata.update(nested)
            elif source is not None:
                for attr in ("title", "path", "url"):
                    try:
                        value = getattr(source, attr, None)
                    except Exception:
                        value = None
                    if value is not None and attr not in metadata:
                        metadata[attr] = value
                try:
                    nested = getattr(source, "full_metadata", None) or getattr(source, "metadata", None)
                except Exception:
                    nested = None
                if isinstance(nested, dict):
                    metadata.update(nested)

        scp_path = metadata.get("scp_path") or metadata.get("selection_path")
        if not scp_path:
            path_value = metadata.get("path") or metadata.get("url") or metadata.get("scp_url")
            path_text = str(path_value or "").strip()
            if path_text.startswith(("scp://", "sftp://")):
                scp_path = self._normalize_remote_path(path_text, default=self._base_path)
        if scp_path:
            base_path = str(metadata.get("base_path") or self._base_path)
            metadata["scp_path"] = self._normalize_remote_path(scp_path, default=base_path)
            metadata.setdefault("selection_path", metadata["scp_path"])

        if metadata.get("scp_path") and not metadata.get("scp_url"):
            metadata["scp_url"] = self._build_url(
                metadata["scp_path"],
                settings={
                    "host": metadata.get("host") or self._host,
                    "instance": metadata.get("instance"),
                },
            )
        if metadata.get("scp_url") and not metadata.get("selection_url"):
            metadata["selection_url"] = metadata["scp_url"]

        is_dir = metadata.get("is_dir")
        if is_dir is None and metadata.get("media_kind"):
            is_dir = str(metadata.get("media_kind") or "").strip().lower() == "folder"
        metadata["is_dir"] = bool(is_dir)
        return metadata