lang_integration.py

import glob
import os
import os.path
import re
import socket
from contextlib import closing
from functools import partial
from string import Template

import sublime
import sublime_plugin

SETTINGS_FILE = "SublimeREPL.sublime-settings"

class ClojureAutoTelnetRepl(sublime_plugin.WindowCommand):
    def is_running(self, port_str):
        """Check if port is open on localhost"""
        port = int(port_str)
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        res = s.connect_ex(("127.0.0.1", port))
        s.close()
        return res == 0

    def choices(self):
        choices = []
        for folder in self.window.folders():
            proj_file = os.path.join(folder, "project.clj")
            try:
                with open(proj_file) as f:
                    data = f.read()
                    port_match = re.search(r":repl-port\s+(\d{1,})", data)
                    if not port_match:
                        continue
                    port = port_match.group(1)
                    description = proj_file
                    desc_match = re.search(r':description\s+"([^"]+)"', data)
                    if desc_match:
                        description = desc_match.group(1)
                    if self.is_running(port):
                        description += " (active)"
                    else:
                        description += " (not responding)"
                    choices.append([description, port])
            except IOError as e:
                pass  # just ignore it, no file or no access

        return choices + [["Custom telnet", "Pick your own telnet port number to Lein REPL"]]

    def run(self):
        choices = self.choices()
        if len(choices) == 1: #only custom telnet action
            self.on_done(choices, 0)
        else:
            on_done = partial(self.on_done, choices)
            self.window.show_quick_panel(self.choices(), on_done)

    def on_done(self, choices, index):
        if index == -1:
            return
        if index == len(choices) - 1:
            self.window.show_input_panel("Enter port number", "",
                                         self.open_telnet_repl,
                                         None, None)
            return
        self.open_telnet_repl(choices[index][1])

    def open_telnet_repl(self, port_str):
        try:
            port = int(port_str)
        except ValueError:
            return
        self.window.run_command("repl_open", {"type":"telnet", "encoding":"utf8", "host":"localhost", "port":port,
                                "external_id":"clojure", "syntax":"Packages/Clojure/Clojure.tmLanguage"})


def is_venv(venv_root):
    """True if `venv_root` is a modern virtualenv: pyvenv.cfg plus an
       interpreter in bin/ (or Scripts/ on Windows)."""
    bin_dir = "Scripts" if os.name == "nt" else "bin"
    python = os.path.join(venv_root, bin_dir, "python")
    if os.name == "nt":
        python += ".exe"
    return os.path.isfile(os.path.join(venv_root, "pyvenv.cfg")) and os.path.isfile(python)


def scan_for_virtualenvs(venv_paths):
    """Find virtualenvs living directly under each configured base dir.

    Detection relies on pyvenv.cfg + interpreter, the layout shared by modern
    tools (python3 -m venv, uv, virtualenv, poetry, pipenv), instead of the
    legacy activate_this.py marker."""
    found_dirs = set()
    for venv_path in venv_paths:
        base = os.path.expanduser(venv_path)
        if is_venv(base):
            found_dirs.add(base)
        for pyvenv_cfg in glob.glob(os.path.join(base, "*", "pyvenv.cfg")):
            venv_root = os.path.dirname(pyvenv_cfg)
            if is_venv(venv_root):
                found_dirs.add(venv_root)
    return sorted(found_dirs)


def _subst_for_anchors(anchors):
    """Minimal substitution map used to expand python_virtualenv_paths
       entries. `$folder`/`$file_path` are overridden per anchor during
       expansion."""
    return {
        "packages": sublime.packages_path(),
        "installed_packages": sublime.installed_packages_path(),
        "folder": anchors[0] if anchors else "",
        "file_path": anchors[0] if anchors else "",
    }


def _expand_venv_paths(venv_paths, anchors, subst):
    """Expand each python_virtualenv_paths entry against the anchors.

    Entries referencing `$folder`/`$file_path` are expanded once per anchor
    (file directory first, then project/workspace folders), which is what
    makes project-local venvs and multi-folder workspaces resolve. Yields
    (path, anchor_index); anchor_index is None for static entries."""
    seen = set()
    expanded = []
    for entry in venv_paths:
        if "$folder" in entry or "$file_path" in entry:
            for i, anchor in enumerate(anchors):
                per_anchor = dict(subst)
                per_anchor["folder"] = anchor
                per_anchor["file_path"] = anchor
                path = Template(entry).safe_substitute(**per_anchor).strip()
                key = (path, i)
                if path and key not in seen:
                    seen.add(key)
                    expanded.append((path, i))
        else:
            path = Template(entry).safe_substitute(**subst).strip()
            key = (path, None)
            if path and key not in seen:
                seen.add(key)
                expanded.append((path, None))
    return expanded


def _categorize_venv_paths(venv_paths, anchors, subst):
    """Returns (anchored_direct, static_direct, base_dirs): virtualenvs the
       config points at directly split by provenance, and base directories to
       scan one level deep."""
    anchored_direct = []
    static_direct = []
    base_dirs = set()
    for path, anchor_index in _expand_venv_paths(venv_paths, anchors, subst):
        p = os.path.abspath(os.path.expanduser(path))
        if is_venv(p):
            if anchor_index is not None:
                anchored_direct.append((anchor_index, p))
            else:
                static_direct.append(p)
        else:
            base_dirs.add(os.path.expanduser(path))
    anchored_direct.sort()
    return anchored_direct, static_direct, base_dirs


def discover_venvs(anchors, venv_paths):
    """All virtualenv roots reachable through python_virtualenv_paths for the
       given anchors, sorted. Used by the virtualenv quick-panel command."""
    if not anchors or not venv_paths:
        return []
    subst = _subst_for_anchors(anchors)
    anchored_direct, static_direct, base_dirs = _categorize_venv_paths(venv_paths, anchors, subst)
    found = set(p for _, p in anchored_direct)
    found.update(static_direct)
    found.update(scan_for_virtualenvs(sorted(base_dirs)))
    return sorted(found)


def resolve_venv_for(anchors, venv_paths):
    """The virtualenv to use for the active file, or None.

    `python_virtualenv_paths` is the single source of truth. A virtualenv hit
    directly by a `$folder`/`$file_path` entry wins in anchor order (the file's
    directory before any project folder), then a directly configured root, then
    a virtualenv under a configured base directory whose name matches an
    anchor (virtualenvwrapper / venv.bash layout and the hashed dirs used by
    poetry and pipenv)."""
    if not anchors or not venv_paths:
        return None
    subst = _subst_for_anchors(anchors)
    anchored_direct, static_direct, base_dirs = _categorize_venv_paths(venv_paths, anchors, subst)
    if anchored_direct:
        return anchored_direct[0][1]
    if static_direct:
        return static_direct[0]
    children = scan_for_virtualenvs(sorted(base_dirs))
    by_name = {}
    for root in children:
        by_name.setdefault(os.path.basename(root), root)
    for anchor in anchors:
        name = os.path.basename(os.path.abspath(anchor))
        if name in by_name:
            return by_name[name]
        for key, root in by_name.items():
            if key.startswith(name + "-") or key.startswith(name + "_"):
                return root
    return None


class PythonVirtualenvRepl(sublime_plugin.WindowCommand):
    def _anchors(self):
        anchors = []
        av = self.window.active_view()
        if av and av.file_name():
            anchors.append(os.path.dirname(av.file_name()))
        for folder in self.window.folders():
            if folder not in anchors:
                anchors.append(folder)
        return anchors

    def _scan(self):
        venv_paths = sublime.load_settings(SETTINGS_FILE).get("python_virtualenv_paths", [])
        return discover_venvs(self._anchors(), venv_paths)

    def run_virtualenv(self, choices, index):
        if index == -1:
            return
        (name, directory) = choices[index]
        bin_dir = "Scripts" if os.name == "nt" else "bin"
        activate_file = os.path.join(directory, bin_dir, "activate_this.py")
        python_executable = os.path.join(directory, bin_dir, "python")
        path_separator = ":"
        if os.name == "nt":
            python_executable += ".exe"
            path_separator = ";"

        extend_env = {
            "PATH": os.path.join(directory, bin_dir) + path_separator + "{PATH}",
            "PYTHONIOENCODING": "utf-8"
        }
        if os.path.isfile(activate_file):
            extend_env["SUBLIMEREPL_ACTIVATE_THIS"] = activate_file

        self.window.run_command("repl_open",
            {
                "encoding": "utf8",
                "type": "subprocess",
                "autocomplete_server": True,
                "extend_env": extend_env,
                "cmd": [python_executable, "-u", "${packages}/SublimeREPL/config/Python/ipy_repl.py"],
                "cwd": "$file_path",
                "syntax": "Packages/Python/Python.tmLanguage",
                "external_id": "python"
             })

    def run(self):
        choices = self._scan()
        nice_choices = [[os.path.basename(path), path] for path in choices]
        self.window.show_quick_panel(nice_choices, partial(self.run_virtualenv, nice_choices))


VENV_SCAN_CODE = """
import os
import glob
import os.path

venv_paths = channel.receive()
bin_dir = "Scripts" if os.name == "nt" else "bin"
found_dirs = set()
for venv_path in venv_paths:
    p = os.path.expanduser(venv_path)
    pattern = os.path.join(p, "*", bin_dir, "activate_this.py")
    found_dirs.update(map(os.path.dirname, glob.glob(pattern)))

channel.send(found_dirs)
channel.close()

"""

class ExecnetVirtualenvRepl(sublime_plugin.WindowCommand):
    def run(self):
        self.window.show_input_panel("SSH connection (eg. user@host)", "", self.on_ssh_select, None, None)

    def on_ssh_select(self, host_string):
        import execnet
        venv_paths = sublime.load_settings(SETTINGS_FILE).get("python_virtualenv_paths", [])
        try:
            gw = execnet.makegateway("ssh=" + host_string)
            ch = gw.remote_exec(VENV_SCAN_CODE)
        except Exception as e:
            sublime.error_message(repr(e))
            return

        with closing(ch):
            ch.send(venv_paths)
            directories = ch.receive(60)
        gw.exit()

        choices = [[host_string + ":" + path.split(os.path.sep)[-2], path] for path in sorted(directories)]
        nice_choices = [["w/o venv", "n/a"]] + choices
        self.window.show_quick_panel(nice_choices, partial(self.run_virtualenv, host_string, nice_choices))

    def run_virtualenv(self, host_string, nice_choices, index):
        if index == -1:
            return
        if index == 0:
            connection_string = "ssh={host}".format(host=host_string)
            ps1 = "({host}@) >>> ".format(host=host_string)
            activate_file = ""
        else:
            (name, directory) = nice_choices[index]
            activate_file = os.path.join(directory, "activate_this.py")
            python_file = os.path.join(directory, "python")
            ps1 = "({name}) >>> ".format(name=name, host=host_string)
            connection_string = "ssh={host}//env:PATH={dir}//python={python}".format(
                host=host_string,
                dir=directory,
                python=python_file
            )

        self.window.run_command("repl_open",
                 {
                  "type": "execnet_repl",
                  "encoding": "utf8",
                  "syntax": "Packages/Python/Python.tmLanguage",
                  "connection_string": connection_string,
                  "activate_file": activate_file,
                  "ps1": ps1
                 })