python_arg_launcher.pyw

"""Windows Explorer launcher that turns conventional argparse declarations into a GUI.

The selected script is parsed as source with ``ast``.  It is never imported or
executed during inspection.  Dynamic parsers and non-argparse programs remain
usable through the raw-arguments field.
"""
from __future__ import annotations

import ast
import ctypes
import os
import shutil
import subprocess
import sys
import threading
import tkinter as tk
from dataclasses import dataclass, field
from pathlib import Path
from tkinter import filedialog, messagebox, ttk
from typing import Any


APP_TITLE = "Run Python with arguments"
RUNNER = Path(__file__).with_name("python_console_runner.py")


@dataclass
class ArgumentSpec:
    flags: tuple[str, ...]
    dest: str
    positional: bool
    required: bool
    action: str = "store"
    nargs: str | int | None = None
    choices: tuple[Any, ...] = ()
    default: Any = None
    type_name: str = "str"
    help: str = ""
    metavar: str = ""

    @property
    def option(self) -> str:
        long_flags = [flag for flag in self.flags if flag.startswith("--")]
        return (long_flags or list(self.flags) or [self.dest])[-1]


@dataclass
class ParserModel:
    description: str = ""
    root: list[ArgumentSpec] = field(default_factory=list)
    commands: dict[str, list[ArgumentSpec]] = field(default_factory=dict)
    dynamic_notes: list[str] = field(default_factory=list)


def _literal(node: ast.AST | None, default: Any = None) -> Any:
    if node is None:
        return default
    try:
        return ast.literal_eval(node)
    except (ValueError, TypeError):
        return default


def _call_name(node: ast.AST) -> str:
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        return node.attr
    return ""


def _target_name(node: ast.AST) -> str | None:
    return node.id if isinstance(node, ast.Name) else None


def _keyword_map(call: ast.Call) -> dict[str, ast.AST]:
    return {kw.arg: kw.value for kw in call.keywords if kw.arg}


def _argument_from_call(call: ast.Call, model: ParserModel) -> ArgumentSpec | None:
    raw_flags: list[str] = []
    for node in call.args:
        value = _literal(node)
        if isinstance(value, str):
            raw_flags.append(value)
        else:
            model.dynamic_notes.append(
                f"Line {getattr(call, 'lineno', '?')}: dynamic argument name"
            )
            return None
    if not raw_flags:
        return None

    kw = _keyword_map(call)
    positional = not any(flag.startswith("-") for flag in raw_flags)
    dest = str(_literal(kw.get("dest"), ""))
    if not dest:
        dest = raw_flags[0] if positional else raw_flags[-1].lstrip("-").replace("-", "_")
    action = str(_literal(kw.get("action"), "store"))
    nargs = _literal(kw.get("nargs"))
    default = _literal(kw.get("default"))
    choices_value = _literal(kw.get("choices"), ())
    choices = tuple(choices_value) if isinstance(choices_value, (list, tuple, range, set)) else ()
    required_kw = bool(_literal(kw.get("required"), False))
    required = required_kw or (positional and nargs not in ("?", "*"))
    type_node = kw.get("type")
    type_name = _call_name(type_node) if type_node is not None else "str"
    help_text = str(_literal(kw.get("help"), "") or "")
    metavar = str(_literal(kw.get("metavar"), "") or "")
    return ArgumentSpec(
        flags=tuple(raw_flags),
        dest=dest,
        positional=positional,
        required=required,
        action=action,
        nargs=nargs,
        choices=choices,
        default=default,
        type_name=type_name,
        help=help_text,
        metavar=metavar,
    )


class _ParserVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.model = ParserModel()
        self.parsers: dict[str, str | None] = {}
        self.subparser_owners: dict[str, str] = {}

    def visit_Assign(self, node: ast.Assign) -> None:
        target = _target_name(node.targets[0]) if len(node.targets) == 1 else None
        call = node.value if isinstance(node.value, ast.Call) else None
        if target and call:
            name = _call_name(call.func)
            if name == "ArgumentParser":
                self.parsers[target] = None
                description = _literal(_keyword_map(call).get("description"), "")
                if description and not self.model.description:
                    self.model.description = str(description)
            elif isinstance(call.func, ast.Attribute) and isinstance(call.func.value, ast.Name):
                owner = call.func.value.id
                if name == "add_subparsers" and owner in self.parsers:
                    self.subparser_owners[target] = owner
                elif name == "add_parser" and owner in self.subparser_owners:
                    command = _literal(call.args[0]) if call.args else None
                    if isinstance(command, str):
                        self.parsers[target] = command
                        self.model.commands.setdefault(command, [])
                    else:
                        self.model.dynamic_notes.append(
                            f"Line {node.lineno}: dynamic subcommand name"
                        )
        self.generic_visit(node)

    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
        if node.value is not None:
            proxy = ast.Assign(targets=[node.target], value=node.value)
            proxy.lineno = node.lineno
            self.visit_Assign(proxy)

    def visit_Call(self, node: ast.Call) -> None:
        if (
            isinstance(node.func, ast.Attribute)
            and node.func.attr == "add_argument"
            and isinstance(node.func.value, ast.Name)
        ):
            parser_name = node.func.value.id
            if parser_name in self.parsers:
                spec = _argument_from_call(node, self.model)
                if spec and spec.action not in ("help", "version"):
                    command = self.parsers[parser_name]
                    if command is None:
                        self.model.root.append(spec)
                    else:
                        self.model.commands.setdefault(command, []).append(spec)
        self.generic_visit(node)


def inspect_source(source: str) -> ParserModel:
    try:
        tree = ast.parse(source)
    except SyntaxError as exc:
        model = ParserModel()
        model.dynamic_notes.append(f"Source could not be parsed: {exc}")
        return model
    visitor = _ParserVisitor()
    visitor.visit(tree)
    if not visitor.parsers:
        visitor.model.dynamic_notes.append("No conventional argparse parser was found.")
    return visitor.model


def inspect_script(path: Path) -> ParserModel:
    try:
        return inspect_source(path.read_text(encoding="utf-8-sig"))
    except UnicodeDecodeError:
        try:
            return inspect_source(path.read_text(encoding=sys.getfilesystemencoding()))
        except Exception as exc:  # noqa: BLE001 - surfaced in the GUI
            return ParserModel(dynamic_notes=[f"Could not read source: {exc}"])
    except OSError as exc:
        return ParserModel(dynamic_notes=[f"Could not read source: {exc}"])


def _console_executable(path: Path) -> Path:
    if path.name.lower() == "pythonw.exe":
        sibling = path.with_name("python.exe")
        if sibling.is_file():
            return sibling
    return path


def interpreter_candidates(script: Path) -> list[str]:
    found: list[Path] = []
    for parent in (script.parent, *list(script.parents)[:7]):
        for env_name in (".venv", "venv", f"{parent.name}_venv"):
            candidate = parent / env_name / "Scripts" / "python.exe"
            if candidate.is_file():
                found.append(candidate)
    found.append(_console_executable(Path(sys.executable)))
    on_path = shutil.which("python")
    if on_path:
        found.append(Path(on_path))
    result: list[str] = []
    seen: set[str] = set()
    for path in found:
        normalized = str(path.resolve())
        key = normalized.casefold()
        if key not in seen:
            seen.add(key)
            result.append(normalized)
    return result


def split_windows_command_line(text: str) -> list[str]:
    text = text.strip()
    if not text:
        return []
    argc = ctypes.c_int()
    command_line_to_argv = ctypes.windll.shell32.CommandLineToArgvW
    command_line_to_argv.argtypes = [ctypes.c_wchar_p, ctypes.POINTER(ctypes.c_int)]
    command_line_to_argv.restype = ctypes.POINTER(ctypes.c_wchar_p)
    argv = command_line_to_argv("launcher " + text, ctypes.byref(argc))
    if not argv:
        raise ValueError("Windows could not parse the argument text")
    try:
        return [argv[index] for index in range(1, argc.value)]
    finally:
        local_free = ctypes.windll.kernel32.LocalFree
        local_free.argtypes = [ctypes.c_void_p]
        local_free.restype = ctypes.c_void_p
        local_free(argv)


class ArgumentField:
    def __init__(self, parent: ttk.Frame, spec: ArgumentSpec, row: int,
                 changed, initial_dir: Path) -> None:
        self.spec = spec
        self.initial_dir = initial_dir
        self.var: tk.Variable
        label = spec.option if not spec.positional else (spec.metavar or spec.dest)
        if spec.required:
            label += "  *"
        ttk.Label(parent, text=label, width=25).grid(row=row, column=0, sticky="nw", padx=(4, 10), pady=4)

        if spec.action in ("store_true", "store_false", "store_const", "append_const"):
            self.var = tk.BooleanVar(value=False)
            widget = ttk.Checkbutton(parent, text=f"include {spec.option}", variable=self.var, command=changed)
        elif spec.choices:
            self.var = tk.StringVar()
            widget = ttk.Combobox(parent, textvariable=self.var, values=[str(v) for v in spec.choices], state="readonly")
            widget.bind("<<ComboboxSelected>>", lambda _event: changed())
        else:
            self.var = tk.StringVar()
            widget = ttk.Entry(parent, textvariable=self.var)
            self.var.trace_add("write", lambda *_args: changed())
        widget.grid(row=row, column=1, sticky="ew", pady=4)

        browse_kind = self._browse_kind()
        if browse_kind:
            ttk.Button(parent, text="Browse…", command=lambda: self._browse(browse_kind)).grid(
                row=row, column=2, padx=(8, 4), pady=4
            )
        detail = spec.help
        if spec.default not in (None, "", False):
            detail += (" · " if detail else "") + f"default: {spec.default}"
        if spec.type_name not in ("", "str"):
            detail += (" · " if detail else "") + f"type: {spec.type_name}"
        if detail:
            ttk.Label(parent, text=detail, foreground="#555555", wraplength=520).grid(
                row=row + 1, column=1, columnspan=2, sticky="w", pady=(0, 5)
            )
        self.rows_used = 2 if detail else 1

    def _browse_kind(self) -> str | None:
        name = self.spec.dest.casefold()
        if "dir" in name or "folder" in name:
            return "dir"
        if any(word in name for word in ("file", "path", "database", "db", "input", "output")):
            return "save" if any(word in name for word in ("output", "out", "save", "dest")) else "open"
        return None

    def _browse(self, kind: str) -> None:
        dialog_options = {"initialdir": str(self.initial_dir)}
        if kind == "dir":
            value = filedialog.askdirectory(**dialog_options)
        elif kind == "save":
            value = filedialog.asksaveasfilename(**dialog_options)
        else:
            value = filedialog.askopenfilename(**dialog_options)
        if value:
            self.var.set(value)

    def arguments(self) -> list[str]:
        if isinstance(self.var, tk.BooleanVar):
            if self.spec.required and not self.var.get():
                raise ValueError(f"{self.spec.option} is required")
            return [self.spec.option] if self.var.get() else []
        value = str(self.var.get()).strip()
        if not value:
            if self.spec.required:
                raise ValueError(f"{self.spec.option} is required")
            return []
        values = split_windows_command_line(value) if self.spec.nargs in ("+", "*", 2, 3, 4) else [value]
        if self.spec.type_name == "int":
            for item in values:
                int(item)
        elif self.spec.type_name == "float":
            for item in values:
                float(item)
        if self.spec.positional:
            return values
        result = [self.spec.option]
        result.extend(values)
        return result


class Launcher:
    def __init__(self, script: Path, supervised: bool = False) -> None:
        self.script = script.resolve()
        self.supervised = supervised
        self.app_title = "Python Guardian" if supervised else APP_TITLE
        self.model = inspect_script(self.script)
        self.root = tk.Tk()
        self.root.title(f"{self.app_title} — {self.script.name}")
        self.root.geometry("920x820" if supervised else "920x720")
        self.root.minsize(760, 560)
        self.fields: list[ArgumentField] = []
        self._build()

    def _build(self) -> None:
        outer = ttk.Frame(self.root, padding=12)
        outer.pack(fill="both", expand=True)
        ttk.Label(outer, text=self.script.name, font=("Segoe UI", 13, "bold")).pack(anchor="w")
        ttk.Label(outer, text=str(self.script), foreground="#555555").pack(anchor="w", pady=(0, 8))
        if self.model.description:
            ttk.Label(outer, text=self.model.description, wraplength=860).pack(anchor="w", pady=(0, 8))

        top = ttk.Frame(outer)
        top.pack(fill="x", pady=(0, 8))
        ttk.Label(top, text="Python interpreter").grid(row=0, column=0, sticky="w", padx=(0, 8))
        candidates = interpreter_candidates(self.script)
        self.interpreter = tk.StringVar(value=candidates[0] if candidates else "python")
        interpreter_box = ttk.Combobox(top, textvariable=self.interpreter, values=candidates)
        interpreter_box.grid(row=0, column=1, sticky="ew")
        self.interpreter.trace_add("write", lambda *_args: self._update_preview())
        top.columnconfigure(1, weight=1)

        self.command = tk.StringVar()
        if self.model.commands:
            command_row = ttk.Frame(outer)
            command_row.pack(fill="x", pady=(0, 8))
            ttk.Label(command_row, text="Subcommand").pack(side="left", padx=(0, 8))
            self.command.set(next(iter(self.model.commands)))
            box = ttk.Combobox(command_row, textvariable=self.command,
                               values=list(self.model.commands), state="readonly", width=30)
            box.pack(side="left")
            box.bind("<<ComboboxSelected>>", lambda _event: self._render_fields())

        canvas_frame = ttk.Frame(outer)
        canvas_frame.pack(fill="both", expand=True)
        self.canvas = tk.Canvas(canvas_frame, highlightthickness=0)
        scroll = ttk.Scrollbar(canvas_frame, orient="vertical", command=self.canvas.yview)
        self.form = ttk.Frame(self.canvas)
        self.form.columnconfigure(1, weight=1)
        window = self.canvas.create_window((0, 0), window=self.form, anchor="nw")
        self.form.bind("<Configure>", lambda _e: self.canvas.configure(scrollregion=self.canvas.bbox("all")))
        self.canvas.bind("<Configure>", lambda e: self.canvas.itemconfigure(window, width=e.width))
        self.canvas.configure(yscrollcommand=scroll.set)
        self.canvas.pack(side="left", fill="both", expand=True)
        scroll.pack(side="right", fill="y")
        self._render_fields()

        raw_frame = ttk.LabelFrame(outer, text="Additional/raw arguments", padding=8)
        raw_frame.pack(fill="x", pady=(8, 6))
        self.raw = tk.StringVar()
        raw_entry = ttk.Entry(raw_frame, textvariable=self.raw)
        raw_entry.pack(fill="x")
        self.raw.trace_add("write", lambda *_args: self._update_preview())
        ttk.Label(raw_frame, text="Fallback for dynamic argparse, Click, Typer, or hand-parsed sys.argv.",
                  foreground="#555555").pack(anchor="w")

        if self.model.dynamic_notes:
            ttk.Label(outer, text="Inspection note: " + "  ".join(self.model.dynamic_notes),
                      foreground="#8a5500", wraplength=860).pack(anchor="w", pady=(0, 6))

        if self.supervised:
            guardian_frame = ttk.LabelFrame(outer, text="Codex read-only preflight", padding=6)
            guardian_frame.pack(fill="x", pady=(2, 8))
            self.guardian_text = tk.Text(guardian_frame, height=8, wrap="word", background="#f4f4f4")
            self.guardian_text.pack(fill="x")
            self._set_guardian_text("Connecting to the resident Guardian and reviewing this script…")
            threading.Thread(target=self._load_preflight, daemon=True).start()

        controls = ttk.Frame(outer)
        controls.pack(fill="x")
        ttk.Label(controls, text="Keep console open").pack(side="left")
        self.keep = tk.StringVar(value="On error")
        ttk.Combobox(controls, textvariable=self.keep,
                     values=["Always", "On error", "Never"], state="readonly", width=12).pack(
                         side="left", padx=(8, 16)
                     )
        self.keep.trace_add("write", lambda *_args: self._update_preview())
        run_label = "Run under supervision" if self.supervised else "Run"
        self.run_button = ttk.Button(controls, text=run_label, command=self._run)
        self.run_button.pack(side="right")
        ttk.Button(controls, text="Cancel", command=self.root.destroy).pack(side="right", padx=8)

        preview_frame = ttk.LabelFrame(outer, text="Command preview", padding=6)
        preview_frame.pack(fill="x", pady=(8, 0))
        self.preview = tk.Text(preview_frame, height=3, wrap="word", background="#f4f4f4")
        self.preview.pack(fill="x")
        self.preview.configure(state="disabled")
        self._update_preview()

    def _render_fields(self) -> None:
        for child in self.form.winfo_children():
            child.destroy()
        self.fields.clear()
        specs = list(self.model.root)
        selected = self.command.get()
        if selected:
            specs.extend(self.model.commands.get(selected, []))
        row = 0
        if not specs:
            ttk.Label(self.form, text="No static argparse fields found; use raw arguments below.").grid(
                row=0, column=0, columnspan=3, sticky="w", padx=4, pady=8
            )
        for spec in specs:
            field = ArgumentField(
                self.form, spec, row, self._update_preview, self.script.parent
            )
            self.fields.append(field)
            row += field.rows_used
        self._update_preview()

    def _script_arguments(self, validate: bool = False) -> list[str]:
        args: list[str] = []
        root_count = len(self.model.root)
        for field in self.fields[:root_count]:
            try:
                args.extend(field.arguments())
            except (ValueError, TypeError):
                if validate:
                    raise
        if self.command.get():
            args.append(self.command.get())
        for field in self.fields[root_count:]:
            try:
                args.extend(field.arguments())
            except (ValueError, TypeError):
                if validate:
                    raise
        try:
            args.extend(split_windows_command_line(self.raw.get()))
        except ValueError:
            if validate:
                raise
        return args

    def _display_command(self) -> str:
        return subprocess.list2cmdline(
            [self.interpreter.get().strip(), str(self.script), *self._script_arguments(False)]
        )

    def _update_preview(self) -> None:
        if not hasattr(self, "preview"):
            return
        self.preview.configure(state="normal")
        self.preview.delete("1.0", "end")
        self.preview.insert("1.0", self._display_command())
        self.preview.configure(state="disabled")

    def _set_guardian_text(self, text: str) -> None:
        if not hasattr(self, "guardian_text"):
            return
        self.guardian_text.configure(state="normal")
        self.guardian_text.delete("1.0", "end")
        self.guardian_text.insert("1.0", text)
        self.guardian_text.configure(state="disabled")

    def _load_preflight(self) -> None:
        try:
            from guardian_client import preflight
            response = preflight(str(self.script))
            text = response.get("text") if response.get("ok") else response.get("error")
            if not text:
                text = "Guardian returned no preflight text."
        except Exception as exc:  # noqa: BLE001 - displayed to the user
            text = f"Preflight unavailable: {type(exc).__name__}: {exc}\nYou may still submit the run."
        try:
            self.root.after(0, self._set_guardian_text, text)
        except tk.TclError:
            pass

    def _run(self) -> None:
        interpreter = Path(self.interpreter.get().strip().strip('"'))
        if not interpreter.is_file():
            messagebox.showerror(self.app_title, f"Python interpreter not found:\n{interpreter}")
            return
        if not RUNNER.is_file():
            messagebox.showerror(self.app_title, f"Console runner not found:\n{RUNNER}")
            return
        try:
            script_args = self._script_arguments(True)
        except (ValueError, TypeError) as exc:
            messagebox.showerror(self.app_title, str(exc))
            return
        keep_mode = {"Always": "always", "On error": "error", "Never": "never"}[self.keep.get()]
        if self.supervised:
            self.run_button.configure(state="disabled")
            self._set_guardian_text("Submitting the run to the resident Guardian…")
            threading.Thread(
                target=self._submit_guardian,
                args=(str(interpreter), script_args, keep_mode),
                daemon=True,
            ).start()
            return
        runner_python = _console_executable(Path(sys.executable))
        command = [
            str(runner_python), str(RUNNER), "--keep", keep_mode,
            "--cwd", str(self.script.parent), "--interpreter", str(interpreter),
            "--", str(self.script), *script_args,
        ]
        try:
            subprocess.Popen(command, creationflags=subprocess.CREATE_NEW_CONSOLE)
        except OSError as exc:
            messagebox.showerror(self.app_title, f"Could not start Python:\n{exc}")
            return
        self.root.destroy()

    def _submit_guardian(self, interpreter: str, script_args: list[str], keep_mode: str) -> None:
        try:
            from guardian_client import submit
            response = submit(str(self.script), interpreter, script_args, keep_mode)
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "Guardian rejected the job"))
        except Exception as exc:  # noqa: BLE001 - displayed to the user
            error_type = type(exc).__name__
            error_text = str(exc)
            def failed() -> None:
                self.run_button.configure(state="normal")
                self._set_guardian_text(f"Submission failed: {error_type}: {error_text}")
                messagebox.showerror(self.app_title, f"Could not submit run:\n{error_text}")
            try:
                self.root.after(0, failed)
            except tk.TclError:
                pass
            return

        def accepted() -> None:
            messagebox.showinfo(
                self.app_title,
                f"Guardian accepted job {response['job_id']}.\n\n"
                "The script will run in a console. If it fails, Codex will open a copyable postmortem.",
            )
            self.root.destroy()
        try:
            self.root.after(0, accepted)
        except tk.TclError:
            pass

    def run(self) -> None:
        self.root.mainloop()


def main() -> int:
    supervised = len(sys.argv) == 3 and sys.argv[1] == "--supervised"
    if len(sys.argv) != (3 if supervised else 2):
        root = tk.Tk()
        root.withdraw()
        messagebox.showerror(APP_TITLE, "Expected one Python script path from File Explorer.")
        return 2
    script = Path(sys.argv[2] if supervised else sys.argv[1])
    if not script.is_file() or script.suffix.casefold() not in (".py", ".pyw"):
        root = tk.Tk()
        root.withdraw()
        messagebox.showerror(APP_TITLE, f"Not a Python file:\n{script}")
        return 2
    Launcher(script, supervised=supervised).run()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())