#!/usr/bin/env python3 """ companion.py — 웹 코드 에디터용 로컬 Python 실행 컴패니언 사용법: pip install websockets python companion.py 웹 에디터(브라우저)가 ws://localhost:9000 에 연결해서 코드 실행 요청을 보내면 이 프로세스가 로컬에서 실행합니다. pygame 등 GUI 라이브러리도 정상 동작합니다. """ import asyncio import json import os import sys import tempfile import threading try: import websockets except ImportError: print('오류: websockets 패키지가 없습니다.') print('설치: pip install websockets') sys.exit(1) HOST = '127.0.0.1' PORT = 9000 IS_WINDOWS = sys.platform == 'win32' async def _safe_send(ws, data: str): try: await ws.send(data) except Exception: pass class Runner: def __init__(self): self.proc = None self.tmp_path = None self.master_fd = None # Unix PTY only self._shell_task = None async def kill(self): if IS_WINDOWS and hasattr(self, '_win_pty') and self._win_pty: try: self._win_pty.close() except Exception: pass self._win_pty = None if self._shell_task: self._shell_task.cancel() try: await self._shell_task except (asyncio.CancelledError, Exception): pass self._shell_task = None if self.master_fd is not None: try: asyncio.get_running_loop().remove_reader(self.master_fd) except Exception: pass try: os.close(self.master_fd) except Exception: pass self.master_fd = None if self.proc and self.proc.returncode is None: try: self.proc.kill() await self.proc.wait() except Exception: pass if self.tmp_path: try: os.unlink(self.tmp_path) except Exception: pass self.tmp_path = None self.proc = None async def start_shell(self, ws, rows: int = 24, cols: int = 80): if IS_WINDOWS: await self._start_shell_windows(ws, rows, cols) else: await self._start_shell_unix(ws, rows, cols) async def _start_shell_unix(self, ws, rows: int, cols: int): import pty, fcntl, termios, struct master_fd, slave_fd = pty.openpty() winsize = struct.pack('HHHH', rows, cols, 0, 0) fcntl.ioctl(slave_fd, termios.TIOCSWINSZ, winsize) fl = fcntl.fcntl(master_fd, fcntl.F_GETFL) fcntl.fcntl(master_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) self.proc = await asyncio.create_subprocess_exec( '/bin/bash', '--login', stdin=slave_fd, stdout=slave_fd, stderr=slave_fd, close_fds=True, env={**os.environ, 'TERM': 'xterm-256color'}, ) os.close(slave_fd) self.master_fd = master_fd await ws.send(json.dumps({'type': 'shell_ready', 'platform': 'windows' if IS_WINDOWS else 'unix', 'home': os.path.expanduser('~')})) self._shell_task = asyncio.create_task(self._stream_shell_unix(ws)) async def _start_shell_windows(self, ws, rows: int = 24, cols: int = 80): try: from winpty import PtyProcess self._win_pty = PtyProcess.spawn( ['cmd.exe'], dimensions=(rows, cols), ) await ws.send(json.dumps({'type': 'shell_ready', 'platform': 'windows' if IS_WINDOWS else 'unix', 'home': os.path.expanduser('~')})) self._shell_task = asyncio.create_task(self._stream_shell_winpty(ws)) print('[셸] winpty 사용', flush=True) except ImportError: print('[셸] pywinpty 없음 → pipe 모드 (pip install pywinpty 권장)', flush=True) self.proc = await asyncio.create_subprocess_exec( 'cmd.exe', '/K', 'chcp 65001 > nul', stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, env={**os.environ}, ) await ws.send(json.dumps({'type': 'shell_ready', 'platform': 'windows' if IS_WINDOWS else 'unix', 'home': os.path.expanduser('~')})) self._shell_task = asyncio.create_task(self._stream_shell_pipe(ws)) async def _stream_shell_unix(self, ws): loop = asyncio.get_running_loop() done = asyncio.Event() def on_readable(): fd = self.master_fd if fd is None: done.set() return try: data = os.read(fd, 4096) loop.create_task(_safe_send(ws, json.dumps({ 'type': 'shell_output', 'data': data.decode('utf-8', errors='replace'), }))) except BlockingIOError: pass except OSError: try: loop.remove_reader(fd) except Exception: pass done.set() loop.add_reader(self.master_fd, on_readable) print(f'[셸스트림] 시작 fd={self.master_fd}', flush=True) try: await done.wait() except asyncio.CancelledError: fd = self.master_fd if fd is not None: try: loop.remove_reader(fd) except Exception: pass return if self.proc: try: await self.proc.wait() except Exception: pass try: await ws.send(json.dumps({ 'type': 'shell_exit', 'code': self.proc.returncode if self.proc else -1, })) except Exception: pass async def _stream_shell_winpty(self, ws): loop = asyncio.get_running_loop() try: while self._win_pty and self._win_pty.isalive(): try: data = await loop.run_in_executor(None, self._win_pty.read, 4096) if data: await _safe_send(ws, json.dumps({ 'type': 'shell_output', 'data': data, })) except EOFError: break except Exception: break except asyncio.CancelledError: pass try: await ws.send(json.dumps({'type': 'shell_exit', 'code': 0})) except Exception: pass async def _stream_shell_pipe(self, ws): try: while self.proc and self.proc.returncode is None: try: chunk = await asyncio.wait_for( self.proc.stdout.read(512), timeout=0.05 ) if chunk: await _safe_send(ws, json.dumps({ 'type': 'shell_output', 'data': chunk.decode('utf-8', errors='replace').replace('\x00', ''), })) except asyncio.TimeoutError: continue except asyncio.CancelledError: pass except Exception as e: print(f'[pipe 오류] {e}', flush=True) if self.proc: try: await self.proc.wait() except Exception: pass try: await ws.send(json.dumps({ 'type': 'shell_exit', 'code': self.proc.returncode if self.proc else -1, })) except Exception: pass def resize(self, cols: int, rows: int): if IS_WINDOWS: if hasattr(self, '_win_pty') and self._win_pty: try: self._win_pty.setwinsize(rows, cols) except Exception: pass return if self.master_fd is None: return try: import fcntl, termios, struct winsize = struct.pack('HHHH', rows, cols, 0, 0) fcntl.ioctl(self.master_fd, termios.TIOCSWINSZ, winsize) except Exception: pass def write_input(self, data: str): if IS_WINDOWS and hasattr(self, '_win_pty') and self._win_pty: try: self._win_pty.write(data) except Exception as e: print(f'[입력오류] {e}', flush=True) elif self.master_fd is not None: try: os.write(self.master_fd, data.encode('utf-8')) except Exception as e: print(f'[입력오류] {e}', flush=True) elif self.proc and self.proc.stdin and self.proc.returncode is None: try: self.proc.stdin.write(data.encode('utf-8')) except Exception: pass async def handle(ws): runner = Runner() addr = ws.remote_address print(f'[연결] {addr}') try: async for raw in ws: try: msg = json.loads(raw) except Exception: continue t = msg.get('type') if t == 'run': await runner.kill() code = msg.get('code', '') filename = msg.get('filename', 'script.py') ext = os.path.splitext(filename)[1] or '.py' with tempfile.NamedTemporaryFile( mode='w', suffix=ext, delete=False, encoding='utf-8' ) as f: f.write(code) runner.tmp_path = f.name print(f'[실행] {filename}') await ws.send(json.dumps({'type': 'started', 'filename': filename})) try: runner.proc = await asyncio.create_subprocess_exec( sys.executable, runner.tmp_path, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, env={**os.environ}, ) while True: chunk = await runner.proc.stdout.read(512) if not chunk: break await ws.send(json.dumps({ 'type': 'output', 'data': chunk.decode('utf-8', errors='replace'), })) await runner.proc.wait() code_ret = runner.proc.returncode print(f'[종료] {filename} → exit {code_ret}') await ws.send(json.dumps({'type': 'exit', 'code': code_ret})) except Exception as e: await ws.send(json.dumps({'type': 'error', 'data': str(e)})) finally: if runner.tmp_path: try: os.unlink(runner.tmp_path) except Exception: pass runner.tmp_path = None elif t == 'shell': await runner.kill() rows = msg.get('rows', 24) cols = msg.get('cols', 80) print(f'[셸] 시작 ({cols}x{rows})') await runner.start_shell(ws, rows=rows, cols=cols) elif t == 'resize': runner.resize(msg.get('cols', 80), msg.get('rows', 24)) elif t == 'input': runner.write_input(msg.get('data', '')) if runner.proc and runner.proc.stdin and runner.proc.returncode is None and IS_WINDOWS: try: await runner.proc.stdin.drain() except Exception: pass elif t == 'kill': await runner.kill() print('[강제 종료]') await ws.send(json.dumps({'type': 'killed'})) elif t == 'ping': await ws.send(json.dumps({'type': 'pong'})) except Exception as e: print(f'[연결오류] {type(e).__name__}: {e}', flush=True) finally: await runner.kill() print(f'[종료] {addr}', flush=True) def _make_tray_icon(): from PIL import Image, ImageDraw size = 64 img = Image.new('RGBA', (size, size), (0, 0, 0, 0)) d = ImageDraw.Draw(img) d.ellipse([2, 2, size - 2, size - 2], fill=(52, 120, 200)) d.ellipse([18, 18, 46, 46], fill='white') d.ellipse([24, 24, 40, 40], fill=(52, 120, 200)) return img def _run_asyncio_in_thread(stop_event): async def _serve(): async with websockets.serve(handle, HOST, PORT): while not stop_event.is_set(): await asyncio.sleep(0.3) try: asyncio.run(_serve()) except Exception as e: print(f'[서버오류] {e}', flush=True) def _run_with_tray(): import pystray stop_event = threading.Event() t = threading.Thread(target=_run_asyncio_in_thread, args=(stop_event,), daemon=True) t.start() def on_quit(icon, item): stop_event.set() icon.stop() icon = pystray.Icon( 'python_companion', _make_tray_icon(), f'Python Companion ws://{HOST}:{PORT}', menu=pystray.Menu( pystray.MenuItem('Python Companion', None, enabled=False), pystray.MenuItem(f'포트: {PORT} | 연결 대기 중', None, enabled=False), pystray.Menu.SEPARATOR, pystray.MenuItem('종료', on_quit), ), ) print('시스템 트레이에서 실행 중 (우클릭 → 종료)', flush=True) icon.run() stop_event.set() async def main(): async with websockets.serve(handle, HOST, PORT): await asyncio.Future() if __name__ == '__main__': print('━' * 40) print(f'🐍 Python Companion ws://{HOST}:{PORT}') print(f'플랫폼: {"Windows" if IS_WINDOWS else "Unix/Linux"}') print('━' * 40) print(f'Python: {sys.executable}') try: import pystray from PIL import Image _run_with_tray() except ImportError: print('웹 에디터에서 연결 대기 중...\nCtrl+C 로 종료\n', flush=True) try: asyncio.run(main()) except KeyboardInterrupt: print('\n컴패니언 종료')