python_console_runner.py

"""Console host for python_arg_launcher.pyw."""
from __future__ import annotations

import argparse
import ctypes
import datetime as dt
import hashlib
import json
import os
import subprocess
import sys
import tempfile
import tkinter as tk
from ctypes import wintypes
from pathlib import Path


def parse_args() -> tuple[argparse.Namespace, list[str]]:
    parser = argparse.ArgumentParser()
    parser.add_argument("--keep", choices=("always", "error", "never"), default="error")
    parser.add_argument("--cwd", required=True)
    parser.add_argument("--interpreter", required=True)
    parser.add_argument("--result-file")
    parser.add_argument("--inject-dir")
    parser.add_argument("--panic-dir")
    parser.add_argument("--crash-dir")
    parser.add_argument("--job-id")
    known, remainder = parser.parse_known_args()
    if remainder and remainder[0] == "--":
        remainder = remainder[1:]
    if not remainder:
        parser.error("missing script path")
    return known, remainder


class _Coord(ctypes.Structure):
    _fields_ = [("X", wintypes.SHORT), ("Y", wintypes.SHORT)]


class _SmallRect(ctypes.Structure):
    _fields_ = [
        ("Left", wintypes.SHORT), ("Top", wintypes.SHORT),
        ("Right", wintypes.SHORT), ("Bottom", wintypes.SHORT),
    ]


class _ConsoleScreenBufferInfo(ctypes.Structure):
    _fields_ = [
        ("dwSize", _Coord),
        ("dwCursorPosition", _Coord),
        ("wAttributes", wintypes.WORD),
        ("srWindow", _SmallRect),
        ("dwMaximumWindowSize", _Coord),
    ]


def read_console_transcript(max_rows: int = 5000) -> str:
    """Read recent text directly from this console's screen/history buffer."""
    if os.name != "nt":
        return ""
    kernel32 = ctypes.windll.kernel32
    get_std_handle = kernel32.GetStdHandle
    get_std_handle.argtypes = [wintypes.DWORD]
    get_std_handle.restype = wintypes.HANDLE
    get_buffer_info = kernel32.GetConsoleScreenBufferInfo
    get_buffer_info.argtypes = [
        wintypes.HANDLE, ctypes.POINTER(_ConsoleScreenBufferInfo)
    ]
    get_buffer_info.restype = wintypes.BOOL
    read_output = kernel32.ReadConsoleOutputCharacterW
    read_output.argtypes = [
        wintypes.HANDLE, wintypes.LPWSTR, wintypes.DWORD,
        _Coord, ctypes.POINTER(wintypes.DWORD),
    ]
    read_output.restype = wintypes.BOOL

    handle = get_std_handle(-11)  # STD_OUTPUT_HANDLE
    info = _ConsoleScreenBufferInfo()
    if handle in (None, 0, wintypes.HANDLE(-1).value) or not get_buffer_info(handle, ctypes.byref(info)):
        return ""
    width = max(1, int(info.dwSize.X))
    final_row = max(0, int(info.dwCursorPosition.Y))
    first_row = max(0, final_row - max_rows + 1)
    lines: list[str] = []
    read = wintypes.DWORD()
    for row in range(first_row, final_row + 1):
        buffer = ctypes.create_unicode_buffer(width + 1)
        ok = read_output(
            handle, buffer, width, _Coord(0, row), ctypes.byref(read)
        )
        lines.append(buffer[: read.value].rstrip() if ok else "")
    while lines and not lines[0]:
        lines.pop(0)
    while lines and not lines[-1]:
        lines.pop()
    return "\n".join(lines)


def save_transcript(text: str, script: str) -> Path:
    safe_stem = "".join(c if c.isalnum() or c in "-_" else "_" for c in Path(script).stem)
    stamp = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"{stamp}_{safe_stem or 'python'}.log"
    roots = [
        Path(os.environ.get("LOCALAPPDATA", Path.home())) / "PythonFileLauncher" / "logs",
        Path(tempfile.gettempdir()) / "PythonFileLauncher" / "logs",
    ]
    last_error: OSError | None = None
    for root in roots:
        try:
            root.mkdir(parents=True, exist_ok=True)
            path = root / filename
            path.write_text(text, encoding="utf-8")
            return path
        except OSError as exc:
            last_error = exc
    raise OSError(f"Could not save transcript: {last_error}")


def write_text_atomic(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(text, encoding="utf-8")
    os.replace(temporary, path)


def write_json_atomic(path: Path, value: dict) -> None:
    write_text_atomic(path, json.dumps(value, indent=2, ensure_ascii=False))


def newest_panic_report(crash_dir: str | None) -> tuple[str | None, dict | None]:
    if not crash_dir:
        return None, None
    root = Path(crash_dir)
    candidates = sorted(root.glob("crash_Global_*.json"), key=lambda item: item.stat().st_mtime)
    if not candidates:
        return None, None
    path = candidates[-1]
    try:
        return str(path), json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return str(path), None


def failure_fingerprint(code: int, panic: dict | None, transcript: str) -> str | None:
    if code == 0 and not panic:
        return None
    if panic:
        traceback_data = panic.get("traceback", [])
        last_frame = traceback_data[-1] if isinstance(traceback_data, list) and traceback_data else ""
        signature = "|".join(
            str(item) for item in (
                panic.get("exception_type"), panic.get("error_message"), last_frame,
            )
        )
    else:
        signature = f"exit={code}|{transcript[-4000:]}"
    return hashlib.sha256(signature.encode("utf-8", errors="replace")).hexdigest()[:16]


def copy_to_clipboard(text: str) -> None:
    root = tk.Tk()
    root.withdraw()
    root.clipboard_clear()
    root.clipboard_append(text)
    root.update()
    root.destroy()


def transcript_menu(transcript: str, log_path: Path) -> None:
    print(f"Transcript saved to: {log_path}")
    while True:
        try:
            answer = input("[C]opy transcript  [O]pen log  [Enter] close: ").strip().casefold()
        except KeyboardInterrupt:
            # At this post-run prompt, make habitual Ctrl+C mean Copy, not exit.
            print()
            answer = "c"
        except EOFError:
            return
        if not answer:
            return
        if answer == "c":
            try:
                copy_to_clipboard(transcript)
                print("Transcript copied to the Windows clipboard.")
            except tk.TclError as exc:
                print(f"Clipboard copy failed: {exc}")
        elif answer == "o":
            try:
                os.startfile(log_path)
            except OSError as exc:
                print(f"Could not open log: {exc}")
        else:
            print("Choose C, O, or Enter.")


def main() -> int:
    options, script_and_args = parse_args()
    command = [options.interpreter, *script_and_args]
    print("Python File Launcher")
    print("Working directory:", options.cwd)
    print("Command:", subprocess.list2cmdline(command))
    print("-" * 78)
    started = dt.datetime.now(dt.timezone.utc).isoformat()
    environment = os.environ.copy()
    if options.inject_dir:
        existing = environment.get("PYTHONPATH", "")
        environment["PYTHONPATH"] = options.inject_dir + (os.pathsep + existing if existing else "")
    if options.panic_dir:
        environment["PYTHON_GUARDIAN_PANIC_DIR"] = options.panic_dir
    if options.crash_dir:
        environment["PYTHON_GUARDIAN_CRASH_DIR"] = options.crash_dir
    if options.job_id:
        environment["PYTHON_GUARDIAN_JOB_ID"] = options.job_id
    try:
        result = subprocess.run(command, cwd=options.cwd, env=environment)
        code = result.returncode
    except OSError as exc:
        print(f"\nCould not launch process: {exc}", file=sys.stderr)
        code = 1
    print("\n" + "-" * 78)
    print(f"Process exited with code {code}.")
    transcript = read_console_transcript()
    if not transcript:
        transcript = (
            "Python File Launcher\n"
            f"Working directory: {options.cwd}\n"
            f"Command: {subprocess.list2cmdline(command)}\n"
            f"Process exited with code {code}.\n"
            "Console screen capture was unavailable."
        )
    log_path: Path | None = None
    if options.result_file:
        result_path = Path(options.result_file)
        log_path = result_path.with_name("transcript.log")
        write_text_atomic(log_path, transcript)
        panic_path, panic = newest_panic_report(options.crash_dir)
        write_json_atomic(result_path, {
            "job_id": options.job_id,
            "script": str((Path(options.cwd) / script_and_args[0]).resolve()
                          if not Path(script_and_args[0]).is_absolute()
                          else Path(script_and_args[0]).resolve()),
            "arguments": script_and_args[1:],
            "command": command,
            "cwd": options.cwd,
            "started_at": started,
            "finished_at": dt.datetime.now(dt.timezone.utc).isoformat(),
            "exit_code": code,
            "incident": code != 0 or panic_path is not None,
            "transcript_path": str(log_path),
            "panic_report": panic_path,
            "fingerprint": failure_fingerprint(code, panic, transcript),
        })
    if options.keep == "always" or (options.keep == "error" and code != 0):
        if log_path is None:
            log_path = save_transcript(transcript, script_and_args[0])
        transcript_menu(transcript, log_path)
    return code


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